Group chats notice a relay that dropped their subscription

A group that misses a commit is stuck until it hears it, and a group
founded by White Noise can live on a single relay. Its subscriptions now
report a close to the chat watchdog, and one made while a relay was being
skipped as dead is rebuilt when that relay comes back.

Co-Authored-By: Claude Fable 5.1 <[email protected]>

3 files changed, 42 insertions, 9 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))],
diff --git a/tools/marmot-chat-test.js b/tools/marmot-chat-test.js
index cb56e46..da56fba 100644
--- a/tools/marmot-chat-test.js
+++ b/tools/marmot-chat-test.js
@@ -65,6 +65,15 @@ const browser = await puppeteer.launch({ executablePath: '/usr/bin/google-chrome
 const page = await browser.newPage();
 await page.setViewport({ width: 1100, height: 850 });
 page.on('pageerror', (e) => { console.log('   page error:', e.message); ok = false; });
+// DEBUG=1 turns on the app's debug switch and echoes the group-chat lines
+if (process.env.DEBUG) {
+  await page.evaluateOnNewDocument(() => localStorage.setItem('coinos-debug', '1'));
+  page.on('console', async (m) => {
+    if (!/^marmot/.test(m.text())) return;
+    const args = await Promise.all(m.args().map((a) => a.evaluate((v) => (v instanceof Error ? v.message : typeof v === 'string' ? v : JSON.stringify(v))).catch(() => '?')));
+    console.log('   [page]', args.join(' ').slice(0, 300));
+  });
+}
 // every relay is the local relay
 if (!SITE) await page.evaluateOnNewDocument((relay) => {
   const Real = window.WebSocket;
@@ -115,6 +124,9 @@ try {
   await page.waitForSelector('textarea');
   await page.type('textarea', mnemonic);
   await click('button', 'Open wallet');
+  // a mainnet wallet is first offered a username: not now
+  await until(async () => /receive|not now/i.test(await bodyText()), 20000);
+  for (let i = 0; i < 3 && await click('button', 'Not now'); i++) await sleep(600);
   await waitText('receive', 15000);
 
   console.log('\n[invitable without doing anything]');
@@ -132,7 +144,18 @@ try {
   await click('button', 'Join');
   check('the group opens', !!await waitText('2 members'));
   wn(['messages', 'send', gid, 'hello from white noise'], alice);
-  check('their message arrives', !!await waitText('hello from white noise'));
+  const arrived = !!await waitText('hello from white noise');
+  check('their message arrives', arrived);
+  if (!arrived && process.env.DEBUG) {
+    const chat = (wn(['chats', 'list'], alice).chats || []).find((c) => c.group_id === gid) || {};
+    const r = chat.nostr_routing || {};
+    console.log('   wn routing', JSON.stringify(r.relays), 'last', JSON.stringify(chat.last_message || {}).slice(0, 200));
+    console.log('   wn messages', JSON.stringify(wn(['messages', 'list', gid, '--limit', '5'], alice)).slice(0, 400));
+    for (const url of r.relays || []) {
+      const evs = await new Promise((res) => { const ws = new WebSocket(url), out = []; ws.onopen = () => ws.send(JSON.stringify(['REQ', 'q', { kinds: [445], '#h': [r.nostr_group_id_hex] }])); ws.onmessage = (m) => { const d = JSON.parse(m.data); if (d[0] === 'EVENT') out.push(d[2]); else res(out); }; setTimeout(() => res(out), 4000); });
+      console.log('   ', url, 'holds', evs.length, 'group events', evs.map((e) => e.created_at).join(','));
+    }
+  }
   await say('hello from coinos');
   check('ours reaches wn', !!await until(() => JSON.stringify(wn(['messages', 'list', gid, '--limit', '20'], alice)).includes('hello from coinos')));
   await page.screenshot({ path: shot('marmot-group.png') });