diff options
Diffstat (limited to 'src')
| -rw-r--r-- | src/features/messages.js | 3 | ||||
| -rw-r--r-- | src/marmot-client.js | 23 |
2 files changed, 18 insertions, 8 deletions
diff --git a/src/features/messages.js b/src/features/messages.js index 917202e..812a124 100644 --- a/src/features/messages.js +++ b/src/features/messages.js @@ -1237,6 +1237,7 @@ export function messagesFeature(ctx) { if (dmStarted) for (const u of liveRelayList(DM_RELAYS)) urls.add(u); for (const room of rooms.values()) if (room.subscribed) for (const u of liveRelayList(room.relays)) urls.add(u); if (wn) for (const u of liveRelayList(wn.relaysInUse())) urls.add(u); + if (wn && wn.starved()) subsDead = true; // a group's relay came back after being skipped const dead = [...urls].filter((u) => !relayAlive(u)); if (!dead.length && !subsDead) { rebuildBackoff = 15_000; return; } const now = Date.now(); @@ -2978,7 +2979,7 @@ export function messagesFeature(ctx) { let relays = null; try { relays = JSON.parse(localStorage.getItem('coinos-group-relays') || 'null'); } catch {} const c = m.marmotClient({ - scope: wallet._cacheKey() + ':' + pk, pubkey: pk, identity, on: onWn, + scope: wallet._cacheKey() + ':' + pk, pubkey: pk, identity, on: onWn, subClosed: noteSubClosed, ...(Array.isArray(relays) && relays.length ? { relays } : {}), }); wn = c; diff --git a/src/marmot-client.js b/src/marmot-client.js index 1bad242..1fef568 100644 --- a/src/marmot-client.js +++ b/src/marmot-client.js @@ -9,7 +9,7 @@ import * as M from './marmot.js'; import { marmotStore } from './marmot-store.js'; -import { subscribeOn, queryOn, publishOn, fetchInboxRelays, PROFILE_RELAYS, finalizeEvent } from './nostr.js'; +import { subscribeOn, queryOn, publishOn, fetchInboxRelays, liveRelayList, PROFILE_RELAYS, finalizeEvent } from './nostr.js'; import { wrapDM } from './dm.js'; import { dlog } from './debug.js'; @@ -24,13 +24,13 @@ const REACH_TTL = 10 * 60_000; const now = () => Math.floor(Date.now() / 1000); const sleep = (ms) => new Promise((r) => setTimeout(r, ms)); -export function marmotClient({ scope, pubkey, identity, on, relays = GROUP_RELAYS }) { +export function marmotClient({ scope, pubkey, identity, on, relays = GROUP_RELAYS, subClosed }) { const store = marmotStore(scope); const groups = new Map(); // mls group id (hex) -> record const subs = new Map(); // route id -> { unsub, gid } const reach = new Map(); // pubkey -> { at, p } const inbox = new Map(); // gid -> queued 445 events - let dev = null, stopped = false, loaded = null, kpBusy = null; + let dev = null, stopped = false, loaded = null, kpBusy = null, rebuilding = false; const dirty = new Set(); let flushTimer = 0, rotateTimer = 0; @@ -139,12 +139,16 @@ export function marmotClient({ scope, pubkey, identity, on, relays = GROUP_RELAY if (stopped || g.removed) return; for (const r of M.routes(g)) { if (subs.has(r.id)) continue; - const unsub = subscribeOn(r.relays, { kinds: [M.KIND.GROUP], '#h': [r.id], since: M.since(g), limit: 500 }, (ev) => queue(g.id, ev)); - subs.set(r.id, { unsub, gid: g.id }); + // a relay that drops the subscription is the caller's watchdog's business: + // a group that misses a commit is stuck until it hears it + const unsub = subscribeOn(r.relays, { kinds: [M.KIND.GROUP], '#h': [r.id], since: M.since(g), limit: 500 }, (ev) => queue(g.id, ev), + subClosed ? { onclose: () => { if (!rebuilding && !stopped) subClosed(); } } : {}); + subs.set(r.id, { unsub, gid: g.id, relays: r.relays, live: liveRelayList(r.relays).length }); } } function unwatch(gid) { - for (const [id, s] of subs) if (s.gid === gid) { try { s.unsub(); } catch {} subs.delete(id); } + rebuilding = true; + try { for (const [id, s] of subs) if (s.gid === gid) { try { s.unsub(); } catch {} subs.delete(id); } } finally { rebuilding = false; } } // Relays replay history newest-first and commits only apply oldest-first: @@ -343,13 +347,15 @@ export function marmotClient({ scope, pubkey, identity, on, relays = GROUP_RELAY } function resubscribe() { - for (const s of subs.values()) { try { s.unsub(); } catch {} } + rebuilding = true; // our own closes are not a relay giving up on us + try { for (const s of subs.values()) { try { s.unsub(); } catch {} } } finally { rebuilding = false; } subs.clear(); for (const g of groups.values()) watch(g); } function stop() { stopped = true; + rebuilding = true; clearTimeout(rotateTimer); for (const s of subs.values()) { try { s.unsub(); } catch {} } subs.clear(); @@ -364,6 +370,9 @@ export function marmotClient({ scope, pubkey, identity, on, relays = GROUP_RELAY peer: (g) => M.directPeer(g), accounts: (g) => M.accounts(g.tip), isAdmin: (g) => M.isAdmin(g), + // A relay that was being skipped as dead when a group subscribed is + // back: that subscription is listening on fewer relays than it could. + starved: () => [...subs.values()].some((s) => liveRelayList(s.relays).length > s.live), // has this device ever been given its standing (a signed proof)? isSetUp: () => !!dev, relaysInUse: () => [...new Set([...groups.values()].filter((g) => !g.removed).flatMap(routeRelays))], |