diff --git a/.env.example b/.env.example index 5dd2e43..83fd3d0 100644 --- a/.env.example +++ b/.env.example @@ -1,4 +1,7 @@ -CONTINUUM_REDIS_URL=redis://localhost:6379 +# 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 CONTINUUM_CONSUMER_GROUP=ai-worker CONTINUUM_CONSUMER_NAME=ai-worker-1 diff --git a/README.md b/README.md index 6816e1b..75d4e88 100644 --- a/README.md +++ b/README.md @@ -1,7 +1,7 @@ # continuum-ai-worker Computer-vision failure detection worker for the Continuum print farm -platform. Consumes camera frame batches pushed onto the Redis Stream +platform. Consumes camera frame batches pushed onto the Valkey Stream `stream:camera_frames` (by `continuum-backend`'s `src/queue/`) and reports failures. @@ -10,8 +10,8 @@ failures. 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 Redis Streams consumer-group loop in -`main.rs` and the Redis pub/sub alert in `reporter/` are real. This lets the +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. @@ -24,7 +24,7 @@ continuum-proxy --(frames)--> continuum-backend --(XADD)--> stream:camera_frames v inference::detect (stub for now) v - reporter::report (Redis pub/sub - real) + reporter::report (Valkey pub/sub - real) ``` ## Getting started @@ -41,5 +41,5 @@ 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/ Redis pub/sub alert (real) + a note for the Postgres write (later) + reporter/ Valkey pub/sub alert (real) + a note for the Postgres write (later) ``` diff --git a/src/config.rs b/src/config.rs index f72b38f..4606f37 100644 --- a/src/config.rs +++ b/src/config.rs @@ -2,7 +2,7 @@ use std::env; #[derive(Debug, Clone)] pub struct Config { - pub redis_url: String, + pub valkey_url: String, pub frame_stream: String, pub consumer_group: String, pub consumer_name: String, @@ -11,7 +11,7 @@ pub struct Config { impl Config { pub fn from_env() -> anyhow::Result { Ok(Self { - redis_url: env_or("CONTINUUM_REDIS_URL", "redis://localhost:6379"), + valkey_url: env_or("CONTINUUM_VALKEY_URL", "redis://localhost:6379"), frame_stream: env_or("CONTINUUM_FRAME_STREAM", "stream:camera_frames"), consumer_group: env_or("CONTINUUM_CONSUMER_GROUP", "ai-worker"), consumer_name: env_or("CONTINUUM_CONSUMER_NAME", "ai-worker-1"), diff --git a/src/main.rs b/src/main.rs index 8f81533..6334380 100644 --- a/src/main.rs +++ b/src/main.rs @@ -18,9 +18,11 @@ async fn main() -> anyhow::Result<()> { dotenvy::dotenv().ok(); let config = Config::from_env()?; - let client = redis::Client::open(config.redis_url.clone())?; + // The `redis` crate is a RESP client; the server it talks to is Valkey. + // See continuum-common/docs/LICENSES.md for why. + let client = redis::Client::open(config.valkey_url.clone())?; let mut conn = client.get_multiplexed_tokio_connection().await?; - let mut reporter = Reporter::connect(&config.redis_url).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. diff --git a/src/reporter/mod.rs b/src/reporter/mod.rs index 02e1baa..1d4cda0 100644 --- a/src/reporter/mod.rs +++ b/src/reporter/mod.rs @@ -7,17 +7,17 @@ const ALERT_CHANNEL: &str = "channel:print_alerts"; /// Fast-paths a failure alert out to whatever's listening. pub struct Reporter { - redis: redis::aio::MultiplexedConnection, + valkey: redis::aio::MultiplexedConnection, } impl Reporter { - pub async fn connect(redis_url: &str) -> anyhow::Result { - let client = redis::Client::open(redis_url)?; - let redis = client.get_multiplexed_tokio_connection().await?; - Ok(Self { redis }) + pub async fn connect(valkey_url: &str) -> anyhow::Result { + let client = redis::Client::open(valkey_url)?; + let valkey = client.get_multiplexed_tokio_connection().await?; + Ok(Self { valkey }) } - /// Publishes to Redis (real - this is what `continuum-backend`'s + /// Publishes to Valkey (real - this is what `continuum-backend`'s /// WebSocket fan-out subscribes to, for live operator-console alerts). /// /// A real version would also `INSERT` into Postgres here via `sqlx` for @@ -32,7 +32,7 @@ impl Reporter { "label": best.label, "confidence": best.confidence, }); - self.redis.publish::<_, _, ()>(ALERT_CHANNEL, payload.to_string()).await?; + self.valkey.publish::<_, _, ()>(ALERT_CHANNEL, payload.to_string()).await?; println!("(would also INSERT this alert into Postgres for the audit trail)");