From 2a8dea267809b69cfb9f1e4cb0d289fe77d5155b Mon Sep 17 00:00:00 2001 From: Patrick Lee Scott Date: Mon, 25 May 2026 05:00:05 -0500 Subject: [PATCH 1/4] Separate outbox worker store boundary Implements specs/outbox-worker-store-boundary --- README.md | 17 +- docs/async-repositories.md | 5 +- docs/read-models.md | 15 + src/bus/mod.rs | 3 +- src/hashmap_repo/mod.rs | 2 +- src/hashmap_repo/repository.rs | 24 +- src/lib.rs | 38 +- src/outbox/commit.rs | 4 +- src/outbox/mod.rs | 5 + src/outbox/table.rs | 213 ++++++++ src/outbox_worker/mod.rs | 19 +- .../{repository_ext.rs => store.rs} | 508 +++++++++++------- src/outbox_worker/thread.rs | 115 ++-- src/outbox_worker/worker.rs | 4 +- src/postgres_repo/mod.rs | 207 +++++-- src/repository/async_repository.rs | 56 +- src/repository/mod.rs | 6 +- src/sqlite_repo/mod.rs | 200 +++++-- src/sqlx_repo/mod.rs | 8 + src/table/mod.rs | 51 ++ src/table/sql.rs | 263 +++++++++ tests/async_repository/main.rs | 17 +- tests/bomberman/main.rs | 5 +- tests/distributed_read_model/main.rs | 5 +- tests/distributed_read_model_board/main.rs | 7 +- tests/hashmap_repository_conformance/main.rs | 18 +- tests/microsvc/convention.rs | 9 +- .../outbox.rs | 118 ++-- tests/postgres_repository_conformance/main.rs | 15 +- tests/sagas/distributed.rs | 28 +- tests/sagas/microsvc_saga.rs | 10 +- tests/sourced_snapshot/main.rs | 4 +- tests/sqlite_repository/main.rs | 34 +- tests/sqlite_repository_conformance/main.rs | 19 +- tests/todos/main.rs | 92 +++- 35 files changed, 1597 insertions(+), 547 deletions(-) create mode 100644 src/outbox/table.rs rename src/outbox_worker/{repository_ext.rs => store.rs} (51%) create mode 100644 src/table/mod.rs create mode 100644 src/table/sql.rs diff --git a/README.md b/README.md index 13721ea25..3f51f603e 100644 --- a/README.md +++ b/README.md @@ -675,27 +675,32 @@ let message = OutboxMessage::encode_for_entity( A separate process claims and publishes pending messages: ```rust -use sourced_rust::{LogPublisher, OutboxRepositoryExt, OutboxWorker}; +use sourced_rust::{ClaimOutboxMessages, LogPublisher, OutboxClaimRef, OutboxStore, OutboxWorker}; use std::time::Duration; let repo = HashMapRepository::new(); +let outbox = repo.outbox_store(); let worker_id = "worker-1"; let mut worker = OutboxWorker::new(LogPublisher::new()) .with_worker_id(worker_id) .with_max_attempts(3); -let claimed = repo.claim_outbox_messages(worker_id, 100, Duration::from_secs(30))?; +let mut claimed = outbox.claim(ClaimOutboxMessages::new(worker_id, 100, Duration::from_secs(30)))?; +let claims = claimed + .iter() + .map(OutboxClaimRef::from_message) + .collect::, _>>()?; -for mut message in claimed { - let result = worker.process_message(&mut message)?; +for (message, claim) in claimed.iter_mut().zip(claims.iter()) { + let result = worker.process_message(message)?; if result.completed { - repo.complete_outbox_message_for_worker(message.id(), worker_id)?; + outbox.complete(claim)?; } else if result.released || result.failed { let error = match message.last_error.as_deref() { Some(error) => error, None => "publish failed", }; - repo.record_outbox_publish_failure(message.id(), worker_id, error, 3)?; + outbox.record_failure(claim, error, 3)?; } } ``` diff --git a/docs/async-repositories.md b/docs/async-repositories.md index 22d44806d..912f51b1b 100644 --- a/docs/async-repositories.md +++ b/docs/async-repositories.md @@ -20,8 +20,9 @@ persistence should override it with an explicit durable name through `AsyncRelationalReadModelQueryStore` mirror the current document and relational read-model surfaces for async adapters. - `AsyncSnapshotStore` keys snapshots by full stream identity. -- `AsyncOutboxRepositoryExt` exposes async worker operations for durable outbox - implementations. +- `AsyncOutboxStore` exposes async claim/update operations for durable outbox + table stores. Aggregate repositories commit outbox rows transactionally, but + workers do not hydrate outbox messages through aggregate repositories. Async methods use an `_async` suffix where a synchronous method with the same name already exists. This keeps `HashMapRepository`, `InMemoryReadModelStore`, diff --git a/docs/read-models.md b/docs/read-models.md index 02bfbabaa..b7c41207e 100644 --- a/docs/read-models.md +++ b/docs/read-models.md @@ -256,6 +256,21 @@ schema changes should be generated or user-authored migrations plus verification; normal repository construction and command handling should not silently sync production schemas. +SQL repositories expose the same lifecycle through neutral table-schema APIs so +read models and operational tables, such as the outbox table, use one metadata +path: + +```rust +let mut registry = TableSchemaRegistry::new(); +registry.register_schema(outbox_message_schema())?; + +let artifacts = repo.generate_table_migration_artifacts(®istry)?; +let bootstrap = repo.bootstrap_table_schema_for_dev(®istry).await?; +``` + +Use generated artifacts as migration input for production tooling such as Atlas. +Reserve `bootstrap_table_schema_for_dev` for tests and local development. + ## Bomberman And Document Views Bomberman `BoardView` is intentionally a document-row read model. It stores a diff --git a/src/bus/mod.rs b/src/bus/mod.rs index f5daac0ef..90f522987 100644 --- a/src/bus/mod.rs +++ b/src/bus/mod.rs @@ -34,7 +34,8 @@ //! //! // 2. Worker drains outbox and publishes via bus //! let bus = Bus::new(kafka_publisher, kafka_subscriber); -//! for msg in repo.claim_outbox_messages(...) { +//! let outbox = repo.outbox_store(); +//! for msg in outbox.claim(...) { //! let event = Event::new(msg.id(), &msg.event_type, msg.payload.clone()); //! bus.publish(event)?; //! } diff --git a/src/hashmap_repo/mod.rs b/src/hashmap_repo/mod.rs index 615623653..f5936c0df 100644 --- a/src/hashmap_repo/mod.rs +++ b/src/hashmap_repo/mod.rs @@ -1,3 +1,3 @@ mod repository; -pub use repository::HashMapRepository; +pub use repository::{HashMapOutboxStore, HashMapRepository}; diff --git a/src/hashmap_repo/repository.rs b/src/hashmap_repo/repository.rs index b972f1181..7395b1de6 100644 --- a/src/hashmap_repo/repository.rs +++ b/src/hashmap_repo/repository.rs @@ -38,6 +38,12 @@ pub struct HashMapRepository { snapshot_store: InMemorySnapshotStore, } +/// In-memory outbox table handle. +#[derive(Clone)] +pub struct HashMapOutboxStore { + pub(crate) storage: Arc>>, +} + impl Default for HashMapRepository { fn default() -> Self { Self::new() @@ -55,10 +61,18 @@ impl HashMapRepository { } } - pub(crate) fn outbox_store(&self) -> &RwLock> { + #[cfg(test)] + pub(crate) fn outbox_storage(&self) -> &RwLock> { self.outbox_store.as_ref() } + /// Access the in-memory outbox table handle. + pub fn outbox_store(&self) -> HashMapOutboxStore { + HashMapOutboxStore { + storage: Arc::clone(&self.outbox_store), + } + } + /// Access the embedded read model store directly. pub fn model_store(&self) -> &InMemoryReadModelStore { &self.model_store @@ -375,6 +389,7 @@ fn reject_duplicate_async_streams(streams: &[AsyncStreamWrite<'_>]) -> Result<() fn reject_duplicate_outbox_messages(messages: &[OutboxMessage]) -> Result<(), RepositoryError> { let mut seen = HashSet::with_capacity(messages.len()); for message in messages { + validate_outbox_table_write(message)?; let id = message.id(); if id.trim().is_empty() { return Err(RepositoryError::Model( @@ -393,6 +408,13 @@ fn reject_duplicate_outbox_messages(messages: &[OutboxMessage]) -> Result<(), Re Ok(()) } +fn validate_outbox_table_write(message: &OutboxMessage) -> Result<(), RepositoryError> { + crate::outbox::outbox_message_insert_plan(message) + .and_then(|plan| plan.validate().map(|()| plan)) + .map(|_| ()) + .map_err(|err| RepositoryError::Model(err.to_string())) +} + fn validate_async_entity_id_matches_identity( streams: &[AsyncStreamWrite<'_>], ) -> Result<(), RepositoryError> { diff --git a/src/lib.rs b/src/lib.rs index 2cb657064..0536f20d7 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -26,6 +26,7 @@ pub mod snapshot; pub mod sqlite_repo; #[cfg(any(feature = "postgres", feature = "sqlite"))] mod sqlx_repo; +pub mod table; // Re-export entity types at crate root for convenience pub use entity::{ @@ -38,11 +39,11 @@ pub type SourcedResult = std::result::Result; // Re-export repository traits at crate root for convenience pub use repository::{ - AsyncCommitBatch, AsyncGetStream, AsyncOutboxRepositoryExt, AsyncReadModelSessionStore, - AsyncReadModelStore, AsyncRelationalReadModelQueryStore, AsyncRepository, AsyncSnapshotStore, - AsyncSnapshotWrite, AsyncStreamWrite, AsyncTransactionalCommit, Commit, CommitBatch, Get, - GetMany, GetOne, Gettable, PreparedEventAppend, Repository, RepositoryError, SnapshotWrite, - StreamIdentity, TransactionalCommit, + AsyncCommitBatch, AsyncGetStream, AsyncReadModelSessionStore, AsyncReadModelStore, + AsyncRelationalReadModelQueryStore, AsyncRepository, AsyncSnapshotStore, AsyncSnapshotWrite, + AsyncStreamWrite, AsyncTransactionalCommit, Commit, CommitBatch, Get, GetMany, GetOne, + Gettable, PreparedEventAppend, Repository, RepositoryError, SnapshotWrite, StreamIdentity, + TransactionalCommit, }; // Re-export aggregate types at crate root for convenience @@ -51,31 +52,35 @@ pub use aggregate::{ AsyncAggregateRepository, CommitAggregate, GetAggregate, GetAllAggregates, RepositoryExt, }; -pub use hashmap_repo::HashMapRepository; +pub use hashmap_repo::{HashMapOutboxStore, HashMapRepository}; #[cfg(feature = "postgres")] -pub use postgres_repo::PostgresRepository; +pub use postgres_repo::{PostgresOutboxStore, PostgresRepository}; #[cfg(feature = "sqlite")] -pub use sqlite_repo::SqliteRepository; +pub use sqlite_repo::{SqliteOutboxStore, SqliteRepository}; // Re-export lock traits and types at crate root for convenience pub use lock::{InMemoryLock, InMemoryLockManager, Lock, LockError, LockManager}; // Outbox: commit concerns (aggregate + outbox in one commit) pub use outbox::{ - AsyncOutboxCommit, OutboxCommit, OutboxCommitExt, OutboxMessage, OutboxMessageStatus, + outbox_message_insert_plan, outbox_message_key, outbox_message_row_values, + outbox_message_schema, AsyncOutboxCommit, OutboxCommit, OutboxCommitExt, OutboxMessage, + OutboxMessageStatus, OUTBOX_MESSAGES_TABLE, }; // Outbox Worker: drain and publish concerns pub use outbox_worker::{ + AsyncOutboxStore, + ClaimOutboxMessages, // Worker DrainResult, // Publishers LogPublisher, LogPublisherError, + OutboxClaimRef, OutboxPublishFailureAction, OutboxPublisher, - // Repository extension for claiming/completing messages - OutboxRepositoryExt, + OutboxStore, OutboxWorker, ProcessOneResult, }; @@ -123,6 +128,17 @@ pub use read_model::{ DEFAULT_READ_MODEL_VERSION_COLUMN, }; +// Neutral table/row primitives shared by read models and operational tables. +pub use table::{ + generate_table_migration_artifacts, table_schema_bootstrap_result, table_schema_statements, + DeleteTableRowMutation, PatchTableRowMutation, TableAdapterCapabilities, TableColumn, + TableCommitOutcome, TableDocumentMutation, TableIndex, TableMigrationArtifact, TableModel, + TableMutation, TableRowMutation, TableSchema, TableSchemaAdapter, + TableSchemaAdapterCapabilities, TableSchemaBootstrap, TableSchemaIssue, TableSchemaIssueKind, + TableSchemaRegistry, TableSchemaRegistryExt, TableSchemaVerification, TableSqlDialect, + TableSqlSchemaAdapter, TableStoreError, TableWritePlan, DEFAULT_TABLE_VERSION_COLUMN, +}; + // CommitBuilder: transactional batches of read models, outbox, and aggregates pub use commit_builder::{ CommitBuilder, CommitBuilderExt, ReadModelSessionCommitExt, StagedCommitBuilder, diff --git a/src/outbox/commit.rs b/src/outbox/commit.rs index b0c883d24..83dc5995e 100644 --- a/src/outbox/commit.rs +++ b/src/outbox/commit.rs @@ -87,7 +87,7 @@ mod tests { use super::*; use crate::{ impl_aggregate, AggregateBuilder, CommitBatch, Entity, EventRecord, HashMapRepository, - OutboxRepositoryExt, TransactionalCommit, + OutboxStore, TransactionalCommit, }; use std::cell::RefCell; @@ -145,7 +145,7 @@ mod tests { repo.outbox(event).commit(&mut aggregate).unwrap(); - let pending = repo.repo().outbox_messages_pending().unwrap(); + let pending = repo.repo().outbox_store().pending().unwrap(); assert_eq!(pending.len(), 1); assert_eq!(pending[0].id(), "msg-1"); } diff --git a/src/outbox/mod.rs b/src/outbox/mod.rs index 9a278eb1d..b83d3262a 100644 --- a/src/outbox/mod.rs +++ b/src/outbox/mod.rs @@ -39,9 +39,14 @@ mod commit; mod message; +mod table; // Outbox message record pub use message::{OutboxMessage, OutboxMessageStatus}; +pub use table::{ + outbox_message_insert_plan, outbox_message_key, outbox_message_row_values, + outbox_message_schema, OUTBOX_MESSAGES_TABLE, +}; // Commit helpers pub use commit::{AsyncOutboxCommit, OutboxCommit, OutboxCommitExt}; diff --git a/src/outbox/table.rs b/src/outbox/table.rs new file mode 100644 index 000000000..a6588bc46 --- /dev/null +++ b/src/outbox/table.rs @@ -0,0 +1,213 @@ +use std::time::{SystemTime, UNIX_EPOCH}; + +use crate::outbox::{OutboxMessage, OutboxMessageStatus}; +use crate::table::{ + ColumnType, ExpectedVersion, PrimaryKey, RowKey, RowValue, RowValues, RowWriteMode, + TableColumn, TableIndex, TableModel, TableMutation, TableRowMutation, TableSchema, + TableStoreError, TableWritePlan, +}; + +pub const OUTBOX_MESSAGES_TABLE: &str = "outbox_messages"; + +/// Schema for the durable outbox delivery table. +pub fn outbox_message_schema() -> TableSchema { + TableSchema { + model_name: "OutboxMessage".into(), + table_name: OUTBOX_MESSAGES_TABLE.into(), + columns: vec![ + table_column("message_id", ColumnType::Text, false), + table_column("event_type", ColumnType::Text, false), + table_column("payload", ColumnType::Bytes, false), + table_column("payload_codec", ColumnType::Text, false), + table_column("payload_codec_version", ColumnType::UnsignedInteger, false), + table_column("destination", ColumnType::Text, true), + table_column("metadata", ColumnType::Json, false), + table_column("status", ColumnType::Text, false), + table_column("created_at", ColumnType::UnsignedInteger, false), + table_column("next_available_at", ColumnType::UnsignedInteger, false), + table_column("claimed_by", ColumnType::Text, true), + table_column("claimed_until", ColumnType::UnsignedInteger, true), + table_column("attempts", ColumnType::UnsignedInteger, false), + table_column("last_error", ColumnType::Text, true), + table_column("published_at", ColumnType::UnsignedInteger, true), + table_column("failed_at", ColumnType::UnsignedInteger, true), + table_column("source_aggregate_type", ColumnType::Text, true), + table_column("source_aggregate_id", ColumnType::Text, true), + table_column("source_sequence", ColumnType::UnsignedInteger, true), + table_column("correlation_id", ColumnType::Text, true), + table_column("causation_id", ColumnType::Text, true), + ], + primary_key: PrimaryKey::new(["message_id"]), + version_column: None, + foreign_keys: Vec::new(), + indexes: vec![ + named_index( + "outbox_messages_claimable_idx", + ["status", "next_available_at", "claimed_until", "created_at"], + ), + named_index( + "outbox_messages_source_idx", + [ + "source_aggregate_type", + "source_aggregate_id", + "source_sequence", + ], + ), + named_index("outbox_messages_destination_idx", ["destination", "status"]), + ], + relationships: Vec::new(), + } +} + +pub fn outbox_message_insert_plan( + message: &OutboxMessage, +) -> Result { + let mutation = TableMutation::UpsertRow(TableRowMutation { + schema: outbox_message_schema(), + key: outbox_message_key(message.id()), + values: outbox_message_row_values(message)?, + expected_version: ExpectedVersion::NotExists, + mode: RowWriteMode::Insert, + }); + Ok(TableWritePlan::new(vec![mutation], Vec::new())) +} + +pub fn outbox_message_key(message_id: &str) -> RowKey { + RowKey::new([("message_id", RowValue::String(message_id.to_string()))]) +} + +pub fn outbox_message_row_values(message: &OutboxMessage) -> Result { + let mut row = RowValues::new(); + row.insert("message_id", RowValue::String(message.id().to_string())); + row.insert("event_type", RowValue::String(message.event_type.clone())); + row.insert("payload", RowValue::Bytes(message.payload.clone())); + row.insert( + "payload_codec", + RowValue::String(message.payload_codec.clone()), + ); + row.insert( + "payload_codec_version", + RowValue::U64(message.payload_codec_version as u64), + ); + row.insert("destination", optional_string(&message.destination)); + row.insert_serde("metadata", &message.metadata)?; + row.insert("status", RowValue::String(message.status.as_str().into())); + let created_at = system_time_epoch_secs(message.created_at)?; + row.insert("created_at", RowValue::U64(created_at)); + row.insert("next_available_at", RowValue::U64(created_at)); + row.insert("claimed_by", optional_string(&message.worker_id)); + row.insert( + "claimed_until", + optional_time_epoch_secs(message.leased_until)?, + ); + row.insert("attempts", RowValue::U64(message.attempts as u64)); + row.insert("last_error", optional_string(&message.last_error)); + row.insert("published_at", RowValue::Null); + row.insert("failed_at", status_failed_at(message)?); + row.insert( + "source_aggregate_type", + optional_string(&message.source_aggregate_type), + ); + row.insert( + "source_aggregate_id", + optional_string(&message.source_aggregate_id), + ); + row.insert("source_sequence", optional_u64(message.source_sequence)); + row.insert( + "correlation_id", + optional_str(message.metadata.get("correlation_id").map(String::as_str)), + ); + row.insert( + "causation_id", + optional_str(message.metadata.get("causation_id").map(String::as_str)), + ); + Ok(row) +} + +impl TableModel for OutboxMessage { + fn table_schema() -> TableSchema { + outbox_message_schema() + } + + fn table_key(&self) -> Result { + Ok(outbox_message_key(self.id())) + } + + fn to_table_row(&self) -> Result { + outbox_message_row_values(self) + } +} + +fn table_column(name: &str, column_type: ColumnType, nullable: bool) -> TableColumn { + let mut column = TableColumn::new(name, name, column_type); + column.nullable = nullable; + column +} + +fn named_index(name: &str, columns: impl IntoIterator) -> TableIndex { + let mut index = TableIndex::new(columns); + index.name = Some(name.into()); + index +} + +fn optional_string(value: &Option) -> RowValue { + optional_str(value.as_deref()) +} + +fn optional_str(value: Option<&str>) -> RowValue { + value + .map(|value| RowValue::String(value.to_string())) + .unwrap_or(RowValue::Null) +} + +fn optional_u64(value: Option) -> RowValue { + value.map(RowValue::U64).unwrap_or(RowValue::Null) +} + +fn optional_time_epoch_secs(value: Option) -> Result { + value + .map(system_time_epoch_secs) + .transpose() + .map(optional_u64) +} + +fn status_failed_at(message: &OutboxMessage) -> Result { + if message.status == OutboxMessageStatus::Failed { + Ok(RowValue::U64(system_time_epoch_secs(SystemTime::now())?)) + } else { + Ok(RowValue::Null) + } +} + +fn system_time_epoch_secs(value: SystemTime) -> Result { + value + .duration_since(UNIX_EPOCH) + .map(|duration| duration.as_secs()) + .map_err(|err| TableStoreError::Metadata(err.to_string())) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn outbox_insert_plan_uses_table_row_mutation() { + let message = OutboxMessage::create("msg-1", "Event", b"{}".to_vec()).unwrap(); + + let plan = outbox_message_insert_plan(&message).unwrap(); + + assert!(plan.processed_messages.is_empty()); + assert_eq!(plan.mutations.len(), 1); + let TableMutation::UpsertRow(mutation) = &plan.mutations[0] else { + panic!("outbox insert should lower to a table row mutation"); + }; + assert_eq!(mutation.schema.table_name, OUTBOX_MESSAGES_TABLE); + assert_eq!(mutation.key, outbox_message_key("msg-1")); + assert_eq!(mutation.expected_version, ExpectedVersion::NotExists); + assert_eq!(mutation.mode, RowWriteMode::Insert); + assert_eq!( + mutation.values.get("event_type"), + Some(&RowValue::String("Event".into())) + ); + } +} diff --git a/src/outbox_worker/mod.rs b/src/outbox_worker/mod.rs index f641323d8..f2e0e777e 100644 --- a/src/outbox_worker/mod.rs +++ b/src/outbox_worker/mod.rs @@ -1,7 +1,7 @@ //! Outbox Worker - Drains and publishes outbox messages. //! //! This module provides the worker infrastructure for processing outbox messages: -//! - `OutboxRepositoryExt` - Repository operations for claiming and completing messages +//! - `OutboxStore` - Store operations for claiming and completing messages //! - `OutboxWorker` - Synchronous message processor //! - `OutboxPublisher` - Trait for publishing to external systems //! - `LogPublisher` - Simple logging publisher for testing @@ -16,28 +16,29 @@ //! ## Example //! //! ```ignore -//! use sourced_rust::{OutboxWorker, OutboxRepositoryExt, LogPublisher}; +//! use sourced_rust::{ClaimOutboxMessages, OutboxClaimRef, OutboxStore, OutboxWorker, LogPublisher}; //! use std::time::Duration; //! //! // Claim pending messages //! let worker_id = "worker-1"; -//! let messages = repo.claim_outbox_messages(worker_id, 10, Duration::from_secs(60))?; +//! let messages = outbox.claim(ClaimOutboxMessages::new(worker_id, 10, Duration::from_secs(60)))?; //! //! // Process with a worker //! let mut worker = OutboxWorker::new(LogPublisher::default()).with_worker_id(worker_id); //! for mut msg in messages { +//! let claim = OutboxClaimRef::from_message(&msg)?; //! let result = worker.process_message(&mut msg)?; //! if result.completed { -//! repo.complete_outbox_message_for_worker(msg.id(), worker_id)?; +//! outbox.complete(&claim)?; //! } else if result.released || result.failed { //! let error = msg.last_error.as_deref().unwrap_or("publish failed"); -//! repo.record_outbox_publish_failure(msg.id(), worker_id, error, 3)?; +//! outbox.record_failure(&claim, error, 3)?; //! } //! } //! ``` mod publisher; -mod repository_ext; +mod store; #[cfg(feature = "bus")] mod thread; mod worker; @@ -49,8 +50,10 @@ pub use publisher::{LogPublisher, LogPublisherError, OutboxPublisher}; // Repository helpers #[cfg(any(feature = "postgres", feature = "sqlite"))] -pub(crate) use repository_ext::ensure_active_claim; -pub use repository_ext::{OutboxPublishFailureAction, OutboxRepositoryExt}; +pub(crate) use store::ensure_active_claim; +pub use store::{ + AsyncOutboxStore, ClaimOutboxMessages, OutboxClaimRef, OutboxPublishFailureAction, OutboxStore, +}; // Worker pub use worker::{DrainResult, OutboxWorker, ProcessOneResult}; diff --git a/src/outbox_worker/repository_ext.rs b/src/outbox_worker/store.rs similarity index 51% rename from src/outbox_worker/repository_ext.rs rename to src/outbox_worker/store.rs index 6c5b2e5e3..cd97082f8 100644 --- a/src/outbox_worker/repository_ext.rs +++ b/src/outbox_worker/store.rs @@ -6,9 +6,9 @@ use std::future::Future; use std::time::{Duration, SystemTime}; -use crate::hashmap_repo::HashMapRepository; +use crate::hashmap_repo::HashMapOutboxStore; use crate::outbox::{OutboxMessage, OutboxMessageStatus}; -use crate::repository::{AsyncOutboxRepositoryExt, RepositoryError}; +use crate::repository::RepositoryError; #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum OutboxPublishFailureAction { @@ -16,61 +16,138 @@ pub enum OutboxPublishFailureAction { Failed, } -/// Extension trait for repositories that expose outbox message operations. -pub trait OutboxRepositoryExt: Send + Sync { +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct ClaimOutboxMessages { + pub worker_id: String, + pub batch_size: usize, + pub lease: Duration, + pub destination: Option, +} + +impl ClaimOutboxMessages { + pub fn new(worker_id: impl Into, batch_size: usize, lease: Duration) -> Self { + Self { + worker_id: worker_id.into(), + batch_size, + lease, + destination: None, + } + } + + pub fn to_destination(mut self, destination: impl Into) -> Self { + self.destination = Some(destination.into()); + self + } +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct OutboxClaimRef { + pub message_id: String, + pub worker_id: String, + pub leased_until: SystemTime, + pub attempt: u32, +} + +impl OutboxClaimRef { + pub fn from_message(message: &OutboxMessage) -> Result { + let worker_id = message + .worker_id + .clone() + .ok_or_else(|| invalid_outbox_state(message, "outbox claim worker"))?; + let leased_until = message + .leased_until + .ok_or_else(|| invalid_outbox_state(message, "outbox claim lease"))?; + + Ok(Self { + message_id: message.id().to_string(), + worker_id, + leased_until, + attempt: message.attempts, + }) + } +} + +/// Store capability for claiming and updating durable outbox messages. +pub trait OutboxStore: Send + Sync { /// Return all outbox messages with the given status. - fn outbox_messages_by_status( + fn messages_by_status( &self, status: OutboxMessageStatus, ) -> Result, RepositoryError>; /// Return all pending outbox messages. - fn outbox_messages_pending(&self) -> Result, RepositoryError> { - self.outbox_messages_by_status(OutboxMessageStatus::Pending) + fn pending(&self) -> Result, RepositoryError> { + self.messages_by_status(OutboxMessageStatus::Pending) } /// Claim pending outbox messages for processing. - fn claim_outbox_messages( - &self, - worker_id: &str, - max: usize, - lease: Duration, - ) -> Result, RepositoryError>; + fn claim(&self, request: ClaimOutboxMessages) -> Result, RepositoryError>; /// Mark an outbox message as completed if it is still claimed by this worker. - fn complete_outbox_message_for_worker( - &self, - message_id: &str, - worker_id: &str, - ) -> Result<(), RepositoryError>; + fn complete(&self, claim: &OutboxClaimRef) -> Result<(), RepositoryError>; /// Release an outbox message if it is still claimed by this worker. - fn release_outbox_message_for_worker( - &self, - message_id: &str, - worker_id: &str, - error: &str, - ) -> Result<(), RepositoryError>; + fn release(&self, claim: &OutboxClaimRef, error: &str) -> Result<(), RepositoryError>; /// Mark an outbox message as permanently failed if it is still claimed by this worker. - fn fail_outbox_message_for_worker( - &self, - message_id: &str, - worker_id: &str, - error: &str, - ) -> Result<(), RepositoryError>; + fn fail(&self, claim: &OutboxClaimRef, error: &str) -> Result<(), RepositoryError>; /// Record a publish failure for a claimed message, releasing it for retry /// or permanently failing it when the attempt ceiling is reached. - fn record_outbox_publish_failure( + fn record_failure( &self, - message_id: &str, - worker_id: &str, + claim: &OutboxClaimRef, error: &str, max_attempts: u32, ) -> Result; } +/// Async store capability for claiming and updating durable outbox messages. +pub trait AsyncOutboxStore: Send + Sync { + fn messages_by_status_async( + &self, + status: OutboxMessageStatus, + ) -> impl Future, RepositoryError>> + Send + '_; + + fn pending_async( + &self, + ) -> impl Future, RepositoryError>> + Send + '_ { + async move { + self.messages_by_status_async(OutboxMessageStatus::Pending) + .await + } + } + + fn claim_async<'a>( + &'a self, + request: ClaimOutboxMessages, + ) -> impl Future, RepositoryError>> + Send + 'a; + + fn complete_async<'a>( + &'a self, + claim: &'a OutboxClaimRef, + ) -> impl Future> + Send + 'a; + + fn release_async<'a>( + &'a self, + claim: &'a OutboxClaimRef, + error: &'a str, + ) -> impl Future> + Send + 'a; + + fn fail_async<'a>( + &'a self, + claim: &'a OutboxClaimRef, + error: &'a str, + ) -> impl Future> + Send + 'a; + + fn record_failure_async<'a>( + &'a self, + claim: &'a OutboxClaimRef, + error: &'a str, + max_attempts: u32, + ) -> impl Future> + Send + 'a; +} + fn outbox_state(message: &OutboxMessage) -> String { format!( "{:?}, worker={:?}, leased_until={:?}, attempts={}", @@ -88,20 +165,27 @@ fn invalid_outbox_state(message: &OutboxMessage, expected: &'static str) -> Repo pub(crate) fn ensure_active_claim( message: &OutboxMessage, - worker_id: Option<&str>, + claim: Option<&OutboxClaimRef>, now: SystemTime, ) -> Result<(), RepositoryError> { if !message.is_in_flight() { return Err(invalid_outbox_state(message, "in-flight outbox message")); } - if let Some(worker_id) = worker_id { - if !message.is_claimed_by(worker_id) { + if let Some(claim) = claim { + if !message.is_claimed_by(&claim.worker_id) { return Err(invalid_outbox_state( message, "outbox claim held by requesting worker", )); } + + if message.attempts != claim.attempt { + return Err(invalid_outbox_state( + message, + "outbox claim attempt held by requesting worker", + )); + } } if message.has_expired_lease_at(now) { @@ -111,14 +195,14 @@ pub(crate) fn ensure_active_claim( Ok(()) } -impl HashMapRepository { +impl HashMapOutboxStore { fn update_outbox_message( &self, message_id: &str, update: impl FnOnce(&mut OutboxMessage) -> Result, ) -> Result { let mut storage = self - .outbox_store() + .storage .write() .map_err(|_| RepositoryError::LockPoisoned("outbox write"))?; @@ -132,13 +216,13 @@ impl HashMapRepository { } } -impl OutboxRepositoryExt for HashMapRepository { - fn outbox_messages_by_status( +impl OutboxStore for HashMapOutboxStore { + fn messages_by_status( &self, status: OutboxMessageStatus, ) -> Result, RepositoryError> { let storage = self - .outbox_store() + .storage .read() .map_err(|_| RepositoryError::LockPoisoned("outbox read"))?; @@ -155,18 +239,13 @@ impl OutboxRepositoryExt for HashMapRepository { Ok(messages) } - fn claim_outbox_messages( - &self, - worker_id: &str, - max: usize, - lease: Duration, - ) -> Result, RepositoryError> { + fn claim(&self, request: ClaimOutboxMessages) -> Result, RepositoryError> { let mut storage = self - .outbox_store() + .storage .write() .map_err(|_| RepositoryError::LockPoisoned("outbox write"))?; - if max == 0 { + if request.batch_size == 0 { return Ok(Vec::new()); } @@ -180,11 +259,16 @@ impl OutboxRepositoryExt for HashMapRepository { }; if message.is_claimable_at(now) { - message.claim_at(worker_id, lease, now)?; + if let Some(destination) = request.destination.as_deref() { + if message.destination.as_deref() != Some(destination) { + continue; + } + } + message.claim_at(&request.worker_id, request.lease, now)?; claimed.push(message.clone()); } - if claimed.len() >= max { + if claimed.len() >= request.batch_size { break; } } @@ -192,53 +276,38 @@ impl OutboxRepositoryExt for HashMapRepository { Ok(claimed) } - fn complete_outbox_message_for_worker( - &self, - message_id: &str, - worker_id: &str, - ) -> Result<(), RepositoryError> { - self.update_outbox_message(message_id, |message| { - ensure_active_claim(message, Some(worker_id), SystemTime::now())?; + fn complete(&self, claim: &OutboxClaimRef) -> Result<(), RepositoryError> { + self.update_outbox_message(&claim.message_id, |message| { + ensure_active_claim(message, Some(claim), SystemTime::now())?; message.complete()?; Ok(()) }) } - fn release_outbox_message_for_worker( - &self, - message_id: &str, - worker_id: &str, - error: &str, - ) -> Result<(), RepositoryError> { - self.update_outbox_message(message_id, |message| { - ensure_active_claim(message, Some(worker_id), SystemTime::now())?; + fn release(&self, claim: &OutboxClaimRef, error: &str) -> Result<(), RepositoryError> { + self.update_outbox_message(&claim.message_id, |message| { + ensure_active_claim(message, Some(claim), SystemTime::now())?; message.release(error.to_string())?; Ok(()) }) } - fn fail_outbox_message_for_worker( - &self, - message_id: &str, - worker_id: &str, - error: &str, - ) -> Result<(), RepositoryError> { - self.update_outbox_message(message_id, |message| { - ensure_active_claim(message, Some(worker_id), SystemTime::now())?; + fn fail(&self, claim: &OutboxClaimRef, error: &str) -> Result<(), RepositoryError> { + self.update_outbox_message(&claim.message_id, |message| { + ensure_active_claim(message, Some(claim), SystemTime::now())?; message.fail(error.to_string())?; Ok(()) }) } - fn record_outbox_publish_failure( + fn record_failure( &self, - message_id: &str, - worker_id: &str, + claim: &OutboxClaimRef, error: &str, max_attempts: u32, ) -> Result { - self.update_outbox_message(message_id, |message| { - ensure_active_claim(message, Some(worker_id), SystemTime::now())?; + self.update_outbox_message(&claim.message_id, |message| { + ensure_active_claim(message, Some(claim), SystemTime::now())?; if message.attempts >= max_attempts { message.fail(error.to_string())?; Ok(OutboxPublishFailureAction::Failed) @@ -250,14 +319,14 @@ impl OutboxRepositoryExt for HashMapRepository { } } -impl AsyncOutboxRepositoryExt for HashMapRepository { - fn outbox_messages_by_status_async( +impl AsyncOutboxStore for HashMapOutboxStore { + fn messages_by_status_async( &self, status: OutboxMessageStatus, ) -> impl Future, RepositoryError>> + Send + '_ { async move { let storage = self - .outbox_store() + .storage .read() .map_err(|_| RepositoryError::LockPoisoned("async outbox read"))?; let mut messages = storage @@ -270,20 +339,18 @@ impl AsyncOutboxRepositoryExt for HashMapRepository { } } - fn claim_outbox_messages_async<'a>( + fn claim_async<'a>( &'a self, - worker_id: &'a str, - max: usize, - lease: Duration, + request: ClaimOutboxMessages, ) -> impl Future, RepositoryError>> + Send + 'a { async move { - if max == 0 { + if request.batch_size == 0 { return Ok(Vec::new()); } let now = SystemTime::now(); let mut storage = self - .outbox_store() + .storage .write() .map_err(|_| RepositoryError::LockPoisoned("async outbox write"))?; let mut ids = storage.keys().cloned().collect::>(); @@ -299,10 +366,16 @@ impl AsyncOutboxRepositoryExt for HashMapRepository { continue; } - message.claim_at(worker_id, lease, now)?; + if let Some(destination) = request.destination.as_deref() { + if message.destination.as_deref() != Some(destination) { + continue; + } + } + + message.claim_at(&request.worker_id, request.lease, now)?; claimed.push(message.clone()); - if claimed.len() >= max { + if claimed.len() >= request.batch_size { break; } } @@ -311,89 +384,89 @@ impl AsyncOutboxRepositoryExt for HashMapRepository { } } - fn complete_outbox_message_for_worker_async<'a>( + fn complete_async<'a>( &'a self, - message_id: &'a str, - worker_id: &'a str, + claim: &'a OutboxClaimRef, ) -> impl Future> + Send + 'a { async move { let mut storage = self - .outbox_store() + .storage .write() .map_err(|_| RepositoryError::LockPoisoned("async outbox write"))?; - let message = storage - .get_mut(message_id) - .ok_or_else(|| RepositoryError::NotFound { - id: message_id.to_string(), - })?; - ensure_active_claim(message, Some(worker_id), SystemTime::now())?; + let message = + storage + .get_mut(&claim.message_id) + .ok_or_else(|| RepositoryError::NotFound { + id: claim.message_id.clone(), + })?; + ensure_active_claim(message, Some(claim), SystemTime::now())?; message.complete()?; Ok(()) } } - fn release_outbox_message_for_worker_async<'a>( + fn release_async<'a>( &'a self, - message_id: &'a str, - worker_id: &'a str, + claim: &'a OutboxClaimRef, error: &'a str, ) -> impl Future> + Send + 'a { async move { let mut storage = self - .outbox_store() + .storage .write() .map_err(|_| RepositoryError::LockPoisoned("async outbox write"))?; - let message = storage - .get_mut(message_id) - .ok_or_else(|| RepositoryError::NotFound { - id: message_id.to_string(), - })?; - ensure_active_claim(message, Some(worker_id), SystemTime::now())?; + let message = + storage + .get_mut(&claim.message_id) + .ok_or_else(|| RepositoryError::NotFound { + id: claim.message_id.clone(), + })?; + ensure_active_claim(message, Some(claim), SystemTime::now())?; message.release(error.to_string())?; Ok(()) } } - fn fail_outbox_message_for_worker_async<'a>( + fn fail_async<'a>( &'a self, - message_id: &'a str, - worker_id: &'a str, + claim: &'a OutboxClaimRef, error: &'a str, ) -> impl Future> + Send + 'a { async move { let mut storage = self - .outbox_store() + .storage .write() .map_err(|_| RepositoryError::LockPoisoned("async outbox write"))?; - let message = storage - .get_mut(message_id) - .ok_or_else(|| RepositoryError::NotFound { - id: message_id.to_string(), - })?; - ensure_active_claim(message, Some(worker_id), SystemTime::now())?; + let message = + storage + .get_mut(&claim.message_id) + .ok_or_else(|| RepositoryError::NotFound { + id: claim.message_id.clone(), + })?; + ensure_active_claim(message, Some(claim), SystemTime::now())?; message.fail(error.to_string())?; Ok(()) } } - fn record_outbox_publish_failure_async<'a>( + fn record_failure_async<'a>( &'a self, - message_id: &'a str, - worker_id: &'a str, + claim: &'a OutboxClaimRef, error: &'a str, max_attempts: u32, ) -> impl Future> + Send + 'a { async move { let mut storage = self - .outbox_store() + .storage .write() .map_err(|_| RepositoryError::LockPoisoned("async outbox write"))?; - let message = storage - .get_mut(message_id) - .ok_or_else(|| RepositoryError::NotFound { - id: message_id.to_string(), - })?; - ensure_active_claim(message, Some(worker_id), SystemTime::now())?; + let message = + storage + .get_mut(&claim.message_id) + .ok_or_else(|| RepositoryError::NotFound { + id: claim.message_id.clone(), + })?; + ensure_active_claim(message, Some(claim), SystemTime::now())?; if message.attempts >= max_attempts { message.fail(error.to_string())?; Ok(OutboxPublishFailureAction::Failed) @@ -408,7 +481,7 @@ impl AsyncOutboxRepositoryExt for HashMapRepository { #[cfg(test)] mod tests { use super::*; - use crate::TransactionalCommit; + use crate::{HashMapRepository, TransactionalCommit}; use std::sync::{Arc, Barrier}; use std::thread; @@ -421,7 +494,12 @@ mod tests { } fn load_message(repo: &HashMapRepository, id: &str) -> OutboxMessage { - repo.outbox_store().read().unwrap().get(id).unwrap().clone() + repo.outbox_storage() + .read() + .unwrap() + .get(id) + .unwrap() + .clone() } #[test] @@ -433,8 +511,13 @@ mod tests { .unwrap(); let id = store_message(&repo, message); - let claimed = repo - .claim_outbox_messages("worker-2", 1, Duration::from_secs(60)) + let store = repo.outbox_store(); + let claimed = store + .claim(ClaimOutboxMessages::new( + "worker-2", + 1, + Duration::from_secs(60), + )) .unwrap(); assert_eq!(claimed.len(), 1); @@ -456,8 +539,13 @@ mod tests { .unwrap(); let id = store_message(&repo, message); - let claimed = repo - .claim_outbox_messages("worker-2", 1, Duration::from_secs(60)) + let store = repo.outbox_store(); + let claimed = store + .claim(ClaimOutboxMessages::new( + "worker-2", + 1, + Duration::from_secs(60), + )) .unwrap(); assert!(claimed.is_empty()); @@ -473,22 +561,30 @@ mod tests { let id = store_message(&repo, message); let barrier = Arc::new(Barrier::new(3)); - let repo_a = repo.clone(); - let repo_b = repo.clone(); + let store_a = repo.outbox_store(); + let store_b = repo.outbox_store(); let barrier_a = Arc::clone(&barrier); let barrier_b = Arc::clone(&barrier); let worker_a = thread::spawn(move || { barrier_a.wait(); - repo_a - .claim_outbox_messages("worker-a", 1, Duration::from_secs(60)) + store_a + .claim(ClaimOutboxMessages::new( + "worker-a", + 1, + Duration::from_secs(60), + )) .unwrap() .len() }); let worker_b = thread::spawn(move || { barrier_b.wait(); - repo_b - .claim_outbox_messages("worker-b", 1, Duration::from_secs(60)) + store_b + .claim(ClaimOutboxMessages::new( + "worker-b", + 1, + Duration::from_secs(60), + )) .unwrap() .len() }); @@ -508,11 +604,16 @@ mod tests { let message = OutboxMessage::create("msg-1", "Event", b"{}".to_vec()).unwrap(); let id = store_message(&repo, message); - repo.claim_outbox_messages("worker-1", 1, Duration::from_secs(60)) - .unwrap(); - let action = repo - .record_outbox_publish_failure(&id, "worker-1", "first failure", 2) + let store = repo.outbox_store(); + let claimed = store + .claim(ClaimOutboxMessages::new( + "worker-1", + 1, + Duration::from_secs(60), + )) .unwrap(); + let claim = OutboxClaimRef::from_message(&claimed[0]).unwrap(); + let action = store.record_failure(&claim, "first failure", 2).unwrap(); assert_eq!(action, OutboxPublishFailureAction::Released); let stored = load_message(&repo, &id); @@ -520,11 +621,15 @@ mod tests { assert_eq!(stored.attempts, 1); assert_eq!(stored.last_error.as_deref(), Some("first failure")); - repo.claim_outbox_messages("worker-1", 1, Duration::from_secs(60)) - .unwrap(); - let action = repo - .record_outbox_publish_failure(&id, "worker-1", "second failure", 2) + let claimed = store + .claim(ClaimOutboxMessages::new( + "worker-1", + 1, + Duration::from_secs(60), + )) .unwrap(); + let claim = OutboxClaimRef::from_message(&claimed[0]).unwrap(); + let action = store.record_failure(&claim, "second failure", 2).unwrap(); assert_eq!(action, OutboxPublishFailureAction::Failed); let stored = load_message(&repo, &id); @@ -535,39 +640,41 @@ mod tests { #[test] fn missing_message_updates_return_not_found() { - let repo = HashMapRepository::new(); + let store = HashMapOutboxStore { + storage: Default::default(), + }; let expected = RepositoryError::NotFound { id: "missing".into(), }; + let claim = OutboxClaimRef { + message_id: "missing".into(), + worker_id: "worker-1".into(), + leased_until: SystemTime::now(), + attempt: 1, + }; - assert_eq!( - repo.complete_outbox_message_for_worker("missing", "worker-1") - .unwrap_err(), - expected - ); - assert_eq!( - repo.release_outbox_message_for_worker("missing", "worker-1", "error") - .unwrap_err(), - expected - ); - assert_eq!( - repo.fail_outbox_message_for_worker("missing", "worker-1", "error") - .unwrap_err(), - expected - ); + assert_eq!(store.complete(&claim).unwrap_err(), expected); + assert_eq!(store.release(&claim, "error").unwrap_err(), expected); + assert_eq!(store.fail(&claim, "error").unwrap_err(), expected); } #[test] fn stale_or_mismatched_claims_cannot_be_completed() { let repo = HashMapRepository::new(); let message = OutboxMessage::create("msg-1", "Event", b"{}".to_vec()).unwrap(); - let id = store_message(&repo, message); - - repo.claim_outbox_messages("worker-1", 1, Duration::from_secs(60)) + let _id = store_message(&repo, message); + + let store = repo.outbox_store(); + let claimed = store + .claim(ClaimOutboxMessages::new( + "worker-1", + 1, + Duration::from_secs(60), + )) .unwrap(); - let err = repo - .complete_outbox_message_for_worker(&id, "worker-2") - .unwrap_err(); + let mut claim = OutboxClaimRef::from_message(&claimed[0]).unwrap(); + claim.worker_id = "worker-2".into(); + let err = store.complete(&claim).unwrap_err(); assert!(matches!(err, RepositoryError::InvalidState { .. })); let mut expired = OutboxMessage::create("msg-2", "Event", b"{}".to_vec()).unwrap(); @@ -575,26 +682,61 @@ mod tests { .claim_at("worker-1", Duration::from_secs(1), SystemTime::UNIX_EPOCH) .unwrap(); let expired_id = store_message(&repo, expired); - let err = repo - .complete_outbox_message_for_worker(&expired_id, "worker-1") - .unwrap_err(); + let expired = load_message(&repo, &expired_id); + let claim = OutboxClaimRef::from_message(&expired).unwrap(); + let err = store.complete(&claim).unwrap_err(); assert!(matches!(err, RepositoryError::InvalidState { .. })); } #[test] - fn already_published_message_is_not_completed_again() { + fn stale_attempt_claims_cannot_complete_later_claims() { let repo = HashMapRepository::new(); let message = OutboxMessage::create("msg-1", "Event", b"{}".to_vec()).unwrap(); - let id = store_message(&repo, message); - - repo.claim_outbox_messages("worker-1", 1, Duration::from_secs(60)) + let _id = store_message(&repo, message); + + let store = repo.outbox_store(); + let claimed = store + .claim(ClaimOutboxMessages::new( + "worker-1", + 1, + Duration::from_secs(60), + )) + .unwrap(); + let stale_claim = OutboxClaimRef::from_message(&claimed[0]).unwrap(); + store.release(&stale_claim, "retry").unwrap(); + + let claimed = store + .claim(ClaimOutboxMessages::new( + "worker-1", + 1, + Duration::from_secs(60), + )) .unwrap(); - repo.complete_outbox_message_for_worker(&id, "worker-1") + let current_claim = OutboxClaimRef::from_message(&claimed[0]).unwrap(); + + let err = store.complete(&stale_claim).unwrap_err(); + assert!(matches!(err, RepositoryError::InvalidState { .. })); + store.complete(¤t_claim).unwrap(); + } + + #[test] + fn already_published_message_is_not_completed_again() { + let repo = HashMapRepository::new(); + let message = OutboxMessage::create("msg-1", "Event", b"{}".to_vec()).unwrap(); + let _id = store_message(&repo, message); + + let store = repo.outbox_store(); + let claimed = store + .claim(ClaimOutboxMessages::new( + "worker-1", + 1, + Duration::from_secs(60), + )) .unwrap(); + let claim = OutboxClaimRef::from_message(&claimed[0]).unwrap(); + store.complete(&claim).unwrap(); - let err = repo - .complete_outbox_message_for_worker(&id, "worker-1") - .unwrap_err(); + let err = store.complete(&claim).unwrap_err(); assert!(matches!(err, RepositoryError::InvalidState { .. })); } } diff --git a/src/outbox_worker/thread.rs b/src/outbox_worker/thread.rs index d8ee7d06d..22a9abbf6 100644 --- a/src/outbox_worker/thread.rs +++ b/src/outbox_worker/thread.rs @@ -9,7 +9,7 @@ use std::time::Duration; use std::{error::Error, fmt}; use crate::bus::{Event, Publisher, Sender as BusSender}; -use crate::OutboxRepositoryExt; +use crate::{ClaimOutboxMessages, OutboxClaimRef, OutboxStore}; const DEFAULT_MAX_ATTEMPTS: u32 = 3; @@ -33,37 +33,39 @@ impl fmt::Display for OutboxWorkerJoinError { impl Error for OutboxWorkerJoinError {} -fn record_publish_success( - repo: &R, - message_id: &str, - worker_id: &str, +fn record_publish_success( + store: &S, + claim: &OutboxClaimRef, stats: &mut WorkerStats, ) { - match repo.complete_outbox_message_for_worker(message_id, worker_id) { + match store.complete(claim) { Ok(()) => { stats.messages_published += 1; } Err(err) => { - eprintln!("outbox worker `{worker_id}` could not complete `{message_id}`: {err}"); + eprintln!( + "outbox worker `{}` could not complete `{}`: {err}", + claim.worker_id, claim.message_id + ); stats.messages_failed += 1; } } } -fn record_publish_failure( - repo: &R, - message_id: &str, - worker_id: &str, +fn record_publish_failure( + store: &S, + claim: &OutboxClaimRef, error: &str, stats: &mut WorkerStats, ) { - match repo.record_outbox_publish_failure(message_id, worker_id, error, DEFAULT_MAX_ATTEMPTS) { + match store.record_failure(claim, error, DEFAULT_MAX_ATTEMPTS) { Ok(_) => { stats.messages_failed += 1; } Err(err) => { eprintln!( - "outbox worker `{worker_id}` could not record publish failure for `{message_id}`: {err}" + "outbox worker `{}` could not record publish failure for `{}`: {err}", + claim.worker_id, claim.message_id ); stats.messages_failed += 1; } @@ -84,7 +86,7 @@ fn record_publish_failure( /// /// // Start the worker /// let worker = OutboxWorkerThread::spawn( -/// repo.clone(), +/// repo.outbox_store(), /// publisher, /// Duration::from_millis(50), /// ); @@ -103,28 +105,28 @@ pub struct OutboxWorkerThread { impl OutboxWorkerThread { /// Spawn a new outbox worker thread. /// - /// The worker will poll the repository for pending outbox messages, + /// The worker will poll the outbox store for pending outbox messages, /// publish them to the given publisher, and mark them as complete. /// - /// The repository must be `Clone + Send + 'static`. For `HashMapRepository`, - /// cloning creates another handle to the same storage (thread-safe via `Arc>`). - pub fn spawn(repo: R, publisher: P, poll_interval: Duration) -> Self + /// The store must be `Clone + Send + 'static`. For `HashMapOutboxStore`, + /// cloning creates another handle to the same storage. + pub fn spawn(store: S, publisher: P, poll_interval: Duration) -> Self where - R: OutboxRepositoryExt + Clone + Send + 'static, + S: OutboxStore + Clone + Send + 'static, P: Publisher + 'static, { - Self::spawn_with_id(repo, publisher, poll_interval, "outbox-worker") + Self::spawn_with_id(store, publisher, poll_interval, "outbox-worker") } /// Spawn a new outbox worker thread with a custom worker ID. - pub fn spawn_with_id( - repo: R, + pub fn spawn_with_id( + store: S, publisher: P, poll_interval: Duration, worker_id: &str, ) -> Self where - R: OutboxRepositoryExt + Clone + Send + 'static, + S: OutboxStore + Clone + Send + 'static, P: Publisher + 'static, { let (stop_tx, stop_rx) = channel(); @@ -144,9 +146,20 @@ impl OutboxWorkerThread { stats.polls += 1; // Claim and process messages - match repo.claim_outbox_messages(&worker_id, 100, lease) { + match store.claim(ClaimOutboxMessages::new(&worker_id, 100, lease)) { Ok(messages) => { for msg in messages { + let claim = match OutboxClaimRef::from_message(&msg) { + Ok(claim) => claim, + Err(err) => { + eprintln!( + "outbox worker `{worker_id}` received invalid claim `{}`: {err}", + msg.id() + ); + stats.messages_failed += 1; + continue; + } + }; let mut event = Event::new(msg.id(), &msg.event_type, msg.payload.clone()); for (k, v) in &msg.metadata { @@ -155,17 +168,11 @@ impl OutboxWorkerThread { match publisher.publish(event) { Ok(()) => { - record_publish_success(&repo, msg.id(), &worker_id, &mut stats); + record_publish_success(&store, &claim, &mut stats); } Err(err) => { let error = err.to_string(); - record_publish_failure( - &repo, - msg.id(), - &worker_id, - &error, - &mut stats, - ); + record_publish_failure(&store, &claim, &error, &mut stats); } } } @@ -191,23 +198,23 @@ impl OutboxWorkerThread { /// /// Messages with a `destination` are sent point-to-point via `Sender::send()`. /// Messages without a destination are published fan-out via `Publisher::publish()`. - pub fn spawn_routed(repo: R, publisher: P, poll_interval: Duration) -> Self + pub fn spawn_routed(store: S, publisher: P, poll_interval: Duration) -> Self where - R: OutboxRepositoryExt + Clone + Send + 'static, + S: OutboxStore + Clone + Send + 'static, P: Publisher + BusSender + 'static, { - Self::spawn_routed_with_id(repo, publisher, poll_interval, "outbox-worker") + Self::spawn_routed_with_id(store, publisher, poll_interval, "outbox-worker") } /// Spawn a routed worker with a custom worker ID. - pub fn spawn_routed_with_id( - repo: R, + pub fn spawn_routed_with_id( + store: S, publisher: P, poll_interval: Duration, worker_id: &str, ) -> Self where - R: OutboxRepositoryExt + Clone + Send + 'static, + S: OutboxStore + Clone + Send + 'static, P: Publisher + BusSender + 'static, { let (stop_tx, stop_rx) = channel(); @@ -225,9 +232,20 @@ impl OutboxWorkerThread { stats.polls += 1; - match repo.claim_outbox_messages(&worker_id, 100, lease) { + match store.claim(ClaimOutboxMessages::new(&worker_id, 100, lease)) { Ok(messages) => { for msg in messages { + let claim = match OutboxClaimRef::from_message(&msg) { + Ok(claim) => claim, + Err(err) => { + eprintln!( + "outbox worker `{worker_id}` received invalid claim `{}`: {err}", + msg.id() + ); + stats.messages_failed += 1; + continue; + } + }; let mut event = Event::new(msg.id(), &msg.event_type, msg.payload.clone()); for (k, v) in &msg.metadata { @@ -242,17 +260,11 @@ impl OutboxWorkerThread { match result { Ok(()) => { - record_publish_success(&repo, msg.id(), &worker_id, &mut stats); + record_publish_success(&store, &claim, &mut stats); } Err(err) => { let error = err.to_string(); - record_publish_failure( - &repo, - msg.id(), - &worker_id, - &error, - &mut stats, - ); + record_publish_failure(&store, &claim, &error, &mut stats); } } } @@ -324,7 +336,12 @@ mod tests { } fn load_message(repo: &HashMapRepository, id: &str) -> OutboxMessage { - repo.outbox_store().read().unwrap().get(id).unwrap().clone() + repo.outbox_storage() + .read() + .unwrap() + .get(id) + .unwrap() + .clone() } #[test] @@ -375,7 +392,7 @@ mod tests { let id = store_message(&repo, message); let worker = OutboxWorkerThread::spawn_with_id( - repo.clone(), + repo.outbox_store(), FailingPublisher, Duration::from_millis(1), "worker-1", diff --git a/src/outbox_worker/worker.rs b/src/outbox_worker/worker.rs index 0cf1b4ad6..6edc342be 100644 --- a/src/outbox_worker/worker.rs +++ b/src/outbox_worker/worker.rs @@ -91,8 +91,8 @@ impl OutboxWorker

{ /// Process a single outbox message. /// /// If the message is pending, it will be claimed by this worker before - /// publishing. Repository-backed worker loops should persist the outcome - /// with the outbox repository completion or failure APIs. + /// publishing. Store-backed worker loops should persist the outcome with + /// the outbox store completion or failure APIs. pub fn process_message( &mut self, message: &mut OutboxMessage, diff --git a/src/postgres_repo/mod.rs b/src/postgres_repo/mod.rs index 0e7342f55..f476e1245 100644 --- a/src/postgres_repo/mod.rs +++ b/src/postgres_repo/mod.rs @@ -18,11 +18,13 @@ use sqlx::{PgPool, Postgres, Row, Transaction}; use crate::entity::Entity; use crate::entity::EventRecord; use crate::outbox::{OutboxMessage, OutboxMessageStatus}; -use crate::outbox_worker::{ensure_active_claim, OutboxPublishFailureAction}; +use crate::outbox_worker::{ + ensure_active_claim, AsyncOutboxStore, ClaimOutboxMessages, OutboxClaimRef, + OutboxPublishFailureAction, +}; use crate::repository::{ - AsyncCommitBatch, AsyncGetStream, AsyncOutboxRepositoryExt, AsyncSnapshotStore, - AsyncSnapshotWrite, AsyncTransactionalCommit, PreparedEventAppend, RepositoryError, - StreamIdentity, + AsyncCommitBatch, AsyncGetStream, AsyncSnapshotStore, AsyncSnapshotWrite, + AsyncTransactionalCommit, PreparedEventAppend, RepositoryError, StreamIdentity, }; use crate::snapshot::SnapshotRecord; use crate::sqlx_repo::{ @@ -36,6 +38,11 @@ use crate::sqlx_repo::{ validate_entity_id_matches_identity, validate_prepared_appends, validate_snapshot_identity, validate_supported_event_codec, }; +use crate::table::{ + generate_table_migration_artifacts, table_schema_bootstrap_result, table_schema_statements, + TableMigrationArtifact, TableSchemaBootstrap, TableSchemaRegistry, TableSqlDialect, + TableSqlSchemaAdapter, TableStoreError, +}; const POSTGRES_SCHEMA: &str = include_str!("../../migrations/postgres/0001_initial.sql"); const POSTGRES_BACKEND: &str = "postgres"; @@ -48,6 +55,12 @@ pub struct PostgresRepository { pool: PgPool, } +/// Postgres-backed outbox table store. +#[derive(Clone)] +pub struct PostgresOutboxStore { + pool: PgPool, +} + impl PostgresRepository { /// Create a repository from an existing migrated pool. pub fn new(pool: PgPool) -> Self { @@ -95,6 +108,77 @@ impl PostgresRepository { pub fn pool(&self) -> &PgPool { &self.pool } + + /// SQL artifact adapter for registered table/read-model schemas. + pub fn table_schema_adapter(&self) -> TableSqlSchemaAdapter { + TableSqlSchemaAdapter::postgres() + } + + /// Generate SQL statements for registered table/read-model schemas. + pub fn generate_table_migration_artifacts( + &self, + registry: &TableSchemaRegistry, + ) -> Result, TableStoreError> { + generate_table_migration_artifacts(registry, TableSqlDialect::Postgres) + } + + /// Explicit dev/test bootstrap for registered table/read-model schemas. + pub async fn bootstrap_table_schema_for_dev( + &self, + registry: &TableSchemaRegistry, + ) -> Result { + for statement in table_schema_statements(registry, TableSqlDialect::Postgres)? { + sqlx::query(&statement) + .execute(&self.pool) + .await + .map_err(|err| table_schema_storage_error("bootstrap table schema", err))?; + } + Ok(table_schema_bootstrap_result(registry)) + } + + /// Access an outbox-store handle backed by this repository's pool. + pub fn outbox_store(&self) -> PostgresOutboxStore { + PostgresOutboxStore { + pool: self.pool.clone(), + } + } +} + +impl PostgresOutboxStore { + pub fn new(pool: PgPool) -> Self { + Self { pool } + } + + pub fn pool(&self) -> &PgPool { + &self.pool + } + + /// SQL artifact adapter for registered table/read-model schemas. + pub fn table_schema_adapter(&self) -> TableSqlSchemaAdapter { + TableSqlSchemaAdapter::postgres() + } + + /// Generate SQL statements for registered table/read-model schemas. + pub fn generate_table_migration_artifacts( + &self, + registry: &TableSchemaRegistry, + ) -> Result, TableStoreError> { + generate_table_migration_artifacts(registry, TableSqlDialect::Postgres) + } + + /// Explicit dev/test bootstrap for registered table/read-model schemas. + pub async fn bootstrap_table_schema_for_dev( + &self, + registry: &TableSchemaRegistry, + ) -> Result { + for statement in table_schema_statements(registry, TableSqlDialect::Postgres)? { + sqlx::query(&statement) + .execute(&self.pool) + .await + .map_err(|err| table_schema_storage_error("bootstrap table schema", err))?; + } + Ok(table_schema_bootstrap_result(registry)) + } } impl AsyncGetStream for PostgresRepository { @@ -229,8 +313,8 @@ impl AsyncTransactionalCommit for PostgresRepository { } } -impl AsyncOutboxRepositoryExt for PostgresRepository { - fn outbox_messages_by_status_async( +impl AsyncOutboxStore for PostgresOutboxStore { + fn messages_by_status_async( &self, status: OutboxMessageStatus, ) -> impl Future, RepositoryError>> + Send + '_ { @@ -245,26 +329,24 @@ impl AsyncOutboxRepositoryExt for PostgresRepository { } } - fn claim_outbox_messages_async<'a>( + fn claim_async<'a>( &'a self, - worker_id: &'a str, - max: usize, - lease: Duration, + request: ClaimOutboxMessages, ) -> impl Future, RepositoryError>> + Send + 'a { async move { - if max == 0 { + if request.batch_size == 0 { return Ok(Vec::new()); } let now = SystemTime::now(); let now_epoch = system_time_to_epoch_secs(now)?; - let claimed_until = now.checked_add(lease).ok_or_else(|| { + let claimed_until = now.checked_add(request.lease).ok_or_else(|| { RepositoryError::Model("failed to compute outbox lease deadline".into()) })?; let claimed_until_epoch = system_time_to_epoch_secs(claimed_until)?; let limit = sqlx_repository_i64_from_u64( POSTGRES_BACKEND, - max as u64, + request.batch_size as u64, "outbox claim limit", BIGINT_STORAGE, )?; @@ -279,16 +361,19 @@ impl AsyncOutboxRepositoryExt for PostgresRepository { WITH candidates AS ( SELECT message_id FROM outbox_messages - WHERE (status = $1 AND next_available_at <= to_timestamp($2)) - OR (status = $3 AND (claimed_until IS NULL OR claimed_until <= to_timestamp($2))) + WHERE ( + (status = $1 AND next_available_at <= to_timestamp($2)) + OR (status = $3 AND (claimed_until IS NULL OR claimed_until <= to_timestamp($2))) + ) + AND ($4::text IS NULL OR destination = $4) ORDER BY created_at ASC, message_id ASC - LIMIT $4 + LIMIT $5 FOR UPDATE SKIP LOCKED ) UPDATE outbox_messages AS message - SET status = $5, - claimed_by = $6, - claimed_until = to_timestamp($7), + SET status = $6, + claimed_by = $7, + claimed_until = to_timestamp($8), attempts = attempts + 1, updated_at = now() FROM candidates @@ -316,9 +401,10 @@ impl AsyncOutboxRepositoryExt for PostgresRepository { .bind(OutboxMessageStatus::Pending.as_str()) .bind(now_epoch) .bind(OutboxMessageStatus::InFlight.as_str()) + .bind(request.destination.as_deref()) .bind(limit) .bind(OutboxMessageStatus::InFlight.as_str()) - .bind(worker_id) + .bind(&request.worker_id) .bind(claimed_until_epoch) .fetch_all(&mut *tx) .await @@ -332,10 +418,9 @@ impl AsyncOutboxRepositoryExt for PostgresRepository { } } - fn complete_outbox_message_for_worker_async<'a>( + fn complete_async<'a>( &'a self, - message_id: &'a str, - worker_id: &'a str, + claim: &'a OutboxClaimRef, ) -> impl Future> + Send + 'a { async move { let now = SystemTime::now(); @@ -353,14 +438,21 @@ impl AsyncOutboxRepositoryExt for PostgresRepository { AND claimed_by = $5 AND claimed_until IS NOT NULL AND claimed_until > to_timestamp($6) + AND attempts = $7 "#, ) .bind(OutboxMessageStatus::Published.as_str()) .bind(now_epoch) - .bind(message_id) + .bind(&claim.message_id) .bind(OutboxMessageStatus::InFlight.as_str()) - .bind(worker_id) + .bind(&claim.worker_id) .bind(now_epoch) + .bind(sqlx_repository_i32_from_u64( + POSTGRES_BACKEND, + u64::from(claim.attempt), + "outbox claim attempt", + INTEGER_STORAGE, + )?) .execute(&self.pool) .await .map_err(|err| repository_storage_error("complete outbox message", err))?; @@ -368,17 +460,16 @@ impl AsyncOutboxRepositoryExt for PostgresRepository { ensure_outbox_update_applied( &self.pool, result.rows_affected(), - message_id, - |message| ensure_active_claim(message, Some(worker_id), now), + &claim.message_id, + |message| ensure_active_claim(message, Some(claim), now), ) .await } } - fn release_outbox_message_for_worker_async<'a>( + fn release_async<'a>( &'a self, - message_id: &'a str, - worker_id: &'a str, + claim: &'a OutboxClaimRef, error: &'a str, ) -> impl Future> + Send + 'a { async move { @@ -398,15 +489,22 @@ impl AsyncOutboxRepositoryExt for PostgresRepository { AND claimed_by = $6 AND claimed_until IS NOT NULL AND claimed_until > to_timestamp($7) + AND attempts = $8 "#, ) .bind(OutboxMessageStatus::Pending.as_str()) .bind(now_epoch) .bind(empty_string_as_none(error)) - .bind(message_id) + .bind(&claim.message_id) .bind(OutboxMessageStatus::InFlight.as_str()) - .bind(worker_id) + .bind(&claim.worker_id) .bind(now_epoch) + .bind(sqlx_repository_i32_from_u64( + POSTGRES_BACKEND, + u64::from(claim.attempt), + "outbox claim attempt", + INTEGER_STORAGE, + )?) .execute(&self.pool) .await .map_err(|err| repository_storage_error("release outbox message", err))?; @@ -414,17 +512,16 @@ impl AsyncOutboxRepositoryExt for PostgresRepository { ensure_outbox_update_applied( &self.pool, result.rows_affected(), - message_id, - |message| ensure_active_claim(message, Some(worker_id), now), + &claim.message_id, + |message| ensure_active_claim(message, Some(claim), now), ) .await } } - fn fail_outbox_message_for_worker_async<'a>( + fn fail_async<'a>( &'a self, - message_id: &'a str, - worker_id: &'a str, + claim: &'a OutboxClaimRef, error: &'a str, ) -> impl Future> + Send + 'a { async move { @@ -444,15 +541,22 @@ impl AsyncOutboxRepositoryExt for PostgresRepository { AND claimed_by = $6 AND claimed_until IS NOT NULL AND claimed_until > to_timestamp($7) + AND attempts = $8 "#, ) .bind(OutboxMessageStatus::Failed.as_str()) .bind(empty_string_as_none(error)) .bind(now_epoch) - .bind(message_id) + .bind(&claim.message_id) .bind(OutboxMessageStatus::InFlight.as_str()) - .bind(worker_id) + .bind(&claim.worker_id) .bind(now_epoch) + .bind(sqlx_repository_i32_from_u64( + POSTGRES_BACKEND, + u64::from(claim.attempt), + "outbox claim attempt", + INTEGER_STORAGE, + )?) .execute(&self.pool) .await .map_err(|err| repository_storage_error("fail outbox message", err))?; @@ -460,35 +564,32 @@ impl AsyncOutboxRepositoryExt for PostgresRepository { ensure_outbox_update_applied( &self.pool, result.rows_affected(), - message_id, - |message| ensure_active_claim(message, Some(worker_id), now), + &claim.message_id, + |message| ensure_active_claim(message, Some(claim), now), ) .await } } - fn record_outbox_publish_failure_async<'a>( + fn record_failure_async<'a>( &'a self, - message_id: &'a str, - worker_id: &'a str, + claim: &'a OutboxClaimRef, error: &'a str, max_attempts: u32, ) -> impl Future> + Send + 'a { async move { - let message = outbox_message_by_id_pool(&self.pool, message_id) + let message = outbox_message_by_id_pool(&self.pool, &claim.message_id) .await? .ok_or_else(|| RepositoryError::NotFound { - id: message_id.to_string(), + id: claim.message_id.clone(), })?; - ensure_active_claim(&message, Some(worker_id), SystemTime::now())?; + ensure_active_claim(&message, Some(claim), SystemTime::now())?; if message.attempts >= max_attempts { - self.fail_outbox_message_for_worker_async(message_id, worker_id, error) - .await?; + self.fail_async(claim, error).await?; Ok(OutboxPublishFailureAction::Failed) } else { - self.release_outbox_message_for_worker_async(message_id, worker_id, error) - .await?; + self.release_async(claim, error).await?; Ok(OutboxPublishFailureAction::Released) } } @@ -1058,3 +1159,7 @@ fn system_time_from_epoch_secs(value: f64) -> Result RepositoryError { sqlx_repo::repository_storage_error(POSTGRES_BACKEND, operation, err) } + +fn table_schema_storage_error(operation: &str, err: sqlx::Error) -> TableStoreError { + TableStoreError::Storage(format!("{POSTGRES_BACKEND} {operation} failed: {err}")) +} diff --git a/src/repository/async_repository.rs b/src/repository/async_repository.rs index cd26be9f8..72c93d2ff 100644 --- a/src/repository/async_repository.rs +++ b/src/repository/async_repository.rs @@ -1,9 +1,7 @@ use std::future::Future; -use std::time::Duration; use crate::entity::{Entity, EventRecord}; -use crate::outbox::{OutboxMessage, OutboxMessageStatus}; -use crate::outbox_worker::OutboxPublishFailureAction; +use crate::outbox::OutboxMessage; use crate::read_model::{ ReadModel, ReadModelAdapterCapabilities, ReadModelCommitOutcome, ReadModelError, ReadModelLoadGraph, ReadModelLoadRequest, ReadModelQueryCapabilities, ReadModelWritePlan, @@ -191,55 +189,3 @@ pub trait AsyncSnapshotStore: Send + Sync { identity: &'a StreamIdentity, ) -> impl Future> + Send + 'a; } - -/// Async worker-facing outbox repository operations. -pub trait AsyncOutboxRepositoryExt: Send + Sync { - fn outbox_messages_by_status_async( - &self, - status: OutboxMessageStatus, - ) -> impl Future, RepositoryError>> + Send + '_; - - fn outbox_messages_pending_async( - &self, - ) -> impl Future, RepositoryError>> + Send + '_ { - async move { - self.outbox_messages_by_status_async(OutboxMessageStatus::Pending) - .await - } - } - - fn claim_outbox_messages_async<'a>( - &'a self, - worker_id: &'a str, - max: usize, - lease: Duration, - ) -> impl Future, RepositoryError>> + Send + 'a; - - fn complete_outbox_message_for_worker_async<'a>( - &'a self, - message_id: &'a str, - worker_id: &'a str, - ) -> impl Future> + Send + 'a; - - fn release_outbox_message_for_worker_async<'a>( - &'a self, - message_id: &'a str, - worker_id: &'a str, - error: &'a str, - ) -> impl Future> + Send + 'a; - - fn fail_outbox_message_for_worker_async<'a>( - &'a self, - message_id: &'a str, - worker_id: &'a str, - error: &'a str, - ) -> impl Future> + Send + 'a; - - fn record_outbox_publish_failure_async<'a>( - &'a self, - message_id: &'a str, - worker_id: &'a str, - error: &'a str, - max_attempts: u32, - ) -> impl Future> + Send + 'a; -} diff --git a/src/repository/mod.rs b/src/repository/mod.rs index a565a1800..d4f29e8a8 100644 --- a/src/repository/mod.rs +++ b/src/repository/mod.rs @@ -6,9 +6,9 @@ mod identity; mod repository; pub use async_repository::{ - AsyncCommitBatch, AsyncGetStream, AsyncOutboxRepositoryExt, AsyncReadModelSessionStore, - AsyncReadModelStore, AsyncRelationalReadModelQueryStore, AsyncRepository, AsyncSnapshotStore, - AsyncSnapshotWrite, AsyncStreamWrite, AsyncTransactionalCommit, PreparedEventAppend, + AsyncCommitBatch, AsyncGetStream, AsyncReadModelSessionStore, AsyncReadModelStore, + AsyncRelationalReadModelQueryStore, AsyncRepository, AsyncSnapshotStore, AsyncSnapshotWrite, + AsyncStreamWrite, AsyncTransactionalCommit, PreparedEventAppend, }; pub use batch::{CommitBatch, SnapshotWrite, TransactionalCommit}; pub use error::RepositoryError; diff --git a/src/sqlite_repo/mod.rs b/src/sqlite_repo/mod.rs index b34a644c5..29d3fd3f2 100644 --- a/src/sqlite_repo/mod.rs +++ b/src/sqlite_repo/mod.rs @@ -17,15 +17,18 @@ use sqlx::{Row, Sqlite, SqlitePool, Transaction}; use crate::entity::{Entity, EventRecord}; use crate::outbox::{OutboxMessage, OutboxMessageStatus}; -use crate::outbox_worker::{ensure_active_claim, OutboxPublishFailureAction}; +use crate::outbox_worker::{ + ensure_active_claim, AsyncOutboxStore, ClaimOutboxMessages, OutboxClaimRef, + OutboxPublishFailureAction, +}; use crate::read_model::{ ProcessedMessageMark, ReadModel, ReadModelAdapterCapabilities, ReadModelCommitOutcome, ReadModelError, ReadModelMutation, ReadModelWritePlan, Versioned, }; use crate::repository::{ - AsyncCommitBatch, AsyncGetStream, AsyncOutboxRepositoryExt, AsyncReadModelSessionStore, - AsyncReadModelStore, AsyncSnapshotStore, AsyncSnapshotWrite, AsyncTransactionalCommit, - PreparedEventAppend, RepositoryError, StreamIdentity, + AsyncCommitBatch, AsyncGetStream, AsyncReadModelSessionStore, AsyncReadModelStore, + AsyncSnapshotStore, AsyncSnapshotWrite, AsyncTransactionalCommit, PreparedEventAppend, + RepositoryError, StreamIdentity, }; use crate::snapshot::SnapshotRecord; use crate::sqlx_repo::{ @@ -38,6 +41,11 @@ use crate::sqlx_repo::{ validate_entity_id_matches_identity, validate_prepared_appends, validate_snapshot_identity, validate_supported_event_codec, }; +use crate::table::{ + generate_table_migration_artifacts, table_schema_bootstrap_result, table_schema_statements, + TableMigrationArtifact, TableSchemaBootstrap, TableSchemaRegistry, TableSqlDialect, + TableSqlSchemaAdapter, TableStoreError, +}; const SQLITE_SCHEMA: &str = include_str!("../../migrations/sqlite/0001_initial.sql"); const SQLITE_BACKEND: &str = "sqlite"; @@ -49,6 +57,12 @@ pub struct SqliteRepository { pool: SqlitePool, } +/// SQLite-backed outbox table store. +#[derive(Clone)] +pub struct SqliteOutboxStore { + pool: SqlitePool, +} + impl SqliteRepository { /// Create a repository from an existing migrated pool. pub fn new(pool: SqlitePool) -> Self { @@ -96,6 +110,77 @@ impl SqliteRepository { pub fn pool(&self) -> &SqlitePool { &self.pool } + + /// SQL artifact adapter for registered table/read-model schemas. + pub fn table_schema_adapter(&self) -> TableSqlSchemaAdapter { + TableSqlSchemaAdapter::sqlite() + } + + /// Generate SQL statements for registered table/read-model schemas. + pub fn generate_table_migration_artifacts( + &self, + registry: &TableSchemaRegistry, + ) -> Result, TableStoreError> { + generate_table_migration_artifacts(registry, TableSqlDialect::Sqlite) + } + + /// Explicit dev/test bootstrap for registered table/read-model schemas. + pub async fn bootstrap_table_schema_for_dev( + &self, + registry: &TableSchemaRegistry, + ) -> Result { + for statement in table_schema_statements(registry, TableSqlDialect::Sqlite)? { + sqlx::query(&statement) + .execute(&self.pool) + .await + .map_err(|err| read_model_storage_error("bootstrap table schema", err))?; + } + Ok(table_schema_bootstrap_result(registry)) + } + + /// Access an outbox-store handle backed by this repository's pool. + pub fn outbox_store(&self) -> SqliteOutboxStore { + SqliteOutboxStore { + pool: self.pool.clone(), + } + } +} + +impl SqliteOutboxStore { + pub fn new(pool: SqlitePool) -> Self { + Self { pool } + } + + pub fn pool(&self) -> &SqlitePool { + &self.pool + } + + /// SQL artifact adapter for registered table/read-model schemas. + pub fn table_schema_adapter(&self) -> TableSqlSchemaAdapter { + TableSqlSchemaAdapter::sqlite() + } + + /// Generate SQL statements for registered table/read-model schemas. + pub fn generate_table_migration_artifacts( + &self, + registry: &TableSchemaRegistry, + ) -> Result, TableStoreError> { + generate_table_migration_artifacts(registry, TableSqlDialect::Sqlite) + } + + /// Explicit dev/test bootstrap for registered table/read-model schemas. + pub async fn bootstrap_table_schema_for_dev( + &self, + registry: &TableSchemaRegistry, + ) -> Result { + for statement in table_schema_statements(registry, TableSqlDialect::Sqlite)? { + sqlx::query(&statement) + .execute(&self.pool) + .await + .map_err(|err| read_model_storage_error("bootstrap table schema", err))?; + } + Ok(table_schema_bootstrap_result(registry)) + } } impl AsyncGetStream for SqliteRepository { @@ -429,8 +514,8 @@ impl AsyncReadModelSessionStore for SqliteRepository { } } -impl AsyncOutboxRepositoryExt for SqliteRepository { - fn outbox_messages_by_status_async( +impl AsyncOutboxStore for SqliteOutboxStore { + fn messages_by_status_async( &self, status: OutboxMessageStatus, ) -> impl Future, RepositoryError>> + Send + '_ { @@ -456,20 +541,18 @@ impl AsyncOutboxRepositoryExt for SqliteRepository { } } - fn claim_outbox_messages_async<'a>( + fn claim_async<'a>( &'a self, - worker_id: &'a str, - max: usize, - lease: Duration, + request: ClaimOutboxMessages, ) -> impl Future, RepositoryError>> + Send + 'a { async move { - if max == 0 { + if request.batch_size == 0 { return Ok(Vec::new()); } let now = SystemTime::now(); let now_epoch = system_time_to_epoch_secs(now)?; - let claimed_until = now.checked_add(lease).ok_or_else(|| { + let claimed_until = now.checked_add(request.lease).ok_or_else(|| { RepositoryError::Model("failed to compute outbox lease deadline".into()) })?; let claimed_until_storage = system_time_to_storage(claimed_until)?; @@ -481,7 +564,7 @@ impl AsyncOutboxRepositoryExt for SqliteRepository { let limit = sqlx_repository_i64_from_u64( SQLITE_BACKEND, - max as u64, + request.batch_size as u64, "outbox claim limit", SIGNED_INTEGER_STORAGE, )?; @@ -489,8 +572,11 @@ impl AsyncOutboxRepositoryExt for SqliteRepository { r#" SELECT message_id FROM outbox_messages - WHERE (status = ? AND CAST(next_available_at AS REAL) <= ?) - OR (status = ? AND (claimed_until IS NULL OR CAST(claimed_until AS REAL) <= ?)) + WHERE ( + (status = ? AND CAST(next_available_at AS REAL) <= ?) + OR (status = ? AND (claimed_until IS NULL OR CAST(claimed_until AS REAL) <= ?)) + ) + AND (? IS NULL OR destination = ?) ORDER BY CAST(created_at AS REAL) ASC, message_id ASC LIMIT ? "#, @@ -499,6 +585,8 @@ impl AsyncOutboxRepositoryExt for SqliteRepository { .bind(now_epoch) .bind(OutboxMessageStatus::InFlight.as_str()) .bind(now_epoch) + .bind(request.destination.as_deref()) + .bind(request.destination.as_deref()) .bind(limit) .fetch_all(&mut *tx) .await @@ -525,16 +613,19 @@ impl AsyncOutboxRepositoryExt for SqliteRepository { AND (claimed_until IS NULL OR CAST(claimed_until AS REAL) <= ?) ) ) + AND (? IS NULL OR destination = ?) "#, ) .bind(OutboxMessageStatus::InFlight.as_str()) - .bind(worker_id) + .bind(&request.worker_id) .bind(&claimed_until_storage) .bind(&message_id) .bind(OutboxMessageStatus::Pending.as_str()) .bind(now_epoch) .bind(OutboxMessageStatus::InFlight.as_str()) .bind(now_epoch) + .bind(request.destination.as_deref()) + .bind(request.destination.as_deref()) .execute(&mut *tx) .await .map_err(|err| repository_storage_error("claim outbox message", err))?; @@ -555,10 +646,9 @@ impl AsyncOutboxRepositoryExt for SqliteRepository { } } - fn complete_outbox_message_for_worker_async<'a>( + fn complete_async<'a>( &'a self, - message_id: &'a str, - worker_id: &'a str, + claim: &'a OutboxClaimRef, ) -> impl Future> + Send + 'a { async move { let now = SystemTime::now(); @@ -576,14 +666,21 @@ impl AsyncOutboxRepositoryExt for SqliteRepository { AND claimed_by = ? AND claimed_until IS NOT NULL AND CAST(claimed_until AS REAL) > ? + AND attempts = ? "#, ) .bind(OutboxMessageStatus::Published.as_str()) .bind(system_time_to_storage(now)?) - .bind(message_id) + .bind(&claim.message_id) .bind(OutboxMessageStatus::InFlight.as_str()) - .bind(worker_id) + .bind(&claim.worker_id) .bind(now_epoch) + .bind(sqlx_repository_i64_from_u64( + SQLITE_BACKEND, + u64::from(claim.attempt), + "outbox claim attempt", + SIGNED_INTEGER_STORAGE, + )?) .execute(&self.pool) .await .map_err(|err| repository_storage_error("complete outbox message", err))?; @@ -591,17 +688,16 @@ impl AsyncOutboxRepositoryExt for SqliteRepository { ensure_outbox_update_applied( &self.pool, result.rows_affected(), - message_id, - |message| ensure_active_claim(message, Some(worker_id), now), + &claim.message_id, + |message| ensure_active_claim(message, Some(claim), now), ) .await } } - fn release_outbox_message_for_worker_async<'a>( + fn release_async<'a>( &'a self, - message_id: &'a str, - worker_id: &'a str, + claim: &'a OutboxClaimRef, error: &'a str, ) -> impl Future> + Send + 'a { async move { @@ -622,15 +718,22 @@ impl AsyncOutboxRepositoryExt for SqliteRepository { AND claimed_by = ? AND claimed_until IS NOT NULL AND CAST(claimed_until AS REAL) > ? + AND attempts = ? "#, ) .bind(OutboxMessageStatus::Pending.as_str()) .bind(now_storage) .bind(empty_string_as_none(error)) - .bind(message_id) + .bind(&claim.message_id) .bind(OutboxMessageStatus::InFlight.as_str()) - .bind(worker_id) + .bind(&claim.worker_id) .bind(now_epoch) + .bind(sqlx_repository_i64_from_u64( + SQLITE_BACKEND, + u64::from(claim.attempt), + "outbox claim attempt", + SIGNED_INTEGER_STORAGE, + )?) .execute(&self.pool) .await .map_err(|err| repository_storage_error("release outbox message", err))?; @@ -638,17 +741,16 @@ impl AsyncOutboxRepositoryExt for SqliteRepository { ensure_outbox_update_applied( &self.pool, result.rows_affected(), - message_id, - |message| ensure_active_claim(message, Some(worker_id), now), + &claim.message_id, + |message| ensure_active_claim(message, Some(claim), now), ) .await } } - fn fail_outbox_message_for_worker_async<'a>( + fn fail_async<'a>( &'a self, - message_id: &'a str, - worker_id: &'a str, + claim: &'a OutboxClaimRef, error: &'a str, ) -> impl Future> + Send + 'a { async move { @@ -668,15 +770,22 @@ impl AsyncOutboxRepositoryExt for SqliteRepository { AND claimed_by = ? AND claimed_until IS NOT NULL AND CAST(claimed_until AS REAL) > ? + AND attempts = ? "#, ) .bind(OutboxMessageStatus::Failed.as_str()) .bind(empty_string_as_none(error)) .bind(system_time_to_storage(now)?) - .bind(message_id) + .bind(&claim.message_id) .bind(OutboxMessageStatus::InFlight.as_str()) - .bind(worker_id) + .bind(&claim.worker_id) .bind(now_epoch) + .bind(sqlx_repository_i64_from_u64( + SQLITE_BACKEND, + u64::from(claim.attempt), + "outbox claim attempt", + SIGNED_INTEGER_STORAGE, + )?) .execute(&self.pool) .await .map_err(|err| repository_storage_error("fail outbox message", err))?; @@ -684,35 +793,32 @@ impl AsyncOutboxRepositoryExt for SqliteRepository { ensure_outbox_update_applied( &self.pool, result.rows_affected(), - message_id, - |message| ensure_active_claim(message, Some(worker_id), now), + &claim.message_id, + |message| ensure_active_claim(message, Some(claim), now), ) .await } } - fn record_outbox_publish_failure_async<'a>( + fn record_failure_async<'a>( &'a self, - message_id: &'a str, - worker_id: &'a str, + claim: &'a OutboxClaimRef, error: &'a str, max_attempts: u32, ) -> impl Future> + Send + 'a { async move { - let message = outbox_message_by_id_pool(&self.pool, message_id) + let message = outbox_message_by_id_pool(&self.pool, &claim.message_id) .await? .ok_or_else(|| RepositoryError::NotFound { - id: message_id.to_string(), + id: claim.message_id.clone(), })?; - ensure_active_claim(&message, Some(worker_id), SystemTime::now())?; + ensure_active_claim(&message, Some(claim), SystemTime::now())?; if message.attempts >= max_attempts { - self.fail_outbox_message_for_worker_async(message_id, worker_id, error) - .await?; + self.fail_async(claim, error).await?; Ok(OutboxPublishFailureAction::Failed) } else { - self.release_outbox_message_for_worker_async(message_id, worker_id, error) - .await?; + self.release_async(claim, error).await?; Ok(OutboxPublishFailureAction::Released) } } diff --git a/src/sqlx_repo/mod.rs b/src/sqlx_repo/mod.rs index aed290d8e..fb7a90a53 100644 --- a/src/sqlx_repo/mod.rs +++ b/src/sqlx_repo/mod.rs @@ -29,6 +29,7 @@ pub(crate) fn reject_duplicate_outbox_messages( ) -> Result<(), RepositoryError> { let mut seen = HashSet::with_capacity(messages.len()); for message in messages { + validate_outbox_table_write(message)?; let id = message.id(); if id.trim().is_empty() { return Err(RepositoryError::Model( @@ -47,6 +48,13 @@ pub(crate) fn reject_duplicate_outbox_messages( Ok(()) } +fn validate_outbox_table_write(message: &OutboxMessage) -> Result<(), RepositoryError> { + crate::outbox::outbox_message_insert_plan(message) + .and_then(|plan| plan.validate().map(|()| plan)) + .map(|_| ()) + .map_err(|err| RepositoryError::Model(err.to_string())) +} + pub(crate) fn validate_entity_id_matches_identity( streams: &[AsyncStreamWrite<'_>], ) -> Result<(), RepositoryError> { diff --git a/src/table/mod.rs b/src/table/mod.rs new file mode 100644 index 000000000..c47c0d354 --- /dev/null +++ b/src/table/mod.rs @@ -0,0 +1,51 @@ +//! Neutral table/row primitives shared by read models and operational tables. +//! +//! The read-model ORM introduced these structures first, but they are not +//! inherently read-model concepts. Outbox storage, inbox/checkpoint tables, and +//! future operational tables can use the same schema and row-write vocabulary. + +mod sql; + +pub use crate::read_model::{ + ColumnDef as TableColumn, ColumnType, DeleteRowMutation as DeleteTableRowMutation, + DocumentMutation as TableDocumentMutation, ExpectedVersion, ForeignKey, IndexDef as TableIndex, + PatchMode, PatchRowMutation as PatchTableRowMutation, PrimaryKey, + ReadModelAdapterCapabilities as TableAdapterCapabilities, + ReadModelCommitOutcome as TableCommitOutcome, ReadModelError as TableStoreError, + ReadModelMigrationArtifact as TableMigrationArtifact, ReadModelMutation as TableMutation, + ReadModelSchema as TableSchema, ReadModelSchemaAdapter as TableSchemaAdapter, + ReadModelSchemaAdapterCapabilities as TableSchemaAdapterCapabilities, + ReadModelSchemaBootstrap as TableSchemaBootstrap, ReadModelSchemaIssue as TableSchemaIssue, + ReadModelSchemaIssueKind as TableSchemaIssueKind, + ReadModelSchemaRegistry as TableSchemaRegistry, + ReadModelSchemaVerification as TableSchemaVerification, ReadModelWritePlan as TableWritePlan, + RelationshipDef, RelationshipKind, RowKey, RowMutation as TableRowMutation, RowPatch, RowValue, + RowValues, RowWriteMode, DEFAULT_READ_MODEL_VERSION_COLUMN as DEFAULT_TABLE_VERSION_COLUMN, +}; +pub use sql::{ + bootstrap_result as table_schema_bootstrap_result, generate_table_migration_artifacts, + table_schema_statements, TableSqlDialect, TableSqlSchemaAdapter, +}; + +/// Opt-in trait for non-read-model types that map to a relational table row. +pub trait TableModel: Clone + Send + Sync + Sized { + fn table_schema() -> TableSchema; + fn table_key(&self) -> Result; + fn to_table_row(&self) -> Result; +} + +/// Extension methods for registering neutral table models. +pub trait TableSchemaRegistryExt { + fn register_table(&mut self) -> Result<&mut Self, TableStoreError> + where + M: TableModel; +} + +impl TableSchemaRegistryExt for TableSchemaRegistry { + fn register_table(&mut self) -> Result<&mut Self, TableStoreError> + where + M: TableModel, + { + self.register_schema(M::table_schema()) + } +} diff --git a/src/table/sql.rs b/src/table/sql.rs new file mode 100644 index 000000000..7bd3725f5 --- /dev/null +++ b/src/table/sql.rs @@ -0,0 +1,263 @@ +use crate::table::{ + ColumnType, TableMigrationArtifact, TableSchema, TableSchemaAdapter, + TableSchemaAdapterCapabilities, TableSchemaBootstrap, TableSchemaRegistry, TableStoreError, +}; + +/// SQL dialect used to render table-schema migration artifacts. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum TableSqlDialect { + Sqlite, + Postgres, +} + +/// Stateless adapter for generating SQL artifacts from table schemas. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct TableSqlSchemaAdapter { + dialect: TableSqlDialect, +} + +impl TableSqlSchemaAdapter { + pub fn sqlite() -> Self { + Self { + dialect: TableSqlDialect::Sqlite, + } + } + + pub fn postgres() -> Self { + Self { + dialect: TableSqlDialect::Postgres, + } + } + + pub fn dialect(&self) -> TableSqlDialect { + self.dialect + } +} + +impl TableSchemaAdapter for TableSqlSchemaAdapter { + fn schema_capabilities(&self) -> TableSchemaAdapterCapabilities { + TableSchemaAdapterCapabilities { + migration_artifacts: true, + schema_verification: false, + dev_bootstrap: false, + } + } + + fn generate_migration_artifacts( + &self, + registry: &TableSchemaRegistry, + ) -> Result, TableStoreError> { + generate_table_migration_artifacts(registry, self.dialect) + } +} + +pub fn generate_table_migration_artifacts( + registry: &TableSchemaRegistry, + dialect: TableSqlDialect, +) -> Result, TableStoreError> { + Ok(vec![TableMigrationArtifact::new( + artifact_name(dialect), + table_schema_statements(registry, dialect)?, + )]) +} + +pub fn table_schema_statements( + registry: &TableSchemaRegistry, + dialect: TableSqlDialect, +) -> Result, TableStoreError> { + registry.validate()?; + + let mut statements = Vec::new(); + for schema in registry.schemas() { + statements.push(create_table_statement(schema, dialect)?); + statements.extend(index_statements(schema)); + } + Ok(statements) +} + +fn create_table_statement( + schema: &TableSchema, + dialect: TableSqlDialect, +) -> Result { + let mut definitions = schema + .columns + .iter() + .map(|column| { + let mut definition = format!( + "{} {}", + quote_identifier(&column.column_name), + sql_type(&column.column_type, column.jsonb, dialect)? + ); + if !column.nullable || column.primary_key { + definition.push_str(" NOT NULL"); + } + if column.has_default { + if let Some(default) = column.default.as_deref() { + definition.push_str(" DEFAULT "); + definition.push_str(default); + } + } + Ok(definition) + }) + .collect::, TableStoreError>>()?; + + if let Some(version_column) = schema.version_column.as_deref() { + definitions.push(format!( + "{} {} NOT NULL DEFAULT 1", + quote_identifier(version_column), + sql_type(&ColumnType::UnsignedInteger, false, dialect)? + )); + } + + definitions.push(format!( + "PRIMARY KEY ({})", + schema + .primary_key + .columns + .iter() + .map(|column| quote_identifier(column)) + .collect::>() + .join(", ") + )); + + for column in &schema.columns { + if let Some(foreign_key) = &column.foreign_key { + definitions.push(format!( + "FOREIGN KEY ({}) REFERENCES {} ({})", + quote_identifier(&column.column_name), + quote_identifier(&foreign_key.table), + quote_identifier(&foreign_key.column) + )); + } + } + + for foreign_key in &schema.foreign_keys { + let already_declared_on_column = schema.columns.iter().any(|column| { + column.column_name == foreign_key.column + && column.foreign_key.as_ref() == Some(foreign_key) + }); + if already_declared_on_column { + continue; + } + definitions.push(format!( + "FOREIGN KEY ({}) REFERENCES {} ({})", + quote_identifier(&foreign_key.column), + quote_identifier(&foreign_key.table), + quote_identifier(&foreign_key.column) + )); + } + + Ok(format!( + "CREATE TABLE IF NOT EXISTS {} (\n {}\n);", + quote_identifier(&schema.table_name), + definitions.join(",\n ") + )) +} + +fn index_statements(schema: &TableSchema) -> impl Iterator + '_ { + schema.indexes.iter().map(|index| { + let name = index + .name + .clone() + .unwrap_or_else(|| format!("{}_{}_idx", schema.table_name, index.columns.join("_"))); + let unique = if index.unique { "UNIQUE " } else { "" }; + format!( + "CREATE {unique}INDEX IF NOT EXISTS {} ON {} ({});", + quote_identifier(&name), + quote_identifier(&schema.table_name), + index + .columns + .iter() + .map(|column| quote_identifier(column)) + .collect::>() + .join(", ") + ) + }) +} + +fn sql_type( + column_type: &ColumnType, + jsonb: bool, + dialect: TableSqlDialect, +) -> Result<&'static str, TableStoreError> { + let type_name = match (dialect, column_type) { + (TableSqlDialect::Sqlite, ColumnType::Text) => "TEXT", + (TableSqlDialect::Sqlite, ColumnType::Boolean) => "INTEGER", + (TableSqlDialect::Sqlite, ColumnType::Integer | ColumnType::UnsignedInteger) => "INTEGER", + (TableSqlDialect::Sqlite, ColumnType::Float) => "REAL", + (TableSqlDialect::Sqlite, ColumnType::Bytes) => "BLOB", + (TableSqlDialect::Sqlite, ColumnType::Json) => "TEXT", + (TableSqlDialect::Postgres, ColumnType::Text) => "text", + (TableSqlDialect::Postgres, ColumnType::Boolean) => "boolean", + (TableSqlDialect::Postgres, ColumnType::Integer | ColumnType::UnsignedInteger) => "bigint", + (TableSqlDialect::Postgres, ColumnType::Float) => "double precision", + (TableSqlDialect::Postgres, ColumnType::Bytes) => "bytea", + (TableSqlDialect::Postgres, ColumnType::Json) if jsonb => "jsonb", + (TableSqlDialect::Postgres, ColumnType::Json) => "jsonb", + (_, ColumnType::Unsupported(type_name)) => { + return Err(TableStoreError::Metadata(format!( + "unsupported table column type `{type_name}`" + ))); + } + }; + Ok(type_name) +} + +fn quote_identifier(value: &str) -> String { + format!("\"{}\"", value.replace('"', "\"\"")) +} + +fn artifact_name(dialect: TableSqlDialect) -> &'static str { + match dialect { + TableSqlDialect::Sqlite => "sqlite-tables", + TableSqlDialect::Postgres => "postgres-tables", + } +} + +pub fn bootstrap_result(registry: &TableSchemaRegistry) -> TableSchemaBootstrap { + TableSchemaBootstrap::new(registry.table_names().map(str::to_string)) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::outbox::{outbox_message_schema, OUTBOX_MESSAGES_TABLE}; + + #[test] + fn renders_outbox_table_schema_for_sqlite() { + let mut registry = TableSchemaRegistry::new(); + registry + .register_schema(outbox_message_schema()) + .expect("schema should register"); + + let artifact = generate_table_migration_artifacts(®istry, TableSqlDialect::Sqlite) + .expect("artifact should render") + .pop() + .expect("artifact should exist"); + + assert_eq!(artifact.name, "sqlite-tables"); + assert!(artifact + .statements + .iter() + .any(|statement| statement.contains("CREATE TABLE IF NOT EXISTS \"outbox_messages\""))); + assert!(artifact + .statements + .iter() + .any(|statement| statement.contains("\"message_id\" TEXT NOT NULL"))); + } + + #[test] + fn bootstrap_result_lists_registered_tables() { + let mut registry = TableSchemaRegistry::new(); + registry + .register_schema(outbox_message_schema()) + .expect("schema should register"); + + let result = bootstrap_result(®istry); + + assert_eq!( + result.bootstrapped_tables, + vec![OUTBOX_MESSAGES_TABLE.to_string()] + ); + } +} diff --git a/tests/async_repository/main.rs b/tests/async_repository/main.rs index f84f17380..83140ba7a 100644 --- a/tests/async_repository/main.rs +++ b/tests/async_repository/main.rs @@ -3,10 +3,10 @@ use std::time::Duration; use serde::{Deserialize, Serialize}; use sourced_rust::{ impl_aggregate, Aggregate, AsyncAggregateBuilder, AsyncCommitBatch, AsyncGetStream, - AsyncOutboxRepositoryExt, AsyncReadModelSessionStore, AsyncReadModelStore, AsyncSnapshotStore, - AsyncStreamWrite, AsyncTransactionalCommit, Entity, EventRecord, HashMapRepository, - InMemorySnapshotStore, OutboxMessage, ProcessedMessageMark, ReadModel, ReadModelSession, - ReadModelWritePlan, RepositoryError, SnapshotRecord, StreamIdentity, + AsyncOutboxStore, AsyncReadModelSessionStore, AsyncReadModelStore, AsyncSnapshotStore, + AsyncStreamWrite, AsyncTransactionalCommit, ClaimOutboxMessages, Entity, EventRecord, + HashMapRepository, InMemorySnapshotStore, OutboxMessage, ProcessedMessageMark, ReadModel, + ReadModelSession, ReadModelWritePlan, RepositoryError, SnapshotRecord, StreamIdentity, }; #[derive(Default)] @@ -209,6 +209,7 @@ async fn async_snapshot_store_uses_full_stream_identity() { #[tokio::test] async fn async_outbox_repository_delegates_worker_operations() { let repo = HashMapRepository::new(); + let outbox = repo.outbox_store(); let message = OutboxMessage::create("msg-1", "Event", b"{}".to_vec()).unwrap(); let mut aggregate = AlphaAggregate::default(); aggregate.touch("outbox-aggregate-1"); @@ -219,8 +220,12 @@ async fn async_outbox_repository_delegates_worker_operations() { .await .unwrap(); - let claimed = repo - .claim_outbox_messages_async("worker-1", 1, Duration::from_secs(60)) + let claimed = outbox + .claim_async(ClaimOutboxMessages::new( + "worker-1", + 1, + Duration::from_secs(60), + )) .await .unwrap(); diff --git a/tests/bomberman/main.rs b/tests/bomberman/main.rs index c159a7b9e..23cd87761 100644 --- a/tests/bomberman/main.rs +++ b/tests/bomberman/main.rs @@ -193,9 +193,10 @@ fn player_killed_by_bomb() { .contains(&"player:p2".to_string())); // Verify outbox message was created (PlayerKilled) - use sourced_rust::{OutboxMessageStatus, OutboxRepositoryExt}; + use sourced_rust::{OutboxMessageStatus, OutboxStore}; let pending = repo2 - .outbox_messages_by_status(OutboxMessageStatus::Pending) + .outbox_store() + .messages_by_status(OutboxMessageStatus::Pending) .unwrap(); assert!(!pending.is_empty()); assert_eq!(pending[0].event_type, "PlayerKilled"); diff --git a/tests/distributed_read_model/main.rs b/tests/distributed_read_model/main.rs index 4e6d2c800..a6cdd7615 100644 --- a/tests/distributed_read_model/main.rs +++ b/tests/distributed_read_model/main.rs @@ -89,12 +89,13 @@ fn seat_checkout_saga_reserves_seat_and_projects_user_screen() { let checkout_store = HashMapRepository::new(); let checkout_service = checkout_saga_service::service(checkout_store.clone().queued().aggregate()); - let checkout_worker = OutboxWorkerThread::spawn(checkout_store.clone(), queue.clone(), poll); + let checkout_worker = + OutboxWorkerThread::spawn(checkout_store.outbox_store(), queue.clone(), poll); let checkout_sub = microsvc::subscribe(checkout_service.clone(), queue.new_subscriber(), poll); let seat_store = HashMapRepository::new(); let seat_service = seat_inventory_service::service(seat_store.clone().queued().aggregate()); - let seat_worker = OutboxWorkerThread::spawn(seat_store.clone(), queue.clone(), poll); + let seat_worker = OutboxWorkerThread::spawn(seat_store.outbox_store(), queue.clone(), poll); let seat_sub = microsvc::subscribe(seat_service.clone(), queue.new_subscriber(), poll); let read_store = InMemoryReadModelStore::new(); diff --git a/tests/distributed_read_model_board/main.rs b/tests/distributed_read_model_board/main.rs index 3be6ab7e1..c5d4c5b4b 100644 --- a/tests/distributed_read_model_board/main.rs +++ b/tests/distributed_read_model_board/main.rs @@ -64,8 +64,11 @@ fn board_service_feeds_a_normalized_card_read_model() { let board_store = HashMapRepository::new(); let board_service = board_service::model_service(board_store.clone().queued().aggregate()); - let worker = - OutboxWorkerThread::spawn(board_store.clone(), queue.clone(), Duration::from_millis(5)); + let worker = OutboxWorkerThread::spawn( + board_store.outbox_store(), + queue.clone(), + Duration::from_millis(5), + ); let read_store = InMemoryReadModelStore::new(); register_schemas(&read_store).expect("relational schemas should register"); diff --git a/tests/hashmap_repository_conformance/main.rs b/tests/hashmap_repository_conformance/main.rs index 562955944..cfc5706a2 100644 --- a/tests/hashmap_repository_conformance/main.rs +++ b/tests/hashmap_repository_conformance/main.rs @@ -45,7 +45,12 @@ async fn snapshots_use_full_stream_identity() { #[tokio::test] async fn high_level_outbox_commit_persists_row_without_stream() { - conformance::outbox::high_level_outbox_commit_persists_row_without_stream(repository()).await; + let repo = repository(); + conformance::outbox::high_level_outbox_commit_persists_row_without_stream( + repo.clone(), + repo.outbox_store(), + ) + .await; } #[tokio::test] @@ -55,10 +60,17 @@ async fn duplicate_outbox_insert_rolls_back_aggregate() { #[tokio::test] async fn aggregate_conflict_rolls_back_outbox() { - conformance::outbox::aggregate_conflict_rolls_back_outbox(repository()).await; + let repo = repository(); + conformance::outbox::aggregate_conflict_rolls_back_outbox(repo.clone(), repo.outbox_store()) + .await; } #[tokio::test] async fn worker_claim_complete_and_retry_lifecycle() { - conformance::outbox::worker_claim_complete_and_retry_lifecycle(repository()).await; + let repo = repository(); + conformance::outbox::worker_claim_complete_and_retry_lifecycle( + repo.clone(), + repo.outbox_store(), + ) + .await; } diff --git a/tests/microsvc/convention.rs b/tests/microsvc/convention.rs index 8cc137076..971fa4fc6 100644 --- a/tests/microsvc/convention.rs +++ b/tests/microsvc/convention.rs @@ -9,7 +9,7 @@ use serde_json::json; use sourced_rust::microsvc::{Service, Session}; -use sourced_rust::{AggregateBuilder, HashMapRepository, OutboxRepositoryExt, Queueable}; +use sourced_rust::{AggregateBuilder, HashMapRepository, OutboxStore, Queueable}; use crate::handlers; use crate::models::counter::Counter; @@ -100,7 +100,7 @@ fn create_persists_outbox_message() { assert_eq!(counter.value, 0); // Outbox message was persisted - let pending = inner.outbox_messages_pending().unwrap(); + let pending = inner.outbox_store().pending().unwrap(); assert_eq!(pending.len(), 1); assert_eq!(pending[0].event_type, "CounterCreated"); } @@ -124,7 +124,8 @@ fn duplicate_create_leaves_single_outbox_message() { .repo() .repo() .inner() - .outbox_messages_pending() + .outbox_store() + .pending() .unwrap(); assert_eq!(pending.len(), 1); } @@ -155,7 +156,7 @@ fn increment_persists_outbox_message() { // Both outbox messages were persisted let inner = service.repo().repo().inner(); - let pending = inner.outbox_messages_pending().unwrap(); + let pending = inner.outbox_store().pending().unwrap(); assert_eq!(pending.len(), 2); let mut event_types: Vec<&str> = pending.iter().map(|m| m.event_type.as_str()).collect(); event_types.sort(); diff --git a/tests/persistent_repository_conformance/outbox.rs b/tests/persistent_repository_conformance/outbox.rs index 95134fccd..91b2887d7 100644 --- a/tests/persistent_repository_conformance/outbox.rs +++ b/tests/persistent_repository_conformance/outbox.rs @@ -1,23 +1,18 @@ use std::time::Duration; use sourced_rust::{ - Aggregate, AsyncAggregateBuilder, AsyncGetStream, AsyncOutboxRepositoryExt, - AsyncTransactionalCommit, OutboxMessage, OutboxMessageStatus, OutboxPublishFailureAction, - RepositoryError, StreamIdentity, + Aggregate, AsyncAggregateBuilder, AsyncGetStream, AsyncOutboxStore, AsyncTransactionalCommit, + ClaimOutboxMessages, OutboxClaimRef, OutboxMessage, OutboxMessageStatus, + OutboxPublishFailureAction, RepositoryError, StreamIdentity, }; use super::scenario::unique_id; use super::seat::Seat; -pub async fn high_level_outbox_commit_persists_row_without_stream(repo: R) +pub async fn high_level_outbox_commit_persists_row_without_stream(repo: R, outbox: S) where - R: AsyncGetStream - + AsyncTransactionalCommit - + AsyncOutboxRepositoryExt - + Clone - + Send - + Sync - + 'static, + R: AsyncGetStream + AsyncTransactionalCommit + Clone + Send + Sync + 'static, + S: AsyncOutboxStore + Send + Sync, { let seat_id = unique_id("outbox-seat"); let message_id = unique_id("outbox-message"); @@ -32,7 +27,7 @@ where .await .expect("aggregate and outbox should commit atomically"); - let stored = find_outbox_by_id(&repo, &message_id) + let stored = find_outbox_by_id(&outbox, &message_id) .await .expect("outbox message should be stored"); assert_eq!(stored.status, OutboxMessageStatus::Pending); @@ -58,13 +53,7 @@ where pub async fn duplicate_outbox_insert_rolls_back_aggregate(repo: R) where - R: AsyncGetStream - + AsyncTransactionalCommit - + AsyncOutboxRepositoryExt - + Clone - + Send - + Sync - + 'static, + R: AsyncGetStream + AsyncTransactionalCommit + Clone + Send + Sync + 'static, { let duplicate_message_id = unique_id("duplicate-outbox"); let mut existing_seat = added_seat(&unique_id("existing-seat")); @@ -106,15 +95,10 @@ where assert_eq!(rollback_seat.entity.committed_version(), 0); } -pub async fn aggregate_conflict_rolls_back_outbox(repo: R) +pub async fn aggregate_conflict_rolls_back_outbox(repo: R, outbox: S) where - R: AsyncGetStream - + AsyncTransactionalCommit - + AsyncOutboxRepositoryExt - + Clone - + Send - + Sync - + 'static, + R: AsyncGetStream + AsyncTransactionalCommit + Clone + Send + Sync + 'static, + S: AsyncOutboxStore + Send + Sync, { let seat_id = unique_id("conflict-outbox-seat"); let seat_repo = repo.clone().async_aggregate::(); @@ -165,18 +149,13 @@ where .expect_err("stale aggregate should reject the batch"); assert!(matches!(err, RepositoryError::ConcurrentWrite { .. })); - assert!(find_outbox_by_id(&repo, &message_id).await.is_none()); + assert!(find_outbox_by_id(&outbox, &message_id).await.is_none()); } -pub async fn worker_claim_complete_and_retry_lifecycle(repo: R) +pub async fn worker_claim_complete_and_retry_lifecycle(repo: R, outbox: S) where - R: AsyncGetStream - + AsyncTransactionalCommit - + AsyncOutboxRepositoryExt - + Clone - + Send - + Sync - + 'static, + R: AsyncGetStream + AsyncTransactionalCommit + Clone + Send + Sync + 'static, + S: AsyncOutboxStore + Send + Sync, { let complete_message_id = unique_id("complete-outbox"); let mut complete_seat = added_seat(&unique_id("complete-seat")); @@ -189,23 +168,35 @@ where .await .expect("complete message should be stored"); - let claimed = repo - .claim_outbox_messages_async("worker-a", 1, Duration::from_secs(60)) + let claimed = outbox + .claim_async(ClaimOutboxMessages::new( + "worker-a", + 1, + Duration::from_secs(60), + )) .await .expect("claim should succeed"); assert_eq!(claimed.len(), 1); assert_eq!(claimed[0].id(), complete_message_id); - let stale_err = repo - .complete_outbox_message_for_worker_async(claimed[0].id(), "worker-b") + let wrong_claim = OutboxClaimRef { + message_id: claimed[0].id().to_string(), + worker_id: "worker-b".into(), + leased_until: claimed[0].leased_until.expect("claim should have lease"), + attempt: claimed[0].attempts, + }; + let stale_err = outbox + .complete_async(&wrong_claim) .await .expect_err("wrong worker should not complete a claim"); assert!(matches!(stale_err, RepositoryError::InvalidState { .. })); - repo.complete_outbox_message_for_worker_async(claimed[0].id(), "worker-a") + let claim = OutboxClaimRef::from_message(&claimed[0]).expect("claim should be valid"); + outbox + .complete_async(&claim) .await .expect("owning worker should complete the claim"); - let published = find_outbox_by_id(&repo, &complete_message_id) + let published = find_outbox_by_id(&outbox, &complete_message_id) .await .expect("completed message should still be queryable"); assert_eq!(published.status, OutboxMessageStatus::Published); @@ -221,34 +212,49 @@ where .await .expect("retry message should be stored"); - let claimed = repo - .claim_outbox_messages_async("worker-r", 1, Duration::from_secs(60)) + let claimed = outbox + .claim_async(ClaimOutboxMessages::new( + "worker-r", + 1, + Duration::from_secs(60), + )) .await .expect("retry claim should succeed"); - let action = repo - .record_outbox_publish_failure_async(claimed[0].id(), "worker-r", "first failure", 2) + let claim = OutboxClaimRef::from_message(&claimed[0]).expect("claim should be valid"); + let action = outbox + .record_failure_async(&claim, "first failure", 2) .await .expect("first failure should be recorded"); assert_eq!(action, OutboxPublishFailureAction::Released); - let released = find_outbox_by_id(&repo, &retry_message_id) + let released = find_outbox_by_id(&outbox, &retry_message_id) .await .expect("released message should exist"); assert_eq!(released.status, OutboxMessageStatus::Pending); assert_eq!(released.attempts, 1); assert_eq!(released.last_error.as_deref(), Some("first failure")); - let claimed = repo - .claim_outbox_messages_async("worker-r", 1, Duration::from_secs(60)) + let claimed = outbox + .claim_async(ClaimOutboxMessages::new( + "worker-r", + 1, + Duration::from_secs(60), + )) .await .expect("second retry claim should succeed"); - let action = repo - .record_outbox_publish_failure_async(claimed[0].id(), "worker-r", "second failure", 2) + let stale_err = outbox + .complete_async(&claim) + .await + .expect_err("stale attempt should not complete a later claim"); + assert!(matches!(stale_err, RepositoryError::InvalidState { .. })); + let claim = OutboxClaimRef::from_message(&claimed[0]).expect("claim should be valid"); + let action = outbox + .record_failure_async(&claim, "second failure", 2) .await .expect("second failure should be recorded"); assert_eq!(action, OutboxPublishFailureAction::Failed); - let failed = find_outbox_by_id(&repo, &retry_message_id) + let failed = find_outbox_by_id(&outbox, &retry_message_id) .await .expect("failed message should exist"); assert_eq!(failed.status, OutboxMessageStatus::Failed); @@ -263,9 +269,9 @@ fn added_seat(id: &str) -> Seat { seat } -async fn find_outbox_by_id(repo: &R, id: &str) -> Option +async fn find_outbox_by_id(outbox: &S, id: &str) -> Option where - R: AsyncOutboxRepositoryExt + Send + Sync, + S: AsyncOutboxStore + Send + Sync, { for status in [ OutboxMessageStatus::Pending, @@ -273,8 +279,8 @@ where OutboxMessageStatus::Published, OutboxMessageStatus::Failed, ] { - let messages = repo - .outbox_messages_by_status_async(status) + let messages = outbox + .messages_by_status_async(status) .await .expect("outbox status lookup should succeed"); if let Some(message) = messages.into_iter().find(|message| message.id() == id) { diff --git a/tests/postgres_repository_conformance/main.rs b/tests/postgres_repository_conformance/main.rs index 078765097..bcaadbc7d 100644 --- a/tests/postgres_repository_conformance/main.rs +++ b/tests/postgres_repository_conformance/main.rs @@ -81,7 +81,11 @@ async fn high_level_outbox_commit_persists_row_without_stream() { let Some(repo) = repository().await else { return; }; - conformance::outbox::high_level_outbox_commit_persists_row_without_stream(repo).await; + conformance::outbox::high_level_outbox_commit_persists_row_without_stream( + repo.clone(), + repo.outbox_store(), + ) + .await; } #[tokio::test] @@ -97,7 +101,8 @@ async fn aggregate_conflict_rolls_back_outbox() { let Some(repo) = repository().await else { return; }; - conformance::outbox::aggregate_conflict_rolls_back_outbox(repo).await; + conformance::outbox::aggregate_conflict_rolls_back_outbox(repo.clone(), repo.outbox_store()) + .await; } #[tokio::test] @@ -105,5 +110,9 @@ async fn worker_claim_complete_and_retry_lifecycle() { let Some(repo) = repository().await else { return; }; - conformance::outbox::worker_claim_complete_and_retry_lifecycle(repo).await; + conformance::outbox::worker_claim_complete_and_retry_lifecycle( + repo.clone(), + repo.outbox_store(), + ) + .await; } diff --git a/tests/sagas/distributed.rs b/tests/sagas/distributed.rs index b463a11d3..e2bae76ee 100644 --- a/tests/sagas/distributed.rs +++ b/tests/sagas/distributed.rs @@ -50,7 +50,7 @@ fn distributed_saga_with_threads() { let order_fulfillment_saga_thread = thread::spawn(move || { let repo = HashMapRepository::new(); let worker = OutboxWorkerThread::spawn( - repo.clone(), + repo.outbox_store(), order_fulfillment_saga_queue.clone(), Duration::from_millis(10), ); @@ -194,8 +194,11 @@ fn distributed_saga_with_threads() { let order_queue = queue.clone(); let order_thread = thread::spawn(move || { let repo = HashMapRepository::new(); - let worker = - OutboxWorkerThread::spawn(repo.clone(), order_queue.clone(), Duration::from_millis(10)); + let worker = OutboxWorkerThread::spawn( + repo.outbox_store(), + order_queue.clone(), + Duration::from_millis(10), + ); let order_repo = repo.queued().aggregate::(); // Create a bus for this service @@ -282,7 +285,7 @@ fn distributed_saga_with_threads() { let inventory_thread = thread::spawn(move || { let repo = HashMapRepository::new(); let worker = OutboxWorkerThread::spawn( - repo.clone(), + repo.outbox_store(), inventory_queue.clone(), Duration::from_millis(10), ); @@ -344,7 +347,7 @@ fn distributed_saga_with_threads() { let payment_thread = thread::spawn(move || { let repo = HashMapRepository::new(); let worker = OutboxWorkerThread::spawn( - repo.clone(), + repo.outbox_store(), payment_queue.clone(), Duration::from_millis(10), ); @@ -459,7 +462,7 @@ fn distributed_saga_with_send_listen() { let repo = HashMapRepository::new(); // spawn_routed: checks msg.destination → send() if set, publish() if not let worker = OutboxWorkerThread::spawn_routed( - repo.clone(), + repo.outbox_store(), order_fulfillment_saga_queue.clone(), Duration::from_millis(10), ); @@ -603,7 +606,7 @@ fn distributed_saga_with_send_listen() { let order_thread = thread::spawn(move || { let repo = HashMapRepository::new(); let worker = OutboxWorkerThread::spawn_routed( - repo.clone(), + repo.outbox_store(), order_queue.clone(), Duration::from_millis(10), ); @@ -709,7 +712,7 @@ fn distributed_saga_with_send_listen() { let inventory_thread = thread::spawn(move || { let repo = HashMapRepository::new(); let worker = OutboxWorkerThread::spawn_routed( - repo.clone(), + repo.outbox_store(), inventory_queue.clone(), Duration::from_millis(10), ); @@ -784,7 +787,7 @@ fn distributed_saga_with_send_listen() { let payment_thread = thread::spawn(move || { let repo = HashMapRepository::new(); let worker = OutboxWorkerThread::spawn_routed( - repo.clone(), + repo.outbox_store(), payment_queue.clone(), Duration::from_millis(10), ); @@ -911,8 +914,11 @@ fn metadata_propagates_across_bus_to_subscriber() { let producer_queue = queue.clone(); let producer_thread = thread::spawn(move || { let repo = HashMapRepository::new(); - let worker = - OutboxWorkerThread::spawn(repo.clone(), producer_queue, Duration::from_millis(10)); + let worker = OutboxWorkerThread::spawn( + repo.outbox_store(), + producer_queue, + Duration::from_millis(10), + ); let order_repo = repo.aggregate::(); // Create an order with metadata on the entity diff --git a/tests/sagas/microsvc_saga.rs b/tests/sagas/microsvc_saga.rs index b9d09ec2c..3e74e5c2a 100644 --- a/tests/sagas/microsvc_saga.rs +++ b/tests/sagas/microsvc_saga.rs @@ -239,7 +239,8 @@ fn saga_distributed() { // === SAGA SERVICE === let saga_repo = HashMapRepository::new(); - let saga_worker = OutboxWorkerThread::spawn_routed(saga_repo.clone(), queue.clone(), poll); + let saga_worker = + OutboxWorkerThread::spawn_routed(saga_repo.outbox_store(), queue.clone(), poll); let saga_svc = Arc::new(sourced_rust::register_handlers!( Service::new(saga_repo.queued().aggregate::()), handlers::saga::start, @@ -252,7 +253,8 @@ fn saga_distributed() { // === ORDER SERVICE === let order_repo = HashMapRepository::new(); - let order_worker = OutboxWorkerThread::spawn_routed(order_repo.clone(), queue.clone(), poll); + let order_worker = + OutboxWorkerThread::spawn_routed(order_repo.outbox_store(), queue.clone(), poll); let order_svc = Arc::new(sourced_rust::register_handlers!( Service::new(order_repo.queued().aggregate::()), handlers::orders::create, @@ -263,7 +265,7 @@ fn saga_distributed() { // === INVENTORY SERVICE === let inventory_repo = HashMapRepository::new(); let inventory_worker = - OutboxWorkerThread::spawn_routed(inventory_repo.clone(), queue.clone(), poll); + OutboxWorkerThread::spawn_routed(inventory_repo.outbox_store(), queue.clone(), poll); // Pre-seed inventory before starting the service { @@ -284,7 +286,7 @@ fn saga_distributed() { // === PAYMENT SERVICE === let payment_repo = HashMapRepository::new(); let payment_worker = - OutboxWorkerThread::spawn_routed(payment_repo.clone(), queue.clone(), poll); + OutboxWorkerThread::spawn_routed(payment_repo.outbox_store(), queue.clone(), poll); let payment_svc = Arc::new(sourced_rust::register_handlers!( Service::new(payment_repo.queued().aggregate::()), handlers::payments::process, diff --git a/tests/sourced_snapshot/main.rs b/tests/sourced_snapshot/main.rs index cedb47596..4f4b11253 100644 --- a/tests/sourced_snapshot/main.rs +++ b/tests/sourced_snapshot/main.rs @@ -2,7 +2,7 @@ mod aggregates; use aggregates::*; use sourced_rust::{ - AggregateBuilder, HashMapRepository, OutboxCommitExt, OutboxMessage, OutboxRepositoryExt, + AggregateBuilder, HashMapRepository, OutboxCommitExt, OutboxMessage, OutboxStore, SnapshotStore, Snapshottable, }; @@ -259,7 +259,7 @@ fn domain_event_commits_with_outbox() { let loaded = repo.get("t1").unwrap().unwrap(); assert_eq!(loaded.snapshot().task, "Ship it"); - let pending = repo.repo().outbox_messages_pending().unwrap(); + let pending = repo.repo().outbox_store().pending().unwrap(); assert_eq!(pending.len(), 1); assert!(pending[0].is_pending()); } diff --git a/tests/sqlite_repository/main.rs b/tests/sqlite_repository/main.rs index 2f50e4070..da3a9fab6 100644 --- a/tests/sqlite_repository/main.rs +++ b/tests/sqlite_repository/main.rs @@ -5,7 +5,7 @@ use sourced_rust::{ impl_aggregate, Aggregate, AsyncAggregateBuilder, AsyncCommitBatch, AsyncGetStream, AsyncReadModelSessionStore, AsyncReadModelStore, AsyncSnapshotStore, AsyncStreamWrite, AsyncTransactionalCommit, Entity, EventRecord, ReadModel, ReadModelSession, RepositoryError, - SnapshotRecord, SqliteRepository, StreamIdentity, + SnapshotRecord, SqliteRepository, StreamIdentity, TableSchemaRegistry, OUTBOX_MESSAGES_TABLE, }; #[derive(Default)] @@ -96,6 +96,38 @@ async fn migration_is_idempotent_and_aggregate_stream_round_trips() { assert_eq!(loaded.entity().events()[0].correlation_id(), Some("corr-1")); } +#[tokio::test] +async fn dev_bootstrap_applies_registered_table_schemas() { + let repo = SqliteRepository::connect("sqlite::memory:").await.unwrap(); + let mut registry = TableSchemaRegistry::new(); + registry + .register_schema(sourced_rust::outbox_message_schema()) + .unwrap(); + + let artifacts = repo.generate_table_migration_artifacts(®istry).unwrap(); + assert!(artifacts[0] + .statements + .iter() + .any(|statement| statement.contains("CREATE TABLE IF NOT EXISTS \"outbox_messages\""))); + + let bootstrap = repo + .bootstrap_table_schema_for_dev(®istry) + .await + .unwrap(); + assert_eq!( + bootstrap.bootstrapped_tables, + vec![OUTBOX_MESSAGES_TABLE.to_string()] + ); + + let row = sqlx::query("SELECT name FROM sqlite_master WHERE type = 'table' AND name = ?") + .bind(OUTBOX_MESSAGES_TABLE) + .fetch_one(repo.pool()) + .await + .unwrap(); + let table_name: String = sqlx::Row::try_get(&row, "name").unwrap(); + assert_eq!(table_name, OUTBOX_MESSAGES_TABLE); +} + #[tokio::test] async fn aggregate_stream_identity_separates_same_id_across_types() { let repo = repository().await; diff --git a/tests/sqlite_repository_conformance/main.rs b/tests/sqlite_repository_conformance/main.rs index 8d3f3ff3b..e1c7356f3 100644 --- a/tests/sqlite_repository_conformance/main.rs +++ b/tests/sqlite_repository_conformance/main.rs @@ -53,8 +53,12 @@ async fn snapshots_use_full_stream_identity() { #[tokio::test] async fn high_level_outbox_commit_persists_row_without_stream() { - conformance::outbox::high_level_outbox_commit_persists_row_without_stream(repository().await) - .await; + let repo = repository().await; + conformance::outbox::high_level_outbox_commit_persists_row_without_stream( + repo.clone(), + repo.outbox_store(), + ) + .await; } #[tokio::test] @@ -64,10 +68,17 @@ async fn duplicate_outbox_insert_rolls_back_aggregate() { #[tokio::test] async fn aggregate_conflict_rolls_back_outbox() { - conformance::outbox::aggregate_conflict_rolls_back_outbox(repository().await).await; + let repo = repository().await; + conformance::outbox::aggregate_conflict_rolls_back_outbox(repo.clone(), repo.outbox_store()) + .await; } #[tokio::test] async fn worker_claim_complete_and_retry_lifecycle() { - conformance::outbox::worker_claim_complete_and_retry_lifecycle(repository().await).await; + let repo = repository().await; + conformance::outbox::worker_claim_complete_and_retry_lifecycle( + repo.clone(), + repo.outbox_store(), + ) + .await; } diff --git a/tests/todos/main.rs b/tests/todos/main.rs index fd40370ae..f8f34b911 100644 --- a/tests/todos/main.rs +++ b/tests/todos/main.rs @@ -2,9 +2,10 @@ mod aggregate; use aggregate::{Todo, TodoSnapshot}; use sourced_rust::{ - AggregateBuilder, Commit, CommitBuilderExt, EventEmitter, GetAggregate, HashMapRepository, - LocalEmitterPublisher, LockError, LogPublisher, OutboxCommitExt, OutboxMessage, - OutboxMessageStatus, OutboxRepositoryExt, OutboxWorker, Queueable, RepositoryError, + AggregateBuilder, ClaimOutboxMessages, Commit, CommitBuilderExt, EventEmitter, GetAggregate, + HashMapRepository, LocalEmitterPublisher, LockError, LogPublisher, OutboxClaimRef, + OutboxCommitExt, OutboxMessage, OutboxMessageStatus, OutboxStore, OutboxWorker, Queueable, + RepositoryError, }; use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::{mpsc, Arc, Mutex}; @@ -20,26 +21,27 @@ fn next_id() -> String { fn complete_published_outbox( repo: &HashMapRepository, - worker_id: &str, messages: &[OutboxMessage], + claims: &[OutboxClaimRef], ) { - for message in messages { + let store = repo.outbox_store(); + for (message, claim) in messages.iter().zip(claims) { if message.is_published() { - repo.complete_outbox_message_for_worker(message.id(), worker_id) - .unwrap(); + store.complete(claim).unwrap(); } } } fn load_outbox_message(repo: &HashMapRepository, id: &str) -> OutboxMessage { + let store = repo.outbox_store(); for status in [ OutboxMessageStatus::Pending, OutboxMessageStatus::InFlight, OutboxMessageStatus::Published, OutboxMessageStatus::Failed, ] { - if let Some(message) = repo - .outbox_messages_by_status(status) + if let Some(message) = store + .messages_by_status(status) .unwrap() .into_iter() .find(|message| message.id() == id) @@ -74,7 +76,7 @@ fn todos() { // Verify the outbox event was captured { - let pending = repo.repo().inner().outbox_messages_pending().unwrap(); + let pending = repo.repo().inner().outbox_store().pending().unwrap(); assert_eq!(pending.len(), 1); assert_eq!(pending[0].event_type, "TodoInitialized"); } @@ -95,7 +97,7 @@ fn todos() { // Verify we now have 2 outbox events { - let pending = repo.repo().inner().outbox_messages_pending().unwrap(); + let pending = repo.repo().inner().outbox_store().pending().unwrap(); assert_eq!(pending.len(), 2); assert!(pending .iter() @@ -219,7 +221,7 @@ fn outbox_records_persisted() { repo.outbox(message).commit(&mut todo).unwrap(); // Check pending outbox messages - let pending = repo.outbox_messages_pending().unwrap(); + let pending = repo.outbox_store().pending().unwrap(); assert_eq!(pending.len(), 1); assert_eq!(pending[0].event_type, "TodoInitialized"); @@ -256,12 +258,22 @@ fn outbox_worker_log_publisher() { .with_max_attempts(3); // Claim pending messages and process - let mut claimed = repo - .claim_outbox_messages("logger-1", 10, Duration::from_secs(30)) + let store = repo.outbox_store(); + let mut claimed = store + .claim(ClaimOutboxMessages::new( + "logger-1", + 10, + Duration::from_secs(30), + )) + .unwrap(); + let claims = claimed + .iter() + .map(OutboxClaimRef::from_message) + .collect::, _>>() .unwrap(); let result = worker.process_batch(&mut claimed).unwrap(); assert_eq!(result.completed, 1); - complete_published_outbox(&repo, "logger-1", &claimed); + complete_published_outbox(&repo, &claimed, &claims); let lines = buffer.lock().unwrap(); assert_eq!(lines.len(), 1); @@ -301,12 +313,22 @@ fn outbox_worker_local_emitter_publisher() { .with_max_attempts(3); // Claim pending messages and process - let mut claimed = repo - .claim_outbox_messages("emitter-1", 10, Duration::from_secs(30)) + let store = repo.outbox_store(); + let mut claimed = store + .claim(ClaimOutboxMessages::new( + "emitter-1", + 10, + Duration::from_secs(30), + )) + .unwrap(); + let claims = claimed + .iter() + .map(OutboxClaimRef::from_message) + .collect::, _>>() .unwrap(); let result = worker.process_batch(&mut claimed).unwrap(); assert_eq!(result.completed, 1); - complete_published_outbox(&repo, "emitter-1", &claimed); + complete_published_outbox(&repo, &claimed, &claims); // LocalEmitterPublisher converts bytes to lossy string, so we just verify something was received let payload = rx.recv_timeout(Duration::from_secs(1)).unwrap(); @@ -582,15 +604,25 @@ fn outbox_worker_process_next_with_commit() { let mut processed = 0; loop { - let mut claimed = repo - .claim_outbox_messages("safe-worker", 1, Duration::from_secs(30)) + let store = repo.outbox_store(); + let mut claimed = store + .claim(ClaimOutboxMessages::new( + "safe-worker", + 1, + Duration::from_secs(30), + )) .unwrap(); if claimed.is_empty() { break; } + let claims = claimed + .iter() + .map(OutboxClaimRef::from_message) + .collect::, _>>() + .unwrap(); let result = worker.process_batch(&mut claimed).unwrap(); processed += result.completed + result.released + result.failed; - complete_published_outbox(&repo, "safe-worker", &claimed); + complete_published_outbox(&repo, &claimed, &claims); } assert_eq!(processed, 3); @@ -598,7 +630,7 @@ fn outbox_worker_process_next_with_commit() { let message = load_outbox_message(&repo, id); assert!(message.is_published()); } - assert_eq!(repo.outbox_messages_pending().unwrap().len(), 0); + assert_eq!(repo.outbox_store().pending().unwrap().len(), 0); let lines = buffer.lock().unwrap(); assert_eq!(lines.len(), 3); @@ -645,12 +677,22 @@ fn metadata_flows_from_entity_through_outbox_to_publisher() { let publisher = LogPublisher::with_buffer(Arc::clone(&buffer)); let mut worker = OutboxWorker::new(publisher).with_worker_id("meta-worker"); - let mut claimed = repo - .repo() - .claim_outbox_messages("meta-worker", 10, Duration::from_secs(30)) + let store = repo.repo().outbox_store(); + let mut claimed = store + .claim(ClaimOutboxMessages::new( + "meta-worker", + 10, + Duration::from_secs(30), + )) + .unwrap(); + let claims = claimed + .iter() + .map(OutboxClaimRef::from_message) + .collect::, _>>() .unwrap(); let result = worker.process_batch(&mut claimed).unwrap(); assert_eq!(result.completed, 1); + complete_published_outbox(repo.repo(), &claimed, &claims); // 6. Verify the publisher received metadata let lines = buffer.lock().unwrap(); From f7a6c7abe29556351fd3d3791963e250c2523a41 Mon Sep 17 00:00:00 2001 From: Patrick Lee Scott Date: Mon, 25 May 2026 13:14:41 -0500 Subject: [PATCH 2/4] Address outbox review feedback --- docs/read-models.md | 2 +- src/hashmap_repo/repository.rs | 10 +-- src/outbox/mod.rs | 1 + src/outbox/table.rs | 17 +++-- src/outbox_worker/store.rs | 77 ++++++++++++++++++++--- src/outbox_worker/thread.rs | 11 ++-- src/read_model/metadata.rs | 1 + src/sqlite_repo/mod.rs | 8 ++- src/sqlx_repo/mod.rs | 12 +--- src/table/sql.rs | 29 +++++++++ tests/read_model_schema_bootstrap/main.rs | 1 + 11 files changed, 128 insertions(+), 41 deletions(-) diff --git a/docs/read-models.md b/docs/read-models.md index b7c41207e..c23ccddf9 100644 --- a/docs/read-models.md +++ b/docs/read-models.md @@ -262,7 +262,7 @@ path: ```rust let mut registry = TableSchemaRegistry::new(); -registry.register_schema(outbox_message_schema())?; +registry.register_schema(sourced_rust::outbox_message_schema())?; let artifacts = repo.generate_table_migration_artifacts(®istry)?; let bootstrap = repo.bootstrap_table_schema_for_dev(®istry).await?; diff --git a/src/hashmap_repo/repository.rs b/src/hashmap_repo/repository.rs index 7395b1de6..f723fb55c 100644 --- a/src/hashmap_repo/repository.rs +++ b/src/hashmap_repo/repository.rs @@ -389,7 +389,8 @@ fn reject_duplicate_async_streams(streams: &[AsyncStreamWrite<'_>]) -> Result<() fn reject_duplicate_outbox_messages(messages: &[OutboxMessage]) -> Result<(), RepositoryError> { let mut seen = HashSet::with_capacity(messages.len()); for message in messages { - validate_outbox_table_write(message)?; + crate::outbox::validate_outbox_message_table_write(message) + .map_err(|err| RepositoryError::Model(err.to_string()))?; let id = message.id(); if id.trim().is_empty() { return Err(RepositoryError::Model( @@ -408,13 +409,6 @@ fn reject_duplicate_outbox_messages(messages: &[OutboxMessage]) -> Result<(), Re Ok(()) } -fn validate_outbox_table_write(message: &OutboxMessage) -> Result<(), RepositoryError> { - crate::outbox::outbox_message_insert_plan(message) - .and_then(|plan| plan.validate().map(|()| plan)) - .map(|_| ()) - .map_err(|err| RepositoryError::Model(err.to_string())) -} - fn validate_async_entity_id_matches_identity( streams: &[AsyncStreamWrite<'_>], ) -> Result<(), RepositoryError> { diff --git a/src/outbox/mod.rs b/src/outbox/mod.rs index b83d3262a..eb446fc48 100644 --- a/src/outbox/mod.rs +++ b/src/outbox/mod.rs @@ -43,6 +43,7 @@ mod table; // Outbox message record pub use message::{OutboxMessage, OutboxMessageStatus}; +pub(crate) use table::validate_outbox_message_table_write; pub use table::{ outbox_message_insert_plan, outbox_message_key, outbox_message_row_values, outbox_message_schema, OUTBOX_MESSAGES_TABLE, diff --git a/src/outbox/table.rs b/src/outbox/table.rs index a6588bc46..63295365d 100644 --- a/src/outbox/table.rs +++ b/src/outbox/table.rs @@ -23,14 +23,14 @@ pub fn outbox_message_schema() -> TableSchema { table_column("destination", ColumnType::Text, true), table_column("metadata", ColumnType::Json, false), table_column("status", ColumnType::Text, false), - table_column("created_at", ColumnType::UnsignedInteger, false), - table_column("next_available_at", ColumnType::UnsignedInteger, false), + table_column("created_at", ColumnType::Timestamp, false), + table_column("next_available_at", ColumnType::Timestamp, false), table_column("claimed_by", ColumnType::Text, true), - table_column("claimed_until", ColumnType::UnsignedInteger, true), + table_column("claimed_until", ColumnType::Timestamp, true), table_column("attempts", ColumnType::UnsignedInteger, false), table_column("last_error", ColumnType::Text, true), - table_column("published_at", ColumnType::UnsignedInteger, true), - table_column("failed_at", ColumnType::UnsignedInteger, true), + table_column("published_at", ColumnType::Timestamp, true), + table_column("failed_at", ColumnType::Timestamp, true), table_column("source_aggregate_type", ColumnType::Text, true), table_column("source_aggregate_id", ColumnType::Text, true), table_column("source_sequence", ColumnType::UnsignedInteger, true), @@ -72,6 +72,13 @@ pub fn outbox_message_insert_plan( Ok(TableWritePlan::new(vec![mutation], Vec::new())) } +pub(crate) fn validate_outbox_message_table_write( + message: &OutboxMessage, +) -> Result<(), TableStoreError> { + let plan = outbox_message_insert_plan(message)?; + plan.validate() +} + pub fn outbox_message_key(message_id: &str) -> RowKey { RowKey::new([("message_id", RowValue::String(message_id.to_string()))]) } diff --git a/src/outbox_worker/store.rs b/src/outbox_worker/store.rs index cd97082f8..2a8c77047 100644 --- a/src/outbox_worker/store.rs +++ b/src/outbox_worker/store.rs @@ -195,6 +195,26 @@ pub(crate) fn ensure_active_claim( Ok(()) } +fn sort_by_claim_order(messages: &mut [OutboxMessage]) { + messages.sort_by(|left, right| { + left.created_at + .cmp(&right.created_at) + .then_with(|| left.id().cmp(right.id())) + }); +} + +fn claim_order_ids<'a>(messages: impl Iterator) -> Vec { + let mut entries = messages + .map(|message| (message.created_at, message.id().to_string())) + .collect::>(); + entries.sort_by(|(left_time, left_id), (right_time, right_id)| { + left_time + .cmp(right_time) + .then_with(|| left_id.cmp(right_id)) + }); + entries.into_iter().map(|(_, id)| id).collect() +} + impl HashMapOutboxStore { fn update_outbox_message( &self, @@ -231,11 +251,7 @@ impl OutboxStore for HashMapOutboxStore { .filter(|message| message.status == status) .cloned() .collect::>(); - messages.sort_by(|left, right| { - left.created_at - .cmp(&right.created_at) - .then_with(|| left.id().cmp(right.id())) - }); + sort_by_claim_order(&mut messages); Ok(messages) } @@ -250,8 +266,7 @@ impl OutboxStore for HashMapOutboxStore { } let now = SystemTime::now(); - let mut ids = storage.keys().cloned().collect::>(); - ids.sort(); + let ids = claim_order_ids(storage.values()); let mut claimed = Vec::new(); for id in ids { let Some(message) = storage.get_mut(&id) else { @@ -334,7 +349,7 @@ impl AsyncOutboxStore for HashMapOutboxStore { .filter(|message| message.status == status) .cloned() .collect::>(); - messages.sort_by_key(|message| message.created_at); + sort_by_claim_order(&mut messages); Ok(messages) } } @@ -353,8 +368,7 @@ impl AsyncOutboxStore for HashMapOutboxStore { .storage .write() .map_err(|_| RepositoryError::LockPoisoned("async outbox write"))?; - let mut ids = storage.keys().cloned().collect::>(); - ids.sort(); + let ids = claim_order_ids(storage.values()); let mut claimed = Vec::new(); for id in ids { @@ -554,6 +568,49 @@ mod tests { assert_eq!(stored.attempts, 1); } + #[test] + fn claim_uses_created_at_before_message_id_order() { + let repo = HashMapRepository::new(); + let mut newer = OutboxMessage::create("msg-a", "Event", b"{}".to_vec()).unwrap(); + newer.created_at = SystemTime::UNIX_EPOCH + Duration::from_secs(10); + let mut older = OutboxMessage::create("msg-z", "Event", b"{}".to_vec()).unwrap(); + older.created_at = SystemTime::UNIX_EPOCH + Duration::from_secs(1); + store_message(&repo, newer); + store_message(&repo, older); + + let claimed = repo + .outbox_store() + .claim(ClaimOutboxMessages::new( + "worker-1", + 1, + Duration::from_secs(60), + )) + .unwrap(); + + assert_eq!(claimed[0].id(), "msg-z"); + } + + #[test] + fn sort_by_claim_order_uses_message_id_tiebreaker() { + let mut later = OutboxMessage::create("msg-c", "Event", b"{}".to_vec()).unwrap(); + later.created_at = SystemTime::UNIX_EPOCH + Duration::from_secs(10); + let mut second = OutboxMessage::create("msg-b", "Event", b"{}".to_vec()).unwrap(); + second.created_at = SystemTime::UNIX_EPOCH + Duration::from_secs(1); + let mut first = OutboxMessage::create("msg-a", "Event", b"{}".to_vec()).unwrap(); + first.created_at = SystemTime::UNIX_EPOCH + Duration::from_secs(1); + let mut messages = vec![later, second, first]; + + sort_by_claim_order(&mut messages); + + assert_eq!( + messages + .iter() + .map(|message| message.id()) + .collect::>(), + vec!["msg-a", "msg-b", "msg-c"] + ); + } + #[test] fn competing_workers_only_claim_message_once() { let repo = HashMapRepository::new(); diff --git a/src/outbox_worker/thread.rs b/src/outbox_worker/thread.rs index 22a9abbf6..f69bbdde2 100644 --- a/src/outbox_worker/thread.rs +++ b/src/outbox_worker/thread.rs @@ -108,11 +108,10 @@ impl OutboxWorkerThread { /// The worker will poll the outbox store for pending outbox messages, /// publish them to the given publisher, and mark them as complete. /// - /// The store must be `Clone + Send + 'static`. For `HashMapOutboxStore`, - /// cloning creates another handle to the same storage. + /// The store must be `Send + 'static`. pub fn spawn(store: S, publisher: P, poll_interval: Duration) -> Self where - S: OutboxStore + Clone + Send + 'static, + S: OutboxStore + Send + 'static, P: Publisher + 'static, { Self::spawn_with_id(store, publisher, poll_interval, "outbox-worker") @@ -126,7 +125,7 @@ impl OutboxWorkerThread { worker_id: &str, ) -> Self where - S: OutboxStore + Clone + Send + 'static, + S: OutboxStore + Send + 'static, P: Publisher + 'static, { let (stop_tx, stop_rx) = channel(); @@ -200,7 +199,7 @@ impl OutboxWorkerThread { /// Messages without a destination are published fan-out via `Publisher::publish()`. pub fn spawn_routed(store: S, publisher: P, poll_interval: Duration) -> Self where - S: OutboxStore + Clone + Send + 'static, + S: OutboxStore + Send + 'static, P: Publisher + BusSender + 'static, { Self::spawn_routed_with_id(store, publisher, poll_interval, "outbox-worker") @@ -214,7 +213,7 @@ impl OutboxWorkerThread { worker_id: &str, ) -> Self where - S: OutboxStore + Clone + Send + 'static, + S: OutboxStore + Send + 'static, P: Publisher + BusSender + 'static, { let (stop_tx, stop_rx) = channel(); diff --git a/src/read_model/metadata.rs b/src/read_model/metadata.rs index cb341fdaf..d0da4624d 100644 --- a/src/read_model/metadata.rs +++ b/src/read_model/metadata.rs @@ -16,6 +16,7 @@ pub enum ColumnType { Float, Bytes, Json, + Timestamp, Unsupported(String), } diff --git a/src/sqlite_repo/mod.rs b/src/sqlite_repo/mod.rs index 29d3fd3f2..b554313b7 100644 --- a/src/sqlite_repo/mod.rs +++ b/src/sqlite_repo/mod.rs @@ -133,7 +133,7 @@ impl SqliteRepository { sqlx::query(&statement) .execute(&self.pool) .await - .map_err(|err| read_model_storage_error("bootstrap table schema", err))?; + .map_err(|err| table_schema_storage_error("bootstrap table schema", err))?; } Ok(table_schema_bootstrap_result(registry)) } @@ -177,7 +177,7 @@ impl SqliteOutboxStore { sqlx::query(&statement) .execute(&self.pool) .await - .map_err(|err| read_model_storage_error("bootstrap table schema", err))?; + .map_err(|err| table_schema_storage_error("bootstrap table schema", err))?; } Ok(table_schema_bootstrap_result(registry)) } @@ -1594,3 +1594,7 @@ fn repository_storage_error(operation: &str, err: sqlx::Error) -> RepositoryErro fn read_model_storage_error(operation: &str, err: sqlx::Error) -> ReadModelError { sqlx_repo::read_model_storage_error(SQLITE_BACKEND, operation, err) } + +fn table_schema_storage_error(operation: &str, err: sqlx::Error) -> TableStoreError { + TableStoreError::Storage(format!("{SQLITE_BACKEND} {operation} failed: {err}")) +} diff --git a/src/sqlx_repo/mod.rs b/src/sqlx_repo/mod.rs index fb7a90a53..6cd5a9822 100644 --- a/src/sqlx_repo/mod.rs +++ b/src/sqlx_repo/mod.rs @@ -3,7 +3,7 @@ use std::collections::{HashMap, HashSet}; use crate::entity::{ EventRecord, EventRecordError, BITCODE_PAYLOAD_CODEC, BITCODE_PAYLOAD_CODEC_VERSION, }; -use crate::outbox::OutboxMessage; +use crate::outbox::{validate_outbox_message_table_write, OutboxMessage}; #[cfg(feature = "sqlite")] use crate::read_model::ReadModelError; use crate::repository::{AsyncStreamWrite, PreparedEventAppend, RepositoryError, StreamIdentity}; @@ -29,7 +29,8 @@ pub(crate) fn reject_duplicate_outbox_messages( ) -> Result<(), RepositoryError> { let mut seen = HashSet::with_capacity(messages.len()); for message in messages { - validate_outbox_table_write(message)?; + validate_outbox_message_table_write(message) + .map_err(|err| RepositoryError::Model(err.to_string()))?; let id = message.id(); if id.trim().is_empty() { return Err(RepositoryError::Model( @@ -48,13 +49,6 @@ pub(crate) fn reject_duplicate_outbox_messages( Ok(()) } -fn validate_outbox_table_write(message: &OutboxMessage) -> Result<(), RepositoryError> { - crate::outbox::outbox_message_insert_plan(message) - .and_then(|plan| plan.validate().map(|()| plan)) - .map(|_| ()) - .map_err(|err| RepositoryError::Model(err.to_string())) -} - pub(crate) fn validate_entity_id_matches_identity( streams: &[AsyncStreamWrite<'_>], ) -> Result<(), RepositoryError> { diff --git a/src/table/sql.rs b/src/table/sql.rs index 7bd3725f5..f9344adb5 100644 --- a/src/table/sql.rs +++ b/src/table/sql.rs @@ -187,6 +187,7 @@ fn sql_type( (TableSqlDialect::Sqlite, ColumnType::Float) => "REAL", (TableSqlDialect::Sqlite, ColumnType::Bytes) => "BLOB", (TableSqlDialect::Sqlite, ColumnType::Json) => "TEXT", + (TableSqlDialect::Sqlite, ColumnType::Timestamp) => "TEXT", (TableSqlDialect::Postgres, ColumnType::Text) => "text", (TableSqlDialect::Postgres, ColumnType::Boolean) => "boolean", (TableSqlDialect::Postgres, ColumnType::Integer | ColumnType::UnsignedInteger) => "bigint", @@ -194,6 +195,7 @@ fn sql_type( (TableSqlDialect::Postgres, ColumnType::Bytes) => "bytea", (TableSqlDialect::Postgres, ColumnType::Json) if jsonb => "jsonb", (TableSqlDialect::Postgres, ColumnType::Json) => "jsonb", + (TableSqlDialect::Postgres, ColumnType::Timestamp) => "timestamptz", (_, ColumnType::Unsupported(type_name)) => { return Err(TableStoreError::Metadata(format!( "unsupported table column type `{type_name}`" @@ -244,6 +246,33 @@ mod tests { .statements .iter() .any(|statement| statement.contains("\"message_id\" TEXT NOT NULL"))); + assert!(artifact + .statements + .iter() + .any(|statement| statement.contains("\"created_at\" TEXT NOT NULL"))); + } + + #[test] + fn renders_outbox_table_schema_for_postgres_with_timestamp_columns() { + let mut registry = TableSchemaRegistry::new(); + registry + .register_schema(outbox_message_schema()) + .expect("schema should register"); + + let artifact = generate_table_migration_artifacts(®istry, TableSqlDialect::Postgres) + .expect("artifact should render") + .pop() + .expect("artifact should exist"); + + assert_eq!(artifact.name, "postgres-tables"); + assert!(artifact + .statements + .iter() + .any(|statement| statement.contains("\"created_at\" timestamptz NOT NULL"))); + assert!(artifact + .statements + .iter() + .any(|statement| statement.contains("\"claimed_until\" timestamptz"))); } #[test] diff --git a/tests/read_model_schema_bootstrap/main.rs b/tests/read_model_schema_bootstrap/main.rs index 4ae6136ae..9d04a39f9 100644 --- a/tests/read_model_schema_bootstrap/main.rs +++ b/tests/read_model_schema_bootstrap/main.rs @@ -161,6 +161,7 @@ fn logical_type_name(column_type: &ColumnType, jsonb: bool) -> &'static str { ColumnType::Float => "float", ColumnType::Bytes => "bytes", ColumnType::Json => "json", + ColumnType::Timestamp => "timestamp", ColumnType::Unsupported(_) => "unsupported", } } From 03b53d507cc4fb8a215c3b50f32194e3fffbac54 Mon Sep 17 00:00:00 2001 From: Patrick Lee Scott Date: Mon, 25 May 2026 14:31:41 -0500 Subject: [PATCH 3/4] Clear clippy debt in test fixtures --- tests/bomberman/commands.rs | 65 ++++++++++++------- tests/bomberman/domain/game_map.rs | 4 +- tests/sagas/order/mod.rs | 4 +- .../order/{order.rs => order_aggregate.rs} | 9 +-- tests/sagas/order/payment.rs | 9 +-- tests/sagas/order/saga.rs | 11 +--- 6 files changed, 54 insertions(+), 48 deletions(-) rename tests/sagas/order/{order.rs => order_aggregate.rs} (95%) diff --git a/tests/bomberman/commands.rs b/tests/bomberman/commands.rs index e80016ef0..776b74f68 100644 --- a/tests/bomberman/commands.rs +++ b/tests/bomberman/commands.rs @@ -13,6 +13,21 @@ use crate::domain::types::{Direction, Tile}; use crate::error::GameError; use crate::views::{build_board, BoardView}; +#[derive(Default)] +struct DamageReport { + blocks_destroyed: Vec<(i32, i32)>, + players_killed: Vec, + chain_detonations: Vec, +} + +impl DamageReport { + fn has_damage(&self) -> bool { + !self.blocks_destroyed.is_empty() + || !self.players_killed.is_empty() + || !self.chain_detonations.is_empty() + } +} + fn load_board(repo: &R, game_id: &str) -> Result { repo.read_models::() .get_by_primary_key(game_id) @@ -117,11 +132,14 @@ pub fn tick( explosion.expand()?; let new_cells = explosion.newly_reached_cells().to_vec(); - let (blocks, killed, chains) = - apply_damage(&new_cells, &mut map, &mut players, &mut bombs, None)?; - - if !blocks.is_empty() || !killed.is_empty() || !chains.is_empty() { - saga.record_damage(blocks, killed, chains)?; + let damage = apply_damage(&new_cells, &mut map, &mut players, &mut bombs, None)?; + + if damage.has_damage() { + saga.record_damage( + damage.blocks_destroyed, + damage.players_killed, + damage.chain_detonations, + )?; } } } @@ -159,7 +177,7 @@ pub fn tick( // Apply center-cell damage (ring 0) let center_cells = explosion.newly_reached_cells().to_vec(); - let (blocks, killed, chains) = + let damage = apply_damage(¢er_cells, &mut map, &mut players, &mut bombs, Some(idx))?; saga.record_detonation(Detonation { @@ -168,8 +186,12 @@ pub fn tick( explosion_id, })?; - if !blocks.is_empty() || !killed.is_empty() || !chains.is_empty() { - saga.record_damage(blocks, killed, chains)?; + if damage.has_damage() { + saga.record_damage( + damage.blocks_destroyed, + damage.players_killed, + damage.chain_detonations, + )?; } explosions.push(explosion); @@ -256,46 +278,43 @@ pub fn tick( } /// Apply damage to cells: destroy blocks, kill players, mark chain detonations. -/// Returns (blocks_destroyed, players_killed, chain_detonations). fn apply_damage( cells: &[(i32, i32)], map: &mut GameMap, players: &mut [Player], bombs: &mut [Bomb], skip_bomb_idx: Option, -) -> Result<(Vec<(i32, i32)>, Vec, Vec), GameError> { - let mut blocks_destroyed = Vec::new(); - let mut players_killed = Vec::new(); - let mut chain_detonations = Vec::new(); +) -> Result { + let mut report = DamageReport::default(); for &(cx, cy) in cells { // Destroy blocks if map.is_in_bounds(cx, cy) && *map.tile_at(cx, cy) == Tile::Block { map.destroy_block(cx, cy)?; - blocks_destroyed.push((cx, cy)); + report.blocks_destroyed.push((cx, cy)); } // Kill players for player in players.iter_mut() { if player.alive && player.x == cx && player.y == cy { player.kill()?; - players_killed.push(player.entity.id().to_string()); + report.players_killed.push(player.entity.id().to_string()); } } // Chain-detonate other bombs - for i in 0..bombs.len() { + for (i, bomb) in bombs.iter_mut().enumerate() { if Some(i) == skip_bomb_idx { continue; } - if !bombs[i].exploded && bombs[i].x == cx && bombs[i].y == cy { - bombs[i].ticks_remaining = 0; - chain_detonations.push(bombs[i].entity.id().to_string()); + if !bomb.exploded && bomb.x == cx && bomb.y == cy { + bomb.ticks_remaining = 0; + report.chain_detonations.push(bomb.entity.id().to_string()); } } } - Ok((blocks_destroyed, players_killed, chain_detonations)) + Ok(report) } // ── Player commands ── @@ -493,7 +512,7 @@ pub fn calculate_blast_rings(bomb: &Bomb, map: &GameMap) -> Vec> let mut cx = bomb.x; let mut cy = bomb.y; - for dist in 1..=radius { + for ring in rings.iter_mut().take(radius + 1).skip(1) { let (nx, ny) = dir.apply(cx, cy); if !map.is_in_bounds(nx, ny) { @@ -503,11 +522,11 @@ pub fn calculate_blast_rings(bomb: &Bomb, map: &GameMap) -> Vec> match map.tile_at(nx, ny) { Tile::Wall => break, Tile::Block => { - rings[dist].push((nx, ny)); + ring.push((nx, ny)); break; } _ => { - rings[dist].push((nx, ny)); + ring.push((nx, ny)); } } diff --git a/tests/bomberman/domain/game_map.rs b/tests/bomberman/domain/game_map.rs index 5999e59df..a2b57822e 100644 --- a/tests/bomberman/domain/game_map.rs +++ b/tests/bomberman/domain/game_map.rs @@ -2,6 +2,8 @@ use sourced_rust::{digest, Entity, SourcedResult}; use super::types::{PowerUp, Tile}; +pub type ParsedMap = (usize, usize, Vec>, Vec<(i32, i32)>); + #[derive(Default)] pub struct GameMap { pub entity: Entity, @@ -91,7 +93,7 @@ impl GameMap { /// - `.` = Block (destructible) /// - ` ` = Floor /// - `1`-`4` = Spawn points (stored as Floor tiles) - pub fn from_ascii(ascii: &str) -> (usize, usize, Vec>, Vec<(i32, i32)>) { + pub fn from_ascii(ascii: &str) -> ParsedMap { let lines: Vec<&str> = ascii.lines().filter(|l| !l.is_empty()).collect(); let height = lines.len(); let width = lines.iter().map(|l| l.len()).max().unwrap_or(0); diff --git a/tests/sagas/order/mod.rs b/tests/sagas/order/mod.rs index f94907d19..9fbdb13c8 100644 --- a/tests/sagas/order/mod.rs +++ b/tests/sagas/order/mod.rs @@ -1,6 +1,6 @@ mod events; mod inventory; -mod order; +mod order_aggregate; mod payment; mod saga; @@ -9,6 +9,6 @@ pub use events::{ OrderFulfillmentStartedPayload, PaymentSucceededPayload, }; pub use inventory::Inventory; -pub use order::{Order, OrderItem, OrderStatus}; +pub use order_aggregate::{Order, OrderItem, OrderStatus}; pub use payment::{Payment, PaymentStatus}; pub use saga::{OrderFulfillmentSaga, SagaStatus}; diff --git a/tests/sagas/order/order.rs b/tests/sagas/order/order_aggregate.rs similarity index 95% rename from tests/sagas/order/order.rs rename to tests/sagas/order/order_aggregate.rs index ee0812330..8207409fe 100644 --- a/tests/sagas/order/order.rs +++ b/tests/sagas/order/order_aggregate.rs @@ -11,8 +11,9 @@ pub struct OrderItem { } /// Order status -#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] pub enum OrderStatus { + #[default] Pending, InventoryReserved, PaymentProcessed, @@ -31,12 +32,6 @@ pub struct Order { failure_reason: Option, } -impl Default for OrderStatus { - fn default() -> Self { - OrderStatus::Pending - } -} - #[allow(dead_code)] impl Order { pub fn new() -> Self { diff --git a/tests/sagas/order/payment.rs b/tests/sagas/order/payment.rs index d99d9c7c6..8263fa098 100644 --- a/tests/sagas/order/payment.rs +++ b/tests/sagas/order/payment.rs @@ -2,8 +2,9 @@ use serde::{Deserialize, Serialize}; use sourced_rust::{digest, Entity}; /// Payment status -#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] pub enum PaymentStatus { + #[default] Pending, Authorized, Captured, @@ -11,12 +12,6 @@ pub enum PaymentStatus { Refunded, } -impl Default for PaymentStatus { - fn default() -> Self { - PaymentStatus::Pending - } -} - /// Payment aggregate - represents a payment attempt for an order #[derive(Default)] pub struct Payment { diff --git a/tests/sagas/order/saga.rs b/tests/sagas/order/saga.rs index 0035dd088..cd0d04388 100644 --- a/tests/sagas/order/saga.rs +++ b/tests/sagas/order/saga.rs @@ -1,11 +1,12 @@ use serde::{Deserialize, Serialize}; use sourced_rust::{digest, Entity}; -use super::order::OrderItem; +use super::OrderItem; /// Saga status - tracks the overall state machine -#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] pub enum SagaStatus { + #[default] Started, InventoryReserved, PaymentProcessed, @@ -14,12 +15,6 @@ pub enum SagaStatus { Failed, } -impl Default for SagaStatus { - fn default() -> Self { - SagaStatus::Started - } -} - /// Tracks what compensating actions are needed #[derive(Clone, Debug, Default, Serialize, Deserialize)] pub struct CompensationState { From 0cdab7b673b6cf7f4c7e592660ee4872e1f52f95 Mon Sep 17 00:00:00 2001 From: Patrick Lee Scott Date: Mon, 25 May 2026 15:11:31 -0500 Subject: [PATCH 4/4] Add updated_at to outbox table schema --- src/outbox/table.rs | 8 ++++++++ src/table/sql.rs | 4 ++++ 2 files changed, 12 insertions(+) diff --git a/src/outbox/table.rs b/src/outbox/table.rs index 63295365d..ade893856 100644 --- a/src/outbox/table.rs +++ b/src/outbox/table.rs @@ -25,6 +25,7 @@ pub fn outbox_message_schema() -> TableSchema { table_column("status", ColumnType::Text, false), table_column("created_at", ColumnType::Timestamp, false), table_column("next_available_at", ColumnType::Timestamp, false), + timestamp_column_with_default("updated_at", "CURRENT_TIMESTAMP"), table_column("claimed_by", ColumnType::Text, true), table_column("claimed_until", ColumnType::Timestamp, true), table_column("attempts", ColumnType::UnsignedInteger, false), @@ -151,6 +152,13 @@ fn table_column(name: &str, column_type: ColumnType, nullable: bool) -> TableCol column } +fn timestamp_column_with_default(name: &str, default: &str) -> TableColumn { + let mut column = table_column(name, ColumnType::Timestamp, false); + column.has_default = true; + column.default = Some(default.into()); + column +} + fn named_index(name: &str, columns: impl IntoIterator) -> TableIndex { let mut index = TableIndex::new(columns); index.name = Some(name.into()); diff --git a/src/table/sql.rs b/src/table/sql.rs index f9344adb5..4ad59b9c0 100644 --- a/src/table/sql.rs +++ b/src/table/sql.rs @@ -250,6 +250,8 @@ mod tests { .statements .iter() .any(|statement| statement.contains("\"created_at\" TEXT NOT NULL"))); + assert!(artifact.statements.iter().any(|statement| statement + .contains("\"updated_at\" TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP"))); } #[test] @@ -273,6 +275,8 @@ mod tests { .statements .iter() .any(|statement| statement.contains("\"claimed_until\" timestamptz"))); + assert!(artifact.statements.iter().any(|statement| statement + .contains("\"updated_at\" timestamptz NOT NULL DEFAULT CURRENT_TIMESTAMP"))); } #[test]