Something went wrong. Try again.
A community based topic aggregation platform built on atproto
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471package posts
import ( "context" "errors" "fmt" "log" "sync" "time")
// The acceptance engine's driver: the thing that decides WHEN the engine runs// and on what (docs/POSTS.md §5.6, §8).//// The engine settles one subject. Nothing until now decided which subjects, in// what order, or how often — the firehose consumer pushes work at it today, and// task 6's synchronous fast path will push more, and neither can see a subject// that was left pending because a credential expired or a lookup blipped. This is the pull side: a periodic pass// over the undecided backlog that eventually reaches every stranded row.//// # IT IS A SINGLE GOROUTINE, AND THAT IS THE PER-COMMUNITY SERIALIZATION//// Task 4 recorded the requirement: serialize the queue per community DID,// because swapCommit is repo-global and sibling workers on one busy community// would starve each other's removals. In v1 that requirement is satisfied// trivially and completely — one goroutine walks one ordered list, so no two// subjects of any community are ever in flight together. The ordering matters// anyway: the list is grouped by community so that when this DOES grow a worker// pool, the partition it has to shard on is already the shape of the data,// rather than something a later change has to introduce.//// # A DEFERRAL IS NOT A FAILURE, AND MUST NOT BECOME A SPIN//// EngineDeferred is the common outcome, not the exceptional one: a community// whose token expired, a post whose content CID has not been decoded yet, a// policy lookup that could not be reached. Every one of those is "look again// later", and a driver that took it as "try again now" would turn one wedged// subject into a hot loop against a PDS that is already unhappy. Deferred// subjects therefore carry a per-subject backoff, and the same subject is never// touched twice in one pass — §8's edit-debounce falls out of the same rule,// since a post being edited in a storm coalesces into one pending_reacceptance// row that this pass sees exactly once.
// AdmissionProcessor settles one subject. Satisfied by *AcceptanceEngine.type AdmissionProcessor interface { ProcessAdmission(ctx context.Context, communityDID, postURI string) (EngineOutcome, error)}
// PendingSubjectLister supplies the backlog. Satisfied by AdmissionRepository.type PendingSubjectLister interface { ListPendingSubjects(ctx context.Context, limit int) ([]PendingSubject, error)}
// PassReport is what one pass did, for logs and for the health surface.//// Deferred and Failed are counted separately because they mean opposite things// to an operator. A pass that defers everything is usually a credential or// connectivity problem that will clear; a pass that FAILS everything is a bug.// Collapsing them into one number would make the first look like the second at// exactly the moment somebody is deciding whether to page.type PassReport struct { // Listed is how many subjects the backlog query returned. Listed int
// Processed is how many were handed to the engine — Listed minus those the // backoff held back. Processed int
// Settled is how many reached a verdict: accepted, rejected, removed or // repinned. Settled int
// Deferred is how many the engine declined to decide yet. Deferred int
// Failed is how many returned an error. Failed int
// StartedAt is when the pass began. StartedAt time.Time}
// QueueSnapshot is the driver's health surface: what the backlog looks like and// when it was last worked.//// LastPassAt is the liveness signal and the reason the snapshot exists at all.// A driver that has stopped running produces no logs and no errors — it simply// stops, and every symptom of that appears somewhere else, as posts that never// become visible. An operator needs one number that says the pass is happening.type QueueSnapshot struct { PendingBacklog int
// OldestPendingAt is the created_at of the oldest subject the last pass // listed, or nil when the backlog was empty. OldestPendingAt *time.Time
// LastPassAt is nil until the first pass completes — distinguishing "the // driver has never run" from "the driver ran and found nothing", which look // identical in every other field. LastPassAt *time.Time
LastPassDeferred int LastPassFailed int
// DeferredSubjects is how many subjects are currently holding a backoff. // // Exposed because it is the only external view of a map that would // otherwise grow for the life of the process: entries are keyed by subject // and a subject settled by somebody else — the fast path, a firehose // acceptance, a moderator — stops being listed without ever telling the // driver to forget it. A number that climbs while the backlog does not is // the leak, visible. DeferredSubjects int}
// QueueDriverOption configures the driver.type QueueDriverOption func(*QueueDriver)
// WithQueueBatchSize bounds how many subjects one pass lists.func WithQueueBatchSize(size int) QueueDriverOption { return func(d *QueueDriver) { d.batchSize = size }}
// WithQueueBackoff sets the delay before a DEFERRED subject is offered to the// engine again, and the ceiling that delay grows to.//// It is per-subject rather than global: one community with dead credentials must// not slow down the queue for everyone else, which is exactly what a global// backoff would do.func WithQueueBackoff(base, max time.Duration) QueueDriverOption { return func(d *QueueDriver) { d.backoffBase, d.backoffMax = base, max }}
// QueueDriver walks the undecided backlog and feeds the engine.type QueueDriver struct { subjects PendingSubjectLister engine AdmissionProcessor now Clock batchSize int
backoffBase time.Duration backoffMax time.Duration
// deferrals holds the per-subject backoff state. In-memory on purpose: it is // a politeness hint, not state anything is allowed to depend on, so a // restart that forgets it costs one extra attempt per subject and nothing // else. // // It is keyed by (community, post) rather than by the whole PendingSubject // because the third field is a time.Time read back from Postgres, and Go // compares those by wall clock AND monotonic reading AND location. Two // reads of one unchanged row can therefore produce values that are equal to // a human and distinct to a map, which would silently defeat the backoff. deferrals map[subjectKey]deferral
// mu guards deferrals and snapshot. Snapshot is read by the health handler // on an HTTP goroutine while RunPass is writing on the job goroutine, so // this is a genuine race rather than a defensive one. mu sync.Mutex snapshot QueueSnapshot}
// subjectKey identifies one subject by the two fields that actually name it.type subjectKey struct { communityDID string postURI string}
func keyOf(subject PendingSubject) subjectKey { return subjectKey{communityDID: subject.CommunityDID, postURI: subject.PostURI}}
// deferral is how long one subject is held back, and until when.//// The delay is carried alongside the deadline so it can GROW: a subject that// defers repeatedly is one whose community is wedged, and re-offering it on a// fixed interval would keep a steady trickle of doomed requests pointed at a// PDS that is already failing.type deferral struct { until time.Time delay time.Duration}
// Default queue bounds, applied when the corresponding option is not given.//// The batch bound is not a nicety: this query runs on a timer against a table// that grows with every submission the instance has ever seen, so a driver// built without one must still not ask for the whole backlog.const ( defaultQueueBatchSize = 100 defaultQueueBackoffBase = time.Minute defaultQueueBackoffMax = 15 * time.Minute)
// NewQueueDriver wires the driver.func NewQueueDriver(subjects PendingSubjectLister, engine AdmissionProcessor, now Clock, opts ...QueueDriverOption) *QueueDriver { d := &QueueDriver{ subjects: subjects, engine: engine, now: now, batchSize: defaultQueueBatchSize, backoffBase: defaultQueueBackoffBase, backoffMax: defaultQueueBackoffMax, deferrals: make(map[subjectKey]deferral), } for _, opt := range opts { opt(d) } return d}
// RunPass processes one batch of the backlog.//// It returns an error ONLY when the backlog itself could not be read. A subject// the engine failed on is counted and the pass continues: one poisonous row// must not stop every other community's posts from being decided, which is what// an early return would do — and the row is still in the backlog next pass.func (d *QueueDriver) RunPass(ctx context.Context) (PassReport, error) { startedAt := d.now() report := PassReport{StartedAt: startedAt}
subjects, err := d.subjects.ListPendingSubjects(ctx, d.fetchSize()) if err != nil { // The one failure a pass has nothing to do about. Every other outcome // below is per-subject and counted; this one means there is no work // list to count against. return report, fmt.Errorf("listing the acceptance backlog: %w", err) } report.Listed = len(subjects)
for _, subject := range groupByCommunity(subjects) { // THE BATCH IS FILLED WITH WORK, not with rows. Skipping a backed-off // subject must not consume a slot, or a stuck prefix would spend the // whole pass on subjects it never touched — see fetchSize. if report.Processed >= d.batchSize { break } if d.heldBack(subject, startedAt) { continue }
outcome, err := d.engine.ProcessAdmission(ctx, subject.CommunityDID, subject.PostURI) report.Processed++
// The ERROR is checked before the outcome, because a failing engine // returns EngineDeferred alongside it and reading the outcome first // would file every failure as a deferral — collapsing the two numbers an // operator uses to decide whether this is credentials or a bug. // // They are COUNTED apart and BACKED OFF the same. The counters are what // an operator reads, and there the difference is the whole point: a pass // that defers everything is usually credentials and will clear, while a // pass that fails everything is a bug. The backoff answers a different // question — "how soon is it worth asking again" — and the honest answer // for a failure is at least as conservative as for a deferral. A wedged // community's PDS returns errors, not deferrals, so exempting failures // would leave the loudest case as the one thing nothing paced. switch { case errors.Is(err, ErrSubjectGone): // NOT A FAILURE, and counting it as one would make an ordinary race // look like an outage. The backlog query excludes tombstoned and // unindexed posts, but a post can be deleted between the listing and // the decision — so this is the queue meeting a subject that stopped // existing while it worked, which is exactly what the exclusion is // for and needs no operator's attention. It is counted as deferred // and backed off like any other "nothing to do yet"; the next pass // will not list it at all. report.Deferred++ d.deferSubject(subject, startedAt) case err != nil: report.Failed++ log.Printf("[ACCEPTANCE-QUEUE] Warning: %s in %s could not be settled: %v", subject.PostURI, subject.CommunityDID, err) d.deferSubject(subject, startedAt) case outcome == EngineDeferred: report.Deferred++ d.deferSubject(subject, startedAt) default: report.Settled++ d.clearDeferral(subject) } }
d.pruneDeferrals(subjects) d.record(subjects, report, startedAt) return report, nil}
// queueOverFetchFactor bounds how far past a backed-off prefix one pass may// reach: a pass never asks the query for more than batchSize × this.//// FOUR, and the number is a trade rather than a preference. The backlog is// ordered oldest-first, so the subjects most likely to be stuck are exactly the// ones at the front of it — a community whose credentials expired weeks ago sits// there forever. Without over-fetching, a pass asks for LIMIT rows, gets LIMIT// stuck ones, skips them all for backoff and does nothing; every pass, while a// healthy post two rows behind is never decided. Over-fetching without a bound// would instead let one pass drag the entire backlog into memory to find one// live subject. Four buys three batches of headroom against a query whose cost// grows with it, and a prefix deeper than that drains as backoffs expire.const queueOverFetchFactor = 4
// fetchSize is how many rows to ask for so the pass can still fill its batch// after skipping the subjects it already knows are held back.//// It asks for the batch plus exactly the number of deferrals currently held —// the measured size of the prefix that may be skipped — rather than always// over-fetching. A driver with nothing backed off has nothing to skip, so it// asks for precisely what it will use.func (d *QueueDriver) fetchSize() int { held := d.deferredCount() size := d.batchSize + held if ceiling := d.batchSize * queueOverFetchFactor; size > ceiling { size = ceiling } return size}
func (d *QueueDriver) deferredCount() int { d.mu.Lock() defer d.mu.Unlock() return len(d.deferrals)}
// pruneDeferrals forgets the backoffs of subjects that are no longer listed.//// A deferral outlives its subject otherwise. The row gets settled by somebody// else — the synchronous fast path, a firehose acceptance, a moderator's// removal — and simply stops being listed, without ever telling the driver. The// map is keyed by subject and swept by nothing, so on a busy instance it is an// unbounded leak held for the life of the process.//// It is also wrong on RE-ENTRY, which is the part a leak metric would not show:// a subject that leaves the backlog and comes back — an edit reopening an// accepted post — would arrive carrying a stale backoff it did nothing to earn,// and wait out a delay that was about a completely different decision.//// Pruning against the LISTING rather than against what the pass processed is// deliberate: a subject skipped for backoff is still in the backlog, and// forgetting it would defeat the backoff on the very next pass.func (d *QueueDriver) pruneDeferrals(listed []PendingSubject) { d.mu.Lock() defer d.mu.Unlock()
if len(d.deferrals) == 0 { return } stillListed := make(map[subjectKey]bool, len(listed)) for _, subject := range listed { stillListed[keyOf(subject)] = true } for key := range d.deferrals { if !stillListed[key] { delete(d.deferrals, key) } }}
// Snapshot returns the driver's health surface as of the last completed pass.func (d *QueueDriver) Snapshot() QueueSnapshot { d.mu.Lock() defer d.mu.Unlock() return d.snapshot}
// heldBack reports whether a subject's backoff has yet to elapse.//// It is judged against the instant the PASS began rather than against the clock// now, so one pass makes one decision about what is due. Reading the clock per// subject would let a long pass treat its first and last subjects by different// rules, and would make which subjects ran depend on how slow the engine was.func (d *QueueDriver) heldBack(subject PendingSubject, at time.Time) bool { d.mu.Lock() defer d.mu.Unlock()
held, ok := d.deferrals[keyOf(subject)] return ok && at.Before(held.until)}
// deferSubject holds a subject back, doubling its wait each consecutive time.func (d *QueueDriver) deferSubject(subject PendingSubject, at time.Time) { d.mu.Lock() defer d.mu.Unlock()
key := keyOf(subject) delay := d.backoffBase if previous, ok := d.deferrals[key]; ok && previous.delay > 0 { delay = previous.delay * 2 } if delay > d.backoffMax { delay = d.backoffMax } d.deferrals[key] = deferral{until: at.Add(delay), delay: delay}}
// clearDeferral forgets a settled subject, so the map tracks the backlog rather// than the history of everything the driver has ever seen.func (d *QueueDriver) clearDeferral(subject PendingSubject) { d.mu.Lock() defer d.mu.Unlock() delete(d.deferrals, keyOf(subject))}
// record publishes the pass's health surface.func (d *QueueDriver) record(subjects []PendingSubject, report PassReport, at time.Time) { d.mu.Lock() defer d.mu.Unlock()
snapshot := QueueSnapshot{ PendingBacklog: report.Listed, LastPassDeferred: report.Deferred, LastPassFailed: report.Failed, DeferredSubjects: len(d.deferrals), } // Taken as a MINIMUM rather than as subjects[0], even though the query // orders by age. The oldest entry's age is the queue's only early warning, // and deriving it from an ordering assumption would make it silently wrong // the first time anything reorders the list. for i, subject := range subjects { if i == 0 || subject.CreatedAt.Before(*snapshot.OldestPendingAt) { oldest := subject.CreatedAt snapshot.OldestPendingAt = &oldest } } passedAt := at snapshot.LastPassAt = &passedAt
d.snapshot = snapshot}
// groupByCommunity returns the subjects with each community's contiguous, and// with duplicates dropped.//// GROUPING. swapCommit is repo-global, so two writers on one community's repo// starve each other — task 4 recorded that before this driver existed. A single// goroutine satisfies it today whatever the order; what the grouping protects is// the NEXT version, where a worker pool has to shard on something, and the only// safe partition is the community. Output that interleaved communities would be// output with no partition in it.//// DEDUPLICATION. A subject appearing twice in one listing is what a racing edit// produces, and §8's edit-debounce is exactly this rule: a post 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.//// The order WITHIN a community is preserved, and so is the order BETWEEN them// (first appearance wins), so the query's oldest-first discipline survives.func groupByCommunity(subjects []PendingSubject) []PendingSubject { var order []string byCommunity := make(map[string][]PendingSubject) seen := make(map[subjectKey]bool, len(subjects))
for _, subject := range subjects { if seen[keyOf(subject)] { continue } seen[keyOf(subject)] = true
if _, known := byCommunity[subject.CommunityDID]; !known { order = append(order, subject.CommunityDID) } byCommunity[subject.CommunityDID] = append(byCommunity[subject.CommunityDID], subject) }
grouped := make([]PendingSubject, 0, len(seen)) for _, communityDID := range order { grouped = append(grouped, byCommunity[communityDID]...) } return grouped}