From 984204685506a83c9e4bec186d8448f2a7a404fc Mon Sep 17 00:00:00 2001 From: Mathias Buus Date: Thu, 15 Oct 2026 01:58:06 +0200 Subject: [PATCH] Share decoded blocks between callers inflating the same node Every caller runs the inflation IO under its own config, so timeout, wait and cancellation are per caller and hypercore dedups the fetches. Decoded blocks land in slots on ptr.inflating shared by all callers, so the first to wake decodes and a late caller only fetches what is still missing. Co-Authored-By: Claude Fable 5.1 --- lib/context.js | 14 ++- lib/inflate.js | 272 ++++++++++++++++++++++++++++++------------------ test/all.js | 1 + test/inflate.js | 133 +++++++++++++++++++++++ 4 files changed, 318 insertions(+), 102 deletions(-) create mode 100644 test/inflate.js diff --git a/lib/context.js b/lib/context.js index fa6c580..5258ca0 100644 --- a/lib/context.js +++ b/lib/context.js @@ -209,6 +209,11 @@ class CoreContext { } async getBlock(seq, core, config, origin) { + const buffer = await this.getBlockBuffer(seq, core, config, origin) + return decodeBlock(buffer, seq) + } + + async getBlockBuffer(seq, core, config, origin) { if (core !== 0 && core - 1 >= this.cores.length) { await this.updateMaybe(config, core, origin) } @@ -219,8 +224,7 @@ class CoreContext { if (config.trace !== null) config.trace(core, seq) - const block = decodeBlock(buffer, seq) - return block + return buffer } getLocalContext() { @@ -247,6 +251,12 @@ class CoreContext { async getContext(core, config, origin) { if (core === 0) return this if (core > this.cores.length) await this.updateMaybe(config, core, origin) + return this.getLoadedContext(core) + } + + // context of a core this context already knows, eg after reading one of its blocks + getLoadedContext(core) { + if (core === 0) return this if (core > this.cores.length) throw new Error('Bad core index: ' + core) const hex = b4a.toString(this.cores[core - 1].key, 'hex') diff --git a/lib/inflate.js b/lib/inflate.js index 653b436..60bca62 100644 --- a/lib/inflate.js +++ b/lib/inflate.js @@ -1,77 +1,26 @@ const b4a = require('b4a') const { Pointer, KeyPointer, ValuePointer, TreeNode } = require('./tree.js') const { DeltaOp, DeltaCohort, OP_COHORT } = require('./compression.js') +const { decodeBlock } = require('./encoding.js') + +// Every caller runs the inflation IO under its own config, so timeout, wait and +// cancellation are per caller and hypercore dedups the fetches. Decoded blocks +// land in slots on ptr.inflating shared by all callers, filled by whoever wakes first. exports.inflate = async function inflate(ptr, config) { if (ptr.value) return ptr.value - if (ptr.inflating) return ptr.inflating + if (ptr.inflating === null) ptr.inflating = new Inflation(ptr) - ptr.inflating = _inflate(ptr, config) + const value = await ptr.inflating.run(config) - try { - return await ptr.inflating - } finally { + if (!ptr.value) { + ptr.value = value ptr.inflating = null } -} - -async function _inflate(ptr, config) { - const [block, context] = await Promise.all([ - ptr.context.getBlock(ptr.seq, ptr.core, config, null), - ptr.context.getContext(ptr.core, config, null) - ]) - const tree = block.tree[ptr.offset] - - const keys = new Array(tree.keys.length) - const children = new Array(tree.children.length) - - for (let i = 0; i < keys.length; i++) { - const d = tree.keys[i] - keys[i] = inflateKey(context, d, ptr, block, config) - } - - for (let i = 0; i < children.length; i++) { - const d = tree.children[i] - children[i] = inflateChild(context, d, ptr, block, config) - } - - const [k, c] = await Promise.all([Promise.all(keys), Promise.all(children)]) - - const value = new TreeNode(block.t, k, c) - if (!ptr.value) ptr.value = value return ptr.value } -function inflateKey(context, d, ptr, block, config) { - if (d.type === OP_COHORT) return inflateKeyCohort(context, d, ptr, block, config) - return inflateKeyDelta(context, d, ptr, block, config) -} - -async function inflateKeyDelta(context, d, ptr, block, config) { - const k = d.pointer - - if (!k) return new DeltaOp(false, d.type, d.index, null) - - const blk = - k.seq === ptr.seq && k.core === 0 && ptr.core === 0 - ? block - : await context.getBlock(k.seq, k.core, config, block) - - const bk = blk.keys[k.offset] - - let vp = null - - if (bk.valuePointer) { - const p = bk.valuePointer - const ctx = await context.getContext(k.core, config, block) - vp = new ValuePointer(ctx, p.core, p.seq, p.offset, p.split) - } - - const kp = new KeyPointer(context, k.core, k.seq, k.offset, false, bk.key, bk.value, vp) - return new DeltaOp(false, d.type, d.index, kp) -} - exports.inflateValue = async function inflateValue(key, config) { if (key.value) return key.value if (!key.valuePointer) return null @@ -96,62 +45,185 @@ exports.inflateValue = async function inflateValue(key, config) { return b4a.concat(splitValue) } -async function inflateKeyCohort(context, d, ptr, block, config) { - const co = d.pointer +class Inflation { + constructor(ptr) { + this.ptr = ptr + this.block = null + this.keys = null // per delta key: its block, or { block, entries } for a cohort + this.children = null // per delta child: { block, entries } for a cohort, else unused + } + + async run(config) { + const ptr = this.ptr + + // round 0, the ptr block + if (this.block === null) { + const buffer = await ptr.context.getBlockBuffer(ptr.seq, ptr.core, config, null) + if (this.block === null) this.block = decodeBlock(buffer, ptr.seq) + } + + if (ptr.value) return ptr.value + + const block = this.block + const context = ptr.context.getLoadedContext(ptr.core) + const tree = block.tree[ptr.offset] + + if (this.keys === null) { + this.keys = new Array(tree.keys.length) + this.children = new Array(tree.children.length) + } + + // round 1, the blocks the keys point at, the cohorts (round 2 inside) and + // the cores the children live in + const promises = [] + + for (let i = 0; i < tree.keys.length; i++) { + const d = tree.keys[i] + if (d.type === OP_COHORT) { + promises.push(this._cohort(this.keys, i, context, d, ptr, block, config, true)) + } else if (d.pointer && this.keys[i] === undefined) { + if (sameBlock(d.pointer, ptr)) this.keys[i] = block + else promises.push(fillBlock(this.keys, i, context, d.pointer, block, config)) + } + } + + for (let i = 0; i < tree.children.length; i++) { + const d = tree.children[i] + if (d.type === OP_COHORT) { + promises.push(this._cohort(this.children, i, context, d, ptr, block, config, false)) + } else if (needsCore(context, d)) { + promises.push(fillCore(context, d, block, config)) + } + } + + await Promise.all(promises) - const blk = - co.seq === ptr.seq && co.core === 0 && ptr.core === 0 - ? block - : await context.getBlock(co.seq, co.core, config, block) + if (ptr.value) return ptr.value - const cohort = blk.cohorts[co.offset] - const promises = new Array(cohort.length) - const ctx = await context.getContext(co.core, config, block) + // done, build the node from the slots + const keys = new Array(tree.keys.length) + const children = new Array(tree.children.length) - for (let i = 0; i < cohort.length; i++) { - const p = cohort[i] - const k = inflateKeyDelta(ctx, p, co, blk, config) - promises[i] = k + for (let i = 0; i < keys.length; i++) { + const d = tree.keys[i] + if (d.type === OP_COHORT) keys[i] = keyCohort(context, d, this.keys[i]) + else keys[i] = key(context, d, this.keys[i]) + } + + for (let i = 0; i < children.length; i++) { + const d = tree.children[i] + if (d.type === OP_COHORT) children[i] = childCohort(context, d, this.children[i]) + else children[i] = child(context, d) + } + + return new TreeNode(block.t, keys, children) + } + + async _cohort(slots, i, context, d, ptr, block, config, isKey) { + const co = d.pointer + + // round 1, the cohort block + if (slots[i] === undefined) { + if (sameBlock(co, ptr)) slots[i] = { block, entries: null } + else await fillCohortBlock(slots, i, context, co, block, config) + } + + const slot = slots[i] + const blk = slot.block + const ctx = context.getLoadedContext(co.core) + const cohort = blk.cohorts[co.offset] + + if (slot.entries === null) slot.entries = new Array(cohort.length) + + // round 2, the blocks the cohort keys point at or the cores its children live in + const promises = [] + + for (let j = 0; j < cohort.length; j++) { + const p = cohort[j] + if (isKey) { + if (p.pointer && slot.entries[j] === undefined) { + if (sameBlock(p.pointer, co)) slot.entries[j] = blk + else promises.push(fillBlock(slot.entries, j, ctx, p.pointer, blk, config)) + } + } else if (needsCore(ctx, p)) { + promises.push(fillCore(ctx, p, blk, config)) + } + } + + await Promise.all(promises) } +} + +// IO, run by every caller. Slots are filled once. + +function sameBlock(p, ptr) { + return p.seq === ptr.seq && p.core === 0 && ptr.core === 0 +} + +async function fillBlock(slots, i, context, p, origin, config) { + const buffer = await context.getBlockBuffer(p.seq, p.core, config, origin) + // callers woken by the same block decode it once + if (slots[i] === undefined) slots[i] = decodeBlock(buffer, p.seq) +} - const p = new Pointer(context, co.core, co.seq, co.offset) - return new DeltaCohort(false, p, await Promise.all(promises)) +async function fillCohortBlock(slots, i, context, co, origin, config) { + const buffer = await context.getBlockBuffer(co.seq, co.core, config, origin) + if (slots[i] === undefined) slots[i] = { block: decodeBlock(buffer, co.seq), entries: null } } -async function inflateChild(context, d, ptr, block, config) { - if (d.type === OP_COHORT) return inflateChildCohort(context, d, ptr, block, config) - if (d.pointer && !context.hasCore(d.pointer.core)) { - await context.updateMaybe(config, d.pointer.core, block) +// a child in a core the context does not know yet needs the checkpoint first +function needsCore(context, d) { + return d.pointer !== null && !context.hasCore(d.pointer.core) +} + +function fillCore(context, d, block, config) { + return context.updateMaybe(config, d.pointer.core, block) +} + +// Building from filled slots, sync. + +function key(context, d, blk) { + const k = d.pointer + if (!k) return new DeltaOp(false, d.type, d.index, null) + + const bk = blk.keys[k.offset] + + let vp = null + + if (bk.valuePointer) { + const p = bk.valuePointer + const ctx = context.getLoadedContext(k.core) + vp = new ValuePointer(ctx, p.core, p.seq, p.offset, p.split) } - return inflateChildDelta(context, d, ptr, block, config) + + const kp = new KeyPointer(context, k.core, k.seq, k.offset, false, bk.key, bk.value, vp) + return new DeltaOp(false, d.type, d.index, kp) } -function inflateChildDelta(context, d, ptr, block, config) { +function keyCohort(context, d, slot) { + const co = d.pointer + const ctx = context.getLoadedContext(co.core) + const cohort = slot.block.cohorts[co.offset] + + const keys = new Array(cohort.length) + for (let j = 0; j < keys.length; j++) keys[j] = key(ctx, cohort[j], slot.entries[j]) + + return new DeltaCohort(false, new Pointer(context, co.core, co.seq, co.offset), keys) +} + +function child(context, d) { const p = d.pointer const c = p && context.createTreeNode(p.core, p.seq, p.offset, false, null) return new DeltaOp(false, d.type, d.index, c) } -async function inflateChildCohort(context, d, ptr, block, config) { +function childCohort(context, d, slot) { const co = d.pointer + const ctx = context.getLoadedContext(co.core) + const cohort = slot.block.cohorts[co.offset] - const blk = - co.seq === ptr.seq && co.core === 0 && ptr.core === 0 - ? block - : await context.getBlock(co.seq, co.core, config, block) - - const cohort = blk.cohorts[co.offset] - const deltas = new Array(cohort.length) - const ctx = await context.getContext(co.core, config, block) - - for (let i = 0; i < cohort.length; i++) { - const c = cohort[i] - if (c.pointer && !ctx.hasCore(c.pointer.core)) { - await ctx.updateMaybe(config, c.pointer.core, blk) - } - deltas[i] = inflateChildDelta(ctx, c, co, blk, config) - } + const children = new Array(cohort.length) + for (let j = 0; j < children.length; j++) children[j] = child(ctx, cohort[j]) - const p = new Pointer(context, co.core, co.seq, co.offset) - return new DeltaCohort(false, p, deltas) + return new DeltaCohort(false, new Pointer(context, co.core, co.seq, co.offset), children) } diff --git a/test/all.js b/test/all.js index 57f28cd..953fdc9 100644 --- a/test/all.js +++ b/test/all.js @@ -13,6 +13,7 @@ async function runTests() { await import('./diff.js') await import('./download.js') await import('./fuzz.js') + await import('./inflate.js') await import('./fuzz-bad-cache.js') await import('./fuzz-diff.js') await import('./fuzz-multiwriter.js') diff --git a/test/inflate.js b/test/inflate.js new file mode 100644 index 0000000..d753de2 --- /dev/null +++ b/test/inflate.js @@ -0,0 +1,133 @@ +const test = require('brittle') +const b4a = require('b4a') +const { create, replicate } = require('./helpers') + +test('destroying one stream does not cancel an inflation shared with another', async function (t) { + const { db, reader, entries } = await createRemote(t) + + const view = reader.checkout({ length: db.head().length }) + + const s1 = view.createReadStream() + const s2 = view.createReadStream() + + const r1 = collect(s1) + const r2 = collect(s2) + + // let both streams reach the root inflation + await new Promise((resolve) => setTimeout(resolve, 50)) + + s1.destroy() + await t.exception(r1, /REQUEST_CANCELLED|destroyed/i) + + await replicate(t, db, reader) + + const got = await r2 + t.is(got.length, entries.length) + for (let i = 0; i < entries.length; i++) t.alike(got[i].key, entries[i].key) + + await view.close() +}) + +test("shared inflation honours each caller's own timeout", async function (t) { + const { db, reader, entries } = await createRemote(t) + + const view = reader.checkout({ length: db.head().length }) + + const fast = view.get(entries[0].key, { timeout: 50 }) + const slow = view.get(entries[0].key, { timeout: 10000 }) + + await t.exception(fast, /REQUEST_TIMEOUT/) + + await replicate(t, db, reader) + + const node = await slow + t.alike(node.value, entries[0].value) + + await view.close() +}) + +test('wait: false fails fast while a shared fetch is in flight', async function (t) { + const { db, reader, entries } = await createRemote(t) + + const view = reader.checkout({ length: db.head().length }) + + const pending = view.get(entries[0].key) + await new Promise((resolve) => setTimeout(resolve, 50)) + + await t.exception(view.get(entries[0].key, { wait: false }), /BLOCK_NOT_AVAILABLE/) + + await replicate(t, db, reader) + + const node = await pending + t.alike(node.value, entries[0].value) + + await view.close() +}) + +test('each caller sharing an inflation traces its own reads', async function (t) { + const { db, reader, entries } = await createRemote(t) + + const view = reader.checkout({ length: db.head().length }) + + const a = new Set() + const b = new Set() + + const ga = view.get(entries[0].key, { trace: (core, seq) => a.add(seq) }) + const gb = view.get(entries[0].key, { trace: (core, seq) => b.add(seq) }) + + await replicate(t, db, reader) + await Promise.all([ga, gb]) + + t.ok(a.size > 0) + t.alike(a, b) + + await view.close() +}) + +test('a late caller reuses the first round and shares the second', async function (t) { + const { db, reader, entries } = await createRemote(t) + + const view = reader.checkout({ length: db.head().length }) + const root = await view.bootstrap(view.config) + + const first = view.inflate(root, view.config) + await new Promise((resolve) => setTimeout(resolve, 50)) + t.ok(root.inflating !== null, 'round 0 in flight') + + await replicate(t, db, reader) + await first + + t.is(root.inflating, null, 'state dropped once inflated') + t.ok(root.value) + + const again = await view.inflate(root, view.config) + t.is(again, root.value) + + await view.close() +}) + +async function createRemote(t) { + const db = await create(t, { t: 4 }) + const entries = [] + + for (let i = 0; i < 100; i++) entries.push({ key: entriesKey(i), value: b4a.from('v' + i) }) + + const w = db.write() + for (const e of entries) w.tryPut(e.key, e.value) + await w.flush() + + const reader = await create(t, { key: db.core.key, t: 4 }) + await reader.ready() + + return { db, reader, entries } +} + +function entriesKey(i) { + return b4a.from('k' + i.toString().padStart(4, '0')) +} + +async function collect(stream) { + const result = [] + for await (const data of stream) result.push(data) + return result +}