diff --git a/README.md b/README.md index c570bed..f0719bf 100644 --- a/README.md +++ b/README.md @@ -65,7 +65,8 @@ Options include: } ``` -If you pass your own `core`, consider increasing the inflight range on it (`256` minimum recommended, e.g. `{ inflightRange: [256, 512] }` when creating the core via Corestore) to match `hyperbee2` constructed cores. +Mappings are persisted next to the core table on the next write, so readers +that open the tree later resolve them without being given them. #### `await db.ready()` @@ -128,6 +129,24 @@ Read only. Boolean indicating whether the [Corestore][corestore] has closed. Calls `replicate` on underlying [Corestore][corestore]. +#### `await db.cores([options])` + +The keys of the cores the tree reads from, mapped: the core the root lives on +first, then the cores it references. Options: + +```js +{ + local: true, // Boolean. Include the local core + recursive: false, // Boolean. Also walk the core tables of every core reached + timeout: ..., // Number. Timeout for the checkpoint reads + wait: ... // Boolean. Wait for checkpoint blocks to download +} +``` + +With `recursive` the result covers everything the tree can reach through core +tables, one checkpoint read per core. It is a superset of what the current +root references, as tables only ever grow. + #### `db.checkout([options])` Returns a new Hyperbee as a view on the underlying [Hypercore][hypercore]. @@ -159,6 +178,8 @@ Options: } ``` +If `key` is mapped the core it maps to is opened instead. + #### `db.snapshot([options])` Returns a new Hyperbee that is a read only view of the current tree. @@ -202,6 +223,9 @@ Options: key: null, // Buffer or null. Key of the Hypercore. autoUpdate: true, // Boolean .Should Hyperbee automatically reflect updates // after each flush()? + mappings: null, // [{ from, to }]. Read core `from` as core `to`, which holds the + // same blocks, and write them into this checkpoint, even with + // nothing else to write. Readers of the tree learn them from it compat: false, // Boolean. Write blocks compatible with Hyperbee 1? type: encoding.TYPE_LATEST, // Number. Block format to use. inlineValueSize: 1024, // Values smaller than this byte length are diff --git a/build.js b/build.js index 9eca427..92a7378 100644 --- a/build.js +++ b/build.js @@ -1,6 +1,6 @@ const Hyperschema = require('hyperschema') -const schema = Hyperschema.from('./spec/hyperschema', { versioned: false }) +const schema = Hyperschema.from('./spec/hyperschema') const bee = schema.namespace('bee') bee.register({ @@ -228,6 +228,23 @@ bee.register({ ] }) +bee.register({ + name: 'mapping', + compact: true, + fields: [ + { + name: 'from', + type: 'fixed32', + required: true + }, + { + name: 'to', + type: 'fixed32', + required: true + } + ] +}) + bee.register({ name: 'block-0', fields: [ @@ -281,6 +298,11 @@ bee.register({ type: '@bee/core', array: true, required: true + }, + { + name: 'mappings', + type: '@bee/mapping', + array: true } ] }) diff --git a/index.js b/index.js index 8915cee..8fb6e63 100644 --- a/index.js +++ b/index.js @@ -196,21 +196,16 @@ class Hyperbee extends EventEmitter { this.emit('close') } - async cores({ local = true, ...opts } = {}) { + async cores({ local = true, recursive = false, ...opts } = {}) { const config = this.config.options(opts) await this.ready() - await this.context.update(config) - const cores = [] + const keys = recursive + ? (await this.context.getAllContexts(config)).map((c) => c.core.key) + : await this.context.getCoreKeys(config) - // index 0 is the core this bee is viewing, the rest are referenced cores - for (let i = 0; i <= this.context.cores.length; i++) { - const key = this.context.getCoreKey(i) - if (!local && b4a.equals(key, this.context.local.key)) continue - cores.push(key) - } - - return cores + if (local) return keys + return keys.filter((k) => !b4a.equals(k, this.context.local.key)) } createReadStream(options) { @@ -339,6 +334,8 @@ class Hyperbee extends EventEmitter { } _setRoot(root, emit) { + if (root !== null && root !== EMPTY) this.context = root.context + if (root === null || !root.equivalentTo(this.root)) { this.root = root if (emit) this.emit('update') diff --git a/lib/context.js b/lib/context.js index 483513c..98cf7be 100644 --- a/lib/context.js +++ b/lib/context.js @@ -4,6 +4,7 @@ const c = require('compact-encoding') const { BLOCK_NOT_AVAILABLE } = require('hypercore-errors') const { TreeNodePointer } = require('./tree.js') const { decodeBlock, TYPE_LATEST } = require('./encoding.js') +const Mappings = require('./mappings.js') const CHECKPOINT_KEY = 'hyperbee/checkpoint' @@ -14,6 +15,7 @@ class CoreContext { cache, core, getEncryptionProvider, + mappings = new Mappings(), lock = new ScopeLock(), other = new Map() ) { @@ -24,6 +26,7 @@ class CoreContext { this.getEncryptionProvider = getEncryptionProvider this.lock = lock this.other = other + this.mappings = mappings this.length = 0 this.checkpoint = 0 this.invalidation = 0 @@ -93,13 +96,17 @@ class CoreContext { const checkpoint = block.checkpoint let cores = null - if (checkpoint > 0) { + if (checkpoint > 0 && checkpoint !== this.checkpoint) { const seq = checkpoint - 1 const buffer = await this.core.get(seq, config) if (buffer === null) throw BLOCK_NOT_AVAILABLE() const block = decodeBlock(buffer, seq) cores = block.metadata ? block.metadata.cores : [] + if (block.metadata && block.metadata.mappings !== null) { + const persisted = b4a.equals(this.core.key, this.local.key) + this.mappings.add(block.metadata.mappings, { persisted }) + } } if (length <= this.length) return @@ -195,24 +202,57 @@ class CoreContext { getCore(index) { if (index === 0) return this.core if (index > this.cores.length) throw new Error('Bad core index: ' + index) - if (this.opened[index - 1] === null) { - const key = this.cores[index - 1].key - this.opened[index - 1] = this.store.get({ - key, - encryption: this.getEncryptionProvider(key), - inflightRange: [256, 512] - }) - } + + const key = this.mappings.resolve(this.cores[index - 1].key) + const opened = this.opened[index - 1] + if (opened !== null && (key === opened.key || b4a.equals(opened.key, key))) return opened + + this.opened[index - 1] = this._openCore(key) return this.opened[index - 1] } getCoreKey(index) { if (index === 0) return this.core.key if (index > this.cores.length) throw new Error('Bad core index: ' + index) - if (this.opened[index - 1] !== null && this.opened[index - 1].key) { - return this.opened[index - 1].key + return this.mappings.resolve(this.cores[index - 1].key) + } + + async getCoreKeys(config) { + await this.update(config) + + const keys = [] + for (let i = 0; i <= this.cores.length; i++) keys.push(this.getCoreKey(i)) + return keys + } + + async getAllContexts(config) { + await this.update(config) + + const all = [this] + const seen = new Set([b4a.toString(this.core.key, 'hex')]) + + for (let i = 0; i < all.length; i++) { + const pending = [] + + for (let j = 1; j <= all[i].cores.length; j++) { + const hex = b4a.toString(all[i].getCoreKey(j), 'hex') + if (seen.has(hex)) continue + seen.add(hex) + pending.push(loadContext(all[i], j, config)) + } + + for (const c of await Promise.all(pending)) all.push(c) } - return this.cores[index - 1].key + + return all + } + + _openCore(key) { + return this.store.get({ + key, + encryption: this.getEncryptionProvider(key), + inflightRange: [256, 512] + }) } async getBlock(seq, core, config, origin) { @@ -246,18 +286,13 @@ class CoreContext { } getContextByKey(key) { + key = this.mappings.resolve(key) if (b4a.equals(key, this.core.key)) return this const hex = b4a.toString(key, 'hex') if (this.other.has(hex)) return this.other.get(hex) - const ctx = this._createContext( - this.store.get({ - key, - encryption: this.getEncryptionProvider(key), - inflightRange: [256, 512] - }) - ) + const ctx = this._createContext(this._openCore(key)) this.other.set(hex, ctx) return ctx } @@ -273,18 +308,25 @@ class CoreContext { 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') + const hc = this.getCore(core) + const hex = b4a.toString(hc.key, 'hex') if (this.other.has(hex)) return this.other.get(hex) - const ctx = this._createContext(this.getCore(core)) + const ctx = this._createContext(hc) this.other.set(hex, ctx) return ctx } + needsFlush() { + return this.changed || this.mappings.dirty + } + flush() { + if (this.mappings.dirty) this.invalidation++ this.changed = false return { - cores: this.cores + cores: this.cores, + mappings: this.mappings.flush() } } @@ -293,10 +335,17 @@ class CoreContext { const local = this.local const cache = this.cache const getEncryptionProvider = this.getEncryptionProvider + const mappings = this.mappings const lock = this.lock const other = this.other - return new CoreContext(store, local, cache, core, getEncryptionProvider, lock, other) + return new CoreContext(store, local, cache, core, getEncryptionProvider, mappings, lock, other) } } +async function loadContext(context, index, config) { + const ctx = await context.getContext(index, config, null) + await ctx.update(config) + return ctx +} + module.exports = CoreContext diff --git a/lib/copy.js b/lib/copy.js index 395af72..a2d55bb 100644 --- a/lib/copy.js +++ b/lib/copy.js @@ -62,7 +62,7 @@ module.exports = async function copyChanges(tree, changes, config, map) { } } - if (local.changed) { + if (local.needsFlush()) { local.checkpoint = base + blocks.length blocks[blocks.length - 1].metadata = local.flush() } @@ -71,6 +71,7 @@ module.exports = async function copyChanges(tree, changes, config, map) { for (let i = 0; i < blocks.length; i++) { blocks[i].checkpoint = local.checkpoint + blocks[i].invalidation = local.invalidation buffers[i] = encodeBlock(blocks[i]) } diff --git a/lib/encoding.js b/lib/encoding.js index 6b90043..e380dd6 100644 --- a/lib/encoding.js +++ b/lib/encoding.js @@ -85,7 +85,7 @@ function decodeBlock0(state, seq) { checkpoint: blk.checkpoint, batch: blk.batch, previous: blk.previous, - metadata: { cores: blk.cores }, + metadata: { cores: blk.cores, mappings: null }, tree, keys: blk.data.map(dataToKey), values: null, diff --git a/lib/mappings.js b/lib/mappings.js new file mode 100644 index 0000000..2552176 --- /dev/null +++ b/lib/mappings.js @@ -0,0 +1,40 @@ +const b4a = require('b4a') + +class Mappings { + constructor() { + this.known = new Map() + this.dirty = false + this._flushed = null + } + + resolve(key) { + if (this.known.size === 0) return key + const known = this.known.get(b4a.toString(key, 'hex')) + return known === undefined ? key : known + } + + add(mappings, { persisted = false } = {}) { + for (const { from, to } of mappings) { + if (b4a.equals(from, to)) continue + const hex = b4a.toString(from, 'hex') + if (this.known.has(hex)) continue + this.known.set(hex, to) + this._flushed = null + if (!persisted) this.dirty = true + } + } + + flush() { + this.dirty = false + if (this.known.size === 0) return null + + if (this._flushed === null) { + this._flushed = [] + for (const [hex, to] of this.known) this._flushed.push({ from: b4a.from(hex, 'hex'), to }) + } + + return this._flushed + } +} + +module.exports = Mappings diff --git a/lib/write.js b/lib/write.js index bfd2832..5d0d39e 100644 --- a/lib/write.js +++ b/lib/write.js @@ -14,6 +14,7 @@ module.exports = class WriteBatch { const { length = -1, key = null, + mappings = null, autoUpdate = true, compat = false, type = compat ? TYPE_COMPAT : TYPE_LATEST, @@ -39,6 +40,7 @@ module.exports = class WriteBatch { this.applied = 0 this.root = null this.ops = [] + this.mappings = mappings } tryPut(key, value) { @@ -356,6 +358,16 @@ module.exports = class WriteBatch { } async _flush() { + const context = this.tree.context.getLocalContext() + const config = this.tree.config + + await context.update(config) + + if (this.mappings !== null) { + context.mappings.add(this.mappings) + if (this.root && !this.root.changed && context.mappings.dirty) this.root.mark() + } + if (!this.root || !this.root.changed) { return } @@ -374,11 +386,6 @@ module.exports = class WriteBatch { const stack = [this.root] const values = [] - const context = this.tree.context.getLocalContext() - const config = this.tree.config - - await context.update(config) - while (stack.length > 0) { const node = stack.pop() @@ -424,7 +431,8 @@ module.exports = class WriteBatch { const length = context.core.length // if noop and not genesis, bail early - if (this.applied === 0 && this.length > 0) return + const mapping = this.mappings !== null && context.mappings.dirty + if (this.applied === 0 && this.length > 0 && !mapping) return if (minValue > -1 && minValue + update.size < this.preferredBlockSize) { // TODO: repack the value into the block @@ -525,13 +533,14 @@ module.exports = class WriteBatch { } // TODO: make this transaction safe - if (context.changed) { + if (context.needsFlush()) { context.checkpoint = context.core.length + blocks.length blocks[blocks.length - 1].metadata = context.flush() } for (let i = 0; i < blocks.length; i++) { blocks[i].checkpoint = context.checkpoint + blocks[i].invalidation = context.invalidation buffers[i] = encodeBlock(blocks[i]) } diff --git a/spec/hyperschema/index.js b/spec/hyperschema/index.js index 3093d62..3a9e555 100644 --- a/spec/hyperschema/index.js +++ b/spec/hyperschema/index.js @@ -1,12 +1,12 @@ // This file is autogenerated by the hyperschema compiler -// Schema Version: 1 +// Schema Version: 2 /* eslint-disable camelcase */ /* eslint-disable quotes */ /* eslint-disable space-before-function-paren */ const { c } = require('hyperschema/runtime') -const VERSION = 1 +const VERSION = 2 // eslint-disable-next-line no-unused-vars let version = VERSION @@ -423,15 +423,25 @@ const encoding13_0 = encoding12_7 const encoding13 = { preencode(state, m) { encoding13_0.preencode(state, m.cores) + state.end++ // max flag is 1 so always one byte + + if (version >= 2 && m.mappings) encoding13_1.preencode(state, m.mappings) }, encode(state, m) { + const flags = version >= 2 && m.mappings ? 1 : 0 + encoding13_0.encode(state, m.cores) + c.uint.encode(state, flags) + + if (version >= 2 && m.mappings) encoding13_1.encode(state, m.mappings) }, decode(state) { const r0 = encoding13_0.decode(state) + const flags = state.start < state.end ? c.uint.decode(state) : 0 return { - cores: r0 + cores: r0, + mappings: version >= 2 && (flags & 1) !== 0 ? encoding13_1.decode(state) : null } } } @@ -522,6 +532,30 @@ const encoding14 = { } } +// @bee/mapping +const encoding15 = { + preencode(state, m) { + c.fixed32.preencode(state, m.from) + c.fixed32.preencode(state, m.to) + }, + encode(state, m) { + c.fixed32.encode(state, m.from) + c.fixed32.encode(state, m.to) + }, + decode(state) { + const r0 = c.fixed32.decode(state) + const r1 = c.fixed32.decode(state) + + return { + from: r0, + to: r1 + } + } +} + +// @bee/metadata.mappings, deferred due to recusive use +const encoding13_1 = c.array(encoding15) + function setVersion(v) { version = v } @@ -575,6 +609,8 @@ function getEncoding(name) { return encoding13 case '@bee/block-1': return encoding14 + case '@bee/mapping': + return encoding15 default: throw new Error('Encoder not found ' + name) } diff --git a/spec/hyperschema/schema.json b/spec/hyperschema/schema.json index 309401c..1fe05a6 100644 --- a/spec/hyperschema/schema.json +++ b/spec/hyperschema/schema.json @@ -1,5 +1,5 @@ { - "version": 1, + "version": 2, "schema": [ { "name": "tree-pointer-0", @@ -325,7 +325,7 @@ "name": "metadata", "namespace": "bee", "compact": false, - "flagsPosition": -1, + "flagsPosition": 1, "fields": [ { "name": "cores", @@ -333,6 +333,12 @@ "array": true, "type": "@bee/core", "version": 1 + }, + { + "name": "mappings", + "array": true, + "type": "@bee/mapping", + "version": 2 } ] }, @@ -411,6 +417,26 @@ "version": 1 } ] + }, + { + "name": "mapping", + "namespace": "bee", + "compact": true, + "flagsPosition": -1, + "fields": [ + { + "name": "from", + "required": true, + "type": "fixed32", + "version": 2 + }, + { + "name": "to", + "required": true, + "type": "fixed32", + "version": 2 + } + ] } ] } diff --git a/test/all.js b/test/all.js index 57f28cd..06f26fd 100644 --- a/test/all.js +++ b/test/all.js @@ -17,6 +17,7 @@ async function runTests() { await import('./fuzz-diff.js') await import('./fuzz-multiwriter.js') await import('./perf.js') + await import('./map-core.js') await import('./prefetch.js') await import('./tree.js') await import('./undo.js') diff --git a/test/basic.js b/test/basic.js index c243efa..0e349c1 100644 --- a/test/basic.js +++ b/test/basic.js @@ -698,7 +698,8 @@ test('reindex preserves batches across splits, value blocks and cores', async fu t.is(remote.length, 5) const source = b.head() - t.absent(b4a.equals(source.key, b.core.key), 'b is viewing c, its writes live in its local core') + t.alike(source.key, b.context.local.key, 'b built on c, its writes live in its local core') + t.absent(b4a.equals(source.key, genesis.key)) // stop at c's change, so the copies keep pointing into c's core for untouched nodes a.move(source) diff --git a/test/map-core.js b/test/map-core.js new file mode 100644 index 0000000..9831182 --- /dev/null +++ b/test/map-core.js @@ -0,0 +1,385 @@ +const test = require('brittle') +const b4a = require('b4a') +const Corestore = require('corestore') +const Hypercore = require('hypercore') +const Bee = require('../') + +const hex = (k) => b4a.toString(k, 'hex') +const key = (i) => b4a.from('key/' + String(i).padStart(4, '0')) +const value = (i) => b4a.from('value-' + i) + +async function source(store, n) { + const a = new Bee(store.namespace('a')) + await a.ready() + const w = a.write() + for (let i = 0; i < n; i++) w.tryPut(key(i), value(i)) + await w.flush() + return a +} + +// a core holding the same blocks, the target of a mapping +async function copyCore(store, src, name) { + const dst = store.get({ name }) + await dst.ready() + for (let i = 0; i < src.length; i++) await dst.append(await src.get(i)) + return dst +} + +async function readAll(bee, n, t, msg) { + for (let i = 0; i < n; i++) { + const node = await bee.get(key(i)) + if (!node || !b4a.equals(node.value, value(i))) return t.fail(msg + ': key ' + i) + } + t.pass(msg) +} + +// a writer whose tree reaches a, and a copy of a to map it to +async function setup(store, n = 50) { + const a = await source(store, n) + + const w = new Bee(store.namespace('w')) + await w.ready() + w.move({ key: a.core.key, length: a.core.length }) + { + const batch = w.write() + batch.tryPut(key(n), value(n)) + await batch.flush() + } + + const copy = await copyCore(store, a.core, 'copy') + const mappings = [{ from: a.core.key, to: copy.key }] + + return { a, w, copy, mappings } +} + +test('mappings - a write with mappings records them, readers read through them', async function (t) { + const store = new Corestore(await t.tmp()) + t.teardown(() => store.close()) + + const { a, w, copy, mappings } = await setup(store) + const local = w.context.local + const length = local.length + + await w.write({ mappings }).flush() + t.is(local.length, length + 1, 'with nothing else to write, one block') + t.alike(w.head(), { key: local.key, length: local.length }, 'the head moved onto it') + + await w.write({ mappings }).flush() + t.is(local.length, length + 1, 'mappings already written write nothing') + await w.close() + + await a.core.clear(0, a.core.length) + + const r = new Bee(store.namespace('w'), { writable: false }) + await r.ready() + await readAll(r, 51, t, 'a reader reads through the mappings of the checkpoint') + t.alike((await r.cores({ local: false })).map(hex), [hex(copy.key)], 'and lists the copy only') + await r.close() + + const again = new Bee(store.namespace('w')) + await again.ready() + await again.write({ mappings }).flush() + t.is(again.context.local.length, length + 1, 'nor after a reopen') + await again.close() +}) + +test('mappings - a write with ops carries the mappings in its checkpoint', async function (t) { + const store = new Corestore(await t.tmp()) + t.teardown(() => store.close()) + + const { a, w, copy, mappings } = await setup(store) + const length = w.context.local.length + + const batch = w.write({ mappings }) + batch.tryPut(key(51), value(51)) + await batch.flush() + t.is(w.context.local.length, length + 1, 'one block for both') + t.absent( + (await w.cores({ local: false })).some((k) => b4a.equals(k, a.core.key)), + 'the original is not among the cores read' + ) + await w.close() + + await a.core.clear(0, a.core.length) + + const r = new Bee(store.namespace('w'), { writable: false }) + await r.ready() + await readAll(r, 52, t, 'a reader reads everything through the copy') + t.ok((await r.cores()).some((k) => b4a.equals(k, copy.key))) + await r.close() +}) + +test('mappings - an undo below the record reads through the mappings of the head', async function (t) { + const store = new Corestore(await t.tmp()) + t.teardown(() => store.close()) + + const { a, w, copy, mappings } = await setup(store) + for (let i = 51; i < 55; i++) { + const batch = w.write() + batch.tryPut(key(i), value(i)) + await batch.flush() + } + await w.write({ mappings }).flush() + await w.close() + + await a.core.clear(0, a.core.length) + + const r = new Bee(store.namespace('w'), { writable: false }) + await r.ready() + await readAll(r, 55, t, 'the head reads through the copy') + + const old = r.undo(4) + await readAll(old, 52, t, 'an undone root reads through the mappings the head carried') + t.is(await old.get(key(54)), null, 'the undone writes are gone') + t.alike((await old.cores({ local: false })).map(hex), [hex(copy.key)]) + await old.close() + await r.close() +}) + +test('mappings - a mapping met in another writer is recorded locally', async function (t) { + const store = new Corestore(await t.tmp()) + t.teardown(() => store.close()) + + const { a, w, mappings } = await setup(store) + await w.write({ mappings }).flush() + + await a.core.clear(0, a.core.length) + + // c is never given the mapping and only ever sees w's head + const c = new Bee(store.namespace('c')) + await c.ready() + c.move({ key: w.context.local.key, length: w.context.local.length }) + { + const batch = c.write() + batch.tryPut(key(51), value(51)) + await batch.flush() + } + await readAll(c, 52, t, 'c reads a through the mapping w recorded') + await c.close() + await w.close() + + const again = new Bee(store.namespace('c'), { writable: false }) + await again.ready() + await readAll(again, 52, t, "a reopen of c reads a through c's own checkpoint") + t.absent( + (await again.cores()).some((k) => b4a.equals(k, a.core.key)), + 'the original is not opened anywhere' + ) + await again.close() +}) + +test('mappings - a remote reads the whole tree after the original data is gone', async function (t) { + const origin = new Corestore(await t.tmp()) + t.teardown(() => origin.close()) + + const { a, w, copy, mappings } = await setup(origin, 200) + const wKey = w.context.local.key + const before = w.context.local.length + await w.write({ mappings }).flush() + await w.close() + + await a.core.clear(0, a.core.length) + + const remote = new Corestore(await t.tmp()) + t.teardown(() => remote.close()) + const stop = replicateStores(origin, remote) + + const reader = new Bee(remote.session(), { key: wKey, writable: false }) + await reader.ready() + await reader.core.update({ wait: true }) + + let count = 0 + for await (const { key: k, value: v } of reader.createReadStream()) { + const expected = b4a.equals(k, key(count)) && b4a.equals(v, value(count)) + if (!expected) return t.fail('bad entry ' + count) + count++ + } + t.is(count, 201, 'the whole tree was read through the mapping') + t.alike( + (await reader.cores()).map(hex).sort(), + [hex(wKey), hex(copy.key)].sort(), + 'only the writer and the copy were opened' + ) + t.absent( + await remote.storage.hasCore(Hypercore.discoveryKey(a.core.key)), + 'the remote never asked for the original' + ) + await reader.close() + await stop() + + // a head from before the record has a table without the mappings + const other = new Corestore(await t.tmp()) + t.teardown(() => other.close()) + const stopOther = replicateStores(origin, other) + { + const core = other.get({ key: wKey }) + await core.update({ wait: true }) + await core.download({ start: 0, end: core.length }).done() + await core.close() + } + + const old = new Bee(other.session(), { key: wKey, writable: false }) + await old.ready() + old.move({ length: before }) + await t.exception( + old.get(key(0), { wait: false }), + /BLOCK_NOT_AVAILABLE/, + 'before the record the original is read' + ) + await old.close() + await stopOther() +}) + +test('mappings - move, checkout and write onto an original resolve once known', async function (t) { + const store = new Corestore(await t.tmp()) + t.teardown(() => store.close()) + + const { a, w, copy, mappings } = await setup(store) + await w.write({ mappings }).flush() + await a.core.clear(0, a.core.length) + + const view = w.checkout({ key: a.core.key, length: a.core.length }) + t.alike(view.head().key, copy.key, 'a checkout onto the original sits on the copy') + await readAll(view, 50, t, 'and reads through it') + await view.close() + + const batch = w.write({ key: a.core.key, length: a.core.length }) + batch.tryPut(key(50), value(50)) + await batch.flush() + await readAll(w, 51, t, 'a batch keyed by the original builds on the copy') + + w.move({ key: a.core.key, length: a.core.length }) + t.alike(w.core.key, copy.key, 'a move onto the original lands on the copy') + await w.close() +}) + +test('mappings - a read in flight survives its core being remapped', async function (t) { + const origin = new Corestore(await t.tmp()) + t.teardown(() => origin.close()) + + const { a, w, copy, mappings } = await setup(origin) + const wKey = w.context.local.key + + // the remote holds the writer but nothing of the core it references + const remote = new Corestore(await t.tmp()) + t.teardown(() => remote.close()) + { + const stop = replicateStores(origin, remote) + const core = remote.get({ key: wKey }) + await core.update({ wait: true }) + await core.download({ start: 0, end: core.length }).done() + await core.close() + await stop() + } + + const reader = new Bee(remote.session(), { key: wKey, writable: false }) + await reader.ready() + + const read = reader.get(key(0)) + for (let i = 0; i < 100 && !reader.context.opened[0]; i++) await immediate() + const stale = reader.context.opened[0] + t.alike(stale.key, a.core.key, 'the read opened the core as referenced') + + // the writer records the mapping, the reader moves onto that head + await w.write({ mappings }).flush() + const stop = replicateStores(origin, remote) + await reader.core.update({ wait: true }) + reader.move({ length: reader.core.length }) + await reader.get(key(50)) + + t.alike(reader.context.getCore(1).key, copy.key, 'the next open resolves to the copy') + t.absent(stale.closing, 'the superseded session is left to the reads holding it') + + t.alike((await read).value, value(0), 'the read issued before the remap still resolved') + await readAll(reader, 51, t, 'later reads go through the mapping') + + await reader.close() + await w.close() + await stop() +}) + +test('mappings - cores({ recursive: true }) walks the tables of every core reached', async function (t) { + const store = new Corestore(await t.tmp()) + t.teardown(() => store.close()) + + // b links a, then clears and rewrites the tree: nothing in b's tree points + // into a any more but b's table still names it, and c builds on b + const a = await source(store, 50) + + const b = new Bee(store.namespace('b')) + await b.ready() + b.move({ key: a.core.key, length: a.core.length }) + { + const batch = b.write() + batch.tryPut(key(50), value(50)) + await batch.flush() + } + { + const batch = b.write() + batch.tryClear() + for (let i = 0; i < 50; i++) batch.tryPut(key(i), value(100 + i)) + await batch.flush() + } + + const c = new Bee(store.namespace('c')) + await c.ready() + c.move({ key: b.context.local.key, length: b.context.local.length }) + { + const batch = c.write() + batch.tryPut(key(50), value(50)) + await batch.flush() + } + + t.alike( + (await c.cores({ local: false })).map(hex), + [hex(b.context.local.key)], + "c's table names b only" + ) + t.alike( + (await c.cores({ local: false, recursive: true })).map(hex).sort(), + [hex(a.core.key), hex(b.context.local.key)].sort(), + 'recursively a is reached through b' + ) + + // find what is reachable, copy it, record it + const copy = await copyCore(store, a.core, 'copy') + await a.core.clear(0, a.core.length) + await c.write({ mappings: [{ from: a.core.key, to: copy.key }] }).flush() + + t.alike( + (await c.cores({ local: false, recursive: true })).map(hex).sort(), + [hex(copy.key), hex(b.context.local.key)].sort(), + 'the walk reports the copy' + ) + for (let i = 0; i < 50; i++) { + const node = await c.get(key(i)) + if (!node || !b4a.equals(node.value, value(100 + i))) return t.fail('bad value for key ' + i) + } + t.alike((await c.get(key(50))).value, value(50), 'every entry still reads') + + await c.close() + await b.close() +}) + +function replicateStores(a, b) { + const s1 = a.replicate(true) + const s2 = b.replicate(false) + s1.pipe(s2).pipe(s1) + s1.on('error', noop) + s2.on('error', noop) + return async () => { + const closed = Promise.all([ + new Promise((resolve) => s1.once('close', resolve)), + new Promise((resolve) => s2.once('close', resolve)) + ]) + s1.destroy() + s2.destroy() + await closed + } +} + +function immediate() { + return new Promise((resolve) => setImmediate(resolve)) +} + +function noop() {} diff --git a/test/undo.js b/test/undo.js index 7bfc503..28e9a69 100644 --- a/test/undo.js +++ b/test/undo.js @@ -1,5 +1,7 @@ const test = require('brittle') const b4a = require('b4a') +const Corestore = require('corestore') +const Bee = require('../') const { create } = require('./helpers') test('undo', async function (t) { @@ -164,3 +166,39 @@ test('move', async function (t) { ]) } }) + +test('undo after writing on top of another core', async function (t) { + const store = new Corestore(await t.tmp()) + t.teardown(() => store.close()) + + const a = new Bee(store.namespace('a')) + await a.ready() + for (const k of ['a', 'b']) { + const w = a.write() + w.tryPut(b4a.from(k), b4a.from(k)) + await w.flush() + } + + const b = new Bee(store.namespace('b')) + await b.ready() + b.move({ key: a.core.key, length: a.core.length }) + for (const k of ['c', 'd']) { + const w = b.write() + w.tryPut(b4a.from(k), b4a.from(k)) + await w.flush() + } + + t.alike(b.core.key, b.context.local.key, 'the tree views the core it wrote to') + + const head = b.checkout() + t.alike((await head.get(b4a.from('d'))).value, b4a.from('d'), 'a checkout sees the writes') + await head.close() + + const undone = b.undo(1) + t.alike((await undone.get(b4a.from('c'))).value, b4a.from('c'), 'undo keeps the earlier write') + t.is(await undone.get(b4a.from('d')), null, 'and drops the last one') + await undone.close() + + await a.close() + await b.close() +})