octoturgeandClaude Opus 5
ae986f371a
Six bugs, found while writing down the `stream:camera_frames` contract this worker is one end of. **It never read the task.** It pulled one field, `printer_id`, from a producer writing `printerId`, and substituted the literal string "unknown" when that missed - which was always. Every other field, including `r2_object_key`, the only thing saying *which picture*, was never looked at. Nothing errored: a stream entry is a flat string map, and asking it for an absent field is not a failure. `src/task.rs` is now the Rust mirror of `continuum-schemas/schemas/continuum/v1/ai/frame_task.schema.json` and a missing field is an error, not a default. Its tests include the regression directly: a camelCase entry must fail to parse rather than yield defaults. **It invented print failures.** `inference::detect` returned a fixed `spaghetti` at 0.91 confidence for every input, including the empty slice it was always handed, so a running worker declared a failure for every frame of every print on evidence it had never looked at. Harmless only because nothing subscribes to the alert channel - the moment anything does, it is a false-alarm generator pointed at an operator. It now abstains. An unimplemented model returning "nothing wrong" is a safe default; one returning "failure" is not. **It discarded the backlog.** The consumer group was created at `$`, meaning "only entries added from now on", so a worker that was down while a farm printed came back up and silently skipped every frame it had missed. For a failure detector that is the opposite of the job. Now `0`. **It stranded entries forever.** An entry delivered to a consumer that died before acking stays pending: it has been handed out, so no other consumer is offered it, and nothing retried it. Now reclaimed via `XPENDING`/`XCLAIM` once idle past `CONTINUUM_CLAIM_IDLE_MS`, and dead-lettered onto `stream:camera_frames:dead` - with the reason and the original fields, so it can be inspected and replayed - once it has been delivered `CONTINUUM_MAX_DELIVERIES` times, so a poison entry can neither be retried forever nor block the queue. **One bad moment killed it.** `xread_options(...).await?` and `reporter.report(...).await?` both propagated out of the loop, so a single transient Valkey error ended the process. Reads now retry with a growing delay and give up only after ten consecutive failures; a report failure leaves the entry unacknowledged for redelivery rather than taking the worker down with it. The `XACK` result is no longer discarded with `let _: RedisResult<i64>`. **`report` would panic** on `detections[0]` for an empty slice. The one caller checks, which makes it a trap for the next one rather than a live bug; it now picks the highest-confidence detection and returns early on none. Also corrects two comments that were simply false: `reporter` claimed `channel:print_alerts` is "what continuum-backend's WebSocket fan-out subscribes to" - nothing anywhere subscribes to it - and the README advertised the fabricated detection as a deliberate aid to understanding the pipeline. The README now states plainly what is real (the loop, tested) and what is not (no frame fetch, no model, no subscriber, no persistence), because "the plumbing works and detects nothing" is the useful thing to know before enabling this. Verified: 10 unit tests, zero clippy warnings, and `continuum-backend/scripts/e2e-ai-frames.sh` - the real producer against this real binary over real Valkey - asserting a well-formed task is consumed and acked, a legacy camelCase entry is dead-lettered naming the field it lacks, an entry stranded on a dead consumer is reclaimed, and nothing is published to the alert channel. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
56 lines
2.2 KiB
Rust
56 lines
2.2 KiB
Rust
use std::env;
|
|
|
|
#[derive(Debug, Clone)]
|
|
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 {
|
|
pub fn from_env() -> anyhow::Result<Self> {
|
|
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)?,
|
|
})
|
|
}
|
|
}
|
|
|
|
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<usize> {
|
|
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}'")),
|
|
}
|
|
}
|