The other end of the `stream:camera_frames` drift. This side wrote
`farmId`, `printerId`, `printJobId`, `r2ObjectKey` and `capturedAt`;
`continuum-ai-worker` read `printer_id` and nothing else. Every task was
delivered and acknowledged, and every frame was attributed to a printer
called "unknown", with no error on either side - a stream entry is a
flat string map, so there is nothing to parse and nothing to fail.
The field names are now snake_case, from
`schemas/continuum/v1/ai/frame_task.schema.json`. That is the convention
this repo already follows for anything shared with a Rust consumer (see
`/ws/edge/v1`), and it is not an inconsistency with the camelCase HTTP
routes: this is not an HTTP route.
`print_job_id` is omitted from the XADD when there is no job rather than
written as `""`. An empty string is a value: a consumer that believes it
looks up job id `""`, or puts it in a uuid column. A stream entry has no
null, so "no value" has to mean "no field".
The producer checks each task against the shared schema before writing
it. The check is not for the database's sake - a stream accepts any
field name - it is to make this exact class of drift loud. It earned its
keep immediately by failing on every task with `Unknown format 'uuid'`:
TypeBox ships no format validators and treats an *unregistered* format
as a failure rather than skipping it, so a schema using `format: "uuid"`
checked outside Elysia rejects everything. `src/validation/formats.ts`
registers the two this codebase uses, guarded by `Has` so Elysia's own
registrations still win for the route validators. (The HTTP routes were
never affected - importing Elysia registers ~24 formats including
`uuid`, which I confirmed by asserting a non-uuid body gets a 422.)
A `frame_batch` now goes onto the stream in one pipelined round trip
instead of one per frame, which was a round trip each while holding up
the WebSocket message handler - and a farm produces roughly a frame per
printer per second. ioredis reports per-command pipeline errors in the
result tuple rather than throwing, so those are checked explicitly; a
pipeline where one XADD failed otherwise looks like success.
`frameTaskToStreamFields` and the compiled check live in their own
module because `src/queue/index.ts` opens a Valkey connection on import.
Keeping the wire format pure means `tests/frame_task_contract.test.ts`
asserts it with nothing running.
Tests, mirroring the worker's own so neither side can agree only with
itself: the exact key set, values carried through unchanged,
`print_job_id` present when there is a job and the key absent when there
is not, no camelCase key ever emitted, and the uuid and date-time
formats actually enforced - that last one fails if the format
registration is lost (everything rejected) or becomes a no-op
(nothing rejected). Plus `PrintAlertSchema`, which this repo will have
to parse once the fan-out route exists.
`scripts/e2e-ai-frames.sh` runs this producer against the real worker
binary over real Valkey and asserts the field names by reading them back
out of Valkey - rather than from either side's source, which is what
made the original bug invisible.
Two faults in the first draft of that script, both worth naming because
a test that passes for the wrong reason is worse than no test: it
stranded *every* entry on a dead consumer, so the ordinary delivery path
was never exercised at all; and it grepped the subscriber output for
"print_alert", which matched `redis-cli`'s own subscription
confirmation line and reported a published alert that did not exist.
octoturge
b42cbac299