From 298d4877821aa75e3c87c6ba1ac668f2574bc489 Mon Sep 17 00:00:00 2001 From: anonsaber Date: Mon, 12 Oct 2026 17:43:26 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20stop=20swallowing=20reconcile=20errors?= =?UTF-8?q?=20=E2=80=94=20surface=20reason=20in=20503=20body?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 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 | 24 ++++++++++++--- cloudflare-worker/test/lb.test.js | 19 +++++++++++- src/notify.rs | 36 ++++++++++++++++------ src/worker.rs | 51 ++++++++++++++++++++++--------- 4 files changed, 101 insertions(+), 29 deletions(-) diff --git a/cloudflare-worker/src/lb.js b/cloudflare-worker/src/lb.js index 8e14af5..064a95e 100644 --- a/cloudflare-worker/src/lb.js +++ b/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", diff --git a/cloudflare-worker/test/lb.test.js b/cloudflare-worker/test/lb.test.js index ba0499a..03f23fb 100644 --- a/cloudflare-worker/test/lb.test.js +++ b/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/); }); diff --git a/src/notify.rs b/src/notify.rs index 77c5acb..c5eed26 100644 --- a/src/notify.rs +++ b/src/notify.rs @@ -334,22 +334,40 @@ async fn reconcile(State(app): State, 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 = 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), diff --git a/src/worker.rs b/src/worker.rs index 6cc2d05..75c2d4a 100644 --- a/src/worker.rs +++ b/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 { - Err(()) + ) -> Result { + Err("reconcile_unsupported") } } @@ -143,8 +148,12 @@ impl WorkerHandler for WorkerHandle { &self, since_state: Option<&str>, initial_changes: usize, - ) -> Result { - let worker = self.current.read().map_err(|_| ())?.clone(); + ) -> Result { + let worker = self + .current + .read() + .map_err(|_| "worker_handle_unavailable")? + .clone(); worker.reconcile(since_state, initial_changes).await } } @@ -344,7 +353,7 @@ impl WorkerHandler for MetadataWorker { &self, since_state: Option<&str>, initial_changes: usize, - ) -> Result { + ) -> Result { // `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 WorkerHandler for MetadataWorker { } 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 WorkerHandler for MetadataWorker { 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 MetadataWorker { &self, cursor: ReconcileCursor, started: tokio::time::Instant, - ) -> Result { + ) -> Result { // 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 MetadataWorker { .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 MetadataWorker { mut state: String, initial_window: usize, started: tokio::time::Instant, - ) -> Result { + ) -> Result { let mut window = initial_window; let mut pages = 0_usize; loop { @@ -444,7 +458,11 @@ impl MetadataWorker { // 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 MetadataWorker { // 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 MetadataWorker { // 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 MetadataWorker { - 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 MetadataWorker { self.state .claim_dedup_and_enqueue(&key, ttl::DEDUP_JMAP_SECONDS, "stalwart:jmap", &payload) .await - .map_err(|_| ())?; + .map_err(|_| "enqueue_failed")?; Ok(()) }