1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
|
// Negentropy (NIP-77): set reconciliation with a relay. The client holds a
// set of (created_at, id) pairs — the events it already has for a filter —
// and the relay holds its own. Instead of the relay re-sending every event
// the filter matches (a DM catch-up re-served 200 gift wraps, half a
// megabyte, on every subscribe; a notification refresh a hundred kilobytes),
// the two sides exchange fingerprints of ranges of their sets, split the
// ranges that differ, and end with the ids each side lacks — a few hundred
// bytes, usually in one round. The events themselves are then fetched by id.
//
// Protocol version 1 (0x61), as strfry speaks it. A port of Doug Hoyte's
// reference client (github.com/hoytech/negentropy, MIT, © 2023), trimmed to
// the initiator's side plus what a test needs to play the relay.
// Hashing is WebCrypto SHA-256, so everything here is async.
const PROTOCOL_VERSION = 0x61;
const ID_SIZE = 32;
const FINGERPRINT_SIZE = 16;
const MODE = { skip: 0, fingerprint: 1, idList: 2 };
const MAX_TS = Number.MAX_SAFE_INTEGER; // "infinity": the bound past every item
// ---- bytes ----------------------------------------------------------------
const hexOf = (a) => { let s = ''; for (let i = 0; i < a.length; i++) s += (a[i] < 16 ? '0' : '') + a[i].toString(16); return s; };
const bytesOf = (h) => { if (h instanceof Uint8Array) return h; if (h.length % 2) throw new Error('odd hex'); const a = new Uint8Array(h.length / 2); for (let i = 0; i < a.length; i++) a[i] = parseInt(h.substr(i * 2, 2), 16); return a; };
const cmpBytes = (a, b) => { const n = Math.min(a.length, b.length); for (let i = 0; i < n; i++) { if (a[i] !== b[i]) return a[i] < b[i] ? -1 : 1; } return a.length === b.length ? 0 : a.length < b.length ? -1 : 1; };
const cmpItem = (a, b) => (a.ts === b.ts ? cmpBytes(a.id, b.id) : a.ts < b.ts ? -1 : 1);
class Out {
constructor() { this.parts = []; this.length = 0; }
push(a) { if (Array.isArray(a)) a = Uint8Array.from(a); this.parts.push(a); this.length += a.length; }
bytes() { const o = new Uint8Array(this.length); let p = 0; for (const a of this.parts) { o.set(a, p); p += a.length; } return o; }
}
class In {
constructor(a) { this.a = a; this.p = 0; }
get left() { return this.a.length - this.p; }
byte() { if (this.p >= this.a.length) throw new Error('negentropy: message ends early'); return this.a[this.p++]; }
take(n) { if (this.p + n > this.a.length) throw new Error('negentropy: message ends early'); const s = this.a.subarray(this.p, this.p + n); this.p += n; return s; }
}
// varints, big-endian 7-bit groups with a continuation bit (the protocol's)
const putVar = (o, n) => { if (n === 0) return o.push([0]); const g = []; while (n > 0) { g.push(n % 128); n = Math.floor(n / 128); } g.reverse(); for (let i = 0; i < g.length - 1; i++) g[i] |= 128; o.push(g); };
const getVar = (i) => { let n = 0; for (;;) { const b = i.byte(); n = n * 128 + (b & 127); if (!(b & 128)) return n; } };
// ---- the fingerprint of a range: the ids summed mod 2^256, hashed with the count
const sha256 = async (bytes) => new Uint8Array(await crypto.subtle.digest('SHA-256', bytes));
function addInto(acc, id) {
let carry = 0;
for (let i = 0; i < ID_SIZE; i++) { const s = acc[i] + id[i] + carry; acc[i] = s & 255; carry = s >> 8; }
}
async function fingerprint(items, begin, end) {
const acc = new Uint8Array(ID_SIZE);
for (let i = begin; i < end; i++) addInto(acc, items[i].id);
const o = new Out(); o.push(acc); putVar(o, end - begin);
return (await sha256(o.bytes())).subarray(0, FINGERPRINT_SIZE);
}
// ---- the set ----------------------------------------------------------------
// `items`: [[created_at, id hex], ...] in any order; duplicates are dropped.
export class NegSet {
constructor(items = []) {
const seen = new Set();
this.items = [];
for (const [ts, id] of items) {
if (typeof id !== 'string' || id.length !== 64 || seen.has(id)) continue;
seen.add(id);
this.items.push({ ts: Math.max(0, Math.floor(ts || 0)), id: bytesOf(id) });
}
this.items.sort(cmpItem);
}
get size() { return this.items.length; }
// the first index in [begin, end) at or past the bound
lowerBound(begin, end, bound) {
let lo = begin, hi = end;
while (lo < hi) { const m = (lo + hi) >> 1; if (cmpItem(this.items[m], bound) < 0) lo = m + 1; else hi = m; }
return lo;
}
}
// ---- one side of a reconciliation ------------------------------------------
export class Negentropy {
constructor(set, { frameSizeLimit = 0 } = {}) {
if (frameSizeLimit && frameSizeLimit < 4096) throw new Error('negentropy: frame size limit too small');
this.set = set;
this.frameSizeLimit = frameSizeLimit;
this.initiator = false;
this.lastTsIn = 0; this.lastTsOut = 0;
}
// the client's opening message (hex)
async initiate() {
this.initiator = true;
const o = new Out(); o.push([PROTOCOL_VERSION]);
await this.splitRange(0, this.set.size, { ts: MAX_TS, id: new Uint8Array(0) }, o);
return hexOf(o.bytes());
}
// One message in, one out. For the initiator the answer is null once the
// sets are reconciled; `have` are ids we hold that the other side lacks,
// `need` the ones it holds that we lack. The non-initiator (a relay, or
// a test playing one) always answers and collects nothing.
async reconcile(hex) {
const q = new In(bytesOf(hex));
const have = [], need = [];
this.lastTsIn = this.lastTsOut = 0;
const full = new Out(); full.push([PROTOCOL_VERSION]);
const version = q.byte();
if (version < 0x60 || version > 0x6f) throw new Error('negentropy: not a negentropy message');
if (version !== PROTOCOL_VERSION) {
if (this.initiator) throw new Error('negentropy: unsupported protocol version ' + (version - 0x60));
return { next: hexOf(full.bytes()), have, need };
}
const items = this.set.items, size = this.set.size;
let prevBound = { ts: 0, id: new Uint8Array(0) }, prevIndex = 0, skip = false;
while (q.left) {
const o = new Out();
const doSkip = () => { if (skip) { skip = false; this.putBound(o, prevBound); putVar(o, MODE.skip); } };
const bound = this.getBound(q);
const mode = getVar(q);
const lower = prevIndex;
let upper = this.set.lowerBound(prevIndex, size, bound);
if (mode === MODE.skip) skip = true;
else if (mode === MODE.fingerprint) {
const theirs = q.take(FINGERPRINT_SIZE);
const ours = await fingerprint(items, lower, upper);
if (cmpBytes(theirs, ours) !== 0) { doSkip(); await this.splitRange(lower, upper, bound, o); }
else skip = true;
} else if (mode === MODE.idList) {
const n = getVar(q);
const theirs = new Map();
for (let i = 0; i < n; i++) { const id = q.take(ID_SIZE); if (this.initiator) theirs.set(hexOf(id), id); }
if (this.initiator) {
skip = true;
for (let i = lower; i < upper; i++) {
const k = hexOf(items[i].id);
if (theirs.has(k)) theirs.delete(k); else have.push(k);
}
for (const k of theirs.keys()) need.push(k);
} else {
doSkip();
const ids = new Out(); let count = 0, endBound = bound;
for (let i = lower; i < upper; i++) {
if (this.overFrame(full.length + ids.length)) { endBound = items[i]; upper = i; break; }
ids.push(items[i].id); count++;
}
this.putBound(o, endBound); putVar(o, MODE.idList); putVar(o, count); o.push(ids.bytes());
full.push(o.bytes()); o.parts = []; o.length = 0;
}
} else throw new Error('negentropy: unexpected mode ' + mode);
if (this.overFrame(full.length + o.length)) {
// the frame is full: one fingerprint for everything left
const rest = await fingerprint(items, upper, size);
this.putBound(full, { ts: MAX_TS, id: new Uint8Array(0) }); putVar(full, MODE.fingerprint); full.push(rest);
break;
}
full.push(o.bytes());
prevIndex = upper; prevBound = bound;
}
return { next: full.length === 1 && this.initiator ? null : hexOf(full.bytes()), have, need };
}
// a range that differs: sixteen fingerprinted buckets, or the ids outright when small
async splitRange(lower, upper, upperBound, o) {
const items = this.set.items, n = upper - lower, buckets = 16;
if (n < buckets * 2) {
this.putBound(o, upperBound); putVar(o, MODE.idList); putVar(o, n);
for (let i = lower; i < upper; i++) o.push(items[i].id);
return;
}
const per = Math.floor(n / buckets), extra = n % buckets;
let cur = lower;
for (let b = 0; b < buckets; b++) {
const size = per + (b < extra ? 1 : 0);
const fp = await fingerprint(items, cur, cur + size);
cur += size;
const next = cur === upper ? upperBound : minimalBound(items[cur - 1], items[cur]);
this.putBound(o, next); putVar(o, MODE.fingerprint); o.push(fp);
}
}
overFrame(n) { return !!this.frameSizeLimit && n > this.frameSizeLimit - 200; }
// bounds: a timestamp as a delta from the last one (+1; 0 = infinity), then a prefix of an id
getBound(q) {
let ts = getVar(q);
ts = ts === 0 ? MAX_TS : ts - 1;
if (this.lastTsIn === MAX_TS || ts === MAX_TS) { this.lastTsIn = MAX_TS; ts = MAX_TS; }
else { ts += this.lastTsIn; this.lastTsIn = ts; }
const len = getVar(q);
if (len > ID_SIZE) throw new Error('negentropy: bound key too long');
return { ts, id: q.take(len) };
}
putBound(o, b) {
if (b.ts === MAX_TS) { this.lastTsOut = MAX_TS; putVar(o, 0); }
else { putVar(o, b.ts - this.lastTsOut + 1); this.lastTsOut = b.ts; }
putVar(o, b.id.length); o.push(b.id);
}
}
// the shortest bound that separates two neighbouring items
function minimalBound(prev, cur) {
if (cur.ts !== prev.ts) return { ts: cur.ts, id: new Uint8Array(0) };
let shared = 0;
while (shared < ID_SIZE && cur.id[shared] === prev.id[shared]) shared++;
return { ts: cur.ts, id: cur.id.subarray(0, shared + 1) };
}
|
