Use Valkey instead of Redis

Redis 7.4+ is RSALv2/SSPL and 8.0+ adds AGPLv3, none of which a
closed-source commercial product can ship. Valkey is BSD-3-Clause and
speaks the same RESP protocol, so the `redis` crate works unchanged -
`CONTINUUM_REDIS_URL` becomes `CONTINUUM_VALKEY_URL`, `Config.redis_url`
becomes `valkey_url`, and `Reporter`'s connection field follows.

The `redis://` scheme and the `redis` crate itself deliberately stay:
both name the RESP wire protocol rather than the server we deploy, and
redis-rs has no `valkey://` scheme to use instead. Commented in place so
the inconsistency reads as intentional.
This commit is contained in:
2026-10-03 10:10:39 +00:00
parent 506b66414a
commit 4fae5eb94e
5 changed files with 22 additions and 17 deletions
+4 -1
View File
@@ -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_FRAME_STREAM=stream:camera_frames
CONTINUUM_CONSUMER_GROUP=ai-worker CONTINUUM_CONSUMER_GROUP=ai-worker
CONTINUUM_CONSUMER_NAME=ai-worker-1 CONTINUUM_CONSUMER_NAME=ai-worker-1
+5 -5
View File
@@ -1,7 +1,7 @@
# continuum-ai-worker # continuum-ai-worker
Computer-vision failure detection worker for the Continuum print farm 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 `stream:camera_frames` (by `continuum-backend`'s `src/queue/`) and reports
failures. failures.
@@ -10,8 +10,8 @@ failures.
Real ONNX Runtime inference (the `ort` crate) and the Postgres audit write Real ONNX Runtime inference (the `ort` crate) and the Postgres audit write
(`sqlx`) are **not** implemented yet - `src/inference/mod.rs` always (`sqlx`) are **not** implemented yet - `src/inference/mod.rs` always
"detects" one fixed example failure, and `src/reporter/mod.rs` only prints "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 where the Postgres write would go. The Valkey Streams consumer-group loop in
`main.rs` and the Redis pub/sub alert in `reporter/` are real. This lets the `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 whole pipeline (read a frame, "detect", report, ack) run and be understood
before adding a real model or a SQL database on top. 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 v
inference::detect (stub for now) inference::detect (stub for now)
v v
reporter::report (Redis pub/sub - real) reporter::report (Valkey pub/sub - real)
``` ```
## Getting started ## Getting started
@@ -41,5 +41,5 @@ src/
main.rs Consumer-group loop: XREADGROUP -> detect -> report -> XACK main.rs Consumer-group loop: XREADGROUP -> detect -> report -> XACK
config.rs Loads settings from environment variables config.rs Loads settings from environment variables
inference/ Detection stub - real ONNX Runtime inference goes here later 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)
``` ```
+2 -2
View File
@@ -2,7 +2,7 @@ use std::env;
#[derive(Debug, Clone)] #[derive(Debug, Clone)]
pub struct Config { pub struct Config {
pub redis_url: String, pub valkey_url: String,
pub frame_stream: String, pub frame_stream: String,
pub consumer_group: String, pub consumer_group: String,
pub consumer_name: String, pub consumer_name: String,
@@ -11,7 +11,7 @@ pub struct Config {
impl Config { impl Config {
pub fn from_env() -> anyhow::Result<Self> { pub fn from_env() -> anyhow::Result<Self> {
Ok(Self { 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"), frame_stream: env_or("CONTINUUM_FRAME_STREAM", "stream:camera_frames"),
consumer_group: env_or("CONTINUUM_CONSUMER_GROUP", "ai-worker"), consumer_group: env_or("CONTINUUM_CONSUMER_GROUP", "ai-worker"),
consumer_name: env_or("CONTINUUM_CONSUMER_NAME", "ai-worker-1"), consumer_name: env_or("CONTINUUM_CONSUMER_NAME", "ai-worker-1"),
+4 -2
View File
@@ -18,9 +18,11 @@ async fn main() -> anyhow::Result<()> {
dotenvy::dotenv().ok(); dotenvy::dotenv().ok();
let config = Config::from_env()?; 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 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 // Consumer groups have to exist before you can read from them. This
// fails with "BUSYGROUP" if it already exists - that's fine, ignore it. // fails with "BUSYGROUP" if it already exists - that's fine, ignore it.
+7 -7
View File
@@ -7,17 +7,17 @@ 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's listening.
pub struct Reporter { pub struct Reporter {
redis: redis::aio::MultiplexedConnection, valkey: redis::aio::MultiplexedConnection,
} }
impl Reporter { impl Reporter {
pub async fn connect(redis_url: &str) -> anyhow::Result<Self> { pub async fn connect(valkey_url: &str) -> anyhow::Result<Self> {
let client = redis::Client::open(redis_url)?; let client = redis::Client::open(valkey_url)?;
let redis = client.get_multiplexed_tokio_connection().await?; let valkey = client.get_multiplexed_tokio_connection().await?;
Ok(Self { redis }) 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). /// WebSocket fan-out subscribes to, for live operator-console alerts).
/// ///
/// A real version would also `INSERT` into Postgres here via `sqlx` for /// A real version would also `INSERT` into Postgres here via `sqlx` for
@@ -32,7 +32,7 @@ impl Reporter {
"label": best.label, "label": best.label,
"confidence": best.confidence, "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)"); println!("(would also INSERT this alert into Postgres for the audit trail)");