From 138e0742a718960a9632a769d2dd07799290461a Mon Sep 17 00:00:00 2001 From: Ocnrb Date: Wed, 30 Sep 2026 16:10:35 +0100 Subject: [PATCH] Confirm on chain before sending a write again when its receipt read failed - A retry now asks the chain first and skips what already landed: stream creation, storage assignment, retention, grants - The check comes after the backoff wait, so a write still pending when its receipt read failed is seen - A chain that cannot be read waits for the next round instead of sending blind - Only retries within one attempt sequence change; a fresh call (repair, wizard) still sends first Co-authored-by: Claude Opus 5.5 --- src/js/streamr.js | 163 +++++++++++++++++++---------- src/js/utils/retry.js | 63 ++++++----- tests/unit/retry.test.js | 82 ++++++++++++--- tests/unit/streamr.core.test.js | 16 +-- tests/unit/streamr.txRetry.test.js | 105 +++++++++++++++++++ 5 files changed, 325 insertions(+), 104 deletions(-) create mode 100644 tests/unit/streamr.txRetry.test.js diff --git a/src/js/streamr.js b/src/js/streamr.js index b9c5bab..e782732 100644 --- a/src/js/streamr.js +++ b/src/js/streamr.js @@ -56,6 +56,11 @@ import { /** A publish grant added after a stream was cached takes this long to be seen. */ const WRITER_CACHE_TTL_MS = 60_000; +/** Streamr's StreamRegistry on Polygon, read directly past the SDK's metadata cache. */ +const STREAM_REGISTRY_ADDRESS = '0x0D483E10612F327FC11965Fc82E90dC19b141641'; +const STREAM_REGISTRY_ABI = ['function getStreamMetadata(string streamId) view returns (string)']; +const REGISTRY_READ_TIMEOUT_MS = 10_000; + // === STREAM CONFIG (DUAL-STREAM ARCHITECTURE) === import { STREAM_CONFIG } from './streamConfig.js'; import { History } from './streamr/History.js'; @@ -595,13 +600,9 @@ class StreamrController { return stream; }, async () => { - // Check if stream was actually created despite error - const existingStream = await this.client.getStream(streamId); - if (existingStream) { - Logger.info(`✓ Stream exists (created despite error): ${existingStream.id}`); - return existingStream; - } - return null; + const existingStream = await this._streamIfExists(streamId); + if (existingStream) Logger.info(`✓ Stream exists (created despite error): ${existingStream.id}`); + return existingStream; }, { maxRetries } ); @@ -978,13 +979,72 @@ class StreamrController { return; } - await executeWithRetry('setStreamPermissions', async () => { + await executeWithRetryAndVerify('setStreamPermissions', async () => { await this.client.setPermissions({ streamId: streamId, assignments: assignments }); Logger.debug('Permissions set successfully (batch)'); - }, { maxRetries: retries }); + return true; + }, () => this._permissionsInPlace(streamId, assignments), { maxRetries: retries }); + } + + /** + * Does the chain already hold exactly these assignments? setPermissions + * replaces a grantee's set, so a grant is in place only when every + * permission matches, the ones it leaves out included. Reads the + * registry directly: throws when the chain cannot be read. + */ + async _permissionsInPlace(streamId, assignments) { + const all = ['publish', 'subscribe', 'edit', 'delete', 'grant']; + for (const assignment of assignments) { + const wanted = new Set(assignment.permissions); + for (const permission of all) { + const held = await this.client.hasPermission(assignment.public + ? { streamId, public: true, permission } + : { streamId, userId: assignment.userId, permission, allowPublic: false }); + if (held !== wanted.has(permission)) return false; + } + } + return true; + } + + /** + * The retention the chain holds now. The SDK caches stream metadata and + * keeps the old value after a write whose receipt read failed, so this + * reads the registry itself. Throws when no endpoint answers. + * @returns {Promise} null when the stream sets none + */ + async _chainStorageDays(streamId) { + let lastError = new Error('No RPC endpoint'); + for (const { url } of getRpcEndpoints()) { + try { + const provider = new ethers.JsonRpcProvider(url, CONFIG.network.chainId, + { staticNetwork: true, batchMaxCount: 1 }); + const registry = new ethers.Contract(STREAM_REGISTRY_ADDRESS, STREAM_REGISTRY_ABI, provider); + const raw = await Promise.race([ + registry.getStreamMetadata(streamId), + new Promise((_, reject) => setTimeout( + () => reject(new Error('registry read timed out')), REGISTRY_READ_TIMEOUT_MS)) + ]); + let metadata = {}; + try { metadata = raw ? JSON.parse(raw) : {}; } catch { /* the chain answered: no retention set */ } + return Number.isFinite(metadata.storageDays) ? metadata.storageDays : null; + } catch (e) { + lastError = e; + } + } + throw lastError; + } + + /** The stream, or null when the chain says it does not exist; throws when the chain cannot be read. */ + async _streamIfExists(streamId) { + try { + return await this.client.getStream(streamId); + } catch (e) { + if (e?.code === 'STREAM_NOT_FOUND') return null; + throw e; + } } /** @@ -996,16 +1056,12 @@ class StreamrController { // If string passed, get the stream object const streamId = typeof stream === 'string' ? stream : stream.id; - await executeWithRetry('grantPublicPermissions', async () => { - await this.client.setPermissions({ - streamId: streamId, - assignments: [{ - public: true, - permissions: ['subscribe', 'publish'] - }] - }); + const assignments = [{ public: true, permissions: ['subscribe', 'publish'] }]; + await executeWithRetryAndVerify('grantPublicPermissions', async () => { + await this.client.setPermissions({ streamId, assignments }); Logger.debug('Public permissions granted successfully'); - }, { maxRetries: retries }); + return true; + }, () => this._permissionsInPlace(streamId, assignments), { maxRetries: retries }); } /** @@ -1015,16 +1071,12 @@ class StreamrController { async grantPublicReadOnlyPermissions(stream, retries = 7) { const streamId = typeof stream === 'string' ? stream : stream.id; - await executeWithRetry('grantPublicReadOnlyPermissions', async () => { - await this.client.setPermissions({ - streamId: streamId, - assignments: [{ - public: true, - permissions: ['subscribe'] - }] - }); + const assignments = [{ public: true, permissions: ['subscribe'] }]; + await executeWithRetryAndVerify('grantPublicReadOnlyPermissions', async () => { + await this.client.setPermissions({ streamId, assignments }); Logger.debug('Public read-only permissions granted successfully'); - }, { maxRetries: retries }); + return true; + }, () => this._permissionsInPlace(streamId, assignments), { maxRetries: retries }); } /** @@ -1035,16 +1087,13 @@ class StreamrController { async grantManyToOnePermissions(stream, retries = 7) { const streamId = typeof stream === 'string' ? stream : stream.id; - await executeWithRetry('grantManyToOnePermissions', async () => { - await this.client.setPermissions({ - streamId: streamId, - assignments: [{ - public: true, - permissions: ['publish'] // Only PUBLISH, not SUBSCRIBE - }] - }); + // Only PUBLISH, not SUBSCRIBE + const assignments = [{ public: true, permissions: ['publish'] }]; + await executeWithRetryAndVerify('grantManyToOnePermissions', async () => { + await this.client.setPermissions({ streamId, assignments }); Logger.debug('Many-to-one permissions granted (public publish, owner-only subscribe)'); - }, { maxRetries: retries }); + return true; + }, () => this._permissionsInPlace(streamId, assignments), { maxRetries: retries }); } /** @@ -1171,10 +1220,7 @@ class StreamrController { }); return stream; }, - async () => { - const s = await this.client.getStream(streamId); - return s || null; - }, + () => this._streamIfExists(streamId), { maxRetries: 7 } ); } @@ -1395,10 +1441,7 @@ class StreamrController { description: metadata, partitions: STREAM_CONFIG.MESSAGE_STREAM.DM_PARTITIONS }), - async () => { - const s = await this.client.getStream(messageStreamId); - return s || null; - }, + () => this._streamIfExists(messageStreamId), { maxRetries: 7 } ); result.messageStream = 'ok'; @@ -1442,10 +1485,7 @@ class StreamrController { description: ephemeralMetadata, partitions: STREAM_CONFIG.EPHEMERAL_STREAM.PARTITIONS }), - async () => { - const s = await this.client.getStream(ephemeralStreamId); - return s || null; - }, + () => this._streamIfExists(ephemeralStreamId), { maxRetries: 7 } ); result.ephemeralStream = 'ok'; @@ -2812,9 +2852,10 @@ class StreamrController { { userId: newAddress.toLowerCase(), permissions: ['publish'] }, ...(oldAddress ? [{ userId: oldAddress.toLowerCase(), permissions: [] }] : []) ]; - await executeWithRetry(`rekeySharedPublishGrants(${streamId.slice(-20)})`, async () => { + await executeWithRetryAndVerify(`rekeySharedPublishGrants(${streamId.slice(-20)})`, async () => { await this.client.setPermissions({ streamId, assignments }); - }); + return true; + }, () => this._permissionsInPlace(streamId, assignments)); } } @@ -2831,8 +2872,14 @@ class StreamrController { channel.interactionsStreamId || deriveInteractionsId(channel.messageStreamId), channel.ephemeralStreamId ].filter(Boolean).map((streamId) => ({ streamId, assignments })); - await executeWithRetry('rekeyInteractionsGrants', async () => { + await executeWithRetryAndVerify('rekeyInteractionsGrants', async () => { await this.client.setPermissions(...items); + return true; + }, async () => { + for (const { streamId } of items) { + if (!await this._permissionsInPlace(streamId, assignments)) return false; + } + return true; }); } @@ -4294,12 +4341,13 @@ class StreamrController { }); try { - await executeWithRetry('enableStorage', async () => { + await executeWithRetryAndVerify('enableStorage', async () => { const stream = await this.client.getStream(messageStreamId); await stream.addToStorageNode(nodeAddress); - try { options.onProgress?.(); } catch (_) { /* progress callback errors must not break flow */ } - Logger.info('Storage enabled:', { stream: messageStreamId, provider: providerId }); - }, { maxRetries: retries }); + return true; + }, () => this.client.isStoredStream(messageStreamId, nodeAddress), { maxRetries: retries }); + try { options.onProgress?.(); } catch (_) { /* progress callback errors must not break flow */ } + Logger.info('Storage enabled:', { stream: messageStreamId, provider: providerId }); } catch (error) { Logger.error('All storage attempts failed for:', messageStreamId); return { success: false, provider: providerId, storageDays: null, retentionApplied: false }; @@ -4311,10 +4359,11 @@ class StreamrController { let retentionApplied = false; if (storageDays && providerConfig.supportsTTL) { try { - await executeWithRetry('setStorageDayCount', async () => { + await executeWithRetryAndVerify('setStorageDayCount', async () => { const stream = await this.client.getStream(messageStreamId); await stream.setStorageDayCount(storageDays); - }, { maxRetries: retries }); + return true; + }, async () => (await this._chainStorageDays(messageStreamId)) === storageDays, { maxRetries: retries }); retentionApplied = true; Logger.debug('Storage days set to:', storageDays); } catch (ttlError) { diff --git a/src/js/utils/retry.js b/src/js/utils/retry.js index c556d07..5f253e6 100644 --- a/src/js/utils/retry.js +++ b/src/js/utils/retry.js @@ -75,12 +75,17 @@ export async function executeWithRetry(operationName, asyncFn, options = {}) { } /** - * Execute with retry, checking if resource exists after error - * Useful for blockchain operations where tx may succeed despite error response - * + * Execute an on-chain write with retry, never sending it twice. + * + * A write whose receipt read failed has usually landed anyway, so each retry + * first asks the chain: `checkExistsFn` resolves truthy when the effect is + * there (returned as the result), falsy when it is missing (send again), and + * throws when the chain could not be read — which waits for the next round + * rather than sending blind. + * * @param {string} operationName - Name for logging - * @param {Function} asyncFn - Async function to execute - * @param {Function} checkExistsFn - Function to check if resource was created despite error + * @param {Function} asyncFn - Async function that sends the write + * @param {Function} checkExistsFn - Is the write's effect already on chain? * @param {Object} options - Retry options (same as executeWithRetry) * @returns {Promise} - Result of asyncFn or checkExistsFn */ @@ -88,36 +93,40 @@ export async function executeWithRetryAndVerify(operationName, asyncFn, checkExi const { maxRetries = CONFIG.retry.maxAttempts, baseDelay = CONFIG.retry.baseDelayMs, - backoffMultiplier = CONFIG.retry.backoffMultiplier + backoffMultiplier = CONFIG.retry.backoffMultiplier, + shouldRetry = () => true } = options; let lastError; + let send = true; - for (let attempt = 1; attempt <= maxRetries; attempt++) { - try { - Logger.debug(`${operationName} (attempt ${attempt}/${maxRetries})...`); - return await asyncFn(); - - } catch (error) { - lastError = error; - Logger.warn(`${operationName} attempt ${attempt} failed:`, error.message); - - // Check if operation actually succeeded despite error + for (let round = 1; round <= maxRetries; round++) { + if (send) { try { - const existing = await checkExistsFn(); - if (existing) { - Logger.info(`${operationName}: resource exists despite error`); - return existing; - } - } catch (checkError) { - // Resource doesn't exist, continue retry + Logger.debug(`${operationName} (attempt ${round}/${maxRetries})...`); + return await asyncFn(); + } catch (error) { + lastError = error; + Logger.warn(`${operationName} attempt ${round} failed:`, error.message); + if (!shouldRetry(error)) throw error; } + } + if (round === maxRetries) break; - if (attempt < maxRetries) { - const delay = baseDelay * Math.pow(backoffMultiplier, attempt - 1); - Logger.debug(`Retrying ${operationName} in ${delay / 1000}s...`); - await new Promise(resolve => setTimeout(resolve, delay)); + // The wait comes first: a write still pending when its receipt read + // failed gets mined meanwhile, and the check then sees it. + const delay = baseDelay * Math.pow(backoffMultiplier, round - 1); + await new Promise(resolve => setTimeout(resolve, delay)); + try { + const existing = await checkExistsFn(); + if (existing) { + Logger.info(`${operationName}: already on chain`); + return existing; } + send = true; + } catch (checkError) { + Logger.warn(`${operationName}: could not confirm on chain, not resending yet:`, checkError.message); + send = false; } } diff --git a/tests/unit/retry.test.js b/tests/unit/retry.test.js index 7a9bc83..6902d36 100644 --- a/tests/unit/retry.test.js +++ b/tests/unit/retry.test.js @@ -153,31 +153,85 @@ describe('retry', () => { expect(asyncFn).toHaveBeenCalledTimes(2); }); - it('should continue retrying if checkExists throws', async () => { + it('never sends a write again that landed while its receipt read failed', async () => { + const asyncFn = vi.fn().mockRejectedValue(new Error('Error while waiting transaction')); + const checkExistsFn = vi.fn().mockResolvedValue(true); + + const result = await executeWithRetryAndVerify('test', asyncFn, checkExistsFn, { + maxRetries: 5, + baseDelay: 10 + }); + + expect(result).toBe(true); + expect(asyncFn).toHaveBeenCalledTimes(1); + }); + + it('checks only after the wait, when a pending write has had time to land', async () => { + const order = []; + const asyncFn = vi.fn(async () => { order.push('send'); throw new Error('receipt read failed'); }); + const checkExistsFn = vi.fn(async () => { order.push('check'); return true; }); + + await executeWithRetryAndVerify('test', asyncFn, checkExistsFn, { maxRetries: 3, baseDelay: 10 }); + + expect(order).toEqual(['send', 'check']); + }); + + it('an unreadable chain sends nothing until the check answers, then sends only if missing', async () => { const asyncFn = vi.fn() - .mockRejectedValueOnce(new Error('fail')) + .mockRejectedValueOnce(new Error('receipt read failed')) .mockResolvedValue('success'); - const checkExistsFn = vi.fn().mockRejectedValue(new Error('check failed')); - - const result = await executeWithRetryAndVerify('test', asyncFn, checkExistsFn, { - maxRetries: 3, - baseDelay: 10 + const checkExistsFn = vi.fn() + .mockRejectedValueOnce(new Error('RPC down')) + .mockRejectedValueOnce(new Error('RPC down')) + .mockResolvedValue(false); + + const result = await executeWithRetryAndVerify('test', asyncFn, checkExistsFn, { + maxRetries: 6, + baseDelay: 10 }); - + expect(result).toBe('success'); + expect(checkExistsFn).toHaveBeenCalledTimes(3); + expect(asyncFn).toHaveBeenCalledTimes(2); + }); + + it('an unreadable chain to the end sends nothing more and fails with the write error', async () => { + const asyncFn = vi.fn().mockRejectedValue(new Error('receipt read failed')); + const checkExistsFn = vi.fn().mockRejectedValue(new Error('RPC down')); + + await expect(executeWithRetryAndVerify('test', asyncFn, checkExistsFn, { + maxRetries: 4, + baseDelay: 10 + })).rejects.toThrow('receipt read failed'); + + expect(asyncFn).toHaveBeenCalledTimes(1); + expect(checkExistsFn).toHaveBeenCalledTimes(3); }); it('should throw after all retries if resource never exists', async () => { const asyncFn = vi.fn().mockRejectedValue(new Error('persistent error')); const checkExistsFn = vi.fn().mockResolvedValue(null); - - await expect(executeWithRetryAndVerify('test', asyncFn, checkExistsFn, { - maxRetries: 2, - baseDelay: 10 + + await expect(executeWithRetryAndVerify('test', asyncFn, checkExistsFn, { + maxRetries: 2, + baseDelay: 10 })).rejects.toThrow('persistent error'); - + expect(asyncFn).toHaveBeenCalledTimes(2); - expect(checkExistsFn).toHaveBeenCalledTimes(2); + expect(checkExistsFn).toHaveBeenCalledTimes(1); + }); + + it('does not retry an error the caller rules out', async () => { + const asyncFn = vi.fn().mockRejectedValue(new Error('insufficient funds')); + const checkExistsFn = vi.fn(); + + await expect(executeWithRetryAndVerify('test', asyncFn, checkExistsFn, { + maxRetries: 3, + baseDelay: 10, + shouldRetry: () => false + })).rejects.toThrow('insufficient funds'); + + expect(checkExistsFn).not.toHaveBeenCalled(); }); }); diff --git a/tests/unit/streamr.core.test.js b/tests/unit/streamr.core.test.js index d0834e7..c72ca8b 100644 --- a/tests/unit/streamr.core.test.js +++ b/tests/unit/streamr.core.test.js @@ -954,7 +954,7 @@ describe('StreamrController Core', () => { }); it('should return failure on error', async () => { - executeWithRetry.mockRejectedValueOnce(new Error('fail')); + executeWithRetryAndVerify.mockRejectedValueOnce(new Error('fail')); const result = await streamrController.enableStorage('stream-1'); expect(result.success).toBe(false); }); @@ -985,13 +985,13 @@ describe('StreamrController Core', () => { it('retries the retention independently of the node assignment', async () => { await streamrController.enableStorage('stream-1', { storageDays: 30 }); - const named = executeWithRetry.mock.calls.map(([name]) => name); + const named = executeWithRetryAndVerify.mock.calls.map(([name]) => name); expect(named).toContain('enableStorage'); expect(named).toContain('setStorageDayCount'); }); it('reports no retention when the node assignment itself failed', async () => { - executeWithRetry.mockRejectedValueOnce(new Error('fail')); + executeWithRetryAndVerify.mockRejectedValueOnce(new Error('fail')); const result = await streamrController.enableStorage('stream-1', { storageDays: 30 }); expect(result.storageDays).toBeNull(); expect(result.retentionApplied).toBe(false); @@ -1687,6 +1687,7 @@ describe('StreamrController Core', () => { const mockEph = { id: '0xmyaddress/abcd1234-2' }; let callCount = 0; executeWithRetryAndVerify.mockImplementation(async (name, fn) => { + if (!name.startsWith('createStream')) return fn(); callCount++; if (callCount === 1) return mockMsg; return mockEph; @@ -1702,6 +1703,7 @@ describe('StreamrController Core', () => { const mockEph = { id: '0xmyaddress/abcd1234-2' }; let callCount = 0; executeWithRetryAndVerify.mockImplementation(async (name, fn) => { + if (!name.startsWith('createStream')) return fn(); callCount++; if (callCount === 1) return mockMsg; return mockEph; @@ -1740,7 +1742,8 @@ describe('StreamrController Core', () => { it('should set many-to-one permissions', async () => { mockClient.getStream.mockRejectedValue(new Error('not found')); let count = 0; - executeWithRetryAndVerify.mockImplementation(async () => { + executeWithRetryAndVerify.mockImplementation(async (name, fn) => { + if (!name.startsWith('createDMStream')) return fn(); count++; return { id: count === 1 ? '0xmyaddress/Pombo-DM-1' : '0xmyaddress/Pombo-DM-2' }; }); @@ -1759,8 +1762,9 @@ describe('StreamrController Core', () => { const result = await streamrController.createDMInbox('pk123'); expect(result.messageStreamId).toBeDefined(); - // Should NOT have called executeWithRetryAndVerify (no creation needed) - expect(executeWithRetryAndVerify).not.toHaveBeenCalled(); + // No creation needed + expect(executeWithRetryAndVerify.mock.calls.filter(([name]) => name.startsWith('createDMStream'))) + .toHaveLength(0); }); }); diff --git a/tests/unit/streamr.txRetry.test.js b/tests/unit/streamr.txRetry.test.js new file mode 100644 index 0000000..c7b7d86 --- /dev/null +++ b/tests/unit/streamr.txRetry.test.js @@ -0,0 +1,105 @@ +/** + * A write whose receipt read failed has usually landed. The retries ask the + * chain before sending again, so a flaky RPC during channel creation stops + * paying for the same storage assignment, retention or grant twice. + */ + +import { describe, it, expect, beforeEach, afterEach, vi } from 'vitest'; + +vi.mock('../../src/js/auth.js', () => ({ authManager: { getSigner: vi.fn() } })); +vi.mock('../../src/js/crypto.js', () => ({ cryptoManager: {} })); + +const { streamrController } = await import('../../src/js/streamr.js'); + +const STREAM = '0xowner/room-1'; +const NODE = '0xae340e799e8151f6a4999d245e466197aa217667'; +const receiptFailed = () => new Error('Error while waiting transaction for contract call, code=SERVER_ERROR'); + +/** Runs a write through its retry waits without sleeping. */ +async function settled(promise) { + const outcome = promise.then((value) => ({ value }), (error) => ({ error })); + await vi.runAllTimersAsync(); + const { value, error } = await outcome; + if (error) throw error; + return value; +} + +describe('on-chain writes retried against the chain', () => { + let client; + let stream; + + afterEach(() => vi.useRealTimers()); + + beforeEach(() => { + vi.useFakeTimers(); + stream = { + id: STREAM, + addToStorageNode: vi.fn().mockRejectedValue(receiptFailed()), + setStorageDayCount: vi.fn().mockResolvedValue(undefined) + }; + client = { + getStream: vi.fn().mockResolvedValue(stream), + isStoredStream: vi.fn().mockResolvedValue(true), + setPermissions: vi.fn(), + hasPermission: vi.fn() + }; + streamrController.client = client; + }); + + it('does not assign the storage node twice when the first assignment landed', async () => { + const result = await settled(streamrController.enableStorage(STREAM, {}, 4)); + + expect(result.success).toBe(true); + expect(stream.addToStorageNode).toHaveBeenCalledTimes(1); + expect(client.isStoredStream).toHaveBeenCalledWith(STREAM, NODE); + }); + + it('does not set the retention twice when the first write landed', async () => { + stream.addToStorageNode.mockResolvedValue(undefined); + stream.setStorageDayCount.mockRejectedValue(receiptFailed()); + const chainDays = vi.spyOn(streamrController, '_chainStorageDays').mockResolvedValue(30); + + const result = await settled(streamrController.enableStorage(STREAM, { storageDays: 30 }, 4)); + + expect(result.retentionApplied).toBe(true); + expect(stream.setStorageDayCount).toHaveBeenCalledTimes(1); + chainDays.mockRestore(); + }); + + it('does not grant twice when the first grant landed', async () => { + client.setPermissions.mockRejectedValue(receiptFailed()); + client.hasPermission.mockImplementation(async ({ permission }) => permission === 'subscribe'); + + await settled(streamrController.grantPublicReadOnlyPermissions(STREAM, 4)); + + expect(client.setPermissions).toHaveBeenCalledTimes(1); + }); + + it('grants again when the chain shows a different set than asked', async () => { + client.setPermissions.mockRejectedValueOnce(receiptFailed()).mockResolvedValue(undefined); + // Holds publish too: the grant asked for subscribe only, so it did not land. + client.hasPermission.mockImplementation(async ({ permission }) => + permission === 'subscribe' || permission === 'publish'); + + await settled(streamrController.grantPublicReadOnlyPermissions(STREAM, 4)); + + expect(client.setPermissions).toHaveBeenCalledTimes(2); + }); + + it('does not grant again while the chain cannot be read', async () => { + client.setPermissions.mockRejectedValue(receiptFailed()); + client.hasPermission.mockRejectedValue(new Error('RPC down')); + + await expect(settled(streamrController.grantPublicReadOnlyPermissions(STREAM, 3))).rejects.toThrow('SERVER_ERROR'); + + expect(client.setPermissions).toHaveBeenCalledTimes(1); + }); + + it('tells a missing stream from an unreadable chain', async () => { + client.getStream.mockRejectedValueOnce(Object.assign(new Error('Stream not found'), { code: 'STREAM_NOT_FOUND' })); + await expect(streamrController._streamIfExists(STREAM)).resolves.toBeNull(); + + client.getStream.mockRejectedValueOnce(new Error('RPC down')); + await expect(streamrController._streamIfExists(STREAM)).rejects.toThrow('RPC down'); + }); +});