Add bounded retry logic for lock publish quorum

src/lock-ops.mjs

@@ -61,6 +61,36 @@ function relayListLabel(relays) {
   return relays.join(', ');
 }
 
+function classifyPublishError(message) {
+  const normalized = String(message || '').toLowerCase();
+  if (normalized.includes('timed out') || normalized.includes('timeout') || normalized.includes('etimedout')) {
+    return 'network_timeout';
+  }
+  if (normalized.includes('econnreset') || normalized.includes('connection reset') || normalized.includes('socket hang up')) {
+    return 'connection_reset';
+  }
+  if (
+    normalized.includes('unavailable') ||
+    normalized.includes('offline') ||
+    normalized.includes('econnrefused') ||
+    normalized.includes('connection refused') ||
+    normalized.includes('enotfound') ||
+    normalized.includes('503')
+  ) {
+    return 'relay_unavailable';
+  }
+  return 'permanent_validation_error';
+}
+
+function isTransientPublishCategory(category) {
+  return category === 'network_timeout' || category === 'connection_reset' || category === 'relay_unavailable';
+}
+
+function calculateBackoffDelayMs(attemptNumber, baseMs, capMs, randomFn = Math.random) {
+  const maxDelay = Math.min(capMs, baseMs * (2 ** Math.max(0, attemptNumber - 1)));
+  return Math.floor(randomFn() * maxDelay);
+}
+
 export async function queryLocks(relays, cadence, dateStr, namespace, deps = {}) {
   const {
     poolFactory = () => new SimplePool(),
@@ -137,26 +167,33 @@ export async function publishLock(relays, event, deps = {}) {
     getPublishTimeoutMsFn = getPublishTimeoutMs,
     getMinSuccessfulRelayPublishesFn = getMinSuccessfulRelayPublishes,
     getRelayFallbacksFn = getRelayFallbacks,
+    retryAttempts = 4,
+    retryBaseDelayMs = 500,
+    retryCapDelayMs = 8_000,
+    sleepFn = (ms) => new Promise((resolve) => setTimeout(resolve, ms)),
+    randomFn = Math.random,
+    telemetryLogger = console.error,
   } = deps;
 
   const pool = poolFactory();
   const publishTimeoutMs = getPublishTimeoutMsFn();
   const minSuccesses = getMinSuccessfulRelayPublishesFn();
   const fallbackRelays = getRelayFallbacksFn().filter((relay) => !relays.includes(relay));
+  const maxAttempts = Math.max(1, Math.floor(retryAttempts));
 
-  const attempted = new Set();
+  const publishOnce = async () => {
-  const publishResults = [];
+    const attempted = new Set();
+    const publishResults = [];
 
-  const attemptPhase = async (phaseRelays, phaseName) => {
+    const attemptPhase = async (phaseRelays, phaseName) => {
-    if (!phaseRelays.length) return;
+      if (!phaseRelays.length) return;
-    for (const relay of phaseRelays) {
+      for (const relay of phaseRelays) {
-      attempted.add(relay);
+        attempted.add(relay);
-    }
+      }
-    const phaseResults = await publishToRelays(pool, phaseRelays, event, publishTimeoutMs, phaseName);
+      const phaseResults = await publishToRelays(pool, phaseRelays, event, publishTimeoutMs, phaseName);
-    publishResults.push(...phaseResults);
+      publishResults.push(...phaseResults);
-  };
+    };
 
-  try {
     await attemptPhase(relays, 'publish:primary');
 
     let successCount = publishResults.filter((result) => result.success).length;
@@ -165,23 +202,63 @@ export async function publishLock(relays, event, deps = {}) {
       successCount = publishResults.filter((result) => result.success).length;
     }
 
-    if (successCount < minSuccesses) {
+    const failures = publishResults
-      const failureLines = publishResults
+      .filter((result) => !result.success)
-        .filter((result) => !result.success)
+      .map((result) => ({
-        .map((result) => `${result.relay} (${result.phase}): ${result.message}`);
+        ...result,
+        category: classifyPublishError(result.message),
+      }));
 
-      throw new Error(
+    return { attempted, publishResults, successCount, failures };
-        `Failed relay publish quorum in publish phase: ${successCount}/${attempted.size} successful ` +
+  };
-        `(required=${minSuccesses}, timeout=${publishTimeoutMs}ms)\n  ${failureLines.join('\n  ')}`,
+
-      );
+  try {
+    let lastAttemptState = null;
+    for (let attemptNumber = 1; attemptNumber <= maxAttempts; attemptNumber += 1) {
+      const startedAtMs = Date.now();
+      lastAttemptState = await publishOnce();
+
+      if (lastAttemptState.successCount >= minSuccesses) {
+        console.error(
+          `  Published to ${lastAttemptState.successCount}/${lastAttemptState.attempted.size} relays ` +
+          `(required=${minSuccesses}, timeout=${publishTimeoutMs}ms)`,
+        );
+        return event;
+      }
+
+      const hasTransientFailure = lastAttemptState.failures.some((failure) => isTransientPublishCategory(failure.category));
+      const hasPermanentFailure = lastAttemptState.failures.some((failure) => !isTransientPublishCategory(failure.category));
+      const canRetry = attemptNumber < maxAttempts && hasTransientFailure && !hasPermanentFailure;
+
+      if (!canRetry) {
+        break;
+      }
+
+      const elapsedMs = Date.now() - startedAtMs;
+      const nextDelayMs = calculateBackoffDelayMs(attemptNumber, retryBaseDelayMs, retryCapDelayMs, randomFn);
+      for (const failure of lastAttemptState.failures) {
+        if (!isTransientPublishCategory(failure.category)) continue;
+        telemetryLogger(JSON.stringify({
+          event: 'lock_publish_retry',
+          attempt: attemptNumber,
+          relayUrl: failure.relay,
+          errorCategory: failure.category,
+          elapsedMs,
+          nextDelayMs,
+        }));
+      }
+
+      await sleepFn(nextDelayMs);
     }
 
-    console.error(
+    const failureLines = lastAttemptState.failures
-      `  Published to ${successCount}/${attempted.size} relays ` +
+      .map((result) => `${result.relay} (${result.phase}, category=${result.category}): ${result.message}`);
-      `(required=${minSuccesses}, timeout=${publishTimeoutMs}ms)`,
+
+    throw new Error(
+      `Failed relay publish quorum in publish phase: ${lastAttemptState.successCount}/${lastAttemptState.attempted.size} successful ` +
+      `(required=${minSuccesses}, timeout=${publishTimeoutMs}ms, attempts=${maxAttempts})\n  ${failureLines.join('\n  ')}`,
     );
-    return event;
   } finally {
-    pool.close([...attempted]);
+    pool.close([...new Set([...relays, ...fallbackRelays])]);
   }
 }

test/lock-ops.test.mjs

@@ -65,23 +65,155 @@ describe('lock-ops', () => {
   });
 
   describe('publishLock', () => {
-    it('succeeds with partial primary success when minimum successes is 1', async () => {
+    it('retries transient failures and succeeds on a later attempt', async () => {
+      const telemetry = [];
+      const sleepCalls = [];
+      let publishCalls = 0;
+
       const mockPool = {
-        publish: (relays) => relays.map((relay) => (
+        publish: (relays) => {
-          relay.includes('ok') ? Promise.resolve('ok') : Promise.reject(new Error('offline'))
+          publishCalls += 1;
-        )),
+          if (publishCalls === 1) {
+            return relays.map(() => Promise.reject(new Error('connection reset by peer')));
+          }
+          return relays.map(() => Promise.resolve('ok'));
+        },
         close: () => {},
       };
 
-      const event = { id: 'evt-1' };
+      const event = { id: 'evt-retry' };
-      const result = await publishLock(['wss://ok-1', 'wss://bad-1'], event, {
+      const result = await publishLock(['wss://primary-a', 'wss://primary-b'], event, {
         poolFactory: () => mockPool,
         getPublishTimeoutMsFn: () => 200,
         getMinSuccessfulRelayPublishesFn: () => 1,
         getRelayFallbacksFn: () => [],
+        retryAttempts: 3,
+        retryBaseDelayMs: 500,
+        retryCapDelayMs: 8_000,
+        randomFn: () => 0.5,
+        sleepFn: async (ms) => {
+          sleepCalls.push(ms);
+        },
+        telemetryLogger: (line) => {
+          telemetry.push(JSON.parse(line));
+        },
       });
 
       assert.strictEqual(result, event);
+      assert.strictEqual(publishCalls, 2);
+      assert.deepStrictEqual(sleepCalls, [250]);
+      assert.strictEqual(telemetry.length, 2);
+      assert.deepStrictEqual(telemetry[0], {
+        event: 'lock_publish_retry',
+        attempt: 1,
+        relayUrl: 'wss://primary-a',
+        errorCategory: 'connection_reset',
+        elapsedMs: telemetry[0].elapsedMs,
+        nextDelayMs: 250,
+      });
+      assert.deepStrictEqual(telemetry[1], {
+        event: 'lock_publish_retry',
+        attempt: 1,
+        relayUrl: 'wss://primary-b',
+        errorCategory: 'connection_reset',
+        elapsedMs: telemetry[1].elapsedMs,
+        nextDelayMs: 250,
+      });
+      assert.ok(Number.isInteger(telemetry[0].elapsedMs));
+      assert.ok(Number.isInteger(telemetry[1].elapsedMs));
+    });
+
+
+    it('fails after retry budget is exhausted for persistent transient failures', async () => {
+      let publishCalls = 0;
+      const sleepCalls = [];
+
+      const mockPool = {
+        publish: (relays) => {
+          publishCalls += 1;
+          return relays.map(() => Promise.reject(new Error('relay unavailable')));
+        },
+        close: () => {},
+      };
+
+      await assert.rejects(
+        () => publishLock(['wss://primary-down'], { id: 'evt-exhausted' }, {
+          poolFactory: () => mockPool,
+          getPublishTimeoutMsFn: () => 200,
+          getMinSuccessfulRelayPublishesFn: () => 1,
+          getRelayFallbacksFn: () => [],
+          retryAttempts: 3,
+          randomFn: () => 1,
+          sleepFn: async (ms) => {
+            sleepCalls.push(ms);
+          },
+          telemetryLogger: () => {},
+        }),
+        /Failed relay publish quorum in publish phase: 0\/1 successful \(required=1, timeout=200ms, attempts=3\)/,
+      );
+
+      assert.strictEqual(publishCalls, 3);
+      assert.deepStrictEqual(sleepCalls, [500, 1000]);
+    });
+
+    it('fails without retry for persistent validation failures', async () => {
+      let publishCalls = 0;
+      const mockPool = {
+        publish: (relays) => {
+          publishCalls += 1;
+          return relays.map(() => Promise.reject(new Error('validation failed: invalid signature')));
+        },
+        close: () => {},
+      };
+
+      await assert.rejects(
+        () => publishLock(['wss://primary-bad'], { id: 'evt-permanent' }, {
+          poolFactory: () => mockPool,
+          getPublishTimeoutMsFn: () => 200,
+          getMinSuccessfulRelayPublishesFn: () => 1,
+          getRelayFallbacksFn: () => [],
+          retryAttempts: 4,
+          sleepFn: async () => {
+            throw new Error('should not sleep for permanent failures');
+          },
+          telemetryLogger: () => {
+            throw new Error('should not emit retry telemetry for permanent failures');
+          },
+        }),
+        /Failed relay publish quorum in publish phase: 0\/1 successful \(required=1, timeout=200ms, attempts=4\)/,
+      );
+
+      assert.strictEqual(publishCalls, 1);
+    });
+
+    it('returns success when mixed relay outcomes still satisfy quorum', async () => {
+      let publishCalls = 0;
+      const mockPool = {
+        publish: (relays) => {
+          publishCalls += 1;
+          return relays.map((relay) => (
+            relay === 'wss://relay-ok'
+              ? Promise.resolve('ok')
+              : Promise.reject(new Error('validation failed: rejected event'))
+          ));
+        },
+        close: () => {},
+      };
+
+      const event = { id: 'evt-mixed' };
+      const result = await publishLock(['wss://relay-ok', 'wss://relay-bad'], event, {
+        poolFactory: () => mockPool,
+        getPublishTimeoutMsFn: () => 200,
+        getMinSuccessfulRelayPublishesFn: () => 1,
+        getRelayFallbacksFn: () => [],
+        retryAttempts: 3,
+        sleepFn: async () => {
+          throw new Error('should not sleep when quorum is satisfied');
+        },
+      });
+
+      assert.strictEqual(result, event);
+      assert.strictEqual(publishCalls, 1);
     });
 
     it('uses fallback relays to satisfy min success quorum', async () => {
@@ -102,22 +234,5 @@ describe('lock-ops', () => {
 
       assert.strictEqual(result, event);
     });
-
-    it('fails when successful publishes stay below configured minimum', async () => {
-      const mockPool = {
-        publish: (relays) => relays.map(() => Promise.reject(new Error('offline'))),
-        close: () => {},
-      };
-
-      await assert.rejects(
-        () => publishLock(['wss://primary-bad'], { id: 'evt-3' }, {
-          poolFactory: () => mockPool,
-          getPublishTimeoutMsFn: () => 200,
-          getMinSuccessfulRelayPublishesFn: () => 2,
-          getRelayFallbacksFn: () => ['wss://fallback-bad'],
-        }),
-        /Failed relay publish quorum in publish phase: 0\/2 successful \(required=2, timeout=200ms\)/,
-      );
-    });
   });
 });