waku-org
/
negentropy
mirror of https://github.com/waku-org/negentropy.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

bring JS version up to new spec

This commit is contained in:
Doug Hoyte 2023-09-02 04:06:05 -04:00
parent 275c6f3215
commit 06300abc78
1 changed files with 202 additions and 212 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

414
js/Negentropy.js
View File

@ -1,116 +1,68 @@
// (C) 2023 Doug Hoyte. MIT license
class WrappedBuffer {
constructor(buffer) {
this._raw = new Uint8Array(buffer) || new Uint8Array(1024);
this._raw = new Uint8Array(buffer || 256);
this.length = buffer ? buffer.length : 0;
this._expansionRate = 2;
}
unwrap() {
return this._raw.subarray(0, this.length)
return this._raw.subarray(0, this.length);
}
get capacity() {
return this._raw.byteLength
return this._raw.byteLength;
}
extend(buf) {
if (buf._raw) buf = buf.unwrap();
const targetSize = buf.length + this.length;
if (this.capacity < targetSize) {
const oldRaw = this._raw;
const newCapacity = Math.max(
Math.floor(this.capacity * this._expansionRate),
targetSize
);
const newCapacity = Math.max(this.capacity * 2, targetSize);
this._raw = new Uint8Array(newCapacity);
this._raw.set(oldRaw);
}
this._raw.set(buf, this.length);
this.length += buf.length;
return this;
}
shift() {
const first = this._raw[0]
this._raw = this._raw.subarray(1)
this.length -= 1
return first
const first = this._raw[0];
this._raw = this._raw.subarray(1);
this.length--;
return first;
}
splice(n = 1) {
const firstSubarray = this._raw.subarray(0, n)
this._raw = this._raw.subarray(n)
this.length -= n
return firstSubarray
shiftN(n = 1) {
const firstSubarray = this._raw.subarray(0, n);
this._raw = this._raw.subarray(n);
this.length -= n;
return firstSubarray;
}
}
class TrieNode {
constructor() {
this.children = new Array(256).fill(null);
this.value = null;
this.isTail = false;
}
}
class HashMap {
constructor() {
this._root = new TrieNode();
}
set(id, value) {
let node = this._root;
for (let byte of id) {
if (!node.children[byte]) {
node.children[byte] = new TrieNode();
}
node = node.children[byte];
};
node.value = value;
node.isTail = true;
}
get(keyUint8Array) {
let node = this._root;
for (let element of keyUint8Array) {
if (!node.children[element]) {
return null; // return null if the key is not in the trie
}
node = node.children[element];
}
return node.value;
}
*[Symbol.iterator]() {
const key = [];
function* walk(node) {
for (let i = 0; i < 256; i++) {
if (!node.children[i]) continue;
key.push(i);
if (node.children[i].isTail) {
yield [new Uint8Array(key), node.children[i].value];
}
yield* walk(node.children[i]);
key.pop();
}
}
yield* walk(this._root);
}
}
class Negentropy {
constructor(idSize) {
constructor(idSize = 16, frameSizeLimit = 0) {
if (idSize < 8 || idSize > 32) throw Error("idSize invalid");
if (frameSizeLimit !== 0 && frameSizeLimit < 4096) throw Error("frameSizeLimit too small");
this.idSize = idSize;
this.frameSizeLimit = frameSizeLimit;
this.items = [];
this.pendingOutputs = [];
}
addItem(timestamp, id) {
if (this.sealed) throw Error("already sealed");
if (id.byteLength > 64 || id.byteLength % 2 !== 0) throw Error("bad length for id");
id = id.subarray(0, this.idSize * 2);
id = this._loadInput(id);
if (id.byteLength > 64 || id.byteLength < this.idSize) throw Error("bad length for id");
if (id.byteLength > this.idSize) id = id.subarray(0, this.idSize);
this.items.push({ timestamp, id });
}
@ -137,44 +89,42 @@ class Negentropy {
return { timestamp: Number.MAX_VALUE, id: new Uint8Array(0) };
}
_loadInput(inp) {
if (typeof(inp) === 'string') inp = hexToUint8Array(inp);
else if (__proto__ !== Uint8Array.prototype) inp = new Uint8Array(inp); // node Buffer?
return inp;
}
initiate() {
if (!this.sealed) throw Error("not sealed");
this.isInitiator = true;
let output = new WrappedBuffer();
let state = this._newState();
this.splitRange(0, this.items.length, this._zeroBound(), this._maxBound(), this.pendingOutputs);
this.splitRange(0, this.items.length, this._zeroBound(), this._maxBound(), state, output);
return output.unwrap();
return this.buildOutput();
}
reconcile(query) {
if (!this.sealed) throw Error("not sealed");
query = new WrappedBuffer(query);
query = new WrappedBuffer(this._loadInput(query));
let haveIds = [], needIds = [];
let output = new WrappedBuffer();
if (!this.sealed) throw Error("not sealed");
this.continuationNeeded = false;
let prevBound = this._zeroBound();
let prevIndex = 0;
let state = this._newState();
let skip = false;
let doSkip = () => {
if (!skip) return;
skip = false;
this.encodeBound(prevBound, state, output);
output.extend(this.encodeVarInt(0)); // mode = Skip
};
let outputs = [];
while (query.length !== 0) {
let currBound = this.decodeBound(query, state);
let mode = this.decodeVarInt(query); // 0 = Skip, 1 = Fingerprint, 2 = IdList, 3 = IdListResponse
let mode = this.decodeVarInt(query); // 0 = Skip, 1 = Fingerprint, 2 = IdList, 3 = deprecated, 4 = Continuation
let lower = prevIndex;
let upper = findUpperBound(this.items, lower, this.items.length, currBound, itemCompare);
if (mode === 0) { // Skip
skip = true;
// Do nothing
} else if (mode === 1) { // Fingerprint
let theirXorSet = this.getBytes(query, this.idSize);
@ -193,74 +143,71 @@ class Negentropy {
}
if (!matches) {
doSkip();
this.splitRange(lower, upper, prevBound, currBound, state, output);
} else {
skip = true;
this.splitRange(lower, upper, prevBound, currBound, outputs);
}
} else if (mode === 2) { // IdList
let numElems = this.decodeVarInt(query);
let numIds = this.decodeVarInt(query);
let theirElems = new HashMap();
for (let i = 0; i < numElems; i++) {
let id = this.getBytes(query, this.idSize);
theirElems.set(id, { offset: i, onBothSides: false });
let theirElems = {}; // stringified Uint8Array -> original Uint8Array
for (let i = 0; i < numIds; i++) {
let e = this.getBytes(query, this.idSize);
theirElems[e] = e;
}
let responseHaveIds = [];
let responseNeedIndices = [];
for (let i = lower; i < upper; i++) {
let id = this.items[i].id;
let e = theirElems.get(id);
let k = this.items[i].id;
if (!e) {
if (!theirElems[k]) {
// ID exists on our side, but not their side
if (this.isInitiator) haveIds.push(id);
else responseHaveIds.push(id);
if (this.isInitiator) haveIds.push(this.wantUint8ArrayOutput ? k : uint8ArrayToHex(k));
} else {
// ID exists on both sides
e.onBothSides = true;
delete theirElems[k];
}
}
for (let [k, v] of theirElems) {
if (!v.onBothSides) {
// ID exists on their side, but not our side
if (this.isInitiator) needIds.push(k);
else responseNeedIndices.push(v.offset);
if (this.isInitiator) {
for (let v of Object.values(theirElems)) {
needIds.push(this.wantUint8ArrayOutput ? v : uint8ArrayToHex(v));
}
}
if (!this.isInitiator) {
doSkip();
this.encodeBound(currBound, state, output);
output.extend(this.encodeVarInt(3)); // mode = IdListResponse
output.extend(this.encodeVarInt(responseHaveIds.length));
for (let id of responseHaveIds) output.extend(id);
let bitField = this.encodeBitField(responseNeedIndices);
output.extend(this.encodeVarInt(bitField.length));
output.extend(bitField);
} else {
skip = true;
}
} else if (mode === 3) { // IdListResponse
if (!this.isInitiator) throw Error("unexpected IdListResponse");
skip = true;
let responseHaveIds = [];
let numIds = this.decodeVarInt(query);
for (let i = 0; i < numIds; i++) {
needIds.push(this.getBytes(query, this.idSize));
}
let it = lower;
let didSplit = false;
let splitBound = this._zeroBound();
let bitFieldSize = this.decodeVarInt(query);
let bitField = this.getBytes(query, bitFieldSize);
let flushIdListOutput = () => {
let payload = this.encodeVarInt(2); // mode = IdList
for (let i = lower; i < upper; i++) {
if (this.bitFieldLookup(bitField, i - lower)) haveIds.push(this.items[i].id);
payload.extend(this.encodeVarInt(responseHaveIds.length));
for (let id of responseHaveIds) payload.extend(id);
let nextSplitBound = (it+1) >= upper ? currBound : this.getMinimalBound(this.items[it], this.items[it+1]);
outputs.push({
start: didSplit ? splitBound : prevBound,
end: nextSplitBound,
payload: payload,
});
splitBound = nextSplitBound;
didSplit = true;
responseHaveIds = [];
};
for (; it < upper; ++it) {
responseHaveIds.push(this.items[it].id);
if (responseHaveIds.length >= 100) flushIdListOutput(); // 100*32 is less than minimum frame size limit of 4k
}
flushIdListOutput();
}
} else if (mode === 3) { // Deprecated
throw Error("other side is speaking old negentropy protocol");
} else if (mode === 4) { // Continuation
this.continuationNeeded = true;
} else {
throw Error("unexpected mode");
}
@ -269,23 +216,28 @@ class Negentropy {
prevBound = currBound;
}
return [output.unwrap(), haveIds, needIds];
while (outputs.length) {
this.pendingOutputs.unshift(outputs.pop());
}
return [this.buildOutput(), haveIds, needIds];
}
splitRange(lower, upper, _lowerBound, upperBound, state, output) {
splitRange(lower, upper, lowerBound, upperBound, outputs) {
let numElems = upper - lower;
let buckets = 16;
if (numElems < buckets * 2) {
this.encodeBound(upperBound, state, output);
output.extend(this.encodeVarInt(2)); // mode = IdList
let payload = this.encodeVarInt(2); // mode = IdList
payload.extend(this.encodeVarInt(numElems));
for (let it = lower; it < upper; ++it) payload.extend(this.items[it].id);
output.extend(this.encodeVarInt(numElems));
for (let it = lower; it < upper; ++it) output.extend(this.items[it].id);
outputs.push({ start: lowerBound, end: upperBound, payload: payload, });
} else {
let itemsPerBucket = Math.floor(numElems / buckets);
let bucketsWithExtra = numElems % buckets;
let curr = lower;
let prevBound = this.items[curr];
for (let i = 0; i < buckets; i++) {
let ourXorSet = new Uint8Array(this.idSize)
@ -293,32 +245,79 @@ class Negentropy {
for (let j = 0; j < this.idSize; j++) ourXorSet[j] ^= this.items[curr].id[j];
}
if (i === buckets - 1) this.encodeBound(upperBound, state, output);
else this.encodeMinimalBound(this.items[curr], this.items[curr - 1], state, output);
let payload = this.encodeVarInt(1); // mode = Fingerprint
payload.extend(ourXorSet);
output.extend(this.encodeVarInt(1)); // mode = Fingerprint
output.extend(ourXorSet);
outputs.push({
start: i === 0 ? lowerBound : prevBound,
end: curr === this.items.length ? upperBound : this.getMinimalBound(this.items[curr - 1], this.items[curr]),
payload: payload,
});
prevBound = outputs[outputs.length - 1].end;
}
outputs[outputs.length - 1].end = upperBound;
}
}
buildOutput() {
let output = new WrappedBuffer();
let currBound = this._zeroBound();
let state = this._newState();
this.pendingOutputs.sort((a,b) => itemCompare(a.start, b.start));
while (this.pendingOutputs.length) {
let o = new WrappedBuffer();
let p = this.pendingOutputs[0];
let cmp = itemCompare(p.start, currBound);
// When bounds are out of order or overlapping, finish and resume next time (shouldn't happen because of sort above)
if (cmp < 0) break;
if (cmp !== 0) {
o.extend(this.encodeBound(p.start, state));
o.extend(this.encodeVarInt(0)); // mode = Skip
}
o.extend(this.encodeBound(p.end, state));
o.extend(p.payload);
if (this.frameSizeLimit && output.length + o.length > this.frameSizeLimit - 5) break; // 5 leaves room for Continuation
output.extend(o);
currBound = p.end;
this.pendingOutputs.shift();
}
// Server indicates that it has more to send, OR ensure client sends a non-empty message
if ((!this.isInitiator && this.pendingOutputs.length) || (this.isInitiator && output.length == 0 && this.continuationNeeded)) {
output.extend(this.encodeBound(this._maxBound(), state));
output.extend(this.encodeVarInt(4)); // mode = Continue
}
let ret = output.unwrap();
if (!this.wantUint8ArrayOutput) ret = uint8ArrayToHex(ret);
return ret;
}
// Decoding
getByte(buf) {
if (buf.length === 0) throw Error("parse ends prematurely");
return buf.shift();
}
getBytes(buf, n) {
if (buf.length < n) throw Error("parse ends prematurely");
return buf.splice(n);
return buf.shiftN(n);
}
decodeVarInt(buf) {
let res = 0;
while (1) {
let byte = this.getByte(buf);
if (buf.length === 0) throw Error("parse ends prematurely");
let byte = buf.shift();
res = (res << 7) | (byte & 127);
if ((byte & 128) === 0) break;
}
@ -349,7 +348,7 @@ class Negentropy {
// Encoding
encodeVarInt(n) {
if (n === 0) return new Uint8Array([0]);
if (n === 0) return new WrappedBuffer([0]);
let o = [];
@ -362,7 +361,7 @@ class Negentropy {
for (let i = 0; i < o.length - 1; i++) o[i] |= 128;
return new Uint8Array(o);
return new WrappedBuffer(o);
}
encodeTimestampOut(timestamp, state) {
@ -377,17 +376,19 @@ class Negentropy {
return this.encodeVarInt(timestamp + 1);
}
encodeBound(key, state, output) {
encodeBound(key, state) {
let output = new WrappedBuffer();
output.extend(this.encodeTimestampOut(key.timestamp, state));
output.extend(this.encodeVarInt(key.id.length));
output.extend(key.id);
return output;
}
encodeMinimalBound(curr, prev, state, output) {
output.extend(this.encodeTimestampOut(curr.timestamp, state));
getMinimalBound(prev, curr) {
if (curr.timestamp !== prev.timestamp) {
output.extend(this.encodeVarInt(0));
return { timestamp: curr.timestamp, id: new Uint8Array(0), };
} else {
let sharedPrefixBytes = 0;
@ -396,44 +397,58 @@ class Negentropy {
sharedPrefixBytes++;
}
output.extend(this.encodeVarInt(sharedPrefixBytes + 1));
output.extend(curr.id.subarray(0, sharedPrefixBytes + 1));
return { timestamp: curr.timestamp, id: curr.id.subarray(0, sharedPrefixBytes + 1), };
}
};
encodeBitField(inds) {
if (inds.length === 0) return [];
let max = Math.max(...inds);
let bitField = new Uint8Array(Math.floor((max + 8) / 8));
for (let ind of inds) bitField[Math.floor(ind / 8)] |= 1 << (ind % 8);
return bitField;
}
bitFieldLookup(bitField, ind) {
if (Math.floor((ind + 8) / 8) > bitField.length) return false;
return !!(bitField[Math.floor(ind / 8)] & 1 << (ind % 8));
}
}
function hexToUint8Array(h) {
if (h.startsWith('0x')) h = h.substr(2);
if (h.length % 2 === 1) throw Error("odd length of hex string");
let arr = new Uint8Array(h.length / 2);
for (let i = 0; i < arr.length; i++) arr[i] = parseInt(h.substr(i * 2, 2), 16);
return arr;
}
const uint8ArrayToHexLookupTable = new Array(256);
{
const hexAlphabet = ['0', '1', '2', '3', '4', '5', '6', '7', '8', '9', 'a', 'b', 'c', 'd', 'e', 'f'];
for (let i = 0; i < 256; i++) {
uint8ArrayToHexLookupTable[i] = hexAlphabet[(i >>> 4) & 0xF] + hexAlphabet[i & 0xF];
}
}
function uint8ArrayToHex(arr) {
let out = '';
for (let i = 0, edx = arr.length; i < edx; i++) {
out += uint8ArrayToHexLookupTable[arr[i]];
}
return out;
}
/**
* @param {Item} a
* @param {Item} b
*
* @returns {number}
*/
function compareUint8Array(a, b) {
for (let i = 0; i < a.byteLength; i++) {
if (a[i] < b[i]) return -1;
if (a[i] > b[i]) return 1;
}
if (a.byteLength > b.byteLength) return 1;
if (a.byteLength < b.byteLength) return -1;
return 0;
}
function itemCompare(a, b) {
if (a.timestamp === b.timestamp) {
return compareUint8Array(a.id, b.id)
return compareUint8Array(a.id, b.id);
}
return a.timestamp - b.timestamp;
}
function binarySearch(arr, first, last, cmp) {
let count = last - first;
@ -461,30 +476,5 @@ function findUpperBound(arr, first, last, value, cmp) {
return binarySearch(arr, first, last, (a) => cmp(value, a) >= 0);
}
/**
* @param {Uint8Array} a
* @param {Uint8Array} b
*/
function compareUint8Array(a, b) {
for (let i = 0; i < a.byteLength; i++) {
if (a[i] < b[i]) {
return -1
}
if (a[i] > b[i]) {
return 1
}
}
if (a.byteLength > b.byteLength) {
return 1
}
if (a.byteLength < b.byteLength) {
return -1
}
return 0
}
module.exports = Negentropy;