// retention.js — the last three hooks of DESIGN §7.6: retain, knownRev, evict. Plus the byte budget.
//
// WHAT THIS OWNS. Everything downstream of verification: writing a verified foreign record to disk,
// folding it into the appview so it renders, remembering enough to answer the skip-test, and enforcing
// the Tier-B budget by evicting least-recently-used records. §7.6 specifies all of this; until now none
// of it existed.
//
// FOREIGN RECORDS ARE FILES, NOT CORES. A retained bsky account costs a directory under an
// OverlayRepoStore, not a Hypercore (hyper-proto-peer/lib/device-node.js:221 uses the same store for
// its own repo). That is what makes "one hop from our records" affordable — 500 followed accounts are
// 500 directories, not 500 cores with RocksDB overhead each.
//
// THE RETENTION INDEX IS BOOKKEEPING, NOT AUTHORITY — and §7.5's invariant is the reason this is safe.
// It records { rev, size, lastSeen } per retained record and gates exactly two things: the skip-test and
// eviction order. Both are CACHE MANAGEMENT. It never gates a write, a signature, or a serve decision:
// lose the whole index and the system is still correct, merely slower (every pointer re-fetches, and
// eviction restarts from a clean slate). That is the test §7.5 demands, and this passes it.
//
// ⛔ BYTE-EXACTNESS IS GUARDED, NOT ASSUMED. OverlayRepoStore.put() takes a DECODED record and
// re-encodes it (`dagCbor.encode(record)`), but what we verified was the ORIGINAL bytes and their CID.
// DAG-CBOR is deterministic so a canonical record round-trips identically — but "should" is not a
// fidelity argument, and §13's whole point is that an encoding divergence federates as corruption
// rather than failing loudly. So we re-encode, re-CID, and REFUSE to retain on any mismatch. A record
// we cannot store byte-exactly is one we decline to hold, not one we quietly mangle.
//
// PURE + INJECTED. The store opener, the appview, the index and the clock are all parameters, so the
// tiering, the skip-test, the budget and the eviction order are testable with no Corestore and no disk.
import * as dagCbor from '@ipld/dag-cbor'
import { cidForBytesSync } from 'atproto-repo/lib/cid.js'
import { isLossySubset } from './appview.js'
// Index key for one retained foreign record.
export function retentionKey ({ did, collection, rkey }) {
return `${did}/${collection}/${rkey}`
}
// An in-memory retention index. Production wraps a Hyperbee (see createBeeIndex below); this is the
// reference implementation and what the tests drive.
export function createMemoryIndex () {
const m = new Map()
return {
async get (k) { return m.get(k) || null },
async put (k, v) { m.set(k, v) },
async del (k) { return m.delete(k) },
async * entries () { for (const [k, v] of m) yield { key: k, value: v } },
get size () { return m.size }
}
}
// A Hyperbee-backed retention index. `bee` should be a sub() dedicated to retention so it never
// collides with the appview's own namespaces (Hyperbee subs do not dedup — see the
// reference_hyperbee-subs-no-alwaysduplicate note; alwaysDuplicate:false belongs on the ROOT bee).
export function createBeeIndex (bee) {
return {
async get (k) { const n = await bee.get(k); return n ? n.value : null },
async put (k, v) { await bee.put(k, v) },
async del (k) { await bee.del(k); return true },
async * entries () { for await (const n of bee.createReadStream()) yield { key: n.key.toString(), value: n.value } }
}
}
export function createRetentionHooks ({
openStore, // async (did) -> OverlayRepoStore (ready)
appview = null, // optional: something with applyCommit(did, ops)
index, // a retention index (memory or bee)
budgetBytes = null, // §7.6 Tier-B byte budget; null = unbounded (caller accepts the risk)
// ⛔ Is this DID one hpp AUTHORS for? §7.6 scopes retention to FOREIGN records, and I did not enforce
// that — so Constellation, which reports ANY record linking to our subjects INCLUDING our own replies
// to our own posts, fed our own records back through this pipeline. They were re-encoded over hpp's
// canonical files, dropped `langs`, and wedged the forward-only sign gate: two of wj62's replies never
// federated. Default is "everything is foreign" so an unconfigured caller cannot silently clobber.
isForeign = () => true,
now = () => Date.now(),
onEvent = null // (event, detail) -> void, for observability
} = {}) {
if (typeof openStore !== 'function') throw new Error('createRetentionHooks requires openStore')
if (!index) throw new Error('createRetentionHooks requires an index')
const stats = { retained: 0, reencodeRefused: 0, lossyRefused: 0, skippedOwn: 0, evicted: 0, evictedForBudget: 0, bytes: 0, storeErrors: 0 }
const emit = (ev, detail) => { if (onEvent) onEvent(ev, detail) }
// Recompute the live byte total from the index. Cheap enough at Tier-B scale and avoids a counter
// drifting out of sync with reality across restarts.
async function totalBytes () {
let sum = 0
for await (const { value } of index.entries()) sum += (value && value.size) || 0
return sum
}
// ---- knownRev -----------------------------------------------------------------------------------
// Per-RECORD, deliberately not per-repo. Tier B is a partial cache: we hold some records of a foreign
// repo, never all of them, so a repo-level head would claim a completeness we do not have.
async function knownRev (pointer) {
const entry = await index.get(retentionKey(pointer))
return entry && typeof entry.rev === 'string' ? entry.rev : null
}
// ---- retainRecord -------------------------------------------------------------------------------
async function retainRecord (record, pointer) {
if (!record || !record.bytes) return false
// NEVER write into a repo hpp authors. hpp is the authority for those records: PointerSubstrate
// hashes the bytes IT writes and commits that hash to the autobase, so any differing re-encode by a
// second writer invalidates a converged pointer and the signer can no longer resolve it.
if (!isForeign(pointer.did)) {
stats.skippedOwn++
emit('skipped-own-did', { pointer })
return false
}
// Decode once, then prove the round-trip before writing anything.
let value
try {
value = dagCbor.decode(record.bytes)
} catch {
stats.reencodeRefused++
emit('decode-failed', { pointer })
return false
}
const reencoded = dagCbor.encode(value)
if (cidForBytesSync(reencoded).toString() !== record.recordCid.toString()) {
// The store would persist bytes that no longer match the CID we verified. Refuse loudly.
stats.reencodeRefused++
emit('reencode-mismatch', { pointer, cid: record.recordCid.toString() })
return false
}
let store
try {
store = await openStore(pointer.did)
// NO LOSSY OVERWRITE — the same rule appview.putPost adopted in 0.2.50, which this path bypassed.
// The byte-exactness guard above only proves our re-encode matches the bytes we FETCHED; it says
// nothing about what is already on disk, which is the only comparison that matters when another
// writer owns the file. Refuse only a strict subset: a changed value or an added field still
// writes (freezing at first-seen version is its own bug), dropping a field is refused.
const existing = await store.getRaw(pointer.collection, pointer.rkey)
if (existing != null && isLossySubset(value, existing.record)) {
stats.lossyRefused++
emit('lossy-refused', {
pointer,
oursBytes: reencoded.length,
canonicalBytes: existing.bytes.length
})
return false
}
await store.put(pointer.collection, pointer.rkey, value)
} catch (err) {
stats.storeErrors++
emit('store-failed', { pointer, error: (err && err.message) || String(err) })
return false
}
// Fold into the appview so the record actually renders. Synthesised as a commit op because that is
// the shape the existing ingest path already understands — no second indexing route.
if (appview) {
try {
await appview.applyCommit(pointer.did, [{
collection: pointer.collection,
rkey: pointer.rkey,
action: 'create',
record: value
}])
} catch (err) {
// The bytes are already safely on disk; a failed index fold is a render gap, not data loss, and
// a re-index will heal it. Never unwind the write for it.
emit('index-failed', { pointer, error: (err && err.message) || String(err) })
}
}
const size = reencoded.length
const prev = await index.get(retentionKey(pointer))
await index.put(retentionKey(pointer), {
did: pointer.did,
collection: pointer.collection,
rkey: pointer.rkey,
// A backfilled pointer carries rev: null (Constellation does not report it). Keep any rev we
// already had rather than downgrading to null, or the skip-test would forget known state.
rev: pointer.rev || (prev && prev.rev) || null,
size,
retainedAt: prev ? prev.retainedAt : now(),
lastSeen: now()
})
stats.retained++
stats.bytes = await totalBytes()
emit('retained', { pointer, size })
if (budgetBytes !== null) await enforceBudget()
return true
}
// ---- evictRecord --------------------------------------------------------------------------------
// Always safe: the authoritative copy lives in the author's PDS and is re-fetchable. That asymmetry
// is what makes §7.6 a budgeting problem rather than a correctness one.
async function evictRecord (pointer) {
const key = retentionKey(pointer)
const entry = await index.get(key)
try {
const store = await openStore(pointer.did)
await store.del(pointer.collection, pointer.rkey)
} catch (err) {
// Fall through: drop the index entry regardless, so a store that has already lost the file does
// not leave a permanently un-evictable row inflating the budget.
stats.storeErrors++
emit('evict-store-failed', { pointer, error: (err && err.message) || String(err) })
}
if (appview) {
try {
await appview.applyCommit(pointer.did, [{
collection: pointer.collection,
rkey: pointer.rkey,
action: 'delete'
}])
} catch (err) {
emit('index-failed', { pointer, error: (err && err.message) || String(err) })
}
}
await index.del(key)
if (entry) {
stats.evicted++
stats.bytes = await totalBytes()
}
emit('evicted', { pointer })
return true
}
// ---- the §7.6 byte budget -----------------------------------------------------------------------
// LRU by lastSeen. Evicting past the budget cannot break correctness — a Tier-B miss degrades to a
// lens-import fetch, which already fails soft — so this is allowed to be aggressive.
async function enforceBudget () {
if (budgetBytes === null) return 0
let total = await totalBytes()
if (total <= budgetBytes) return 0
const all = []
for await (const { value } of index.entries()) if (value) all.push(value)
all.sort((a, b) => (a.lastSeen || 0) - (b.lastSeen || 0)) // oldest touch first
let evicted = 0
for (const entry of all) {
if (total <= budgetBytes) break
await evictRecord({ did: entry.did, collection: entry.collection, rkey: entry.rkey })
total -= entry.size || 0
evicted++
stats.evictedForBudget++
}
if (evicted > 0) emit('budget-enforced', { evicted, budgetBytes, total })
stats.bytes = total
return evicted
}
// Mark a record as freshly used, so LRU reflects RENDER activity and not merely write order. The
// caller should call this when a retained record is actually served.
async function touch (pointer) {
const key = retentionKey(pointer)
const entry = await index.get(key)
if (!entry) return false
entry.lastSeen = now()
await index.put(key, entry)
return true
}
return { retainRecord, knownRev, evictRecord, enforceBudget, touch, totalBytes, stats }
}