fix(manager-core): F-Q-009 promote dispatcher tick + async-exec timeout to env-overridable
TICK_INTERVAL (100ms) and ASYNC_EXEC_TIMEOUT (300s) were `const`. Every other timing knob in the file (cron tick, queue reclaim, retry policy) is env-overridable via TriggerConfig::from_env. Operators on a constrained Pi or a busier instance could tune retries but not dispatcher cadence. Add env knobs: - PICLOUD_DISPATCHER_TICK_INTERVAL_MS (default 100) - PICLOUD_DISPATCHER_ASYNC_EXEC_TIMEOUT_SEC (default 300) Read via tick_interval_from_env() / async_exec_timeout_from_env() at dispatcher startup. Invalid values fall back with a tracing-warn. AUDIT.md anchor: F-Q-009. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -70,14 +70,46 @@ const CLAIM_BATCH: i64 = 8;
|
|||||||
|
|
||||||
/// Polling cadence. Short enough that fan-out feels instant; long
|
/// Polling cadence. Short enough that fan-out feels instant; long
|
||||||
/// enough that an idle dispatcher doesn't burn cycles.
|
/// enough that an idle dispatcher doesn't burn cycles.
|
||||||
const TICK_INTERVAL: Duration = Duration::from_millis(100);
|
/// F-Q-009: env-overridable via `PICLOUD_DISPATCHER_TICK_INTERVAL_MS`.
|
||||||
|
const DEFAULT_TICK_INTERVAL: Duration = Duration::from_millis(100);
|
||||||
|
|
||||||
/// Hard cap on the wall-clock budget passed to the executor for an
|
/// Hard cap on the wall-clock budget passed to the executor for an
|
||||||
/// async-dispatched script. Sync HTTP gets a per-script timeout via
|
/// async-dispatched script. Sync HTTP gets a per-script timeout via
|
||||||
/// the orchestrator path; async rows don't have one, so we apply a
|
/// the orchestrator path; async rows don't have one, so we apply a
|
||||||
/// platform-wide ceiling here. Matches `LocalExecutorClient`'s own
|
/// platform-wide ceiling here. Matches `LocalExecutorClient`'s own
|
||||||
/// 5-minute cap.
|
/// 5-minute cap.
|
||||||
const ASYNC_EXEC_TIMEOUT: Duration = Duration::from_secs(300);
|
/// F-Q-009: env-overridable via `PICLOUD_DISPATCHER_ASYNC_EXEC_TIMEOUT_SEC`.
|
||||||
|
const DEFAULT_ASYNC_EXEC_TIMEOUT: Duration = Duration::from_secs(300);
|
||||||
|
|
||||||
|
/// Read `PICLOUD_DISPATCHER_TICK_INTERVAL_MS`; invalid values fall back
|
||||||
|
/// to the documented default with a tracing-warn.
|
||||||
|
fn tick_interval_from_env() -> Duration {
|
||||||
|
if let Ok(v) = std::env::var("PICLOUD_DISPATCHER_TICK_INTERVAL_MS") {
|
||||||
|
match v.trim().parse::<u64>() {
|
||||||
|
Ok(n) if n > 0 => return Duration::from_millis(n),
|
||||||
|
_ => tracing::warn!(
|
||||||
|
value = %v,
|
||||||
|
"ignoring invalid PICLOUD_DISPATCHER_TICK_INTERVAL_MS (want a positive integer)"
|
||||||
|
),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
DEFAULT_TICK_INTERVAL
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Read `PICLOUD_DISPATCHER_ASYNC_EXEC_TIMEOUT_SEC`; invalid values
|
||||||
|
/// fall back to the documented default with a tracing-warn.
|
||||||
|
fn async_exec_timeout_from_env() -> Duration {
|
||||||
|
if let Ok(v) = std::env::var("PICLOUD_DISPATCHER_ASYNC_EXEC_TIMEOUT_SEC") {
|
||||||
|
match v.trim().parse::<u64>() {
|
||||||
|
Ok(n) if n > 0 => return Duration::from_secs(n),
|
||||||
|
_ => tracing::warn!(
|
||||||
|
value = %v,
|
||||||
|
"ignoring invalid PICLOUD_DISPATCHER_ASYNC_EXEC_TIMEOUT_SEC (want a positive integer)"
|
||||||
|
),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
DEFAULT_ASYNC_EXEC_TIMEOUT
|
||||||
|
}
|
||||||
|
|
||||||
/// F-P-007: per-tick concurrency cap for queue dispatches. Matches the
|
/// F-P-007: per-tick concurrency cap for queue dispatches. Matches the
|
||||||
/// default execution-gate concurrency so the gate is what bounds real
|
/// default execution-gate concurrency so the gate is what bounds real
|
||||||
@@ -114,7 +146,7 @@ impl Dispatcher {
|
|||||||
}
|
}
|
||||||
|
|
||||||
async fn run(self) {
|
async fn run(self) {
|
||||||
let mut ticker = tokio::time::interval(TICK_INTERVAL);
|
let mut ticker = tokio::time::interval(tick_interval_from_env());
|
||||||
// Skip the immediate first fire so we don't race startup.
|
// Skip the immediate first fire so we don't race startup.
|
||||||
ticker.tick().await;
|
ticker.tick().await;
|
||||||
loop {
|
loop {
|
||||||
@@ -279,7 +311,7 @@ impl Dispatcher {
|
|||||||
};
|
};
|
||||||
let outcome = self
|
let outcome = self
|
||||||
.executor
|
.executor
|
||||||
.execute_with_identity(identity, &script.source, exec_req, ASYNC_EXEC_TIMEOUT)
|
.execute_with_identity(identity, &script.source, exec_req, async_exec_timeout_from_env())
|
||||||
.await;
|
.await;
|
||||||
drop(permit);
|
drop(permit);
|
||||||
|
|
||||||
@@ -446,7 +478,7 @@ impl Dispatcher {
|
|||||||
};
|
};
|
||||||
let outcome = self
|
let outcome = self
|
||||||
.executor
|
.executor
|
||||||
.execute_with_identity(identity, &source, exec_req, ASYNC_EXEC_TIMEOUT)
|
.execute_with_identity(identity, &source, exec_req, async_exec_timeout_from_env())
|
||||||
.await;
|
.await;
|
||||||
drop(permit);
|
drop(permit);
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user