2732ae1 · atproto-store

Changed files (4) +266 -2

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

…	…	@@ -0,0 +1,105 @@
	1	// BlobPointerStore — the blob half of the "pointers in the autobase, bytes in the overlay" design.
	2	// This is to blobs exactly what PointerSubstrate is to records, and it exists because the records
	3	// side got this right and the blobs side did not.
	4	//
	5	// THE GAP IT CLOSES. OverlayBlobStore keeps a per-blob metadata record { mimeType, size, oid } in a
	6	// Hyperbee, and that bee is NODE-LOCAL — in the live daemon it lives in the membrane corestore
	7	// (device-node.js: `overlay-blobs:<did>` on membraneStore), which does not replicate. Its get() reads
	8	// that bee FIRST and returns null when the entry is missing, BEFORE it ever looks at the file or the
	9	// overlay. So a blob uploaded on node A produces two artifacts with very different fates:
	10	//
	11	// the FILE + overlay registration -> content-addressed, fetchable by `oid` from any peer
	12	// the bee entry { mimeType,size,oid } -> stays on node A, forever
	13	//
	14	// Node B therefore cannot serve the blob even when the bytes are reachable, because the one thing it
	15	// is missing is the cid -> oid mapping needed to ask for them. Since the atproto blob CID (raw codec,
	16	// cidForBlobSync) and the overlay oid (blake2b) are DIFFERENT hashes, B cannot derive one from the
	17	// other — the mapping has to be replicated, not recomputed. Symptom in production: an avatar that
	18	// serves or 404s depending on which co-owner the bridge happens to front.
	19	//
	20	// THE FIX is the pattern PointerSubstrate already proves for records: put a tiny pointer in the SHARED
	21	// autobase so every co-owner converges on which blobs exist and the oid to fetch each one by, while the
	22	// bytes stay out of the append-only log. Pointer key is `__blob__/<cid>`, in the RESERVED "__"
	23	// namespace — RepoBase.splitKey and PointerSubstrate both return null for a collection starting with
	24	// "__", so these entries can never be mistaken for records or materialized into a signed commit.
	25	//
	26	// Drop-in: same public surface as BlobStore / OverlayBlobStore (ready / put / get / has / close), so it
	27	// composes under DualBlobStore unchanged.
	28
	29	const BLOB_PREFIX = '__blob__/'
	30
	31	export class BlobPointerStore {
	32	// statebase: the shared StateBase (set/get). blobs: an OverlayBlobStore (the local bytes + bee).
	33	constructor (statebase, blobs, { did = null } = {}) {
	34	if (!statebase \|\| typeof statebase.set !== 'function' \|\| typeof statebase.get !== 'function') {
	35	throw new Error('BlobPointerStore: a StateBase (set/get) is required for the shared pointer set')
	36	}
	37	if (!blobs \|\| typeof blobs.put !== 'function' \|\| typeof blobs.adoptMeta !== 'function') {
	38	throw new Error('BlobPointerStore: an OverlayBlobStore (put/get/adoptMeta) is required for the bytes')
	39	}
	40	this.sb = statebase
	41	this.blobs = blobs
	42	this._did = did
	43	}
	44
	45	static key (cidStr) { return BLOB_PREFIX + cidStr }
	46
	47	async ready () { return this } // the caller readies the underlying OverlayBlobStore
	48
	49	// Write the bytes locally (file + overlay registration + local bee), then publish the SHARED pointer
	50	// so any co-owner can resolve this blob by oid.
	51	//
	52	// A failed pointer write does NOT fail the upload. The blob is already durably stored and servable
	53	// from this node, so throwing here would turn a cross-node limitation into a total upload failure —
	54	// strictly worse than today. But it is logged LOUDLY rather than swallowed, because the resulting
	55	// state (blob local-only) is exactly the silent condition this module exists to end. The common
	56	// cause is a non-writable device: sb.set appends to the autobase, so only an admitted writer can
	57	// publish a pointer.
	58	async put (bytes, mimeType) {
	59	const r = await this.blobs.put(bytes, mimeType)
	60	const cidStr = r.cid.toString()
	61	if (r.oid) {
	62	try {
	63	await this.sb.set(BlobPointerStore.key(cidStr), { $oid: r.oid, size: r.size, mimeType: r.mimeType })
	64	} catch (err) {
	65	this._warn(cidStr, 'pointer write failed — blob is stored LOCAL-ONLY and will not serve from other co-owners', err && err.message)
	66	}
	67	} else {
	68	this._warn(cidStr, 'no overlay oid — blob cannot be published as a shared pointer (is the overlay attached?)', null)
	69	}
	70	return r
	71	}
	72
	73	// Local first (bee hit -> file, or fetch-by-oid). On a miss, consult the SHARED pointer: adopt its
	74	// metadata into the local bee, then re-enter the ordinary get() path, which now finds the mapping and
	75	// fetches the bytes by oid. Reusing get() rather than fetching here keeps ONE byte-resolution path.
	76	async get (cidStr) {
	77	const cid = String(cidStr)
	78	const local = await this.blobs.get(cid)
	79	if (local) return local
	80	const ptr = await this._pointer(cid)
	81	if (!ptr \|\| !ptr.$oid) return null
	82	await this.blobs.adoptMeta(cid, { mimeType: ptr.mimeType, size: ptr.size, oid: ptr.$oid })
	83	return this.blobs.get(cid)
	84	}
	85
	86	// Known here OR known to the swarm. A shared pointer means the bytes are fetchable even if this node
	87	// has never seen them, so reporting only the local bee would understate availability.
	88	async has (cidStr) {
	89	const cid = String(cidStr)
	90	if (await this.blobs.has(cid)) return true
	91	return !!(await this._pointer(cid))
	92	}
	93
	94	async _pointer (cidStr) {
	95	try { return await this.sb.get(BlobPointerStore.key(cidStr)) } catch { return null }
	96	}
	97
	98	_warn (cid, msg, error) {
	99	try { console.error(JSON.stringify({ msg: 'blob pointer: ' + msg, did: this._did, cid, error })) } catch {}
	100	}
	101
	102	async close () { try { await this.blobs.close() } catch {} }
	103	}
	104
	105	export { BLOB_PREFIX }

lib/overlay-blob-store.js +15 -1

…	…	@@ -59,7 +59,21 @@ export class OverlayBlobStore {
59	59	oid = r ? r.contentHash : null
60	60	}
61	61	await this.bee.put(cidStr, b4a.from(JSON.stringify({ mimeType, size: buf.length, oid }), 'utf-8'))
62		return { cid, mimeType, size: buf.length }
	62	// `oid` is returned so a caller can publish it as a SHARED pointer (see BlobPointerStore). The bee
	63	// here is node-local and never replicates, so without propagating the oid a co-owner that receives
	64	// the blob FILE still cannot serve it — get() below returns null on a missing bee entry.
	65	return { cid, mimeType, size: buf.length, oid }
	66	}
	67
	68	// Record metadata for a blob this node did NOT upload — learned from a replicated source (the shared
	69	// pointer set). Writing the bee entry is what makes get() able to resolve the blob at all: get() reads
	70	// the bee FIRST and returns null when it is absent, even if the bytes are already on disk. After this,
	71	// the ordinary get() path applies unchanged (local file hit, else fetchFile by oid).
	72	async adoptMeta (cidStr, { mimeType = 'application/octet-stream', size = 0, oid = null } = {}) {
	73	const existing = await this.bee.get(cidStr)
	74	if (existing) return false // never clobber locally-known metadata
	75	await this.bee.put(cidStr, b4a.from(JSON.stringify({ mimeType, size, oid }), 'utf-8'))
	76	return true
63	77	}
64	78
65	79	// Fetch by CID string; returns { mimeType, size, bytes } or null. Serves the local file if present;

package.json +1 -1

…	…	@@ -1,6 +1,6 @@
1	1	{
2	2	"name": "atproto-store",
3		"version": "0.0.11",
	3	"version": "0.0.12",
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 +145 -0

…	…	@@ -0,0 +1,145 @@
	1	// BlobPointerStore — proves the cross-node gap it exists to close, and proves the pointer stays out
	2	// of the signed repo.
	3	//
	4	// The gap: OverlayBlobStore's metadata bee ({mimeType,size,oid}) is node-local and never replicates,
	5	// and its get() returns null on a missing bee entry BEFORE looking at the file or the overlay. So a
	6	// co-owner that can reach the bytes still cannot serve them — it lacks the cid -> oid mapping, and
	7	// since the atproto blob CID (raw) and the overlay oid (blake2b) are different hashes it cannot
	8	// derive one from the other. Test 2 below is the load-bearing one: it runs node B twice, WITHOUT and
	9	// WITH the shared pointer, so the fix is measured against its own negative control.
	10	//
	11	// The fake overlay models exactly the property that matters — bytes are fetchable by oid from
	12	// anywhere, metadata is not — without needing two real daemons.
	13
	14	import test from 'brittle'
	15	import { mkdtempSync, rmSync, readFileSync, writeFileSync } from 'node:fs'
	16	import { join } from 'node:path'
	17	import { tmpdir } from 'node:os'
	18	import { createHash } from 'node:crypto'
	19	import { OverlayBlobStore } from '../lib/overlay-blob-store.js'
	20	import { BlobPointerStore, BLOB_PREFIX } from '../lib/blob-pointer-store.js'
	21	import { PointerSubstrate } from '../lib/pointer-substrate.js'
	22	import { OverlayRepoStore } from '../lib/overlay-repo-store.js'
	23
	24	const BYTES = Buffer.from('\xff\xd8\xff\xe0 pretend this is an avatar', 'binary')
	25	const MIME = 'image/jpeg'
	26
	27	function tmp (t) { const d = mkdtempSync(join(tmpdir(), 'bps-')); t.teardown(() => rmSync(d, { recursive: true, force: true })); return d }
	28
	29	// Content-addressed byte transport shared by every "node" — the property a real overlay provides.
	30	// sha256 stands in for blake2b; only distinctness from the atproto CID matters here.
	31	function mkOverlay (net) {
	32	return {
	33	async registerFile (overlayPath, diskPath) {
	34	const bytes = readFileSync(diskPath)
	35	const oid = createHash('sha256').update(bytes).digest('hex')
	36	net.set(oid, bytes)
	37	return { contentHash: oid }
	38	},
	39	async fetchFile (oid, { destPath } = {}) {
	40	const bytes = net.get(oid)
	41	if (!bytes) return null
	42	writeFileSync(destPath, bytes)
	43	return { path: destPath }
	44	}
	45	}
	46	}
	47
	48	// The replicated autobase view: every node reads the same map.
	49	function mkStatebase (shared) {
	50	return {
	51	async set (k, v) { shared.set(k, v) },
	52	async get (k) { return shared.has(k) ? shared.get(k) : null },
	53	async list (prefix = '') {
	54	const out = {}
	55	for (const [k, v] of shared) if (k.startsWith(prefix)) out[k] = v
	56	return out
	57	}
	58	}
	59	}
	60
	61	async function mkNode (t, net) {
	62	return new OverlayBlobStore(tmp(t), { overlay: mkOverlay(net), indexStore: tmp(t), name: 'blobs' }).ready()
	63	}
	64
	65	test('put publishes a shared {$oid,size,mimeType} pointer under the reserved __blob__ namespace', async (t) => {
	66	const net = new Map(); const shared = new Map()
	67	const store = new BlobPointerStore(mkStatebase(shared), await mkNode(t, net))
	68
	69	const { cid, size } = await store.put(BYTES, MIME)
	70	const ptr = shared.get(BLOB_PREFIX + cid.toString())
	71
	72	t.ok(ptr, 'a pointer was published')
	73	t.ok(ptr.$oid, 'pointer carries the overlay oid (what a peer fetches by)')
	74	t.is(ptr.mimeType, MIME, 'pointer carries the mime type')
	75	t.is(ptr.size, size, 'pointer carries the size')
	76	t.not(ptr.$oid, cid.toString(), 'oid and atproto CID are DIFFERENT hashes — the mapping cannot be recomputed')
	77
	78	await store.close()
	79	})
	80
	81	test('a co-owner that never uploaded the blob can serve it — and CANNOT without the pointer', async (t) => {
	82	const net = new Map(); const shared = new Map()
	83
	84	// Node A uploads. Bytes enter the shared content-addressed transport; A's bee is A's alone.
	85	const a = new BlobPointerStore(mkStatebase(shared), await mkNode(t, net))
	86	const { cid } = await a.put(BYTES, MIME)
	87	const cidStr = cid.toString()
	88
	89	// NEGATIVE CONTROL — node B's bare OverlayBlobStore, bytes reachable in `net`, no shared pointer.
	90	const bare = await mkNode(t, net)
	91	t.is(await bare.get(cidStr), null, 'CONTROL: bytes are reachable but B cannot serve them (no cid->oid mapping)')
	92
	93	// Same node B, now reading through the pointer store.
	94	const b = new BlobPointerStore(mkStatebase(shared), await mkNode(t, net))
	95	const got = await b.get(cidStr)
	96
	97	t.ok(got, 'B resolves the blob via the shared pointer')
	98	t.ok(Buffer.from(got.bytes).equals(BYTES), 'bytes are byte-identical to what A uploaded')
	99	t.is(got.mimeType, MIME, 'mime type survived the hop')
	100	t.ok(await b.has(cidStr), 'has() reports swarm-wide availability')
	101
	102	// Self-healing: the metadata is now local, so the next read needs no pointer lookup.
	103	t.ok(await b.blobs.get(cidStr), 'B adopted the metadata into its own bee')
	104
	105	await a.close(); await bare.close(); await b.close()
	106	})
	107
	108	test('a failed pointer write does not fail the upload (blob stays locally servable)', async (t) => {
	109	const net = new Map()
	110	const readOnly = { async set () { throw new Error('not writable') }, async get () { return null } }
	111	const store = new BlobPointerStore(readOnly, await mkNode(t, net))
	112
	113	const r = await store.put(BYTES, MIME) // a non-writable device must still be able to store a blob
	114	t.ok(r.cid, 'upload succeeded despite the pointer write failing')
	115	t.ok(await store.get(r.cid.toString()), 'and the blob still serves from this node')
	116
	117	await store.close()
	118	})
	119
	120	test('adoptMeta never clobbers metadata this node already has', async (t) => {
	121	const net = new Map()
	122	const blobs = await mkNode(t, net)
	123	const { cid } = await blobs.put(BYTES, MIME)
	124
	125	const adopted = await blobs.adoptMeta(cid.toString(), { mimeType: 'application/octet-stream', size: 1, oid: 'bogus' })
	126	t.is(adopted, false, 'adoptMeta declined to overwrite')
	127	t.is((await blobs.get(cid.toString())).mimeType, MIME, 'original metadata intact')
	128
	129	await blobs.close()
	130	})
	131
	132	test('blob pointers are invisible to the record set (never enter a signed commit)', async (t) => {
	133	const net = new Map(); const shared = new Map()
	134	const sb = mkStatebase(shared)
	135
	136	const records = new PointerSubstrate(sb, await new OverlayRepoStore(tmp(t), { overlay: mkOverlay(net) }).ready())
	137	await records.set('app.bsky.feed.post/3kaaa', { $type: 'app.bsky.feed.post', text: 'hello' })
	138	await new BlobPointerStore(sb, await mkNode(t, net)).put(BYTES, MIME)
	139
	140	const listed = await records.list('')
	141	t.is(Object.keys(listed).length, 1, 'exactly one record — the blob pointer is not one')
	142	t.ok(listed['app.bsky.feed.post/3kaaa'], 'the real record is present')
	143	t.ok([...shared.keys()].some((k) => k.startsWith(BLOB_PREFIX)), 'the blob pointer IS in the shared state')
	144	t.ok(BLOB_PREFIX.startsWith('__'), 'and it lives in the reserved namespace that splitKey rejects')
	145	})

atproto-store commits / 2732ae1

Changed files (4) +266 -2