transcribe_audio: wait for the service's reply deadline, cap the read, say what each failure is

hive-agent/src/transcribe.rs

@@ -8,8 +8,10 @@
 
 use std::path::Path;
 
+use async_nats::client::RequestErrorKind;
 use hive_agent_sock::Response;
 use swarm_wire::stt;
+use tokio::io::AsyncReadExt as _;
 
 /// The largest file sent. The queue carries 8 MiB a message; the service
 /// refuses what it was configured not to take, so this only keeps the harness
@@ -27,14 +29,30 @@ pub async fn run(path: &str) -> Response {
         Err(message) => return err(message),
     };
     let client = crate::swarm_queue::client().await;
-    let request = client.request(stt::request_subject(&agent), bytes.into());
+    // `client.request` would wait the connection's default 10 seconds, long
-    match tokio::time::timeout(stt::REPLY_DEADLINE, request).await {
+    // before a queued or long memo is done; this request names its own wait.
-        Err(_) => err("the speech-to-text service did not answer in time"),
+    match client
-        Ok(Err(e)) => err(request_failure(&e)),
+        .send_request(stt::request_subject(&agent), request(bytes))
-        Ok(Ok(reply)) => from_wire(&reply.payload),
+        .await
+    {
+        Ok(reply) => from_wire(&reply.payload),
+        Err(e) => err(request_failure(e.kind())),
     }
 }
 
+/// How long a request waits for the answer: as long as the service may take,
+/// not the connection's default, which `async-nats` sets to 10 seconds.
+const fn reply_wait() -> std::time::Duration {
+    stt::REPLY_DEADLINE
+}
+
+/// The request for `bytes`, waiting [`reply_wait`] for the answer.
+fn request(bytes: Vec<u8>) -> async_nats::Request {
+    async_nats::Request::new()
+        .payload(bytes.into())
+        .timeout(Some(reply_wait()))
+}
+
 /// The file's bytes, if it is a regular file small enough to send.
 async fn read_audio(path: &Path) -> Result<Vec<u8>, String> {
     let meta = tokio::fs::metadata(path)
@@ -50,17 +68,35 @@ async fn read_audio(path: &Path) -> Result<Vec<u8>, String> {
             meta.len()
         ));
     }
-    tokio::fs::read(path)
+    // The size check above can go stale (the file may grow, or be a
+    // pseudo-file that reports nothing), so the read is capped as well.
+    let cannot = |e: std::io::Error| format!("cannot read {}: {e}", path.display());
+    let file = tokio::fs::File::open(path).await.map_err(cannot)?;
+    let mut bytes = Vec::new();
+    file.take(MAX_BYTES + 1)
+        .read_to_end(&mut bytes)
         .await
-        .map_err(|e| format!("cannot read {}: {e}", path.display()))
+        .map_err(cannot)?;
+    if bytes.len() as u64 > MAX_BYTES {
+        return Err(format!(
+            "{} is over the {MAX_BYTES} byte limit",
+            path.display()
+        ));
+    }
+    Ok(bytes)
 }
 
 /// What a failed publish-and-wait means to the agent.
-fn request_failure(e: &async_nats::RequestError) -> &'static str {
+fn request_failure(kind: RequestErrorKind) -> &'static str {
-    if e.kind() == async_nats::client::RequestErrorKind::NoResponders {
+    match kind {
-        "no speech-to-text service is running in this swarm"
+        RequestErrorKind::NoResponders => "no speech-to-text service is running in this swarm",
-    } else {
+        RequestErrorKind::TimedOut => "the speech-to-text service did not answer in time",
-        "the speech-to-text service could not be reached"
+        RequestErrorKind::MaxPayloadExceeded => {
+            "the audio is too large for the swarm queue to carry"
+        }
+        RequestErrorKind::InvalidSubject | RequestErrorKind::Other => {
+            "the speech-to-text service could not be reached"
+        }
     }
 }
 
@@ -98,6 +134,27 @@ mod tests {
         assert!(matches!(from_wire(b"nonsense"), Response::Err { .. }));
     }
 
+    #[test]
+    fn a_request_names_the_services_reply_deadline_not_the_connections_default() {
+        assert_eq!(reply_wait(), stt::REPLY_DEADLINE);
+        assert!(stt::REPLY_DEADLINE > std::time::Duration::from_secs(10));
+        // `request` builds with that wait; the request type has no getter, so
+        // this only pins that it builds.
+        let _ = request(vec![1, 2, 3]);
+    }
+
+    #[test]
+    fn each_failure_kind_says_what_it_is() {
+        for (kind, said) in [
+            (RequestErrorKind::NoResponders, "no speech-to-text service"),
+            (RequestErrorKind::TimedOut, "did not answer in time"),
+            (RequestErrorKind::MaxPayloadExceeded, "too large"),
+            (RequestErrorKind::Other, "could not be reached"),
+        ] {
+            assert!(request_failure(kind).contains(said), "{kind:?}");
+        }
+    }
+
     #[tokio::test]
     async fn only_a_small_regular_file_is_read() {
         let dir = tempfile::tempdir().unwrap();