aboutsummaryrefslogtreecommitdiffstats
path: root/src/nostr.js
diff options
context:
space:
mode:
Diffstat (limited to 'src/nostr.js')
-rw-r--r--src/nostr.js252
1 files changed, 227 insertions, 25 deletions
diff --git a/src/nostr.js b/src/nostr.js
index b76e865..731c076 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';
@@ -185,7 +186,12 @@ pool.ensureRelay = async (url, params) => {
relay._coinosRL = true;
const notice = relay.onnotice;
relay.onnotice = (msg) => {
- if (RATE_LIMITED.test(String(msg))) markRelayRateLimited(relay.url, String(msg));
+ // Primal's "too many concurrent REQs" is advice, not a refusal: every
+ // REQ past its twenty is answered all the same (measured: 60 at once,
+ // 60 EOSEs, 12 notices). Cooling it off for 30 s at every boot — when
+ // the relay lists, profiles and the feed all ask at once — only lost
+ // us its answers. A CLOSED with that reason still counts (below).
+ if (RATE_LIMITED.test(String(msg)) && !/concurrent/i.test(String(msg))) markRelayRateLimited(relay.url, String(msg));
try { notice?.call(relay, msg); } catch {}
};
const subscribe = relay.subscribe.bind(relay);
@@ -283,6 +289,54 @@ export function resetRelay(url) {
try { r.close(); } catch {}
pool.relays.delete(n);
}
+// Back from a freeze (a phone's backgrounded tab: timers stopped, the
+// network's idea of every socket long gone), the pool's sockets still read
+// OPEN, and anything sent on them vanishes. Left to itself the library
+// notices one in up to 29 s (its ping) + 20 s (the ping's timeout), then
+// waits 10 s more before redialing — a minute of a feed that won't catch up
+// ("no new notes since 11:35?", Adam, 2026-11-01). So on resume every open
+// socket is asked one tiny REQ; one that answers within the deadline is
+// fine, one that doesn't is closed and redialed now, its subscriptions
+// refired by the library with `since` set to the last event each saw. A
+// relay already waiting out a reconnect backoff is dialed now too.
+const WAKE_PING_MS = 2500;
+const DUMMY_ID = 'a'.repeat(64);
+function pingRelay(r, ms) {
+ return new Promise((resolve) => {
+ let done = false;
+ const fin = (ok) => { if (done) return; done = true; clearTimeout(t); resolve(ok); };
+ const t = setTimeout(() => fin(false), ms);
+ try {
+ const sub = r.subscribe([{ ids: [DUMMY_ID], limit: 0 }], {
+ label: '<wake-ping>',
+ oneose: () => { fin(true); try { sub.close(); } catch {} },
+ onclose: () => fin(true), // a relay refusing the REQ is a relay that answers
+ eoseTimeout: ms + 1000, // ours fires first: the library's would count as an EOSE
+ });
+ } catch { fin(false); }
+ });
+}
+function redialRelay(r, why) {
+ console.warn('nostr: ' + r.url + ' ' + why + ' — redialing');
+ // the dead socket's close would reach the library long after (a closing
+ // handshake nobody answers), and undo the new connection: detached first
+ const ws = r.ws;
+ try { if (ws) { ws.onopen = ws.onclose = ws.onerror = ws.onmessage = null; ws.close(); } } catch {}
+ try { r.handleHardClose(why); } catch {} // the library's own path: subscriptions kept, a reconnect queued
+ if (r.reconnectTimeoutHandle) { clearTimeout(r.reconnectTimeoutHandle); r.reconnectTimeoutHandle = undefined; }
+ r.connect().catch(() => {}); // now, not after the backoff
+}
+export async function wakeRelays(ms = WAKE_PING_MS) {
+ const checks = [];
+ for (const r of pool.relays.values()) {
+ if (r.connected && r.ws && r.ws.readyState === 1) checks.push(pingRelay(r, ms).then((ok) => { if (!ok) redialRelay(r, 'silent after resume'); }));
+ else if (!r.connectionPromise && r.reconnectTimeoutHandle) {
+ clearTimeout(r.reconnectTimeoutHandle); r.reconnectTimeoutHandle = undefined;
+ r.connect().catch(() => {});
+ }
+ }
+ await Promise.all(checks);
+}
// the relays a subscription on this list actually dials (sick ones skipped)
export const liveRelayList = (list) => liveRelays(list);
// ...and the ones a one-shot read goes to (rate-limited ones skipped too)
@@ -353,34 +407,183 @@ async function authRelay(relay, onauth) {
}
return relay.auth(onauth);
}
+// ---- one-shot reads: per relay, closed on its own EOSE, a few at a time --
+// The pool's own querySync keeps every relay's REQ open until the SLOWEST
+// relay has answered or the wait runs out: asked beside a relay still
+// dialing, our own relay — done in fifty milliseconds — carried each REQ
+// for five seconds, and a feed's opening asks stacked up on it by the
+// dozen. Each relay's REQ now closes on that relay's EOSE. And no relay
+// carries more than REQ_SLOTS of them at once: the rest wait their turn,
+// which is a beat on a relay that answers and no longer than the wait on
+// one that doesn't. (A Following feed opened from the laptop sent some 560
+// REQs in twenty seconds across 57 relays, over a hundred each to damus
+// and primal; damus answered the home IP "banned: too many rate-limit
+// violations" on 2026-11-02, and primal notices past twenty at once.)
+const REQ_SLOTS = 10;
+const slots = new Map(); // normalized url -> { n, q: [resolve] }
+function takeSlot(url) {
+ const k = relayKey(url);
+ let s = slots.get(k);
+ if (!s) slots.set(k, (s = { n: 0, q: [] }));
+ if (s.n < REQ_SLOTS) { s.n++; return Promise.resolve(false); }
+ const at = Date.now();
+ return new Promise((resolve) => s.q.push(() => resolve(Date.now() - at))); // how long it waited
+}
+function freeSlot(url) {
+ const s = slots.get(relayKey(url));
+ if (!s) return;
+ const next = s.q.shift();
+ if (next) next(); // the slot passes straight to the next in line
+ else s.n--;
+}
+export const reqSlotsOf = (url) => { const s = slots.get(relayKey(url)); return s ? { busy: s.n, waiting: s.q.length } : { busy: 0, waiting: 0 }; };
+// One REQ carrying `filters` on one relay. Settles at the relay's EOSE, its
+// CLOSED, a failed dial, or maxWait — whichever is first — and sends CLOSE.
+function askRelay(url, filters, maxWait, onEvent) {
+ return new Promise((resolve) => {
+ let finished = false, sub = null, timer = 0;
+ const finish = () => {
+ if (finished) return;
+ finished = true;
+ clearTimeout(timer);
+ if (sub) { try { sub.close('done'); } catch {} }
+ freeSlot(url);
+ resolve();
+ };
+ takeSlot(url).then((waited) => {
+ // a turn that came only after the ask's own wait had run out: the
+ // caller has moved on, and a storm of late REQs helps nobody
+ if (waited > maxWait) { finish(); return; }
+ timer = setTimeout(finish, maxWait + 1000);
+ try {
+ sub = pool.subscribeMap(filters.map((filter) => ({ url, filter })), { maxWait, onevent: onEvent, oneose: finish, onclose: finish });
+ if (finished) { try { sub.close('done'); } catch {} }
+ } catch { finish(); }
+ });
+ });
+}
+// All relays asked at once; an event is counted once however many hold it.
+async function askRelays(relays, filters, maxWait, onEvent) {
+ const seen = new Set();
+ await Promise.all(relays.map((url) => askRelay(url, filters, maxWait, (ev) => {
+ if (!ev || seen.has(ev.id)) return;
+ seen.add(ev.id);
+ onEvent(ev);
+ })));
+}
export async function queryOn(relays, filter, maxWait = 1500) {
probeRelays(relays);
try {
const live = readRelays(relays);
if (!live.length) return [];
- const evs = await pool.querySync(live, filter, { maxWait });
+ const evs = [];
+ await askRelays(live, [filter], maxWait, (ev) => evs.push(ev));
const oks = await verifyEventsAsync(evs);
return evs.filter((_, i) => oks[i]);
} catch { return []; }
}
// 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 {
const live = readRelays(relays);
if (!live.length || !filters.length) return [];
- const evs = await new Promise((resolve) => {
- const got = [];
- let sub = null, done = false;
- const finish = () => { if (!done) { done = true; resolve(got); } };
- sub = pool.subscribeMap(live.flatMap((url) => filters.map((filter) => ({ url, filter }))), {
- maxWait,
- onevent: (e) => got.push(e),
- oneose: () => { try { sub?.close('closed automatically on eose'); } catch {} finish(); },
- onclose: finish,
- });
- });
+ const evs = [];
+ await askRelays(live, filters, maxWait, (ev) => evs.push(ev));
const oks = await verifyEventsAsync(evs);
return evs.filter((_, i) => oks[i]);
} catch { return []; }
@@ -393,18 +596,12 @@ export async function queryStreamOn(relays, filter, onEvent, maxWait = 1500) {
try {
const live = readRelays(relays);
if (!live.length) return;
- await new Promise((resolve) => {
- pool.subscribeManyEose(live, filter, {
- maxWait,
- onevent(event) {
- pending.push(verifyOneAsync(event).then((ok) => ok && onEvent(event)).catch(() => {}));
- },
- onclose: resolve,
- });
+ await askRelays(live, [filter], maxWait, (event) => {
+ pending.push(verifyOneAsync(event).then((ok) => ok && onEvent(event)).catch(() => {}));
});
} catch {} finally { await Promise.all(pending); }
}
-export async function publishOn(relays, evt, { onauth = null } = {}) {
+export async function publishOn(relays, evt, { onauth = null, onrefused = null } = {}) {
try {
const live = liveRelays(relays);
// A rejected publish is a relay refusing the event (policy, rate limit,
@@ -426,7 +623,12 @@ export async function publishOn(relays, evt, { onauth = null } = {}) {
// "relay connection errored/closed" means the socket never came up —
// that's a sick relay, not a rejected event.
if (/connect|closed|errored|timeout|network/i.test(msg)) markRelaySick(live[i], msg);
- else console.warn(`nostr: publish kind ${evt.kind} refused by ${live[i]}:`, msg);
+ else {
+ console.warn(`nostr: publish kind ${evt.kind} refused by ${live[i]}:`, msg);
+ // the relay said no to this event (policy, a rate limit, a ban): a
+ // caller can stop asking it
+ if (onrefused) try { onrefused(live[i], msg); } catch {}
+ }
return false;
}));
// The first acceptance IS delivery — one durable copy is enough, and
@@ -505,7 +707,7 @@ export const PROFILE_RELAYS = HOME_RELAYS;
// Fetch a recipient's profile (name + picture) for a pubkey, newest across relays.
export async function fetchNostrProfile(pubkeyHex, relays = PROFILE_RELAYS) {
let events;
- try { events = await pool.querySync(relays, { kinds: [0], authors: [pubkeyHex] }, { maxWait: 5000 }); } catch { return null; }
+ try { events = await queryOn(relays, { kinds: [0], authors: [pubkeyHex] }, 5000); } catch { return null; }
const best = events.sort((a, b) => b.created_at - a.created_at)[0];
if (!best) return null;
try {
@@ -527,7 +729,7 @@ export async function fetchNostrProfile(pubkeyHex, relays = PROFILE_RELAYS) {
// 10050), else their NIP-65 read relays (kind 10002), else [] (caller falls back).
export async function fetchInboxRelays(pubkeyHex, relays = PROFILE_RELAYS) {
let events;
- try { events = await pool.querySync(relays, { kinds: [10002, 10050], authors: [pubkeyHex] }, { maxWait: 5000 }); } catch { return []; }
+ try { events = await queryOn(relays, { kinds: [10002, 10050], authors: [pubkeyHex] }, 5000); } catch { return []; }
const newest = (kind) => events.filter((e) => e.kind === kind).sort((a, b) => b.created_at - a.created_at)[0];
const dm = newest(10050);
if (dm) { const r = dm.tags.filter((t) => t[0] === 'relay' && t[1]).map((t) => t[1]); if (r.length) return r; }
generated by cgit v1.2.3 (git 2.51.0) at 2026-11-02 19:53:10 +0000