refactor(delivery): share recovery primitives

vincentkoc · vincentkoc · commit c782fa98aa88 · 2026-06-23T03:10:30.000+08:00
diff --git a/src/infra/delivery-recovery.shared.ts b/src/infra/delivery-recovery.shared.ts
@@ -0,0 +1,30 @@
+const RECOVERY_BACKOFF_MS: readonly number[] = [5_000, 25_000, 120_000, 600_000];
+
+export function computeBackoffMs(retryCount: number): number {
+  if (retryCount <= 0) {
+    return 0;
+  }
+  return (
+    RECOVERY_BACKOFF_MS[Math.min(retryCount - 1, RECOVERY_BACKOFF_MS.length - 1)] ??
+    RECOVERY_BACKOFF_MS.at(-1) ??
+    0
+  );
+}
+
+export function getErrnoCode(err: unknown): string | null {
+  return err && typeof err === "object" && "code" in err
+    ? String((err as { code?: unknown }).code)
+    : null;
+}
+
+export function claimRecoveryEntry(entriesInProgress: Set<string>, entryId: string): boolean {
+  if (entriesInProgress.has(entryId)) {
+    return false;
+  }
+  entriesInProgress.add(entryId);
+  return true;
+}
+
+export function releaseRecoveryEntry(entriesInProgress: Set<string>, entryId: string): void {
+  entriesInProgress.delete(entryId);
+}
diff --git a/src/infra/outbound/delivery-queue-recovery.ts b/src/infra/outbound/delivery-queue-recovery.ts
@@ -9,6 +9,12 @@ import type {
   ChannelMessageUnknownSendReconciliationResult,
 } from "../../channels/message/types.js";
 import type { OpenClawConfig } from "../../config/types.openclaw.js";
+import {
+  claimRecoveryEntry as claimSharedRecoveryEntry,
+  computeBackoffMs,
+  getErrnoCode,
+  releaseRecoveryEntry as releaseSharedRecoveryEntry,
+} from "../delivery-recovery.shared.js";
 import { formatErrorMessage } from "../errors.js";
 import { resolveOutboundChannelMessageAdapter } from "./channel-resolution.js";
 import type { OutboundDeliveryResult } from "./deliver-types.js";
@@ -26,6 +32,8 @@ import {
   type QueuedDeliveryPayload,
 } from "./delivery-queue-storage.js";
 
+export { computeBackoffMs };
+
 export type RecoverySummary = {
   recovered: number;
   failed: number;
@@ -61,14 +69,6 @@ export type ActiveDeliveryClaimResult<T> =
 
 const MAX_RETRIES = 5;
 
-/** Backoff delays in milliseconds indexed by retry count (1-based). */
-const BACKOFF_MS: readonly number[] = [
-  5_000, // retry 1: 5s
-  25_000, // retry 2: 25s
-  120_000, // retry 3: 2m
-  600_000, // retry 4: 10m
-];
-
 const PERMANENT_ERROR_PATTERNS: readonly RegExp[] = [
   /no conversation reference found/i,
   /chat not found/i,
@@ -97,12 +97,6 @@ function resolveRecoveryDeadlineMs(maxRecoveryMs: number | undefined): number {
   return resolveExpiresAtMsFromDurationMs(durationMs) ?? resolveDateTimestampMs(Date.now());
 }
 
-function getErrnoCode(err: unknown): string | null {
-  return err && typeof err === "object" && "code" in err
-    ? String((err as { code?: unknown }).code)
-    : null;
-}
-
 function createEmptyRecoverySummary(): RecoverySummary {
   return {
     recovered: 0,
@@ -112,30 +106,18 @@ function createEmptyRecoverySummary(): RecoverySummary {
   };
 }
 
-function claimRecoveryEntry(entryId: string): boolean {
-  if (entriesInProgress.has(entryId)) {
-    return false;
-  }
-  entriesInProgress.add(entryId);
-  return true;
-}
-
-function releaseRecoveryEntry(entryId: string): void {
-  entriesInProgress.delete(entryId);
-}
-
 export async function withActiveDeliveryClaim<T>(
   entryId: string,
   fn: () => Promise<T>,
 ): Promise<ActiveDeliveryClaimResult<T>> {
-  if (!claimRecoveryEntry(entryId)) {
+  if (!claimSharedRecoveryEntry(entriesInProgress, entryId)) {
     return { status: "claimed-by-other-owner" };
   }
 
   try {
     return { status: "claimed", value: await fn() };
   } finally {
-    releaseRecoveryEntry(entryId);
+    releaseSharedRecoveryEntry(entriesInProgress, entryId);
   }
 }
 
@@ -326,14 +308,6 @@ async function moveEntryToFailedWithLogging(
   }
 }
 
-/** Compute the backoff delay in ms for a given retry count. */
-export function computeBackoffMs(retryCount: number): number {
-  if (retryCount <= 0) {
-    return 0;
-  }
-  return BACKOFF_MS[Math.min(retryCount - 1, BACKOFF_MS.length - 1)] ?? BACKOFF_MS.at(-1) ?? 0;
-}
-
 export function isEntryEligibleForRecoveryRetry(
   entry: QueuedDelivery,
   now: number,
@@ -513,7 +487,7 @@ export async function drainPendingDeliveries(opts: {
     );
 
     for (const entry of matchingEntries) {
-      if (!claimRecoveryEntry(entry.id)) {
+      if (!claimSharedRecoveryEntry(entriesInProgress, entry.id)) {
         opts.log.info(`${opts.logLabel}: entry ${entry.id} is already being recovered`);
         continue;
       }
@@ -582,7 +556,7 @@ export async function drainPendingDeliveries(opts: {
           );
         }
       } finally {
-        releaseRecoveryEntry(entry.id);
+        releaseSharedRecoveryEntry(entriesInProgress, entry.id);
       }
     }
   } finally {
@@ -622,7 +596,7 @@ export async function recoverPendingDeliveries(opts: {
       break;
     }
 
-    if (!claimRecoveryEntry(entry.id)) {
+    if (!claimSharedRecoveryEntry(entriesInProgress, entry.id)) {
       opts.log.info(`Recovery skipped for delivery ${entry.id}: already being processed`);
       continue;
     }
@@ -677,7 +651,7 @@ export async function recoverPendingDeliveries(opts: {
         continue;
       }
     } finally {
-      releaseRecoveryEntry(entry.id);
+      releaseSharedRecoveryEntry(entriesInProgress, entry.id);
     }
   }
 
diff --git a/src/infra/session-delivery-queue-recovery.ts b/src/infra/session-delivery-queue-recovery.ts
@@ -4,6 +4,12 @@ import {
   resolveExpiresAtMsFromDurationMs,
   resolveNonNegativeIntegerOption,
 } from "@openclaw/normalization-core/number-coercion";
+import {
+  claimRecoveryEntry as claimSharedRecoveryEntry,
+  computeBackoffMs,
+  getErrnoCode,
+  releaseRecoveryEntry as releaseSharedRecoveryEntry,
+} from "./delivery-recovery.shared.js";
 import { formatErrorMessage } from "./errors.js";
 import {
   ackSessionDelivery,
@@ -38,16 +44,9 @@ interface PendingSessionDeliveryDrainDecision {
 
 const MAX_SESSION_DELIVERY_RETRIES = 5;
 
-const BACKOFF_MS: readonly number[] = [5_000, 25_000, 120_000, 600_000];
 const drainInProgress = new Map<string, boolean>();
 const entriesInProgress = new Set<string>();
 
-function getErrnoCode(err: unknown): string | null {
-  return err && typeof err === "object" && "code" in err
-    ? String((err as { code?: unknown }).code)
-    : null;
-}
-
 function createEmptyRecoverySummary(): SessionDeliveryRecoverySummary {
   return {
     recovered: 0,
@@ -57,25 +56,6 @@ function createEmptyRecoverySummary(): SessionDeliveryRecoverySummary {
   };
 }
 
-function claimRecoveryEntry(entryId: string): boolean {
-  if (entriesInProgress.has(entryId)) {
-    return false;
-  }
-  entriesInProgress.add(entryId);
-  return true;
-}
-
-function releaseRecoveryEntry(entryId: string): void {
-  entriesInProgress.delete(entryId);
-}
-
-function computeSessionDeliveryBackoffMs(retryCount: number): number {
-  if (retryCount <= 0) {
-    return 0;
-  }
-  return BACKOFF_MS[Math.min(retryCount - 1, BACKOFF_MS.length - 1)] ?? BACKOFF_MS.at(-1) ?? 0;
-}
-
 function resolveSessionDeliveryMaxRetries(entry: QueuedSessionDelivery): number {
   return entry.maxRetries ?? MAX_SESSION_DELIVERY_RETRIES;
 }
@@ -92,7 +72,7 @@ export function isSessionDeliveryEligibleForRetry(
   entry: QueuedSessionDelivery,
   now: number,
 ): { eligible: true } | { eligible: false; remainingBackoffMs: number } {
-  const backoff = computeSessionDeliveryBackoffMs(entry.retryCount);
+  const backoff = computeBackoffMs(entry.retryCount);
   if (backoff <= 0) {
     return { eligible: true };
   }
@@ -160,7 +140,7 @@ export async function drainPendingSessionDeliveries(opts: {
       .toSorted((a, b) => a.enqueuedAt - b.enqueuedAt);
 
     for (const entry of matchingEntries) {
-      if (!claimRecoveryEntry(entry.id)) {
+      if (!claimSharedRecoveryEntry(entriesInProgress, entry.id)) {
         opts.log.info(`${opts.logLabel}: entry ${entry.id} is already being recovered`);
         continue;
       }
@@ -207,7 +187,7 @@ export async function drainPendingSessionDeliveries(opts: {
           },
         });
       } finally {
-        releaseRecoveryEntry(entry.id);
+        releaseSharedRecoveryEntry(entriesInProgress, entry.id);
       }
     }
   } finally {
@@ -239,7 +219,7 @@ export async function recoverPendingSessionDeliveries(opts: {
       opts.log.warn("Session delivery recovery time budget exceeded — remaining entries deferred");
       break;
     }
-    if (!claimRecoveryEntry(entry.id)) {
+    if (!claimSharedRecoveryEntry(entriesInProgress, entry.id)) {
       continue;
     }
 
@@ -285,7 +265,7 @@ export async function recoverPendingSessionDeliveries(opts: {
         opts.log.info(`Recovered session delivery ${currentEntry.id}`);
       }
     } finally {
-      releaseRecoveryEntry(entry.id);
+      releaseSharedRecoveryEntry(entriesInProgress, entry.id);
     }
   }
 
