holepunchto / autobase Goto Github PK

I feel like I have fully grasped what inputs, localInput and localOutput does. But, I am having a really hard time grasping on what exactly multiple outputs are needed for. Having one makes sense to me. In my mind, inputs converge data from multiple sources. localInput allows for the 'local client' to participate in the convergence, the apply processes what converged from the inputs at a given clock as a view, localOutput persists the view into the output.... but at this point, doesn't this mean that all outputs are pretty much the same? Why would I need to have multiple output cores that will just get the same persisted view? Is there a way or going to be a way to corral certain views to different outputs? I've looked through the examples and tests, but still for the life of me wrap my mind around what I need the multiple outputs for. Thank you for any advice.

tl;dr My intuition of it is, the inputs converge to an assimilated data structure, but I don't see mechanisms to diverge outputs to distinctive data structures.

I would love to know more about why your not using the Readable Writeable webstream implementations

they offer everything that you implemented here with a nice generic cross platform compatible api so i guess i do not get what are the benefits of this.

you could write your clocking abstractions which i by the way do not like as the engine offers clockings this can be done with a simple transform stream they exist also on the web.

it would help me to understand your design process. as we have a similar product but i wonder why you choose to abstract everything and not depend on the generic offered apis?

Running the sample code provided in the readme resulted in the following error: TypeError: Cannot read properties of null (reading 'get'). It appears that localOutput must be defined to get the linearized view via base.view.get(). Would love to submit a PR for any of the following to help with developer experience

• Update sample code to include localOutput for completeness
• Throw error in base.view.update() if no output is defined (wasn't obvious why update returns but get throws?)
• Default in-memory value for localOutput if not defined

import Autobase from "autobase";
import Hypercore from "hypercore";
import ram from "random-access-memory";

const inputA = new Hypercore(ram);
const inputB = new Hypercore(ram);
const inputC = new Hypercore(ram);
const base = new Autobase({
  inputs: [inputA, inputB, inputC],
  localInput: inputA,
  autostart: true,
});
await base.append("hello");
await base.append("world");
// view.update() returns
await base.view.update();
// view.get() throws
await base.view.get(0);

Moved from hypercore-protocol/hypercore-next#59

Repro repo: https://github.com/pfrazee/hyper-replication-bug

Something is failing in the replication code causing the program to end prematurely and without any information.

What's happening in the repro repo is that I'm creating a set of "Nodes" which are corestores, a set of cores, and autobases. I'm then randomly connecting and disconnecting them using core.replicate(), and also creating put() or del() operations. There are no reads occurring yet, so the rebased-hypercore index isnt being touched yet (no apply calls).

The failure seems to occur during an oplog append, and I traced it as follows:

• head() is calling _getInputNode() on a remote core,
• which in turn is calling core.get().
• That cache misses,
• so it calls out to the replicator which is creating a request.

At that point, you get into the complexities of the request code and I figured it'd be better to pass this off to yall.

Another interesting piece of info: if I connect all of my "nodes" corestores in replication, then the error doesn't occur

I get a HypercoreError: SESSION_CLOSED error when I .close() and then .ready() an autobase instance. It appears the Hypercores are not automatically re-opened when restarting autobase.

HypercoreError: SESSION_CLOSED: Cannot make sessions on a closing core
    at Hypercore.session (/Documents/GitHub.nosync/nebula/node_modules/hypercore/index.js:197:13)
    at Autobase._addInput (/Documents/GitHub.nosync/nebula/node_modules/autobase/index.js:83:27)
    at Autobase._open (/Documents/GitHub.nosync/nebula/node_modules/autobase/index.js:65:12)
    at processTicksAndRejections (node:internal/process/task_queues:96:5) {
  code: 'SESSION_CLOSED'
}

const firstUser = new Hypercore('./core1')
const firstOutput = new Hypercore('./core2')

const inputs = [firstUser]

const base = new Autobase({
  inputs,
  localInput: firstUser,
  localOutput: firstOutput
})

await base.ready()

await base.close()

await base.ready()

await base.append(Buffer.from('foo')) // This breaks

Got the following error message and did some sleuthing but don't know enough about how everything works to fix it:

.../node_modules/autobase/lib/linearize.js:588
    if (!this.lastUpdate) await this.ready()
                                     ^

TypeError: this.ready is not a function
    at LinearizedCore.get (.../node_modules/autobase/lib/linearize.js:588:38)
    at LinearizedCoreSession.get (.../node_modules/autobase/lib/linearize.js:778:34)
    at Batch.getBlock (.../node_modules/hyperbee/index.js:540:35)
    at Batch.getRoot (.../node_modules/hyperbee/index.js:528:24)
    at processTicksAndRejections (node:internal/process/task_queues:96:5)
    at async TreeIterator.open (.../node_modules/hyperbee/iterators/diff.js:81:18)
    at async Promise.all (index 1)
    at async DiffIterator.open (.../node_modules/hyperbee/iterators/diff.js:156:5)

My Sleuthing

I took a look at the LinearizedCore class and didn't see a ready function and couldn't find where it would be added dynamically. So I git blamed the offending line to find it was added here:

const base = new Autobase(store, {
  apply: async (batch, view) => {
    throw Error('and i did throw')
  }
})
await base.append('foo')
// Ensure apply ran
await base.update()
// Never throws

The error isn't thrown even when no error event is registered.

I think I tracked it down to this:

1. Error is thrown and caught calling ._onError(err) in catch here:
   https://github.com/holepunchto/autobase/blob/main/index.js#L1426
   This calls .close() starting the closing process which sets .closing and then rethrows via EventEmitter's throwing on unhandled error events.
2. The rethrow is caught here:
   https://github.com/holepunchto/autobase/blob/main/index.js#L905
   Again this calls ._onError(err) but now .closing is truthy causing it to return before calling .emit('error', err) again.
   https://github.com/holepunchto/autobase/blob/main/index.js#L339

Every now and then I get keys is not iterable error when starting up Autobase. After the error starts showing up, it comes up on every subsequent run, which sounds to me that one of the cores is somehow corrupted and truncate() can't handle it.

The error...

Error: TypeError: keys is not iterable
    at KeyCompressor.truncate (../autobase/lib/compression.js:79:25)
    at Output.truncate (../autobase/lib/output.js:30:21)
    at LinearizedCore._commitUpdate (../autobase/lib/linearize.js:500:27)
    at processTicksAndRejections (node:internal/process/task_queues:96:5)
    at async LinearizedCore._rebuild (../autobase/lib/linearize.js:473:7)
    at async LinearizedCore._update (../autobase/lib/linearize.js:534:5)

The issue seems to be that this.bySeq.get() returns null. The underlying core is ready, so should be good to go.

truncate (length) {
    const badSeqs = []
    for (const seq of this.bySeq.keys()) {
      if (seq < length) continue
      badSeqs.push(seq)
    }
    for (const seq of badSeqs) {
      const keys = this.bySeq.get(seq)
      this.bySeq.delete(seq)
      for (const key of keys) {
        this.byKey.delete(b.toString(key, 'hex'))
      } 
    }
  }

Are there any side-effects if I add a simple check whether keys is null?

I've been trying to get Autobase to work in a browser environment and I discovered a bug of some sorts.

Test code

const RAM = require("random-access-memory")
const Corestore = require("corestore")
const Hyperbee = require("hyperbee")
const Autobase = require("autobase")

;(async () => {
  const store = new Corestore(RAM)
  const writer = await store.get({ name: "writer" })
  const index = await store.get({ name: "index" })
  await writer.ready()
  await index.ready()

  const autobase = new Autobase({
    localInput: writer,
    inputs: [writer],
    outputs: [],
    localOutput: index,
    autostart: true,
    unwrap: true,
    async apply(batch) {
      console.log("apply")
    },
  })
  await autobase.ready()

  const db = new Hyperbee(autobase.view, {
    extension: false,
    keyEncoding: "utf-8",
    valueEncoding: "binary",
  })

  await autobase.append('hello')
  await autobase.view.update()
})()

node test.js outputs "apply" as it gets to the apply function.

Compiled with npx browserify test.js > test.build.js and running it in browser outputs this error:

Uncaught (in promise) TypeError: Cannot read properties of null (reading 'clock')
    at Linearizer.update (test.build.js:4130:26)
    at async LinearizedView._update (test.build.js:4288:5)
    at async LinearizedView.debounced [as update] (test.build.js:7113:14)
    at async test.build.js:24032:3

I'm using a rebased index with Hyperbee, and when I attempt to bee.createHistoryStream({ live: true }) I get the following error:

node:events:346
      throw er; // Unhandled 'error' event
      ^

TypeError: Cannot read property 'value' of undefined
    at RebasedHypercore.get (/Users/gh/Documents/tests/autobase/node_modules/autobase/lib/rebase.js:216:19)
    at processTicksAndRejections (node:internal/process/task_queues:94:5)
    at async HyperBee.getBlock (/Users/gh/Documents/tests/autobase/node_modules/hyperbee/index.js:357:21)
    at async Batch.getBlock (/Users/gh/Documents/tests/autobase/node_modules/hyperbee/index.js:527:9)
    at async HistoryIterator.next (/Users/gh/Documents/tests/autobase/node_modules/hyperbee/iterators/history.js:32:18)
Emitted 'error' event on Readable instance at:
    at ReadableState.afterDestroy (/Users/gh/Documents/tests/autobase/node_modules/streamx/index.js:442:19)
    at Readable._destroy (/Users/gh/Documents/tests/autobase/node_modules/streamx/index.js:535:5)
    at ReadableState.updateNonPrimary (/Users/gh/Documents/tests/autobase/node_modules/streamx/index.js:349:16)
    at ReadableState.update (/Users/gh/Documents/tests/autobase/node_modules/streamx/index.js:333:69)
    at ReadableState.afterRead (/Users/gh/Documents/tests/autobase/node_modules/streamx/index.js:472:58)

Example code used:

const Hyperbee = require("hyperbee")
const Autobase = require("autobase")
const Corestore = require("corestore")
const ram = require("random-access-memory")

const store = new Corestore(ram)
const writer = store.get({ name: 'writer' })
const base = new Autobase([writer], { input: writer })

const index = base.createRebasedIndex({
  unwrap: true,
  apply: _apply.bind(this)
})

const hb = new Hyperbee(index, {
  keyEncoding: 'utf-8',
  valueEncoding: 'json',
  extension: false
})

const stream = hb.createHistoryStream({ live: true })

stream.on('data', data => {
  // do something
})

await put('hi', { hello: 'world' })

async function put(key, value, opts) {
  const op = Buffer.from(JSON.stringify({ type: 'put', key, value }))
  return await base.append(op, opts)
}

async function _apply(batch) {
  const b = hb.batch({ update: false })
  for (const node of batch) {
    const op = JSON.parse(node.value.toString())
    if (op.type === 'put') await b.put(op.key, op.value)
  }
  await b.flush()
}

I showed this to @mafintosh, who said it's because the history stream is reading out of bounds on the Hypercore.