diff --git a/hive-agent/src/transcribe.rs b/hive-agent/src/transcribe.rs index 7fcb19dfc..fe154a839 100644 --- a/hive-agent/src/transcribe.rs +++ b/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()); - match tokio::time::timeout(stt::REPLY_DEADLINE, request).await { - Err(_) => err("the speech-to-text service did not answer in time"), - Ok(Err(e)) => err(request_failure(&e)), - Ok(Ok(reply)) => from_wire(&reply.payload), + // `client.request` would wait the connection's default 10 seconds, long + // before a queued or long memo is done; this request names its own wait. + match client + .send_request(stt::request_subject(&agent), request(bytes)) + .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) -> 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, String> { let meta = tokio::fs::metadata(path) @@ -50,17 +68,35 @@ async fn read_audio(path: &Path) -> Result, 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 { - if e.kind() == async_nats::client::RequestErrorKind::NoResponders { - "no speech-to-text service is running in this swarm" - } else { - "the speech-to-text service could not be reached" +fn request_failure(kind: RequestErrorKind) -> &'static str { + match kind { + RequestErrorKind::NoResponders => "no speech-to-text service is running in this swarm", + RequestErrorKind::TimedOut => "the speech-to-text service did not answer in time", + 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();