fix(ingest): send Content-Length on S3 PUTs so the WAL heartbeat lands

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<u8>, 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()));
         }

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