| … | … | @@ -21,6 +21,7 @@ import { join } from 'bare-path' |
| 21 | 21 | import { createSwarmIdentity } from 'atproto-identity' |
| 22 | 22 | import { StateBaseV3 } from 'atproto-room/lib/state-base.js' |
| 23 | 23 | import { SwarmDevice } from './swarm-device.js' |
| 24 | import { nextRetry } from './reconcile-retry.js' | |
| 24 | 25 | import { serveDevice } from './serve-device.js' |
| 25 | 26 | import { serveDeviceUds } from './serve-device-uds.js' |
| 26 | 27 | import { serveMembrane } from 'atproto-room/lib/serve-membrane.js' |
| … | … | @@ -400,13 +401,47 @@ export async function bootSwarmDevice ({ storage, deviceMnemonic, did, signingKe |
| 400 | 401 | // `if (controlEnabled)`, and reconcile() is defined before it. Null until the blob stores exist. |
| 401 | 402 | let blobStore = null |
| 402 | 403 | let reconciling = false |
| 403 | const reconcile = async () => { | |
| 404 | ||
| 405 | // ⛔ A FAILED FETCH IS NOT "NOTHING TO DO", AND THAT COST A POST. The pass loop below used to break on | |
| 406 | // `!r.fetched`, which collapses two opposite outcomes: "everything resolved" and "I tried and could not | |
| 407 | // reach the bytes". The comment said the next event would retry — and events are the ONLY driver here | |
| 408 | // (zero polling, by design: sb.watch on a StateBase advance or a presence change, plus reconcile() on | |
| 409 | // overlay attach when a peer connects). For the MOST RECENTLY converged pointer the advance that would | |
| 410 | // retry it has already happened, so a single transient miss leaves that record unresolved until some | |
| 411 | // unrelated write or a new peer connection shakes it loose. | |
| 412 | // | |
| 413 | // Measured on the handset 2026-08-18: a post converged at 21:26, its one fetch failed, and nothing | |
| 414 | // retried for 13 minutes. At the next boot the same fetch failed once more and SUCCEEDED 3 seconds | |
| 415 | // later — the miss was transient all along (the peer had not registered the oid yet: hyper-overlay's | |
| 416 | // oid→disk map is in-memory and rebuilt per boot, atproto-store invariant #5). Downstream, favela-web's | |
| 417 | // appview correctly refuses to author bytes hpp owns, so the record indexed but could not render: every | |
| 418 | // layer behaving as designed, and the post invisible. | |
| 419 | // | |
| 420 | // So retry on FAILURE specifically, bounded and self-disarming: | |
| 421 | // - one pending retry at a time, never a queue (a burst of events must not stack timers) | |
| 422 | // - exponential backoff, because a peer that is genuinely absent must not be hammered | |
| 423 | // - a clean pass disarms and resets the budget, so steady state is still zero timers | |
| 424 | // - a real EVENT resets the budget too: the world changed, so the previous exhaustion is stale | |
| 425 | // - unref'd, because a convergence retry must never be the reason a process stays alive | |
| 426 | // The arithmetic and the budget live in lib/reconcile-retry.js (pure, tested in | |
| 427 | // test/reconcile-retry.test.mjs). This owns only the timer — and unrefs it, because a convergence | |
| 428 | // retry must never be the reason a process stays alive. | |
| 429 | let retryTimer = null | |
| 430 | let retryDelay = 0 | |
| 431 | let retryAttempts = 0 | |
| 432 | ||
| 433 | const reconcile = async ({ fromRetry = false } = {}) => { | |
| 404 | 434 | if (reconciling) return |
| 405 | 435 | reconciling = true |
| 436 | if (retryTimer) { clearTimeout(retryTimer); retryTimer = null } | |
| 437 | let failedRecords = 0 | |
| 438 | let failedBlobs = 0 | |
| 406 | 439 | try { |
| 407 | 440 | for (let pass = 0; pass < 64; pass++) { |
| 408 | 441 | const r = await device.reconcileBytes() |
| 409 | if (!r || r.skipped || !r.more || !r.fetched) break // done, or no progress → wait for an event | |
| 442 | if (!r || r.skipped) break | |
| 443 | failedRecords = r.failed || 0 | |
| 444 | if (!r.more || !r.fetched) break // resolved everything reachable this pass | |
| 410 | 445 | } |
| 411 | 446 | // BLOBS converge on the same events. reconcileBytes walks the RECORD pointer set and skips |
| 412 | 447 | // `__`-prefixed keys, so `__blob__/<cid>` could never be covered by it — without this, a blob's |
| … | … | @@ -414,10 +449,24 @@ export async function bootSwarmDevice ({ storage, deviceMnemonic, did, signingKe |
| 414 | 449 | if (blobStore && typeof blobStore.reconcileBlobs === 'function') { |
| 415 | 450 | for (let pass = 0; pass < 64; pass++) { |
| 416 | 451 | const b = await blobStore.reconcileBlobs() |
| 417 | if (!b || b.skipped || !b.more || !b.fetched) break | |
| 452 | if (!b || b.skipped) break | |
| 453 | failedBlobs = b.failed || 0 | |
| 454 | if (!b.more || !b.fetched) break | |
| 418 | 455 | } |
| 419 | 456 | } |
| 420 | 457 | } catch {} finally { reconciling = false } |
| 458 | ||
| 459 | const d = nextRetry({ failed: failedRecords, failedBlobs, fromRetry, attempts: retryAttempts, delay: retryDelay }) | |
| 460 | retryAttempts = d.attempts | |
| 461 | retryDelay = d.delay | |
| 462 | if (d.giveUp) { | |
| 463 | console.log(JSON.stringify({ msg: 'reconcile: giving up until the next event', did: device.did, records: failedRecords, blobs: failedBlobs, attempts: d.attempts })) | |
| 464 | return | |
| 465 | } | |
| 466 | if (!d.arm) return | |
| 467 | console.log(JSON.stringify({ msg: 'reconcile: bytes unresolved — retry armed', did: device.did, records: failedRecords, blobs: failedBlobs, attempt: d.attempts, inMs: d.delay })) | |
| 468 | retryTimer = setTimeout(() => { retryTimer = null; reconcile({ fromRetry: true }).catch(() => {}) }, d.delay) | |
| 469 | if (retryTimer.unref) retryTimer.unref() | |
| 421 | 470 | } |
| 422 | 471 | |
| 423 | 472 | let ticking = false |
| … | … | @@ -0,0 +1,58 @@ |
| 1 | // The retry decision for byte convergence — pure, so it can be tested without booting a device. | |
| 2 | // | |
| 3 | // ⛔ WHY THIS IS ITS OWN MODULE. The rule it replaces lived inside a closure in `bootSwarmDevice`, where | |
| 4 | // nothing could reach it, and it was wrong in a way no test could have caught there: the pass loop ended | |
| 5 | // with `if (!r.fetched) break`, which treats "everything resolved" and "I tried and could not reach the | |
| 6 | // bytes" as the same outcome. The driver is deliberately zero-polling — it runs on a StateBase advance, a | |
| 7 | // signer-presence change, or a peer's overlay attach — so "wait for an event" meant the most recently | |
| 8 | // converged pointer waited for an event that had already happened. | |
| 9 | // | |
| 10 | // Measured on the handset 2026-08-18: a post converged at 21:26, its one fetch failed, nothing retried | |
| 11 | // for 13 minutes; at the next boot the same fetch failed once and succeeded 3 seconds later. Transient | |
| 12 | // all along. See test/reconcile-retry.test.mjs, whose first test pins the old decision. | |
| 13 | // | |
| 14 | // The caller owns the timer (and must unref it — a convergence retry may never be the reason a process | |
| 15 | // stays alive). This owns only the arithmetic and the budget. | |
| 16 | ||
| 17 | /** | |
| 18 | * @param {object} opts | |
| 19 | * @param {number} opts.failed records that could not be fetched this pass | |
| 20 | * @param {number} [opts.failedBlobs] blobs that could not be fetched this pass (same class of fault: | |
| 21 | * reconcileBytes skips `__`-prefixed keys, so blobs converge in a | |
| 22 | * separate pass and would otherwise be the half that never retries) | |
| 23 | * @param {boolean} opts.fromRetry true when this pass WAS a retry; false when an event drove it | |
| 24 | * @param {number} opts.attempts retries already spent in this run of failures | |
| 25 | * @param {number} opts.delay the delay used last time, 0 if none | |
| 26 | * @param {number} [opts.baseMs] | |
| 27 | * @param {number} [opts.maxMs] | |
| 28 | * @param {number} [opts.maxAttempts] | |
| 29 | * @returns {{arm: boolean, delay: number, attempts: number, giveUp: boolean}} | |
| 30 | */ | |
| 31 | export function nextRetry ({ | |
| 32 | failed = 0, | |
| 33 | failedBlobs = 0, | |
| 34 | fromRetry = false, | |
| 35 | attempts = 0, | |
| 36 | delay = 0, | |
| 37 | baseMs = 5000, | |
| 38 | maxMs = 300000, | |
| 39 | maxAttempts = 8 | |
| 40 | } = {}) { | |
| 41 | // AN EVENT IS NEW INFORMATION; A RETRY IS THE SAME INFORMATION AGAIN. Only the former earns a fresh | |
| 42 | // budget — otherwise a node that hit a bad patch would spend its attempts once and never try again, no | |
| 43 | // matter what changed afterwards (a peer arriving is exactly such a change, and it arrives as an event). | |
| 44 | if (!fromRetry) { attempts = 0; delay = 0 } | |
| 45 | ||
| 46 | if (!failed && !failedBlobs) return { arm: false, delay: 0, attempts: 0, giveUp: false } | |
| 47 | ||
| 48 | // Bounded. A peer that genuinely does not hold these bytes must not be asked forever — but the caller | |
| 49 | // is TOLD, because the defect this module exists to fix was precisely a node giving up in silence. | |
| 50 | if (attempts >= maxAttempts) return { arm: false, delay, attempts, giveUp: true } | |
| 51 | ||
| 52 | return { | |
| 53 | arm: true, | |
| 54 | delay: delay ? Math.min(delay * 2, maxMs) : baseMs, | |
| 55 | attempts: attempts + 1, | |
| 56 | giveUp: false | |
| 57 | } | |
| 58 | } |
| … | … | @@ -1,7 +1,7 @@ |
| 1 | 1 | { |
| 2 | 2 | "name": "hyper-proto-peer", |
| 3 | "version": "0.3.118", | |
| 4 | "description": "Self-custody, P2P-native atproto peer \u2014 holds the user's key + content as Hypercores and authors+signs its own repo. The source of truth in the peer/bridge split (Model B); atproto-bridge projects it to bsky.", | |
| 3 | "version": "0.3.119", | |
| 4 | "description": "Self-custody, P2P-native atproto peer — holds the user's key + content as Hypercores and authors+signs its own repo. The source of truth in the peer/bridge split (Model B); atproto-bridge projects it to bsky.", | |
| 5 | 5 | "type": "module", |
| 6 | 6 | "main": "lib/peer.js", |
| 7 | 7 | "bin": { |
| … | … | @@ -23,6 +23,7 @@ import './serve-device.test.js' |
| 23 | 23 | import './dual-blob-store.test.js' |
| 24 | 24 | import './blob-pointer-wiring.test.mjs' |
| 25 | 25 | import './reconcile-bytes.test.mjs' |
| 26 | import './reconcile-retry.test.mjs' | |
| 26 | 27 | import './pointer-diag.test.mjs' |
| 27 | 28 | import './coowner.test.mjs' |
| 28 | 29 | import './coowner-join-role.test.js' |
| … | … | @@ -0,0 +1,96 @@ |
| 1 | // When a byte-convergence pass FAILS, what happens next. | |
| 2 | // | |
| 3 | // ⛔ THE CONTROL BELOW IS THE BUG. The reconcile driver used to end each pass with | |
| 4 | // | |
| 5 | // if (!r || r.skipped || !r.more || !r.fetched) break // done, or no progress → wait for an event | |
| 6 | // | |
| 7 | // which collapses two opposite outcomes: "everything resolved" and "I tried and could not reach the | |
| 8 | // bytes". Both stopped, and neither scheduled anything. Since the driver is deliberately zero-polling — | |
| 9 | // it runs on a StateBase advance, a presence change, or a peer's overlay attach — the retry for the MOST | |
| 10 | // RECENTLY converged pointer never came: the advance that would have triggered it already happened. | |
| 11 | // | |
| 12 | // Measured on the handset 2026-08-18: a post converged at 21:26, its single fetch failed, and nothing | |
| 13 | // retried for 13 minutes. On the next boot the same fetch failed once and succeeded 3 seconds later, so | |
| 14 | // the miss was transient the whole time. Downstream the appview correctly refuses to author bytes hpp | |
| 15 | // owns, so the record indexed but could not render — every layer correct, the post invisible. | |
| 16 | // | |
| 17 | // So the first test pins the old decision, and the rest assert the policy that replaces it. | |
| 18 | // | |
| 19 | // Run: bare test/reconcile-retry.test.mjs | |
| 20 | import '../lib/bare-globals.js' | |
| 21 | import test from 'brittle' | |
| 22 | import { nextRetry } from '../lib/reconcile-retry.js' | |
| 23 | ||
| 24 | const P = { baseMs: 5000, maxMs: 300000, maxAttempts: 8 } | |
| 25 | const clean = { failed: 0, fromRetry: false, attempts: 0, delay: 0, ...P } | |
| 26 | ||
| 27 | test('control: the old rule could not tell "resolved" from "could not reach it"', (t) => { | |
| 28 | // Both of these ended the pass identically, and neither armed anything. | |
| 29 | const resolvedEverything = { fetched: 3, failed: 0 } | |
| 30 | const couldNotReachIt = { fetched: 0, failed: 1 } | |
| 31 | const oldDecision = (r) => (!r.fetched ? 'stop' : 'continue') | |
| 32 | t.is(oldDecision(couldNotReachIt), 'stop', 'a failed fetch stopped the loop') | |
| 33 | t.is(oldDecision({ ...resolvedEverything, fetched: 0 }), 'stop', '…exactly as a finished one did') | |
| 34 | }) | |
| 35 | ||
| 36 | test('a clean pass arms nothing and holds steady state at zero timers', (t) => { | |
| 37 | const d = nextRetry(clean) | |
| 38 | t.is(d.arm, false) | |
| 39 | t.is(d.attempts, 0, 'budget reset') | |
| 40 | t.is(d.delay, 0) | |
| 41 | t.is(d.giveUp, false) | |
| 42 | }) | |
| 43 | ||
| 44 | test('a failure after an event arms the first retry at the base delay', (t) => { | |
| 45 | const d = nextRetry({ ...clean, failed: 1 }) | |
| 46 | t.is(d.arm, true) | |
| 47 | t.is(d.delay, 5000) | |
| 48 | t.is(d.attempts, 1) | |
| 49 | }) | |
| 50 | ||
| 51 | test('a failed retry backs off exponentially', (t) => { | |
| 52 | let d = nextRetry({ ...clean, failed: 1 }) | |
| 53 | const seen = [d.delay] | |
| 54 | for (let i = 0; i < 4; i++) { | |
| 55 | d = nextRetry({ ...P, failed: 1, fromRetry: true, attempts: d.attempts, delay: d.delay }) | |
| 56 | seen.push(d.delay) | |
| 57 | } | |
| 58 | t.alike(seen, [5000, 10000, 20000, 40000, 80000], 'doubling, not a fixed interval') | |
| 59 | }) | |
| 60 | ||
| 61 | test('backoff is capped — an absent peer is never hammered, and never waits forever', (t) => { | |
| 62 | let d = { attempts: 3, delay: 200000 } | |
| 63 | d = nextRetry({ ...P, failed: 1, fromRetry: true, attempts: d.attempts, delay: d.delay }) | |
| 64 | t.is(d.delay, 300000, 'capped at maxMs') | |
| 65 | d = nextRetry({ ...P, failed: 1, fromRetry: true, attempts: d.attempts, delay: d.delay }) | |
| 66 | t.is(d.delay, 300000, 'stays at the cap rather than growing') | |
| 67 | }) | |
| 68 | ||
| 69 | test('the budget is bounded, and exhausting it is SAID rather than silent', (t) => { | |
| 70 | const d = nextRetry({ ...P, failed: 1, fromRetry: true, attempts: 8, delay: 300000 }) | |
| 71 | t.is(d.arm, false, 'no further timer') | |
| 72 | t.is(d.giveUp, true, 'and the caller is told, because giving up silently is the original defect') | |
| 73 | }) | |
| 74 | ||
| 75 | test('an EVENT after exhaustion earns a fresh budget', (t) => { | |
| 76 | // A retry is the same information again; an event is new information. Without this a node that had a | |
| 77 | // bad patch would burn its 8 attempts once and never try again, however much the world changed. | |
| 78 | const d = nextRetry({ ...P, failed: 1, fromRetry: false, attempts: 8, delay: 300000 }) | |
| 79 | t.is(d.arm, true) | |
| 80 | t.is(d.attempts, 1, 'counted from scratch') | |
| 81 | t.is(d.delay, 5000, 'and backed off from scratch') | |
| 82 | }) | |
| 83 | ||
| 84 | test('a clean pass after failures disarms and resets, so the next incident gets the full budget', (t) => { | |
| 85 | const d = nextRetry({ ...P, failed: 0, fromRetry: true, attempts: 5, delay: 80000 }) | |
| 86 | t.is(d.arm, false) | |
| 87 | t.is(d.attempts, 0) | |
| 88 | t.is(d.delay, 0) | |
| 89 | }) | |
| 90 | ||
| 91 | test('blob failures count the same as record failures', (t) => { | |
| 92 | // reconcileBytes skips `__`-prefixed keys, so blobs converge through a separate pass. A blob that | |
| 93 | // cannot be fetched is the same class of fault and must not be the one that goes unretried. | |
| 94 | t.is(nextRetry({ ...clean, failed: 0, failedBlobs: 2 }).arm, true) | |
| 95 | t.is(nextRetry({ ...clean, failed: 0, failedBlobs: 0 }).arm, false) | |
| 96 | }) |