Handle partial watch history publish retries

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 = {}) {

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,

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);