// link drain — tier routing, the rev skip-test, fail-soft accounting, concurrency bound.
// Run: node test/link-drain.test.mjs
//
// Every pipeline step is injected, so §7.6's routing rules are tested without a network or a store.
// The assertions that matter most are the NEGATIVE ones: a Tier-C link must cost zero fetches, and an
// unverified record must never reach retain. Both are silent-by-nature in production.
import { LinkDrain, pointerKey, shouldSkipRev } from '../lib/link-drain.js'
import { SOURCES } from '../lib/spacedust.js'
let pass = 0
const ok = (c, m) => { if (!c) { console.error('FAIL:', m); process.exit(1) } console.log(' ok -', m); pass++ }
const ptr = (over = {}) => ({
operation: 'create',
source: SOURCES.REPLY_PARENT,
subject: 'at://did:plc:wj62/app.bsky.feed.post/3kxyz',
did: 'did:plc:replier',
collection: 'app.bsky.feed.post',
rkey: '3kdef',
rev: '3lbbb',
contentBearing: true,
origin: 'live',
...over
})
// A client stub — the drain only needs .drain(n).
const stubClient = (batches = []) => ({
drain (n) { return batches.length ? batches.shift().slice(0, n) : [] },
pending: 0
})
// ── pure helpers ─────────────────────────────────────────────────────────────────────────────────
ok(pointerKey(ptr()) === 'did:plc:replier/app.bsky.feed.post/3kdef',
'dedup is by RECORD, not by link — one record legitimately emits several link events (reply.parent AND reply.root)')
ok(shouldSkipRev('3laaa', '3lbbb') === true, 'an older rev is already-known state')
ok(shouldSkipRev('3lccc', '3lbbb') === false, 'a newer rev is new state')
ok(shouldSkipRev('3lbbb', '3lbbb') === true, 'the same rev is already held')
ok(shouldSkipRev(null, '3lbbb') === false, 'a MISSING incoming rev is treated as new — skipping on absent data would silently lose writes')
ok(shouldSkipRev('3lccc', null) === false, 'nothing known yet means nothing to skip')
// ── tier routing: Tier C must cost nothing ───────────────────────────────────────────────────────
{
let fetches = 0
const d = new LinkDrain(stubClient(), { fetchRecord: async () => { fetches++; return null } })
await d._handleOne(ptr({ contentBearing: false, source: SOURCES.LIKE }))
ok(fetches === 0, 'a LIKE triggers ZERO fetches — this is §7.6\'s pressure valve; a 50k-like post must not become 50k fetches')
ok(d.stats.skippedTierC === 1 && d.stats.retained === 0, 'and it is counted as a Tier-C skip, not silently ignored')
}
// ── deletes evict rather than fetch ──────────────────────────────────────────────────────────────
{
let fetches = 0, evicted = null
const d = new LinkDrain(stubClient(), {
fetchRecord: async () => { fetches++; return null },
evictRecord: async (p) => { evicted = p }
})
await d._handleOne(ptr({ operation: 'delete' }))
ok(fetches === 0 && evicted !== null, 'a delete link evicts; fetching a record that was just deleted would be pointless')
ok(d.stats.deleted === 1, 'and is counted')
}
// ── the rev skip-test short-circuits before the network ──────────────────────────────────────────
{
let fetches = 0
const d = new LinkDrain(stubClient(), {
knownRev: async () => '3lzzz',
fetchRecord: async () => { fetches++; return null }
})
await d._handleOne(ptr({ rev: '3laaa' }))
ok(fetches === 0 && d.stats.skippedRev === 1, 'state we already hold costs no fetch — the skip happens BEFORE the network, not after')
}
// ── verification gates retention ─────────────────────────────────────────────────────────────────
{
let retained = 0
const d = new LinkDrain(stubClient(), {
fetchRecord: async () => ({ bytes: new Uint8Array([1]) }),
verifyRecord: async () => false,
retainRecord: async () => { retained++ }
})
await d._handleOne(ptr())
ok(retained === 0, 'an UNVERIFIED record is never retained — this is the whole agency claim, not a nicety')
ok(d.stats.verifyFailed === 1 && d.stats.retained === 0, 'and the failure is counted')
}
{
let retained = 0
const d = new LinkDrain(stubClient(), {
fetchRecord: async () => null,
verifyRecord: async () => { throw new Error('should not be called') },
retainRecord: async () => { retained++ }
})
await d._handleOne(ptr())
ok(retained === 0 && d.stats.fetchFailed === 1, 'a failed fetch stops the pipeline before verify')
}
{
let got = null
const d = new LinkDrain(stubClient(), {
fetchRecord: async () => ({ bytes: new Uint8Array([1, 2]) }),
verifyRecord: async () => true,
retainRecord: async (rec, p) => { got = { rec, p } }
})
await d._handleOne(ptr())
ok(got !== null && got.p.rkey === '3kdef' && d.stats.retained === 1, 'the happy path retains, with the pointer alongside the record')
}
// ── fail soft: one bad pointer must not stop the drain ────────────────────────────────────────────
{
const d = new LinkDrain(stubClient(), {
fetchRecord: async (p) => { if (p.rkey === 'boom') throw new Error('kaboom'); return { bytes: new Uint8Array([1]) } },
verifyRecord: async () => true,
retainRecord: async () => {}
})
await d._handleBatch([ptr({ rkey: 'boom' }), ptr({ rkey: 'fine' })])
ok(d.stats.errored === 1 && d.stats.retained === 1,
'a throwing pointer is counted and the REST of the batch still completes — losing the drain is unrecoverable, losing one pointer is not')
}
// ── concurrency is bounded ───────────────────────────────────────────────────────────────────────
{
let inFlight = 0, peak = 0
const d = new LinkDrain(stubClient(), {
concurrency: 3,
fetchRecord: async () => {
inFlight++; peak = Math.max(peak, inFlight)
await new Promise((r) => setTimeout(r, 5))
inFlight--
return { bytes: new Uint8Array([1]) }
},
verifyRecord: async () => true,
retainRecord: async () => {}
})
await d._handleBatch(Array.from({ length: 12 }, (_, i) => ptr({ rkey: `rk${i}` })))
ok(peak <= 3, `fetch concurrency is capped (peak ${peak} <= 3) — a burst must not monopolise the event loop`)
ok(d.stats.retained === 12, 'and every pointer in the batch is still handled')
}
// ── the loop yields and can be stopped ───────────────────────────────────────────────────────────
{
const batch = [ptr({ rkey: 'a' }), ptr({ rkey: 'b' })]
const d = new LinkDrain(stubClient([batch]), {
idleMs: 1,
fetchRecord: async () => ({ bytes: new Uint8Array([1]) }),
verifyRecord: async () => true,
retainRecord: async () => {}
})
d.start()
await new Promise((r) => setTimeout(r, 40))
await d.stop()
ok(d.stats.retained === 2, 'the running loop drains what the client offers')
ok(d.stats.seen === 2, 'and accounts for everything it saw')
}
// ── defaults are safe ────────────────────────────────────────────────────────────────────────────
{
const d = new LinkDrain(stubClient(), {})
await d._handleOne(ptr())
ok(d.stats.fetchFailed === 1 && d.stats.retained === 0,
'with no hooks wired the drain retains NOTHING rather than defaulting to trusting the index')
}
console.log(`\nLINK-DRAIN OK: ${pass} assertions`)