// BlobPointerStore — proves the cross-node gap it exists to close, and proves the pointer stays out
// of the signed repo.
//
// The gap: OverlayBlobStore's metadata bee ({mimeType,size,oid}) is node-local and never replicates,
// and its get() returns null on a missing bee entry BEFORE looking at the file or the overlay. So a
// co-owner that can reach the bytes still cannot serve them — it lacks the cid -> oid mapping, and
// since the atproto blob CID (raw) and the overlay oid (blake2b) are different hashes it cannot
// derive one from the other. Test 2 below is the load-bearing one: it runs node B twice, WITHOUT and
// WITH the shared pointer, so the fix is measured against its own negative control.
//
// The fake overlay models exactly the property that matters — bytes are fetchable by oid from
// anywhere, metadata is not — without needing two real daemons.
import test from 'brittle'
import { mkdtempSync, rmSync, readFileSync, writeFileSync } from 'node:fs'
import { join } from 'node:path'
import { tmpdir } from 'node:os'
import { createHash } from 'node:crypto'
import { OverlayBlobStore } from '../lib/overlay-blob-store.js'
import { BlobPointerStore, BLOB_PREFIX } from '../lib/blob-pointer-store.js'
import { PointerSubstrate } from '../lib/pointer-substrate.js'
import { OverlayRepoStore } from '../lib/overlay-repo-store.js'
const BYTES = Buffer.from('\xff\xd8\xff\xe0 pretend this is an avatar', 'binary')
const MIME = 'image/jpeg'
function tmp (t) { const d = mkdtempSync(join(tmpdir(), 'bps-')); t.teardown(() => rmSync(d, { recursive: true, force: true })); return d }
// Content-addressed byte transport shared by every "node" — the property a real overlay provides.
// sha256 stands in for blake2b; only distinctness from the atproto CID matters here.
function mkOverlay (net) {
return {
async registerFile (overlayPath, diskPath) {
const bytes = readFileSync(diskPath)
const oid = createHash('sha256').update(bytes).digest('hex')
net.set(oid, bytes)
return { contentHash: oid }
},
async fetchFile (oid, { destPath } = {}) {
const bytes = net.get(oid)
if (!bytes) return null
writeFileSync(destPath, bytes)
return { path: destPath }
}
}
}
// The replicated autobase view: every node reads the same map.
function mkStatebase (shared) {
return {
async set (k, v) { shared.set(k, v) },
async get (k) { return shared.has(k) ? shared.get(k) : null },
async list (prefix = '') {
const out = {}
for (const [k, v] of shared) if (k.startsWith(prefix)) out[k] = v
return out
}
}
}
async function mkNode (t, net) {
return new OverlayBlobStore(tmp(t), { overlay: mkOverlay(net), indexStore: tmp(t), name: 'blobs' }).ready()
}
// Same, but hands back the blob root so a test can simulate on-disk state the bee doesn't know about.
async function mkNodeAt (t, net) {
const root = tmp(t)
const blobs = await new OverlayBlobStore(root, { overlay: mkOverlay(net), indexStore: tmp(t), name: 'blobs' }).ready()
return { blobs, root }
}
test('put publishes a shared {$oid,size,mimeType} pointer under the reserved __blob__ namespace', async (t) => {
const net = new Map(); const shared = new Map()
const store = new BlobPointerStore(mkStatebase(shared), await mkNode(t, net))
const { cid, size } = await store.put(BYTES, MIME)
const ptr = shared.get(BLOB_PREFIX + cid.toString())
t.ok(ptr, 'a pointer was published')
t.ok(ptr.$oid, 'pointer carries the overlay oid (what a peer fetches by)')
t.is(ptr.mimeType, MIME, 'pointer carries the mime type')
t.is(ptr.size, size, 'pointer carries the size')
t.not(ptr.$oid, cid.toString(), 'oid and atproto CID are DIFFERENT hashes — the mapping cannot be recomputed')
await store.close()
})
test('a co-owner that never uploaded the blob can serve it — and CANNOT without the pointer', async (t) => {
const net = new Map(); const shared = new Map()
// Node A uploads. Bytes enter the shared content-addressed transport; A's bee is A's alone.
const a = new BlobPointerStore(mkStatebase(shared), await mkNode(t, net))
const { cid } = await a.put(BYTES, MIME)
const cidStr = cid.toString()
// NEGATIVE CONTROL — node B's bare OverlayBlobStore, bytes reachable in `net`, no shared pointer.
const bare = await mkNode(t, net)
t.is(await bare.get(cidStr), null, 'CONTROL: bytes are reachable but B cannot serve them (no cid->oid mapping)')
// Same node B, now reading through the pointer store.
const b = new BlobPointerStore(mkStatebase(shared), await mkNode(t, net))
const got = await b.get(cidStr)
t.ok(got, 'B resolves the blob via the shared pointer')
t.ok(Buffer.from(got.bytes).equals(BYTES), 'bytes are byte-identical to what A uploaded')
t.is(got.mimeType, MIME, 'mime type survived the hop')
t.ok(await b.has(cidStr), 'has() reports swarm-wide availability')
// Self-healing: the metadata is now local, so the next read needs no pointer lookup.
t.ok(await b.blobs.get(cidStr), 'B adopted the metadata into its own bee')
await a.close(); await bare.close(); await b.close()
})
test('a failed pointer write does not fail the upload (blob stays locally servable)', async (t) => {
const net = new Map()
const readOnly = { async set () { throw new Error('not writable') }, async get () { return null } }
const store = new BlobPointerStore(readOnly, await mkNode(t, net))
const r = await store.put(BYTES, MIME) // a non-writable device must still be able to store a blob
t.ok(r.cid, 'upload succeeded despite the pointer write failing')
t.ok(await store.get(r.cid.toString()), 'and the blob still serves from this node')
await store.close()
})
test('adoptMeta never clobbers metadata this node already has', async (t) => {
const net = new Map()
const blobs = await mkNode(t, net)
const { cid } = await blobs.put(BYTES, MIME)
const adopted = await blobs.adoptMeta(cid.toString(), { mimeType: 'application/octet-stream', size: 1, oid: 'bogus' })
t.is(adopted, false, 'adoptMeta declined to overwrite')
t.is((await blobs.get(cid.toString())).mimeType, MIME, 'original metadata intact')
await blobs.close()
})
test('blob pointers are invisible to the record set (never enter a signed commit)', async (t) => {
const net = new Map(); const shared = new Map()
const sb = mkStatebase(shared)
const records = new PointerSubstrate(sb, await new OverlayRepoStore(tmp(t), { overlay: mkOverlay(net) }).ready())
await records.set('app.bsky.feed.post/3kaaa', { $type: 'app.bsky.feed.post', text: 'hello' })
await new BlobPointerStore(sb, await mkNode(t, net)).put(BYTES, MIME)
const listed = await records.list('')
t.is(Object.keys(listed).length, 1, 'exactly one record — the blob pointer is not one')
t.ok(listed['app.bsky.feed.post/3kaaa'], 'the real record is present')
t.ok([...shared.keys()].some((k) => k.startsWith(BLOB_PREFIX)), 'the blob pointer IS in the shared state')
t.ok(BLOB_PREFIX.startsWith('__'), 'and it lives in the reserved namespace that splitKey rejects')
})
test('backfill publishes pointers for blobs uploaded before this module existed', async (t) => {
const net = new Map(); const shared = new Map()
const blobs = await mkNode(t, net)
// Two blobs written straight to the byte layer — bytes + local metadata, but never announced.
const { cid: c1 } = await blobs.put(BYTES, MIME)
const { cid: c2 } = await blobs.put(Buffer.from('a second blob'), 'image/png')
t.is(shared.size, 0, 'no pointers exist yet (this is the pre-migration state)')
const store = new BlobPointerStore(mkStatebase(shared), blobs)
const r = await store.backfill()
t.is(r.written, 2, 'both pre-existing blobs were announced')
t.is(r.unpublishable, 0, 'both had a local oid to point at')
t.ok(shared.get(BLOB_PREFIX + c1.toString()) && shared.get(BLOB_PREFIX + c2.toString()), 'pointers are in the shared state')
t.is(shared.get(BLOB_PREFIX + c2.toString()).mimeType, 'image/png', 'each pointer kept its own mime type')
const again = await store.backfill()
t.is(again.written, 0, 'idempotent — a second run announces nothing new')
t.is(again.skipped, 2, 'and leaves the existing pointers alone')
await store.close()
})
test('backfill never clobbers a pointer a co-owner already published', async (t) => {
const net = new Map(); const shared = new Map()
const blobs = await mkNode(t, net)
const { cid } = await blobs.put(BYTES, MIME)
const theirs = { $oid: 'published-by-a-co-owner', size: 99, mimeType: 'image/webp' }
shared.set(BLOB_PREFIX + cid.toString(), theirs)
const r = await new BlobPointerStore(mkStatebase(shared), blobs).backfill()
t.is(r.written, 0, 'nothing written')
t.is(r.skipped, 1, 'the already-announced blob was skipped')
t.alike(shared.get(BLOB_PREFIX + cid.toString()), theirs, 'the co-owner pointer survived untouched')
await blobs.close()
})
test('reindexOverlay re-registers blob files after index loss, without inventing metadata', async (t) => {
const net = new Map()
const { blobs, root } = await mkNodeAt(t, net)
const { cid } = await blobs.put(BYTES, MIME)
// A blob file present on disk that the local bee knows nothing about — the impoverished-index case.
writeFileSync(join(root, 'bafkreiorphan00000000000000000000000000000000000000000'), Buffer.from('orphan bytes'))
// Index loss is a PEER-visible failure, not a local one: the files are still on disk, so this node
// keeps serving its own blobs either way. What disappears is every other node's ability to fetch them
// by oid — which is exactly what gates a co-owner serving an avatar.
net.clear()
t.is(net.size, 0, 'CONTROL: with the index lost, NOTHING is fetchable by a peer')
t.ok(await blobs.get(cid.toString()), 'while this node still serves its own file from disk (local reads are unaffected)')
const r = await blobs.reindexOverlay()
t.is(net.size, 2, 'after repair both blobs are fetchable by a peer again')
t.is(r.registered, 2, 'both files on disk were re-registered as servable-by-oid')
t.is(r.unknownMeta, 1, 'the orphan is reported, NOT given a guessed mimeType')
t.is(r.failed, 0, 'no failures')
const got = await blobs.get(cid.toString())
t.ok(got && Buffer.from(got.bytes).equals(BYTES), 'the known blob resolves again after repair')
t.is(got.mimeType, MIME, 'and its real mime type was preserved, not overwritten')
await blobs.close()
})
test('reconcile PULLS blobs a co-owner announced but this node lacks', async (t) => {
const net = new Map(); const shared = new Map()
// Node A uploads two blobs — bytes enter the shared transport, pointers enter the shared state.
const a = new BlobPointerStore(mkStatebase(shared), await mkNode(t, net))
const { cid: c1 } = await a.put(BYTES, MIME)
const { cid: c2 } = await a.put(Buffer.from('second blob bytes'), 'image/png')
// Node B has the same replicated state, but none of the bytes.
const { blobs: bBlobs } = await mkNodeAt(t, net)
const b = new BlobPointerStore(mkStatebase(shared), bBlobs)
t.absent(bBlobs.hasLocalFile(c1.toString()), 'precondition: B holds no bytes for blob 1')
t.absent(bBlobs.hasLocalFile(c2.toString()), 'precondition: B holds no bytes for blob 2')
const r = await b.reconcile()
t.is(r.checked, 2, 'both announced blobs were checked')
t.is(r.missing, 2, 'both were missing locally')
t.is(r.fetched, 2, 'both were PULLED — this is what never happened before')
t.ok(bBlobs.hasLocalFile(c1.toString()) && bBlobs.hasLocalFile(c2.toString()), 'B now holds the bytes on disk')
t.ok(Buffer.from((await b.get(c1.toString())).bytes).equals(BYTES), 'and they are byte-identical')
const again = await b.reconcile()
t.is(again.missing, 0, 'idempotent — a converged node has nothing to pull')
t.is(again.fetched, 0, 'and does no work')
await a.close(); await b.close()
})
test('reconcile is BOUNDED per pass and reports `more`', async (t) => {
const net = new Map(); const shared = new Map()
const a = new BlobPointerStore(mkStatebase(shared), await mkNode(t, net))
for (let i = 0; i < 5; i++) await a.put(Buffer.from('blob number ' + i), 'image/png')
const { blobs: bBlobs } = await mkNodeAt(t, net)
const b = new BlobPointerStore(mkStatebase(shared), bBlobs)
const first = await b.reconcile({ limit: 2 })
t.is(first.fetched, 2, 'a bounded pass pulls at most `limit`')
t.ok(first.more, 'and reports there is more to do')
let guard = 0
while ((await b.reconcile({ limit: 2 })).more && guard++ < 10) { /* drain */ }
t.is((await b.reconcile({ limit: 2 })).missing, 0, 'successive passes converge the backlog')
await a.close(); await b.close()
})
test('reconcile is PULL-ONLY — it never writes a pointer (safe on a non-writable replica)', async (t) => {
const net = new Map(); const shared = new Map()
const a = new BlobPointerStore(mkStatebase(shared), await mkNode(t, net))
await a.put(BYTES, MIME)
const before = [...shared.keys()].sort()
// A statebase whose set() THROWS — exactly a non-writable co-owner. reconcile must still converge.
const readOnly = {
async set () { throw new Error('not writable') },
async get (k) { return shared.has(k) ? shared.get(k) : null },
async list (prefix = '') { const o = {}; for (const [k, v] of shared) if (k.startsWith(prefix)) o[k] = v; return o }
}
const { blobs: bBlobs } = await mkNodeAt(t, net)
const b = new BlobPointerStore(readOnly, bBlobs)
const r = await b.reconcile()
t.is(r.fetched, 1, 'a NON-WRITABLE replica still pulls the bytes')
t.alike([...shared.keys()].sort(), before, 'and wrote nothing to the shared state')
await a.close(); await b.close()
})
test('an unresolvable pointer is counted, not thrown, and does not block the rest', async (t) => {
const net = new Map(); const shared = new Map()
const a = new BlobPointerStore(mkStatebase(shared), await mkNode(t, net))
const { cid } = await a.put(BYTES, MIME)
shared.set(BLOB_PREFIX + 'bafkreighost0000000000000000000000000000000000000000', { $oid: 'nobody-has-this', size: 5, mimeType: 'image/png' })
const { blobs: bBlobs } = await mkNodeAt(t, net)
const b = new BlobPointerStore(mkStatebase(shared), bBlobs)
const r = await b.reconcile()
t.is(r.fetched, 1, 'the resolvable blob was pulled')
t.is(r.failed, 1, 'the unresolvable one was counted')
t.ok(bBlobs.hasLocalFile(cid.toString()), 'and the good blob is present')
await a.close(); await b.close()
})