chadtmiller.com/knotless
git that lives in your Atmosphere account
1
10
XRPC failures log their cause, and a query whose connection broke is asked again
.changeset/reads-survive-a-dropped-connection.mdnew+7−0
| @@ -0,0 +1,7 @@ | ||
--- | ||
default: patch | ||
--- | ||
| ||
# A read whose connection dropped is asked again | ||
| ||
A read from a PDS or the appview whose connection broke before it was answered, as one kept open that the server had since closed, is now asked again once instead of failing the command. Writes are not sent again, since they may have been done. | ||
crates/knotless/src/xrpc.rs+20−3
| @@ -184,7 +184,10 @@ impl Xrpc { | ||
Ok(out.blob) | ||
} | ||
| ||
/// Sends the request `make` builds, with `auth`'s headers. A DPoP | ||
/// Sends the request `make` builds, with `auth`'s headers. A query | ||
/// whose connection broke before it was answered, as one kept open | ||
/// that the server had closed does, is sent again once: reading again | ||
/// changes nothing, where a procedure may have been done. A DPoP | ||
/// request the server refuses for want of a fresh nonce is sent again | ||
/// once, with the nonce it named. | ||
async fn send( | ||
| @@ -198,9 +201,16 @@ impl Xrpc { | ||
let headers = auth.map(|a| a.headers(method, url)).unwrap_or_default(); | ||
telemetry::send(with(make(), &headers), resend) | ||
}; | ||
let res = attempt(0).await?; | ||
let mut resent = 0; | ||
let res = match attempt(resent).await { | ||
Err(e) if method == "GET" && broken(&e) => { | ||
resent += 1; | ||
attempt(resent).await? | ||
} | ||
res => res?, | ||
}; | ||
if matches!(auth, Some(Auth::Dpop { .. })) && oauth::nonce_refused(url, &res) { | ||
return Ok(attempt(1).await?); | ||
return Ok(attempt(resent + 1).await?); | ||
} | ||
Ok(res) | ||
} | ||
| @@ -238,6 +248,13 @@ fn with(mut req: reqwest::RequestBuilder, headers: &[(String, String)]) -> reqwe | ||
req | ||
} | ||
| ||
/// Whether `err` is a connection that broke once it was made, before an | ||
/// answer came: not one that could not be made, nor one that timed out, | ||
/// which sending again at once would only repeat. | ||
fn broken(err: &reqwest::Error) -> bool { | ||
err.is_request() && !err.is_connect() && !err.is_timeout() | ||
} | ||
| ||
async fn ok(res: reqwest::Response) -> Result<bytes::Bytes> { | ||
let status = res.status().as_u16(); | ||
let server = res.url().origin().ascii_serialization(); | ||
crates/knotless/tests/library.rs+75−0
| @@ -29,6 +29,24 @@ fn mock(rt: &tokio::runtime::Runtime, router: Router) -> String { | ||
addr | ||
} | ||
| ||
/// Serves `router` on a free port, but for the first connection, closed | ||
/// once its request is read, unanswered: as a server does that dropped a | ||
/// connection kept open. Returns `host:port`. | ||
fn mock_dropping_first(rt: &tokio::runtime::Runtime, router: Router) -> String { | ||
let listener = rt | ||
.block_on(tokio::net::TcpListener::bind("127.0.0.1:0")) | ||
.unwrap(); | ||
let addr = listener.local_addr().unwrap().to_string(); | ||
rt.spawn(async move { | ||
let (mut first, _) = listener.accept().await.unwrap(); | ||
let mut asked = [0u8; 1024]; | ||
let _ = tokio::io::AsyncReadExt::read(&mut first, &mut asked).await; | ||
drop(first); | ||
axum::serve(listener, router).await.unwrap() | ||
}); | ||
addr | ||
} | ||
| ||
async fn connect(ctx: &Ctx, url: &str) -> Remote { | ||
Remote::connect(ctx, &RepoUrl::parse(url).unwrap()) | ||
.await | ||
| @@ -275,6 +293,63 @@ fn xrpc_failures() { | ||
}); | ||
} | ||
| ||
#[test] | ||
fn broken_connections() { | ||
let rt = rt(); | ||
let answering = || { | ||
let answer = || async { axum::Json(json!({"ok": true})) }; | ||
Router::new().route("/xrpc/x.y", get(answer).post(answer)) | ||
}; | ||
let at = | ||
|addr: String| knotless::xrpc::Xrpc::new(reqwest::Client::new(), &format!("http://{addr}")); | ||
// A query whose connection broke is asked again, and answered. | ||
let x = at(mock_dropping_first(&rt, answering())); | ||
let read = rt.block_on(x.query::<serde_json::Value>("x.y", &[], None)); | ||
assert_eq!(read.unwrap()["ok"], true); | ||
// A procedure may have been done before its connection broke: it is | ||
// not sent again. | ||
let x = at(mock_dropping_first(&rt, answering())); | ||
let wrote = rt.block_on(x.procedure::<serde_json::Value>("x.y", None, None)); | ||
assert!(wrote.is_err()); | ||
} | ||
| ||
/// A query refused for want of a DPoP nonce is sent again with the nonce | ||
/// the server named, after a resend for a broken connection too. | ||
#[test] | ||
fn dpop_nonces() { | ||
use axum::response::IntoResponse; | ||
use base64::Engine as _; | ||
let rt = rt(); | ||
let answer = |headers: axum::http::HeaderMap| async move { | ||
let proof = headers.get("dpop").and_then(|v| v.to_str().ok()); | ||
let claims = proof.unwrap_or_default().split('.').nth(1); | ||
let claims = base64::engine::general_purpose::URL_SAFE_NO_PAD | ||
.decode(claims.unwrap_or_default()) | ||
.unwrap_or_default(); | ||
let claims: serde_json::Value = serde_json::from_slice(&claims).unwrap_or_default(); | ||
match claims["nonce"] == "fresh" { | ||
true => axum::Json(json!({"ok": true})).into_response(), | ||
false => ( | ||
axum::http::StatusCode::UNAUTHORIZED, | ||
[ | ||
("www-authenticate", r#"DPoP error="use_dpop_nonce""#), | ||
("dpop-nonce", "fresh"), | ||
], | ||
"{}", | ||
) | ||
.into_response(), | ||
} | ||
}; | ||
let addr = mock_dropping_first(&rt, Router::new().route("/xrpc/x.y", get(answer))); | ||
let x = knotless::xrpc::Xrpc::new(reqwest::Client::new(), &format!("http://{addr}")); | ||
let auth = knotless::xrpc::Auth::Dpop { | ||
token: "t".into(), | ||
key: std::sync::Arc::new(knotless::crypto::P256Key::generate()), | ||
}; | ||
let read = rt.block_on(x.query::<serde_json::Value>("x.y", &[], Some(&auth))); | ||
assert_eq!(read.unwrap()["ok"], true); | ||
} | ||
| ||
#[test] | ||
fn session_refresh_and_fallbacks() { | ||
let mut s = Sandbox::new(); | ||
crates/knotless-serve/src/web.rs+35−2
| @@ -903,7 +903,8 @@ pub async fn origin(req: Request, next: Next) -> Response { | ||
} | ||
| ||
/// A failure an XRPC method answered as text, as XRPC has one instead: | ||
/// its kind, by name, and why. | ||
/// its kind, by name, and why. Its extensions go with it, the cause its | ||
/// request's log line gives among them. | ||
pub async fn xrpc_failures(req: Request, next: Next) -> Response { | ||
let xrpc = req.uri().path().starts_with("/xrpc/"); | ||
let res = next.run(req).await; | ||
| @@ -920,7 +921,9 @@ pub async fn xrpc_failures(req: Request, next: Next) -> Response { | ||
error: failure_name(parts.status).into(), | ||
message: String::from_utf8_lossy(&said).into_owned(), | ||
}; | ||
(parts.status, axum::Json(failure)).into_response() | ||
let mut answer = (parts.status, axum::Json(failure)).into_response(); | ||
*answer.extensions_mut() = parts.extensions; | ||
answer | ||
} | ||
| ||
fn origin_of(headers: &HeaderMap) -> String { | ||
| @@ -4218,6 +4221,36 @@ mod tests { | ||
); | ||
} | ||
| ||
/// A failure answered as text is answered as XRPC's, and its request's | ||
/// log line still reads why it failed. | ||
#[tokio::test] | ||
async fn xrpc_failures_keep_their_cause() { | ||
let seen = Arc::new(Mutex::new(None)); | ||
let kept = seen.clone(); | ||
let failing = || async { AppError::internal(anyhow::anyhow!("disk on fire")) }; | ||
let router = Router::new() | ||
.route("/xrpc/x.y", get(failing)) | ||
.layer(middleware::from_fn(xrpc_failures)) | ||
.layer(middleware::from_fn(move |req: Request, next: Next| { | ||
let kept = kept.clone(); | ||
async move { | ||
let res = next.run(req).await; | ||
let cause = res.extensions().get::<crate::telemetry::Cause>(); | ||
*kept.lock().unwrap() = cause.map(|c| c.0.clone()); | ||
res | ||
} | ||
})); | ||
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); | ||
let at = format!("http://{}/xrpc/x.y", listener.local_addr().unwrap()); | ||
tokio::spawn(async move { axum::serve(listener, router).await.unwrap() }); | ||
let res = reqwest::get(at).await.unwrap(); | ||
assert_eq!(res.status(), StatusCode::INTERNAL_SERVER_ERROR); | ||
let body: serde_json::Value = res.json().await.unwrap(); | ||
assert_eq!(body["error"], "InternalServerError"); | ||
let cause = seen.lock().unwrap().clone().expect("the cause is kept"); | ||
assert_eq!(body["message"], cause.as_str()); | ||
} | ||
| ||
#[tokio::test] | ||
async fn addresses_in_full() { | ||
let mut headers = HeaderMap::new(); | ||
docs/observability.md+4−1
| @@ -41,7 +41,10 @@ to read it is in [deploy/README.md](../deploy/README.md#logs). | ||
a handle's host, from the one place every call goes through | ||
(`knotless::telemetry::send`): host, path, XRPC method, status, error, | ||
resend count, duration. At debug, so a busy minute does not drown the | ||
log; failed and slow ones at warn. | ||
log; failed and slow ones at warn. An XRPC query whose connection broke | ||
before it was answered, as one kept open that the server had closed | ||
does, is sent again once, its line with a resend count of 1; a | ||
procedure, which may have been done, is not. | ||
- **A line for each slow SQL statement**: one of the index's that took | ||
a tenth of `KNOTLESS_SLOW_MS` (100 ms) or more, at warn, with its SQL | ||
as written (`db.query.text`, its parameters unbound, so no value it | ||