diff --git a/apps/ingest/src/aws.rs b/apps/ingest/src/aws.rs index d99885656..7ecb89a74 100644 --- a/apps/ingest/src/aws.rs +++ b/apps/ingest/src/aws.rs @@ -442,10 +442,16 @@ impl S3Client { /// the key not existing, which S3 answers with 412 when it does — the /// primitive orphan claiming is built on. pub async fn put(&self, key: &str, body: Vec, if_none_match: bool) -> Result<(), S3Error> { - let mut headers = vec![( - "content-type".to_owned(), - "application/octet-stream".to_owned(), - )]; + // Explicit because hyper omits `Content-Length: 0` for an empty body, and + // S3 answers a PUT without it with 411 — the heartbeat's empty PUT never + // landed, so no owner was ever claimable. + let mut headers = vec![ + ( + "content-type".to_owned(), + "application/octet-stream".to_owned(), + ), + ("content-length".to_owned(), body.len().to_string()), + ]; if if_none_match { headers.push(("if-none-match".to_owned(), "*".to_owned())); } diff --git a/apps/ingest/src/telemetry.rs b/apps/ingest/src/telemetry.rs index 57add0742..1f6c3a52c 100644 --- a/apps/ingest/src/telemetry.rs +++ b/apps/ingest/src/telemetry.rs @@ -6986,6 +6986,11 @@ mod tests { let mut objects = state.objects.lock().unwrap(); match method { axum::http::Method::PUT => { + // Real S3 refuses a PUT without Content-Length (411 + // MissingContentLength); accepting one hid the heartbeat bug. + if !headers.contains_key("content-length") { + return (StatusCode::LENGTH_REQUIRED, Vec::new()); + } if headers.contains_key("if-none-match") && objects.contains_key(&key) { return (StatusCode::PRECONDITION_FAILED, Vec::new()); } @@ -7209,6 +7214,27 @@ mod tests { ); } + #[tokio::test] + async fn a_heartbeat_lands_and_makes_its_owner_claimable_once_stale() { + // The heartbeat is an EMPTY PUT, the one request shape the fake used to + // accept and S3 refused. Without it no owner object exists, so a dead + // task is never seen as stale and nothing it shipped is ever recovered. + let (endpoint, bucket) = fake_s3::spawn("maple-wal-test").await; + let store = test_wal_store(&endpoint); + store.heartbeat().await.expect("heartbeat PUT succeeds"); + + let owner_key = format!("wal/v1/owners/{}", store.owner()); + assert!(bucket.get(&owner_key).is_some(), "heartbeat object exists"); + + let successor = test_wal_store(&endpoint); + let later = Utc::now() + chrono::Duration::hours(1); + assert_eq!( + successor.stale_owners(later).await.unwrap(), + vec![store.owner().to_owned()], + "a heartbeat that stops refreshing is found as a stale owner" + ); + } + /// Cross-language contract with the Prometheus scraper (apps/scraper). /// /// The scraper converts scraped exposition text into OTLP/JSON