| … | … | @@ -0,0 +1,77 @@ |
| 1 | // FirehoseSequencer — the atproto firehose as a Hypercore. | |
| 2 | // | |
| 3 | // Each subscribeRepos message (a framed dag-cbor #commit/#identity/#account) is one | |
| 4 | // append to a single-writer Hypercore; the block index + 1 IS the atproto `seq` | |
| 5 | // (1-based, monotonic). `subscribe(cursor)` replays persisted frames with seq > cursor | |
| 6 | // then live-tails — so a freshly-attached relay gets full history from cursor=0 (the | |
| 7 | // fix for the non-archival-relay root cause, POSTMORTEM.md §Federation). | |
| 8 | // | |
| 9 | // This is both the atproto sequenced-firehose model and the P2P-native thesis: the | |
| 10 | // firehose is not a side-channel, it's a replicable append-only log. | |
| 11 | ||
| 12 | import { commitFrame, identityFrame, accountFrame } from './firehose.js' | |
| 13 | ||
| 14 | export class FirehoseSequencer { | |
| 15 | constructor (core) { this.core = core } // a Hypercore, binary values | |
| 16 | ||
| 17 | async ready () { await this.core.ready(); return this } | |
| 18 | get length () { return this.core.length } // == the latest seq | |
| 19 | async close () { await this.core.close() } | |
| 20 | ||
| 21 | // Append a pre-framed message; returns its 1-based seq. | |
| 22 | async append (frameBytes) { | |
| 23 | const seq = this.core.length + 1 | |
| 24 | await this.core.append(frameBytes) | |
| 25 | return seq | |
| 26 | } | |
| 27 | ||
| 28 | // Sequence a #commit: assign the seq FIRST (frame body carries it), then append. | |
| 29 | async sequenceCommit (write, { time }) { | |
| 30 | const seq = this.core.length + 1 | |
| 31 | const frame = await commitFrame(write, { seq, time }) | |
| 32 | await this.core.append(frame) | |
| 33 | return { seq, frame } | |
| 34 | } | |
| 35 | ||
| 36 | async sequenceIdentity ({ did, handle, time }) { | |
| 37 | const seq = this.core.length + 1 | |
| 38 | const frame = identityFrame({ seq, did, handle, time }) | |
| 39 | await this.core.append(frame) | |
| 40 | return { seq, frame } | |
| 41 | } | |
| 42 | ||
| 43 | async sequenceAccount ({ did, active = true, status, time }) { | |
| 44 | const seq = this.core.length + 1 | |
| 45 | const frame = accountFrame({ seq, did, active, status, time }) | |
| 46 | await this.core.append(frame) | |
| 47 | return { seq, frame } | |
| 48 | } | |
| 49 | ||
| 50 | // Async iterator over { seq, bytes }: backfill frames with seq > cursor, then | |
| 51 | // live-tail new appends. cursor is the last seq the consumer holds (0/absent = | |
| 52 | // from the start). Pass an AbortSignal to stop (WS disconnect). | |
| 53 | async * subscribe (cursor = 0, { signal } = {}) { | |
| 54 | let next = Math.max(0, cursor | 0) // seq > cursor <=> index >= cursor | |
| 55 | while (true) { | |
| 56 | if (signal && signal.aborted) return | |
| 57 | if (next < this.core.length) { | |
| 58 | const bytes = await this.core.get(next) | |
| 59 | yield { seq: next + 1, bytes } | |
| 60 | next++ | |
| 61 | continue | |
| 62 | } | |
| 63 | // caught up — wait for the next append (or abort) | |
| 64 | await new Promise((resolve) => { | |
| 65 | const onAppend = () => { cleanup(); resolve() } | |
| 66 | const onAbort = () => { cleanup(); resolve() } | |
| 67 | const cleanup = () => { | |
| 68 | this.core.off('append', onAppend) | |
| 69 | if (signal) signal.removeEventListener('abort', onAbort) | |
| 70 | } | |
| 71 | this.core.on('append', onAppend) | |
| 72 | if (signal) signal.addEventListener('abort', onAbort, { once: true }) | |
| 73 | }) | |
| 74 | if (signal && signal.aborted) return | |
| 75 | } | |
| 76 | } | |
| 77 | } |
| … | … | @@ -0,0 +1,77 @@ |
| 1 | // FirehoseSequencer: seq assignment, cursor=0 full backfill, cursor>0 partial, and | |
| 2 | // live-tail (a frame appended AFTER subscription arrives) + clean abort. | |
| 3 | ||
| 4 | import test from 'brittle' | |
| 5 | import { mkdtempSync, rmSync } from 'node:fs' | |
| 6 | import { join } from 'node:path' | |
| 7 | import { tmpdir } from 'node:os' | |
| 8 | import { RepoStore } from '../lib/repo-store.js' | |
| 9 | import { FirehoseSequencer } from '../lib/sequencer.js' | |
| 10 | ||
| 11 | async function newSeq (t) { | |
| 12 | const dir = mkdtempSync(join(tmpdir(), 'seq-')) | |
| 13 | const store = await new RepoStore(dir).ready() | |
| 14 | const core = store.corestore.get({ name: 'firehose' }) | |
| 15 | await core.ready() | |
| 16 | const seq = await new FirehoseSequencer(core).ready() | |
| 17 | t.teardown(async () => { await seq.close(); await store.close(); rmSync(dir, { recursive: true, force: true }) }) | |
| 18 | return seq | |
| 19 | } | |
| 20 | ||
| 21 | const F = (n) => new Uint8Array([n]) // a tiny distinguishable "frame" | |
| 22 | ||
| 23 | // 1. seq is 1-based and monotonic | |
| 24 | test('append returns 1-based monotonic seq', async (t) => { | |
| 25 | const s = await newSeq(t) | |
| 26 | const a = await s.append(F(10)) | |
| 27 | const b = await s.append(F(20)) | |
| 28 | const c = await s.append(F(30)) | |
| 29 | t.ok(a === 1 && b === 2 && c === 3 && s.length === 3, 'append returns 1-based monotonic seq') | |
| 30 | }) | |
| 31 | ||
| 32 | // 2. cursor=0 backfills the full history, in order | |
| 33 | test('cursor=0 replays all frames in seq order', async (t) => { | |
| 34 | const s = await newSeq(t) | |
| 35 | for (let i = 1; i <= 4; i++) await s.append(F(i)) | |
| 36 | const got = [] | |
| 37 | const ac = new AbortController() | |
| 38 | for await (const { seq, bytes } of s.subscribe(0, { signal: ac.signal })) { | |
| 39 | got.push([seq, bytes[0]]) | |
| 40 | if (got.length === 4) { ac.abort(); break } | |
| 41 | } | |
| 42 | t.ok(JSON.stringify(got) === JSON.stringify([[1, 1], [2, 2], [3, 3], [4, 4]]), | |
| 43 | 'cursor=0 replays all frames in seq order') | |
| 44 | }) | |
| 45 | ||
| 46 | // 3. cursor>0 backfills only frames with seq > cursor | |
| 47 | test('cursor=2 replays only seq 3,4', async (t) => { | |
| 48 | const s = await newSeq(t) | |
| 49 | for (let i = 1; i <= 4; i++) await s.append(F(i)) | |
| 50 | const got = [] | |
| 51 | const ac = new AbortController() | |
| 52 | for await (const { seq } of s.subscribe(2, { signal: ac.signal })) { | |
| 53 | got.push(seq) | |
| 54 | if (got.length === 2) { ac.abort(); break } | |
| 55 | } | |
| 56 | t.ok(JSON.stringify(got) === JSON.stringify([3, 4]), 'cursor=2 replays only seq 3,4') | |
| 57 | }) | |
| 58 | ||
| 59 | // 4. live-tail: a frame appended AFTER subscribe() started is delivered | |
| 60 | test('live frame appended after subscribe is delivered', async (t) => { | |
| 61 | const s = await newSeq(t) | |
| 62 | await s.append(F(1)) | |
| 63 | const ac = new AbortController() | |
| 64 | const got = [] | |
| 65 | const consumer = (async () => { | |
| 66 | for await (const { seq, bytes } of s.subscribe(0, { signal: ac.signal })) { | |
| 67 | got.push([seq, bytes[0]]) | |
| 68 | if (got.length === 2) { ac.abort(); break } | |
| 69 | } | |
| 70 | })() | |
| 71 | // give the consumer a tick to drain backfill and park on 'append', then append live | |
| 72 | await new Promise(r => setTimeout(r, 50)) | |
| 73 | await s.append(F(2)) | |
| 74 | await consumer | |
| 75 | t.ok(JSON.stringify(got) === JSON.stringify([[1, 1], [2, 2]]), | |
| 76 | 'live frame appended after subscribe is delivered') | |
| 77 | }) |