fix(arr): cache the remote-command probe verdict
Reused for 5 minutes (#225) instead of running the configured command on every /api/health poll.
This commit is contained in:
+123
-29
@@ -21,8 +21,8 @@
|
||||
use std::{
|
||||
io::ErrorKind,
|
||||
process::Stdio,
|
||||
sync::{atomic::AtomicU64, atomic::Ordering, Arc},
|
||||
time::Duration,
|
||||
sync::{atomic::AtomicU64, atomic::Ordering, Arc, Mutex},
|
||||
time::{Duration, Instant},
|
||||
};
|
||||
|
||||
use serde::{Deserialize, Serialize};
|
||||
@@ -54,6 +54,15 @@ fn language_tag(language: &Language) -> &str {
|
||||
/// How long one command may run when the configuration does not say.
|
||||
pub const DEFAULT_TIMEOUT: Duration = Duration::from_secs(120);
|
||||
|
||||
/// How long a probe verdict is trusted before the next call runs the
|
||||
/// template again (#225). `/api/health` is polled every 15 seconds by the
|
||||
/// settings view, and unlike the other lamps this one is an arbitrary
|
||||
/// command — `ssh box claude -p` is the example DESIGN.md §15 gives —
|
||||
/// which turned "poll the lamp" into roughly 5,700 runs of that command a
|
||||
/// day. Serving a cached verdict is what keeps [`Backend::probe`]'s
|
||||
/// contract that a probe costs none of the quota translation spends.
|
||||
pub const PROBE_CACHE_TTL: Duration = Duration::from_secs(300);
|
||||
|
||||
/// 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
|
||||
@@ -79,6 +88,9 @@ pub struct Command {
|
||||
/// this backend to the settings row. Milliseconds so the tests — and an
|
||||
/// impatient operator — can go below one second.
|
||||
timeout_ms: Arc<AtomicU64>,
|
||||
/// The last probe verdict and when it ran, reused for [`PROBE_CACHE_TTL`]
|
||||
/// (#225).
|
||||
probe_cache: Arc<Mutex<Option<(Instant, ProbeOutcome)>>>,
|
||||
}
|
||||
|
||||
impl Command {
|
||||
@@ -100,6 +112,7 @@ impl Command {
|
||||
)),
|
||||
config,
|
||||
id: BackendId::new(BACKEND_NAME),
|
||||
probe_cache: Arc::new(Mutex::new(None)),
|
||||
})
|
||||
}
|
||||
|
||||
@@ -157,6 +170,44 @@ impl Command {
|
||||
}
|
||||
}
|
||||
|
||||
/// Runs the template once over an empty batch: the lamp is the command
|
||||
/// starting and exiting cleanly, not what it says. Uncached — callers go
|
||||
/// through [`Command::probe`], which is what applies [`PROBE_CACHE_TTL`].
|
||||
async fn run_probe(&self) -> Result<()> {
|
||||
let payload = format!(
|
||||
"{}\n[]\n",
|
||||
system_prompt(
|
||||
&Language::Other("en".to_owned()),
|
||||
&Language::PortuguesePortugal
|
||||
)
|
||||
);
|
||||
let mut tokens = argv(&self.config.template);
|
||||
for token in &mut tokens {
|
||||
*token = token
|
||||
.replace("{source}", language_tag(&Language::Other("en".to_owned())))
|
||||
.replace("{target}", language_tag(&Language::PortuguesePortugal));
|
||||
}
|
||||
let (program, args) = tokens.split_first().expect("the constructor rejects empty");
|
||||
|
||||
let timeout = Duration::from_millis(self.timeout_ms.load(Ordering::Relaxed));
|
||||
match time::timeout(timeout, run(program, args, payload, self.id.clone())).await {
|
||||
Err(_) => Err(Error::Transport {
|
||||
backend: self.id.clone(),
|
||||
source: format!("timed out after {timeout:?}").into(),
|
||||
}),
|
||||
Ok(Err(err)) => Err(err),
|
||||
Ok(Ok(_)) => Ok(()),
|
||||
}
|
||||
}
|
||||
|
||||
/// The cached verdict, if the last probe is still within
|
||||
/// [`PROBE_CACHE_TTL`] (#225).
|
||||
fn cached_probe(&self) -> Option<ProbeOutcome> {
|
||||
let cache = self.probe_cache.lock().expect("probe cache lock poisoned");
|
||||
let (at, outcome) = cache.as_ref()?;
|
||||
(at.elapsed() < PROBE_CACHE_TTL).then(|| outcome.clone())
|
||||
}
|
||||
|
||||
fn parse_reply(id: &BackendId, out: &[u8]) -> Result<Vec<TranslatedCue>> {
|
||||
let malformed = |detail: String| Error::Malformed {
|
||||
backend: id.clone(),
|
||||
@@ -211,39 +262,55 @@ impl Backend for Command {
|
||||
Box::pin(async move { self.translate_inner(batch).await })
|
||||
}
|
||||
|
||||
/// Runs the template once over an empty batch (#200): the lamp is the
|
||||
/// command starting and exiting cleanly, not what it says. The same
|
||||
/// timeout cell bounds it, so a wedged template cannot pile up probes.
|
||||
/// Runs the template once over an empty batch (#200), same timeout cell
|
||||
/// as a real batch so a wedged template cannot pile up probes — but only
|
||||
/// when the cache is stale. `/api/health` polls every 15 seconds and
|
||||
/// this backend's probe is a real command, so a cached verdict is served
|
||||
/// to every call inside [`PROBE_CACHE_TTL`] (#225).
|
||||
fn probe(&self) -> ProbeFuture<'_> {
|
||||
Box::pin(async move {
|
||||
let payload = format!(
|
||||
"{}\n[]\n",
|
||||
system_prompt(
|
||||
&Language::Other("en".to_owned()),
|
||||
&Language::PortuguesePortugal
|
||||
)
|
||||
);
|
||||
let mut tokens = argv(&self.config.template);
|
||||
for token in &mut tokens {
|
||||
*token = token
|
||||
.replace("{source}", language_tag(&Language::Other("en".to_owned())))
|
||||
.replace("{target}", language_tag(&Language::PortuguesePortugal));
|
||||
}
|
||||
let (program, args) = tokens.split_first().expect("the constructor rejects empty");
|
||||
|
||||
let timeout = Duration::from_millis(self.timeout_ms.load(Ordering::Relaxed));
|
||||
match time::timeout(timeout, run(program, args, payload, self.id.clone())).await {
|
||||
Err(_) => Err(Error::Transport {
|
||||
backend: self.id(),
|
||||
source: format!("timed out after {timeout:?}").into(),
|
||||
}),
|
||||
Ok(Err(err)) => Err(err),
|
||||
Ok(Ok(_)) => Ok(()),
|
||||
if let Some(outcome) = self.cached_probe() {
|
||||
return outcome.into_result(self.id.clone());
|
||||
}
|
||||
let result = self.run_probe().await;
|
||||
*self.probe_cache.lock().expect("probe cache lock poisoned") =
|
||||
Some((Instant::now(), ProbeOutcome::from_result(&result)));
|
||||
result
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
/// A probe verdict, cheap to store and clone — [`Error`] is not [`Clone`]
|
||||
/// because its transport source is a boxed trait object, and a cached
|
||||
/// verdict must survive past the call that produced it (#225).
|
||||
#[derive(Clone, Debug)]
|
||||
enum ProbeOutcome {
|
||||
Ok,
|
||||
Unauthorized,
|
||||
Failed(String),
|
||||
}
|
||||
|
||||
impl ProbeOutcome {
|
||||
fn from_result(result: &Result<()>) -> Self {
|
||||
match result {
|
||||
Ok(()) => Self::Ok,
|
||||
Err(Error::Unauthorized { .. }) => Self::Unauthorized,
|
||||
Err(err) => Self::Failed(err.to_string()),
|
||||
}
|
||||
}
|
||||
|
||||
fn into_result(self, backend: BackendId) -> Result<()> {
|
||||
match self {
|
||||
Self::Ok => Ok(()),
|
||||
Self::Unauthorized => Err(Error::Unauthorized { backend }),
|
||||
Self::Failed(detail) => Err(Error::Transport {
|
||||
backend,
|
||||
source: detail.into(),
|
||||
}),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Spawn the command once, feed it `payload`, and collect its stdout.
|
||||
///
|
||||
/// The child is killed when this future is dropped — including on timeout.
|
||||
@@ -330,7 +397,9 @@ async fn run(program: &str, args: &[String], payload: String, id: BackendId) ->
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::{argv, Command, CommandConfig};
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
use super::{argv, Command, CommandConfig, ProbeOutcome, PROBE_CACHE_TTL};
|
||||
use crate::translate::Error;
|
||||
|
||||
#[test]
|
||||
@@ -350,4 +419,29 @@ mod tests {
|
||||
assert_eq!(argv(" claude "), ["claude"]);
|
||||
assert!(argv(" ").is_empty());
|
||||
}
|
||||
|
||||
/// #225: a fresh verdict is served straight from the cache; a stale one
|
||||
/// is not. Back-dates the cache directly rather than sleeping — the real
|
||||
/// TTL is five minutes.
|
||||
#[test]
|
||||
fn a_cached_probe_expires_after_its_ttl() {
|
||||
let backend = Command::new(CommandConfig {
|
||||
template: "true".to_owned(),
|
||||
timeout: super::DEFAULT_TIMEOUT,
|
||||
})
|
||||
.expect("backend constructs");
|
||||
|
||||
let fresh = Instant::now();
|
||||
*backend.probe_cache.lock().expect("lock") = Some((fresh, ProbeOutcome::Ok));
|
||||
assert!(backend.cached_probe().is_some(), "a fresh entry is served");
|
||||
|
||||
let stale = fresh
|
||||
.checked_sub(PROBE_CACHE_TTL + Duration::from_secs(1))
|
||||
.expect("the test has been running long enough to back-date");
|
||||
*backend.probe_cache.lock().expect("lock") = Some((stale, ProbeOutcome::Ok));
|
||||
assert!(
|
||||
backend.cached_probe().is_none(),
|
||||
"a stale entry must not be served"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -40,7 +40,10 @@ pub mod sync;
|
||||
pub mod translate;
|
||||
|
||||
#[cfg(feature = "translate-command")]
|
||||
pub use command::{Command, CommandConfig, DEFAULT_TIMEOUT as COMMAND_DEFAULT_TIMEOUT};
|
||||
pub use command::{
|
||||
Command, CommandConfig, DEFAULT_TIMEOUT as COMMAND_DEFAULT_TIMEOUT,
|
||||
PROBE_CACHE_TTL as COMMAND_PROBE_CACHE_TTL,
|
||||
};
|
||||
#[cfg(feature = "translate-deepl")]
|
||||
pub use deepl::{DeepL, DeepLConfig};
|
||||
pub use error::{Error, Result};
|
||||
|
||||
@@ -156,7 +156,9 @@ pub trait Backend: fmt::Debug + Send + Sync {
|
||||
/// runs and exits cleanly. Cheap by contract — the endpoint carrying it
|
||||
/// is polled, and it must spend none of the quota translation spends.
|
||||
/// [`Error::Unauthorized`] means the credentials were refused; anything
|
||||
/// else is an outage.
|
||||
/// else is an outage. `Command` cannot make one run of an arbitrary
|
||||
/// command free, so it keeps this contract by caching the verdict
|
||||
/// (#225) instead.
|
||||
fn probe(&self) -> ProbeFuture<'_>;
|
||||
}
|
||||
|
||||
|
||||
@@ -292,3 +292,29 @@ async fn a_probe_runs_the_template_and_wants_a_clean_exit() {
|
||||
.expect_err("non-zero exit");
|
||||
assert!(matches!(error, Error::Transport { .. }), "got {error:?}");
|
||||
}
|
||||
|
||||
/// #225: `/api/health` polls every 15 seconds, and unlike the other lamps
|
||||
/// this one is a real command. A second call within the TTL must not spawn
|
||||
/// it again. TTL expiry itself is a unit test in `command.rs` — it needs
|
||||
/// [`Command`]'s private cache to back-date it, rather than a real sleep.
|
||||
#[tokio::test]
|
||||
async fn a_probe_within_the_ttl_is_not_rerun() {
|
||||
let dir = tempfile::tempdir().expect("tempdir");
|
||||
let counter = dir.path().join("count");
|
||||
let script = stub(
|
||||
dir.path(),
|
||||
"count.sh",
|
||||
&format!("cat > /dev/null\necho x >> {}\n", counter.display()),
|
||||
);
|
||||
let backend =
|
||||
Command::new(config(&script.display().to_string(), DEFAULT)).expect("backend constructs");
|
||||
|
||||
backend.probe().await.expect("first probe runs the command");
|
||||
backend.probe().await.expect("second probe hits the cache");
|
||||
|
||||
let runs = fs::read_to_string(&counter)
|
||||
.expect("counter written")
|
||||
.lines()
|
||||
.count();
|
||||
assert_eq!(runs, 1, "the cached call must not re-run the template");
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user