fix(analysis): harden readiness gate and vision-manager per self-review
Backend: - Log on readiness state transitions (ready<->not-ready) so a misconfigured ANALYSIS_VISION_HEALTH_URL, which otherwise parks the worker silently forever, is diagnosable. - Add unit tests for HttpVisionReadiness: 2xx->ready, non-2xx->not-ready, connection error->not-ready (the production status mapping had no test). vision-manager: - Guard the backlog count against non-numeric output before `-gt`, so a stray value can't exit-2 and kill the loop under `set -e`. - Throttle the crawl-mutex "deferring start" log to once per episode. - Only reset the idle/uptime timers when `docker stop` actually succeeds, so a failed stop retries next tick instead of waiting a full debounce window. - Decouple the per-probe curl timeout (HEALTH_TIMEOUT) from the warm-up poll cadence. Docs: - Correct the docker-socket-proxy comment: CONTAINERS+POST permits the full container lifecycle (not just start/stop); state the real trust boundary. - Document that the externally-defined mangalord-vision container must share the compose network for name resolution. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -16,6 +16,15 @@ container lifecycle: start it when work appears, stop it when work is gone.
|
|||||||
> still-loading vision never burns a job's retries or writes a `failed` row
|
> still-loading vision never burns a job's retries or writes a `failed` row
|
||||||
> (the gotcha called out below). Configure via the `ai`-profile vars in
|
> (the gotcha called out below). Configure via the `ai`-profile vars in
|
||||||
> [.env.example](.env.example).
|
> [.env.example](.env.example).
|
||||||
|
>
|
||||||
|
> **Operator prerequisite:** the `mangalord-vision` container is defined
|
||||||
|
> *outside* this compose project (the manager only drives it by name). For the
|
||||||
|
> manager and backend to resolve it by name for the `/health` probe, attach it
|
||||||
|
> to this project's default network (`<project>_default`, e.g.
|
||||||
|
> `mangalord_default`) — or point `VISION_HEALTH_URL` /
|
||||||
|
> `ANALYSIS_VISION_HEALTH_URL` at an address that resolves. If the container is
|
||||||
|
> unreachable the manager silently treats it as "not running / not ready" and
|
||||||
|
> will loop trying to start a container it cannot see.
|
||||||
|
|
||||||
## Chosen approach — Option 2: a "vision-manager" sidecar
|
## Chosen approach — Option 2: a "vision-manager" sidecar
|
||||||
|
|
||||||
|
|||||||
@@ -132,6 +132,9 @@ struct WorkerContext {
|
|||||||
|
|
||||||
impl WorkerContext {
|
impl WorkerContext {
|
||||||
async fn run(self) {
|
async fn run(self) {
|
||||||
|
// Last observed readiness, so we log only on transitions (not every
|
||||||
|
// poll). `None` until the first probe.
|
||||||
|
let mut was_ready: Option<bool> = None;
|
||||||
loop {
|
loop {
|
||||||
if self.cancel.is_cancelled() {
|
if self.cancel.is_cancelled() {
|
||||||
tracing::info!(worker = self.id, "analysis worker: shutdown");
|
tracing::info!(worker = self.id, "analysis worker: shutdown");
|
||||||
@@ -142,7 +145,21 @@ impl WorkerContext {
|
|||||||
// costs a job its retries. Probe *before* the lease — leasing
|
// costs a job its retries. Probe *before* the lease — leasing
|
||||||
// increments `attempts` in SQL and can't be undone.
|
// increments `attempts` in SQL and can't be undone.
|
||||||
if let Some(readiness) = &self.readiness {
|
if let Some(readiness) = &self.readiness {
|
||||||
if !readiness.ready().await {
|
let ready = readiness.ready().await;
|
||||||
|
// Log on flip so a typo'd health URL (which parks the worker
|
||||||
|
// forever) is diagnosable, without spamming every poll.
|
||||||
|
if was_ready != Some(ready) {
|
||||||
|
if ready {
|
||||||
|
tracing::info!(worker = self.id, "analysis worker: vision ready — resuming");
|
||||||
|
} else {
|
||||||
|
tracing::warn!(
|
||||||
|
worker = self.id,
|
||||||
|
"analysis worker: vision not ready — parking until /health is 2xx"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
was_ready = Some(ready);
|
||||||
|
}
|
||||||
|
if !ready {
|
||||||
if self.sleep_or_cancel(READINESS_POLL).await {
|
if self.sleep_or_cancel(READINESS_POLL).await {
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
@@ -380,3 +397,56 @@ pub mod test_support {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[cfg(test)]
|
||||||
|
mod tests {
|
||||||
|
use super::*;
|
||||||
|
use tokio::io::{AsyncReadExt, AsyncWriteExt};
|
||||||
|
use tokio::net::TcpListener;
|
||||||
|
|
||||||
|
/// Bind an ephemeral port that answers every request with `status_line`
|
||||||
|
/// (e.g. `"200 OK"`), and return its `/health` URL. The probe under test
|
||||||
|
/// only inspects the status code, so a zero-length body is enough.
|
||||||
|
async fn serve(status_line: &'static str) -> String {
|
||||||
|
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
|
||||||
|
let addr = listener.local_addr().unwrap();
|
||||||
|
tokio::spawn(async move {
|
||||||
|
while let Ok((mut sock, _)) = listener.accept().await {
|
||||||
|
let mut buf = [0u8; 1024];
|
||||||
|
let _ = sock.read(&mut buf).await;
|
||||||
|
let resp = format!(
|
||||||
|
"HTTP/1.1 {status_line}\r\nContent-Length: 0\r\nConnection: close\r\n\r\n"
|
||||||
|
);
|
||||||
|
let _ = sock.write_all(resp.as_bytes()).await;
|
||||||
|
}
|
||||||
|
});
|
||||||
|
format!("http://{addr}/health")
|
||||||
|
}
|
||||||
|
|
||||||
|
fn probe(url: String) -> HttpVisionReadiness {
|
||||||
|
HttpVisionReadiness {
|
||||||
|
http: reqwest::Client::builder()
|
||||||
|
.timeout(Duration::from_millis(500))
|
||||||
|
.build()
|
||||||
|
.unwrap(),
|
||||||
|
health_url: url,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn http_readiness_2xx_is_ready() {
|
||||||
|
assert!(probe(serve("200 OK").await).ready().await);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn http_readiness_non_2xx_is_not_ready() {
|
||||||
|
// llama-server answers 503 while the model is still loading.
|
||||||
|
assert!(!probe(serve("503 Service Unavailable").await).ready().await);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn http_readiness_connection_error_is_not_ready() {
|
||||||
|
// Nothing listening (vision stopped) → not ready, never an error.
|
||||||
|
assert!(!probe("http://127.0.0.1:1/health".to_string()).ready().await);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
@@ -126,16 +126,22 @@ services:
|
|||||||
# The vision container itself is NOT defined here (it lives elsewhere on the
|
# The vision container itself is NOT defined here (it lives elsewhere on the
|
||||||
# host); the manager drives it by name. See VISION-AUTOSCALE.md.
|
# host); the manager drives it by name. See VISION-AUTOSCALE.md.
|
||||||
|
|
||||||
# Scoped Docker access for the manager: exposes ONLY the /containers API and
|
# Scoped Docker access for the manager. The proxy gates by API *section*
|
||||||
# write methods (start/stop) over TCP on an internal-only network. The raw
|
# (not per-method), so CONTAINERS=1 + POST=1 permits the full /containers
|
||||||
# host socket is mounted HERE and nowhere else — never on the backend.
|
# lifecycle — inspect/start/stop, but also create/kill/restart/update/
|
||||||
|
# rename/remove. It does NOT expose exec, images, volumes, networks, swarm,
|
||||||
|
# etc. (all default-denied). The trust boundary is therefore: (a) this is an
|
||||||
|
# internal-only network reachable solely by vision-manager, and (b) the raw
|
||||||
|
# host socket is mounted HERE and nowhere else — never on the backend. A
|
||||||
|
# backend RCE still cannot reach the Docker API. If you need true start/stop-
|
||||||
|
# only granularity, front the socket with an allow-list reverse proxy instead.
|
||||||
docker-socket-proxy:
|
docker-socket-proxy:
|
||||||
image: tecnativa/docker-socket-proxy:latest
|
image: tecnativa/docker-socket-proxy:latest
|
||||||
profiles: ["ai"]
|
profiles: ["ai"]
|
||||||
environment:
|
environment:
|
||||||
CONTAINERS: 1 # allow /containers/* (inspect, start, stop)
|
CONTAINERS: 1 # allow the /containers/* section
|
||||||
POST: 1 # allow write methods (start/stop are POSTs)
|
POST: 1 # allow write methods (start/stop are POSTs)
|
||||||
# Everything else stays denied (images, exec, networks, volumes, ...).
|
# Everything else stays at its default-deny (EXEC, IMAGES, NETWORKS, ...).
|
||||||
volumes:
|
volumes:
|
||||||
- /var/run/docker.sock:/var/run/docker.sock:ro
|
- /var/run/docker.sock:/var/run/docker.sock:ro
|
||||||
networks:
|
networks:
|
||||||
|
|||||||
@@ -24,6 +24,7 @@ POLL_INTERVAL="${POLL_INTERVAL:-20}" # seconds between backlog checks
|
|||||||
STOP_DEBOUNCE="${STOP_DEBOUNCE:-600}" # idle seconds before stopping vision
|
STOP_DEBOUNCE="${STOP_DEBOUNCE:-600}" # idle seconds before stopping vision
|
||||||
START_HEALTH_TIMEOUT="${START_HEALTH_TIMEOUT:-300}" # max wait for /health 200
|
START_HEALTH_TIMEOUT="${START_HEALTH_TIMEOUT:-300}" # max wait for /health 200
|
||||||
HEALTH_POLL_INTERVAL="${HEALTH_POLL_INTERVAL:-5}" # poll cadence while warming
|
HEALTH_POLL_INTERVAL="${HEALTH_POLL_INTERVAL:-5}" # poll cadence while warming
|
||||||
|
HEALTH_TIMEOUT="${HEALTH_TIMEOUT:-5}" # per-probe curl timeout, seconds
|
||||||
MAX_UPTIME="${MAX_UPTIME:-0}" # >0: force-stop after N idle-or-not seconds running (backstop); 0 disables
|
MAX_UPTIME="${MAX_UPTIME:-0}" # >0: force-stop after N idle-or-not seconds running (backstop); 0 disables
|
||||||
# When >0, refuse to start vision while a crawl is running (Chromium + vision
|
# When >0, refuse to start vision while a crawl is running (Chromium + vision
|
||||||
# together OOM the 8 GiB box). Set to 0 on roomier hosts.
|
# together OOM the 8 GiB box). Set to 0 on roomier hosts.
|
||||||
@@ -60,7 +61,7 @@ vision_running() {
|
|||||||
}
|
}
|
||||||
|
|
||||||
vision_ready() {
|
vision_ready() {
|
||||||
curl -fsS -m "$HEALTH_POLL_INTERVAL" -o /dev/null "$VISION_HEALTH_URL" 2>/dev/null
|
curl -fsS -m "$HEALTH_TIMEOUT" -o /dev/null "$VISION_HEALTH_URL" 2>/dev/null
|
||||||
}
|
}
|
||||||
|
|
||||||
start_vision() {
|
start_vision() {
|
||||||
@@ -84,24 +85,34 @@ start_vision() {
|
|||||||
|
|
||||||
stop_vision() {
|
stop_vision() {
|
||||||
log "stopping $VISION_CONTAINER (idle ${STOP_DEBOUNCE}s)"
|
log "stopping $VISION_CONTAINER (idle ${STOP_DEBOUNCE}s)"
|
||||||
docker stop "$VISION_CONTAINER" >/dev/null 2>&1 || log "WARNING: docker stop failed"
|
if docker stop "$VISION_CONTAINER" >/dev/null 2>&1; then
|
||||||
|
return 0
|
||||||
|
fi
|
||||||
|
log "WARNING: docker stop failed; will retry next tick"
|
||||||
|
return 1
|
||||||
}
|
}
|
||||||
|
|
||||||
# ---- Main loop -------------------------------------------------------------
|
# ---- Main loop -------------------------------------------------------------
|
||||||
log "started: container=$VISION_CONTAINER health=$VISION_HEALTH_URL poll=${POLL_INTERVAL}s debounce=${STOP_DEBOUNCE}s docker_host=$DOCKER_HOST"
|
log "started: container=$VISION_CONTAINER health=$VISION_HEALTH_URL poll=${POLL_INTERVAL}s debounce=${STOP_DEBOUNCE}s docker_host=$DOCKER_HOST"
|
||||||
|
|
||||||
idle_for=0 # seconds the queue has been empty while vision is running
|
idle_for=0 # seconds the queue has been empty while vision is running
|
||||||
up_for=0 # seconds vision has been running (for MAX_UPTIME backstop)
|
up_for=0 # seconds vision has been running (for MAX_UPTIME backstop)
|
||||||
|
crawl_deferred=0 # 1 while a start is held off by the crawl mutex (log on flip only)
|
||||||
|
|
||||||
while true; do
|
while true; do
|
||||||
pending="$(pending_analysis)"
|
pending="$(pending_analysis)"
|
||||||
running="$(vision_running)"
|
running="$(vision_running)"
|
||||||
|
|
||||||
if [ "$pending" = "ERR" ]; then
|
# Treat anything non-numeric (the "ERR" sentinel, or a stray psql notice on
|
||||||
log "WARNING: backlog query failed; will retry next tick"
|
# stdout) as "query failed" — never feed it to `-gt`, which under
|
||||||
sleep "$POLL_INTERVAL"
|
# `set -e` would otherwise exit-2 and kill the loop.
|
||||||
continue
|
case "$pending" in
|
||||||
fi
|
''|*[!0-9]*)
|
||||||
|
log "WARNING: backlog query failed (got '${pending}'); will retry next tick"
|
||||||
|
sleep "$POLL_INTERVAL"
|
||||||
|
continue
|
||||||
|
;;
|
||||||
|
esac
|
||||||
|
|
||||||
if [ "$running" = "true" ]; then
|
if [ "$running" = "true" ]; then
|
||||||
up_for=$((up_for + POLL_INTERVAL))
|
up_for=$((up_for + POLL_INTERVAL))
|
||||||
@@ -113,21 +124,34 @@ while true; do
|
|||||||
idle_for=0
|
idle_for=0
|
||||||
if [ "$running" != "true" ]; then
|
if [ "$running" != "true" ]; then
|
||||||
if [ "$RESPECT_CRAWL_MUTEX" != "0" ] && [ "$(crawl_running)" != "0" ]; then
|
if [ "$RESPECT_CRAWL_MUTEX" != "0" ] && [ "$(crawl_running)" != "0" ]; then
|
||||||
log "deferring start: $pending analysis job(s) pending but a crawl is running (RAM mutex)"
|
# Log once per deferral episode, not every poll, so a long crawl
|
||||||
|
# doesn't flood the log.
|
||||||
|
if [ "$crawl_deferred" != "1" ]; then
|
||||||
|
log "deferring start: $pending analysis job(s) pending but a crawl is running (RAM mutex)"
|
||||||
|
crawl_deferred=1
|
||||||
|
fi
|
||||||
else
|
else
|
||||||
|
crawl_deferred=0
|
||||||
start_vision || true
|
start_vision || true
|
||||||
fi
|
fi
|
||||||
|
else
|
||||||
|
crawl_deferred=0
|
||||||
fi
|
fi
|
||||||
else
|
else
|
||||||
|
crawl_deferred=0
|
||||||
# No work. Debounce the stop so bursty enqueues don't thrash the load
|
# No work. Debounce the stop so bursty enqueues don't thrash the load
|
||||||
# cycle, and own the stop on our OWN timer (never wait for a backend
|
# cycle, and own the stop on our OWN timer (never wait for a backend
|
||||||
# "drained" signal — leak safety).
|
# "drained" signal — leak safety).
|
||||||
if [ "$running" = "true" ]; then
|
if [ "$running" = "true" ]; then
|
||||||
idle_for=$((idle_for + POLL_INTERVAL))
|
idle_for=$((idle_for + POLL_INTERVAL))
|
||||||
if [ "$idle_for" -ge "$STOP_DEBOUNCE" ]; then
|
if [ "$idle_for" -ge "$STOP_DEBOUNCE" ]; then
|
||||||
stop_vision
|
# Only reset the timers if the stop actually took; otherwise let them
|
||||||
idle_for=0
|
# keep counting so we retry on the next tick rather than waiting out
|
||||||
up_for=0
|
# another full debounce window.
|
||||||
|
if stop_vision; then
|
||||||
|
idle_for=0
|
||||||
|
up_for=0
|
||||||
|
fi
|
||||||
fi
|
fi
|
||||||
else
|
else
|
||||||
idle_for=0
|
idle_for=0
|
||||||
@@ -137,9 +161,10 @@ while true; do
|
|||||||
# Backstop: bound how long vision can stay up regardless of debounce state.
|
# Backstop: bound how long vision can stay up regardless of debounce state.
|
||||||
if [ "$MAX_UPTIME" -gt 0 ] && [ "$running" = "true" ] && [ "$up_for" -ge "$MAX_UPTIME" ]; then
|
if [ "$MAX_UPTIME" -gt 0 ] && [ "$running" = "true" ] && [ "$up_for" -ge "$MAX_UPTIME" ]; then
|
||||||
log "MAX_UPTIME ${MAX_UPTIME}s reached; force-stopping $VISION_CONTAINER"
|
log "MAX_UPTIME ${MAX_UPTIME}s reached; force-stopping $VISION_CONTAINER"
|
||||||
stop_vision
|
if stop_vision; then
|
||||||
idle_for=0
|
idle_for=0
|
||||||
up_for=0
|
up_for=0
|
||||||
|
fi
|
||||||
fi
|
fi
|
||||||
|
|
||||||
sleep "$POLL_INTERVAL"
|
sleep "$POLL_INTERVAL"
|
||||||
|
|||||||
Reference in New Issue
Block a user