From 50454b551a40e632755190a9ff962f371386b94f Mon Sep 17 00:00:00 2001 From: Mathias Buus Date: Wed, 21 Oct 2026 16:18:06 +0200 Subject: [PATCH 1/2] Add invalidation pointer to blocks Block-1 gains an optional invalidation uint. Every decode path returns it (0 for block-0 and bee1 compat blocks), and the writer carries the last loaded value forward into new blocks. The core context tracks the pointer it loaded its table with and reloads the table when a block carries a newer one, in addition to the existing reload when a referenced core is missing. Co-Authored-By: Claude Fable 5.1 --- build.js | 4 ++ lib/compat.js | 3 +- lib/context.js | 17 +++++-- lib/encoding.js | 3 +- lib/write.js | 3 +- spec/hyperschema/index.js | 20 ++++++-- spec/hyperschema/schema.json | 5 ++ test/checkpoint.js | 94 ++++++++++++++++++++++++++++++++++++ 8 files changed, 140 insertions(+), 9 deletions(-) diff --git a/build.js b/build.js index ca5c6b1..9eca427 100644 --- a/build.js +++ b/build.js @@ -339,6 +339,10 @@ bee.register({ { name: 't', type: 'uint' + }, + { + name: 'invalidation', // TODO: in future block-2 should be required + type: 'uint' } ] }) diff --git a/lib/compat.js b/lib/compat.js index ca48a8d..a5e5e6d 100644 --- a/lib/compat.js +++ b/lib/compat.js @@ -25,7 +25,8 @@ function decodeCompat(buffer, seq) { keys: node ? [{ key: node.key, value: node.value, valuePointer: null }] : [], values: [], cohorts: [], - t: exports.t + t: exports.t, + invalidation: 0 } for (const lvl of index.levels) { diff --git a/lib/context.js b/lib/context.js index fa6c580..ad425c1 100644 --- a/lib/context.js +++ b/lib/context.js @@ -26,6 +26,7 @@ class CoreContext { this.other = other this.length = 0 this.checkpoint = 0 + this.invalidation = 0 this.opened = [] this.cores = [] this.changed = false @@ -50,11 +51,11 @@ class CoreContext { async updateMaybe(config, core, origin) { await this.core.ready() - if (this.hasCore(core)) return + if (!this.needsUpdate(core, origin)) return if (origin !== null) { await this._inflateCheckpoint(origin.seq + 1, origin, config, false) - if (this.hasCore(core)) return + if (!this.needsUpdate(core, origin)) return } const buf = await this.core.getUserData(CHECKPOINT_KEY) @@ -62,7 +63,7 @@ class CoreContext { if (cached > this.length && cached <= this.core.length) { await this._inflateCheckpointFromLength(cached, config, false) - if (this.hasCore(core)) return + if (!this.needsUpdate(core, origin)) return } await this.update(config) @@ -72,6 +73,11 @@ class CoreContext { return core <= this.opened.length } + needsUpdate(core, origin) { + if (!this.hasCore(core)) return true + return origin !== null && origin.invalidation > this.invalidation + } + async _inflateCheckpointFromLength(length, config, persist) { const seq = length - 1 const buffer = await this.core.get(seq, config) @@ -108,6 +114,7 @@ class CoreContext { } this.checkpoint = checkpoint + this.invalidation = block.invalidation this.length = length if (persist) { @@ -220,6 +227,10 @@ class CoreContext { if (config.trace !== null) config.trace(core, seq) const block = decodeBlock(buffer, seq) + + const ctx = core === 0 ? this : await this.getContext(core, config, origin) + if (block.invalidation > ctx.invalidation) await ctx.updateMaybe(config, 0, block) + return block } diff --git a/lib/encoding.js b/lib/encoding.js index 6553e02..6b90043 100644 --- a/lib/encoding.js +++ b/lib/encoding.js @@ -90,7 +90,8 @@ function decodeBlock0(state, seq) { keys: blk.data.map(dataToKey), values: null, cohorts, - t: T + t: T, + invalidation: 0 } } diff --git a/lib/write.js b/lib/write.js index ebeccfb..bfd2832 100644 --- a/lib/write.js +++ b/lib/write.js @@ -462,7 +462,8 @@ module.exports = class WriteBatch { keys: null, values: null, cohorts: null, - t + t, + invalidation: context.invalidation } for (const k of update.values) { diff --git a/spec/hyperschema/index.js b/spec/hyperschema/index.js index b96029a..3093d62 100644 --- a/spec/hyperschema/index.js +++ b/spec/hyperschema/index.js @@ -450,10 +450,20 @@ const encoding14_9 = c.array(encoding3) // @bee/block-1 const encoding14 = { preencode(state, m) { + const flags = + (m.previous ? 1 : 0) | + (m.metadata ? 2 : 0) | + (m.tree ? 4 : 0) | + (m.keys ? 8 : 0) | + (m.values ? 16 : 0) | + (m.cohorts ? 32 : 0) | + (m.t ? 64 : 0) | + (m.invalidation ? 128 : 0) + c.uint.preencode(state, m.type) c.uint.preencode(state, m.checkpoint) encoding6.preencode(state, m.batch) - state.end++ // max flag is 64 so always one byte + c.uint.preencode(state, flags) if (m.previous) encoding5.preencode(state, m.previous) if (m.metadata) encoding14_5.preencode(state, m.metadata) @@ -462,6 +472,7 @@ const encoding14 = { if (m.values) encoding14_8.preencode(state, m.values) if (m.cohorts) encoding14_9.preencode(state, m.cohorts) if (m.t) c.uint.preencode(state, m.t) + if (m.invalidation) c.uint.preencode(state, m.invalidation) }, encode(state, m) { const flags = @@ -471,7 +482,8 @@ const encoding14 = { (m.keys ? 8 : 0) | (m.values ? 16 : 0) | (m.cohorts ? 32 : 0) | - (m.t ? 64 : 0) + (m.t ? 64 : 0) | + (m.invalidation ? 128 : 0) c.uint.encode(state, m.type) c.uint.encode(state, m.checkpoint) @@ -485,6 +497,7 @@ const encoding14 = { if (m.values) encoding14_8.encode(state, m.values) if (m.cohorts) encoding14_9.encode(state, m.cohorts) if (m.t) c.uint.encode(state, m.t) + if (m.invalidation) c.uint.encode(state, m.invalidation) }, decode(state) { const r0 = c.uint.decode(state) @@ -503,7 +516,8 @@ const encoding14 = { keys: (flags & 8) !== 0 ? encoding14_7.decode(state) : null, values: (flags & 16) !== 0 ? encoding14_8.decode(state) : null, cohorts: (flags & 32) !== 0 ? encoding14_9.decode(state) : null, - t: (flags & 64) !== 0 ? c.uint.decode(state) : 0 + t: (flags & 64) !== 0 ? c.uint.decode(state) : 0, + invalidation: (flags & 128) !== 0 ? c.uint.decode(state) : 0 } } } diff --git a/spec/hyperschema/schema.json b/spec/hyperschema/schema.json index 8c917a9..309401c 100644 --- a/spec/hyperschema/schema.json +++ b/spec/hyperschema/schema.json @@ -404,6 +404,11 @@ "name": "t", "type": "uint", "version": 1 + }, + { + "name": "invalidation", + "type": "uint", + "version": 1 } ] } diff --git a/test/checkpoint.js b/test/checkpoint.js index 062ec87..9c128ce 100644 --- a/test/checkpoint.js +++ b/test/checkpoint.js @@ -2,6 +2,7 @@ const test = require('brittle') const b4a = require('b4a') const Corestore = require('corestore') const Bee = require('../') +const { encodeBlock, decodeBlock } = require('../lib/encoding.js') const { create, replicate } = require('./helpers') test('offline reader can move to cached checkpoint after learning a newer length', async function (t) { @@ -161,3 +162,96 @@ function replicateManual(a, b) { await closed2 } } + +test('invalidation pointer forces a core table reload', async function (t) { + const db = await create(t) + { + const w = db.write() + w.tryPut(b4a.from('a'), b4a.from('1')) + await w.flush() + } + + const context = db.context + const config = db.config + + await context.update(config) + + const length = context.length + + t.is(context.invalidation, 0) + t.is(context.cores.length, 0) + + const cores = [{ key: b4a.alloc(32, 1), fork: 0, length: 0, treeHash: null }] + + // newer core table, but no invalidation pointer: the loaded table is kept + await db.core.append(encodeBlock(metadataBlock(length + 1, cores, 0))) + await context.getBlock(length, 0, config, null) + + t.is(context.cores.length, 0, 'no reload without an invalidation pointer') + t.is(context.length, length) + + // invalidation pointer newer than the one the table was loaded with: reload + await db.core.append(encodeBlock(metadataBlock(length + 2, cores, length + 2))) + await context.getBlock(length + 1, 0, config, null) + + t.is(context.invalidation, length + 2) + t.is(context.length, length + 2) + t.alike(context.cores, cores, 'table reloaded from the invalidating block') + + // invalidation pointer equal to the last one seen: no reload + await db.core.append(encodeBlock(metadataBlock(length + 3, [], length + 2))) + await context.getBlock(length + 2, 0, config, null) + + t.is(context.invalidation, length + 2) + t.is(context.length, length + 2) + t.alike(context.cores, cores, 'table kept') + + // rereading the invalidating block is a noop + await context.getBlock(length + 1, 0, config, null) + + t.is(context.length, length + 2) + t.alike(context.cores, cores) +}) + +test('new blocks carry the invalidation pointer forward', async function (t) { + const db = await create(t) + { + const w = db.write() + w.tryPut(b4a.from('a'), b4a.from('1')) + await w.flush() + } + + await db.context.update(db.config) + + const length = db.context.length + const cores = db.context.cores + + await db.core.append(encodeBlock(metadataBlock(length + 1, cores, length + 1))) + + { + const w = db.write() + w.tryPut(b4a.from('b'), b4a.from('2')) + await w.flush() + } + + const head = decodeBlock(await db.core.get(db.core.length - 1), db.core.length - 1) + + t.is(head.invalidation, length + 1) + t.alike((await db.get(b4a.from('b'))).value, b4a.from('2')) +}) + +function metadataBlock(checkpoint, cores, invalidation) { + return { + type: 1, + checkpoint, + batch: { start: 0, end: 0 }, + previous: null, + metadata: { cores }, + tree: null, + keys: null, + values: null, + cohorts: null, + t: 0, + invalidation + } +} From 4d61c46aeaa9a8a886b3e179d873555e029633fc Mon Sep 17 00:00:00 2001 From: Mathias Buus Date: Wed, 21 Oct 2026 16:38:25 +0200 Subject: [PATCH 2/2] Split getBlock into getBlockBuffer and add getLoadedContext Ports the two context APIs from #105 so the invalidation check in getBlock resolves the owning context synchronously. Co-Authored-By: Claude Fable 5.1 --- lib/context.js | 25 +++++++++++++++++++------ 1 file changed, 19 insertions(+), 6 deletions(-) diff --git a/lib/context.js b/lib/context.js index ad425c1..483513c 100644 --- a/lib/context.js +++ b/lib/context.js @@ -216,6 +216,18 @@ class CoreContext { } async getBlock(seq, core, config, origin) { + const buffer = await this.getBlockBuffer(seq, core, config, origin) + const block = decodeBlock(buffer, seq) + + if (block.invalidation > 0) { + const ctx = this.getLoadedContext(core) + if (block.invalidation > ctx.invalidation) await ctx.updateMaybe(config, 0, block) + } + + return block + } + + async getBlockBuffer(seq, core, config, origin) { if (core !== 0 && core - 1 >= this.cores.length) { await this.updateMaybe(config, core, origin) } @@ -226,12 +238,7 @@ class CoreContext { if (config.trace !== null) config.trace(core, seq) - const block = decodeBlock(buffer, seq) - - const ctx = core === 0 ? this : await this.getContext(core, config, origin) - if (block.invalidation > ctx.invalidation) await ctx.updateMaybe(config, 0, block) - - return block + return buffer } getLocalContext() { @@ -258,6 +265,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')