Recover workflow wakes after interrupted authority bodies

Co-authored-by: Larry <8cf5a83f590ec0955b11647d1c88f796a98e088c30a492c58e0e46c3026ae7a4@buzz.block.builderlab.xyz>

Signed-off-by: Logan Johnson <loganj@squareup.com>
Signed-off-by: Larry <8cf5a83f590ec0955b11647d1c88f796a98e088c30a492c58e0e46c3026ae7a4@buzz.block.builderlab.xyz>
Signed-off-by: Logan Johnson <loganj@squareup.com>

crates/buzz-acp/src/relay.rs

@@ -522,11 +522,21 @@ impl RestClient {
         message_id: &nostr::EventId,
     ) -> Result<crate::workflow_wake::WorkflowWakeAuthority, RelayError> {
         let path = format!("/workflow-wakes/{run_id}/{}", message_id.to_hex());
-        self.bridge_get(&path)
-            .await?
-            .json()
-            .await
-            .map_err(|error| RelayError::Http(error.to_string()))
+        // A successful status is not a complete authority response. Read the
+        // body separately so an interrupted transfer remains recoverable while
+        // a complete but malformed authority document stays terminal.
+        let response = self.bridge_get(&path).await?;
+        let body = match response.bytes().await {
+            Ok(body) => body,
+            Err(error) => {
+                // Replay retries this read through the normal verification path.
+                // Pace body failures too: headers may have arrived immediately,
+                // bypassing request_with_retry's backoff entirely.
+                tokio::time::sleep(jittered_duration(REST_RETRY_BASE_DELAYS[0])).await;
+                return Err(RelayError::TransientHttp(error.to_string()));
+            }
+        };
+        serde_json::from_slice(&body).map_err(|error| RelayError::Http(error.to_string()))
     }
 
     /// Query events via the HTTP bridge: `POST /query` with NIP-98 auth.

crates/buzz-acp/src/workflow_wake_recovery_tests.rs

@@ -6,6 +6,27 @@ use tokio::io::{AsyncReadExt, AsyncWriteExt};
 
 #[tokio::test]
 async fn exhausted_authority_failure_replays_exact_wake_and_verifies_before_dispatch() {
+    assert_authority_recovery(AuthorityFailure::Status).await;
+}
+
+#[tokio::test]
+async fn truncated_authority_body_replays_before_dispatch() {
+    assert_authority_recovery(AuthorityFailure::TruncatedBody).await;
+}
+
+#[tokio::test]
+async fn stalled_authority_body_replays_before_dispatch() {
+    assert_authority_recovery(AuthorityFailure::StalledBody).await;
+}
+
+#[derive(Clone, Copy)]
+enum AuthorityFailure {
+    Status,
+    TruncatedBody,
+    StalledBody,
+}
+
+async fn assert_authority_recovery(failure: AuthorityFailure) {
     let agent = Keys::generate();
     let owner = Keys::generate();
     let relay_key = Keys::generate();
@@ -37,24 +58,44 @@ async fn exhausted_authority_failure_replays_exact_wake_and_verifies_before_disp
     .to_string();
     let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
     let address = listener.local_addr().unwrap();
-    // Four failures exhaust the entire bounded request budget; the fifth
-    // request is possible only after the transport has replayed the wake.
+    // Status failures exhaust the HTTP retry budget. Body failures arrive
+    // after successful headers and must independently reopen transport replay.
+    let failures = if matches!(failure, AuthorityFailure::Status) {
+        4
+    } else {
+        1
+    };
     let http_server = tokio::spawn(async move {
-        for index in 0..5 {
+        for index in 0..=failures {
             let (mut stream, _) = listener.accept().await.unwrap();
             let mut request = [0u8; 4096];
             let len = stream.read(&mut request).await.unwrap();
             let request = String::from_utf8_lossy(&request[..len]);
             assert!(request.starts_with(&format!("GET /workflow-wakes/{run}/")));
-            let (status, response) = if index < 4 {
-                (503, "")
+            if index < failures {
+                match failure {
+                    AuthorityFailure::Status => {
+                        stream.write_all(b"HTTP/1.1 503 test\r\nContent-Length: 0\r\nConnection: close\r\n\r\n").await.unwrap();
+                    }
+                    AuthorityFailure::TruncatedBody | AuthorityFailure::StalledBody => {
+                        stream.write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 1000\r\nConnection: close\r\n\r\n{").await.unwrap();
+                        if matches!(failure, AuthorityFailure::StalledBody) {
+                            tokio::spawn(async move {
+                                tokio::time::sleep(Duration::from_secs(5)).await;
+                                drop(stream);
+                            });
+                        }
+                    }
+                }
             } else {
-                (200, body.as_str())
-            };
-            stream.write_all(format!("HTTP/1.1 {status} test\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{response}", response.len()).as_bytes()).await.unwrap();
+                stream.write_all(format!("HTTP/1.1 200 OK\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}", body.len()).as_bytes()).await.unwrap();
+            }
         }
     });
-    let http = reqwest::Client::new();
+    let http = reqwest::Client::builder()
+        .timeout(Duration::from_secs(1))
+        .build()
+        .unwrap();
     let rest = RestClient {
         http: http.clone(),
         base_url: format!("http://{address}"),
@@ -112,7 +153,7 @@ async fn exhausted_authority_failure_replays_exact_wake_and_verifies_before_disp
     let error = rest
         .workflow_wake_authority(authenticated.run_id(), &authenticated.message_event_id())
         .await
-        .expect_err("all four requests fail");
+        .expect_err("authority transfer fails");
     assert!(error.is_transient());
     assert!(
         harness.event_rx.try_recv().is_err(),
@@ -180,3 +221,43 @@ async fn next_data_frame(server: &mut WebSocketStream<tokio::net::TcpStream>) ->
     .await
     .expect("data frame before timeout")
 }
+
+#[tokio::test]
+async fn complete_malformed_authority_body_is_terminal() {
+    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let address = listener.local_addr().unwrap();
+    let server = tokio::spawn(async move {
+        for body in ["{", "{}"] {
+            let (mut stream, _) = listener.accept().await.unwrap();
+            let mut request = [0u8; 4096];
+            stream.read(&mut request).await.unwrap();
+            stream
+                .write_all(
+                    format!(
+                        "HTTP/1.1 200 OK\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
+                        body.len()
+                    )
+                    .as_bytes(),
+                )
+                .await
+                .unwrap();
+        }
+    });
+    let rest = RestClient {
+        http: reqwest::Client::new(),
+        base_url: format!("http://{address}"),
+        keys: Keys::generate(),
+        auth_tag_json: None,
+    };
+    for _ in 0..2 {
+        let error = rest
+            .workflow_wake_authority(Uuid::new_v4(), &nostr::EventId::all_zeros())
+            .await
+            .unwrap_err();
+        assert!(
+            !error.is_transient(),
+            "complete malformed authority must not replay"
+        );
+    }
+    server.await.unwrap();
+}