From 44cf906b295f61d254f558607ea4cff1797b7a4a Mon Sep 17 00:00:00 2001 From: Ocnrb Date: Thu, 24 Sep 2026 17:46:20 +0100 Subject: [PATCH 1/2] Rebuild the Streamr client after a failed node start and subscribe again on any new client Co-authored-by: Claude Opus 5.5 --- src/js/app.js | 42 +++++++- src/js/dm.js | 12 +++ src/js/streamr.js | 133 +++++++++++++++++++---- src/js/streamr/NodeRevival.js | 112 +++++++++++++++++++ src/js/subscriptionManager.js | 11 ++ src/js/ui/SettingsUI.js | 1 - tests/unit/nodeRevival.test.js | 148 ++++++++++++++++++++++++++ tests/unit/streamr.core.test.js | 1 + tests/unit/streamr.disconnect.test.js | 122 +++++++++++++++++++-- tests/unit/streamr.nodeWatch.test.js | 78 ++++++++++++++ 10 files changed, 626 insertions(+), 34 deletions(-) create mode 100644 src/js/streamr/NodeRevival.js create mode 100644 tests/unit/nodeRevival.test.js create mode 100644 tests/unit/streamr.nodeWatch.test.js diff --git a/src/js/app.js b/src/js/app.js index 953dac1..e5f5949 100644 --- a/src/js/app.js +++ b/src/js/app.js @@ -94,12 +94,11 @@ class App { secureStorage.onSentDataChanged = () => syncManager.scheduleAutoPush(5000); settingsUI.setDependencies({ - connectWallet: () => walletFlows.connectWallet(), - onSyncTransportReconnected: () => this.pullSyncedStateAndBlobs().catch((error) => { - Logger.debug('Sync: Auto-pull after reconnect failed (non-critical):', error.message); - }) + connectWallet: () => walletFlows.connectWallet() }); + streamrController.onClientReplaced(() => this.resumeAfterClientReplaced()); + // Wire wallet flows with app-level callbacks walletFlows.init({ disconnectWallet: (opts) => this.disconnectWallet(opts), @@ -266,6 +265,7 @@ class App { syncManager.stopSnapshotWatch(); pushOnHide(); } else if (document.visibilityState === 'visible') { + streamrController.revival.kick(); if (!syncManager.isAutoSyncAllowed('foreground')) return; syncManager.startSnapshotWatch(); syncManager.checkForNewSnapshot().catch((error) => { @@ -273,6 +273,40 @@ class App { }); } }); + + window.addEventListener('online', () => streamrController.revival.kick()); + } + + /** + * A replaced Streamr client took every subscription with it: make them + * again, and retry what only the network can carry (owed rotations, the + * inbox push registration, sync). + */ + async resumeAfterClientReplaced() { + if (!authManager.isConnected()) return; + const steps = [ + ['DM inbox', () => dmManager.resubscribeInbox()], + ['active channel', () => subscriptionManager.resubscribeActive()], + ['preview', async () => { + const preview = previewModeUI.isInPreviewMode() ? previewModeUI.getPreviewChannel() : null; + if (preview) await previewModeUI.enterPreviewWithoutHistory(preview.streamId, preview.channelInfo); + }], + ['owed rotations', async () => { + if (!authManager.isGuestMode()) channelManager.resumeOwedRotations(); + }], + ['inbox push', () => dmManager.subscribeInboxPush()], + ['sync', async () => { + if (syncManager.isAutoSyncAllowed('foreground')) await syncManager.smartSync(); + }] + ]; + for (const [label, step] of steps) { + try { + await step(); + } catch (error) { + Logger.warn(`Resume after client replacement (${label}) failed:`, error?.message || error); + } + } + Logger.info('Subscriptions resumed on the new Streamr client'); } /** diff --git a/src/js/dm.js b/src/js/dm.js index b46fdd6..f2b0dc0 100644 --- a/src/js/dm.js +++ b/src/js/dm.js @@ -297,6 +297,18 @@ class DMManager { } } + /** + * Subscribe the inbox again on a replaced client, whose predecessor took + * these handles with it. The open conversation's ephemeral comes back + * with the active channel. + */ + async resubscribeInbox() { + this.inboxSubscription = null; + this.inboxNotificationSub = null; + this.inboxEphemeralSubscription = null; + if (this.inboxMessageStreamId) await this.subscribeToInbox(); + } + /** * Unsubscribe from inbox (when leaving DM tab or disconnecting) */ diff --git a/src/js/streamr.js b/src/js/streamr.js index eb93cb0..c481126 100644 --- a/src/js/streamr.js +++ b/src/js/streamr.js @@ -26,6 +26,7 @@ import { CONFIG, getRpcEndpoints } from './config.js'; import { executeWithRetry, executeWithRetryAndVerify } from './utils/retry.js'; import { isRpcError, createPermissionResult } from './utils/rpcErrors.js'; import { authManager } from './auth.js'; +import { NodeRevival } from './streamr/NodeRevival.js'; import { recoverPublisherAccount, applyAccount, stripLocalFields, dropLocalState, clearPublisherProofCache } from './publisherProof.js'; @@ -121,6 +122,13 @@ class StreamrController { this.messages = new MessagePipeline(this); this._writers = new Map(); // streamId -> { public, writers:Set, ts } this._writerFetches = new Map(); // streamId -> in-flight promise + this._clientReplacedHandlers = []; + this._replacing = Promise.resolve(); + this._session = 0; + this.revival = new NodeRevival({ + networkUp: () => this._networkReachable(), + rebuild: () => this.replaceClient() + }); } /** @@ -233,8 +241,10 @@ class StreamrController { /** * Initialize Streamr client with signer * @param {Object} signer - Ethers signer from wallet (must have privateKey) + * @param {Object} [options] + * @param {boolean} [options.resubscribe] - run the onClientReplaced handlers once the node is up */ - async init(signer) { + async init(signer, { resubscribe = false } = {}) { try { // Get StreamrClient from the global window object (exposed by the // self-hosted vendor bundle — see src/streamr-bundle.js) @@ -246,7 +256,7 @@ class StreamrController { throw new Error('Signer must have a privateKey'); } - this.client = new StreamrClient({ + const client = new StreamrClient({ auth: { privateKey: signer.privateKey }, @@ -283,9 +293,11 @@ class StreamrController { rpcQuorum: 1 } }); - - this.address = await this.client.getAddress(); + this.client = client; + + this.address = await client.getAddress(); Logger.info('Streamr client initialized with address:', this.address); + this._watchNode(client, resubscribe); // Account identity for publishAs-based ACCOUNT publishes (keys // stream). The -4 stream must publish as the account — its grant is @@ -308,6 +320,91 @@ class StreamrController { } } + /** + * The SDK keeps a failed node start for the client's whole life, so the + * start is watched and a failure hands the client to the revival. + */ + _watchNode(client, resubscribe = false) { + client.getNodeId().then(() => { + if (this.client !== client) return; + this.revival.onAlive(); + if (resubscribe) this._resubscribe(client); + }, (error) => { + if (this.client !== client) return; + Logger.warn('Streamr node failed to start:', error?.message || error); + this.revival.onDead(); + }); + } + + /** One JSON-RPC round trip to any of the chosen endpoints. */ + async _networkReachable() { + const endpoints = getRpcEndpoints().slice(0, 3); + const probe = ({ url }) => fetch(url, { + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: JSON.stringify({ jsonrpc: '2.0', id: 1, method: 'eth_chainId', params: [] }), + signal: AbortSignal.timeout(5000) + }).then((response) => { + if (!response.ok) throw new Error(`HTTP ${response.status}`); + }); + try { + await Promise.any(endpoints.map(probe)); + return true; + } catch { + return false; + } + } + + /** Called after every client replacement, to subscribe again. */ + onClientReplaced(handler) { + this._clientReplacedHandlers.push(handler); + } + + /** + * Before the node is up a subscribe only waits on its start, and after a + * failed start there is nothing to subscribe on: so this runs once it is up. + */ + async _resubscribe(client) { + for (const handler of this._clientReplacedHandlers) { + if (this.client !== client) return; + try { + await handler(); + } catch (e) { + Logger.warn('Resubscribe after client replacement failed:', e?.message || e); + } + } + } + + /** + * A new SDK client for this same session: the pseudonyms and caches of the + * session stay, the subscriptions go with the old client and every + * onClientReplaced handler makes its own again once the new node is up. + */ + replaceClient() { + const run = this._replacing.catch(() => {}).then(() => this._replaceClientNow()); + this._replacing = run; + return run; + } + + async _replaceClientNow() { + const session = this._session; + const signer = authManager.getSigner(); + if (!signer) throw new Error('No signer to rebuild the Streamr client with'); + const old = this.client; + this.client = null; + this.subscriptions.clear(); + if (old) { + // A stop that hangs must not hold the new client back. + await Promise.race([ + old.destroy().catch((e) => Logger.warn('Streamr client destroy error (ignored):', e.message)), + new Promise((resolve) => setTimeout(resolve, 5000)) + ]); + } + // Logged out or switched account meanwhile: the next login makes its own client. + if (session !== this._session) return; + await this.init(signer, { resubscribe: true }); + } + /** * Pre-warm the network node for faster first subscription. * Subscribes briefly to a stream to force node startup. @@ -3939,7 +4036,14 @@ class StreamrController { * Disconnect Streamr client */ async disconnect() { - if (this.client) { + this._session++; + this.revival.stop(); + const client = this.client; + if (client) { + // Out before anything is awaited: a node start still pending settles + // during the teardown, and its watch must not see this client as current. + this.client = null; + // Unsubscribe from all streams for (const streamId of this.subscriptions.keys()) { try { @@ -3950,11 +4054,10 @@ class StreamrController { } try { - await this.client.destroy(); + await client.destroy(); } catch (e) { Logger.warn('Streamr client destroy error (ignored):', e.message); } - this.client = null; Logger.info('Streamr client disconnected'); } @@ -3973,26 +4076,12 @@ class StreamrController { /** * Reconnect Streamr client with new RPC endpoints - * Gets signer from authManager (avoids storing private key) - * Note: Active subscriptions are cleared - user needs to rejoin channels * @returns {Promise} - Success status */ async reconnect() { - const signer = authManager.getSigner(); - if (!signer) { - Logger.warn('Cannot reconnect: no signer available from authManager'); - return false; - } - try { Logger.info('Reconnecting Streamr client with new RPC endpoints...'); - - // Disconnect current client (clears subscriptions) - await this.disconnect(); - - // Re-initialize with fresh signer (will use new RPC endpoints) - await this.init(signer); - + await this.replaceClient(); Logger.info('Streamr client reconnected successfully'); return true; } catch (error) { diff --git a/src/js/streamr/NodeRevival.js b/src/js/streamr/NodeRevival.js new file mode 100644 index 0000000..b42dd1a --- /dev/null +++ b/src/js/streamr/NodeRevival.js @@ -0,0 +1,112 @@ +/** + * Rebuilds the Streamr client when its node failed to start (twin of the + * Android NodeRevival). The SDK starts the node once per client and keeps a + * failed start for the client's whole life: every publish and subscribe then + * fails at once, and only a new client starts again. + * + * One rebuild at a time, and only after a real request got through: the first + * at once, the next ones on a backoff. The browser coming back online, or the + * page back in view, skips what is left of the wait. + */ + +import { Logger } from '../logger.js'; + +export const NODE_REVIVAL_DELAYS_MS = [15000, 30000, 60000, 120000, 300000]; +export const NODE_REPORT_TIMEOUT_MS = 180000; + +export class NodeRevival { + /** + * @param {Object} host + * @param {() => Promise} host.networkUp - a real request; navigator.onLine is true behind a captive portal + * @param {() => Promise} host.rebuild - new client; its node's start comes back as onAlive/onDead + * @param {Object} [options] + * @param {number[]} [options.delaysMs] + * @param {number} [options.reportTimeoutMs] + */ + constructor(host, { delaysMs = NODE_REVIVAL_DELAYS_MS, reportTimeoutMs = NODE_REPORT_TIMEOUT_MS } = {}) { + this.host = host; + this.delaysMs = delaysMs; + this.reportTimeoutMs = reportTimeoutMs; + this.dead = false; + this.failures = 0; + this.busy = false; + this.rebuilding = false; + this.waitTimer = null; + this.watchdog = null; + } + + onDead() { + this.dead = true; + if (this.rebuilding) { + this.rebuilding = false; + this.busy = false; + this.failures++; + clearTimeout(this.watchdog); + this.watchdog = null; + } + if (!this.busy) this._schedule(); + } + + onAlive() { + this.dead = false; + this.rebuilding = false; + this.busy = false; + this.failures = 0; + clearTimeout(this.waitTimer); + this.waitTimer = null; + clearTimeout(this.watchdog); + this.watchdog = null; + } + + /** Logged out: nothing left to revive. */ + stop() { + this.onAlive(); + } + + /** Back online or back in view: no reason to wait longer. */ + kick() { + if (!this.dead || this.busy) return; + clearTimeout(this.waitTimer); + this.waitTimer = null; + this._attempt(); + } + + _schedule() { + clearTimeout(this.waitTimer); + const delay = this.failures === 0 + ? 0 + : this.delaysMs[Math.min(this.failures - 1, this.delaysMs.length - 1)]; + this.waitTimer = setTimeout(() => { + this.waitTimer = null; + this._attempt(); + }, delay); + } + + async _attempt() { + if (!this.dead || this.busy) return; + this.busy = true; + let up = false; + try { up = await this.host.networkUp(); } catch { up = false; } + if (!this.dead) { + this.busy = false; + return; + } + if (!up) { + this.busy = false; + this.failures++; + this._schedule(); + return; + } + this.rebuilding = true; + // A start that never reports would block every later rebuild. + this.watchdog = setTimeout(() => { + if (this.rebuilding) this.onDead(); + }, this.reportTimeoutMs); + try { + await this.host.rebuild(); + } catch (e) { + Logger.warn('Streamr client rebuild failed:', e?.message || e); + if (this.rebuilding) this.onDead(); + } + } +} diff --git a/src/js/subscriptionManager.js b/src/js/subscriptionManager.js index e77ed17..987ee81 100644 --- a/src/js/subscriptionManager.js +++ b/src/js/subscriptionManager.js @@ -137,6 +137,17 @@ class SubscriptionManager { this._startMemberCatchUp(messageStreamId); } + /** + * Subscribe the open channel again on a replaced client. Not through + * setActiveChannel, which returns early for the channel already active. + */ + async resubscribeActive() { + const messageStreamId = this.activeChannelId; + if (!messageStreamId) return; + const channel = channelManager.getChannel(messageStreamId); + await channelManager.subscribeToChannel(messageStreamId, channel?.password ?? null); + } + /** A member's keys and messages need a raw sweep the owner does not. */ _startMemberCatchUp(messageStreamId) { try { diff --git a/src/js/ui/SettingsUI.js b/src/js/ui/SettingsUI.js index 3a189a8..c4c3788 100644 --- a/src/js/ui/SettingsUI.js +++ b/src/js/ui/SettingsUI.js @@ -1963,7 +1963,6 @@ class SettingsUI { this.showNotification('Reconnecting with new RPC...', 'info'); const success = await this.streamrController.reconnect(); if (success) { - await this.deps.onSyncTransportReconnected?.(); this.showNotification('RPC updated successfully', 'success'); } else { this.showNotification('RPC saved. Reload page to apply.', 'warning'); diff --git a/tests/unit/nodeRevival.test.js b/tests/unit/nodeRevival.test.js new file mode 100644 index 0000000..6054da4 --- /dev/null +++ b/tests/unit/nodeRevival.test.js @@ -0,0 +1,148 @@ +import { describe, it, expect, beforeEach, afterEach, vi } from 'vitest'; + +vi.mock('../../src/js/logger.js', () => ({ Logger: { debug: vi.fn(), info: vi.fn(), warn: vi.fn(), error: vi.fn() } })); + +const { NodeRevival, NODE_REPORT_TIMEOUT_MS } = await import('../../src/js/streamr/NodeRevival.js'); + +let network, checks, rebuilds, rebuildFails, revival; + +const elapse = (ms) => vi.advanceTimersByTimeAsync(ms); + +beforeEach(() => { + vi.useFakeTimers(); + network = true; + checks = 0; + rebuilds = 0; + rebuildFails = false; + revival = new NodeRevival({ + networkUp: async () => { checks++; return network; }, + rebuild: async () => { + rebuilds++; + if (rebuildFails) throw new Error('no signer'); + } + }); +}); + +afterEach(() => vi.useRealTimers()); + +describe('NodeRevival', () => { + it('rebuilds a failed start at once when the network answers', async () => { + revival.onDead(); + await elapse(0); + + expect(checks).toBe(1); + expect(rebuilds).toBe(1); + expect(revival.rebuilding).toBe(true); + }); + + it('waits longer after each failed rebuild, up to the last step', async () => { + revival.onDead(); + await elapse(0); + for (const wait of [15000, 30000, 60000, 120000, 300000, 300000]) { + const before = rebuilds; + revival.onDead(); + await elapse(wait - 1); + expect(rebuilds).toBe(before); + await elapse(1); + expect(rebuilds).toBe(before + 1); + } + }); + + it('does not rebuild while the network check fails, and keeps checking on the backoff', async () => { + network = false; + revival.onDead(); + await elapse(0); + expect(checks).toBe(1); + + await elapse(15000); + expect(checks).toBe(2); + await elapse(29999); + expect(checks).toBe(2); + await elapse(1); + expect(checks).toBe(3); + expect(rebuilds).toBe(0); + + network = true; + revival.kick(); + await elapse(0); + + expect(rebuilds).toBe(1); + }); + + it('a kick skips what is left of the wait', async () => { + revival.onDead(); + await elapse(0); + revival.onDead(); + + revival.kick(); + await elapse(0); + + expect(rebuilds).toBe(2); + }); + + it('never runs two rebuilds at once', async () => { + revival.onDead(); + await elapse(0); + revival.kick(); + revival.kick(); + revival.onDead(); + await elapse(0); + + expect(rebuilds).toBe(1); + expect(revival.rebuilding).toBe(false); + }); + + it('a node that comes up resets the backoff and ignores kicks', async () => { + revival.onDead(); + await elapse(0); + revival.onDead(); + await elapse(15000); + revival.onAlive(); + + revival.kick(); + await elapse(0); + expect(rebuilds).toBe(2); + + revival.onDead(); + await elapse(0); + + expect(rebuilds).toBe(3); + }); + + it('a node that comes up during the network check is left alone', async () => { + let answer; + revival = new NodeRevival({ + networkUp: () => new Promise((r) => { answer = r; }), + rebuild: async () => { rebuilds++; } + }); + revival.onDead(); + await elapse(0); + revival.onAlive(); + answer(true); + await elapse(0); + + expect(rebuilds).toBe(0); + }); + + it('a rebuild that never reports counts as failed', async () => { + revival.onDead(); + await elapse(0); + await elapse(NODE_REPORT_TIMEOUT_MS); + + expect(revival.rebuilding).toBe(false); + await elapse(14999); + expect(rebuilds).toBe(1); + await elapse(1); + expect(rebuilds).toBe(2); + }); + + it('a rebuild that throws counts as failed', async () => { + rebuildFails = true; + revival.onDead(); + await elapse(0); + + expect(revival.rebuilding).toBe(false); + await elapse(15000); + expect(rebuilds).toBe(2); + }); +}); diff --git a/tests/unit/streamr.core.test.js b/tests/unit/streamr.core.test.js index 212fbb4..e928ecb 100644 --- a/tests/unit/streamr.core.test.js +++ b/tests/unit/streamr.core.test.js @@ -51,6 +51,7 @@ function createMockClient(overrides = {}) { subscribe: vi.fn().mockResolvedValue({ unsubscribe: vi.fn().mockResolvedValue(undefined) }), getStream: vi.fn().mockResolvedValue(createMockStream()), getAddress: vi.fn().mockResolvedValue('0xmyaddress'), + getNodeId: vi.fn().mockResolvedValue('node-id'), createStream: vi.fn().mockResolvedValue({ id: '0xowner/abcd1234-1' }), deleteStream: vi.fn().mockResolvedValue(undefined), setPermissions: vi.fn().mockResolvedValue(undefined), diff --git a/tests/unit/streamr.disconnect.test.js b/tests/unit/streamr.disconnect.test.js index 2e21f3d..ab8b9b2 100644 --- a/tests/unit/streamr.disconnect.test.js +++ b/tests/unit/streamr.disconnect.test.js @@ -157,6 +157,8 @@ describe('streamrController.reconnect()', () => { afterEach(() => { streamrController.client = null; streamrController.address = null; + streamrController._clientReplacedHandlers = []; + vi.useRealTimers(); vi.clearAllMocks(); }); @@ -179,24 +181,130 @@ describe('streamrController.reconnect()', () => { streamrController.init = originalInit; }); - it('should call disconnect before reinitializing', async () => { - // Mock init to avoid actual StreamrClient creation + it('destroys the old client before making the new one, and drops its subscriptions', async () => { + const originalInit = streamrController.init.bind(streamrController); + const order = []; + mockClient.destroy.mockImplementation(async () => { order.push('destroy'); }); + streamrController.init = vi.fn().mockImplementation(async () => { order.push('init'); return true; }); + streamrController.subscriptions.set('stream-a', { 0: { unsubscribe: vi.fn() } }); + + await streamrController.reconnect(); + + expect(order).toEqual(['destroy', 'init']); + expect(streamrController.init).toHaveBeenCalledWith(mockSigner, { resubscribe: true }); + expect(streamrController.subscriptions.size).toBe(0); + + streamrController.init = originalInit; + }); + + it('stays in the session: no logout, so the pseudonyms are kept', async () => { const originalInit = streamrController.init.bind(streamrController); streamrController.init = vi.fn().mockResolvedValue(true); - - // Spy on disconnect const disconnectSpy = vi.spyOn(streamrController, 'disconnect'); await streamrController.reconnect(); - expect(disconnectSpy).toHaveBeenCalled(); - expect(streamrController.init).toHaveBeenCalledWith(mockSigner); + expect(disconnectSpy).not.toHaveBeenCalled(); - // Restore streamrController.init = originalInit; disconnectSpy.mockRestore(); }); + // Stands in for init: installs a client whose node start is `start` and watches it. + const initWithNode = (start) => vi.fn().mockImplementation(async (_signer, options) => { + const client = { getNodeId: vi.fn(start), destroy: vi.fn().mockResolvedValue(undefined) }; + streamrController.client = client; + streamrController._watchNode(client, options?.resubscribe); + return true; + }); + + const quietRevival = () => [ + vi.spyOn(streamrController.revival, 'onAlive').mockImplementation(() => {}), + vi.spyOn(streamrController.revival, 'onDead').mockImplementation(() => {}) + ]; + + it('asks every handler to subscribe again once the new node is up, even after one fails', async () => { + const originalInit = streamrController.init.bind(streamrController); + const spies = quietRevival(); + streamrController.init = initWithNode(() => Promise.resolve('node-id')); + const later = vi.fn(); + streamrController.onClientReplaced(async () => { throw new Error('inbox'); }); + streamrController.onClientReplaced(later); + + const result = await streamrController.reconnect(); + + expect(result).toBe(true); + await vi.waitFor(() => expect(later).toHaveBeenCalledTimes(1)); + + streamrController.init = originalInit; + spies.forEach((spy) => spy.mockRestore()); + }); + + it('leaves the subscribing to the next rebuild when the new node fails to start', async () => { + const originalInit = streamrController.init.bind(streamrController); + const spies = quietRevival(); + streamrController.init = initWithNode(() => Promise.reject(new Error('Failed to connect to the entrypoints after 7 attempts'))); + const handler = vi.fn(); + streamrController.onClientReplaced(handler); + + await streamrController.reconnect(); + await new Promise((resolve) => setTimeout(resolve, 0)); + + expect(handler).not.toHaveBeenCalled(); + + streamrController.init = originalInit; + spies.forEach((spy) => spy.mockRestore()); + }); + + it('a node start that never settles does not hold the next replacement back', async () => { + const originalInit = streamrController.init.bind(streamrController); + const spies = quietRevival(); + streamrController.init = initWithNode(() => new Promise(() => {})); + + expect(await streamrController.reconnect()).toBe(true); + expect(await streamrController.reconnect()).toBe(true); + + expect(streamrController.init).toHaveBeenCalledTimes(2); + + streamrController.init = originalInit; + spies.forEach((spy) => spy.mockRestore()); + }); + + it('gives up when the account changes while the old client stops', async () => { + const originalInit = streamrController.init.bind(streamrController); + streamrController.init = vi.fn().mockResolvedValue(true); + const handler = vi.fn(); + streamrController.onClientReplaced(handler); + let stopped; + mockClient.destroy.mockReturnValue(new Promise((resolve) => { stopped = resolve; })); + + const done = streamrController.reconnect(); + await vi.waitFor(() => expect(mockClient.destroy).toHaveBeenCalled()); + await streamrController.disconnect(); + stopped(); + await done; + + expect(streamrController.init).not.toHaveBeenCalled(); + expect(handler).not.toHaveBeenCalled(); + + streamrController.init = originalInit; + }); + + it('does not wait on an old client that never finishes stopping', async () => { + vi.useFakeTimers(); + const originalInit = streamrController.init.bind(streamrController); + streamrController.init = vi.fn().mockResolvedValue(true); + mockClient.destroy.mockReturnValue(new Promise(() => {})); + + const done = streamrController.reconnect(); + await vi.advanceTimersByTimeAsync(5000); + + expect(await done).toBe(true); + expect(streamrController.init).toHaveBeenCalledTimes(1); + + streamrController.init = originalInit; + }); + it('should return true on successful reconnect', async () => { const originalInit = streamrController.init.bind(streamrController); streamrController.init = vi.fn().mockResolvedValue(true); diff --git a/tests/unit/streamr.nodeWatch.test.js b/tests/unit/streamr.nodeWatch.test.js new file mode 100644 index 0000000..2a904a0 --- /dev/null +++ b/tests/unit/streamr.nodeWatch.test.js @@ -0,0 +1,78 @@ +import { describe, it, expect, beforeEach, afterEach, vi } from 'vitest'; + +vi.mock('../../src/js/auth.js', () => ({ + authManager: { getSigner: vi.fn() } +})); + +import { streamrController } from '../../src/js/streamr.js'; + +const flush = () => new Promise((r) => setTimeout(r, 0)); + +function clientWithStart(start) { + return function () { + this.getAddress = vi.fn().mockResolvedValue('0xme'); + this.getNodeId = vi.fn(start); + }; +} + +describe('streamrController node start watch', () => { + let onDead, onAlive; + + beforeEach(() => { + onDead = vi.spyOn(streamrController.revival, 'onDead').mockImplementation(() => {}); + onAlive = vi.spyOn(streamrController.revival, 'onAlive').mockImplementation(() => {}); + }); + + afterEach(() => { + streamrController.client = null; + vi.restoreAllMocks(); + }); + + it('hands a client whose node failed to start to the revival', async () => { + window.StreamrClient = clientWithStart(() => Promise.reject(new Error('Failed to connect to the entrypoints after 7 attempts'))); + + await streamrController.init({ privateKey: '0xabc' }); + await flush(); + + expect(onDead).toHaveBeenCalledTimes(1); + expect(onAlive).not.toHaveBeenCalled(); + }); + + it('reports a node that came up', async () => { + window.StreamrClient = clientWithStart(() => Promise.resolve('node-id')); + + await streamrController.init({ privateKey: '0xabc' }); + await flush(); + + expect(onAlive).toHaveBeenCalledTimes(1); + expect(onDead).not.toHaveBeenCalled(); + }); + + it('a start that fails while logging out does not wake the revival', async () => { + let fail; + window.StreamrClient = clientWithStart(() => new Promise((_, reject) => { fail = reject; })); + await streamrController.init({ privateKey: '0xabc' }); + let destroyed; + streamrController.client.destroy = vi.fn(() => new Promise((resolve) => { destroyed = resolve; })); + + const leaving = streamrController.disconnect(); + fail(new Error('Failed to connect to the entrypoints after 7 attempts')); + await flush(); + destroyed(); + await leaving; + + expect(onDead).not.toHaveBeenCalled(); + }); + + it('ignores the start of a client that was already replaced', async () => { + let fail; + window.StreamrClient = clientWithStart(() => new Promise((_, reject) => { fail = reject; })); + + await streamrController.init({ privateKey: '0xabc' }); + streamrController.client = { replacement: true }; + fail(new Error('connectionCount>0')); + await flush(); + + expect(onDead).not.toHaveBeenCalled(); + }); +}); From 6ab44030244617cf13e6686ee6afd8cca25f05ab Mon Sep 17 00:00:00 2001 From: Ocnrb Date: Thu, 24 Sep 2026 19:09:47 +0100 Subject: [PATCH 2/2] =?UTF-8?q?Show=20a=20red=20dot=20on=20the=20logo=20an?= =?UTF-8?q?d=20Connecting=E2=80=A6=20in=20the=20chat=20header=20while=20th?= =?UTF-8?q?e=20network=20node=20is=20down?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-authored-by: Claude Opus 5.5 --- index.html | 8 ++++++-- src/js/app.js | 6 ++++++ src/js/streamr.js | 8 ++++++++ src/js/ui/HeaderUI.js | 5 +++++ src/styles/components.css | 33 +++++++++++++++++++++++++++++++++ 5 files changed, 58 insertions(+), 2 deletions(-) diff --git a/index.html b/index.html index 4459ed9..b468365 100644 --- a/index.html +++ b/index.html @@ -80,7 +80,10 @@
- + + + +
Pombo @@ -453,7 +456,8 @@

Explore Cha