Something went wrong. Try again.
A collaborative coding-agent orchestrator for atproto radl.app
Something went wrong. Try again.
TypeScript
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169// Quarantine, and the want-list that makes evicting from it safe.//// An envelope whose signing key this observer has never seen is not wrong — a `device` record is// published to a public repo, and that repo may simply not have been polled yet. So it waits. But// waiting is unbounded memory unless it is bounded, and bounding it means eviction, and eviction on// the private path is not the harmless thing it is on the public one: **no PDS holds a re-fetchable// copy**. Dropping an envelope silently is dropping a record from this replica's history for good,// and the fork would be invisible — version summaries would agree, because neither side knows the// record was ever there.//// Hence the rule this file exists to enforce: *eviction is never silent loss*. Every eviction, and// every refusal for capacity, writes a durable want entry naming exactly// `(did, collection, rkey, recordCid)`, and catch-up re-requests wants EXPLICITLY rather than// inferring completeness from summaries. The cost of a burst is a re-fetch. It is never a permanent// divergence.//// What is NOT durable, stated so nobody has to infer it from the absence of a table: the held set// itself. A restart drops whatever was still waiting on a key, and writes no want for it. That is// covered, but by a different mechanism — a quarantined envelope was never admitted, so it is absent// from `summarize()`, and the next `catchUp` asks for it again as a matter of course. The want list// exists for the case that argument does NOT cover: an envelope this replica evicted while its peers// believe it landed, which no summary diff would ever surface. `PrivateSpaceIngestor`'s gossip inbox// has the same property for the same reason.
import type { PrivateEnvelope } from './envelope.js'import { encode } from './cbor.js'import { envelopeKey } from './envelope.js'import { wantFor, wantKey, type WantEntry } from './admission.js'
/** * Records this observer knows it is missing. Durable — a want that only lived in memory would be * lost by the restart that most often follows the burst that caused it. */export interface WantList { add(want: WantEntry): void /** Called when the record finally lands, from catch-up or from any peer's gossip. */ clear(want: WantEntry): void all(): WantEntry[]}
export class MemoryWantList implements WantList { readonly #wants = new Map<string, WantEntry>()
add(want: WantEntry): void { this.#wants.set(wantKey(want), { ...want }) }
clear(want: WantEntry): void { this.#wants.delete(wantKey(want)) }
all(): WantEntry[] { return [...this.#wants.values()].sort((left, right) => wantKey(left) < wantKey(right) ? -1 : wantKey(left) > wantKey(right) ? 1 : 0, ) }}
export interface QuarantineLimits { maxEnvelopes: number maxBytes: number}
export const DEFAULT_QUARANTINE_LIMITS: QuarantineLimits = { maxEnvelopes: 1000, maxBytes: 8 * 1024 * 1024,}
interface Held { envelope: PrivateEnvelope bytes: number /** Insertion order, so eviction is oldest-first and deterministic. */ sequence: number}
/** * A bounded holding area for envelopes waiting on a device key. * * Oldest-first eviction, by both count and bytes. Deterministic on purpose: two replicas fed the * same burst evict the same envelopes and write the same wants, so a test can assert the recovery * rather than hope for it. */export class Quarantine { readonly #held = new Map<string, Held>() readonly #limits: QuarantineLimits readonly #wants: WantList #sequence = 0 #bytes = 0
constructor(wants: WantList, limits: QuarantineLimits = DEFAULT_QUARANTINE_LIMITS) { this.#wants = wants this.#limits = limits }
get size(): number { return this.#held.size }
get bytes(): number { return this.#bytes }
/** * Hold an envelope, evicting older ones if it does not fit. Returns what was evicted — every * eviction has already been written to the want list, so a caller that ignores the return value * still cannot lose the record. */ hold(envelope: PrivateEnvelope): WantEntry[] { const key = envelopeKey(envelope) if (this.#held.has(key)) return [] // The whole ENVELOPE, not just its record: what is retained here is the envelope, and measuring // the record alone let a peer pin gigabytes behind a 207-byte accounting by padding the fields // around it. `admit()` caps those fields too, so both halves of the bound are real. const bytes = encode(envelope).length // An envelope that could never fit is refused outright rather than emptying the queue for a // record it would then evict anyway — but it still leaves a want, because it is still a record // this replica does not have. if (bytes > this.#limits.maxBytes) { const want = wantFor(envelope) this.#wants.add(want) return [want] } this.#sequence += 1 this.#held.set(key, { envelope, bytes, sequence: this.#sequence }) this.#bytes += bytes const evicted: WantEntry[] = [] while ( this.#held.size > this.#limits.maxEnvelopes || this.#bytes > this.#limits.maxBytes ) { const oldest = [...this.#held.entries()].sort( ([, left], [, right]) => left.sequence - right.sequence, )[0] if (!oldest) break const [oldestKey, held] = oldest this.#held.delete(oldestKey) this.#bytes -= held.bytes const want = wantFor(held.envelope) this.#wants.add(want) evicted.push(want) } return evicted }
/** * Take out everything signed by a device that has just become known, so the caller can re-run * admission over it. Draining is by `(did, deviceKeyId)` because that is exactly what changed. */ drain(did: string, deviceKeyId: string): PrivateEnvelope[] { const drained: PrivateEnvelope[] = [] for (const [key, held] of [...this.#held.entries()].sort( ([, left], [, right]) => left.sequence - right.sequence, )) { if (held.envelope.did !== did || held.envelope.deviceKeyId !== deviceKeyId) continue this.#held.delete(key) this.#bytes -= held.bytes drained.push(held.envelope) } return drained }
/** Everything still waiting, oldest first — for diagnostics, so residual state is visible. */ waiting(): PrivateEnvelope[] { return [...this.#held.values()] .sort((left, right) => left.sequence - right.sequence) .map((held) => held.envelope) }}