fix(arr): re-read the command timeout every batch
This commit is contained in:
@@ -18,7 +18,12 @@
|
||||
//! exit, timeout and unparseable output fail loudly here, and a reply with
|
||||
//! the wrong cue count or numbering never becomes a subtitle.
|
||||
|
||||
use std::{io::ErrorKind, process::Stdio, time::Duration};
|
||||
use std::{
|
||||
io::ErrorKind,
|
||||
process::Stdio,
|
||||
sync::{atomic::AtomicU64, atomic::Ordering, Arc},
|
||||
time::Duration,
|
||||
};
|
||||
|
||||
use serde::{Deserialize, Serialize};
|
||||
use tokio::{
|
||||
@@ -52,7 +57,10 @@ pub const DEFAULT_TIMEOUT: Duration = Duration::from_secs(120);
|
||||
/// Settings for the remote-command backend (DESIGN.md §10, issue #198).
|
||||
///
|
||||
/// The template is not a credential, but it names a host — it arrives from
|
||||
/// bootstrap config like the API keys do, never from the database.
|
||||
/// bootstrap config like the API keys do, never from the database. The
|
||||
/// timeout is the *initial* value only: the running backend re-reads a
|
||||
/// shared cell per batch ([`Command::timeout_cell`]), so the operator's edit
|
||||
/// of `remote_command_timeout_seconds` applies without a restart (#219).
|
||||
#[derive(Clone, Debug, PartialEq, Eq)]
|
||||
pub struct CommandConfig {
|
||||
/// The command to run per batch, e.g. `ssh box claude -p`. May carry
|
||||
@@ -67,6 +75,10 @@ pub struct CommandConfig {
|
||||
pub struct Command {
|
||||
config: CommandConfig,
|
||||
id: BackendId,
|
||||
/// The live per-batch timeout in milliseconds, shared with whoever wired
|
||||
/// this backend to the settings row. Milliseconds so the tests — and an
|
||||
/// impatient operator — can go below one second.
|
||||
timeout_ms: Arc<AtomicU64>,
|
||||
}
|
||||
|
||||
impl Command {
|
||||
@@ -83,11 +95,25 @@ impl Command {
|
||||
});
|
||||
}
|
||||
Ok(Self {
|
||||
timeout_ms: Arc::new(AtomicU64::new(
|
||||
u64::try_from(config.timeout.as_millis()).unwrap_or(u64::MAX),
|
||||
)),
|
||||
config,
|
||||
id: BackendId::new(BACKEND_NAME),
|
||||
})
|
||||
}
|
||||
|
||||
/// The cell this backend re-reads for every batch, in milliseconds.
|
||||
///
|
||||
/// Whoever constructs the backend — the daemon, at startup — hands a
|
||||
/// clone of this to the settings API, which stores the row's
|
||||
/// `remote_command_timeout_seconds` into it on every edit. That is the
|
||||
/// path from the database to the running process #219 found missing.
|
||||
#[must_use]
|
||||
pub fn timeout_cell(&self) -> Arc<AtomicU64> {
|
||||
Arc::clone(&self.timeout_ms)
|
||||
}
|
||||
|
||||
async fn translate_inner(&self, batch: &Batch) -> Result<Vec<TranslatedCue>> {
|
||||
let request_cues: Vec<WireCue> = batch
|
||||
.cues
|
||||
@@ -114,14 +140,17 @@ impl Command {
|
||||
}
|
||||
let (program, args) = tokens.split_first().expect("the constructor rejects empty");
|
||||
|
||||
// Re-read per call: the settings API writes this cell when the
|
||||
// operator edits `remote_command_timeout_seconds` (#219).
|
||||
let timeout = Duration::from_millis(self.timeout_ms.load(Ordering::Relaxed));
|
||||
let work = run(program, args, payload, self.id.clone());
|
||||
|
||||
// Dropping `work` on timeout drops the child, whose `kill_on_drop`
|
||||
// ends the command — nothing is left resident (DESIGN.md §15).
|
||||
match time::timeout(self.config.timeout, work).await {
|
||||
match time::timeout(timeout, work).await {
|
||||
Err(_) => Err(Error::Transport {
|
||||
backend: self.id.clone(),
|
||||
source: format!("timed out after {:?}", self.config.timeout).into(),
|
||||
source: format!("timed out after {timeout:?}").into(),
|
||||
}),
|
||||
Ok(Err(err)) => Err(err),
|
||||
Ok(Ok(out)) => Self::parse_reply(&self.id, &out),
|
||||
|
||||
@@ -237,4 +237,36 @@ async fn a_hanging_command_is_killed_at_the_timeout() {
|
||||
}
|
||||
}
|
||||
|
||||
/// Issue #219: the timeout is not frozen at construction. The settings API
|
||||
/// stores the row's value into [`Command::timeout_cell`] on every edit, and
|
||||
/// the next batch must already run under it.
|
||||
#[tokio::test]
|
||||
async fn an_edited_timeout_reaches_the_next_call() {
|
||||
let dir = tempfile::tempdir().expect("tempdir");
|
||||
let script = stub(dir.path(), "hang.sh", "sleep 30\n");
|
||||
let backend =
|
||||
Command::new(config(&script.display().to_string(), DEFAULT)).expect("backend constructs");
|
||||
|
||||
// The edit that a PUT of remote_command_timeout_seconds performs.
|
||||
backend
|
||||
.timeout_cell()
|
||||
.store(150, std::sync::atomic::Ordering::Relaxed);
|
||||
|
||||
let error = translate(
|
||||
&backend,
|
||||
&cues(&["hello"]),
|
||||
&Language::Other("en".to_owned()),
|
||||
&Language::PortuguesePortugal,
|
||||
)
|
||||
.await
|
||||
.expect_err("the edited timeout must apply without a restart");
|
||||
|
||||
match error {
|
||||
Error::Transport { source, .. } => {
|
||||
assert!(source.to_string().contains("150ms"), "{source}");
|
||||
}
|
||||
other => panic!("expected Transport, got {other:?}"),
|
||||
}
|
||||
}
|
||||
|
||||
const DEFAULT: Duration = arr_subs::COMMAND_DEFAULT_TIMEOUT;
|
||||
|
||||
Reference in New Issue
Block a user