Add invalidation pointer to blocks

build.js

@@ -339,6 +339,10 @@ bee.register({
     {
       name: 't',
       type: 'uint'
+    },
+    {
+      name: 'invalidation', // TODO: in future block-2 should be required
+      type: 'uint'
     }
   ]
 })

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) {

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,19 +51,19 @@ 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)
     const cached = buf ? c.decode(c.uint, buf) : 0
 
     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) {
@@ -209,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)
     }
@@ -219,8 +238,7 @@ class CoreContext {
 
     if (config.trace !== null) config.trace(core, seq)
 
-    const block = decodeBlock(buffer, seq)
-    return block
+    return buffer
   }
 
   getLocalContext() {
@@ -247,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')

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
   }
 }
 

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) {

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
     }
   }
 }

spec/hyperschema/schema.json

@@ -404,6 +404,11 @@
           "name": "t",
           "type": "uint",
           "version": 1
+        },
+        {
+          "name": "invalidation",
+          "type": "uint",
+          "version": 1
         }
       ]
     }

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
+  }
+}