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. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
+4
-1
@@ -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
|
||||
|
||||
@@ -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)
|
||||
```
|
||||
|
||||
+2
-2
@@ -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<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"),
|
||||
consumer_group: env_or("CONTINUUM_CONSUMER_GROUP", "ai-worker"),
|
||||
consumer_name: env_or("CONTINUUM_CONSUMER_NAME", "ai-worker-1"),
|
||||
|
||||
+4
-2
@@ -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.
|
||||
|
||||
+7
-7
@@ -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<Self> {
|
||||
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<Self> {
|
||||
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)");
|
||||
|
||||
|
||||
Reference in New Issue
Block a user