diff --git a/js/nostr.js b/js/nostr.js index 0be0469f..2b6511f7 100644 --- a/js/nostr.js +++ b/js/nostr.js @@ -35,7 +35,6 @@ import { publishEventToRelay, publishEventToRelays, assertAnyRelayAccepted, - summarizePublishResults, } from "./nostrPublish.js"; /** @@ -2346,6 +2345,54 @@ class NostrClient { const chunkResults = []; const chunkAddresses = []; let anyChunkRejected = false; + let anyChunkPartial = false; + + const formatRelayStatus = (results = []) => { + const normalized = Array.isArray(results) ? results : []; + const statuses = []; + const byUrl = new Map(); + + for (const entry of normalized) { + const url = typeof entry?.url === "string" ? entry.url : ""; + if (!url) { + continue; + } + const reasonValue = (() => { + const error = entry?.error; + if (!error) { + return null; + } + if (error instanceof Error) { + return error.message || "publish failed"; + } + if (typeof error === "string" && error.trim()) { + return error.trim(); + } + try { + return JSON.stringify(error); + } catch (_) { + return String(error); + } + })(); + + byUrl.set(url, { + url, + success: !!entry?.success, + reason: reasonValue, + }); + } + + for (const relayUrl of relays) { + const existing = byUrl.get(relayUrl); + if (existing) { + statuses.push(existing); + } else { + statuses.push({ url: relayUrl, success: false, reason: "no-result" }); + } + } + + return statuses; + }; for (let index = 0; index < chunks.length; index += 1) { const chunkItems = Array.isArray(chunks[index]) ? chunks[index] : []; @@ -2395,9 +2442,13 @@ class NostrClient { return { ok: false, error: "signing-failed", retryable: false }; } - const publishResults = await publishEventToRelays(this.pool, relays, signedEvent); - const summary = summarizePublishResults(publishResults); - const acceptedCount = summary.accepted.length; + const publishResults = await publishEventToRelays( + this.pool, + relays, + signedEvent, + ); + const relayStatus = formatRelayStatus(publishResults); + const acceptedCount = relayStatus.filter((entry) => entry.success).length; if (acceptedCount === 0) { anyChunkRejected = true; @@ -2406,9 +2457,21 @@ class NostrClient { publishResults, ); } else { - console.info( - `[nostr] Watch history chunk ${index} accepted by ${acceptedCount}/${relays.length} relay(s).`, - ); + const logMessage = + acceptedCount === relays.length + ? "accepted" + : "partially accepted"; + if (acceptedCount === relays.length) { + console.info( + `[nostr] Watch history chunk ${index} accepted by ${acceptedCount}/${relays.length} relay(s).`, + ); + } else { + anyChunkPartial = true; + console.warn( + `[nostr] Watch history chunk ${index} ${logMessage} by ${acceptedCount}/${relays.length} relay(s).`, + publishResults, + ); + } } const address = eventToAddressPointer(signedEvent); @@ -2420,6 +2483,7 @@ class NostrClient { event: signedEvent, publishResults, acceptedCount, + relayStatus, }); } @@ -2440,23 +2504,66 @@ class NostrClient { return { ok: false, error: "signing-failed", retryable: false }; } - const pointerResults = await publishEventToRelays(this.pool, relays, signedPointerEvent); - const pointerSummary = summarizePublishResults(pointerResults); - const pointerAccepted = pointerSummary.accepted.length > 0; + const pointerResults = await publishEventToRelays( + this.pool, + relays, + signedPointerEvent, + ); + const pointerRelayStatus = formatRelayStatus(pointerResults); + const pointerAcceptedCount = pointerRelayStatus.filter((entry) => entry.success) + .length; + const pointerAccepted = pointerAcceptedCount > 0; - if (pointerAccepted) { + if (pointerAcceptedCount === relays.length) { console.info( - `[nostr] Watch history pointer accepted by ${pointerSummary.accepted.length}/${relays.length} relay(s).`, + `[nostr] Watch history pointer accepted by ${pointerAcceptedCount}/${relays.length} relay(s).`, + ); + } else if (pointerAccepted) { + console.warn( + `[nostr] Watch history pointer partially accepted by ${pointerAcceptedCount}/${relays.length} relay(s).`, + pointerResults, ); } else { - console.warn("[nostr] Watch history pointer rejected by all relays:", pointerResults); + console.warn( + "[nostr] Watch history pointer rejected by all relays:", + pointerResults, + ); } this.watchHistoryLastCreatedAt = createdAtCursor; - const success = pointerAccepted && !anyChunkRejected; + const chunkStatuses = chunkResults.map((entry) => entry.relayStatus); + const chunkAcceptedEverywhere = chunkResults.every( + (entry) => entry.acceptedCount === relays.length, + ); + const chunkRejectedEverywhere = chunkResults.some( + (entry) => entry.acceptedCount === 0, + ); + const pointerRejectedEverywhere = pointerAcceptedCount === 0; + const pointerPartial = + pointerAccepted && pointerAcceptedCount < relays.length; + const partialAcceptance = pointerPartial || anyChunkPartial; + const success = + !pointerRejectedEverywhere && pointerAcceptedCount === relays.length && + chunkAcceptedEverywhere && + !anyChunkRejected; - return { + let errorCode = null; + if (!success) { + if (pointerRejectedEverywhere && chunkRejectedEverywhere) { + errorCode = "pointer-and-chunk-rejected"; + } else if (pointerRejectedEverywhere) { + errorCode = "pointer-rejected"; + } else if (chunkRejectedEverywhere || anyChunkRejected) { + errorCode = "chunk-rejected"; + } else if (partialAcceptance) { + errorCode = "partial-relay-acceptance"; + } else { + errorCode = "publish-rejected"; + } + } + + const result = { ok: success, retryable: !success, actor: actorPubkey, @@ -2467,10 +2574,21 @@ class NostrClient { publishResults: { pointer: pointerResults, chunks: chunkResults.map((entry) => entry.publishResults), + relayStatus: { + pointer: pointerRelayStatus, + chunks: chunkStatuses, + }, }, skippedCount: skipped.length, source: options.source || "manual", + partial: partialAcceptance, }; + + if (!success && errorCode) { + result.error = errorCode; + } + + return result; } async updateWatchHistoryList(rawItems = [], options = {}) { diff --git a/js/watchHistoryService.js b/js/watchHistoryService.js index cbbcdcba..1aeb3bef 100644 --- a/js/watchHistoryService.js +++ b/js/watchHistoryService.js @@ -753,9 +753,14 @@ function scheduleRepublishForQueue(actorKey) { source: `${queue.lastSnapshotReason || "session"}-retry`, } ); + if (publishResult?.snapshotId) { + queue.pendingSnapshotId = publishResult.snapshotId; + } if (publishResult?.ok) { queue.pendingSnapshotId = null; queue.republishScheduled = false; + queue.items.clear(); + queue.throttle.clear(); await updateFingerprintCache( actorKey, publishResult.items || latestItems, diff --git a/tests/watch-history.test.mjs b/tests/watch-history.test.mjs index 10c9a2a1..8558c243 100644 --- a/tests/watch-history.test.mjs +++ b/tests/watch-history.test.mjs @@ -46,6 +46,8 @@ const originalWatchHistoryLastCreatedAt = nostrClient.watchHistoryLastCreatedAt; const originalRecordVideoView = nostrClient.recordVideoView; const originalWatchHistoryCache = nostrClient.watchHistoryCache; const originalWatchHistoryStorage = nostrClient.watchHistoryStorage; +const originalScheduleWatchHistoryRepublish = + nostrClient.scheduleWatchHistoryRepublish; nostrClient.watchHistoryCache = new Map(); nostrClient.watchHistoryStorage = { @@ -605,6 +607,211 @@ async function testPublishSnapshotFailureRetry() { } } +async function testWatchHistoryPartialRelayRetry() { + poolHarness.reset(); + watchHistoryService.resetProgress(); + nostrClient.watchHistoryLastCreatedAt = 0; + + const actor = "partial-actor"; + const restoreCrypto = installSessionCrypto({ privateKey: "partial-priv" }); + const originalEnsure = nostrClient.ensureSessionActor; + const originalSession = nostrClient.sessionActor; + const originalPub = nostrClient.pubkey; + const originalRelays = Array.isArray(nostrClient.relays) + ? [...nostrClient.relays] + : nostrClient.relays; + const originalWriteRelays = Array.isArray(nostrClient.writeRelays) + ? [...nostrClient.writeRelays] + : nostrClient.writeRelays; + const originalPublishSnapshot = nostrClient.publishWatchHistorySnapshot; + const originalSchedule = nostrClient.scheduleWatchHistoryRepublish; + const originalRecordView = nostrClient.recordVideoView; + + try { + nostrClient.pubkey = ""; + nostrClient.sessionActor = { pubkey: actor, privateKey: "partial-priv" }; + nostrClient.ensureSessionActor = async () => actor; + + const relaySet = [ + "wss://relay.one", + "wss://relay.two", + "wss://relay.three", + ]; + nostrClient.relays = [...relaySet]; + nostrClient.writeRelays = [...relaySet]; + + let attemptIndex = 0; + let currentAttempt = 0; + nostrClient.publishWatchHistorySnapshot = async function publishWithTracking( + ...args + ) { + attemptIndex += 1; + currentAttempt = attemptIndex; + return originalPublishSnapshot.apply(this, args); + }; + + nostrClient.recordVideoView = async () => ({ + ok: true, + event: { + id: `view-${Date.now()}`, + pubkey: actor, + created_at: Math.floor(Date.now() / 1000), + }, + }); + + await watchHistoryService.publishView( + { type: "e", value: "partial-pointer" }, + Math.floor(Date.now() / 1000), + { actor }, + ); + + const queuedBefore = watchHistoryService.getQueuedPointers(actor); + assert.equal( + queuedBefore.length, + 1, + "queue should contain the partial pointer before snapshot", + ); + assert.deepEqual( + nostrClient.writeRelays, + relaySet, + "write relays should include all configured endpoints", + ); + + const attemptPlans = [ + { + "wss://relay.one": true, + "wss://relay.two": true, + "wss://relay.three": false, + }, + { + "wss://relay.one": true, + "wss://relay.two": false, + "wss://relay.three": true, + }, + { + "wss://relay.one": true, + "wss://relay.two": true, + "wss://relay.three": true, + }, + ]; + + poolHarness.setResolver(({ relays }) => { + const relayUrl = Array.isArray(relays) && relays.length ? relays[0] : ""; + const index = Math.min( + Math.max(currentAttempt, 1) - 1, + attemptPlans.length - 1, + ); + const plan = attemptPlans[index] || {}; + const accept = plan?.[relayUrl]; + if (accept) { + return { ok: true }; + } + return { ok: false, error: new Error(`reject-${relayUrl || "unknown"}`) }; + }); + + const scheduledRuns = []; + nostrClient.scheduleWatchHistoryRepublish = (snapshotId, operation) => { + const promise = (async () => { + let attempt = 1; + let result = null; + for (; attempt <= attemptPlans.length + 1; attempt += 1) { + result = await operation(attempt); + if (result?.ok || !result?.retryable) { + break; + } + } + return result; + })(); + scheduledRuns.push({ snapshotId, promise }); + return { attempt: 1, delay: 0 }; + }; + + let thrownError = null; + try { + await watchHistoryService.snapshot(null, { actor, reason: "partial-test" }); + } catch (error) { + thrownError = error; + } + + assert(thrownError, "snapshot should throw when partial acceptance occurs"); + assert.equal( + thrownError?.result?.retryable, + true, + "partial failures should be marked retryable", + ); + assert.equal( + thrownError?.result?.error, + "partial-relay-acceptance", + "partial failures should expose the partial acceptance error code", + ); + assert(thrownError?.result?.partial, "result should report partial acceptance"); + + const initialPointerStatus = + thrownError?.result?.publishResults?.relayStatus?.pointer || []; + assert( + initialPointerStatus.some((entry) => entry && entry.success === false), + "initial relay status should capture pointer rejections", + ); + + assert( + attemptIndex >= 1, + "initial snapshot attempt should increment the attempt counter", + ); + assert.equal( + scheduledRuns.length, + 1, + "partial failure should schedule a republish operation", + ); + + const finalResult = await scheduledRuns[0].promise; + assert(finalResult?.ok, "republish attempts should converge to success"); + assert.equal( + attemptIndex, + 3, + "republish loop should retry until every relay accepts", + ); + assert.equal( + finalResult?.partial, + false, + "final result should not mark the publish as partial", + ); + + const finalPointerStatus = + finalResult?.publishResults?.relayStatus?.pointer || []; + assert.equal( + finalPointerStatus.filter((entry) => entry?.success).length, + relaySet.length, + "final pointer publish should succeed on all relays", + ); + for (const chunkStatus of finalResult?.publishResults?.relayStatus?.chunks || []) { + assert.equal( + chunkStatus.filter((entry) => entry?.success).length, + relaySet.length, + "final chunk publish should succeed on all relays", + ); + } + + const remainingQueue = watchHistoryService.getQueuedPointers(actor); + assert.equal( + remainingQueue.length, + 0, + "queue should remain empty after successful retries", + ); + } finally { + nostrClient.scheduleWatchHistoryRepublish = originalSchedule; + nostrClient.publishWatchHistorySnapshot = originalPublishSnapshot; + nostrClient.relays = originalRelays; + nostrClient.writeRelays = originalWriteRelays; + nostrClient.recordVideoView = originalRecordView; + nostrClient.ensureSessionActor = originalEnsure; + nostrClient.sessionActor = originalSession; + nostrClient.pubkey = originalPub; + poolHarness.setResolver(() => ({ ok: true })); + watchHistoryService.resetProgress(actor); + restoreCrypto.restore(); + } +} + async function testResolveWatchHistoryBatchingWindow() { const actor = "batch-window-actor"; const originalEnsure = nostrClient.ensureSessionActor; @@ -776,6 +983,7 @@ async function testWatchHistoryServiceIntegration() { await testPublishSnapshotCanonicalizationAndChunking(); await testPublishSnapshotUsesExtensionCrypto(); await testPublishSnapshotFailureRetry(); +await testWatchHistoryPartialRelayRetry(); await testResolveWatchHistoryBatchingWindow(); await testWatchHistoryServiceIntegration(); @@ -794,6 +1002,8 @@ nostrClient.watchHistoryLastCreatedAt = originalWatchHistoryLastCreatedAt; nostrClient.recordVideoView = originalRecordVideoView; nostrClient.watchHistoryCache = originalWatchHistoryCache; nostrClient.watchHistoryStorage = originalWatchHistoryStorage; +nostrClient.scheduleWatchHistoryRepublish = + originalScheduleWatchHistoryRepublish; if (!originalFlag) { setWatchHistoryV2Enabled(false);