poll: stop on errors a retry cannot clear
Assisted-by: Claude Opus 5.5 via Claude Code
diff --git a/src/bin/docket/main.rs b/src/bin/docket/main.rs
index 0e7fee2..29d8f46 100644
--- a/src/bin/docket/main.rs
+++ b/src/bin/docket/main.rs
@@ -57,13 +57,14 @@ async fn run(config: config::Config) -> miette::Result<()> {
// to quietly leave out. Each sync then polls for changes in the
// background.
let jmap = docket::jmap::Client::fastmail().into_diagnostic()?;
+ let mut polls = tokio::task::JoinSet::new();
for credential in config.credentials.iter().cloned() {
let sync = jmap
.sync_account(&credential, &store)
.await
.into_diagnostic()
.wrap_err_with(|| format!("opening the JMAP session for {}", credential.name))?;
- poll::spawn(jmap.clone(), credential, store.clone(), sync);
+ poll::spawn(&mut polls, jmap.clone(), credential, store.clone(), sync);
}
let state = docket::routes::AppState::new(store);
#[cfg(feature = "dev")]
@@ -74,12 +75,22 @@ async fn run(config: config::Config) -> miette::Result<()> {
let listener = TcpListener::bind(config.bind).await.into_diagnostic()?;
tracing::info!(addr = %config.bind, "listening");
- axum::serve(listener, app)
- .with_graceful_shutdown(shutdown_signal())
- .await
- .into_diagnostic()?;
+ // A stopped poll ends the process, so a broken credential fails
+ // loudly instead of serving stale mail.
+ let serve = axum::serve(listener, app).with_graceful_shutdown(shutdown_signal());
+ tokio::select! {
+ served = serve => served.into_diagnostic(),
+ stopped = first_stopped(&mut polls) => Err(stopped),
+ }
+}
- Ok(())
+/// Waits for a poll loop to end, forever when there are none.
+async fn first_stopped(polls: &mut tokio::task::JoinSet<poll::Stopped>) -> miette::Report {
+ match polls.join_next().await {
+ Some(Ok(stopped)) => miette::Report::new(stopped),
+ Some(Err(err)) => miette::Report::msg(format!("a poll task died: {err}")),
+ None => std::future::pending().await,
+ }
}
async fn shutdown_signal() {
diff --git a/src/bin/docket/poll.rs b/src/bin/docket/poll.rs
index 63dc448..b19d8bb 100644
--- a/src/bin/docket/poll.rs
+++ b/src/bin/docket/poll.rs
@@ -6,17 +6,35 @@
use std::time::Duration;
+use docket::Error;
use docket::jmap::{Client, Credential, Sync};
use docket::store::Store;
+use tokio::task::JoinSet;
/// "Within a minute or so" with headroom for the fetches a cycle makes.
const INTERVAL: Duration = Duration::from_secs(30);
-/// Runs the loop in the background. Errors are logged and retried on
-/// the next tick — a blip must not take syncing down with it — at
-/// error level, so each failure reaches Sentry.
-pub fn spawn(client: Client, credential: Credential, store: Store, sync: Sync) {
- tokio::spawn(async move {
+/// Why a credential's loop ended.
+#[derive(Debug, thiserror::Error, miette::Diagnostic)]
+#[error("polling {credential} stopped: {source}")]
+pub struct Stopped {
+ credential: String,
+ #[source]
+ source: Error,
+}
+
+/// Runs the loop in the background. A retryable error is logged and
+/// retried on the next tick — a blip must not take syncing down with
+/// it — at error level, so each failure reaches Sentry. Any other error
+/// ends the loop, since every tick would fail the same way.
+pub fn spawn(
+ polls: &mut JoinSet<Stopped>,
+ client: Client,
+ credential: Credential,
+ store: Store,
+ sync: Sync,
+) {
+ polls.spawn(async move {
let name = credential.name.clone();
let mut sync = sync;
loop {
@@ -28,9 +46,15 @@ pub fn spawn(client: Client, credential: Credential, store: Store, sync: Sync) {
Ok(counts) => {
tracing::info!(credential = %name, ?counts, "poll");
}
- Err(err) => {
+ Err(err) if err.is_retryable() => {
tracing::error!(credential = %name, %err, "poll failed; retrying next tick");
}
+ Err(source) => {
+ return Stopped {
+ credential: name,
+ source,
+ };
+ }
}
}
});
diff --git a/src/error.rs b/src/error.rs
index 31760b4..ca945f9 100644
--- a/src/error.rs
+++ b/src/error.rs
@@ -51,6 +51,25 @@ impl IntoResponse for Error {
}
}
+impl Error {
+ /// Whether a background job hitting this should try again rather
+ /// than stop.
+ pub fn is_retryable(&self) -> bool {
+ match self {
+ Error::Jmap(err) => err.is_retryable(),
+ Error::Db(err) => matches!(
+ err.sqlite_error_code(),
+ Some(rusqlite::ErrorCode::DatabaseBusy | rusqlite::ErrorCode::DatabaseLocked)
+ ),
+ Error::Io(_)
+ | Error::Unauthenticated
+ | Error::NotFound(_)
+ | Error::Forbidden(_)
+ | Error::BadRequest(_) => false,
+ }
+ }
+}
+
#[cfg(test)]
mod tests {
use super::*;
@@ -75,4 +94,38 @@ mod tests {
assert_eq!(err.into_response().status().as_u16(), status);
}
}
+
+ #[test]
+ fn only_contention_and_transient_jmap_failures_retry() {
+ let sqlite = |code| {
+ Error::Db(rusqlite::Error::SqliteFailure(
+ rusqlite::ffi::Error::new(code),
+ None,
+ ))
+ };
+ assert!(sqlite(rusqlite::ffi::SQLITE_BUSY).is_retryable());
+ assert!(sqlite(rusqlite::ffi::SQLITE_LOCKED).is_retryable());
+ assert!(!sqlite(rusqlite::ffi::SQLITE_CORRUPT).is_retryable());
+ assert!(!Error::Db(rusqlite::Error::QueryReturnedNoRows).is_retryable());
+
+ let method = |kind: &str| {
+ Error::Jmap(crate::jmap::JmapError::Method {
+ method: "Email/get",
+ kind: kind.into(),
+ description: None,
+ })
+ };
+ assert!(method("serverUnavailable").is_retryable());
+ assert!(!method("invalidArguments").is_retryable());
+
+ for err in [
+ Error::Io(std::io::Error::other("disk")),
+ Error::Unauthenticated,
+ Error::NotFound("thread"),
+ Error::Forbidden("read-only"),
+ Error::BadRequest("empty"),
+ ] {
+ assert!(!err.is_retryable(), "{err}");
+ }
+ }
}
diff --git a/src/jmap.rs b/src/jmap.rs
index fd885b7..23ebb19 100644
--- a/src/jmap.rs
+++ b/src/jmap.rs
@@ -57,6 +57,14 @@ pub enum JmapError {
detail: String,
},
+ /// A method-level error reply (RFC 8620 §3.6.2).
+ #[error("{method} failed: {kind}{}", described(.description))]
+ Method {
+ method: &'static str,
+ kind: String,
+ description: Option<String>,
+ },
+
#[error("{method} cannot calculate changes since the given state")]
CannotCalculateChanges { method: &'static str },
@@ -67,6 +75,43 @@ pub enum JmapError {
Unauthorized,
}
+impl JmapError {
+ /// Whether asking again could succeed. Everything else fails the
+ /// same way until someone changes the code, config, or token.
+ pub fn is_retryable(&self) -> bool {
+ match self {
+ JmapError::Http(err) => match err.status() {
+ Some(status) => {
+ status.is_server_error()
+ || status == reqwest::StatusCode::TOO_MANY_REQUESTS
+ || status == reqwest::StatusCode::REQUEST_TIMEOUT
+ }
+ None => err.is_timeout() || err.is_connect() || err.is_request() || err.is_body(),
+ },
+ // RFC 8620 §3.6.2's transient method errors.
+ JmapError::Method { kind, .. } => {
+ matches!(
+ kind.as_str(),
+ "serverUnavailable" | "serverFail" | "serverPartialFail"
+ )
+ }
+ JmapError::TokenFile(_)
+ | JmapError::Malformed { .. }
+ | JmapError::Reply { .. }
+ | JmapError::CannotCalculateChanges { .. }
+ | JmapError::NoMailAccount { .. }
+ | JmapError::Unauthorized => false,
+ }
+ }
+}
+
+fn described(description: &Option<String>) -> String {
+ description
+ .as_deref()
+ .map(|description| format!(": {description}"))
+ .unwrap_or_default()
+}
+
/// A token's session, as the server reports it (RFC 8620 §3). Only what
/// Docket derives rights from is kept.
#[derive(Debug, Clone, Deserialize)]
@@ -212,9 +257,10 @@ fn parse_reply(method: &'static str, json: &str) -> Result<Value, JmapError> {
Err(if err.kind == "cannotCalculateChanges" {
JmapError::CannotCalculateChanges { method }
} else {
- match err.description {
- Some(description) => detail(format!("{}: {description}", err.kind)),
- None => detail(err.kind),
+ JmapError::Method {
+ method,
+ kind: err.kind,
+ description: err.description,
}
})
}
diff --git a/tests/jmap.rs b/tests/jmap.rs
index e5e5dcc..19d9bc4 100644
--- a/tests/jmap.rs
+++ b/tests/jmap.rs
@@ -870,6 +870,56 @@ async fn a_rejected_token_fails_the_sync() {
assert!(store.accounts().unwrap().is_empty());
}
+#[tokio::test]
+async fn only_transient_http_failures_are_retryable() {
+ let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+ let addr = listener.local_addr().unwrap();
+ let app = Router::new()
+ .route(
+ "/status/{code}",
+ get(
+ |axum::extract::Path(code): axum::extract::Path<u16>| async move {
+ StatusCode::from_u16(code).unwrap()
+ },
+ ),
+ )
+ .route("/garbled", get(|| async { "{" }));
+ tokio::spawn(async move {
+ axum::serve(listener, app).await.unwrap();
+ });
+ let dir = tempfile::tempdir().unwrap();
+ let household = credential(&dir, "household", HOUSEHOLD_TOKEN);
+ let store = Store::open_in_memory(Clock::System).unwrap();
+ let retryable = async |url: String| {
+ Client::new(url)
+ .unwrap()
+ .sync_account(&household, &store)
+ .await
+ .unwrap_err()
+ .is_retryable()
+ };
+
+ for code in [500, 503, 429, 408] {
+ assert!(
+ retryable(format!("http://{addr}/status/{code}")).await,
+ "{code}"
+ );
+ }
+ for code in [401, 403, 404] {
+ assert!(
+ !retryable(format!("http://{addr}/status/{code}")).await,
+ "{code}"
+ );
+ }
+ assert!(!retryable(format!("http://{addr}/garbled")).await);
+
+ // Nothing listens on a port whose listener is gone.
+ let closed = TcpListener::bind("127.0.0.1:0").await.unwrap();
+ let gone = closed.local_addr().unwrap();
+ drop(closed);
+ assert!(retryable(format!("http://{gone}/session")).await);
+}
+
/// The imported message with the given Message-ID, if any.
fn the(store: &Store, message_id: &str) -> Option<docket::model::Message> {
store