diff --git a/.env.example b/.env.example index 83fd3d0..f639551 100644 --- a/.env.example +++ b/.env.example @@ -2,8 +2,25 @@ # The `redis://` scheme is the RESP wire protocol; redis-rs has no # `valkey://` scheme to use instead. CONTINUUM_VALKEY_URL=redis://localhost:6379 + CONTINUUM_FRAME_STREAM=stream:camera_frames +# Where entries go that must not be retried: a malformed task, or one +# redelivered CONTINUUM_MAX_DELIVERIES times. A separate stream so a poison +# entry can neither block the queue nor vanish. +CONTINUUM_DEAD_LETTER_STREAM=stream:camera_frames:dead + CONTINUUM_CONSUMER_GROUP=ai-worker +# Must be unique per process: two consumers sharing a name share a pending +# list, so each would reclaim entries the other is working on. CONTINUUM_CONSUMER_NAME=ai-worker-1 +# How long XREADGROUP blocks waiting for new entries. +CONTINUUM_BLOCK_MS=5000 +# How long an unacknowledged entry must sit before another consumer may claim +# it. Must comfortably exceed the time to process one frame - once inference +# is real that is a model run, not the no-op it is today. +CONTINUUM_CLAIM_IDLE_MS=60000 +# Deliveries before an entry is dead-lettered instead of retried again. +CONTINUUM_MAX_DELIVERIES=5 + RUST_LOG=info,continuum_ai_worker=debug diff --git a/README.md b/README.md index 75d4e88..3d70a3f 100644 --- a/README.md +++ b/README.md @@ -1,31 +1,77 @@ # continuum-ai-worker Computer-vision failure detection worker for the Continuum print farm -platform. Consumes camera frame batches pushed onto the Valkey Stream -`stream:camera_frames` (by `continuum-backend`'s `src/queue/`) and reports -failures. +platform. Consumes camera frame tasks from the Valkey Stream +`stream:camera_frames` (written by `continuum-backend`'s `src/queue/`) and +reports failures. -## This is a learning-stage boilerplate +## What is real and what is not -Real ONNX Runtime inference (the `ort` crate) and the Postgres audit write -(`sqlx`) are **not** implemented yet - `src/inference/mod.rs` always -"detects" one fixed example failure, and `src/reporter/mod.rs` only prints -where the Postgres write would go. The Valkey Streams consumer-group loop in -`main.rs` and the Valkey pub/sub alert in `reporter/` are real. This lets the -whole pipeline (read a frame, "detect", report, ack) run and be understood -before adding a real model or a SQL database on top. +The consumer-group loop is real and robust: it reads, parses, recovers +entries stranded by a dead consumer, dead-letters what it cannot process, +and acknowledges. The *detection* is not. -## Pipeline +- `src/inference/mod.rs` **abstains**. It has no model and it is handed no + frame, so it returns no detections. It used to return a fixed + `spaghetti` at 0.91 confidence for every call - including the empty slice + it was always given - which made a running worker a false-alarm generator + declaring a failure for every frame of every print, on evidence it had + never looked at. A stub that returns "nothing wrong" is a safe default; a + stub that returns "failure" is not. +- **Nothing fetches the frame.** `FrameTask::r2_object_key` says where the + JPEG is in R2 and no code downloads it. This is the next piece of work and + the reason inference cannot do anything yet. +- `src/reporter/mod.rs` publishes a real `PrintAlert` to + `channel:print_alerts`, but **nothing subscribes to that channel**, and + PUBLISH has no persistence - so an alert published today goes nowhere. The + fan-out is a `continuum-backend` route that does not exist yet. +- **Nothing is persisted.** A detection's permanent record belongs in + Postgres; `frame_embeddings` exists for the vectors, with a pgvector + ivfflat index, and has never been written to. -``` -continuum-proxy --(frames)--> continuum-backend --(XADD)--> stream:camera_frames - | - XREADGROUP (this worker) - v - inference::detect (stub for now) - v - reporter::report (Valkey pub/sub - real) -``` +In short: the plumbing works and is tested; the worker does not yet detect +anything. Do not enable it expecting alerts, and do not wire a console to +`channel:print_alerts` expecting the alerts to mean something. + +## The contract + +`stream:camera_frames` and `channel:print_alerts` are the only interface +between this worker and `continuum-backend`. Both are specified in +`continuum-schemas/schemas/continuum/v1/ai/`, which is canonical: + +- `frame_task.schema.json` - one stream entry. A **flat map of strings**: + a stream entry has no nesting, no numbers and no null, which is why a + field with no value is *omitted* from the XADD rather than written as + `""`. +- `print_alert.schema.json` - the JSON payload of a PUBLISH. + +`src/task.rs` is the Rust mirror; `ts-types/src/schemas/ai.ts` is the +TypeScript one. Change the JSON Schema first, then both mirrors. + +This is written down because it drifted when it was not: the backend wrote +`printerId`, `farmId`, `r2ObjectKey` and `capturedAt`, and this worker read +a single field called `printer_id` and ignored the rest. Every task was +delivered, acknowledged and attributed to a printer called `"unknown"`, and +nothing errored - asking a string map for a field that is not there is not a +failure. Both ends now have a contract test, and +`continuum-backend/scripts/e2e-ai-frames.sh` runs the real producer against +this real binary over real Valkey. + +## Delivery semantics + +| situation | what happens | +| --- | --- | +| a well-formed task | scored, then `XACK`ed | +| a malformed task | copied to `stream:camera_frames:dead` with a reason, then `XACK`ed - it will never become valid, and retrying it would block the queue | +| reporting fails | left unacknowledged, reclaimed after `CONTINUUM_CLAIM_IDLE_MS` | +| a consumer dies mid-task | its pending entries are reclaimed by `XPENDING`/`XCLAIM` on the next pass | +| a task fails `CONTINUUM_MAX_DELIVERIES` times | dead-lettered, so a poison entry cannot be redelivered forever | +| Valkey blips | logged and retried with a growing delay; the worker exits only after `MAX_CONSECUTIVE_FAILURES` consecutive read failures | + +The consumer group is created at `0`, not `$`: `$` means "only entries added +from now on", so a worker that was down while a farm printed would come back +up and silently skip every frame it missed. For a failure detector, dropping +the backlog is the opposite of the job. ## Getting started @@ -34,12 +80,51 @@ cp .env.example .env cargo run ``` +```bash +cargo test # the task-parsing contract +../continuum-backend/scripts/e2e-ai-frames.sh # against real Valkey +``` + +## Configuration + +| variable | default | meaning | +| --- | --- | --- | +| `CONTINUUM_VALKEY_URL` | `redis://localhost:6379` | RESP server (Valkey, not Redis - see `continuum-common/docs/LICENSES.md`) | +| `CONTINUUM_FRAME_STREAM` | `stream:camera_frames` | the work queue | +| `CONTINUUM_DEAD_LETTER_STREAM` | `stream:camera_frames:dead` | where unprocessable entries go | +| `CONTINUUM_CONSUMER_GROUP` | `ai-worker` | consumer group name | +| `CONTINUUM_CONSUMER_NAME` | `ai-worker-1` | must be unique per process | +| `CONTINUUM_BLOCK_MS` | `5000` | how long `XREADGROUP` blocks | +| `CONTINUUM_CLAIM_IDLE_MS` | `60000` | how long before a pending entry may be reclaimed; must exceed the time to process one frame | +| `CONTINUUM_MAX_DELIVERIES` | `5` | deliveries before dead-lettering | + +## Pipeline + +``` +continuum-proxy --(frame_batch)--> continuum-backend --(XADD)--> stream:camera_frames + | | + no frame capture XREADGROUP (this worker) + path yet either v + FrameTask::from_stream_fields + v + (missing: fetch r2_object_key from R2) + v + inference::detect -- abstains + v + reporter::report -> channel:print_alerts (no subscriber) +``` + +Both ends of this pipeline are stubs: `continuum-proxy` restreams cameras +through go2rtc and has no frame-capture path, so nothing sends `frame_batch` +and the stream is empty in practice. The middle is what this repo tests. + ## Structure ``` src/ - main.rs Consumer-group loop: XREADGROUP -> detect -> report -> XACK - config.rs Loads settings from environment variables - inference/ Detection stub - real ONNX Runtime inference goes here later - reporter/ Valkey pub/sub alert (real) + a note for the Postgres write (later) + main.rs Consumer-group loop: recover -> read -> parse -> score -> settle + task.rs FrameTask: the Rust mirror of the frame_task schema + config.rs Settings from environment variables + inference/ Detection - abstains; real ONNX Runtime inference goes here + reporter/ Valkey pub/sub alert (real, but unsubscribed) ``` diff --git a/src/config.rs b/src/config.rs index 4606f37..5d66891 100644 --- a/src/config.rs +++ b/src/config.rs @@ -4,8 +4,21 @@ use std::env; pub struct Config { pub valkey_url: String, pub frame_stream: String, + /// Where entries go that must not be retried: a malformed task, or one + /// that has been redelivered too many times. A separate stream so a + /// poison message can neither block the queue nor vanish. + pub dead_letter_stream: String, pub consumer_group: String, pub consumer_name: String, + /// How long XREADGROUP blocks waiting for new entries. + pub block_ms: usize, + /// How long an unacknowledged entry must sit before another consumer may + /// claim it. Must comfortably exceed the time it takes to process one + /// frame, or two consumers will work the same entry. + pub claim_idle_ms: usize, + /// How many times an entry may be delivered before it is dead-lettered. + /// Without a cap, an entry that always fails is redelivered forever. + pub max_deliveries: usize, } impl Config { @@ -13,8 +26,15 @@ impl Config { Ok(Self { valkey_url: env_or("CONTINUUM_VALKEY_URL", "redis://localhost:6379"), frame_stream: env_or("CONTINUUM_FRAME_STREAM", "stream:camera_frames"), + dead_letter_stream: env_or( + "CONTINUUM_DEAD_LETTER_STREAM", + "stream:camera_frames:dead", + ), consumer_group: env_or("CONTINUUM_CONSUMER_GROUP", "ai-worker"), consumer_name: env_or("CONTINUUM_CONSUMER_NAME", "ai-worker-1"), + block_ms: env_num("CONTINUUM_BLOCK_MS", 5_000)?, + claim_idle_ms: env_num("CONTINUUM_CLAIM_IDLE_MS", 60_000)?, + max_deliveries: env_num("CONTINUUM_MAX_DELIVERIES", 5)?, }) } } @@ -22,3 +42,14 @@ impl Config { fn env_or(key: &str, default: &str) -> String { env::var(key).unwrap_or_else(|_| default.to_string()) } + +fn env_num(key: &str, default: usize) -> anyhow::Result { + match env::var(key) { + Err(_) => Ok(default), + // Refuse rather than fall back to the default: a typo in a tuning + // value should not quietly run with different tuning. + Ok(raw) => raw + .parse() + .map_err(|_| anyhow::anyhow!("{key} must be a non-negative integer, got '{raw}'")), + } +} diff --git a/src/inference/mod.rs b/src/inference/mod.rs index 9fa5b73..e7a70e4 100644 --- a/src/inference/mod.rs +++ b/src/inference/mod.rs @@ -7,13 +7,59 @@ pub struct Detection { } /// Stand-in for real ONNX Runtime inference (the `ort` crate, using a model -/// trained on failure images). Always reports one fixed "failure" so the -/// rest of the pipeline - reporting, acking the stream entry - has -/// something to do and can be tested without a real model file. +/// trained on failure images). /// -/// Swap this out once you're comfortable with the pipeline around it: -/// decode `frame_bytes` with the `image` crate, run it through an `ort` -/// session, and turn the model's output into real `Detection`s. -pub fn detect(_frame_bytes: &[u8]) -> Vec { - vec![Detection { label: "spaghetti".to_string(), confidence: 0.91 }] +/// **It abstains.** Given no frame bytes it returns no detections, and since +/// nothing fetches frames yet, that is every call. +/// +/// This used to return a fixed `spaghetti` at 0.91 confidence for every +/// input, including the empty slice it was always handed, on the reasoning +/// that the rest of the pipeline needed something to do. The effect was a +/// worker that declared a print failure for every frame of every print, on +/// evidence it had never looked at. A stub that fabricates findings is worse +/// than one that does nothing: the moment anything subscribes to the alert +/// channel, it is a false-alarm generator pointed at an operator, and the +/// pipeline it was meant to exercise is exercised with data that proves +/// nothing about it. +/// +/// Tests that need a detection should construct one directly rather than +/// making this lie. +/// +/// To finish this: decode `frame_bytes` with the `image` crate, run it +/// through an `ort` session, and turn the model's output into `Detection`s. +/// The `r2_object_key` that says where the bytes are is already carried +/// through `FrameTask`; what is missing is the fetch. +pub fn detect(frame_bytes: &[u8]) -> Vec { + if frame_bytes.is_empty() { + return Vec::new(); + } + + // Unreachable until something fetches frames, and deliberately not a + // fabricated detection when it becomes reachable - an unimplemented + // model that returns "nothing wrong" is a safe default; one that returns + // "failure" is not. + tracing::warn!( + bytes = frame_bytes.len(), + "inference is not implemented; frame scored as no-detection" + ); + Vec::new() +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn abstains_when_there_is_no_frame() { + assert!(detect(&[]).is_empty()); + } + + /// The regression test for the fabricated-alert bug: no input may ever + /// produce a detection from the stub. + #[test] + fn never_invents_a_detection() { + assert!(detect(&[]).is_empty()); + assert!(detect(&[0xFF, 0xD8, 0xFF]).is_empty()); + assert!(detect(&vec![0u8; 1024]).is_empty()); + } } diff --git a/src/main.rs b/src/main.rs index 6334380..c7de125 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,13 +1,48 @@ mod config; mod inference; mod reporter; +mod task; -use redis::streams::{StreamKey, StreamReadOptions, StreamReadReply}; +use std::collections::HashMap; +use std::time::Duration; + +use redis::streams::{ + StreamClaimReply, StreamId, StreamKey, StreamPendingCountReply, StreamRangeReply, + StreamReadOptions, StreamReadReply, +}; use redis::AsyncCommands; use tracing_subscriber::EnvFilter; use crate::config::Config; use crate::reporter::Reporter; +use crate::task::FrameTask; + +/// Entries already on the stream when the group is first created. +/// +/// `0`, not `$`. `$` means "only entries added from now on", which silently +/// discards whatever is already queued - so a worker that is down while a +/// farm prints comes back up and never looks at the frames it missed. For a +/// failure detector, dropping the backlog is the opposite of the job. +const GROUP_START_ID: &str = "0"; + +/// How many consecutive loop failures to tolerate before giving up. A +/// transient Valkey blip should not kill the worker; a permanently wrong URL +/// should not have it spin and log forever. +const MAX_CONSECUTIVE_FAILURES: usize = 10; + +/// What to do with an entry once it has been looked at. +enum Outcome { + /// Processed. Acknowledge it. + Done, + /// Will never succeed - a malformed task. Dead-letter it and acknowledge, + /// because redelivering it would block the queue behind something that + /// cannot be fixed by trying again. + Poison(String), + /// Failed for a reason that might not recur. Leave it unacknowledged so + /// `recover_pending` reclaims it later; it is dead-lettered once it has + /// been delivered `max_deliveries` times. + Retry(String), +} #[tokio::main] async fn main() -> anyhow::Result<()> { @@ -24,39 +59,266 @@ async fn main() -> anyhow::Result<()> { let mut conn = client.get_multiplexed_tokio_connection().await?; let mut reporter = Reporter::connect(&config.valkey_url).await?; - // Consumer groups have to exist before you can read from them. This - // fails with "BUSYGROUP" if it already exists - that's fine, ignore it. - let _: redis::RedisResult = conn - .xgroup_create_mkstream(&config.frame_stream, &config.consumer_group, "$") - .await; + // Consumer groups have to exist before you can read from them. BUSYGROUP + // means it already does, which is the normal case on every restart. + match conn + .xgroup_create_mkstream::<_, _, _, String>( + &config.frame_stream, + &config.consumer_group, + GROUP_START_ID, + ) + .await + { + Ok(_) => tracing::info!(group = %config.consumer_group, "created consumer group"), + Err(err) if err.code() == Some("BUSYGROUP") => { + tracing::debug!("consumer group already exists") + } + // Anything else - a wrong URL, no permission, a key of the wrong + // type - is worth failing on rather than entering a read loop that + // cannot possibly work. + Err(err) => return Err(err).context_msg("could not create the consumer group"), + } - tracing::info!(stream = %config.frame_stream, group = %config.consumer_group, "waiting for camera frames"); + tracing::info!( + stream = %config.frame_stream, + group = %config.consumer_group, + consumer = %config.consumer_name, + "waiting for camera frames" + ); let read_opts = StreamReadOptions::default() .group(&config.consumer_group, &config.consumer_name) - .block(5000); + .block(config.block_ms); + + let mut consecutive_failures = 0usize; loop { - let reply: StreamReadReply = conn.xread_options(&[&config.frame_stream], &[">"], &read_opts).await?; + // Anything a dead consumer left behind, first: an entry stuck pending + // is one a printer may have been failing during. + if let Err(err) = recover_pending(&mut conn, &mut reporter, &config).await { + tracing::warn!(error = %err, "could not recover pending entries this pass"); + } + + let reply: StreamReadReply = match conn + .xread_options(&[&config.frame_stream], &[">"], &read_opts) + .await + { + Ok(reply) => reply, + Err(err) => { + consecutive_failures += 1; + tracing::error!( + error = %err, + attempt = consecutive_failures, + "reading the frame stream failed" + ); + if consecutive_failures >= MAX_CONSECUTIVE_FAILURES { + // Previously this was `?`, so the first transient error + // ended the process. + anyhow::bail!("gave up after {consecutive_failures} consecutive read failures"); + } + tokio::time::sleep(Duration::from_millis(500 * consecutive_failures as u64)).await; + continue; + } + }; + consecutive_failures = 0; for StreamKey { key, ids } in reply.keys { for entry in ids { - let printer_id = printer_id_field(&entry.map); - - let detections = inference::detect(&[]); - if !detections.is_empty() { - reporter.report(&printer_id, &detections).await?; - } - - let _: redis::RedisResult = conn.xack(&key, &config.consumer_group, &[entry.id.as_str()]).await; + // One entry's failure must not skip the rest of the batch, + // and must not leave the loop. + handle_and_settle(&mut conn, &mut reporter, &config, &key, &entry).await; } } } } -fn printer_id_field(fields: &std::collections::HashMap) -> String { - match fields.get("printer_id") { - Some(redis::Value::BulkString(bytes)) => String::from_utf8_lossy(bytes).to_string(), - _ => "unknown".to_string(), +/// Processes one entry and then acknowledges, dead-letters or leaves it +/// pending according to the outcome. +async fn handle_and_settle( + conn: &mut redis::aio::MultiplexedConnection, + reporter: &mut Reporter, + config: &Config, + stream: &str, + entry: &StreamId, +) { + match process(reporter, &entry.map).await { + Outcome::Done => ack(conn, config, stream, &entry.id).await, + Outcome::Poison(reason) => { + tracing::error!(id = %entry.id, reason = %reason, "dead-lettering a malformed frame task"); + dead_letter(conn, config, &entry.id, &entry.map, &reason).await; + ack(conn, config, stream, &entry.id).await; + } + Outcome::Retry(reason) => { + // Deliberately not acknowledged: it stays in the group's pending + // list and `recover_pending` will claim it after `claim_idle_ms`. + tracing::warn!(id = %entry.id, reason = %reason, "leaving a frame task for redelivery"); + } + } +} + +/// Scores one frame and reports anything found. +async fn process(reporter: &mut Reporter, fields: &HashMap) -> Outcome { + let task = match FrameTask::from_stream_fields(fields) { + Ok(task) => task, + Err(err) => return Outcome::Poison(err.to_string()), + }; + + // NOTE: the frame bytes are not fetched yet - `r2_object_key` says where + // they are and nothing downloads them. `inference::detect` is given an + // empty slice and, as of 2026-10-11, abstains rather than inventing a + // detection, so this is a no-op that acknowledges its input. See the + // module docs in src/inference/mod.rs. + let frame_bytes: Vec = Vec::new(); + + let detections = inference::detect(&frame_bytes); + if detections.is_empty() { + tracing::debug!(printer_id = %task.printer_id, "no detections"); + return Outcome::Done; + } + + match reporter.report(&task, &detections).await { + Ok(()) => Outcome::Done, + // A publish failure is transient by nature - the connection dropped, + // the server is loaded - so this is a retry and not a poison. + Err(err) => Outcome::Retry(format!("reporting failed: {err}")), + } +} + +/// Reclaims entries that were delivered to a consumer which never +/// acknowledged them - the usual cause being that it was killed mid-frame. +/// +/// Without this, an entry delivered to a consumer that dies stays pending +/// forever: it has been handed out, so no other consumer is offered it, and +/// nothing retries it. +async fn recover_pending( + conn: &mut redis::aio::MultiplexedConnection, + reporter: &mut Reporter, + config: &Config, +) -> anyhow::Result<()> { + let pending: StreamPendingCountReply = conn + .xpending_count(&config.frame_stream, &config.consumer_group, "-", "+", 64usize) + .await?; + + for entry in pending.ids { + // Only entries that have been idle long enough that whoever holds + // them is plainly not working on them any more. + if entry.last_delivered_ms < config.claim_idle_ms { + continue; + } + + if entry.times_delivered >= config.max_deliveries { + let reason = format!( + "delivered {} times without being acknowledged (limit {})", + entry.times_delivered, config.max_deliveries + ); + tracing::error!(id = %entry.id, reason = %reason, "dead-lettering a repeatedly failing frame task"); + // Fetch the fields before acknowledging, so the dead-letter entry + // carries the original task rather than only its id. + let range: StreamRangeReply = conn + .xrange(&config.frame_stream, &entry.id, &entry.id) + .await + .unwrap_or(StreamRangeReply { ids: Vec::new() }); + let fields = range.ids.first().map(|e| e.map.clone()).unwrap_or_default(); + dead_letter(conn, config, &entry.id, &fields, &reason).await; + ack(conn, config, &config.frame_stream, &entry.id).await; + continue; + } + + let claimed: StreamClaimReply = conn + .xclaim( + &config.frame_stream, + &config.consumer_group, + &config.consumer_name, + config.claim_idle_ms, + &[entry.id.as_str()], + ) + .await?; + + for reclaimed in &claimed.ids { + tracing::info!( + id = %reclaimed.id, + delivery = entry.times_delivered + 1, + "reclaimed a pending frame task" + ); + handle_and_settle(conn, reporter, config, &config.frame_stream, reclaimed).await; + } + } + + Ok(()) +} + +/// Acknowledges an entry, and says so if that fails. +/// +/// The result used to be discarded with `let _: RedisResult`. A failed +/// XACK means the entry stays pending and will be reclaimed and reprocessed +/// later, which is survivable but is not nothing - it should not be invisible. +async fn ack( + conn: &mut redis::aio::MultiplexedConnection, + config: &Config, + stream: &str, + id: &str, +) { + match conn + .xack::<_, _, _, i64>(stream, &config.consumer_group, &[id]) + .await + { + Ok(1) => {} + Ok(count) => tracing::warn!(id = %id, count, "XACK acknowledged an unexpected number of entries"), + Err(err) => tracing::error!(id = %id, error = %err, "XACK failed; the entry stays pending"), + } +} + +/// Copies an entry onto the dead-letter stream, with why it went there. +async fn dead_letter( + conn: &mut redis::aio::MultiplexedConnection, + config: &Config, + id: &str, + fields: &HashMap, + reason: &str, +) { + // The original fields verbatim, so a dead-lettered task can be inspected + // and replayed, plus provenance. Prefixed keys cannot collide with the + // task's own, which `additionalProperties: false` in the schema keeps to + // a known set. + let mut pairs: Vec<(String, String)> = vec![ + ("dead_original_id".to_string(), id.to_string()), + ("dead_reason".to_string(), reason.to_string()), + ("dead_stream".to_string(), config.frame_stream.clone()), + ]; + for (key, value) in fields { + let rendered = match value { + redis::Value::BulkString(bytes) => String::from_utf8_lossy(bytes).into_owned(), + redis::Value::SimpleString(text) => text.clone(), + redis::Value::Int(number) => number.to_string(), + other => format!("{other:?}"), + }; + pairs.push((key.clone(), rendered)); + } + + let args: Vec<(&str, &str)> = pairs.iter().map(|(k, v)| (k.as_str(), v.as_str())).collect(); + if let Err(err) = conn + .xadd::<_, _, _, _, String>(&config.dead_letter_stream, "*", &args) + .await + { + // Nothing left to do but be loud: the entry is about to be + // acknowledged, so this is the last chance to record it anywhere. + tracing::error!( + id = %id, + error = %err, + reason = %reason, + "could not write to the dead-letter stream; this task is being dropped" + ); + } +} + +/// `anyhow::Context` for `RedisError` without pulling the trait into scope at +/// every call site. +trait ContextMsg { + fn context_msg(self, message: &'static str) -> anyhow::Result; +} + +impl ContextMsg for Result { + fn context_msg(self, message: &'static str) -> anyhow::Result { + self.map_err(|err| anyhow::anyhow!("{message}: {err}")) } } diff --git a/src/reporter/mod.rs b/src/reporter/mod.rs index 1d4cda0..f6fab40 100644 --- a/src/reporter/mod.rs +++ b/src/reporter/mod.rs @@ -2,10 +2,13 @@ use redis::AsyncCommands; use tracing::info; use crate::inference::Detection; +use crate::task::FrameTask; +/// Mirrors `x-channel` in +/// `schemas/continuum/v1/ai/print_alert.schema.json`. const ALERT_CHANNEL: &str = "channel:print_alerts"; -/// Fast-paths a failure alert out to whatever's listening. +/// Fast-paths a failure alert out to whatever is listening. pub struct Reporter { valkey: redis::aio::MultiplexedConnection, } @@ -17,24 +20,61 @@ impl Reporter { Ok(Self { valkey }) } - /// Publishes to Valkey (real - this is what `continuum-backend`'s - /// WebSocket fan-out subscribes to, for live operator-console alerts). + /// Publishes a `PrintAlert` onto `channel:print_alerts`, in the shape of + /// `schemas/continuum/v1/ai/print_alert.schema.json`. /// - /// A real version would also `INSERT` into Postgres here via `sqlx` for - /// the permanent audit trail - that's left as a `println!` for now; - /// come back to it once async SQL feels comfortable. - pub async fn report(&mut self, printer_id: &str, detections: &[Detection]) -> anyhow::Result<()> { - let best = &detections[0]; - info!(printer_id, label = %best.label, confidence = best.confidence, "reporting print failure"); + /// ## Nothing subscribes to this yet + /// + /// The previous version of this comment said the channel was "what + /// `continuum-backend`'s WebSocket fan-out subscribes to". It is not: + /// as of 2026-10-11 no process anywhere subscribes, and PUBLISH has no + /// persistence, so every alert published so far went nowhere. Finishing + /// that is a backend route, not a change here. + /// + /// ## And nothing persists it + /// + /// A detection's permanent record belongs in Postgres - `frame_embeddings` + /// exists for the vectors and has never been written to either. PUBLISH + /// is fire-and-forget: an alert published with no subscriber is gone, so + /// this channel can only ever be a live-console feed, never the record. + pub async fn report(&mut self, task: &FrameTask, detections: &[Detection]) -> anyhow::Result<()> { + // `detections[0]` would panic here. The caller does check, but a + // public method that panics on an empty slice is a trap for the next + // caller. + let Some(best) = detections + .iter() + .max_by(|a, b| a.confidence.total_cmp(&b.confidence)) + else { + return Ok(()); + }; - let payload = serde_json::json!({ - "printerId": printer_id, + info!( + printer_id = %task.printer_id, + label = %best.label, + confidence = best.confidence, + "reporting print failure" + ); + + // Built field by field rather than by deriving Serialize on a struct, + // because `print_job_id` must be *absent* when there is no job rather + // than `null` - same rule as the stream task, and the schema says so. + let mut payload = serde_json::json!({ + "printer_id": task.printer_id, "label": best.label, "confidence": best.confidence, + // From the frame, not from now: an operator needs to know when + // the printer looked like this, not when a GPU got round to it. + "captured_at": task.captured_at, + // So a console can show the picture, not just a label. + "r2_object_key": task.r2_object_key, }); - self.valkey.publish::<_, _, ()>(ALERT_CHANNEL, payload.to_string()).await?; + if let Some(job_id) = &task.print_job_id { + payload["print_job_id"] = serde_json::Value::String(job_id.clone()); + } - println!("(would also INSERT this alert into Postgres for the audit trail)"); + self.valkey + .publish::<_, _, ()>(ALERT_CHANNEL, payload.to_string()) + .await?; Ok(()) } diff --git a/src/task.rs b/src/task.rs new file mode 100644 index 0000000..8d04ca4 --- /dev/null +++ b/src/task.rs @@ -0,0 +1,249 @@ +//! The Rust mirror of `schemas/continuum/v1/ai/frame_task.schema.json`. +//! +//! The JSON Schema is canonical and the TypeScript mirror is in +//! `continuum-schemas/packages/ts-types/src/schemas/ai.ts`. Change the schema +//! first, then both mirrors. +//! +//! This module exists because the field names were previously nowhere: the +//! backend wrote `printerId`, `farmId`, `r2ObjectKey` and `capturedAt`, and +//! this worker read one field called `printer_id` and ignored the rest. The +//! read never matched, so every frame was attributed to a printer called +//! "unknown" and the object key - the only thing that says *which picture* - +//! was never looked at. Nothing errored, because a stream entry is just a +//! string map and asking it for a field that is not there is not a failure. + +use std::collections::HashMap; +use std::fmt; + +/// One entry off `stream:camera_frames`. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct FrameTask { + pub farm_id: String, + /// `printers.id` - the backend's UUID. Not the gateway's own printer id, + /// which is only meaningful inside its own farm. + pub printer_id: String, + /// `None` when the frame belongs to no tracked job: a printer may be + /// streaming while idle, or running something queued outside Continuum. + pub print_job_id: Option, + pub r2_object_key: String, + /// The gateway's clock at capture, not enqueue time. A gateway that + /// buffered through an uplink outage produces tasks minutes older than + /// their stream id suggests. + pub captured_at: String, +} + +/// Why a task could not be read. Every variant is terminal: a stream entry +/// does not become well-formed by being retried, so these are dead-lettered +/// rather than redelivered. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum TaskError { + MissingField(&'static str), + /// Present but empty. Called out separately from missing because it means + /// a producer wrote `""` where it meant "no value" - the exact bug the + /// schema's "omit the field" rule exists to prevent - and the message + /// should say so rather than claiming the field was absent. + EmptyField(&'static str), + NotAUuid { field: &'static str, value: String }, +} + +impl fmt::Display for TaskError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::MissingField(name) => write!(f, "missing required field '{name}'"), + Self::EmptyField(name) => { + write!(f, "field '{name}' is present but empty; a producer with no value must omit it") + } + Self::NotAUuid { field, value } => { + write!(f, "field '{field}' is not a uuid: '{value}'") + } + } + } +} + +impl std::error::Error for TaskError {} + +/// A stream entry's values are opaque bulk strings; this is the only place +/// that turns one into a `String`. +fn read_field(fields: &HashMap, name: &str) -> Option { + match fields.get(name) { + Some(redis::Value::BulkString(bytes)) => Some(String::from_utf8_lossy(bytes).into_owned()), + Some(redis::Value::SimpleString(text)) => Some(text.clone()), + Some(redis::Value::Int(number)) => Some(number.to_string()), + _ => None, + } +} + +fn required(fields: &HashMap, name: &'static str) -> Result { + match read_field(fields, name) { + None => Err(TaskError::MissingField(name)), + Some(value) if value.is_empty() => Err(TaskError::EmptyField(name)), + Some(value) => Ok(value), + } +} + +/// A deliberately cheap uuid check: the right length, hyphens in the right +/// four places, hex everywhere else. Not a parse, and no `uuid` dependency +/// for something that is never taken apart here. +/// +/// It is worth doing at all because these ids are passed straight into an +/// alert payload and, once the Postgres write exists, into uuid columns. +/// Rejecting a malformed id here turns a bad row or a nonsense alert into one +/// dead-lettered entry. +fn looks_like_uuid(value: &str) -> bool { + if value.len() != 36 { + return false; + } + value.bytes().enumerate().all(|(index, byte)| match index { + 8 | 13 | 18 | 23 => byte == b'-', + _ => byte.is_ascii_hexdigit(), + }) +} + +fn required_uuid( + fields: &HashMap, + name: &'static str, +) -> Result { + let value = required(fields, name)?; + if looks_like_uuid(&value) { + Ok(value) + } else { + Err(TaskError::NotAUuid { field: name, value }) + } +} + +impl FrameTask { + /// Reads a task out of a stream entry's field map. + pub fn from_stream_fields(fields: &HashMap) -> Result { + // An absent key and an empty value are treated alike, per the + // schema: the producer is supposed to omit what it has no value for, + // and a consumer that distinguishes the two would break the moment a + // producer got that wrong - which one did. + let print_job_id = match read_field(fields, "print_job_id") { + None => None, + Some(value) if value.is_empty() => None, + Some(value) if looks_like_uuid(&value) => Some(value), + Some(value) => { + return Err(TaskError::NotAUuid { field: "print_job_id", value }); + } + }; + + Ok(Self { + farm_id: required_uuid(fields, "farm_id")?, + printer_id: required_uuid(fields, "printer_id")?, + print_job_id, + r2_object_key: required(fields, "r2_object_key")?, + captured_at: required(fields, "captured_at")?, + }) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + const FARM: &str = "11111111-1111-1111-1111-111111111111"; + const PRINTER: &str = "22222222-2222-2222-2222-222222222222"; + const JOB: &str = "33333333-3333-3333-3333-333333333333"; + + fn fields(pairs: &[(&str, &str)]) -> HashMap { + pairs + .iter() + .map(|(k, v)| ((*k).to_string(), redis::Value::BulkString(v.as_bytes().to_vec()))) + .collect() + } + + fn complete() -> Vec<(&'static str, &'static str)> { + vec![ + ("farm_id", FARM), + ("printer_id", PRINTER), + ("r2_object_key", "farms/x/frames/1.jpg"), + ("captured_at", "2026-10-11T12:00:00Z"), + ] + } + + #[test] + fn reads_a_complete_task() { + let task = FrameTask::from_stream_fields(&fields(&complete())).unwrap(); + assert_eq!(task.farm_id, FARM); + assert_eq!(task.printer_id, PRINTER); + assert_eq!(task.print_job_id, None); + assert_eq!(task.r2_object_key, "farms/x/frames/1.jpg"); + } + + #[test] + fn reads_an_optional_job_id() { + let mut pairs = complete(); + pairs.push(("print_job_id", JOB)); + let task = FrameTask::from_stream_fields(&fields(&pairs)).unwrap(); + assert_eq!(task.print_job_id, Some(JOB.to_string())); + } + + #[test] + fn an_empty_job_id_is_no_job_id() { + // The backend used to write `""` for "no job". Readers have to accept + // that as absent rather than carrying an empty string forward. + let mut pairs = complete(); + pairs.push(("print_job_id", "")); + let task = FrameTask::from_stream_fields(&fields(&pairs)).unwrap(); + assert_eq!(task.print_job_id, None); + } + + /// The regression test for the bug this module was written for. The old + /// consumer read `printer_id` from a producer writing `printerId` and + /// substituted the string "unknown" - so this has to be an error, not a + /// default. + #[test] + fn camel_case_field_names_are_an_error_not_a_default() { + let camel = fields(&[ + ("farmId", FARM), + ("printerId", PRINTER), + ("r2ObjectKey", "farms/x/frames/1.jpg"), + ("capturedAt", "2026-10-11T12:00:00Z"), + ]); + let error = FrameTask::from_stream_fields(&camel).unwrap_err(); + assert_eq!(error, TaskError::MissingField("farm_id")); + } + + #[test] + fn every_required_field_is_actually_required() { + for missing in ["farm_id", "printer_id", "r2_object_key", "captured_at"] { + let pairs: Vec<_> = complete().into_iter().filter(|(k, _)| *k != missing).collect(); + let error = FrameTask::from_stream_fields(&fields(&pairs)).unwrap_err(); + assert_eq!(error, TaskError::MissingField(missing), "for {missing}"); + } + } + + #[test] + fn an_empty_required_field_says_so() { + let mut pairs = complete(); + pairs.retain(|(k, _)| *k != "r2_object_key"); + pairs.push(("r2_object_key", "")); + assert_eq!( + FrameTask::from_stream_fields(&fields(&pairs)).unwrap_err(), + TaskError::EmptyField("r2_object_key"), + ); + } + + #[test] + fn a_malformed_printer_id_is_refused() { + let mut pairs = complete(); + pairs.retain(|(k, _)| *k != "printer_id"); + pairs.push(("printer_id", "unknown")); + assert!(matches!( + FrameTask::from_stream_fields(&fields(&pairs)).unwrap_err(), + TaskError::NotAUuid { field: "printer_id", .. } + )); + } + + #[test] + fn uuid_check_accepts_real_ids_and_rejects_near_misses() { + assert!(looks_like_uuid(FARM)); + assert!(looks_like_uuid("0f8fad5b-d9cb-469f-a165-70867728950e")); + assert!(!looks_like_uuid("")); + assert!(!looks_like_uuid("unknown")); + // Right length, hyphens in the wrong places. + assert!(!looks_like_uuid("0f8fad5bd-9cb-469f-a165-70867728950")); + // Right shape, non-hex digit. + assert!(!looks_like_uuid("0f8fad5b-d9cb-469f-a165-70867728950g")); + } +}