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(()) }