<commit_before># Epidemic Broadcast Trees

This module is loosely based on plumtree Epidemic Broadcast Trees
[EBT paper], but adapted to also replicate logs, and optimized
to achive a minimal overhead (the cost of the protocol is linear with
the number of messages to be sent)

It's a algorithm that combines the robustness of a flooding epidemic
gossip broadcast, with the efficiency of a tree model. It's intended
for implementing realtime protocols (such as chat, scuttlebutt, also
radio/video) over networks with random topology - or networks where
otherwise peers may be unable to all connect to each other or to a
central hub.

Although the primary motivation for this module is to use it in secure
scuttlebutt, it's intended to be decoupled sufficiently to use for
other applications.

## Example

implement a simple in memory log replicator.

``` js
var clocks = {}
var logs = {}

function append (msg, cb) {
  var log = logs[msg.author] || {}
  //check that this is the next expected message.
  if(msg.sequence != Object.keys(log).length + 1)
    cb(new Error('out of order, found:'+msg.sequence+', expected:'+log.length))
  else {
    log[msg.sequence] = msg
    ebt.onAppend(msg)
    cb()
  }
}

var ebt = EBT({
  //NOTE: in this example, we are using readable strings for clarity
  //but ideally you'd use cryptographic ids, like public keys.
  id: 'alice',
  getClock: function (id, cb) {
    //load the peer clock for id.
    cb(null, clocks[id] || {})
  },
  setClock: function (id, clock) {
    //set clock doesn't have take a cb, but it's okay to be async.
    clocks[id] = clock
  },
  getAt: function (pair, cb) {
    //load a message particular message, by id:sequence
    if(!logs[pair.id] || !logs[pair.id][pair.sequence])
      cb(new Error('not found'))
    else
      cb(null, logs[pair.id][pair.sequence])
  },
  append: append
})

ebt.append({
  author: 'alice', sequence: 1, content: {}
}, function () {})

//must explicitly say we are replicating which peers.
ebt.request('alice', true)
ebt.request('bob', true)

//create a stream and pipe it to another instance
//isClient and version are required.
var stream = ebt.createStream('bob', version=3, isClient = true)
stream.pipe(remote_stream).pipe(stream)
```

> note about push-stream: push-stream is only new, so you'll probably
  need to convert this to a pull-stream to connect stream to a network
  io stream and serialization

``` js
var pushToPull = require('push-stream-to-pull-stream')
var stream = pushToPull(ebt.createStream(remote_id, 3, isCient = true))
pull(stream, remote_pull_stream, stream)
```

## API

### EBT(opts) => ebt

where opts provides the necessary things to connect ebt
to your system.

```
opts = {
  id: string,
  timeout: 3000, //default,
  getClock: function (id, cb),
  setClock: function (id, clock),
  getAt: function ({id:string, sequence:number}, cb),
  append: function (msg, cb),
  isFeed: function (id),
  isMsg: function(data),
  getMsgAuthor: function(msg),
  getMsgSequence: function(msg)
}
```

Create a new EBT instance. `id` is a unique identifier of the current
peer.  In [secure-scuttlebutt](https://scuttlebutt.nz) this is a
ed25519 public key.

`getClock(id, cb)` and `setClock(id, clock)` save a peer's clock
object.  This is used to save bandwidth when reconnecting to a peer
again.

`getAt({id, sequence}, cb)` retrives a message in a feed and an
sequence.  messages must have `{author, sequence, content}` fields.

`append(msg, cb)` append a particular message to the log.

`timeout` is used to decide when to switch a feed to another peer.
This is essential to detecting when a peer may have stalled.

`isFeed(id)` is a validation function that returns true if `id` is a
valid feed identifier. If not, it is ignored'

`isMsg(id)` is a validation function used to distinguish between data
messages and status messages. A message must contain an `author` field
that corresponds to the feed identifier and a `sequence` field. The
field is optional for backwards compatibility.

`getMsgAuthor(msg)` is a function that given a message returns the
author. The field is optional for backwards compatibility.

`getMsgSequence(msg)` is a function that given a message returns the
sequence. The field is optional for backwards compatibility.

### ebt.onAppend (msg)

When a message is appended to the database, tell ebt about it.  this
must be called whenever a message is successfully appended to the
database.

### ebt.createStream(id, version, isClient) => PushStream

Create a stream for replication. returns a [push-stream]. The current
version is 3, and `isClient` must be either true or false.  On the
client side stream, it will wait for the server to send their vector
clock, before replying. This means that if the server doesn't actually
support this api, you give them a change to send back an error before<commit_msg>Minor tidy of optional opts<commit_after># Epidemic Broadcast Trees

This module is loosely based on plumtree Epidemic Broadcast Trees
[EBT paper], but adapted to also replicate logs, and optimized
to achive a minimal overhead (the cost of the protocol is linear with
the number of messages to be sent)

It's a algorithm that combines the robustness of a flooding epidemic
gossip broadcast, with the efficiency of a tree model. It's intended
for implementing realtime protocols (such as chat, scuttlebutt, also
radio/video) over networks with random topology - or networks where
otherwise peers may be unable to all connect to each other or to a
central hub.

Although the primary motivation for this module is to use it in secure
scuttlebutt, it's intended to be decoupled sufficiently to use for
other applications.

## Example

implement a simple in memory log replicator.

``` js
var clocks = {}
var logs = {}

function append (msg, cb) {
  var log = logs[msg.author] || {}
  //check that this is the next expected message.
  if(msg.sequence != Object.keys(log).length + 1)
    cb(new Error('out of order, found:'+msg.sequence+', expected:'+log.length))
  else {
    log[msg.sequence] = msg
    ebt.onAppend(msg)
    cb()
  }
}

var ebt = EBT({
  //NOTE: in this example, we are using readable strings for clarity
  //but ideally you'd use cryptographic ids, like public keys.
  id: 'alice',
  getClock: function (id, cb) {
    //load the peer clock for id.
    cb(null, clocks[id] || {})
  },
  setClock: function (id, clock) {
    //set clock doesn't have take a cb, but it's okay to be async.
    clocks[id] = clock
  },
  getAt: function (pair, cb) {
    //load a message particular message, by id:sequence
    if(!logs[pair.id] || !logs[pair.id][pair.sequence])
      cb(new Error('not found'))
    else
      cb(null, logs[pair.id][pair.sequence])
  },
  append: append
})

ebt.append({
  author: 'alice', sequence: 1, content: {}
}, function () {})

//must explicitly say we are replicating which peers.
ebt.request('alice', true)
ebt.request('bob', true)

//create a stream and pipe it to another instance
//isClient and version are required.
var stream = ebt.createStream('bob', version=3, isClient = true)
stream.pipe(remote_stream).pipe(stream)
```

> note about push-stream: push-stream is only new, so you'll probably
  need to convert this to a pull-stream to connect stream to a network
  io stream and serialization

``` js
var pushToPull = require('push-stream-to-pull-stream')
var stream = pushToPull(ebt.createStream(remote_id, 3, isCient = true))
pull(stream, remote_pull_stream, stream)
```

## API

### EBT(opts) => ebt

where opts provides the necessary things to connect ebt
to your system.

```
opts = {
  id: string,
  timeout: 3000, //default,
  getClock: function (id, cb),
  setClock: function (id, clock),
  getAt: function ({id:string, sequence:number}, cb),
  append: function (msg, cb),
  isFeed: function (id),
  isMsg: function(data),
  getMsgAuthor: function(msg),
  getMsgSequence: function(msg)
}
```

Create a new EBT instance. `id` is a unique identifier of the current
peer.  In [secure-scuttlebutt](https://scuttlebutt.nz) this is a
ed25519 public key.

`getClock(id, cb)` and `setClock(id, clock)` save a peer's clock
object.  This is used to save bandwidth when reconnecting to a peer
again.

`getAt({id, sequence}, cb)` retrives a message in a feed and an
sequence.  messages must have `{author, sequence, content}` fields.

`append(msg, cb)` append a particular message to the log.

`timeout` is used to decide when to switch a feed to another peer.
This is essential to detecting when a peer may have stalled.

`isFeed(id)` is a validation function that returns true if `id` is a
valid feed identifier. If not, it is ignored'

### optional for backwards compatibility

`isMsg(data)` is a validation function used to distinguish between data
messages and status messages. A message must contain an `author` field
that corresponds to the feed identifier and a `sequence` field.

`getMsgAuthor(msg)` is a function that given a message returns the
author.

`getMsgSequence(msg)` is a function that given a message returns the
sequence.

### ebt.onAppend (msg)

When a message is appended to the database, tell ebt about it.  this
must be called whenever a message is successfully appended to the
database.

### ebt.createStream(id, version, isClient) => PushStream

Create a stream for replication. returns a [push-stream]. The current
version is 3, and `isClient` must be either true or false.  On the
client side stream, it will wait for the server to send their vector
clock, before replying. This means that if the server doesn't actually
support this api, you give them a change to send back an error before
{'fraction_non_alphanumeric': 0.0961198377197545, 'fraction_numerical': 0.0029127223551440756, 'mean_word_length': 4.132941804591565, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 9, 'https://': 2, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '153101', 'n_tokens_mistral': 2919, 'n_tokens_neox': 2791, 'n_words': 1292}