octoturge
43fd50dc99
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.
27 lines
1.2 KiB
Bash
27 lines
1.2 KiB
Bash
# Valkey, not Redis (licensing - see continuum-common/docs/LICENSES.md).
|
|
# 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
|