diff options
| -rw-r--r-- | src/features/messages.js | 125 | ||||
| -rw-r--r-- | src/negentropy.js | 197 | ||||
| -rw-r--r-- | src/nostr.js | 94 | ||||
| -rw-r--r-- | tools/negentropy-sync-test.js | 159 | ||||
| -rw-r--r-- | tools/negentropy-test.js | 64 |
5 files changed, 622 insertions, 17 deletions
diff --git a/src/features/messages.js b/src/features/messages.js index a40f7d5..ad5c32c 100644 --- a/src/features/messages.js +++ b/src/features/messages.js @@ -17,6 +17,7 @@ import { npubOf, neventOf, parseNostrPubkey, parseNostrRef, generateSecretKey, getPublicKey, finalizeEvent, nip44, PROFILE_RELAYS, openWrapsOffthread, unwrapDMsOffthread, verifyEventsAsync, setSyncConfig, DEFAULT_SYNC_RELAYS, } from '../nostr.js'; +import { reconcileOn, syncOn, negentropyOff } from '../nostr.js'; // NIP-77: fetch only what we lack import { channelKey, channelStream, channelEpoch, channelIsPrivate, controlKey, guestbookKey, openWrap, wrapRumor, rumorWithId, foldControl, foldGuestbook, observeAuthor, eventMs, msTags, makeEdition, @@ -688,23 +689,57 @@ export function messagesFeature(ctx) { healSince = 0; } const healing = (wrap) => !!healSince && wrap.created_at >= healSince && !healed.has(wrap.id) && !!healed.add(wrap.id); - function rememberWrap(id) { - const k = wrapKey(id); - if (!wrapLog) wrapLog = []; - if (!seenWraps.has(k)) wrapLog.push(k); - seenWraps.add(k); + // The wraps this device holds — full id and time — for reconciliation + // (NIP-77) with a relay that speaks it: a subscribe then fetches only the + // wraps we lack instead of the relay's newest two hundred (half a + // megabyte, every boot and every resubscribe), and the live subscription + // opens with limit 0, nothing re-served. The prefix log above stays the + // "skip this one" memory; a wrap it knows but this list doesn't (logged + // before the list existed) is added as the relay re-serves it, once. + const WRAP_ITEMS_MAX = 600; // the relay reconciles its newest 400: that many held, with room (46 KB of a phone's near-full localStorage) + let wrapItems = null; // id -> created_at, oldest first + function loadWrapItems() { + if (wrapItems) return; + wrapItems = new Map(); + try { for (const [id, ts] of st().wrapItems || []) if (/^[0-9a-f]{64}$/.test(id)) wrapItems.set(id, ts | 0); } catch {} + } + function noteWrapItem(id, ts) { + if (!id || !ts) return; + loadWrapItems(); + if (wrapItems.has(id)) return; + wrapItems.set(id, ts); + while (wrapItems.size > WRAP_ITEMS_MAX) wrapItems.delete(wrapItems.keys().next().value); + scheduleWrapSave(); + } + const wrapItemsHeld = () => { loadWrapItems(); return [...wrapItems].map(([id, ts]) => [ts, id]); }; + function scheduleWrapSave() { clearTimeout(wrapsTimer); wrapsTimer = setTimeout(() => { try { // another tab may have logged wraps since this one loaded: keep theirs const s = st(); - const mine = new Set(wrapLog); - s.wraps = [...(s.wraps || []).filter((k) => !mine.has(k)), ...wrapLog].slice(-WRAPS_MAX); - wrapLog = s.wraps; + if (wrapLog) { + const mine = new Set(wrapLog); + s.wraps = [...(s.wraps || []).filter((k) => !mine.has(k)), ...wrapLog].slice(-WRAPS_MAX); + wrapLog = s.wraps; + } + if (wrapItems) { + const all = new Map(s.wrapItems || []); + for (const [id, ts] of wrapItems) all.set(id, ts); + s.wrapItems = [...all].slice(-WRAP_ITEMS_MAX); + } save(s); } catch {} }, 1500); } + function rememberWrap(id, ts) { + const k = wrapKey(id); + if (!wrapLog) wrapLog = []; + if (!seenWraps.has(k)) wrapLog.push(k); + seenWraps.add(k); + if (ts) noteWrapItem(id, ts); + scheduleWrapSave(); + } let dmStarted = false; let allUnsubs = []; let dmUnsubs = []; // the DM wrap subscriptions alone, so they can be rebuilt @@ -3055,13 +3090,14 @@ export function messagesFeature(ctx) { async function handleInboxWrap(wrap) { let heal = false; if (seenWraps.has(wrap.id) || seenWraps.has(wrapKey(wrap.id))) { + noteWrapItem(wrap.id, wrap.created_at); // known by its prefix alone: now held in full if (!healing(wrap)) return; heal = true; } seenWraps.add(wrap.id); const r = await openInboxWrap(wrap, heal); if (r === 'stale') return; - if (r === true) { rememberWrap(wrap.id); return; } + if (r === true) { rememberWrap(wrap.id, wrap.created_at); return; } if (pendingWraps.size >= PENDING_MAX) pendingWraps.delete(pendingWraps.keys().next().value); pendingWraps.set(wrap.id, { wrap, heal, tries: r !== 'later' && decryptorsComplete() ? 1 : 0 }); scheduleDrain(8000); @@ -3101,9 +3137,9 @@ export function messagesFeature(ctx) { if (complete && !decryptorsComplete()) break; const r = await openInboxWrap(p.wrap, p.heal); if (r === 'stale') break; // the account changed under this pass - if (r === true) { pendingWraps.delete(id); rememberWrap(id); } + if (r === true) { pendingWraps.delete(id); rememberWrap(id, p.wrap && p.wrap.created_at); } else if (r === 'later') break; // the signer went quiet — nothing learned, try the rest later - else if (complete && ++p.tries >= PENDING_TRIES) { pendingWraps.delete(id); rememberWrap(id); } + else if (complete && ++p.tries >= PENDING_TRIES) { pendingWraps.delete(id); rememberWrap(id, p.wrap && p.wrap.created_at); } if (performance.now() - chunkStart > 8) { await new Promise((r) => setTimeout(r)); chunkStart = performance.now(); @@ -3160,12 +3196,42 @@ export function messagesFeature(ctx) { } bumpMsgRev(); if (swept) save(s); - const dmSub = (relays) => { - const u = subscribeOn(relays, { kinds: [1059], '#p': pks, limit: 400 }, (wrap) => { - handleInboxWrap(wrap).catch(() => {}); - }, { onclose: noteSubClosed }); + const onWrap = (wrap) => { handleInboxWrap(wrap).catch(() => {}); }; + const plainDmSub = (relays) => { + const u = subscribeOn(relays, { kinds: [1059], '#p': pks, limit: 400 }, onWrap, { onclose: noteSubClosed }); allUnsubs.push(u); dmUnsubs.push(u); }; + // Relay by relay. One that speaks negentropy gets the live subscription + // with limit 0 — opened first, so nothing slips between — and a + // reconciliation of the wraps we hold against its newest four hundred, + // the ones we lack fetched by id. One that doesn't (nos.lol has it + // disabled; remembered for a day) is asked the old way, its newest + // wraps re-served and skipped by the prefix log. So is every relay + // while the one-time heal is running: that pass wants them re-served. + const dmSub = (relays) => { + for (const relay of relays) { + if (healSince || negentropyOff(relay)) { plainDmSub([relay]); continue; } + const u = subscribeOn([relay], { kinds: [1059], '#p': pks, limit: 0 }, onWrap, { onclose: noteSubClosed }); + allUnsubs.push(u); dmUnsubs.push(u); + reconcileWraps(relay).then((ok) => { + if (ok || !dmStarted || !dmUnsubs.includes(u)) return; + // it doesn't speak it after all: the plain subscription, with its backlog + try { u(); } catch {} + allUnsubs = allUnsubs.filter((x) => x !== u); dmUnsubs = dmUnsubs.filter((x) => x !== u); + plainDmSub([relay]); + }).catch(() => {}); + } + }; + async function reconcileWraps(relay) { + const r = await reconcileOn(relay, { kinds: [1059], '#p': pks, limit: 400 }, wrapItemsHeld()); + if (!r) return false; + dlog('dm: ' + relay + ' reconciled: ' + r.need.length + ' wraps to fetch, ' + r.have.length + ' held it lacks'); + for (let i = 0; i < Math.min(r.need.length, 600) && dmStarted; i += 100) { + const evs = await queryOn([relay], { ids: r.need.slice(i, i + 100) }, 6000); + for (const w of evs) onWrap(w); + } + return true; + } dmSub(DM_RELAYS); // Senders deliver to the relays our kind-10050 advertises — a list other // clients may have published with relays beyond our defaults. Read those @@ -15092,6 +15158,28 @@ export function messagesFeature(ctx) { let notif = null; // { status, items } const NOTIF_V = 2; let notifAt = 0, notifUnsub = null; + // Every notification event seen, id and time, for reconciliation (NIP-77) + // with the relays that speak it: a refresh then fetches only what's new, + // not the newest hundred and twenty again — a hundred kilobytes a visit. + // All of them, not the rows kept: a zap receipt folded into its twin, or + // a hidden author's reply, would otherwise come back on every refresh. + const NOTIF_SEEN_MAX = 300; // the relay reconciles its newest 120 + let notifHeld = null; // id -> created_at, oldest first + function notifHeldItems() { + if (!notifHeld) { + notifHeld = new Map(); + try { const s = st(); if (s.notifHeldPk === identityPk()) for (const [id, ts] of s.notifHeld || []) notifHeld.set(id, ts | 0); } catch {} + for (const x of notifNow().items) if (x && x.id && x.ts && !notifHeld.has(x.id)) notifHeld.set(x.id, x.ts); + } + return [...notifHeld].map(([id, ts]) => [ts, id]); + } + function noteNotifHeld(evs) { + notifHeldItems(); + let grew = false; + for (const ev of evs || []) if (ev && ev.id && !notifHeld.has(ev.id)) { notifHeld.set(ev.id, ev.created_at | 0); grew = true; } + while (notifHeld.size > NOTIF_SEEN_MAX) notifHeld.delete(notifHeld.keys().next().value); + return grew; + } const notifNoteAsking = new Set(); const notifSeen = () => st().notifSeen || 0; // Notifications are about the IDENTITY, not the wallet: the stored list @@ -15151,6 +15239,7 @@ export function messagesFeature(ctx) { for (const id of wanted) if (notifNotes.has(id)) notes[id] = savedNotifNote(notifNotes.get(id)); const s2 = st(); s2.notifs = notif.items; s2.notifsPk = identityPk(); s2.notifsV = NOTIF_V; + if (notifHeld) { s2.notifHeld = [...notifHeld]; s2.notifHeldPk = identityPk(); } s2.notifNoteCache = notes; s2.notifNotesPk = identityPk(); save(s2); } @@ -15263,9 +15352,11 @@ export function messagesFeature(ctx) { notifAt = Date.now(); const g = session; try { - const evs = await queryOn(zapRelays(), { kinds: NOTIF_KINDS, '#p': my, limit: 120 }, 5000); + const evs = await syncOn(zapRelays(), { kinds: NOTIF_KINDS, '#p': my, limit: 120 }, notifHeldItems(), { maxWait: 5000 }); if (g !== session) return; // asked for the previous account + const grew = noteNotifHeld(evs); if (mergeNotifs(evs)) scheduleRepaint(); + if (grew) persistNotifs(); } catch {} finally { if (g === session) { if (notif) notif.status = 'ready'; @@ -15817,7 +15908,7 @@ export function messagesFeature(ctx) { noteCountsReady.clear(); // what happened to THEIR posts stays with them — the next identity // starts its list empty and asks the relays under its own key - stopNotifWatch(); notif = null; notifAt = 0; notifNotes.clear(); + stopNotifWatch(); notif = null; notifHeld = null; notifAt = 0; notifNotes.clear(); session++; zapWho.clear(); whoOpenIds.clear(); clearTimeout(zapSaveT); zapSaveT = null; zapSeed = null; diff --git a/src/negentropy.js b/src/negentropy.js new file mode 100644 index 0000000..83f65ae --- /dev/null +++ b/src/negentropy.js @@ -0,0 +1,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) }; +} diff --git a/src/nostr.js b/src/nostr.js index b76e865..8dcab01 100644 --- a/src/nostr.js +++ b/src/nostr.js @@ -18,6 +18,7 @@ import { normalizeURL } from 'nostr-tools/utils'; import { wrapEvent as nip17WrapEvent } from 'nostr-tools/nip17'; import { SimplePool } from 'nostr-tools/pool'; import { dlog } from './debug.js'; +import { NegSet, Negentropy } from './negentropy.js'; import { randomBytes, bytesToHex } from '@noble/hashes/utils'; import { base64urlnopad } from '@scure/base'; @@ -365,6 +366,99 @@ export async function queryOn(relays, filter, maxWait = 1500) { } // Several filters in ONE REQ per relay (NIP-01 allows a list) — one // subscription against the relay's REQ budget instead of one per filter. +// ---- negentropy (NIP-77): fetch only what we lack --------------------------- +// A relay that speaks it (strfry with it enabled: relay.coinos.io does; +// nos.lol, damus and primal answer "negentropy disabled") compares the ids +// it holds for a filter with the ids we hold and tells us the difference — +// a few hundred bytes — instead of re-sending the newest N events every +// time we ask. The events we lack are then fetched by id. A relay that +// doesn't speak it is remembered for a day, so the plain query goes +// straight out there next time rather than after a wasted socket. +const NEG_OFF_KEY = 'coinos-neg-off'; +const NEG_OFF_MS = 24 * 60 * 60_000; +const negOff = new Map(); // normalized url -> until +try { + for (const [u, until] of Object.entries(JSON.parse(localStorage.getItem(NEG_OFF_KEY) || '{}'))) if (until > Date.now()) negOff.set(u, until); +} catch {} +function negRefused(url) { + negOff.set(relayKey(url), Date.now() + NEG_OFF_MS); + try { localStorage.setItem(NEG_OFF_KEY, JSON.stringify(Object.fromEntries([...negOff].slice(-100)))); } catch {} +} +export const negentropyOff = (url) => (negOff.get(relayKey(url)) || 0) > Date.now(); +// Reconcile `items` ([[created_at, id], ...]: the events we hold) with one +// relay under `filter`. Resolves { need, have } — the ids it holds that we +// don't, and ours it lacks — or null when it can't be done there: the relay +// doesn't speak it (a NOTICE, a CLOSED, a socket shut on the opening +// message, or silence — remembered for a day), refuses this query (NEG-ERR +// blocked/closed — not remembered), is sick, or we're offline. Its own +// socket: the pool's parser has no ear for NEG-* frames. +export async function reconcileOn(url, filter, items, { timeout = 8000 } = {}) { + const n = relayKey(url); + if (sick.has(n) || negentropyOff(url) || offline()) return null; + let ws; + try { ws = new WebSocket(url); } catch { return null; } + const neg = new Negentropy(new NegSet(items)); + const id = 'neg' + Math.random().toString(36).slice(2, 10); + return new Promise((resolve) => { + let done = false, heard = false; + const need = [], have = []; + const finish = (r, refused) => { + if (done) return; + done = true; clearTimeout(t); + if (refused) negRefused(url); + try { if (ws.readyState === 1) ws.send(JSON.stringify(['NEG-CLOSE', id])); } catch {} + try { ws.close(); } catch {} + resolve(r); + }; + const t = setTimeout(() => finish(null, !heard), timeout); + ws.onerror = () => finish(null, false); + ws.onclose = () => finish(null, !heard && !offline() && !hiddenTab()); + ws.onopen = async () => { + try { ws.send(JSON.stringify(['NEG-OPEN', id, filter, await neg.initiate()])); } catch { finish(null, false); } + }; + ws.onmessage = async (e) => { + let m; try { m = JSON.parse(e.data); } catch { return; } + if (!Array.isArray(m)) return; + if (m[0] === 'NEG-MSG' && m[1] === id) { + heard = true; + try { + const r = await neg.reconcile(m[2]); + need.push(...r.need); have.push(...r.have); + if (r.next) ws.send(JSON.stringify(['NEG-MSG', id, r.next])); + else finish({ need, have }, false); + } catch (err) { dlog('nostr: negentropy with ' + url + ' failed: ' + (err?.message || err)); finish(null, false); } + } else if (m[0] === 'NEG-ERR' && m[1] === id) { dlog('nostr: negentropy refused by ' + url + ': ' + m[2]); finish(null, false); } + else if (m[0] === 'NOTICE' && /negentropy|NEG-OPEN|bad msg/i.test(String(m[1] || ''))) finish(null, true); + else if (m[0] === 'CLOSED' && m[1] === id) finish(null, true); + }; + }); +} +// What `filter` matches on these relays, each event once, verified: by +// reconciliation where a relay speaks it (only the events we don't hold +// come, by id) and by the plain query where it doesn't. `items` are the +// events held, [[created_at, id], ...]. +export async function syncOn(relays, filter, items, { maxWait = 5000 } = {}) { + probeRelays(relays); + const live = readRelays(relays); + if (!live.length) return []; + const plain = [], byRelay = new Map(), seen = new Set(); + await Promise.all(live.map(async (url) => { + const r = await reconcileOn(url, filter, items).catch(() => null); + if (!r) { plain.push(url); return; } + for (const id of r.need) { + if (seen.has(id)) continue; // one copy is enough: the first relay that has it + seen.add(id); + if (!byRelay.has(url)) byRelay.set(url, []); + byRelay.get(url).push(id); + } + })); + const jobs = []; + if (plain.length) jobs.push(queryOn(plain, filter, maxWait)); + for (const [url, ids] of byRelay) for (let i = 0; i < Math.min(ids.length, 1000); i += 200) jobs.push(queryOn([url], { ids: ids.slice(i, i + 200) }, maxWait)); + const out = new Map(); + for (const evs of await Promise.all(jobs)) for (const e of evs || []) if (e && e.id && !out.has(e.id)) out.set(e.id, e); + return [...out.values()]; +} export async function queryManyOn(relays, filters, maxWait = 1500) { probeRelays(relays); try { diff --git a/tools/negentropy-sync-test.js b/tools/negentropy-sync-test.js new file mode 100644 index 0000000..e0257ca --- /dev/null +++ b/tools/negentropy-sync-test.js @@ -0,0 +1,159 @@ +// Catch-up by reconciliation (NIP-77). A relay that speaks negentropy — +// relay.coinos.io does — is no longer asked to re-serve its newest events: +// the DM subscription opens live with limit 0 and the wraps we lack are +// fetched by id after a reconcile; the Notifications refresh reconciles +// its newest 120 against what we hold and fetches only the new ones, and +// a refresh holding everything fetches nothing. A relay that answers +// "negentropy disabled" (nos.lol) gets the plain subscription with its +// backlog, and is remembered as such for a day: the next ask there goes +// straight out as a REQ. +// Run: bun tools/negentropy-sync-test.js +import puppeteer from './puppeteer.js'; +import { generateSecretKey, getPublicKey, finalizeEvent } from 'nostr-tools/pure'; +import { privateKeyFromSeedWords } from 'nostr-tools/nip06'; +import { buildHtml } from '../build.js'; +const sleep = (ms) => new Promise((r) => setTimeout(r, ms)); +let ok = true; +const check = (n, c, d = '') => { console.log(` ${c ? '✓' : '✗'} ${n}${d ? ' — ' + d : ''}`); if (!c) ok = false; }; +const key = () => { const sk = generateSecretKey(); return { sk, pk: getPublicKey(sk) }; }; +const FRIEND = key(), JOHNNY = key(), SENDER = key(); +const now = Math.floor(Date.now() / 1000); +const html = await buildHtml({ minify: true, pwa: false }); +// the client library plays the relay's side in the page +const negSrc = (await Bun.file('src/negentropy.js').text()).replace(/^export /gm, '') + '\nreturn { NegSet, Negentropy };'; +const server = Bun.serve({ port: 5337, fetch: (req) => { + const path = new URL(req.url).pathname; + if (path.startsWith('/punks') || path === '/verify-worker.js' || path.startsWith('/locales')) return new Response(Bun.file('dist' + path)); + return new Response(html, { headers: { 'content-type': 'text/html' } }); +} }); +const browser = await puppeteer.launch({ executablePath: '/usr/bin/google-chrome', headless: 'new', args: ['--no-sandbox'] }); +const page = await browser.newPage(); +const errs = []; +page.on('pageerror', (e) => errs.push(String(e))); +const click = (t) => page.evaluate((x) => { const e = [...document.querySelectorAll('button')].find((n) => n.textContent.trim().toLowerCase().includes(x)); if (e) { e.click(); return true; } return false; }, t); +const text = () => page.evaluate(() => document.body.innerText); +const waitText = async (x, ms = 25000) => { for (let i = 0; i < ms / 250; i++) { if ((await text()).toLowerCase().includes(x)) return true; await sleep(250); } return false; }; +const waitFor = async (fn, arg, ms = 10000, step = 100) => { for (let i = 0; i < ms / step; i++) { if (await page.evaluate(fn, arg)) return true; await sleep(step); } return false; }; +const sent = () => page.evaluate(() => window.__sent); +const rows = () => page.evaluate(() => [...document.querySelectorAll('.alert-row')].map((r) => r.innerText.replace(/\s+/g, ' ').trim())); +const host = (u) => u.replace(/^wss:\/\//, '').replace(/\/$/, ''); +const shape = (m) => { + const f = m[0] === 'REQ' ? m[2] : m[0] === 'NEG-OPEN' ? m[2] : null; + return m[0] + (f ? ' kinds=' + (f.kinds || []).join(',') + (f.ids ? ' ids=' + f.ids.length : '') + ('limit' in f ? ' limit=' + f.limit : '') + (f.since ? ' since=' + (f.since - Math.floor(Date.now() / 1000)) : '') : ''); +}; + +try { + await page.setViewport({ width: 390, height: 844 }); + await page.goto('http://localhost:5337/', { waitUntil: 'domcontentloaded' }); + await page.evaluate(() => localStorage.setItem('btc-wallet-network', 'regtest')); + await page.reload({ waitUntil: 'domcontentloaded' }); await sleep(400); + await click('create a new wallet'); await sleep(600); + await page.waitForSelector('.words .w .t'); + const mn = (await page.$$eval('.words .w .t', (l) => l.map((e) => e.textContent.trim()))).join(' '); + await click('skip verification'); + await waitText('receive', 20000); + const ME = getPublicKey(privateKeyFromSeedWords(mn)); + const MINE = finalizeEvent({ kind: 1, created_at: now - 600, tags: [], content: 'my post, replied to' }, FRIEND.sk); + const reply = (who, content, ago) => finalizeEvent({ kind: 1, created_at: now - ago, tags: [['e', MINE.id, '', 'root'], ['p', ME, '', 'mention']], content }, who.sk); + const NOTIFS = [reply(JOHNNY, 'johnny says hello there', 100), finalizeEvent({ kind: 7, created_at: now - 90, tags: [['e', MINE.id], ['p', ME]], content: '+' }, JOHNNY.sk), reply(FRIEND, 'a reply from a friend', 70)]; + // gift wraps to me (sealed to nobody: they will not open, which is beside the point here) + const WRAPS = [1, 2, 3].map((i) => finalizeEvent({ kind: 1059, created_at: now - i * 3600, tags: [['p', ME]], content: 'wrapped ' + i }, SENDER.sk)); + const PROFILES = [finalizeEvent({ kind: 0, created_at: now - 1000, tags: [], content: JSON.stringify({ name: 'Johnny' }) }, JOHNNY.sk), finalizeEvent({ kind: 0, created_at: now - 1000, tags: [], content: JSON.stringify({ name: 'Friend' }) }, FRIEND.sk)]; + await page.evaluateOnNewDocument((fx, negSrc) => { + const { NegSet, Negentropy } = new Function(negSrc)(); + window.__sent = []; + const Real = window.WebSocket; + const say = (ws, m) => { if (ws.readyState === 1) ws.dispatchEvent(new MessageEvent('message', { data: JSON.stringify(m) })); }; + const store = [...fx.notifs, ...fx.wraps, fx.mine, ...fx.profiles]; + const matches = (f) => store.filter((e) => (!f.kinds || f.kinds.includes(e.kind)) && (!f.authors || f.authors.includes(e.pubkey)) && (!f.ids || f.ids.includes(e.id)) + && (!f.since || e.created_at >= f.since) && (!f.until || e.created_at <= f.until) + && (!f['#p'] || e.tags.some((t) => t[0] === 'p' && f['#p'].includes(t[1]))) && (!f['#e'] || e.tags.some((t) => t[0] === 'e' && f['#e'].includes(t[1])))); + const newest = (evs, f) => { const s = [...evs].sort((a, b) => b.created_at - a.created_at); return 'limit' in f ? s.slice(0, f.limit) : s; }; + window.WebSocket = function (...a) { + const ws = new Real(...a); + const url = String(a[0]); + const coinos = /relay\.coinos\.io/.test(url); + const negs = new Map(); // subid -> relay side + const send = ws.send.bind(ws); + ws.send = (data) => { + try { + const m = JSON.parse(data); + window.__sent.push({ url, m }); + if (m[0] === 'REQ') { + const hit = newest(matches(m[2]), m[2]); + setTimeout(() => { for (const e of hit) say(ws, ['EVENT', m[1], e]); say(ws, ['EOSE', m[1]]); }, 30); + return; + } + if (m[0] === 'NEG-OPEN') { + if (!coinos) { setTimeout(() => say(ws, ['NOTICE', 'ERROR: bad msg: negentropy disabled']), 30); return; } + const items = newest(matches(m[2]), m[2]).map((e) => [e.created_at, e.id]); + const n = new Negentropy(new NegSet(items)); + negs.set(m[1], n); + n.reconcile(m[3]).then((r) => say(ws, ['NEG-MSG', m[1], r.next])); + return; + } + if (m[0] === 'NEG-MSG') { const n = negs.get(m[1]); if (n) n.reconcile(m[2]).then((r) => say(ws, ['NEG-MSG', m[1], r.next])); return; } + if (m[0] === 'NEG-CLOSE') { negs.delete(m[1]); return; } + if (m[0] === 'EVENT') { setTimeout(() => say(ws, ['OK', m[1].id, true, '']), 0); return; } + } catch {} + return send(data); + }; + return ws; + }; + window.WebSocket.prototype = Real.prototype; Object.assign(window.WebSocket, Real); + }, { me: ME, mine: MINE, notifs: NOTIFS, wraps: WRAPS, profiles: PROFILES }, negSrc); + await page.reload({ waitUntil: 'domcontentloaded' }); + await waitText('receive', 20000); + + console.log('\n[the DM subscription]'); + await waitFor(() => window.__sent.some((x) => /coinos/.test(x.url) && x.m[0] === 'REQ' && (x.m[2].ids || []).length), null, 15000); + await sleep(500); + let log = await sent(); + const at = (re) => log.filter((x) => re.test(x.url)).map((x) => shape(x.m)); + const coinosDm = at(/coinos/).filter((s) => /kinds=1059/.test(s)); + check('relay.coinos.io: the live wrap subscription opens with limit 0', coinosDm.some((s) => s === 'REQ kinds=1059 limit=0'), coinosDm.join(' | ')); + check('…and a reconcile of the newest 400', coinosDm.some((s) => s === 'NEG-OPEN kinds=1059 limit=400')); + check('…the relay is never asked to re-serve its backlog', !coinosDm.some((s) => /^REQ kinds=1059 limit=400/.test(s))); + const idsAsk = log.find((x) => /coinos/.test(x.url) && x.m[0] === 'REQ' && (x.m[2].ids || []).length); + check('…and the wraps we lack are fetched by id', !!idsAsk && WRAPS.every((w) => idsAsk.m[2].ids.includes(w.id)), idsAsk ? idsAsk.m[2].ids.length + ' ids' : 'no ask by id'); + const nosDm = at(/nos\.lol/).filter((s) => /kinds=1059/.test(s)); + check('nos.lol says negentropy is disabled: it gets the plain subscription, backlog and all', nosDm.includes('NEG-OPEN kinds=1059 limit=400') && nosDm.includes('REQ kinds=1059 limit=400'), nosDm.join(' | ')); + check('…and is remembered as such', await page.evaluate(() => /nos\.lol/.test(localStorage.getItem('coinos-neg-off') || ''))); + + console.log('\n[Notifications]'); + const before = log.length; + await page.evaluate(() => document.querySelector('.app-nav-button[aria-label="Notifications"]')?.click()); + let got = []; + for (let i = 0; i < 60; i++) { got = await rows(); if (got.length >= 3) break; await sleep(250); } + check('the three notifications show', got.length === 3, got.join(' | ')); + log = (await sent()).slice(before); + const coinosN = log.filter((x) => /coinos/.test(x.url)).map((x) => shape(x.m)).filter((s) => /kinds=1,/.test(s) || /ids=/.test(s)); + console.log(' coinos:', log.filter((x) => /coinos/.test(x.url)).map((x) => shape(x.m) + (x.m[0] === 'REQ' && x.m[2].ids ? ':' + x.m[2].ids.map((i) => i.slice(0, 6)).join(',') : '') + (x.m[0] === 'NEG-MSG' ? ' ' + x.m[2].length + 'ch' : '')).join(' | ')); + console.log(' notif ids:', NOTIFS.map((n) => n.id.slice(0, 6)).join(','), 'mine', MINE.id.slice(0, 6)); + check('relay.coinos.io: reconciled, its newest 120', coinosN.some((s) => /^NEG-OPEN kinds=1,1111,6,7.* limit=120$/.test(s)), coinosN.join(' | ')); + check('…not asked for them outright', !coinosN.some((s) => /^REQ kinds=1,1111,6,7.* limit=120$/.test(s))); + // (the first refresh ran at boot, for the unread count: that is the one that fetched them) + const all = await sent(); + const nAsk = all.find((x) => /coinos/.test(x.url) && x.m[0] === 'REQ' && (x.m[2].ids || []).length && NOTIFS.some((n) => x.m[2].ids.includes(n.id))); + check('…the three we lacked fetched by id, once, at the boot refresh', !!nAsk && nAsk.m[2].ids.length === 3 && all.indexOf(nAsk) < before, nAsk ? nAsk.m[2].ids.length + ' ids at ' + all.indexOf(nAsk) + ' (page opened at ' + before + ')' : 'none'); + check('…and the page\'s own refresh, holding them, fetched nothing', !log.some((x) => /coinos/.test(x.url) && x.m[0] === 'REQ' && (x.m[2].ids || []).some((id) => NOTIFS.some((n) => n.id === id)))); + const nosN = log.filter((x) => /nos\.lol/.test(x.url)).map((x) => shape(x.m)); + check('nos.lol: the plain query, straight out, no reconcile tried', nosN.some((s) => /^REQ kinds=1,1111,6,7.* limit=120$/.test(s)) && !nosN.some((s) => s.startsWith('NEG-OPEN')), nosN.join(' | ')); + + console.log('\n[a refresh holding everything, after a reload]'); + await page.reload({ waitUntil: 'domcontentloaded' }); + await waitText('receive', 20000); + const b2 = (await sent()).length; + await page.evaluate(() => document.querySelector('.app-nav-button[aria-label="Notifications"]')?.click()); + for (let i = 0; i < 60; i++) { got = await rows(); if (got.length >= 3) break; await sleep(250); } + check('the notifications are there from the cache', got.length === 3, got.join(' | ')); + await waitFor((n) => window.__sent.slice(n).some((x) => /coinos/.test(x.url) && x.m[0] === 'NEG-CLOSE'), b2, 10000); + await sleep(800); + log = (await sent()).slice(b2); + const again = log.filter((x) => /coinos/.test(x.url)).map((x) => shape(x.m)).filter((s) => /kinds=1,/.test(s) || /ids=/.test(s)); + check('relay.coinos.io reconciled again…', again.some((s) => /^NEG-OPEN kinds=1,1111,6,7.* limit=120$/.test(s)), again.join(' | ')); + check('…and nothing was fetched: we held it all', !log.some((x) => /coinos/.test(x.url) && x.m[0] === 'REQ' && (x.m[2].ids || []).some((id) => NOTIFS.some((n) => n.id === id)))); + check('no page errors', errs.length === 0, errs.slice(0, 2).join(' | ')); +} catch (e) { console.error(e); ok = false; } finally { await browser.close(); server.stop(true); } +console.log(ok ? '\n✅ catch-up by reconciliation where the relay speaks it, the plain query where it does not' : '\n❌ failed'); +process.exit(ok ? 0 : 1); diff --git a/tools/negentropy-test.js b/tools/negentropy-test.js new file mode 100644 index 0000000..28a1f30 --- /dev/null +++ b/tools/negentropy-test.js @@ -0,0 +1,64 @@ +// The negentropy client (NIP-77, src/negentropy.js) on the wire as strfry +// and the reference implementation speak it: the opening message for a +// fixed set is byte-identical to the reference's (checked against +// hoytech/negentropy's JS on 2026-10-16: seven pairings of ours and theirs +// as client and relay agreed on every have/need), and a reconciliation +// against our own relay side finds exactly what each side lacks — in one +// round for a small difference, in several when the relay's frame is +// small — and ends with nothing to say once the sets match. +// Run: bun tools/negentropy-test.js +import assert from 'node:assert/strict'; +import { createHash } from 'node:crypto'; +import { NegSet, Negentropy } from '../src/negentropy.js'; +const id = (n) => createHash('sha256').update('item-' + n).digest('hex'); +const sha = (s) => createHash('sha256').update(s).digest('hex'); + +// the fixed sets the reference was run on +const A = Array.from({ length: 120 }, (_, i) => [1000 + Math.floor(i / 3), id(i)]); +const B = [...A.filter((_, i) => i % 17 !== 0), ...Array.from({ length: 9 }, (_, i) => [1000 + i * 4, id(1000 + i)])]; +const first = await new Negentropy(new NegSet(A)).initiate(); +assert.equal(first.length, 634, 'the opening message for the fixed set is 317 bytes'); +assert.equal(first.slice(0, 2), '61', 'protocol version 1'); +assert.equal(sha(first), 'c926c013652f6e2f727abff15c0838aebd7f613fb4749ee691952dfc8be8ff81', 'the opening message is the reference implementation\'s, byte for byte'); + +async function talk(client, relay) { + let msg = await client.initiate(), rounds = 0; + const have = [], need = []; + while (msg) { + rounds++; + const r = await relay.reconcile(msg); + assert(r.next, 'the relay side always answers'); + const c = await client.reconcile(r.next); + have.push(...c.have); need.push(...c.need); msg = c.next; + } + return { rounds, have: have.sort(), need: need.sort() }; +} +{ + const r = await talk(new Negentropy(new NegSet(A)), new Negentropy(new NegSet(B))); + assert.equal(r.rounds, 1, 'a small difference reconciles in one round'); + assert.deepEqual(r.have, A.filter((_, i) => i % 17 === 0).map((x) => x[1]).sort(), 'have: the ids the relay lacks'); + assert.deepEqual(r.need, Array.from({ length: 9 }, (_, i) => id(1000 + i)).sort(), 'need: the ids we lack'); +} +{ + const r = await talk(new Negentropy(new NegSet(A)), new Negentropy(new NegSet(A))); + assert.equal(r.rounds, 1); assert.deepEqual(r.have, []); assert.deepEqual(r.need, []); + const r2 = await talk(new Negentropy(new NegSet([])), new Negentropy(new NegSet(B))); + assert.equal(r2.need.length, B.length, 'holding nothing, we need everything'); + const r3 = await talk(new Negentropy(new NegSet(B)), new Negentropy(new NegSet([]))); + assert.equal(r3.have.length, B.length, 'the relay holding nothing lacks everything'); +} +{ + // a big set and a relay with a small frame: many rounds, the same answer + const big = Array.from({ length: 3000 }, (_, i) => [1000 + i, id(i)]); + const theirs = [...big.filter((_, i) => i % 101 !== 0), ...Array.from({ length: 25 }, (_, i) => [1000 + i * 7, id(5000 + i)])]; + const r = await talk(new Negentropy(new NegSet(big)), new Negentropy(new NegSet(theirs), { frameSizeLimit: 4096 })); + assert(r.rounds > 1, 'a small frame takes several rounds: ' + r.rounds); + assert.equal(r.have.length, 30); assert.equal(r.need.length, 25); +} +{ + // the set drops duplicates and bad ids, and sorts by time then id + const s = new NegSet([[5, id(1)], [5, id(1)], [3, 'nope'], [4, id(2)], [5, id(0)]]); + assert.equal(s.size, 3); + assert.deepEqual(s.items.map((x) => x.ts), [4, 5, 5]); +} +console.log('✓ negentropy: the reference\'s bytes, one round for a small difference, several for a small frame, nothing to say when equal'); |