Something went wrong. Try again.
A community based topic aggregation platform built on atproto
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438package posts
import ( "context" "errors" "fmt" "sync" "testing" "time"
"github.com/stretchr/testify/assert" "github.com/stretchr/testify/require")
// The acceptance engine's queue driver (docs/POSTS.md §5.6,// §8): what turns "the engine can settle a subject" into "every stranded// subject eventually gets settled".//// T0, because the driver is a scheduling policy over two interfaces and nothing// it decides needs a database to be true. The backlog QUERY it drives is T1// (internal/db/postgres/admission_queue_repo_test.go); this is the half that// decides what to do with the answer.//// # THE THREE PROPERTIES, AND WHY EACH IS A REAL FAILURE MODE//// - ONE SUBJECT AT A TIME, GROUPED BY COMMUNITY. swapCommit is repo-global, so// two workers on one community's repo starve each other's writes — task 4// recorded this before the driver existed. v1 satisfies it trivially with a// single goroutine, but the ORDER still matters: the partition a worker pool// would eventually shard on has to already be the shape of the data.// - A DEFERRAL IS NOT A RETRY SIGNAL. EngineDeferred is the common outcome —// an expired token, an undecoded CID, an unreachable lookup — and a driver// that re-offered it immediately would hammer a PDS that is already// unhappy. This is the difference between a queue and a spin loop.// - ONE BAD SUBJECT MUST NOT STOP THE PASS. An error on one row aborting the// loop means one poisoned admission freezes every other community's posts,// and the symptom appears as "posts stopped becoming visible" with nothing// pointing at the row that caused it.
// fakeSubjects is the backlog.type fakeSubjects struct { batches [][]PendingSubject err error
limits []int calls int}
func (f *fakeSubjects) ListPendingSubjects(_ context.Context, limit int) ([]PendingSubject, error) { f.limits = append(f.limits, limit) f.calls++ if f.err != nil { return nil, f.err } if len(f.batches) == 0 { return nil, nil } batch := f.batches[0] if len(f.batches) > 1 { f.batches = f.batches[1:] } return batch, nil}
// fakeEngine records the subjects it was handed, in order, and answers each// with a scripted outcome.type fakeEngine struct { mu sync.Mutex seen []PendingSubject outcomes map[string]EngineOutcome errs map[string]error}
func newFakeEngine() *fakeEngine { return &fakeEngine{ outcomes: map[string]EngineOutcome{}, errs: map[string]error{}, }}
func (e *fakeEngine) ProcessAdmission(_ context.Context, communityDID, postURI string) (EngineOutcome, error) { e.mu.Lock() defer e.mu.Unlock() e.seen = append(e.seen, PendingSubject{CommunityDID: communityDID, PostURI: postURI}) if err := e.errs[postURI]; err != nil { return EngineDeferred, err } if outcome, ok := e.outcomes[postURI]; ok { return outcome, nil } return EngineAccepted, nil}
func (e *fakeEngine) uris() []string { e.mu.Lock() defer e.mu.Unlock() uris := make([]string, 0, len(e.seen)) for _, subject := range e.seen { uris = append(uris, subject.PostURI) } return uris}
func (e *fakeEngine) communityOrder() []string { e.mu.Lock() defer e.mu.Unlock() var order []string for _, subject := range e.seen { if len(order) == 0 || order[len(order)-1] != subject.CommunityDID { order = append(order, subject.CommunityDID) } } return order}
// queueClock is a mutable instant the driver reads through Clock.type queueClock struct{ at time.Time }
func (c *queueClock) now() Clock { return func() time.Time { return c.at } }func (c *queueClock) advance(d time.Duration) { c.at = c.at.Add(d) }func newQueueClock() *queueClock { return &queueClock{at: time.Date(2026, 8, 8, 9, 0, 0, 0, time.UTC)} }
func subject(community, rkey string) PendingSubject { return PendingSubject{ CommunityDID: community, PostURI: "at://did:plc:queueauthor/social.coves.community.postv2/" + rkey, CreatedAt: time.Date(2026, 8, 8, 8, 0, 0, 0, time.UTC), }}
func TestQueueDriver_ProcessesEverySubjectOncePerPass(t *testing.T) { t.Parallel()
clock := newQueueClock() first, second := subject("did:plc:qa", "a"), subject("did:plc:qa", "b") subjects := &fakeSubjects{batches: [][]PendingSubject{{first, second}}} engine := newFakeEngine()
report, err := NewQueueDriver(subjects, engine, clock.now(), WithQueueBatchSize(10)). RunPass(context.Background())
require.NoError(t, err) assert.Equal(t, []string{first.PostURI, second.PostURI}, engine.uris(), "every listed subject must be offered to the engine, in the order the backlog returned them") assert.Equal(t, 2, report.Listed) assert.Equal(t, 2, report.Processed) assert.Equal(t, 2, report.Settled) assert.Equal(t, []int{10}, subjects.limits, "the batch size must reach the query: a driver that listed the whole backlog would hold one transaction open across a table that grows forever")}
func TestQueueDriver_GroupsWorkByCommunity(t *testing.T) { t.Parallel()
clock := newQueueClock() // Interleaved on the way in — the backlog is ordered by age across // communities, so this is the shape it genuinely returns. subjects := &fakeSubjects{batches: [][]PendingSubject{{ subject("did:plc:qa", "a1"), subject("did:plc:qb", "b1"), subject("did:plc:qa", "a2"), subject("did:plc:qb", "b2"), }}} engine := newFakeEngine()
_, err := NewQueueDriver(subjects, engine, clock.now()).RunPass(context.Background()) require.NoError(t, err)
// Each community's subjects arrive contiguously. In v1 the single goroutine // already makes this safe, so what the assertion protects is the NEXT // version: the moment this grows a worker pool, the shard key has to be the // community, and a driver whose output interleaves communities is one that // would be sharded on nothing. order := engine.communityOrder() assert.Len(t, order, 2, "a community's subjects must be contiguous; interleaving them (%v) means a future worker pool has no partition to shard on and would run two writers against one repo, where swapCommit is repo-global", order) assert.ElementsMatch(t, []string{"did:plc:qa", "did:plc:qb"}, order) assert.Len(t, engine.uris(), 4, "grouping must not drop or duplicate work")}
func TestQueueDriver_DoesNotTouchOneSubjectTwiceInAPass(t *testing.T) { t.Parallel()
clock := newQueueClock() hot := subject("did:plc:qa", "hot") // The same subject twice in one listing: exactly what a duplicated row or a // racing edit produces, and the shape §8's edit-debounce has to survive. subjects := &fakeSubjects{batches: [][]PendingSubject{{hot, hot, subject("did:plc:qa", "cold")}}} engine := newFakeEngine() engine.outcomes[hot.PostURI] = EngineDeferred
report, err := NewQueueDriver(subjects, engine, clock.now()).RunPass(context.Background()) require.NoError(t, err)
seen := 0 for _, uri := range engine.uris() { if uri == hot.PostURI { seen++ } } assert.Equalf(t, 1, seen, "a subject was processed %d times in one pass. §8's edit-debounce is exactly this rule: a post being edited in a storm coalesces "+ "into one pending_reacceptance row, and a driver that re-decided it per occurrence would re-run the whole policy per keystroke", seen) assert.Equal(t, 1, report.Deferred)}
func TestQueueDriver_HoldsADeferredSubjectBackUntilItsBackoffElapses(t *testing.T) { t.Parallel()
clock := newQueueClock() stuck := subject("did:plc:qa", "stuck") healthy := subject("did:plc:qb", "healthy") // The same backlog every pass — which is what the query genuinely returns // while the subject stays undecided. subjects := &fakeSubjects{batches: [][]PendingSubject{{stuck, healthy}}} engine := newFakeEngine() engine.outcomes[stuck.PostURI] = EngineDeferred
driver := NewQueueDriver(subjects, engine, clock.now(), WithQueueBackoff(time.Minute, 10*time.Minute)) ctx := context.Background()
first, err := driver.RunPass(ctx) require.NoError(t, err) assert.Equal(t, 2, first.Processed) assert.Equal(t, 1, first.Deferred)
// Immediately again. A community with a dead token defers on every pass, and // a driver that re-offered it at once would turn a scheduled job into a spin // against a PDS that is already failing. clock.advance(time.Second) second, err := driver.RunPass(ctx) require.NoError(t, err)
countIn := func(uris []string, target string) int { n := 0 for _, uri := range uris { if uri == target { n++ } } return n } assert.Equalf(t, 1, countIn(engine.uris(), stuck.PostURI), "a subject deferred one second ago was offered again; the backoff exists so one wedged community does not become a hot loop")
// The healthy subject keeps flowing. This is the half that makes the backoff // per-SUBJECT rather than global: one community's dead credentials must not // stall everyone else's posts, which is precisely what a global pause does. assert.Equalf(t, 2, countIn(engine.uris(), healthy.PostURI), "the backoff must be per-subject: a healthy subject stopped being processed because a DIFFERENT one deferred") assert.Equal(t, 1, second.Processed, "only the healthy subject was due")
// Past the backoff, the deferred subject is due again — a deferral is "look // again later", and a driver that never looked again would strand it forever. clock.advance(2 * time.Minute) third, err := driver.RunPass(ctx) require.NoError(t, err) assert.Equalf(t, 2, countIn(engine.uris(), stuck.PostURI), "a deferred subject must be retried once its backoff elapses; holding it forever is how a recovered community's posts stay pending") assert.Equal(t, 2, third.Processed)}
func TestQueueDriver_AFailedSubjectDoesNotAbortThePass(t *testing.T) { t.Parallel()
clock := newQueueClock() poison := subject("did:plc:qa", "poison") after := subject("did:plc:qb", "after") subjects := &fakeSubjects{batches: [][]PendingSubject{{poison, after}}} engine := newFakeEngine() engine.errs[poison.PostURI] = errors.New("the decision blew up")
report, err := NewQueueDriver(subjects, engine, clock.now()).RunPass(context.Background())
require.NoError(t, err, "a pass is only an error when the BACKLOG could not be read; a subject that failed is a counted outcome") assert.Containsf(t, engine.uris(), after.PostURI, "the pass stopped at the first failing subject. One poisoned admission would then freeze every other community's posts, "+ "and the symptom — 'posts stopped appearing' — points nowhere near the row that caused it") assert.Equal(t, 1, report.Failed) assert.Equal(t, 1, report.Settled)
// Failures and deferrals are counted apart because they mean opposite things // to whoever is deciding whether to page: a pass that defers everything is // usually credentials and will clear, a pass that FAILS everything is a bug. assert.Zero(t, report.Deferred, "an error is not a deferral")}
func TestQueueDriver_ReportsAnUnreadableBacklog(t *testing.T) { t.Parallel()
clock := newQueueClock() subjects := &fakeSubjects{err: errors.New("admissions table unreachable")}
report, err := NewQueueDriver(subjects, newFakeEngine(), clock.now()).RunPass(context.Background())
require.Error(t, err, "a backlog that could not be read is the one failure a pass has nothing to do about") assert.Zero(t, report.Processed)}
func TestQueueDriver_SnapshotReportsTheBacklogAndTheLastPass(t *testing.T) { t.Parallel()
clock := newQueueClock() oldest := subject("did:plc:qa", "oldest") oldest.CreatedAt = clock.at.Add(-90 * time.Minute) newer := subject("did:plc:qa", "newer") newer.CreatedAt = clock.at.Add(-5 * time.Minute)
subjects := &fakeSubjects{batches: [][]PendingSubject{{oldest, newer}}} engine := newFakeEngine() engine.outcomes[newer.PostURI] = EngineDeferred
driver := NewQueueDriver(subjects, engine, clock.now())
// Before the first pass, LastPassAt is nil — the one field that distinguishes // "the driver has never run" from "the driver ran and found nothing". A // driver that has died produces no error and no log; it simply stops, and // every symptom shows up somewhere else as posts that never become visible. require.Nil(t, driver.Snapshot().LastPassAt, "before any pass, lastPassAt must be nil: a zero time would read as a pass that happened at the epoch")
_, err := driver.RunPass(context.Background()) require.NoError(t, err)
snapshot := driver.Snapshot() assert.Equal(t, 2, snapshot.PendingBacklog) require.NotNil(t, snapshot.OldestPendingAt) assert.Equal(t, oldest.CreatedAt, *snapshot.OldestPendingAt, "the oldest undecided subject's age is the queue's only early warning: a backlog that is merely big is a busy instance, one whose oldest entry keeps ageing is an engine that has stopped settling anything") require.NotNil(t, snapshot.LastPassAt) assert.Equal(t, clock.at, *snapshot.LastPassAt) assert.Equal(t, 1, snapshot.LastPassDeferred) assert.Zero(t, snapshot.LastPassFailed)}
func TestQueueDriver_OverFetchesPastBackedOffSubjects(t *testing.T) { t.Parallel()
// BATCH-PREFIX STARVATION. The backlog is ordered oldest-first, so the // subjects most likely to be stuck — a community whose credentials expired // weeks ago, a post whose content never decoded — are exactly the ones that // sit at the front of it forever. Ask for LIMIT rows, get LIMIT stuck ones, // skip all of them for backoff, and the pass does nothing. Every pass. The // queue is not empty and the driver is not broken; it simply never sees past // its own prefix, and a healthy post behind them is never decided. // // The fix stays in the DRIVER: over-fetch and keep skipping until the batch // is filled. Pushing the backoff into the query would mean persisting // retry-not-before, and the backoff is deliberately a disposable in-memory // hint that a restart may forget (see QueueDriver.deferrals). clock := newQueueClock()
const batch = 3 stuck := make([]PendingSubject, batch) for i := range stuck { stuck[i] = subject("did:plc:qstuck", fmt.Sprintf("stuck%d", i)) } young := subject("did:plc:qyoung", "young")
subjects := &fakeSubjects{batches: [][]PendingSubject{append(append([]PendingSubject{}, stuck...), young)}} engine := newFakeEngine() for _, s := range stuck { engine.outcomes[s.PostURI] = EngineDeferred }
driver := NewQueueDriver(subjects, engine, clock.now(), WithQueueBatchSize(batch), WithQueueBackoff(time.Minute, 10*time.Minute)) ctx := context.Background()
// Pass one settles nothing and backs the whole prefix off. first, err := driver.RunPass(ctx) require.NoError(t, err) require.Equal(t, batch, first.Deferred, "fixture: the whole prefix must defer")
// Pass two, inside the backoff window. Every stuck subject is held back, so // a driver that asked for exactly LIMIT rows would process nothing at all. clock.advance(time.Second) second, err := driver.RunPass(ctx) require.NoError(t, err)
assert.Contains(t, engine.uris(), young.PostURI, "the young subject was never reached: the pass asked for exactly the batch size, got a prefix of backed-off "+ "subjects, and skipped all of them — so a stuck community at the head of the backlog starves everything behind it forever") assert.Equalf(t, 1, second.Processed, "the pass must fill its batch past the held-back prefix, processing the young subject and nothing else; got %d processed", second.Processed)
require.GreaterOrEqual(t, len(subjects.limits), 2) assert.Greaterf(t, subjects.limits[len(subjects.limits)-1], batch, "the query must be asked for MORE than the batch size (limits seen: %v). The driver cannot skip past what it "+ "never fetched, and the over-fetch factor is what bounds how deep a stuck prefix it can see past", subjects.limits)}
func TestQueueDriver_PrunesDeferralsForSubjectsThatLeaveTheBacklog(t *testing.T) { t.Parallel()
// A deferral outlives its subject. The row is settled by somebody else — the // synchronous fast path, a firehose acceptance, a moderator's removal — and // it stops being listed, but its entry stays in the map forever. On a busy // instance that map is then an unbounded leak keyed by every subject the // driver ever deferred, held for the lifetime of the process. // // It is also wrong on re-entry: a subject that leaves the backlog and comes // back (an edit reopening an accepted post) arrives carrying a stale backoff // it did nothing to earn, and waits out a delay that was about a completely // different decision. clock := newQueueClock() leaving := subject("did:plc:qleave", "leaving") staying := subject("did:plc:qstay", "staying")
subjects := &fakeSubjects{batches: [][]PendingSubject{ {leaving, staying}, // Second pass: `leaving` was settled elsewhere and is gone from the // backlog. `staying` is still owed a decision. {staying}, }} engine := newFakeEngine() engine.outcomes[leaving.PostURI] = EngineDeferred engine.outcomes[staying.PostURI] = EngineDeferred
driver := NewQueueDriver(subjects, engine, clock.now(), WithQueueBackoff(time.Minute, 10*time.Minute)) ctx := context.Background()
_, err := driver.RunPass(ctx) require.NoError(t, err) require.Equal(t, 2, driver.Snapshot().DeferredSubjects, "fixture: both subjects must be holding a deferral")
clock.advance(time.Second) _, err = driver.RunPass(ctx) require.NoError(t, err)
assert.Equalf(t, 1, driver.Snapshot().DeferredSubjects, "the deferral for a subject that has left the backlog was kept. The map is keyed by subject and never swept, so "+ "it grows for the life of the process — and a subject that comes back (an edit reopening an accepted post) "+ "inherits a stale backoff it did nothing to earn")}