OverviewFilesCommitsIssuesDocsReleasesPeers

atproto-store commits / b52ced8

0.0.15: blobs converge PROACTIVELY — reconcile the shared __blob__ pointers
I shipped half a feature. 0.0.12 published `__blob__/<cid>` so a co-owner COULD fetch a blob by oid, and then nothing ever walked those pointers. Blobs moved only LAZILY, when something happened to call get(). Two peers that never request each other's blobs never exchanged them. Observed exactly that on the fleet: images resolve to bsky (the bridge relays get-blob by RPC straight to whoever holds the bytes, so no replication is involved) while silently never replicating between nodes. SwarmDevice.reconcileBytes could never cover this — it walks the RECORD pointer set and skips `__`-prefixed keys by design, which is precisely where blob pointers live. BlobPointerStore.reconcile({limit}) closes it: - PULL-ONLY: never writes a pointer, so it is safe on a NON-WRITABLE replica — exactly the peer that needs to catch up (pinned by a test whose statebase set() throws). - presence via the new OverlayBlobStore.hasLocalFile (a statSync), NOT get(), so a converged node does zero reads instead of pulling every blob into memory to discover it already has it. `has()` was the wrong test: it only says the local bee knows the cid, and a blob can have an ADOPTED bee entry with no bytes. - bounded per pass with `more`; quiet once converged; an unresolvable pointer is counted, not thrown, so one dead blob cannot stall the rest. Tests: +4 (12 total, 54 asserts) — pulls announced blobs, idempotent, bounded, pull-only on a read-only statebase, and failure isolation. Does NOT help ashatron: it runs single-DID on the LEGACY backend (no storeBackend:'overlay'), so it has no overlay, no content-addressed blobs and no __blob__ pointers. That is a separate config migration. Nor does it address cross-DID blob availability — __blob__ lives in one DID's shared autobase, so it only converges co-owners of the SAME DID. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
b52ced8e1226f4ec1926dea5623a4bf3137865fd
by plankatron <fleeky@gmail.com> · 2026-07-27T00:51:53.000Z

Changed files (4) +134 -1

lib/blob-pointer-store.js +39 -0
……@@ -119,6 +119,45 @@ export class BlobPointerStore {
119119 return { written, skipped, unpublishable, total }
120120 }
121121
122 // PROACTIVE blob convergence — the half that was missing.
123 //
124 // Publishing `__blob__/<cid>` makes a blob FETCHABLE by any co-owner, but nothing was walking those
125 // pointers, so blobs only ever moved LAZILY: when something happened to call get(). Two peers that
126 // never request each other's blobs therefore never exchanged them — images resolved to bsky (the
127 // bridge relays get-blob by RPC to whoever holds the bytes) while silently never replicating between
128 // nodes. This is the blob counterpart of SwarmDevice.reconcileBytes, which deliberately skips
129 // `__`-prefixed keys and so could never cover them.
130 //
131 // PULL-ONLY: it never writes a pointer, so it is safe on a NON-WRITABLE replica — exactly the peer
132 // that needs to catch up. Presence is tested with hasLocalFile (a stat), not get(), so a converged
133 // node does no reads at all. Bounded per pass with `more`, so a large backlog cannot stall the caller.
134 async reconcile ({ limit = 8 } = {}) {
135 if (typeof this.blobs.hasLocalFile !== 'function') return { skipped: 'not-supported' }
136 let map
137 try {
138 map = await ((typeof this.sb.listNow === 'function') ? this.sb.listNow(BLOB_PREFIX) : this.sb.list(BLOB_PREFIX))
139 } catch (e) { return { skipped: 'list-failed', error: e.message } }
140 let checked = 0; let missing = 0; let fetched = 0; let failed = 0; let more = false
141 for (const key of Object.keys(map)) {
142 if (!key.startsWith(BLOB_PREFIX)) continue
143 const cid = key.slice(BLOB_PREFIX.length)
144 if (!cid) continue
145 checked++
146 if (this.blobs.hasLocalFile(cid)) continue // bytes already here — a stat, not a read
147 missing++
148 if (fetched + failed >= limit) { more = true; continue }
149 let got = null
150 try { got = await this.get(cid) } catch {}
151 if (got) fetched++; else failed++
152 }
153 const sig = checked + ':' + missing + ':' + fetched + ':' + failed
154 if ((fetched || missing) && sig !== this._lastReconcileSig) {
155 try { console.log(JSON.stringify({ msg: 'blob reconcile: pulling shared-pointer blobs', did: this._did, checked, missing, fetched, failed, more })) } catch {}
156 }
157 this._lastReconcileSig = sig
158 return { checked, missing, fetched, failed, more }
159 }
160
122161 // Pass through to the byte layer's repair (re-register blob files so peers can fetch them by oid).
123162 async reindexOverlay () {
124163 if (typeof this.blobs.reindexOverlay !== 'function') return { skipped: 'not-supported' }
lib/overlay-blob-store.js +8 -0
……@@ -99,6 +99,14 @@ export class OverlayBlobStore {
9999
100100 async has (cidStr) { return !!(await this.bee.get(cidStr)) }
101101
102 // Do we hold the BYTES on disk? Distinct from has(), which only reports that the local metadata bee
103 // knows this cid — a blob can have an ADOPTED bee entry (learned from a shared pointer) with no file
104 // yet. Reconciliation needs the byte-level answer, and needs it CHEAP: a stat, not a read, so a pass
105 // over hundreds of blobs does not pull every one into memory just to discover it is already present.
106 hasLocalFile (cidStr) {
107 try { return fs.statSync(this._pathFor(String(cidStr))).isFile() } catch { return false }
108 }
109
102110 // Everything this node knows locally: { cid, mimeType, size, oid }. The input to pointer backfill.
103111 async * entries () {
104112 for await (const node of this.bee.createReadStream()) {
package.json +1 -1
……@@ -1,6 +1,6 @@
11{
22 "name": "atproto-store",
3 "version": "0.0.14",
3 "version": "0.0.15",
44 "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.",
55 "type": "module",
66 "main": "lib/repo-store.js",
test/blob-pointer-store.test.js +86 -0
……@@ -218,3 +218,89 @@ test('reindexOverlay re-registers blob files after index loss, without inventing
218218
219219 await blobs.close()
220220})
221
222test('reconcile PULLS blobs a co-owner announced but this node lacks', async (t) => {
223 const net = new Map(); const shared = new Map()
224
225 // Node A uploads two blobs — bytes enter the shared transport, pointers enter the shared state.
226 const a = new BlobPointerStore(mkStatebase(shared), await mkNode(t, net))
227 const { cid: c1 } = await a.put(BYTES, MIME)
228 const { cid: c2 } = await a.put(Buffer.from('second blob bytes'), 'image/png')
229
230 // Node B has the same replicated state, but none of the bytes.
231 const { blobs: bBlobs } = await mkNodeAt(t, net)
232 const b = new BlobPointerStore(mkStatebase(shared), bBlobs)
233 t.absent(bBlobs.hasLocalFile(c1.toString()), 'precondition: B holds no bytes for blob 1')
234 t.absent(bBlobs.hasLocalFile(c2.toString()), 'precondition: B holds no bytes for blob 2')
235
236 const r = await b.reconcile()
237 t.is(r.checked, 2, 'both announced blobs were checked')
238 t.is(r.missing, 2, 'both were missing locally')
239 t.is(r.fetched, 2, 'both were PULLED — this is what never happened before')
240 t.ok(bBlobs.hasLocalFile(c1.toString()) && bBlobs.hasLocalFile(c2.toString()), 'B now holds the bytes on disk')
241 t.ok(Buffer.from((await b.get(c1.toString())).bytes).equals(BYTES), 'and they are byte-identical')
242
243 const again = await b.reconcile()
244 t.is(again.missing, 0, 'idempotent — a converged node has nothing to pull')
245 t.is(again.fetched, 0, 'and does no work')
246
247 await a.close(); await b.close()
248})
249
250test('reconcile is BOUNDED per pass and reports `more`', async (t) => {
251 const net = new Map(); const shared = new Map()
252 const a = new BlobPointerStore(mkStatebase(shared), await mkNode(t, net))
253 for (let i = 0; i < 5; i++) await a.put(Buffer.from('blob number ' + i), 'image/png')
254
255 const { blobs: bBlobs } = await mkNodeAt(t, net)
256 const b = new BlobPointerStore(mkStatebase(shared), bBlobs)
257
258 const first = await b.reconcile({ limit: 2 })
259 t.is(first.fetched, 2, 'a bounded pass pulls at most `limit`')
260 t.ok(first.more, 'and reports there is more to do')
261
262 let guard = 0
263 while ((await b.reconcile({ limit: 2 })).more && guard++ < 10) { /* drain */ }
264 t.is((await b.reconcile({ limit: 2 })).missing, 0, 'successive passes converge the backlog')
265
266 await a.close(); await b.close()
267})
268
269test('reconcile is PULL-ONLY — it never writes a pointer (safe on a non-writable replica)', async (t) => {
270 const net = new Map(); const shared = new Map()
271 const a = new BlobPointerStore(mkStatebase(shared), await mkNode(t, net))
272 await a.put(BYTES, MIME)
273 const before = [...shared.keys()].sort()
274
275 // A statebase whose set() THROWS — exactly a non-writable co-owner. reconcile must still converge.
276 const readOnly = {
277 async set () { throw new Error('not writable') },
278 async get (k) { return shared.has(k) ? shared.get(k) : null },
279 async list (prefix = '') { const o = {}; for (const [k, v] of shared) if (k.startsWith(prefix)) o[k] = v; return o }
280 }
281 const { blobs: bBlobs } = await mkNodeAt(t, net)
282 const b = new BlobPointerStore(readOnly, bBlobs)
283
284 const r = await b.reconcile()
285 t.is(r.fetched, 1, 'a NON-WRITABLE replica still pulls the bytes')
286 t.alike([...shared.keys()].sort(), before, 'and wrote nothing to the shared state')
287
288 await a.close(); await b.close()
289})
290
291test('an unresolvable pointer is counted, not thrown, and does not block the rest', async (t) => {
292 const net = new Map(); const shared = new Map()
293 const a = new BlobPointerStore(mkStatebase(shared), await mkNode(t, net))
294 const { cid } = await a.put(BYTES, MIME)
295 shared.set(BLOB_PREFIX + 'bafkreighost0000000000000000000000000000000000000000', { $oid: 'nobody-has-this', size: 5, mimeType: 'image/png' })
296
297 const { blobs: bBlobs } = await mkNodeAt(t, net)
298 const b = new BlobPointerStore(mkStatebase(shared), bBlobs)
299 const r = await b.reconcile()
300
301 t.is(r.fetched, 1, 'the resolvable blob was pulled')
302 t.is(r.failed, 1, 'the unresolvable one was counted')
303 t.ok(bBlobs.hasLocalFile(cid.toString()), 'and the good blob is present')
304
305 await a.close(); await b.close()
306})