| … | … | @@ -22,6 +22,7 @@ import { join } from '#node:path' |
| 22 | 22 | import { tidNow } from './tid.js' |
| 23 | 23 | |
| 24 | 24 | const EXT = '.dagcbor' |
| 25 | const KEEP_MAX = 256 // superseded/ entries kept (see _keep) |
| 25 | 26 | const byteOrder = (a, b) => b4a.compare(b4a.from(a), b4a.from(b)) // match Hyperbee's key comparison exactly |
| 26 | 27 | // atproto collection NSIDs + rkeys are single path segments (no '/'); reject anything else so a |
| 27 | 28 | // record key can never escape the records/ dir or collide with a parent path. |
| … | … | @@ -34,6 +35,9 @@ export class OverlayRepoStore { |
| 34 | 35 | if (typeof root !== 'string') throw new Error('OverlayRepoStore: root dir path required') |
| 35 | 36 | this._root = root |
| 36 | 37 | this._recordsDir = join(root, 'records') |
| 38 | // Bytes a re-fetch REPLACED, kept content-addressed (<oid>.dagcbor) beside records/ — never inside it, |
| 39 | // where records() would read the dir as a collection. See _keep / _fromKept. |
| 40 | this._keptDir = join(root, 'superseded') |
| 37 | 41 | this._headFile = join(root, 'head.json') |
| 38 | 42 | this._overlay = opts.overlay || null |
| 39 | 43 | this._did = opts.did || null |
| … | … | @@ -163,6 +167,10 @@ export class OverlayRepoStore { |
| 163 | 167 | // unchanged wherever verification is impossible. |
| 164 | 168 | if (!h || h.oid === oid) return local // verified against the converged pointer — the common path |
| 165 | 169 | this._logStaleLocal(collection, rkey, oid, h ? h.oid : null) |
| 170 | // The pointer may name bytes THIS node once held and replaced (a newer local write the lagging view |
| 171 | // had overwritten — loser-club 2026-10-14). Kept copies are verified and installed before the network. |
| 172 | const kept = await this._fromKept(collection, rkey, oid) |
| 173 | if (kept != null) return kept |
| 166 | 174 | const fetched = await this._refetchOver(collection, rkey, oid) |
| 167 | 175 | // Deliberately null, NOT the stale record, when the authoritative bytes cannot be obtained. |
| 168 | 176 | // Returning the superseded copy is what published wrong content: this record then reads as |
| … | … | @@ -174,6 +182,8 @@ export class OverlayRepoStore { |
| 174 | 182 | return fetched |
| 175 | 183 | } |
| 176 | 184 | if (!this._overlay || !oid) return null |
| 185 | const kept = await this._fromKept(collection, rkey, oid) |
| 186 | if (kept != null) return kept |
| 177 | 187 | const disk = this._pathFor(collection, rkey) |
| 178 | 188 | try { |
| 179 | 189 | fs.mkdirSync(join(this._recordsDir, collection), { recursive: true }) |
| … | … | @@ -183,6 +193,45 @@ export class OverlayRepoStore { |
| 183 | 193 | return this.get(collection, rkey) |
| 184 | 194 | } |
| 185 | 195 | |
| 196 | // Keep the bytes currently at `disk` under superseded/<oid>.dagcbor before they are replaced. Bounded: |
| 197 | // only the newest KEEP_MAX entries stay (records are small; this exists to bridge a view that lags a |
| 198 | // local write by seconds, not to archive history). Best-effort — a failure here never blocks the repair. |
| 199 | async _keep (disk) { |
| 200 | try { |
| 201 | const h = await this._overlay.hashFile(disk) |
| 202 | if (!h || !h.contentHash) return |
| 203 | fs.mkdirSync(this._keptDir, { recursive: true }) |
| 204 | const dest = join(this._keptDir, h.contentHash + EXT) |
| 205 | if (!fs.existsSync(dest)) fs.copyFileSync(disk, dest) |
| 206 | const entries = fs.readdirSync(this._keptDir).filter((n) => n.endsWith(EXT)) |
| 207 | if (entries.length > KEEP_MAX) { |
| 208 | const byAge = entries.map((n) => ({ n, t: fs.statSync(join(this._keptDir, n)).mtimeMs })).sort((a, b) => a.t - b.t) |
| 209 | for (const { n } of byAge.slice(0, entries.length - KEEP_MAX)) { try { fs.unlinkSync(join(this._keptDir, n)) } catch {} } |
| 210 | } |
| 211 | } catch {} |
| 212 | } |
| 213 | |
| 214 | // Install the kept copy of `oid` as this record's file, if we have one AND it hashes to `oid`. Same |
| 215 | // temp-then-rename discipline as _refetchOver: the current file is untouched unless the copy verifies. |
| 216 | async _fromKept (collection, rkey, oid) { |
| 217 | if (!this._overlay || typeof this._overlay.hashFile !== 'function' || !/^[0-9a-f]{16,}$/i.test(String(oid || ''))) return null |
| 218 | const src = join(this._keptDir, oid + EXT) |
| 219 | if (!fs.existsSync(src)) return null |
| 220 | const disk = this._pathFor(collection, rkey) |
| 221 | const tmp = disk + '.kept-' + String(oid).slice(0, 12) |
| 222 | try { |
| 223 | fs.mkdirSync(join(this._recordsDir, collection), { recursive: true }) |
| 224 | fs.copyFileSync(src, tmp) |
| 225 | const h = await this._overlay.hashFile(tmp) |
| 226 | if (!h || h.contentHash !== oid) { try { fs.unlinkSync(tmp) } catch {}; return null } |
| 227 | if (fs.existsSync(disk)) await this._keep(disk) |
| 228 | fs.renameSync(tmp, disk) |
| 229 | try { await this._overlay.registerFile(this._overlayPath(collection, rkey), disk, { size: h.size }) } catch {} |
| 230 | 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 {} |
| 231 | return this.get(collection, rkey) |
| 232 | } catch { try { fs.unlinkSync(tmp) } catch {}; return null } |
| 233 | } |
| 234 | |
| 186 | 235 | // Replace a SUPERSEDED local record file with the authoritative bytes for `oid`, without destroying |
| 187 | 236 | // what we hold until the replacement is verified. Fetches to a sibling temp path (reSeed:false so the |
| 188 | 237 | // overlay never indexes the temp), re-asserts the hash, then renames into place atomically and |
| … | … | @@ -201,6 +250,10 @@ export class OverlayRepoStore { |
| 201 | 250 | this._logFetchMiss(collection, rkey, oid, 'fetched bytes did not hash to the requested oid') |
| 202 | 251 | return null |
| 203 | 252 | } |
| 253 | // ⛔ KEEP WHAT WE REPLACE. "Local does not match the pointer" can mean we are BEHIND (a stale copy) or |
| 254 | // AHEAD (our own write; the view has not linearized its pointer yet). We cannot tell which, so the |
| 255 | // outgoing bytes are kept content-addressed and a later pointer to them resolves from _fromKept. |
| 256 | await this._keep(disk) |
| 204 | 257 | fs.renameSync(tmp, disk) |
| 205 | 258 | try { await this._overlay.registerFile(this._overlayPath(collection, rkey), disk, { size: h.size }) } catch {} |
| 206 | 259 | try { |