Skip to content

Confirm on chain before sending a write again when its receipt read failed #103

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Ocnrb merged 1 commit into from
Oct 5, 2026
Merged

Confirm on chain before sending a write again when its receipt read failed #103

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
163 changes: 106 additions & 57 deletions src/js/streamr.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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';
Expand Down Expand Up @@ -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 }
);
Expand Down Expand Up @@ -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<number|null>} 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;
}
}

/**
Expand All @@ -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 });
}

/**
Expand All @@ -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 });
}

/**
Expand All @@ -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 });
}

/**
Expand Down Expand Up @@ -1171,10 +1220,7 @@ class StreamrController {
});
return stream;
},
async () => {
const s = await this.client.getStream(streamId);
return s || null;
},
() => this._streamIfExists(streamId),
{ maxRetries: 7 }
);
}
Expand Down Expand Up @@ -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';
Expand Down Expand Up @@ -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';
Expand Down Expand Up @@ -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));
}
}

Expand All @@ -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;
});
}

Expand Down Expand Up @@ -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 };
Expand All @@ -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) {
Expand Down
63 changes: 36 additions & 27 deletions src/js/utils/retry.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -75,49 +75,58 @@ 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<any>} - Result of asyncFn or checkExistsFn
*/
export async function executeWithRetryAndVerify(operationName, asyncFn, checkExistsFn, options = {}) {
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;
}
}

Expand Down
Loading
Loading