// OverlayRepoStore — a RepoStore backend that keeps records as canonical DAG-CBOR FILES
// on disk (one file per record at records/<collection>/<rkey>.dagcbor) instead of in an
// append-only Hyperbee. Overwrite-in-place on edit, unlink on delete → the on-disk size is
// O(CURRENT records), not O(operations-ever). MST/commit/CAR are DERIVED on read by the
// atproto-repo codec (read-view.js / repo-manager.js consume only this interface), so nothing
// atproto-specific is stored beyond the record bytes + a tiny head.json.
//
// Byte-exactness (the determinism contract): `put` stores dagCbor.encode(record) VERBATIM —
// byte-identical to the Hyperbee RepoStore — so cidForBytesSync(bytes) and the derived MST +
// commit are identical. records() returns records in Hyperbee key order (utf-8 byte order via
// b4a.compare) so the derivation is order-stable.
//
// The canonical bytes ARE the files on disk. An optional hyper-overlay instance (opts.overlay)
// indexes + replicates them (bytes stay on disk — no corestore balloon; hyper-overlay/index.js).
// Without it the store is a fully-functional local file tree (used by the determinism tests).
// Drop-in for atproto-store/lib/repo-store.js's RepoStore (same method surface).
import * as dagCbor from '@ipld/dag-cbor'
import b4a from 'b4a'
import fs from '#node:fs' // dual-runtime: bare-fs under Bare (the daemon), builtin fs under Node (tests)
import { join } from '#node:path'
import { tidNow } from './tid.js'
const EXT = '.dagcbor'
const KEEP_MAX = 256 // superseded/ entries kept (see _keep)
const byteOrder = (a, b) => b4a.compare(b4a.from(a), b4a.from(b)) // match Hyperbee's key comparison exactly
// atproto collection NSIDs + rkeys are single path segments (no '/'); reject anything else so a
// record key can never escape the records/ dir or collide with a parent path.
const isSafeSeg = (s) => typeof s === 'string' && s.length > 0 && !s.includes('/') && s !== '.' && s !== '..'
export class OverlayRepoStore {
// root: the per-DID dir (<storage>/dids/<did>). opts.overlay: optional HyperOverlay to
// register files with (bytes-on-disk indexing/replication). opts.did: for logging/namespacing.
constructor (root, opts = {}) {
if (typeof root !== 'string') throw new Error('OverlayRepoStore: root dir path required')
this._root = root
this._recordsDir = join(root, 'records')
// Bytes a re-fetch REPLACED, kept content-addressed (<oid>.dagcbor) beside records/ — never inside it,
// where records() would read the dir as a collection. See _keep / _fromKept.
this._keptDir = join(root, 'superseded')
this._headFile = join(root, 'head.json')
this._overlay = opts.overlay || null
this._did = opts.did || null
}
static key (collection, rkey) { return `${collection}/${rkey}` }
async ready () {
fs.mkdirSync(this._recordsDir, { recursive: true })
// Restore servability across restart. The overlay's path maps are in-memory and only registerFile
// fills them, so without this a restarted node serves ONLY records written since boot — every file
// still on disk, yet peers get "no connected peer served this oid" (75+ hours of it on wj62), which
// blocks signing. Seeding from the PERSISTED index costs one scan with no hashing and no file
// reads, and it avoids the protocol's tier-3 full-index re-scan on every content request.
// Our overlay paths are disk paths relative to root, so the resolver is a plain join.
if (this._overlay && typeof this._overlay.seedFilePathsFromIndex === 'function') {
try {
const n = await this._overlay.seedFilePathsFromIndex((overlayPath) => join(this._root, overlayPath))
if (n) console.log(JSON.stringify({ msg: 'overlay: seeded record paths from the persisted index (servable-by-oid after restart)', did: this._did, seeded: n }))
} catch (err) { console.error(JSON.stringify({ msg: 'overlay path seed FAILED — this node may not serve its records by oid until re-registered', did: this._did, error: err.message })) }
}
return this
}
// EXPLICIT REPAIR — not a boot step. ready() seeds the path maps from the persisted index, which is
// the cheap normal path. This is the heavier fallback for a store whose INDEX itself is missing or
// stale (records written before overlay indexing existed, or an index rebuilt/lost): it re-hashes
// every record file and re-registers it. Needed once on nodes whose indexes predate the seed fix;
// deliberately NOT wired into ready(), because re-hashing every file at every boot is the cold-start
// cost that made pgh-sidecar take ~190s to bind its socket.
//
// Re-register every record file with the overlay so peers can fetch it BY OID.
//
// hyper-overlay's serve path resolves contentHash -> disk path through an IN-MEMORY map
// (overlay.js `_contentHashPaths`, populated only by registerFile and handed to protocol.js by
// reference). Nothing rebuilds it at boot, and the persisted FileIndex stores { contentHash, size,
// mtime } WITHOUT the disk path — so the overlay cannot repair itself; only the consumer knows the
// layout. Net effect before this: after a restart a node could serve ONLY the records it had
// written since boot ("registering a file makes it servable", overlay.js:15), even though every
// file was still on disk.
//
// That silently breaks cross-node byte resolution, which GATES SIGNING: the forward-only sign gate
// skips while a converged pointer's bytes are unresolved. Measured on the live fleet — wj62's
// app.bsky.feed.post/3mr6v6r2ofs22 sat unresolvable for 75+ hours while the peer holding it had the
// exact bytes on disk (hash verified equal to the pointer's oid), reporting only "no connected peer
// served this oid"; both co-owners held different incomplete subsets for the same reason.
//
// Idempotent and safe to run every boot: registerFile re-asserts an identical entry. Deliberately
// passes NO `rev` (a fresh TID would make a re-registration look like a newer write to rev-wins
// conflict resolution) and `prepare: false` (the serve path chunks lazily on first request, so we
// avoid a second full read of every file at boot).
async reindexOverlay () {
if (!this._overlay) return { skipped: 'no-overlay' }
let registered = 0
let failed = 0
const files = await this.records()
for (const { key, bytes } of files) {
const i = key.indexOf('/')
const collection = key.slice(0, i)
const rkey = key.slice(i + 1)
try {
const r = await this._overlay.registerFile(this._overlayPath(collection, rkey), this._pathFor(collection, rkey), { size: bytes.length, prepare: false })
if (r) registered++; else failed++
} catch { failed++ }
}
if (files.length) {
try { console.log(JSON.stringify({ msg: 'overlay reindex: record files re-registered as servable-by-oid', did: this._did, registered, failed, total: files.length })) } catch {}
}
return { registered, failed, total: files.length }
}
_pathFor (collection, rkey) {
if (!isSafeSeg(collection) || !isSafeSeg(rkey)) throw new Error(`OverlayRepoStore: unsafe record key "${collection}/${rkey}"`)
return join(this._recordsDir, collection, rkey + EXT)
}
// READ-ONLY: the overlay oid the record file CURRENTLY on disk hashes to, or null when absent.
// Writes nothing (no registration, no index entry) — safe against a live store.
//
// This is the missing half of the pointer design's integrity story. PointerSubstrate stores
// { $oid, size } in the shared autobase and resolves it through ensureLocal, which returns a local
// file hit WITHOUT comparing it to $oid — so a node holding a superseded copy of a record serves it
// forever and never re-fetches, while its pointer set looks perfectly converged. Nothing in the
// stack could even ask "does my file match the pointer?", which is why the wj62 profile divergence
// (two co-owners, identical 139 pointers, different avatar blob refs) was only visible by diffing
// two signed CARs against each other. Compare this against the pointer's $oid to answer it directly.
async localOid (collection, rkey) {
if (!this._overlay || typeof this._overlay.hashFile !== 'function') return null
let p
try { p = this._pathFor(collection, rkey) } catch { return null }
const r = await this._overlay.hashFile(p)
return r ? { oid: r.contentHash, size: r.size } : null
}
// Returns { oid, size }: oid = the overlay content hash (blake2b) that a PEER fetches the bytes by
// (PointerSubstrate stores {oid,size} as the symmetric autobase pointer). null oid when no overlay.
async put (collection, rkey, record) {
const bytes = b4a.from(dagCbor.encode(record))
const p = this._pathFor(collection, rkey)
fs.mkdirSync(join(this._recordsDir, collection), { recursive: true })
fs.writeFileSync(p, bytes)
// Stamp a monotonic per-write version (TID) so cross-device conflict resolution is atproto-native
// (rev-wins: the later write wins). Records carry no rev at put time — the commit rev comes at sign.
// Only affects the overlay change feed, never the canonical record file → determinism is preserved.
let oid = null
if (this._overlay) { try { const r = await this._overlay.registerFile(this._overlayPath(collection, rkey), p, { size: bytes.length, rev: tidNow() }); oid = r ? r.contentHash : null } catch {} }
return { oid, size: bytes.length }
}
// Ensure the record BYTES are present locally, fetching from a connected peer by overlay oid if this
// node holds only the pointer (a follower resolving the converged autobase set). Returns the decoded
// record or null. Local-hit is the common path; the fetch is the cross-node byte-resolution.
async ensureLocal (collection, rkey, oid) {
const local = await this.get(collection, rkey)
if (local != null) {
// A local hit used to return here UNCONDITIONALLY, without ever comparing the bytes to the oid
// the caller asked for. That made the pointer design's central promise unenforced: the autobase
// converges { $oid, size }, but a node holding a SUPERSEDED copy resolved it happily, never
// re-fetched, and signed it as truth. Live on wj62 — two co-owners, identical 139 converged
// pointers, both resolving all 139, different signed repos, differing on exactly one record
// (app.bsky.actor.profile/self, two avatar blob refs) — so the published profile flipped avatar
// on every signer change. Content-addressing was being used to FETCH but never to VERIFY.
if (!oid || !this._overlay || typeof this._overlay.hashFile !== 'function') return local
const h = await this.localOid(collection, rkey)
// Only a POSITIVE mismatch is staleness. If the bytes cannot be hashed at all we cannot judge
// them, and inventing a verdict would re-resolve every record on every pass — behaviour must be
// unchanged wherever verification is impossible.
if (!h || h.oid === oid) return local // verified against the converged pointer — the common path
this._logStaleLocal(collection, rkey, oid, h ? h.oid : null)
// The pointer may name bytes THIS node once held and replaced (a newer local write the lagging view
// had overwritten — loser-club 2026-10-14). Kept copies are verified and installed before the network.
const kept = await this._fromKept(collection, rkey, oid)
if (kept != null) return kept
const fetched = await this._refetchOver(collection, rkey, oid)
// Deliberately null, NOT the stale record, when the authoritative bytes cannot be obtained.
// Returning the superseded copy is what published wrong content: this record then reads as
// UNRESOLVED, which is precisely the state the forward-only sign gate is built to handle
// ("pointer present, bytes unreplicated -> behind -> SKIP"). A node that cannot prove its bytes
// match the group's pointer must not sign them. Note the trade: if NO peer holds the
// authoritative bytes, the record stays unresolved and the chain stalls rather than advancing
// with content nobody agreed on — loudly (see _logStaleLocal) instead of silently.
return fetched
}
if (!this._overlay || !oid) return null
const kept = await this._fromKept(collection, rkey, oid)
if (kept != null) return kept
const disk = this._pathFor(collection, rkey)
try {
fs.mkdirSync(join(this._recordsDir, collection), { recursive: true })
const r = await this._overlay.fetchFile(oid, { destPath: disk })
if (!r) { this._logFetchMiss(collection, rkey, oid, 'fetchFile returned null'); return null }
} catch (err) { this._logFetchMiss(collection, rkey, oid, (err && err.message) || String(err)); return null }
return this.get(collection, rkey)
}
// Keep the bytes currently at `disk` under superseded/<oid>.dagcbor before they are replaced. Bounded:
// only the newest KEEP_MAX entries stay (records are small; this exists to bridge a view that lags a
// local write by seconds, not to archive history). Best-effort — a failure here never blocks the repair.
async _keep (disk) {
try {
const h = await this._overlay.hashFile(disk)
if (!h || !h.contentHash) return
fs.mkdirSync(this._keptDir, { recursive: true })
const dest = join(this._keptDir, h.contentHash + EXT)
if (!fs.existsSync(dest)) fs.copyFileSync(disk, dest)
const entries = fs.readdirSync(this._keptDir).filter((n) => n.endsWith(EXT))
if (entries.length > KEEP_MAX) {
const byAge = entries.map((n) => ({ n, t: fs.statSync(join(this._keptDir, n)).mtimeMs })).sort((a, b) => a.t - b.t)
for (const { n } of byAge.slice(0, entries.length - KEEP_MAX)) { try { fs.unlinkSync(join(this._keptDir, n)) } catch {} }
}
} catch {}
}
// Install the kept copy of `oid` as this record's file, if we have one AND it hashes to `oid`. Same
// temp-then-rename discipline as _refetchOver: the current file is untouched unless the copy verifies.
async _fromKept (collection, rkey, oid) {
if (!this._overlay || typeof this._overlay.hashFile !== 'function' || !/^[0-9a-f]{16,}$/i.test(String(oid || ''))) return null
const src = join(this._keptDir, oid + EXT)
if (!fs.existsSync(src)) return null
const disk = this._pathFor(collection, rkey)
const tmp = disk + '.kept-' + String(oid).slice(0, 12)
try {
fs.mkdirSync(join(this._recordsDir, collection), { recursive: true })
fs.copyFileSync(src, tmp)
const h = await this._overlay.hashFile(tmp)
if (!h || h.contentHash !== oid) { try { fs.unlinkSync(tmp) } catch {}; return null }
if (fs.existsSync(disk)) await this._keep(disk)
fs.renameSync(tmp, disk)
try { await this._overlay.registerFile(this._overlayPath(collection, rkey), disk, { size: h.size }) } catch {}
try { console.log(JSON.stringify({ msg: 'ensureLocal: restored record bytes this node had kept (a write the view had lagged)', did: this._did, key: collection + '/' + rkey, oid: String(oid).slice(0, 16) })) } catch {}
return this.get(collection, rkey)
} catch { try { fs.unlinkSync(tmp) } catch {}; return null }
}
// Replace a SUPERSEDED local record file with the authoritative bytes for `oid`, without destroying
// what we hold until the replacement is verified. Fetches to a sibling temp path (reSeed:false so the
// overlay never indexes the temp), re-asserts the hash, then renames into place atomically and
// registers the FINAL path so this node serves the correct bytes afterwards. On any failure the
// existing file is left exactly as it was and null is returned. Returns the decoded record or null.
async _refetchOver (collection, rkey, oid) {
const disk = this._pathFor(collection, rkey)
const tmp = disk + '.incoming-' + String(oid).slice(0, 12)
try {
const r = await this._overlay.fetchFile(oid, { destPath: tmp, reSeed: false })
if (!r) { this._logFetchMiss(collection, rkey, oid, 'no peer served the authoritative bytes for a superseded local file'); return null }
// fetchFile verifies incrementally during transfer; re-assert before overwriting persistent state.
const h = await this._overlay.hashFile(tmp)
if (!h || h.contentHash !== oid) {
try { fs.unlinkSync(tmp) } catch {}
this._logFetchMiss(collection, rkey, oid, 'fetched bytes did not hash to the requested oid')
return null
}
// ⛔ KEEP WHAT WE REPLACE. "Local does not match the pointer" can mean we are BEHIND (a stale copy) or
// AHEAD (our own write; the view has not linearized its pointer yet). We cannot tell which, so the
// outgoing bytes are kept content-addressed and a later pointer to them resolves from _fromKept.
await this._keep(disk)
fs.renameSync(tmp, disk)
try { await this._overlay.registerFile(this._overlayPath(collection, rkey), disk, { size: h.size }) } catch {}
try {
console.log(JSON.stringify({ msg: 'ensureLocal: replaced superseded record bytes with the authoritative copy', did: this._did, key: collection + '/' + rkey, oid: String(oid).slice(0, 16) }))
} catch {}
return this.get(collection, rkey)
} catch (err) {
try { fs.unlinkSync(tmp) } catch {}
this._logFetchMiss(collection, rkey, oid, (err && err.message) || String(err))
return null
}
}
// A local file that does not match the converged pointer is an ANOMALY, not routine lag — it means
// this node has been resolving (and possibly signing) content the group did not agree on. Reported
// once per key+oid so a persistent case is visible without spamming the resolve loop.
_logStaleLocal (collection, rkey, oid, localOid) {
if (!this._staleSeen) this._staleSeen = new Set()
const k = collection + '/' + rkey + ':' + oid
if (this._staleSeen.has(k)) return
this._staleSeen.add(k)
try {
console.error(JSON.stringify({
msg: 'ensureLocal: LOCAL BYTES DO NOT MATCH THE CONVERGED POINTER — this node held a superseded copy; re-fetching the authoritative bytes. Until it succeeds this record reads as unresolved and will not be signed.',
did: this._did, key: collection + '/' + rkey, pointerOid: String(oid).slice(0, 16), localOid: localOid ? String(localOid).slice(0, 16) : null
}))
} catch {}
}
// Byte-resolution failures used to be swallowed entirely (`catch { return null }`), which made a
// record whose bytes never replicated indistinguishable from any ordinary miss. That matters because
// this path GATES SIGNING: the forward-only sign gate skips while a converged pointer's bytes are
// unresolved, so a permanent miss silently stops a co-owner signing at all, with no diagnosable
// cause. Observed on wj62 — one record unresolved for 75+ hours with nothing logged anywhere.
// Deduped per oid, because the resolve loop calls this once per record on every list().
// `fetchFile` returns null for two ROOT-DISTINCT reasons and the old message asserted only one of
// them ("no connected peer served this oid"), which is why a transport-level fault read as a content
// fault for days. overlay.js:307-315 waits peerWaitMs for a peer and bails if none ever attaches;
// otherwise it asks the attached peers and they may simply not hold it. Reporting `overlayPeers`
// separates those: 0 => the overlay channel never attached (a WIRING fault — look at attachProtocol);
// >0 => genuinely nobody serving that content (a DISTRIBUTION fault — look at the holder's index).
//
// Dedup is keyed on oid + peerCount, not oid alone. Deduping per-oid-per-process meant the line was
// emitted once at boot and never again, so a stale line was indistinguishable from a live failure and
// could not be used to judge whether a repair had worked. Keying on the peer count keeps the log quiet
// while still re-emitting the moment connectivity changes — which is exactly when the answer changes.
_logFetchMiss (collection, rkey, oid, reason) {
if (!this._fetchMisses) this._fetchMisses = new Set()
const peers = (this._overlay && typeof this._overlay.peerCount === 'number') ? this._overlay.peerCount : null
const k = String(oid)
const dedupKey = k + ':' + peers
if (this._fetchMisses.has(dedupKey)) return
this._fetchMisses.add(dedupKey)
try {
console.error(JSON.stringify({
msg: peers === 0
? 'overlay byte-resolution FAILED — NO overlay peers attached (transport/wiring fault, not content)'
: 'overlay byte-resolution FAILED — peers attached but none served this oid (distribution fault)',
did: this._did, key: `${collection}/${rkey}`, oid: k.slice(0, 16), overlayPeers: peers, reason
}))
} catch {}
}
async get (collection, rkey) {
const raw = await this.getRaw(collection, rkey)
return raw ? raw.record : null
}
// Raw stored bytes + decoded record. The bridge CIDs the EXACT bytes (never a re-encode) —
// so these must be the same bytes `put` wrote.
async getRaw (collection, rkey) {
let bytes
try { bytes = fs.readFileSync(this._pathFor(collection, rkey)) } catch (e) { if (e && e.code === 'ENOENT') return null; throw e }
return { bytes, record: dagCbor.decode(bytes) }
}
async del (collection, rkey) {
try { fs.unlinkSync(this._pathFor(collection, rkey)) } catch (e) { if (!e || e.code !== 'ENOENT') throw e }
// Record the deletion in the overlay's change feed so it propagates to peers (attachReplication).
if (this._overlay) { try { await this._overlay.unregisterFile(this._overlayPath(collection, rkey)) } catch {} }
try { fs.rmdirSync(join(this._recordsDir, collection)) } catch {} // best-effort: drop an empty collection dir
}
// Yield { rkey, record } for one collection, in key (byte) order.
async * list (collection) {
const dir = join(this._recordsDir, collection)
let names
try { names = fs.readdirSync(dir) } catch (e) { if (e && e.code === 'ENOENT') return; throw e }
for (const n of names.filter((x) => x.endsWith(EXT)).sort(byteOrder)) {
const bytes = fs.readFileSync(join(dir, n))
yield { rkey: n.slice(0, -EXT.length), record: dagCbor.decode(bytes) }
}
}
// All records as { key, bytes, record } in key (byte) order — the sole input to the MST build.
async records () {
const out = []
let collections
try { collections = fs.readdirSync(this._recordsDir) } catch (e) { if (e && e.code === 'ENOENT') return out; throw e }
for (const collection of collections) {
const dir = join(this._recordsDir, collection)
let st
try { st = fs.statSync(dir) } catch { continue }
if (!st.isDirectory()) continue
for (const n of fs.readdirSync(dir)) {
if (!n.endsWith(EXT)) continue
const bytes = fs.readFileSync(join(dir, n))
out.push({ key: `${collection}/${n.slice(0, -EXT.length)}`, bytes, record: dagCbor.decode(bytes) })
}
}
out.sort((a, b) => byteOrder(a.key, b.key))
return out
}
// Persist the commit head ({ rev, commitCid, dataCid }) — a tiny file, NOT a record, so it never
// appears in records()/the MST. This is what keeps the head stable across restart: the rev is
// RESTORED, not re-minted (re-minting forges a phantom commit — see RepoManager.init).
async saveHead (head) {
fs.mkdirSync(this._root, { recursive: true })
fs.writeFileSync(this._headFile, JSON.stringify(head))
}
async loadHead () {
try { return JSON.parse(fs.readFileSync(this._headFile, 'utf8')) } catch (e) { if (e && e.code === 'ENOENT') return null; throw e }
}
async close () {}
_overlayPath (collection, rkey) { return `/records/${collection}/${rkey}${EXT}` }
}