| … | … | @@ -272,6 +272,9 @@ export class Feed { |
| 272 | 272 | async resolve (collection, name) { return this._log.resolve(collection, name) } |
| 273 | 273 | async get (collection, name) { return this._log.get(collection, name) } |
| 274 | 274 | async list (collection = null) { return this._log.list(collection) } |
| 275 | // Live pointers, WITHOUT fetching any bytes — see RecordLog.pointers. This is what lets the | |
| 276 | // mirror decide which records to go and get instead of getting all of them at once. | |
| 277 | async pointers (collection = null) { return this._log.pointers(collection) } | |
| 275 | 278 | async unresolved (collection = null) { return this._log.unresolved(collection) } |
| 276 | 279 | |
| 277 | 280 | // Admitted WRITER cores, excluding any whose device has been revoked. |
| … | … | @@ -120,6 +120,31 @@ export class RecordLog { |
| 120 | 120 | ({ key, collection, name, record })) |
| 121 | 121 | } |
| 122 | 122 | |
| 123 | /** | |
| 124 | * Every live pointer, WITHOUT resolving any of them. | |
| 125 | * | |
| 126 | * ⛔ The one enumeration that does not fetch. `list()` and `unresolved()` both go through | |
| 127 | * `_scan`, which calls `resolve()` on every record — so both of them attempt a network fetch for | |
| 128 | * everything missing, all at once. That is right for a reader who wants the records, and wrong | |
| 129 | * for anything that needs to decide WHICH bytes to go and get: you cannot bound a fetch you have | |
| 130 | * already made by asking the question. | |
| 131 | * | |
| 132 | * Tombstones are omitted: a deleted record has no bytes to hold, and offering it here would send | |
| 133 | * a mirror looking for something that is positively known not to exist. | |
| 134 | */ | |
| 135 | async pointers (collection = null) { | |
| 136 | const prefix = collection ? collection + SEP : '' | |
| 137 | const map = await this.kv.list(prefix) | |
| 138 | const out = [] | |
| 139 | for (const key of Object.keys(map).filter((k) => splitKey(k) !== null).sort(byteOrder)) { | |
| 140 | const pointer = map[key] | |
| 141 | if (pointer == null || isTombstone(pointer)) continue | |
| 142 | const { collection: c, name } = splitKey(key) | |
| 143 | out.push({ key, collection: c, name, pointer }) | |
| 144 | } | |
| 145 | return out | |
| 146 | } | |
| 147 | ||
| 123 | 148 | // Pointers we hold whose bytes we cannot obtain. Never merge this into list() — a record that is |
| 124 | 149 | // silently missing from a listing is indistinguishable from one that was never written, and that |
| 125 | 150 | // is the ambiguity this whole layer exists to refuse. |
| … | … | @@ -0,0 +1,115 @@ |
| 1 | // mirror.js — hold the bytes of feeds you follow, before anyone asks to read them. | |
| 2 | // | |
| 3 | // ⛔ THE OUTAGE, measured 2026-09-08. tabertron held five pointers from ashatron's feed and could | |
| 4 | // read none of them. ashatron is a laptop, it went offline, and it was the only machine holding the | |
| 5 | // bytes. Nothing was broken: a record's bytes are fetched only when a reader asks for them, nobody | |
| 6 | // had asked, so no second copy existed. The surface reported it honestly — "the log arrived, the | |
| 7 | // bytes did not" — and all five resolved the moment ashatron came back. | |
| 8 | // | |
| 9 | // ★ SO THIS IS ONLY A TRIGGER. Two things already work and are deliberately NOT rebuilt here: | |
| 10 | // | |
| 11 | // 1. A fetch WRITES the record into that feed's working tree, so the copy is durable. | |
| 12 | // 2. hyper-overlay RE-SEEDS on fetch — "register the verified blob locally so this node now | |
| 13 | // SERVES these chunks to other peers too" — so holding implies serving, automatically. | |
| 14 | // | |
| 15 | // Reading a record once already makes this node a mirror of it. Nothing reads unprompted, and that | |
| 16 | // is the entire gap. hyper-overlay's `mirrorSpool` is NOT needed for us: it exists for "a node with | |
| 17 | // no working tree of its own", and we have one per feed. | |
| 18 | // | |
| 19 | // ⛔ FOLLOWED FEEDS ONLY. `peer.timeline` already excludes peeked feeds because including them | |
| 20 | // "would make reading a stranger's page indistinguishable from following them". Mirroring a peek | |
| 21 | // would be that same collapse with disk attached — every profile glanced at becomes a permanent | |
| 22 | // copy. Following is the decision to keep someone's bytes; peeking is not. | |
| 23 | // | |
| 24 | // Design: hyper-social/docs/HOLDING-AND-TELLING.md | |
| 25 | ||
| 26 | const DEFAULT_CONCURRENCY = 4 | |
| 27 | ||
| 28 | /** | |
| 29 | * @param feeds () => [{ key, feed }] — the FOLLOWED feeds, supplied by the caller so this | |
| 30 | * module never has to know how following is decided. | |
| 31 | * @param concurrency files in flight. ⚠ Bounds FILES, not chunks: the layer below already holds an | |
| 32 | * RTT-scaled inflight window per peer, and re-implementing that would be worse | |
| 33 | * than what exists. What is unbounded is the file count, and a freshly followed | |
| 34 | * feed is all of them at once. | |
| 35 | */ | |
| 36 | export function createMirror ({ feeds, concurrency = DEFAULT_CONCURRENCY, log = () => {} }) { | |
| 37 | if (typeof feeds !== 'function') throw new Error('createMirror: feeds() is required') | |
| 38 | ||
| 39 | let stopped = false | |
| 40 | let running = false | |
| 41 | ||
| 42 | async function mirrorFeed (key, feed) { | |
| 43 | // ⛔ `pointers()` and not `list()` or `unresolved()`. Both of those RESOLVE every record, which | |
| 44 | // means both of them fetch — unbounded, all at once. Enumerating without fetching is what lets | |
| 45 | // this pass be bounded at all. | |
| 46 | let rows | |
| 47 | try { | |
| 48 | rows = await feed.pointers() | |
| 49 | } catch (err) { | |
| 50 | // A feed whose view is not ready yet is not a failure of the pass. It will be there next time. | |
| 51 | log({ msg: 'mirror: could not list feed', feed: key, error: err.message }) | |
| 52 | return { fetched: 0, failed: 0 } | |
| 53 | } | |
| 54 | ||
| 55 | // Ask the STORE, not the log: `get` is a local read and answers without touching the network, | |
| 56 | // so a record already on disk costs nothing and is skipped before any slot is taken. | |
| 57 | const missing = [] | |
| 58 | for (const r of rows) { | |
| 59 | if (stopped) break | |
| 60 | try { | |
| 61 | if ((await feed.store.get(r.collection, r.name)) == null) missing.push(r) | |
| 62 | } catch { missing.push(r) } | |
| 63 | } | |
| 64 | ||
| 65 | let fetched = 0 | |
| 66 | let failed = 0 | |
| 67 | let next = 0 | |
| 68 | ||
| 69 | const worker = async () => { | |
| 70 | while (!stopped) { | |
| 71 | const i = next++ | |
| 72 | if (i >= missing.length) return | |
| 73 | const r = missing[i] | |
| 74 | try { | |
| 75 | // ⛔ One unfetchable record must not end the pass. A peer that is away is not a peer that | |
| 76 | // is gone: the record stays unresolved, the surface says so, and the next pass retries. | |
| 77 | const got = await feed.store.ensureLocal(r.collection, r.name, r.pointer && r.pointer.$oid) | |
| 78 | if (got == null) failed++ | |
| 79 | else fetched++ | |
| 80 | } catch { | |
| 81 | failed++ | |
| 82 | } | |
| 83 | } | |
| 84 | } | |
| 85 | ||
| 86 | await Promise.all(Array.from({ length: Math.min(concurrency, missing.length) }, worker)) | |
| 87 | if (fetched || failed) log({ msg: 'mirror', feed: key, fetched, failed, missing: missing.length }) | |
| 88 | return { fetched, failed } | |
| 89 | } | |
| 90 | ||
| 91 | return { | |
| 92 | /** One sweep over every followed feed. Safe to call concurrently — a second call is a no-op. */ | |
| 93 | async pass () { | |
| 94 | if (stopped || running) return { fetched: 0, failed: 0 } | |
| 95 | running = true | |
| 96 | let fetched = 0 | |
| 97 | let failed = 0 | |
| 98 | try { | |
| 99 | for (const { key, feed } of feeds()) { | |
| 100 | if (stopped) break | |
| 101 | const r = await mirrorFeed(key, feed) | |
| 102 | fetched += r.fetched | |
| 103 | failed += r.failed | |
| 104 | } | |
| 105 | } finally { | |
| 106 | running = false | |
| 107 | } | |
| 108 | return { fetched, failed } | |
| 109 | }, | |
| 110 | ||
| 111 | stop () { stopped = true } | |
| 112 | } | |
| 113 | } | |
| 114 | ||
| 115 | export default { createMirror } |
| … | … | @@ -29,6 +29,7 @@ import { Namespace } from './namespace.js' |
| 29 | 29 | import { loadNamespace, saveNamespace, namespaceSlot, loadRooms, saveRooms } from './device.js' |
| 30 | 30 | import { Room } from './room.js' |
| 31 | 31 | import { readAttested } from './attestation.js' |
| 32 | import { createMirror } from './mirror.js' | |
| 32 | 33 | |
| 33 | 34 | const hex = (k) => (typeof k === 'string' ? k : b4a.toString(k, 'hex')) |
| 34 | 35 | |
| … | … | @@ -217,6 +218,33 @@ export class FavelaNode { |
| 217 | 218 | for (const name of await this.peer.following()) { |
| 218 | 219 | try { await this.hold(name) } catch {} |
| 219 | 220 | } |
| 221 | ||
| 222 | // ⛔ HOLD THE BYTES OF FEEDS WE FOLLOW, before anyone asks to read them. | |
| 223 | // | |
| 224 | // MEASURED 2026-09-08: this node held five pointers from ashatron and could read none of them. | |
| 225 | // ashatron is a laptop, it went offline, and it was the only machine with the bytes — because a | |
| 226 | // record's bytes are fetched only when a reader asks, and nobody had. All five resolved the | |
| 227 | // moment it came back. | |
| 228 | // | |
| 229 | // Following is the decision to KEEP someone's bytes; peeking is not, which is why this reads | |
| 230 | // `following()` rather than every readable feed. `peer.timeline` draws the same line for the | |
| 231 | // same reason. See hyper-social/docs/HOLDING-AND-TELLING.md. | |
| 232 | this._mirror = createMirror({ | |
| 233 | feeds: () => { | |
| 234 | const out = [] | |
| 235 | for (const [name, feed] of this.peer.feeds) { | |
| 236 | if (this.peer.peeked.has(name) || this.peer._blocked.has(name)) continue | |
| 237 | out.push({ key: name, feed }) | |
| 238 | } | |
| 239 | return out | |
| 240 | }, | |
| 241 | // ⚠ No logger on this class, and a mirror pass must not invent one. Failures surface where | |
| 242 | // they already do: a record whose bytes did not arrive stays UNRESOLVED and the surface says so. | |
| 243 | }) | |
| 244 | // ⚠ NOT awaited. A first sweep of a large followed feed must not sit inside boot — the daemon | |
| 245 | // has to be answering its socket while this runs. Failures are logged and retried next pass. | |
| 246 | this._mirror.pass().catch(() => {}) | |
| 247 | ||
| 220 | 248 | return this |
| 221 | 249 | } |
| 222 | 250 | |
| … | … | @@ -694,6 +722,9 @@ export class FavelaNode { |
| 694 | 722 | } |
| 695 | 723 | |
| 696 | 724 | async close () { |
| 725 | // Stop the mirror before anything it touches is torn down — a pass mid-fetch holding a | |
| 726 | // corestore session while the store closes throws into nobody's hands. | |
| 727 | if (this._mirror) this._mirror.stop() | |
| 697 | 728 | // ⚠ Settle in-flight peeks FIRST. A background open still holding a corestore session while |
| 698 | 729 | // this tears the store down throws "RocksDB session is closed" into nobody's hands — the open |
| 699 | 730 | // has no caller left to reject to. In production these settle promptly: `hold()` awaits a feed |
| … | … | @@ -1,6 +1,6 @@ |
| 1 | 1 | { |
| 2 | 2 | "name": "favela-peer", |
| 3 | "version": "0.0.38", | |
| 3 | "version": "0.0.39", | |
| 4 | 4 | "description": "The favela daemon — feeds and documents of markdown records over Hypercore. No atproto, no ActivityPub.", |
| 5 | 5 | "type": "module", |
| 6 | 6 | "scripts": { |
| … | … | @@ -0,0 +1,133 @@ |
| 1 | // mirror.test.js — hold the bytes of feeds you follow, before anyone asks to read them. | |
| 2 | // | |
| 3 | // ⛔ THE OUTAGE THIS CLOSES, measured 2026-09-08. tabertron held five pointers from ashatron and | |
| 4 | // could read none of them: ashatron is a laptop, it went offline, and it was the only machine | |
| 5 | // holding the bytes. Nothing was broken — bytes are fetched only when a reader asks, nobody had | |
| 6 | // asked, so no second copy existed. The surface said so honestly: "the log arrived, the bytes did | |
| 7 | // not." | |
| 8 | // | |
| 9 | // ★ AND THE FIX IS ONLY A TRIGGER. Two things already work and must not be rebuilt: | |
| 10 | // - a fetch WRITES the record into this feed's working tree, so the copy is durable | |
| 11 | // - hyper-overlay RE-SEEDS on fetch, so this node then serves those bytes to others | |
| 12 | // So reading a record once already makes you a mirror of it. Nothing reads unprompted; that is all | |
| 13 | // that is missing. See hyper-social/docs/HOLDING-AND-TELLING.md. | |
| 14 | // | |
| 15 | // ⛔ FOLLOWED FEEDS ONLY. `timeline` already excludes peeked feeds because putting them in "would | |
| 16 | // make reading a stranger's page indistinguishable from following them". Mirroring a peek would be | |
| 17 | // that same collapse with disk attached: every profile you glanced at becomes a permanent copy. | |
| 18 | ||
| 19 | import test from 'brittle' | |
| 20 | import { createMirror } from '../lib/mirror.js' | |
| 21 | ||
| 22 | // A feed with pointers and a store that knows what is local. `pointers()` never fetches — that is | |
| 23 | // the whole reason it exists, since list() and unresolved() both resolve every record. | |
| 24 | function fakeFeed (names, { local = [], fail = [], onFetch = () => {} } = {}) { | |
| 25 | const localSet = new Set(local) | |
| 26 | return { | |
| 27 | async pointers () { | |
| 28 | return names.map((n) => ({ key: 'post/' + n, collection: 'post', name: n, pointer: { $oid: 'oid-' + n, size: 1 } })) | |
| 29 | }, | |
| 30 | store: { | |
| 31 | async get (_c, n) { return localSet.has(n) ? { body: n } : null }, | |
| 32 | async ensureLocal (_c, n) { | |
| 33 | onFetch(n) | |
| 34 | if (fail.includes(n)) throw new Error('no peer has ' + n) | |
| 35 | localSet.add(n) | |
| 36 | return { body: n } | |
| 37 | } | |
| 38 | } | |
| 39 | } | |
| 40 | } | |
| 41 | ||
| 42 | test('★★ it fetches records nobody asked to read', async (t) => { | |
| 43 | const fetched = [] | |
| 44 | const feed = fakeFeed(['a', 'b', 'c'], { onFetch: (n) => fetched.push(n) }) | |
| 45 | const m = createMirror({ feeds: () => [{ key: 'f1', feed }], log: () => {} }) | |
| 46 | ||
| 47 | await m.pass() | |
| 48 | ||
| 49 | t.alike(fetched.sort(), ['a', 'b', 'c'], 'all three, with no reader involved') | |
| 50 | }) | |
| 51 | ||
| 52 | test('★ a record already on disk is not fetched again', async (t) => { | |
| 53 | const fetched = [] | |
| 54 | const feed = fakeFeed(['a', 'b'], { local: ['a'], onFetch: (n) => fetched.push(n) }) | |
| 55 | const m = createMirror({ feeds: () => [{ key: 'f1', feed }], log: () => {} }) | |
| 56 | ||
| 57 | await m.pass() | |
| 58 | ||
| 59 | t.alike(fetched, ['b'], 'only the missing one') | |
| 60 | }) | |
| 61 | ||
| 62 | // ⚠ The layer below already bounds CHUNKS — the scheduler holds an RTT-scaled inflight window per | |
| 63 | // peer. What is unbounded is the number of FILES, and a freshly followed feed is all of them. | |
| 64 | test('★★ never more than `concurrency` fetches in flight', async (t) => { | |
| 65 | let now = 0 | |
| 66 | let peak = 0 | |
| 67 | const names = Array.from({ length: 20 }, (_, i) => 'r' + i) | |
| 68 | const feed = fakeFeed(names, { onFetch: () => {} }) | |
| 69 | // Wrap ensureLocal to hold the slot open across a tick so overlap is observable. | |
| 70 | const inner = feed.store.ensureLocal | |
| 71 | feed.store.ensureLocal = async (c, n) => { | |
| 72 | now++; peak = Math.max(peak, now) | |
| 73 | await new Promise((r) => setTimeout(r, 5)) | |
| 74 | now-- | |
| 75 | return inner.call(feed.store, c, n) | |
| 76 | } | |
| 77 | ||
| 78 | const m = createMirror({ feeds: () => [{ key: 'f1', feed }], concurrency: 3, log: () => {} }) | |
| 79 | await m.pass() | |
| 80 | ||
| 81 | t.is(peak, 3, 'the cap held') | |
| 82 | t.absent(now, 'and everything drained') | |
| 83 | }) | |
| 84 | ||
| 85 | test('⛔ one unfetchable record does not stop the rest', async (t) => { | |
| 86 | const fetched = [] | |
| 87 | const feed = fakeFeed(['a', 'b', 'c'], { fail: ['b'], onFetch: (n) => fetched.push(n) }) | |
| 88 | const m = createMirror({ feeds: () => [{ key: 'f1', feed }], log: () => {} }) | |
| 89 | ||
| 90 | await m.pass() | |
| 91 | ||
| 92 | t.alike(fetched.sort(), ['a', 'b', 'c'], 'every one was attempted') | |
| 93 | // b stays missing and will be retried on the next pass — a peer that is away is not a peer that | |
| 94 | // is gone, and this is exactly the state the surface reports as unresolved. | |
| 95 | }) | |
| 96 | ||
| 97 | test('★ it mirrors every followed feed, not just the first', async (t) => { | |
| 98 | const one = []; const two = [] | |
| 99 | const m = createMirror({ | |
| 100 | feeds: () => [ | |
| 101 | { key: 'f1', feed: fakeFeed(['a'], { onFetch: (n) => one.push(n) }) }, | |
| 102 | { key: 'f2', feed: fakeFeed(['b'], { onFetch: (n) => two.push(n) }) } | |
| 103 | ], | |
| 104 | log: () => {} | |
| 105 | }) | |
| 106 | ||
| 107 | await m.pass() | |
| 108 | ||
| 109 | t.alike(one, ['a']) | |
| 110 | t.alike(two, ['b']) | |
| 111 | }) | |
| 112 | ||
| 113 | test('★ stop() ends the pass, and a stopped mirror does nothing', async (t) => { | |
| 114 | const fetched = [] | |
| 115 | const feed = fakeFeed(['a', 'b', 'c'], { onFetch: (n) => fetched.push(n) }) | |
| 116 | const m = createMirror({ feeds: () => [{ key: 'f1', feed }], log: () => {} }) | |
| 117 | ||
| 118 | m.stop() | |
| 119 | await m.pass() | |
| 120 | ||
| 121 | t.is(fetched.length, 0, 'nothing after stop') | |
| 122 | }) | |
| 123 | ||
| 124 | test('⛔ a feed that cannot be listed is skipped, not fatal', async (t) => { | |
| 125 | const fetched = [] | |
| 126 | const bad = { async pointers () { throw new Error('view not ready') } } | |
| 127 | const good = fakeFeed(['a'], { onFetch: (n) => fetched.push(n) }) | |
| 128 | const m = createMirror({ feeds: () => [{ key: 'bad', feed: bad }, { key: 'ok', feed: good }], log: () => {} }) | |
| 129 | ||
| 130 | await m.pass() | |
| 131 | ||
| 132 | t.alike(fetched, ['a'], 'the healthy feed still mirrored') | |
| 133 | }) |