diff --git a/.github/workflows/on-pr-quality.yaml b/.github/workflows/on-pr-quality.yaml index 1cdba6b6..f7a65654 100644 --- a/.github/workflows/on-pr-quality.yaml +++ b/.github/workflows/on-pr-quality.yaml @@ -22,3 +22,6 @@ jobs: kafka: uses: ./.github/workflows/integration-kafka.yaml + + distributed-cli: + uses: ./.github/workflows/integration-distributed-cli.yaml diff --git a/distributed_cli/src/cli.rs b/distributed_cli/src/cli.rs index fe68621f..3358f9ab 100644 --- a/distributed_cli/src/cli.rs +++ b/distributed_cli/src/cli.rs @@ -723,7 +723,7 @@ fn harness_main_rs(entrypoint: &str, mode: HarnessMode) -> String { let envelope = distributed::DistributedManifestEnvelope::new(manifest); let statements = envelope .project - .sql_statements(distributed::TableSqlDialect::{dialect}) + .sql_statements(distributed::table::TableSqlDialect::{dialect}) .expect("manifest SQL should render"); if !statements.is_empty() {{ println!("{{}}", statements.join("\n\n")); @@ -1031,6 +1031,23 @@ mod tests { assert!(cargo_toml.contains("name = \"dsvc-manifest-harness-schema-postgres\"")); } + #[test] + fn schema_harness_uses_public_table_module_sql_dialect() { + let main_rs = harness_main_rs( + "orders_service::distributed_manifest", + HarnessMode::SchemaSql(SchemaDialect::Postgres), + ); + + assert!( + main_rs.contains("distributed::table::TableSqlDialect::Postgres"), + "main.rs: {main_rs}" + ); + assert!( + !main_rs.contains("distributed::TableSqlDialect"), + "main.rs: {main_rs}" + ); + } + #[test] fn atlas_spec_uses_secret_ref_by_default() { let spec = atlas_spec_from_flags(&schema_args(), "CREATE TABLE orders (id text);".into()) diff --git a/docs/async-transports.md b/docs/async-transports.md index 1a64bfa5..cf2126dd 100644 --- a/docs/async-transports.md +++ b/docs/async-transports.md @@ -126,6 +126,7 @@ bus.subscribe(service.clone(), RunOptions::idempotent()).await?; // fan-out let namespace = "orders-prod"; // let bus = NatsBus::connect("nats://localhost:4222").namespace(namespace).await?; // let bus = PostgresBus::new(pool); +// let bus = SqliteBus::new(pool); // let bus = RabbitBus::connect("amqp://localhost:5672/%2f").namespace(namespace).await?; // let bus = KafkaBus::connect("localhost:9092").namespace(namespace).await?; ``` @@ -137,8 +138,9 @@ of one service deployment use the same value; independent event consumers use different values so each gets its own event copy. Direct `Handlers` or manual `listen`/`subscribe` calls can set the group with `bus.group(..)` or `Handlers::named(..)`. `namespace` scopes streams, subjects, topics, queues, or -exchanges on a shared broker. `PostgresBus` does not take `namespace` because the -database/schema behind `pool` already scopes its bus tables. +exchanges on a shared broker. `PostgresBus` and `SqliteBus` do not take +`namespace` because the database/schema/file behind `pool` already scopes their +bus tables. Topology names are validated before broker use. Keep groups/service names to portable deployment IDs (`A-Z`, `a-z`, `0-9`, `_`, `-`); namespaces may also use @@ -155,6 +157,7 @@ handlers use distinct `group`s when each service needs its own copy: | `InMemoryBus` | (always) | named queue, popped once | retained log + per-subscriber cursor | | `NatsBus` | `nats` | shared durable `{group}_cmd` on the stream | durable `{group}_evt` per group | | `PostgresBus` | `postgres` | `bus_queue`, `FOR UPDATE SKIP LOCKED` | `bus_log` + `bus_offset` per `group` (Kafka-style) | +| `SqliteBus` | `sqlite` | `bus_queue`, atomic `UPDATE ... RETURNING` lease claim | `bus_log` + `bus_offset` per `group` | | `RabbitBus` | `rabbitmq` | default exchange → durable queue `{ns}.cmd.{name}` | topic exchange → queue `{ns}.evt.{group}` per group | | `KafkaBus` | `kafka` | shared consumer group `{ns}.{group}.cmd` | consumer group per service `{ns}.{group}.evt` | | `KnativeBus` | `http` | POST CloudEvent → `{target}-commands` broker-ingress | POST → own `{source}-events` broker; consume via generated Triggers | @@ -171,6 +174,12 @@ the uniform drain-to-idle `run_source` model the facade shares; its `bus_log` + `bus_offset` fan-out gives single-DB transactional effectively-once (the offset advances with the effects). See `specs/transport-bus-facade`. +`SqliteBus` is the same single-database pattern scaled down to a local SQLite +file: `bus_queue` is claimed with a conditional `UPDATE ... RETURNING` because +SQLite has no `FOR UPDATE SKIP LOCKED`, and `bus_log`/`bus_offset` provide +fan-out. It is intended for local durable transport, tests, demos, and small +single-node deployments, not as a high-throughput broker replacement. + ## Testing The reusable conformance harness (`tests/transport_conformance/`) proves the @@ -183,6 +192,7 @@ docker compose up -d # postgres, rabbitmq, kafka, nats (see compose.yaml) DATABASE_URL=postgres://sourced:sourced@localhost:5432/distributed \ cargo test --test postgres_transport --features postgres +cargo test --test sqlite_transport --features sqlite NATS_URL=nats://localhost:4222 cargo test --test nats_transport --features nats AMQP_URL=amqp://guest:guest@localhost:5672/%2f \ cargo test --test rabbitmq_transport --features rabbitmq @@ -198,9 +208,10 @@ on push to `main`. ## Status Implemented and verified: the core contracts, the source runner, the publisher / -outbox dispatcher, the conformance harness, the Postgres / NATS / RabbitMQ / -Kafka adapters, the Knative ingress, and the **bus facade** (`Bus` + -`BusConsumer` with `InMemoryBus` / `NatsBus` / `PostgresBus` / `RabbitBus` / -`KafkaBus` / `KnativeBus`, each with real-broker competing-vs-fan-out tests). +outbox dispatcher, the conformance harness, the Postgres / SQLite / NATS / +RabbitMQ / Kafka adapters, the Knative ingress, and the **bus facade** (`Bus` + +`BusConsumer` with `InMemoryBus` / `NatsBus` / `PostgresBus` / `SqliteBus` / +`RabbitBus` / `KafkaBus` / `KnativeBus`, each with competing-vs-fan-out +integration tests against its broker or local database). Still open: migrating the in-repo examples to showcase these APIs. See `tasks/transport-docs-examples-cutover`. diff --git a/src/bus/mod.rs b/src/bus/mod.rs index 6cae8022..2769b0e4 100644 --- a/src/bus/mod.rs +++ b/src/bus/mod.rs @@ -112,6 +112,8 @@ mod router; mod run_options; mod runner; mod source; +#[cfg(feature = "sqlite")] +mod sqlite_bus; mod stable_id; mod topology; @@ -151,6 +153,8 @@ pub use router::MessageRouter; pub use run_options::{ConsumerDeliveryMode, InboxHook, NoInbox, RunOptions}; pub use runner::run_source; pub use source::{AsyncMessageSource, ReceivedMessage}; +#[cfg(feature = "sqlite")] +pub use sqlite_bus::{SqliteBus, SqliteLogReceived, SqliteQueueReceived}; pub use stable_id::{validate_stable_message_id, StableMessageIdError, MAX_STABLE_MESSAGE_ID_LEN}; pub use topology::{ resolve_consumer_group, validate_consumer_group, validate_namespace, BusTopologyConfig, diff --git a/src/bus/sqlite_bus.rs b/src/bus/sqlite_bus.rs new file mode 100644 index 00000000..8131efef --- /dev/null +++ b/src/bus/sqlite_bus.rs @@ -0,0 +1,560 @@ +//! SQLite [`Bus`] + [`BusConsumer`] — a local durable bus. +//! +//! SQLite covers both bus modes with the same app-facing shape as +//! [`PostgresBus`](super::PostgresBus), but with SQLite-native concurrency: +//! +//! - **`send` / `listen` (point-to-point, competing):** a durable work-queue +//! table (`bus_queue`) claimed by one atomic `UPDATE ... RETURNING` under a +//! lease. SQLite serializes writers, so one of N competing `listen`ers handles +//! each command and the row is deleted on success. +//! - **`publish` / `subscribe` (fan-out):** an append-only `bus_log` table plus +//! a per-consumer offset table (`bus_offset`). Each `group` reads independently +//! past its own offset, so every group sees every event. +//! +//! This is a no-extra-process transport for local development, tests, demos, and +//! small single-node deployments. It is not a high-throughput broker replacement. +//! +//! Requires the `sqlite` feature. Integration-tested in `tests/sqlite_transport`. + +use std::sync::Arc; +use std::time::Duration; + +use sqlx::sqlite::SqliteRow; +use sqlx::{QueryBuilder, Row, Sqlite, SqlitePool}; + +use crate::sqlx_repo::is_sqlx_transient; + +use super::source::{AsyncMessageSource, ReceivedMessage}; +use super::{ + run_source, Bus, BusConsumer, BusTopologyConfig, MessageRouter, RunOptions, TransportError, + TransportErrorKind, +}; +use super::{Message, MessageKind}; + +const DEFAULT_LEASE: Duration = Duration::from_secs(30); + +const SCHEMA: &str = "\ +CREATE TABLE IF NOT EXISTS bus_queue ( + seq INTEGER PRIMARY KEY AUTOINCREMENT, + claim_token TEXT, + name TEXT NOT NULL, + message_id TEXT, + kind TEXT NOT NULL, + payload BLOB NOT NULL, + content_type TEXT NOT NULL DEFAULT 'application/json', + metadata TEXT NOT NULL DEFAULT '[]', + available_at REAL NOT NULL DEFAULT (unixepoch('now','subsec')), + locked_until REAL, + attempts INTEGER NOT NULL DEFAULT 0, + CHECK (claim_token IS NULL OR claim_token <> ''), + CHECK (name <> ''), + CHECK (kind IN ('command', 'event')), + CHECK (content_type <> ''), + CHECK (attempts >= 0) +); +CREATE INDEX IF NOT EXISTS bus_queue_claim_idx + ON bus_queue (name, available_at, locked_until, seq); +CREATE TABLE IF NOT EXISTS bus_log ( + seq INTEGER PRIMARY KEY AUTOINCREMENT, + name TEXT NOT NULL, + message_id TEXT, + kind TEXT NOT NULL, + payload BLOB NOT NULL, + content_type TEXT NOT NULL DEFAULT 'application/json', + metadata TEXT NOT NULL DEFAULT '[]', + appended_at REAL NOT NULL DEFAULT (unixepoch('now','subsec')), + CHECK (name <> ''), + CHECK (kind IN ('command', 'event')), + CHECK (content_type <> '') +); +CREATE INDEX IF NOT EXISTS bus_log_name_seq_idx ON bus_log (name, seq); +CREATE TABLE IF NOT EXISTS bus_offset ( + consumer TEXT PRIMARY KEY, + last_seq INTEGER NOT NULL DEFAULT 0, + CHECK (consumer <> ''), + CHECK (last_seq >= 0) +)"; + +fn db_err(context: &str, err: sqlx::Error) -> TransportError { + let kind = if is_sqlx_transient(&err) { + TransportErrorKind::Retryable + } else { + TransportErrorKind::Permanent + }; + TransportError::new(kind, format!("sqlite bus {context}: {err}")).with_source(err) +} + +fn push_name_filter(query: &mut QueryBuilder, names: &[String]) { + query.push("(name IN ("); + { + let mut separated = query.separated(", "); + for name in names { + separated.push_bind(name.as_str()); + } + } + query.push(") OR name IS NULL)"); +} + +fn metadata_json(message: &Message) -> String { + serde_json::to_string(&message.metadata).unwrap_or_else(|_| "[]".into()) +} + +fn corrupt_row(message: impl Into) -> TransportError { + TransportError::permanent(format!("sqlite bus corrupt row: {}", message.into())) +} + +fn decode_err(column: &str, err: sqlx::Error) -> TransportError { + corrupt_row(format!( + "required column '{column}' failed to decode: {err}" + )) +} + +fn parse_message_kind(value: &str) -> Result { + match value { + "command" => Ok(MessageKind::Command), + "event" => Ok(MessageKind::Event), + _ => Err(corrupt_row(format!( + "required column 'kind' has unsupported value {value:?}" + ))), + } +} + +fn parse_metadata(value: &str) -> Result, TransportError> { + serde_json::from_str(value).map_err(|err| { + corrupt_row(format!( + "required column 'metadata' failed to parse as JSON metadata: {err}" + )) + }) +} + +/// Reconstruct a [`Message`] from a claimed `bus_queue`/`bus_log` row. +/// +/// Required-column decode/parsing failures are permanent corruption. The row has +/// already been selected or claimed, so callers surface the error through +/// [`ReceivedMessage::decode_error`] and let the runner settle it through the +/// configured failure policy. +fn message_from_row(row: &SqliteRow) -> Result { + let name: String = row.try_get("name").map_err(|err| decode_err("name", err))?; + let kind: String = row.try_get("kind").map_err(|err| decode_err("kind", err))?; + let payload: Vec = row + .try_get("payload") + .map_err(|err| decode_err("payload", err))?; + let content_type: String = row + .try_get("content_type") + .map_err(|err| decode_err("content_type", err))?; + if content_type.is_empty() { + return Err(corrupt_row("required column 'content_type' is empty")); + } + let metadata_json: String = row + .try_get("metadata") + .map_err(|err| decode_err("metadata", err))?; + let metadata = parse_metadata(&metadata_json)?; + + Ok(Message { + id: row + .try_get::, _>("message_id") + .unwrap_or(None), + name, + kind: parse_message_kind(&kind)?, + payload, + content_type, + metadata, + }) +} + +struct ReceivedRow { + seq: i64, + message: Message, + decode_error: Option, +} + +impl ReceivedRow { + fn from_row(row: &SqliteRow) -> Self { + let seq = row.try_get("seq").unwrap_or_default(); + let (message, decode_error) = decode_or_placeholder(row); + Self { + seq, + message, + decode_error, + } + } + + fn message(&self) -> &Message { + &self.message + } + + fn decode_error(&self) -> Option<&TransportError> { + self.decode_error.as_ref() + } +} + +/// SQLite [`Bus`] + [`BusConsumer`]. Cheap to clone (the pool is an `Arc`). +#[derive(Clone)] +pub struct SqliteBus { + pool: SqlitePool, + topology: BusTopologyConfig, + lease: Duration, +} + +impl SqliteBus { + /// Build a bus over an existing pool. + /// + /// For event subscriptions, `subscribe` uses the router's consumer identity + /// as the durable SQLite log offset. Service consumers usually get that + /// identity from [`Service::named`](crate::microsvc::Service::named). Direct + /// consumers can set it with [`group`](Self::group). Commands are claimed + /// from `bus_queue` by message name, so command replicas compete by listening + /// to the same registered command names. + pub fn new(pool: SqlitePool) -> Self { + Self { + pool, + topology: BusTopologyConfig::default(), + lease: DEFAULT_LEASE, + } + } + + /// Build a bus with an explicit group for direct/low-level use. + pub fn new_with_group(pool: SqlitePool, group: impl Into) -> Self { + Self::new(pool).group(group) + } + + /// Set an explicit durable event subscription group. + pub fn group(mut self, group: impl Into) -> Self { + self.topology = self.topology.group(group); + self + } + + /// Override the claim lease for `listen` (how long a claimed command stays + /// invisible to other workers before it is eligible for redelivery). + pub fn with_lease(mut self, lease: Duration) -> Self { + self.lease = lease; + self + } + + /// Create the bus tables (`bus_queue`, `bus_log`, `bus_offset`) if absent. + /// + /// Called by `listen`/`subscribe`; producers must ensure the tables exist + /// before `send`/`publish`, either by calling this or through migrations. + pub async fn ensure_tables(&self) -> Result<(), TransportError> { + for statement in SCHEMA.split(';') { + let statement = statement.trim(); + if statement.is_empty() { + continue; + } + sqlx::query(statement) + .execute(&self.pool) + .await + .map_err(|err| db_err("ensure_tables", err))?; + } + Ok(()) + } + + async fn enqueue(&self, message: Message) -> Result<(), TransportError> { + self.insert_message( + "INSERT INTO bus_queue (name, message_id, kind, payload, content_type, metadata) \ + VALUES (?, ?, ?, ?, ?, ?)", + "enqueue", + message, + ) + .await + } + + async fn append(&self, message: Message) -> Result<(), TransportError> { + self.insert_message( + "INSERT INTO bus_log (name, message_id, kind, payload, content_type, metadata) \ + VALUES (?, ?, ?, ?, ?, ?)", + "append", + message, + ) + .await + } + + async fn insert_message( + &self, + sql: &'static str, + context: &'static str, + message: Message, + ) -> Result<(), TransportError> { + let metadata = metadata_json(&message); + sqlx::query(sql) + .bind(&message.name) + .bind(&message.id) + .bind(message.kind.as_str()) + .bind(&message.payload) + .bind(&message.content_type) + .bind(metadata) + .execute(&self.pool) + .await + .map_err(|err| db_err(context, err))?; + Ok(()) + } +} + +impl Bus for SqliteBus { + async fn send(&self, name: &str, payload: Vec) -> Result<(), TransportError> { + self.enqueue(Message::new(name, MessageKind::Command, payload)) + .await + } + + async fn publish(&self, name: &str, payload: Vec) -> Result<(), TransportError> { + self.append(Message::new(name, MessageKind::Event, payload)) + .await + } + + async fn send_message(&self, message: Message) -> Result<(), TransportError> { + self.enqueue(message).await + } + + async fn publish_message(&self, message: Message) -> Result<(), TransportError> { + self.append(message).await + } +} + +impl BusConsumer for SqliteBus { + async fn listen( + &self, + router: Arc, + options: RunOptions, + ) -> Result<(), TransportError> { + self.ensure_tables().await?; + let names = router.subscription_plan().commands; + if names.is_empty() { + return Ok(()); + } + let source = QueueSource { + pool: self.pool.clone(), + names, + lease_secs: self.lease.as_secs_f64(), + }; + run_source(router, source, options).await + } + + async fn subscribe( + &self, + router: Arc, + options: RunOptions, + ) -> Result<(), TransportError> { + self.ensure_tables().await?; + let names = router.subscription_plan().events; + if names.is_empty() { + return Ok(()); + } + let group = self + .topology + .resolve_consumer_group(router.as_ref(), "sqlite")?; + let source = LogSource { + pool: self.pool.clone(), + names, + consumer: group, + }; + run_source(router, source, options).await + } +} + +/// Competing-consumer source over `bus_queue`. +struct QueueSource { + pool: SqlitePool, + names: Vec, + lease_secs: f64, +} + +impl AsyncMessageSource for QueueSource { + type Received = SqliteQueueReceived; + + async fn recv(&mut self) -> Result, TransportError> { + let mut query = QueryBuilder::::new( + "UPDATE bus_queue \ + SET locked_until = unixepoch('now','subsec') + ", + ); + query.push_bind(self.lease_secs); + query.push( + ", claim_token = lower(hex(randomblob(16))), \ + attempts = attempts + 1 \ + WHERE seq = ( \ + SELECT seq FROM bus_queue \ + WHERE ", + ); + push_name_filter(&mut query, &self.names); + query.push( + " AND available_at <= unixepoch('now','subsec') \ + AND (locked_until IS NULL OR locked_until <= unixepoch('now','subsec')) \ + ORDER BY seq LIMIT 1 \ + ) \ + RETURNING seq, claim_token, name, message_id, kind, payload, content_type, metadata", + ); + + let row = query + .build() + .fetch_optional(&self.pool) + .await + .map_err(|err| db_err("claim", err))?; + + if let Some(row) = row { + let claim_token = row + .try_get("claim_token") + .map_err(|err| db_err("claim token", err))?; + Ok(Some(SqliteQueueReceived { + pool: self.pool.clone(), + row: ReceivedRow::from_row(&row), + claim_token, + })) + } else { + Ok(None) + } + } +} + +fn decode_or_placeholder(row: &SqliteRow) -> (Message, Option) { + match message_from_row(row) { + Ok(message) => (message, None), + Err(error) => ( + Message::new("", MessageKind::Event, Vec::new()), + Some(error), + ), + } +} + +/// A claimed `bus_queue` row. +/// +/// `ack` deletes it, `nack` releases the lease for redelivery, and +/// `dead_letter`/`park` delete it so poison rows do not redeliver forever. +pub struct SqliteQueueReceived { + pool: SqlitePool, + row: ReceivedRow, + claim_token: String, +} + +impl SqliteQueueReceived { + async fn delete(&self) -> Result<(), TransportError> { + sqlx::query("DELETE FROM bus_queue WHERE seq = ? AND claim_token = ?") + .bind(self.row.seq) + .bind(&self.claim_token) + .execute(&self.pool) + .await + .map_err(|err| db_err("delete", err))?; + Ok(()) + } +} + +impl ReceivedMessage for SqliteQueueReceived { + fn message(&self) -> &Message { + self.row.message() + } + + fn decode_error(&self) -> Option<&TransportError> { + self.row.decode_error() + } + + async fn ack(self) -> Result<(), TransportError> { + self.delete().await + } + + async fn nack(self, _reason: &str) -> Result<(), TransportError> { + sqlx::query( + "UPDATE bus_queue \ + SET locked_until = NULL, claim_token = NULL \ + WHERE seq = ? AND claim_token = ?", + ) + .bind(self.row.seq) + .bind(&self.claim_token) + .execute(&self.pool) + .await + .map_err(|err| db_err("nack", err))?; + Ok(()) + } + + async fn dead_letter(self, _reason: &str) -> Result<(), TransportError> { + self.delete().await + } + + async fn park(self, _reason: &str) -> Result<(), TransportError> { + self.delete().await + } +} + +/// Fan-out source over `bus_log`. +struct LogSource { + pool: SqlitePool, + names: Vec, + consumer: String, +} + +impl AsyncMessageSource for LogSource { + type Received = SqliteLogReceived; + + async fn recv(&mut self) -> Result, TransportError> { + let mut query = QueryBuilder::::new( + "SELECT seq, name, message_id, kind, payload, content_type, metadata \ + FROM bus_log \ + WHERE ", + ); + push_name_filter(&mut query, &self.names); + query.push(" AND seq > COALESCE((SELECT last_seq FROM bus_offset WHERE consumer = "); + query.push_bind(&self.consumer); + query.push("), 0) ORDER BY seq LIMIT 1"); + + let row = query + .build() + .fetch_optional(&self.pool) + .await + .map_err(|err| db_err("log read", err))?; + + Ok(row.map(|row| SqliteLogReceived { + pool: self.pool.clone(), + consumer: self.consumer.clone(), + row: ReceivedRow::from_row(&row), + })) + } +} + +/// A `bus_log` entry. +/// +/// `ack` advances this consumer's offset, `nack` leaves the offset unchanged, +/// and `dead_letter`/`park` advance past poison entries. +pub struct SqliteLogReceived { + pool: SqlitePool, + consumer: String, + row: ReceivedRow, +} + +impl SqliteLogReceived { + async fn advance_offset(&self) -> Result<(), TransportError> { + sqlx::query( + "INSERT INTO bus_offset (consumer, last_seq) VALUES (?, ?) \ + ON CONFLICT (consumer) DO UPDATE SET last_seq = excluded.last_seq \ + WHERE bus_offset.last_seq < excluded.last_seq", + ) + .bind(&self.consumer) + .bind(self.row.seq) + .execute(&self.pool) + .await + .map_err(|err| db_err("advance offset", err))?; + Ok(()) + } +} + +impl ReceivedMessage for SqliteLogReceived { + fn message(&self) -> &Message { + self.row.message() + } + + fn decode_error(&self) -> Option<&TransportError> { + self.row.decode_error() + } + + async fn ack(self) -> Result<(), TransportError> { + self.advance_offset().await + } + + async fn nack(self, _reason: &str) -> Result<(), TransportError> { + Ok(()) + } + + async fn dead_letter(self, _reason: &str) -> Result<(), TransportError> { + self.advance_offset().await + } + + async fn park(self, _reason: &str) -> Result<(), TransportError> { + self.advance_offset().await + } +} diff --git a/tests/sqlite_transport/main.rs b/tests/sqlite_transport/main.rs new file mode 100644 index 00000000..bc98b2f9 --- /dev/null +++ b/tests/sqlite_transport/main.rs @@ -0,0 +1,618 @@ +//! SQLite bus transport integration tests. +//! +//! Exercises `SqliteBus` over a local SQLite file: command work-queue claims, +//! event log offsets, retry/nack behavior, and corrupt-row handling. +#![cfg(feature = "sqlite")] + +// Shared broker-test helpers (recording_for, named_recording_for). +#[path = "../transport_conformance/mod.rs"] +mod conformance; +use conformance::{named_recording_for, recording_for}; + +use std::path::PathBuf; +use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering}; +use std::sync::{Arc, Mutex}; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; + +use distributed::bus::{ + Bus, BusConsumer, Handlers, Message, MessageKind, RunOptions, SqliteBus, TransportError, +}; +use sqlx::sqlite::{SqliteConnectOptions, SqlitePoolOptions}; +use sqlx::SqlitePool; +use tokio::sync::Notify; + +static DB_SEQ: AtomicU64 = AtomicU64::new(0); + +const COMMAND_NAME: &str = "order.initialize"; +const EVENT_NAME: &str = "order.initialized"; +const PAYLOAD: &[u8] = b"{}"; + +struct TempDb { + path: PathBuf, +} + +impl TempDb { + fn new() -> Self { + let nanos = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_nanos(); + let seq = DB_SEQ.fetch_add(1, Ordering::Relaxed); + let mut path = std::env::temp_dir(); + path.push(format!("distributed_sqlite_bus_test_{nanos}_{seq}.db")); + Self { path } + } + + async fn pool(&self) -> SqlitePool { + self.pool_with_timeout(Duration::from_secs(5)).await + } + + async fn pool_with_timeout(&self, busy_timeout: Duration) -> SqlitePool { + let options = SqliteConnectOptions::new() + .filename(&self.path) + .create_if_missing(true) + .busy_timeout(busy_timeout); + SqlitePoolOptions::new() + .max_connections(5) + .connect_with(options) + .await + .expect("sqlite test pool") + } +} + +impl Drop for TempDb { + fn drop(&mut self) { + let _ = std::fs::remove_file(&self.path); + for suffix in ["-wal", "-shm"] { + let mut sidecar = self.path.clone(); + if let Some(file_name) = sidecar.file_name() { + let name = format!("{}{suffix}", file_name.to_string_lossy()); + sidecar.set_file_name(name); + let _ = std::fs::remove_file(sidecar); + } + } + } +} + +async fn bus() -> (TempDb, SqlitePool, SqliteBus) { + let db = TempDb::new(); + let pool = db.pool().await; + let bus = SqliteBus::new(pool.clone()).group("orders"); + bus.ensure_tables().await.expect("ensure tables"); + (db, pool, bus) +} + +fn command(id: impl Into) -> Message { + Message::new(COMMAND_NAME, MessageKind::Command, PAYLOAD.to_vec()).with_id(id) +} + +fn event(id: impl Into) -> Message { + Message::new(EVENT_NAME, MessageKind::Event, PAYLOAD.to_vec()).with_id(id) +} + +fn expected_ids(prefix: &str, total: usize) -> Vec { + (0..total).map(|i| format!("{prefix}{i}")).collect() +} + +fn recorded_ids(rec: &Arc>>) -> Vec { + let mut ids = rec.lock().unwrap().clone(); + ids.sort(); + ids +} + +async fn send_commands(bus: &SqliteBus, total: usize) { + for message in expected_ids("c", total).into_iter().map(command) { + bus.send_message(message).await.expect("send command"); + } +} + +async fn publish_events(bus: &SqliteBus, total: usize) { + for message in expected_ids("e", total).into_iter().map(event) { + bus.publish_message(message).await.expect("publish event"); + } +} + +async fn recreate_nullable_queue_table(pool: &SqlitePool) { + sqlx::query("DROP TABLE IF EXISTS bus_queue") + .execute(pool) + .await + .expect("drop bus_queue"); + sqlx::query( + r#" + CREATE TABLE bus_queue ( + seq INTEGER PRIMARY KEY AUTOINCREMENT, + claim_token TEXT, + name TEXT, + message_id TEXT, + kind TEXT NOT NULL, + payload BLOB NOT NULL, + content_type TEXT NOT NULL DEFAULT 'application/json', + metadata TEXT NOT NULL DEFAULT '[]', + available_at REAL NOT NULL DEFAULT (unixepoch('now','subsec')), + locked_until REAL, + attempts INTEGER NOT NULL DEFAULT 0 + ) + "#, + ) + .execute(pool) + .await + .expect("create nullable bus_queue"); + sqlx::query( + "CREATE INDEX bus_queue_claim_idx ON bus_queue (name, available_at, locked_until, seq)", + ) + .execute(pool) + .await + .expect("create queue index"); +} + +async fn recreate_nullable_log_table(pool: &SqlitePool) { + sqlx::query("DROP TABLE IF EXISTS bus_log") + .execute(pool) + .await + .expect("drop bus_log"); + sqlx::query( + r#" + CREATE TABLE bus_log ( + seq INTEGER PRIMARY KEY AUTOINCREMENT, + name TEXT, + message_id TEXT, + kind TEXT NOT NULL, + payload BLOB NOT NULL, + content_type TEXT DEFAULT 'application/json', + metadata TEXT NOT NULL DEFAULT '[]', + appended_at REAL NOT NULL DEFAULT (unixepoch('now','subsec')) + ) + "#, + ) + .execute(pool) + .await + .expect("create nullable bus_log"); + sqlx::query("CREATE INDEX bus_log_name_seq_idx ON bus_log (name, seq)") + .execute(pool) + .await + .expect("create log index"); +} + +async fn corrupt_latest_queue_name(pool: &SqlitePool) { + sqlx::query("UPDATE bus_queue SET name = NULL WHERE seq = (SELECT max(seq) FROM bus_queue)") + .execute(pool) + .await + .expect("null out queue name"); +} + +async fn corrupt_latest_queue_kind(pool: &SqlitePool) { + sqlx::query("UPDATE bus_queue SET kind = 'bogus' WHERE seq = (SELECT max(seq) FROM bus_queue)") + .execute(pool) + .await + .expect("corrupt queue kind"); +} + +async fn corrupt_latest_log_name(pool: &SqlitePool) { + sqlx::query("UPDATE bus_log SET name = NULL WHERE seq = (SELECT max(seq) FROM bus_log)") + .execute(pool) + .await + .expect("null out log name"); +} + +async fn corrupt_latest_log_kind(pool: &SqlitePool) { + sqlx::query("UPDATE bus_log SET kind = 'bogus' WHERE seq = (SELECT max(seq) FROM bus_log)") + .execute(pool) + .await + .expect("corrupt log kind"); +} + +async fn corrupt_latest_log_metadata(pool: &SqlitePool) { + sqlx::query( + "UPDATE bus_log SET metadata = 'not-json' WHERE seq = (SELECT max(seq) FROM bus_log)", + ) + .execute(pool) + .await + .expect("corrupt log metadata"); +} + +async fn corrupt_latest_log_content_type(pool: &SqlitePool) { + sqlx::query( + "UPDATE bus_log SET content_type = NULL WHERE seq = (SELECT max(seq) FROM bus_log)", + ) + .execute(pool) + .await + .expect("corrupt log content type"); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn bus_send_listen_is_point_to_point_across_a_group() { + let (_db, _pool, bus) = bus().await; + + let total = 6usize; + send_commands(&bus, total).await; + + let rec = Arc::new(Mutex::new(Vec::new())); + let bus_a = bus.clone(); + let bus_b = bus.clone(); + let (ra, rb) = tokio::join!( + bus_a.listen( + recording_for(COMMAND_NAME, MessageKind::Command, rec.clone()), + RunOptions::idempotent() + ), + bus_b.listen( + recording_for(COMMAND_NAME, MessageKind::Command, rec.clone()), + RunOptions::idempotent() + ), + ); + ra.expect("replica a drains"); + rb.expect("replica b drains"); + + assert_eq!( + recorded_ids(&rec), + expected_ids("c", total), + "every command handled exactly once across the group" + ); +} + +#[tokio::test] +async fn bus_publish_subscribe_fans_out_across_groups() { + let (_db, _pool, producer) = bus().await; + let total = 4usize; + publish_events(&producer, total).await; + + let expected = expected_ids("e", total); + for group in ["projections", "audit"] { + let bus = producer.clone().group(group); + let rec = Arc::new(Mutex::new(Vec::new())); + bus.subscribe( + recording_for(EVENT_NAME, MessageKind::Event, rec.clone()), + RunOptions::idempotent(), + ) + .await + .expect("subscriber drains"); + assert_eq!( + recorded_ids(&rec), + expected, + "group {group} sees every event" + ); + } +} + +#[tokio::test] +async fn bus_subscribe_uses_named_service_as_consumer_group() { + let (_db, pool, producer) = bus().await; + publish_events(&producer, 3).await; + + let rec = Arc::new(Mutex::new(Vec::new())); + SqliteBus::new(pool) + .subscribe( + named_recording_for( + "order-projection", + EVENT_NAME, + MessageKind::Event, + rec.clone(), + ), + RunOptions::idempotent(), + ) + .await + .expect("subscriber drains"); + + assert_eq!(recorded_ids(&rec), expected_ids("e", 3)); +} + +#[tokio::test] +async fn retryable_command_failure_redelivers_then_completes() { + let (_db, pool, bus) = bus().await; + bus.send_message(command("c1")).await.expect("send command"); + + let attempts = Arc::new(AtomicUsize::new(0)); + let seen = attempts.clone(); + let handlers = Arc::new( + Handlers::new().on_command(COMMAND_NAME, move |_: &Message| { + let seen = seen.clone(); + async move { + let previous = seen.fetch_add(1, Ordering::SeqCst); + if previous == 0 { + Err(TransportError::retryable("transient")) + } else { + Ok(()) + } + } + }), + ); + + bus.listen(handlers, RunOptions::idempotent()) + .await + .expect("listener drains after retry"); + + assert_eq!( + attempts.load(Ordering::SeqCst), + 2, + "message was retried once after nack" + ); + let remaining: i64 = sqlx::query_scalar("SELECT count(*) FROM bus_queue") + .fetch_one(&pool) + .await + .expect("count queue"); + assert_eq!(remaining, 0, "retried command was acked and deleted"); +} + +#[tokio::test] +async fn retryable_event_failure_does_not_advance_offset() { + let (_db, pool, bus) = bus().await; + bus.publish_message(event("e1")) + .await + .expect("publish event"); + + let attempts = Arc::new(AtomicUsize::new(0)); + let seen = attempts.clone(); + let handlers = Arc::new(Handlers::new().named("projections").on_event( + EVENT_NAME, + move |_: &Message| { + let seen = seen.clone(); + async move { + let previous = seen.fetch_add(1, Ordering::SeqCst); + if previous == 0 { + Err(TransportError::retryable("transient")) + } else { + Ok(()) + } + } + }, + )); + + SqliteBus::new(pool.clone()) + .subscribe(handlers, RunOptions::idempotent()) + .await + .expect("subscriber drains after retry"); + + assert_eq!( + attempts.load(Ordering::SeqCst), + 2, + "event was reread once because nack left the offset unmoved" + ); + let offset: Option = + sqlx::query_scalar("SELECT last_seq FROM bus_offset WHERE consumer = 'projections'") + .fetch_optional(&pool) + .await + .expect("read offset"); + assert_eq!(offset, Some(1), "offset advanced only after success"); +} + +#[tokio::test] +async fn bus_schema_rejects_unsupported_message_kind() { + let (_db, pool, _bus) = bus().await; + + let queue_err = sqlx::query("INSERT INTO bus_queue (name, kind, payload) VALUES (?, ?, ?)") + .bind(COMMAND_NAME) + .bind("bogus") + .bind(PAYLOAD.to_vec()) + .execute(&pool) + .await + .expect_err("queue kind check rejects unsupported message kind"); + assert!( + queue_err.to_string().contains("CHECK"), + "unexpected queue kind error: {queue_err}" + ); + + let log_err = sqlx::query("INSERT INTO bus_log (name, kind, payload) VALUES (?, ?, ?)") + .bind(EVENT_NAME) + .bind("bogus") + .bind(PAYLOAD.to_vec()) + .execute(&pool) + .await + .expect_err("log kind check rejects unsupported message kind"); + assert!( + log_err.to_string().contains("CHECK"), + "unexpected log kind error: {log_err}" + ); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn busy_or_locked_writer_contention_is_retryable() { + let db = TempDb::new(); + let setup_pool = db.pool().await; + SqliteBus::new(setup_pool.clone()) + .ensure_tables() + .await + .expect("ensure tables"); + + let lock_pool = db.pool_with_timeout(Duration::from_millis(0)).await; + let send_pool = db.pool_with_timeout(Duration::from_millis(0)).await; + let mut conn = lock_pool.acquire().await.expect("lock connection"); + sqlx::query("BEGIN EXCLUSIVE") + .execute(&mut *conn) + .await + .expect("begin exclusive"); + + let result = tokio::time::timeout( + Duration::from_secs(2), + SqliteBus::new(send_pool).send(COMMAND_NAME, PAYLOAD.to_vec()), + ) + .await + .expect("send should not hang behind busy lock"); + let err = result.expect_err("send should fail while database is locked"); + assert!( + err.is_retryable(), + "busy/locked contention must be retryable, got {err}" + ); + + sqlx::query("ROLLBACK") + .execute(&mut *conn) + .await + .expect("rollback exclusive lock"); +} + +#[tokio::test] +async fn bus_listen_dead_letters_corrupt_queue_row_not_silently() { + let (_db, pool, bus) = bus().await; + recreate_nullable_queue_table(&pool).await; + + bus.send_message(command("poison")) + .await + .expect("send poison"); + corrupt_latest_queue_name(&pool).await; + bus.send_message(command("poison-kind")) + .await + .expect("send poison kind"); + corrupt_latest_queue_kind(&pool).await; + bus.send_message(command("ok")).await.expect("send ok"); + + let rec = Arc::new(Mutex::new(Vec::new())); + bus.listen( + recording_for(COMMAND_NAME, MessageKind::Command, rec.clone()), + RunOptions::idempotent(), + ) + .await + .expect("listen drains without surfacing corrupt row as fatal"); + + assert_eq!( + recorded_ids(&rec), + vec!["ok".to_string()], + "only the valid row handled" + ); + let remaining: i64 = sqlx::query_scalar("SELECT count(*) FROM bus_queue") + .fetch_one(&pool) + .await + .expect("count queue"); + assert_eq!( + remaining, 0, + "corrupt row routed through policy, not redelivered forever" + ); +} + +#[tokio::test] +async fn bus_subscribe_dead_letters_corrupt_log_row_not_silently() { + let (_db, pool, bus) = bus().await; + recreate_nullable_log_table(&pool).await; + + bus.publish_message(event("poison")) + .await + .expect("publish leading poison"); + corrupt_latest_log_name(&pool).await; + bus.publish_message(event("ok")).await.expect("publish ok"); + bus.publish_message(event("poison-tail")) + .await + .expect("publish trailing poison"); + corrupt_latest_log_name(&pool).await; + bus.publish_message(event("poison-kind")) + .await + .expect("publish corrupt kind"); + corrupt_latest_log_kind(&pool).await; + bus.publish_message(event("poison-metadata")) + .await + .expect("publish corrupt metadata"); + corrupt_latest_log_metadata(&pool).await; + bus.publish_message(event("poison-content-type")) + .await + .expect("publish corrupt content type"); + corrupt_latest_log_content_type(&pool).await; + + let rec = Arc::new(Mutex::new(Vec::new())); + SqliteBus::new(pool.clone()) + .group("projections") + .subscribe( + recording_for(EVENT_NAME, MessageKind::Event, rec.clone()), + RunOptions::idempotent(), + ) + .await + .expect("subscribe drains past corrupt entries"); + + assert_eq!( + recorded_ids(&rec), + vec!["ok".to_string()], + "only the valid event handled" + ); + let offset: Option = + sqlx::query_scalar("SELECT last_seq FROM bus_offset WHERE consumer = 'projections'") + .fetch_optional(&pool) + .await + .expect("read offset"); + let max_seq: i64 = sqlx::query_scalar("SELECT max(seq) FROM bus_log") + .fetch_one(&pool) + .await + .expect("max seq"); + assert_eq!( + offset, + Some(max_seq), + "offset advanced past corrupt entries through the failure policy" + ); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn expired_queue_claim_cannot_be_settled_by_stale_worker() { + let (_db, pool, bus) = bus().await; + let bus = bus.with_lease(Duration::from_millis(250)); + bus.send_message(command("c1")).await.expect("send command"); + + let attempts = Arc::new(AtomicUsize::new(0)); + let first_claimed = Arc::new(Notify::new()); + let second_claimed = Arc::new(Notify::new()); + let allow_second_finish = Arc::new(Notify::new()); + + let handlers = Arc::new({ + let attempts = attempts.clone(); + let first_claimed = first_claimed.clone(); + let second_claimed = second_claimed.clone(); + let allow_second_finish = allow_second_finish.clone(); + Handlers::new().on_command(COMMAND_NAME, move |_: &Message| { + let attempt = attempts.fetch_add(1, Ordering::SeqCst); + let first_claimed = first_claimed.clone(); + let second_claimed = second_claimed.clone(); + let allow_second_finish = allow_second_finish.clone(); + async move { + match attempt { + 0 => { + first_claimed.notify_one(); + tokio::time::sleep(Duration::from_millis(420)).await; + Ok(()) + } + 1 => { + second_claimed.notify_one(); + allow_second_finish.notified().await; + Err(TransportError::retryable("second claim releases for retry")) + } + _ => Ok(()), + } + } + }) + }); + + let first = tokio::spawn({ + let bus = bus.clone(); + let handlers = handlers.clone(); + async move { bus.listen(handlers, RunOptions::idempotent()).await } + }); + tokio::time::timeout(Duration::from_secs(2), first_claimed.notified()) + .await + .expect("first worker claimed the command"); + + tokio::time::sleep(Duration::from_millis(300)).await; + let second = tokio::spawn({ + let bus = bus.clone(); + let handlers = handlers.clone(); + async move { bus.listen(handlers, RunOptions::idempotent()).await } + }); + tokio::time::timeout(Duration::from_secs(2), second_claimed.notified()) + .await + .expect("second worker reclaimed the expired lease"); + + tokio::time::timeout(Duration::from_secs(2), first) + .await + .expect("stale first worker finished") + .expect("first worker joined") + .expect("first listener drains"); + + allow_second_finish.notify_waiters(); + tokio::time::timeout(Duration::from_secs(2), second) + .await + .expect("second worker finished") + .expect("second worker joined") + .expect("second listener drains"); + + assert_eq!( + attempts.load(Ordering::SeqCst), + 3, + "stale ack did not delete the newer claim before it could be retried" + ); + let remaining: i64 = sqlx::query_scalar("SELECT count(*) FROM bus_queue") + .fetch_one(&pool) + .await + .expect("count queue"); + assert_eq!(remaining, 0, "retried command was eventually acked"); +}