// link-drain.js — consume Spacedust pointers and run DESIGN §7.6's retention pipeline.
//
// WHERE THIS SITS. `spacedust.js` owns the socket and does exactly one thing per message: parse and
// enqueue. This module owns everything after that — tier routing, dedup, fetch, verify, retain. The
// split exists because spacedust DROPS a subscriber that lags, so the socket handler must never do work.
//
// WHY NOT A THREAD (asked and answered 2026-07-30). `bare-thread` runs an entry point on a dedicated OS
// thread with its module graph bundled AHEAD OF TIME, in isolation — its API is join/suspend/wakeup/
// resume/terminate, with no message passing (that is `bare-channel`). The tail of this pipeline WRITES:
// OverlayRepoStore file handles plus the appview Hyperbee over a Corestore, and a Corestore holds an
// EXCLUSIVE RocksDB lock. Store handles are not transferable across threads — only serializable messages
// are — so putting the drain on a thread would need a second store opened over the same directory, which
// is precisely the lock conflict the stack already documents. The lag hazard is event-loop OCCUPANCY,
// not instruction count, so the fix is to YIELD, not to parallelise.
//
// THE THREAD SEAM, IF A PROFILE EVER JUSTIFIES ONE. Only `verifyRecord` is a candidate: it is pure
// (bytes in, boolean out), CPU-bound (signature + hash), and touches no store. It could move behind a
// `bare-channel` to a `bare-thread` worker without changing anything else here. Everything that touches
// a store must stay on this thread. Do not move it without a measurement — see the standing rule about
// unmeasured numbers.
//
// FAILS SOFT, ALWAYS. A pointer that cannot be fetched or verified is counted and dropped, never thrown.
// Losing one pointer is recoverable: §7.6 pairs the live subscription with Constellation precisely so a
// gap can be backfilled by (subject, source). Losing the drain loop is not recoverable.
//
// PURE + INJECTED. Every pipeline step is a parameter, so tier routing, dedup, the rev skip-test, the
// yield cadence and the failure accounting are all unit-testable with no network, no store and no clock.
const DEFAULTS = {
concurrency: 4, // parallel fetch+verify; bounded so a burst cannot monopolise the loop
batchSize: 16, // pointers handled between explicit yields
idleMs: 250, // poll interval when the queue is empty
maxRetained: null // optional cap; null = caller enforces the §7.6 byte budget
}
// Yield the macrotask queue. `await` on I/O already yields, but a run of cache hits or synchronous
// verifies would not — this guarantees the loop cannot be occupied for a whole batch.
function yieldToLoop () {
return new Promise((resolve) => setTimeout(resolve, 0))
}
// Stable identity for a linking record. Dedup is by record, not by link, because the same record can
// legitimately produce several link events (a post links to reply.parent AND reply.root).
export function pointerKey (p) {
return `${p.did}/${p.collection}/${p.rkey}`
}
// The backfilling guide's skip test: a pointer whose rev is not newer than what we already hold for that
// record is already-known state. Revs are TIDs — lexicographically sortable, so string compare is right.
// Unknown/absent revs are treated as NEW (fetch and let verification decide) rather than skipped, because
// skipping on missing data would silently lose writes.
export function shouldSkipRev (incomingRev, knownRev) {
if (typeof incomingRev !== 'string' || typeof knownRev !== 'string') return false
return incomingRev <= knownRev
}
export class LinkDrain {
// `client` is a SpacedustClient (anything with .drain(n) and .pending).
//
// Pipeline hooks — all optional, all async, all fail-soft:
// fetchRecord({did, collection, rkey}) -> { bytes, cid } | null (resolveMiniDoc -> getRecord)
// verifyRecord(record, pointer) -> boolean (signature + MST; the thread seam)
// retainRecord(record, pointer) -> void (OverlayRepoStore + appview index)
// evictRecord(pointer) -> void (a 'delete' link arrived)
// knownRev(pointer) -> string | null (rev we already hold, for the skip)
constructor (client, opts = {}) {
this.client = client
this.opts = { ...DEFAULTS, ...opts }
this._fetchRecord = opts.fetchRecord || (async () => null)
this._verifyRecord = opts.verifyRecord || (async () => true)
this._retainRecord = opts.retainRecord || (async () => {})
this._evictRecord = opts.evictRecord || (async () => {})
this._knownRev = opts.knownRev || (async () => null)
this._running = false
this._loop = null
this.stats = {
seen: 0,
skippedTierC: 0, // count-bearing (likes/reposts) — reference only, never retained
skippedRev: 0, // already have this state
deleted: 0,
fetchFailed: 0,
verifyFailed: 0,
retained: 0,
errored: 0
}
}
start () {
if (this._running) return this
this._running = true
this._loop = this._run()
return this
}
async stop () {
this._running = false
if (this._loop !== null) {
try {
await this._loop
} catch {
// the loop never rejects, but do not let teardown throw
}
this._loop = null
}
}
async _run () {
while (this._running) {
const batch = this.client.drain(this.opts.batchSize)
if (batch.length === 0) {
await new Promise((resolve) => setTimeout(resolve, this.opts.idleMs))
continue
}
await this._handleBatch(batch)
await yieldToLoop() // never hold the loop across batches
}
}
// Bounded-concurrency pool. Deliberately simple: N workers pulling from one array beats a scheduler,
// and the bound is what stops a burst from monopolising the event loop.
async _handleBatch (batch) {
const queue = batch.slice()
const workers = []
const width = Math.min(this.opts.concurrency, queue.length)
for (let i = 0; i < width; i++) {
workers.push((async () => {
while (queue.length > 0) {
const pointer = queue.shift()
try {
await this._handleOne(pointer)
} catch {
this.stats.errored++ // fail soft — one bad pointer must not stop the drain
}
}
})())
}
await Promise.all(workers)
}
async _handleOne (pointer) {
this.stats.seen++
// TIER ROUTING (§7.6). Count-bearing links — likes, reposts, follows — are never retained. This is
// the pressure valve: a post with 50k likes costs us nothing here, and "how many" is answered on
// demand by Constellation's getBacklinksCount. Only content-bearing links proceed.
if (!pointer.contentBearing) {
this.stats.skippedTierC++
return
}
if (pointer.operation === 'delete') {
this.stats.deleted++
await this._evictRecord(pointer)
return
}
// Skip state we already hold.
const known = await this._knownRev(pointer)
if (shouldSkipRev(pointer.rev, known)) {
this.stats.skippedRev++
return
}
// FETCH from the authoritative PDS. Spacedust ships only a locator by design, so the bytes come from
// the record's own repo — never from the index that told us about it.
const record = await this._fetchRecord(pointer)
if (!record) {
this.stats.fetchFailed++
return
}
// VERIFY before retaining. This is the whole agency claim — replicate signed data, verify locally,
// rather than trusting a third-party index. An unverified record is not retained, full stop.
// (This call is the thread seam described in the header.)
const ok = await this._verifyRecord(record, pointer)
if (!ok) {
this.stats.verifyFailed++
return
}
await this._retainRecord(record, pointer)
this.stats.retained++
}
}