fix: stop swallowing reconcile errors — surface reason in 503 body

Two layers were masking the real cause of reconcile failures:

1. Backend: worker.rs reconciled all JMAP/Redis failures into Err(()),
   so handle_reconcile returned 503 with {"error":"reconcile_retry"} but
   no indication of WHICH step broke. Now reconcile() returns
   Result<String, &'static str> with diagnostic codes:
   - jmap_session_failed — session_state() failed (JMAP unreachable/auth)
   - jmap_list_failed — list_emails_page() failed
   - reconcile_budget_exhausted — budget/pages exhausted before any progress
   - reconcile_no_progress — walk made no progress
   - reconcile_lease_lost — background heartbeat lost the lock
   - reconcile_state_write_failed — set_reconcile_state() Redis failure
   - enqueue_failed — claim_dedup_and_enqueue() Redis failure

   The 503 body now includes a "reason" field alongside the documented
   "reconcile_retry" error code, so operators can see the cause without
   server-side logs. The error code is unchanged (stable contract for
   cron-drain.sh and other callers).

2. Cloudflare Worker LB: forwardWithFailover treated all 5xx as backend
   failure and replaced the real response body with synthesized
   "all backends failed: ..." text. Now the last 5xx HTTP response is
   kept and passed through, so the backend's real error body (including
   the new reason field) reaches the client. Pure network errors (no HTTP
   response at all) still get the synthesized 503 fallback.

cloudflare-worker/src/lb.js

@@ -108,6 +108,10 @@ export async function forwardWithFailover(origins, request, opts) {
   const queue = origins.slice(start).concat(origins.slice(0, start));
   const attempts = Math.max(1, Math.min(maxAttempts, queue.length));
   let lastFailure = "no backends configured";
+  // 保留最后一次 5xx HTTP 响应（未 cancel body），全部 origin 失败时透传它，
+  // 让客户端看到后端真实错误 body（如 reconcile 的 reason 字段），而非合成文本。
+  let last5xxResp = null;
+  let last5xxOrigin = "";
   // 一次性读 body 快照：让每次故障转移尝试可回灌同一请求体（body 流只能读一次）。
   const snapshot = await snapshotBody(request);
 
@@ -118,20 +122,32 @@ export async function forwardWithFailover(origins, request, opts) {
       const resp = await forwardRequest(origin, request, io, snapshot);
       // 非 5xx / 非网络错误：直接返回（4xx、2xx、3xx 均原样透传，不故障转移）。
       if (!isRetryableError(null, resp.status)) {
+        // 成功返回前释放之前保留的 5xx body（如果有），避免泄漏。
+        if (last5xxResp) last5xxResp.body?.cancel?.().catch(() => {});
         return mark(resp, origin);
       }
-      // 5xx：释放 body，换下一个实例。
-      resp.body?.cancel?.().catch(() => {});
+      // 5xx：释放上一次保留的 5xx body，保留这一次（最后一次胜出）。
+      // 全部 origin 失败时透传最后一次 5xx——后端的错误码/reason 字段
+      // 比「all backends failed」对排障有用得多。
+      if (last5xxResp) last5xxResp.body?.cancel?.().catch(() => {});
+      last5xxResp = resp;
+      last5xxOrigin = origin;
       lastFailure = `backend ${origin} returned HTTP ${resp.status}`;
       if (log) log(`retrying ${request.method} ${new URL(request.url).pathname}: ${lastFailure}`);
     } catch (err) {
       lastFailure = `backend ${origin} unavailable (${err && err.name ? err.name : "error"})`;
       if (log) log(lastFailure);
-      // 仅「可故障转移」的网络类失败（超时/连接错误）才换实例；其它错误 fail-fast，避免无限重试。
+      // 网络错误不清除 last5xxResp：如果之前某个 origin 返回了 5xx body，
+      // 那个 body 比网络错误描述更有诊断价值。
+      // 仅「可故障转移」的网络类失败（超时/连接错误）才换实例；其它错误 fail-fast。
       if (!isRetryableError(err, 0)) break;
     }
   }
-  // 全部 origin 失败 → 5xx（交由 Telegram/Stalwart 自动重投兜底，不丢消息）。
+  // 全部 origin 失败：如有 5xx HTTP 响应，透传它（带后端真实 body + status）。
+  if (last5xxResp) {
+    return mark(last5xxResp, last5xxOrigin);
+  }
+  // 全部是网络错误（无 HTTP 响应）→ 合成 5xx（交由 Telegram/Stalwart 自动重投兜底，不丢消息）。
   return new Response(`all backends failed: ${lastFailure}`, {
     status: 503,
     statusText: "All Backends Unavailable",

cloudflare-worker/test/lb.test.js

@@ -178,7 +178,8 @@ test("timeout(AbortError) triggers failover and recovery", async () => {
   assert.equal(calls.length, 2);
 });
 
-test("all backends 5xx/timeout => aggregated 503 (Telegram redelivery fallback)", async () => {
+test("all backends fail: HTTP 5xx 透传最后一次真实 body（不被合成文本覆盖）", async () => {
+  // 网络错误 + HTTP 5xx 混合：保留 5xx 响应并透传，客户端能看到后端真实错误 body。
   const { fetch, calls } = mockFetch([{ error: abortError() }, { status: 503, body: "x" }]);
   const res = await forwardWithFailover(["https://a.example", "https://b.example"], req("/push/jmap"), {
     ...noOpTimers,
@@ -189,6 +190,22 @@ test("all backends 5xx/timeout => aggregated 503 (Telegram redelivery fallback)"
   });
   assert.equal(res.status, 503);
   assert.equal(calls.length, 2);
+  assert.equal(await res.text(), "x", "5xx 透传：后端真实 body 不被合成文本覆盖");
+  assert.equal(res.headers.get("x-lb-backend"), "https://b.example");
+});
+
+test("all backends network-error only => 合成 503 (Telegram redelivery fallback)", async () => {
+  // 纯网络错误（无 HTTP 响应）→ 无真实 body 可透传 → 合成 503。
+  const { fetch, calls } = mockFetch([{ error: abortError() }]);
+  const res = await forwardWithFailover(["https://a.example", "https://b.example"], req("/push/jmap"), {
+    ...noOpTimers,
+    fetch,
+    maxAttempts: 2,
+    timeoutMs: 5000,
+    rng: () => 0,
+  });
+  assert.equal(res.status, 503);
+  assert.equal(calls.length, 2);
   assert.match(await res.text(), /all backends failed/);
 });
 

src/notify.rs

@@ -334,22 +334,40 @@ async fn reconcile(State(app): State<AppState>, headers: HeaderMap) -> Response
                 .reconcile(since.as_deref(), RECONCILE_INITIAL_CHANGES)
                 .await;
             heartbeat.abort();
-            let response = if lease_lost.load(Ordering::Acquire) {
-                StatusCode::SERVICE_UNAVAILABLE
+            // Keep the failure reason instead of collapsing everything into a
+            // bare 503: without it operators cannot tell a JMAP outage from a
+            // Redis error, and (behind the Worker LB) the real cause is lost.
+            let outcome: Result<StatusCode, &'static str> = if lease_lost.load(Ordering::Acquire) {
+                Err("reconcile_lease_lost")
             } else {
                 match result {
                     Ok(new_state) => match app.state.set_reconcile_state(&new_state).await {
-                        Ok(()) => StatusCode::NO_CONTENT,
-                        Err(_) => StatusCode::SERVICE_UNAVAILABLE,
+                        Ok(()) => Ok(StatusCode::NO_CONTENT),
+                        Err(_) => Err("reconcile_state_write_failed"),
                     },
-                    Err(()) => StatusCode::SERVICE_UNAVAILABLE,
+                    Err(reason) => Err(reason),
                 }
             };
             let _ = app.state.release_lock("lock:reconcile", &lock_owner).await;
-            if response == StatusCode::SERVICE_UNAVAILABLE {
-                error_response(response, "reconcile_retry", true)
-            } else {
-                response.into_response()
+            match outcome {
+                Ok(status) if status == StatusCode::NO_CONTENT => status.into_response(),
+                Ok(status) => error_response(status, "reconcile_retry", true),
+                Err(reason) => {
+                    // `reconcile_retry` stays the documented error code (stable
+                    // contract for cron-drain.sh and other callers); `reason`
+                    // identifies the broken step (e.g. `jmap_session_failed`).
+                    let body = axum::Json(serde_json::json!({
+                        "error": "reconcile_retry",
+                        "reason": reason,
+                        "request_id": reconcile_lock_owner(),
+                    }));
+                    (
+                        StatusCode::SERVICE_UNAVAILABLE,
+                        [(header::RETRY_AFTER, "30")],
+                        body,
+                    )
+                        .into_response()
+                }
             }
         }
         Ok(false) => error_response(StatusCode::CONFLICT, "conflict", false),

src/worker.rs

@@ -40,12 +40,17 @@ pub trait WorkerHandler: Send + Sync {
     ///
     /// `initial_changes` sizes the first `/changes` call only. The pass may
     /// widen the window afterwards, which is why it is not called `max_changes`.
+    ///
+    /// On failure, returns a `&'static str` diagnostic code identifying which
+    /// step broke (e.g. `"jmap_session_failed"`, `"jmap_list_failed"`). The HTTP
+    /// layer surfaces this as a `reason` field in the 503 body so operators can
+    /// see the cause without server-side logs.
     async fn reconcile(
         &self,
         _since_state: Option<&str>,
         _initial_changes: usize,
-    ) -> Result<String, ()> {
-        Err(())
+    ) -> Result<String, &'static str> {
+        Err("reconcile_unsupported")
     }
 }
 
@@ -143,8 +148,12 @@ impl WorkerHandler for WorkerHandle {
         &self,
         since_state: Option<&str>,
         initial_changes: usize,
-    ) -> Result<String, ()> {
-        let worker = self.current.read().map_err(|_| ())?.clone();
+    ) -> Result<String, &'static str> {
+        let worker = self
+            .current
+            .read()
+            .map_err(|_| "worker_handle_unavailable")?
+            .clone();
         worker.reconcile(since_state, initial_changes).await
     }
 }
@@ -344,7 +353,7 @@ impl<B: JmapBackend> WorkerHandler for MetadataWorker<B> {
         &self,
         since_state: Option<&str>,
         initial_changes: usize,
-    ) -> Result<String, ()> {
+    ) -> Result<String, &'static str> {
         // `maxChanges` must be at least 1: a zero window is not a valid
         // `/changes` request, and rejecting it would waste the pass.
         let window = initial_changes.max(1);
@@ -355,7 +364,12 @@ impl<B: JmapBackend> WorkerHandler for MetadataWorker<B> {
         } else {
             // Capture a stable baseline before paging. Changes occurring while
             // the listing is in flight are replayed from it before committing.
-            ReconcileCursor::walk_from(self.jmap.session_state().await.map_err(|_| ())?)
+            ReconcileCursor::walk_from(
+                self.jmap
+                    .session_state()
+                    .await
+                    .map_err(|_| "jmap_session_failed")?,
+            )
         };
 
         // Phase 1: drain the position walk, the only path that can see email
@@ -364,7 +378,7 @@ impl<B: JmapBackend> WorkerHandler for MetadataWorker<B> {
         match self.walk_baseline(cursor, started).await? {
             WalkOutcome::ReplayFrom(state) => self.replay_changes(state, window, started).await,
             WalkOutcome::Persist(cursor) => Ok(cursor.to_persisted()),
-            WalkOutcome::NoProgress => Err(()),
+            WalkOutcome::NoProgress => Err("reconcile_no_progress"),
         }
     }
 }
@@ -377,7 +391,7 @@ impl<B: JmapBackend> MetadataWorker<B> {
         &self,
         cursor: ReconcileCursor,
         started: tokio::time::Instant,
-    ) -> Result<WalkOutcome, ()> {
+    ) -> Result<WalkOutcome, &'static str> {
         // A cursor in replay mode has nothing to walk: nothing was appended to
         // the collection since it was committed.
         let (state, mut position) = match cursor {
@@ -405,7 +419,7 @@ impl<B: JmapBackend> MetadataWorker<B> {
                 .jmap
                 .list_emails_page(None, position, BASELINE_PAGE_SIZE)
                 .await
-                .map_err(|_| ())?;
+                .map_err(|_| "jmap_list_failed")?;
             if page.is_empty() {
                 return Ok(WalkOutcome::ReplayFrom(state));
             }
@@ -436,7 +450,7 @@ impl<B: JmapBackend> MetadataWorker<B> {
         mut state: String,
         initial_window: usize,
         started: tokio::time::Instant,
-    ) -> Result<String, ()> {
+    ) -> Result<String, &'static str> {
         let mut window = initial_window;
         let mut pages = 0_usize;
         loop {
@@ -444,7 +458,11 @@ impl<B: JmapBackend> MetadataWorker<B> {
                 // Every event from the last completed page is already enqueued.
                 // Returning that cursor lets the next invocation continue from
                 // it instead of retrying the same bounded window forever.
-                return if pages == 0 { Err(()) } else { Ok(state) };
+                return if pages == 0 {
+                    Err("reconcile_budget_exhausted")
+                } else {
+                    Ok(state)
+                };
             }
             pages += 1;
             let changes = match self.jmap.email_changes(&state, window).await {
@@ -455,7 +473,7 @@ impl<B: JmapBackend> MetadataWorker<B> {
                     // carries `oldState`/`newState`/`hasMoreChanges`/`created`/
                     // `updated`, so a `/changes` failure is
                     // indistinguishable from "the server no longer replays from
-                    // this `sinceState`". Returning `Err(())` leaves the cursor
+                    // this `sinceState`". Returning an error leaves the cursor
                     // untouched, so every later invocation retries the same dead
                     // state and `reconcile:state` freezes forever (a persistent
                     // 503 `reconcile_retry` with no forward progress). The
@@ -465,7 +483,10 @@ impl<B: JmapBackend> MetadataWorker<B> {
                     // dedup key makes the resulting re-read at-most-duplicate,
                     // never-loss.
                     let rebaselined = ReconcileCursor::walk_from(
-                        self.jmap.session_state().await.map_err(|_| ())?,
+                        self.jmap
+                            .session_state()
+                            .await
+                            .map_err(|_| "jmap_session_failed")?,
                     );
                     return Ok(rebaselined.to_persisted());
                 }
@@ -507,7 +528,7 @@ fn decode_baseline_cursor(value: &str) -> Option<(String, usize)> {
 }
 
 impl<B: JmapBackend> MetadataWorker<B> {
-    async fn enqueue_reconcile_event(&self, email_id: &str) -> Result<(), ()> {
+    async fn enqueue_reconcile_event(&self, email_id: &str) -> Result<(), &'static str> {
         let key = format!("dedup:jmap:{}:{}", self.jmap.account_id(), email_id);
         let payload = serde_json::json!({
             "accountId": self.jmap.account_id(),
@@ -521,7 +542,7 @@ impl<B: JmapBackend> MetadataWorker<B> {
         self.state
             .claim_dedup_and_enqueue(&key, ttl::DEDUP_JMAP_SECONDS, "stalwart:jmap", &payload)
             .await
-            .map_err(|_| ())?;
+            .map_err(|_| "enqueue_failed")?;
         Ok(())
     }