fix(stats): retry failed syncs instead of waiting an hour (#57516)

packages/stats/core/src/r2-sql.test.ts

@@ -1,5 +1,6 @@
 import { Database } from "bun:sqlite"
-import { Effect } from "effect"
+import { Effect, Fiber } from "effect"
+import { TestClock } from "effect/testing"
 import { expect, test } from "bun:test"
 import { queryR2SqlPages, R2SqlQueryError, type R2SqlData } from "./r2-sql"
 
@@ -97,6 +98,54 @@ test("retries a transient timeout on the same page without repeating earlier row
   expect(calls[0]).not.toBe(calls[1])
 }, 10000)
 
+for (const [name, error] of [
+  [
+    "an intermittent 60104 response",
+    new R2SqlQueryError({ message: "Unprocessable Entity", status: 422, code: 60104 }),
+  ],
+  [
+    "a dropped connection",
+    new R2SqlQueryError({
+      message: "Failed to run R2 SQL stats query",
+      cause: new TypeError("The socket connection was closed unexpectedly"),
+    }),
+  ],
+] as const) {
+  test(`retries ${name} on the same page`, async () => {
+    const calls: string[] = []
+    const rows = await Effect.runPromise(
+      Effect.gen(function* () {
+        const running = yield* queryR2SqlPages("SELECT * FROM aggregates", undefined, (query) => {
+          calls.push(query)
+          return calls.length === 1 ? Effect.fail(error) : Effect.succeed([{ model: "a" }])
+        }).pipe(Effect.forkChild({ startImmediately: true }))
+        yield* Effect.yieldNow
+        yield* TestClock.adjust("5 seconds")
+        return yield* Fiber.join(running)
+      }).pipe(Effect.provide(TestClock.layer())),
+    )
+    expect(rows).toEqual([{ model: "a" }])
+    expect(calls).toHaveLength(2)
+    expect(calls[1]).toBe(calls[0])
+  })
+}
+
+test("does not retry the request timeout", async () => {
+  const calls: string[] = []
+  const error = new R2SqlQueryError({
+    message: "Failed to run R2 SQL stats query",
+    cause: new DOMException("The operation timed out.", "TimeoutError"),
+  })
+  const result = await Effect.runPromise(
+    queryR2SqlPages("SELECT * FROM aggregates", undefined, (query) => {
+      calls.push(query)
+      return Effect.fail(error)
+    }).pipe(Effect.flip),
+  )
+  expect(result).toBe(error)
+  expect(calls).toHaveLength(1)
+})
+
 test("bounds transient retries and preserves the final error", async () => {
   const calls: string[] = []
   const error = new R2SqlQueryError({ message: "unavailable", status: 503 })

packages/stats/core/src/r2-sql.ts

@@ -66,13 +66,19 @@ export const queryR2SqlPages = Effect.fn("R2Sql.query")(function* (
     const page = yield* Effect.suspend(() =>
       fetchRows(columns?.length ? pageQuery(query, columns, cursor) : query),
     ).pipe(
-      // Retry only this page; a transient R2 timeout must not discard the whole
-      // display-window backfill. Syntax, auth, and row-limit errors still fail.
+      // Retry only this page; a transient R2 failure must not discard the whole
+      // pass. 40005 is a query timeout, 60104 an intermittent "Unprocessable
+      // Entity", and fetch rejects with a TypeError when the connection drops.
+      // Syntax, auth, row-limit, and 15-minute request timeout errors still fail.
       Effect.retry({
         times: 2,
         schedule: Schedule.exponential("5 seconds"),
         while: (error) =>
-          error.code === 40005 || error.status === 429 || (error.status !== undefined && error.status >= 500),
+          error.code === 40005 ||
+          error.code === 60104 ||
+          error.status === 429 ||
+          (error.status !== undefined && error.status >= 500) ||
+          error.cause instanceof TypeError,
       }),
     )
     if (page.length >= R2_SQL_MAX_ROWS && !columns?.length)

packages/stats/server/src/stat-sync.ts

@@ -33,9 +33,13 @@ const daemon = Effect.gen(function* () {
     }
     yield* syncStats({ full: false })
   }).pipe(
-    Effect.catchCause((cause) =>
+    Effect.tapCause((cause) =>
       Effect.logWarning(`stats sync failed ${JSON.stringify({ cause: Cause.pretty(cause) })}`),
     ),
+    // Without a retry, one transient R2 SQL failure leaves the site stale until the
+    // next hourly pass. Two retries cap the extra query spend per interval.
+    Effect.retry({ times: 2, schedule: Schedule.spaced("10 minutes") }),
+    Effect.ignoreCause,
   )
   yield* pass.pipe(Effect.repeat(Schedule.fixed(SYNC_INTERVAL)))
 }).pipe(Effect.forkScoped)