29d1bb3 · atproto-store

Changed files (2) +38 -3

lib/firehose.js +15 -2

…	…	@@ -7,12 +7,14 @@
7	7
8	8	import * as dagCbor from '@ipld/dag-cbor'
9	9	import { writeCar } from 'atproto-repo/lib/repo.js'
	10	import { commitCid } from 'atproto-repo/lib/commit.js'
10	11
11	12	// subscribeRepos frame = dag-cbor(header {t, op:1}) ++ dag-cbor(body). Each header is
12	13	// a constant byte prefix (dag-cbor canonicalises map order), so we emit/detect by it.
13	14	export const COMMIT_HEADER = dagCbor.encode({ t: '#commit', op: 1 })
14	15	export const IDENTITY_HEADER = dagCbor.encode({ t: '#identity', op: 1 })
15	16	export const ACCOUNT_HEADER = dagCbor.encode({ t: '#account', op: 1 })
	17	export const SYNC_HEADER = dagCbor.encode({ t: '#sync', op: 1 })
16	18
17	19	function frame (header, body) {
18	20	const bodyBytes = dagCbor.encode(body)
…	…	@@ -56,7 +58,18 @@ export function accountFrame ({ seq, did, active = true, status, time }) {
56	58	return frame(ACCOUNT_HEADER, body)
57	59	}
58	60
59		const HEADERS = [['#commit', COMMIT_HEADER], ['#identity', IDENTITY_HEADER], ['#account', ACCOUNT_HEADER]]
	61	// #sync (Sync 1.1) — tells a relay to RE-BASELINE the repo to `signedCommit` (re-anchor via
	62	// getRepo) instead of applying incremental deltas. This is the sanctioned signal for a repo
	63	// whose commit chain is discontinuous vs the relay's last-known state (e.g. a custodial→
	64	// self-custody migration that re-authored the repo). `blocks` is a CAR carrying ONLY the
	65	// commit block (root = its CID), ≤10KB. `rev`/`did` MUST match the commit object.
	66	export async function syncFrame ({ seq, did, signedCommit, time }) {
	67	const cid = commitCid(signedCommit)
	68	const blocks = await writeCar(cid, [{ cid, bytes: dagCbor.encode(signedCommit) }])
	69	return frame(SYNC_HEADER, { seq, did, blocks, rev: signedCommit.rev, time })
	70	}
	71
	72	const HEADERS = [['#commit', COMMIT_HEADER], ['#identity', IDENTITY_HEADER], ['#account', ACCOUNT_HEADER], ['#sync', SYNC_HEADER]]
60	73
61	74	// Re-stamp a frame's `seq` (relay egress): a relay aggregates many PDS firehoses and MUST
62	75	// present its own monotonic seq in each frame (the cursor contract). Decodes, replaces
…	…	@@ -64,7 +77,7 @@ const HEADERS = [['#commit', COMMIT_HEADER], ['#identity', IDENTITY_HEADER], ['#
64	77	export function restampFrame (frameBytes, seq) {
65	78	const d = decodeFrame(frameBytes)
66	79	if (!d) return frameBytes
67		const header = d.t === '#commit' ? COMMIT_HEADER : d.t === '#identity' ? IDENTITY_HEADER : ACCOUNT_HEADER
	80	const header = d.t === '#commit' ? COMMIT_HEADER : d.t === '#identity' ? IDENTITY_HEADER : d.t === '#account' ? ACCOUNT_HEADER : SYNC_HEADER
68	81	d.body.seq = seq
69	82	const bodyBytes = dagCbor.encode(d.body)
70	83	const out = new Uint8Array(header.length + bodyBytes.length)

test/firehose.test.js +23 -1

…	…	@@ -8,9 +8,10 @@ import { join } from 'node:path'
8	8	import { tmpdir } from 'node:os'
9	9	import { RepoStore } from '../lib/repo-store.js'
10	10	import { RepoManager } from '../lib/repo-manager.js'
11		import { commitFrame, decodeCommitFrame, COMMIT_HEADER } from '../lib/firehose.js'
	11	import { commitFrame, decodeCommitFrame, decodeFrame, syncFrame, restampFrame, COMMIT_HEADER, SYNC_HEADER } from '../lib/firehose.js'
12	12	import { isValidTid } from '../lib/tid.js'
13	13	import { readCar, loadCommit } from 'atproto-repo/lib/repo.js'
	14	import { commitCid } from 'atproto-repo/lib/commit.js'
14	15	import { getRecord } from '../lib/read-view.js'
15	16
16	17	const DID = 'did:plc:yu4kgcktm7qtrwo5ofyiaf4c'
…	…	@@ -79,6 +80,27 @@ test('delete: op cid null, rev advances, read-view reflects removal', async (t)
79	80	'delete: op cid null, rev advances, read-view reflects removal')
80	81	})
81	82
	83	// 4b. #sync frame (Sync 1.1 re-baseline): header byte-exact, body fields, CAR carries the
	84	// commit block (root = commit CID), rev matches, and restampFrame updates its seq.
	85	test('#sync frame: header + body + CAR-with-commit + restamp', async (t) => {
	86	const { mgr } = await newMgr(t)
	87	const w = await mgr.applyWrite({ action: 'create', collection: 'app.bsky.feed.post', rkey: '3sss', record: { $type: 'app.bsky.feed.post', text: 's', createdAt: iso } })
	88	const signed = w.signedCommit
	89	const frame = await syncFrame({ seq: 7, did: DID, signedCommit: signed, time: iso })
	90	const headerOk = Buffer.from(frame.subarray(0, SYNC_HEADER.length)).equals(Buffer.from(SYNC_HEADER))
	91	const dec = decodeFrame(frame)
	92	const car = await readCar(dec.body.blocks)
	93	const commit = loadCommit(car)
	94	const cid = commitCid(signed)
	95	const restamped = decodeFrame(restampFrame(frame, 99))
	96	t.ok(
	97	headerOk && dec.t === '#sync' &&
	98	dec.body.did === DID && dec.body.seq === 7 && dec.body.rev === signed.rev &&
	99	car.roots[0].equals(cid) && commit.value.rev === signed.rev && commit.value.did === DID &&
	100	restamped.body.seq === 99,
	101	'#sync frame: header + body + CAR-with-commit + restamp')
	102	})
	103
82	104	// The live subscribeRepos stream leaves a refed TLSSocket that ws.close() doesn't
83	105	// release; brittle ends the run on `beforeExit` (loop drain), so unref it or the
84	106	// runner hangs after all asserts pass. (This is why the pre-brittle version force-exited.)

atproto-store commits / 29d1bb3

Changed files (2) +38 -3