ec478c3 · atproto-store

Changed files (4) +163 -1

lib/blob-pointer-store.js +34 -0

…	…	@@ -91,6 +91,40 @@ export class BlobPointerStore {
91	91	return !!(await this._pointer(cid))
92	92	}
93	93
	94	// MIGRATION: publish a shared pointer for every blob this node already holds but has never announced.
	95	// The counterpart of PointerSubstrate.backfill for records, and the reason blobs uploaded BEFORE this
	96	// module existed are not stranded: their bytes and metadata are already here, only the announcement is
	97	// missing. Idempotent and convergence-safe — SKIPS any cid that already has a pointer (a co-owner may
	98	// have published a newer one; never clobber it) and never deletes.
	99	//
	100	// WRITABLE-ONLY: sb.set appends to the autobase, so the caller must gate on base.writable.
	101	// `unpublishable` counts blobs with no local oid — nothing to point at; those need reindexOverlay()
	102	// first to re-register the file and recover an oid.
	103	async backfill () {
	104	if (typeof this.blobs.entries !== 'function') {
	105	throw new Error('BlobPointerStore.backfill: the blob store must enumerate entries() to migrate blobs → pointers')
	106	}
	107	let written = 0
	108	let skipped = 0
	109	let unpublishable = 0
	110	let total = 0
	111	for await (const e of this.blobs.entries()) {
	112	total++
	113	if (!e.oid) { unpublishable++; continue }
	114	const key = BlobPointerStore.key(e.cid)
	115	if ((await this.sb.get(key)) != null) { skipped++; continue }
	116	await this.sb.set(key, { $oid: e.oid, size: e.size, mimeType: e.mimeType })
	117	written++
	118	}
	119	return { written, skipped, unpublishable, total }
	120	}
	121
	122	// Pass through to the byte layer's repair (re-register blob files so peers can fetch them by oid).
	123	async reindexOverlay () {
	124	if (typeof this.blobs.reindexOverlay !== 'function') return { skipped: 'not-supported' }
	125	return this.blobs.reindexOverlay()
	126	}
	127
94	128	async _pointer (cidStr) {
95	129	try { return await this.sb.get(BlobPointerStore.key(cidStr)) } catch { return null }
96	130	}

lib/overlay-blob-store.js +53 -0

…	…	@@ -99,6 +99,59 @@ export class OverlayBlobStore {
99	99
100	100	async has (cidStr) { return !!(await this.bee.get(cidStr)) }
101	101
	102	// Everything this node knows locally: { cid, mimeType, size, oid }. The input to pointer backfill.
	103	async * entries () {
	104	for await (const node of this.bee.createReadStream()) {
	105	const cid = typeof node.key === 'string' ? node.key : b4a.toString(node.key, 'utf-8')
	106	let meta
	107	try { meta = JSON.parse(b4a.toString(node.value, 'utf-8')) } catch { continue }
	108	yield { cid, mimeType: meta.mimeType, size: meta.size, oid: meta.oid \|\| null }
	109	}
	110	}
	111
	112	// EXPLICIT REPAIR — the blob counterpart of OverlayRepoStore.reindexOverlay, for a store whose overlay
	113	// INDEX is missing or stale. That method walks records/ ONLY and never touches blobs, so a repaired
	114	// node would resume serving records by oid while silently still failing every blob.
	115	//
	116	// Walks the blob FILES (the canonical bytes; each filename IS its atproto CID) rather than the bee,
	117	// because the bee is exactly what may be impoverished. Deliberately never INVENTS metadata: a file
	118	// with no bee entry is registered — so its bytes become fetchable by oid — but is reported as
	119	// `unknownMeta` rather than given a guessed mimeType, which would be served to clients as fact. Such a
	120	// blob is recoverable once its shared pointer replicates (BlobPointerStore.get adopts mime from there).
	121	// `prepare:false` matches the record path: the serve path chunks lazily on first request.
	122	async reindexOverlay () {
	123	if (!this._overlay) return { skipped: 'no-overlay' }
	124	let names
	125	try { names = fs.readdirSync(this._blobRoot) } catch (e) { if (e && e.code === 'ENOENT') return { registered: 0, failed: 0, metaUpdated: 0, unknownMeta: 0, total: 0 }; throw e }
	126	let registered = 0
	127	let failed = 0
	128	let metaUpdated = 0
	129	let unknownMeta = 0
	130	for (const cidStr of names) {
	131	const path = this._pathFor(cidStr)
	132	let size = 0
	133	try { const st = fs.statSync(path); if (!st.isFile()) continue; size = st.size } catch { continue }
	134	let oid = null
	135	try {
	136	const r = await this._overlay.registerFile('/blobs/' + cidStr, path, { size, prepare: false })
	137	oid = r ? r.contentHash : null
	138	} catch { failed++; continue }
	139	if (!oid) { failed++; continue }
	140	registered++
	141	const node = await this.bee.get(cidStr)
	142	if (!node) { unknownMeta++; continue }
	143	try {
	144	const meta = JSON.parse(b4a.toString(node.value, 'utf-8'))
	145	if (meta.oid !== oid) {
	146	meta.oid = oid
	147	await this.bee.put(cidStr, b4a.from(JSON.stringify(meta), 'utf-8'))
	148	metaUpdated++
	149	}
	150	} catch {}
	151	}
	152	return { registered, failed, metaUpdated, unknownMeta, total: names.length }
	153	}
	154
102	155	async close () {
103	156	try { if (this.bee) await this.bee.close() } catch {}
104	157	if (this._ownStore) { try { await this._store.close() } catch {} }

package.json +1 -1

…	…	@@ -1,6 +1,6 @@
1	1	{
2	2	"name": "atproto-store",
3		"version": "0.0.12",
	3	"version": "0.0.13",
4	4	"description": "Holepunch-native atproto repo store — records in a classic Hyperbee (collection/rkey -> DAG-CBOR). The P2P home; the atproto-repo module is the bridge serializer over it.",
5	5	"type": "module",
6	6	"main": "lib/repo-store.js",

test/blob-pointer-store.test.js +75 -0

…	…	@@ -62,6 +62,13 @@ async function mkNode (t, net) {
62	62	return new OverlayBlobStore(tmp(t), { overlay: mkOverlay(net), indexStore: tmp(t), name: 'blobs' }).ready()
63	63	}
64	64
	65	// Same, but hands back the blob root so a test can simulate on-disk state the bee doesn't know about.
	66	async function mkNodeAt (t, net) {
	67	const root = tmp(t)
	68	const blobs = await new OverlayBlobStore(root, { overlay: mkOverlay(net), indexStore: tmp(t), name: 'blobs' }).ready()
	69	return { blobs, root }
	70	}
	71
65	72	test('put publishes a shared {$oid,size,mimeType} pointer under the reserved __blob__ namespace', async (t) => {
66	73	const net = new Map(); const shared = new Map()
67	74	const store = new BlobPointerStore(mkStatebase(shared), await mkNode(t, net))
…	…	@@ -143,3 +150,71 @@ test('blob pointers are invisible to the record set (never enter a signed commit
143	150	t.ok([...shared.keys()].some((k) => k.startsWith(BLOB_PREFIX)), 'the blob pointer IS in the shared state')
144	151	t.ok(BLOB_PREFIX.startsWith('__'), 'and it lives in the reserved namespace that splitKey rejects')
145	152	})
	153
	154	test('backfill publishes pointers for blobs uploaded before this module existed', async (t) => {
	155	const net = new Map(); const shared = new Map()
	156	const blobs = await mkNode(t, net)
	157
	158	// Two blobs written straight to the byte layer — bytes + local metadata, but never announced.
	159	const { cid: c1 } = await blobs.put(BYTES, MIME)
	160	const { cid: c2 } = await blobs.put(Buffer.from('a second blob'), 'image/png')
	161	t.is(shared.size, 0, 'no pointers exist yet (this is the pre-migration state)')
	162
	163	const store = new BlobPointerStore(mkStatebase(shared), blobs)
	164	const r = await store.backfill()
	165
	166	t.is(r.written, 2, 'both pre-existing blobs were announced')
	167	t.is(r.unpublishable, 0, 'both had a local oid to point at')
	168	t.ok(shared.get(BLOB_PREFIX + c1.toString()) && shared.get(BLOB_PREFIX + c2.toString()), 'pointers are in the shared state')
	169	t.is(shared.get(BLOB_PREFIX + c2.toString()).mimeType, 'image/png', 'each pointer kept its own mime type')
	170
	171	const again = await store.backfill()
	172	t.is(again.written, 0, 'idempotent — a second run announces nothing new')
	173	t.is(again.skipped, 2, 'and leaves the existing pointers alone')
	174
	175	await store.close()
	176	})
	177
	178	test('backfill never clobbers a pointer a co-owner already published', async (t) => {
	179	const net = new Map(); const shared = new Map()
	180	const blobs = await mkNode(t, net)
	181	const { cid } = await blobs.put(BYTES, MIME)
	182
	183	const theirs = { $oid: 'published-by-a-co-owner', size: 99, mimeType: 'image/webp' }
	184	shared.set(BLOB_PREFIX + cid.toString(), theirs)
	185
	186	const r = await new BlobPointerStore(mkStatebase(shared), blobs).backfill()
	187	t.is(r.written, 0, 'nothing written')
	188	t.is(r.skipped, 1, 'the already-announced blob was skipped')
	189	t.alike(shared.get(BLOB_PREFIX + cid.toString()), theirs, 'the co-owner pointer survived untouched')
	190
	191	await blobs.close()
	192	})
	193
	194	test('reindexOverlay re-registers blob files after index loss, without inventing metadata', async (t) => {
	195	const net = new Map()
	196	const { blobs, root } = await mkNodeAt(t, net)
	197	const { cid } = await blobs.put(BYTES, MIME)
	198
	199	// A blob file present on disk that the local bee knows nothing about — the impoverished-index case.
	200	writeFileSync(join(root, 'bafkreiorphan00000000000000000000000000000000000000000'), Buffer.from('orphan bytes'))
	201
	202	// Index loss is a PEER-visible failure, not a local one: the files are still on disk, so this node
	203	// keeps serving its own blobs either way. What disappears is every other node's ability to fetch them
	204	// by oid — which is exactly what gates a co-owner serving an avatar.
	205	net.clear()
	206	t.is(net.size, 0, 'CONTROL: with the index lost, NOTHING is fetchable by a peer')
	207	t.ok(await blobs.get(cid.toString()), 'while this node still serves its own file from disk (local reads are unaffected)')
	208
	209	const r = await blobs.reindexOverlay()
	210	t.is(net.size, 2, 'after repair both blobs are fetchable by a peer again')
	211	t.is(r.registered, 2, 'both files on disk were re-registered as servable-by-oid')
	212	t.is(r.unknownMeta, 1, 'the orphan is reported, NOT given a guessed mimeType')
	213	t.is(r.failed, 0, 'no failures')
	214
	215	const got = await blobs.get(cid.toString())
	216	t.ok(got && Buffer.from(got.bytes).equals(BYTES), 'the known blob resolves again after repair')
	217	t.is(got.mimeType, MIME, 'and its real mime type was preserved, not overwritten')
	218
	219	await blobs.close()
	220	})

atproto-store commits / ec478c3

Changed files (4) +163 -1