diff --git a/.gitignore b/.gitignore index f414b3b..c4e94ad 100644 --- a/.gitignore +++ b/.gitignore @@ -4,7 +4,6 @@ .conflict-files .conflict-side-*/ .DS_Store -.sqlx/ .stack-work/ .tmp/ .worktrees/ diff --git a/.sqlx/query-4d7bad65f4441d98a87a97cf251c6f71485fa5b5e22a38ef47f4cee6c9f88e84.json b/.sqlx/query-4d7bad65f4441d98a87a97cf251c6f71485fa5b5e22a38ef47f4cee6c9f88e84.json new file mode 100644 index 0000000..a6a30c2 --- /dev/null +++ b/.sqlx/query-4d7bad65f4441d98a87a97cf251c6f71485fa5b5e22a38ef47f4cee6c9f88e84.json @@ -0,0 +1,21 @@ +{ + "db_name": "SQLite", + "query": "\n SELECT MAX(sequence) AS \"sequence?: i64\"\n FROM events\n WHERE aggregate_type = ?1\n AND aggregate_id = ?2\n ", + "describe": { + "columns": [ + { + "name": "sequence?: i64", + "ordinal": 0, + "type_info": "Integer", + "origin": "Expression" + } + ], + "parameters": { + "Right": 2 + }, + "nullable": [ + true + ] + }, + "hash": "4d7bad65f4441d98a87a97cf251c6f71485fa5b5e22a38ef47f4cee6c9f88e84" +} diff --git a/.sqlx/query-587fa628a9fe4642ac0e22825a97e81d388af674f7251a9fb699f96901dcb710.json b/.sqlx/query-587fa628a9fe4642ac0e22825a97e81d388af674f7251a9fb699f96901dcb710.json deleted file mode 100644 index 529205f..0000000 --- a/.sqlx/query-587fa628a9fe4642ac0e22825a97e81d388af674f7251a9fb699f96901dcb710.json +++ /dev/null @@ -1,12 +0,0 @@ -{ - "db_name": "SQLite", - "query": "ROLLBACK", - "describe": { - "columns": [], - "parameters": { - "Right": 0 - }, - "nullable": [] - }, - "hash": "587fa628a9fe4642ac0e22825a97e81d388af674f7251a9fb699f96901dcb710" -} diff --git a/.sqlx/query-79663c1d3b43ceaf9ee728e501c64b57ae2e4d6cb36ff5c65eddcda1de27342f.json b/.sqlx/query-79663c1d3b43ceaf9ee728e501c64b57ae2e4d6cb36ff5c65eddcda1de27342f.json deleted file mode 100644 index 05b635e..0000000 --- a/.sqlx/query-79663c1d3b43ceaf9ee728e501c64b57ae2e4d6cb36ff5c65eddcda1de27342f.json +++ /dev/null @@ -1,12 +0,0 @@ -{ - "db_name": "SQLite", - "query": "COMMIT", - "describe": { - "columns": [], - "parameters": { - "Right": 0 - }, - "nullable": [] - }, - "hash": "79663c1d3b43ceaf9ee728e501c64b57ae2e4d6cb36ff5c65eddcda1de27342f" -} diff --git a/.sqlx/query-930a7770399087898ae6ac96ce5375048117486e06b21da4523d2c3c75113c32.json b/.sqlx/query-930a7770399087898ae6ac96ce5375048117486e06b21da4523d2c3c75113c32.json deleted file mode 100644 index 6db3f93..0000000 --- a/.sqlx/query-930a7770399087898ae6ac96ce5375048117486e06b21da4523d2c3c75113c32.json +++ /dev/null @@ -1,12 +0,0 @@ -{ - "db_name": "SQLite", - "query": "BEGIN IMMEDIATE", - "describe": { - "columns": [], - "parameters": { - "Right": 0 - }, - "nullable": [] - }, - "hash": "930a7770399087898ae6ac96ce5375048117486e06b21da4523d2c3c75113c32" -} diff --git a/Cargo.lock b/Cargo.lock index 76b9169..6fa516d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -93,9 +93,9 @@ dependencies = [ [[package]] name = "anyhow" -version = "1.0.103" +version = "1.0.102" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2a4385e2e34eb35d6b3efe798b9eb88096925d87726c0798709bf56d9ed84af3" +checksum = "7f202df86484c868dbad7eaa557ef785d5c66295e41b460ef922eca0723b842c" [[package]] name = "apalis-core" @@ -604,9 +604,14 @@ version = "0.4.0" dependencies = [ "cbindgen", "ciborium", + "cqrs-es", "event-sorcery", + "serde", + "serde_json", "sqlx", + "thiserror", "tokio", + "url", ] [[package]] @@ -1639,9 +1644,9 @@ dependencies = [ [[package]] name = "spin" -version = "0.9.9" +version = "0.9.8" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3763264f6b73151db08c50ff20d7d8a0b8796e021cdea7ceedad07b80155fa0e" +checksum = "6980e8d7511241f8acf4aebddbb1ff938df5eebe98691418c4468d0b72a96a67" dependencies = [ "lock_api", ] diff --git a/Cargo.toml b/Cargo.toml index cc57a23..7085531 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -55,3 +55,4 @@ sqlx = { version = "0.9.0", features = [ thiserror = "2.0.18" tokio = { version = "1.52.3", features = ["full"] } tracing = "0.1.44" +url = "2.5.8" diff --git a/crates/event-sorcery-ffi/Cargo.toml b/crates/event-sorcery-ffi/Cargo.toml index 4b2f996..60fe284 100644 --- a/crates/event-sorcery-ffi/Cargo.toml +++ b/crates/event-sorcery-ffi/Cargo.toml @@ -11,9 +11,14 @@ crate-type = ["rlib", "staticlib"] [dependencies] ciborium.workspace = true +cqrs-es.workspace = true event-sorcery.workspace = true +serde.workspace = true +serde_json.workspace = true sqlx.workspace = true +thiserror.workspace = true tokio.workspace = true +url.workspace = true [build-dependencies] cbindgen.workspace = true diff --git a/crates/event-sorcery-ffi/src/lib.rs b/crates/event-sorcery-ffi/src/lib.rs index 3a5260a..eaeaf48 100644 --- a/crates/event-sorcery-ffi/src/lib.rs +++ b/crates/event-sorcery-ffi/src/lib.rs @@ -1,12 +1,10 @@ -//! Unsafe C ABI for the captive event-sorcery engine. -//! -//! Inputs remain caller-owned for each call. Outputs remain engine-owned until -//! released with [`es_buf_free`]. The owner cell initialized by [`es_open`] must -//! remain at a stable address; [`es_close`] linearizes concurrent closes through -//! that cell, waits for acquired calls, and destroys the engine exactly once. +//! Stable C ABI over the shared event-sorcery engine. use std::collections::HashMap; -use std::io::Cursor; +use std::fmt; +use std::io::{self, Cursor, Write}; +use std::num::NonZeroUsize; +use std::num::TryFromIntError; use std::panic::{AssertUnwindSafe, catch_unwind}; use std::ptr; use std::str::FromStr; @@ -14,23 +12,159 @@ use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::{Arc, Condvar, Mutex, OnceLock}; use std::time::Duration; -use event_sorcery::Engine; +#[cfg(test)] +use cqrs_es::persist::SerializedEvent; +use serde::de::{DeserializeOwned, Error as DeserializeError, IgnoredAny, SeqAccess, Visitor}; +use serde::{Deserialize, Deserializer, Serialize, Serializer}; use sqlx::sqlite::{SqliteConnectOptions, SqliteJournalMode, SqlitePoolOptions, SqliteSynchronous}; +use url::form_urlencoded; + +#[cfg(test)] +use event_sorcery::CommitRequest; +use event_sorcery::{ + Engine, EngineError, OpaqueCommitRequest, OpaqueProposedEvent, OpaqueStoredEvent, + SqliteJobError, StreamIdentity, +}; const ABI_MAJOR: u32 = 0; -const ABI_MINOR: u32 = 1; +const ABI_MINOR: u32 = 2; const ES_OK: i32 = 0; const ES_ERR_DECODE: i32 = 1; +const ES_ERR_CONFLICT: i32 = 2; const ES_ERR_STORAGE: i32 = 4; const ES_ERR_STATE: i32 = 5; const ES_ERR_RESOURCE_LIMIT: i32 = 6; const ES_ERR_PANIC: i32 = 100; const MAX_REQUEST_BYTES: usize = 16 * 1024 * 1024; +const MAX_RESPONSE_BYTES: usize = 64 * 1024 * 1024; +/// Stored JSON can expand each opaque byte to four characters plus envelope +/// overhead before the exact CBOR response limit is applied. +const MAX_STORED_EVENT_PAGE_BYTES: usize = 4 * MAX_RESPONSE_BYTES + MAX_REQUEST_BYTES; const MAX_CBOR_DEPTH: usize = 32; +const MAX_PAYLOAD_BYTES: usize = 1024 * 1024; +const MAX_COMMIT_EVENTS: usize = 1024; +const MAX_LIST_ITEMS: usize = 4096; +const MAX_ERROR_TEXT_BYTES: usize = 4 * 1024; /// Maximum worker threads accepted from the fixed-width C ABI options product. const MAX_RUNTIME_THREADS: u32 = 256; static STORE_REGISTRY: OnceLock>> = OnceLock::new(); +type ProposedEventWire = (String, String, OpaqueBytes); +type CommitWire = (u8, String, String, u64, CommitEvents); +type StoredEventWire = (u64, String, String, OpaqueBytes); +#[cfg(test)] +type StoredEventsWire = (u8, Vec); + +#[derive(Debug, PartialEq, Eq)] +struct OpaqueBytes(Vec); + +#[derive(Debug)] +struct CommitEvents { + values: Vec, + observed: usize, +} + +impl Serialize for OpaqueBytes { + fn serialize( + &self, + serializer: SerializerType, + ) -> Result + where + SerializerType: Serializer, + { + let Self(bytes) = self; + serializer.serialize_bytes(bytes) + } +} + +impl Serialize for CommitEvents { + fn serialize( + &self, + serializer: SerializerType, + ) -> Result + where + SerializerType: Serializer, + { + self.values.serialize(serializer) + } +} + +impl<'de> Deserialize<'de> for CommitEvents { + fn deserialize( + deserializer: DeserializerType, + ) -> Result + where + DeserializerType: Deserializer<'de>, + { + deserializer.deserialize_seq(CommitEventsVisitor) + } +} + +struct CommitEventsVisitor; + +impl<'de> Visitor<'de> for CommitEventsVisitor { + type Value = CommitEvents; + + fn expecting(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter.write_str("a bounded sequence of proposed events") + } + + fn visit_seq(self, mut sequence: Sequence) -> Result + where + Sequence: SeqAccess<'de>, + { + let capacity = sequence.size_hint().unwrap_or(0).min(MAX_COMMIT_EVENTS); + let mut values = Vec::with_capacity(capacity); + while values.len() < MAX_COMMIT_EVENTS { + let Some(event) = sequence.next_element()? else { + return Ok(CommitEvents { + observed: values.len(), + values, + }); + }; + values.push(event); + } + + let observed = if sequence.next_element::()?.is_some() { + values + .len() + .checked_add(1) + .ok_or_else(|| Sequence::Error::custom("commit event count overflow"))? + } else { + values.len() + }; + + Ok(CommitEvents { values, observed }) + } +} + +impl<'de> Deserialize<'de> for OpaqueBytes { + fn deserialize(deserializer: D) -> Result + where + D: Deserializer<'de>, + { + deserializer.deserialize_byte_buf(OpaqueBytesVisitor) + } +} + +struct OpaqueBytesVisitor; + +impl Visitor<'_> for OpaqueBytesVisitor { + type Value = OpaqueBytes; + + fn expecting(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter.write_str("a CBOR byte string") + } + + fn visit_bytes(self, value: &[u8]) -> Result { + Ok(OpaqueBytes(value.to_vec())) + } + + fn visit_byte_buf(self, value: Vec) -> Result { + Ok(OpaqueBytes(value)) + } +} + #[repr(C)] #[derive(Clone, Copy)] pub struct EsBuf { @@ -43,6 +177,8 @@ pub struct EsStore { state_changed: Condvar, inner: Mutex>>, poisoned: AtomicBool, + #[cfg(test)] + panic_on_close: AtomicBool, } #[unsafe(no_mangle)] @@ -55,29 +191,32 @@ pub const extern "C" fn es_abi_version() -> u32 { /// /// # Safety /// -/// `options` must reference readable bytes for the duration of the call. +/// `options` must reference a readable buffer for the duration of the call. /// `out_store` must be a stable, unowned writable cell that remains at the same /// address until close completes. `out_error` must be null or writable. +/// `out_store` and `out_error` must be distinct. The options buffer may alias +/// `out_error`; options are decoded before either output is written. pub unsafe extern "C" fn es_open( options: *const EsBuf, out_store: *mut *mut EsStore, out_error: *mut EsBuf, ) -> i32 { - if !out_store.is_null() { - unsafe { out_store.write(ptr::null_mut()) }; - } ffi_call(None, out_error, || { + let options = decode_open_options(options); + if !out_store.is_null() { + unsafe { out_store.write(ptr::null_mut()) }; + } if out_store.is_null() { return Err(AbiError::State("out_store is null")); } - let options = decode_open_options(options)?; + let options = options?; let runtime = tokio::runtime::Builder::new_multi_thread() .worker_threads(options.runtime_threads) .enable_all() .build() - .map_err(|_| AbiError::State("runtime initialization failed"))?; - let connect = SqliteConnectOptions::from_str(&options.path) - .map_err(|_| AbiError::MalformedInput)? + .map_err(AbiError::Runtime)?; + let connect = options + .connect_options .create_if_missing(true) .journal_mode(SqliteJournalMode::Wal) .synchronous(SqliteSynchronous::Full) @@ -88,27 +227,205 @@ pub unsafe extern "C" fn es_open( .max_connections(options.pool_size) .connect_with(connect), ) - .map_err(|_| AbiError::Storage)?; + .map_err(AbiError::Sql)?; let engine = Engine::new(pool); runtime .block_on(engine.migrate()) - .map_err(|_| AbiError::Storage)?; + .map_err(AbiError::Migration)?; let store = Arc::new(EsStore { state: Mutex::new(StoreState::Open { active_calls: 0 }), state_changed: Condvar::new(), - inner: Mutex::new(Some(Arc::new(StoreInner { - _runtime: runtime, - _engine: engine, - }))), + inner: Mutex::new(Some(Arc::new(StoreInner { runtime, engine }))), poisoned: AtomicBool::new(false), + #[cfg(test)] + panic_on_close: AtomicBool::new(false), }); publish_store(out_store, store) }) } #[unsafe(no_mangle)] -/// Closes a store returned by [`es_open`]. A null owner or null handle is a -/// no-op. +/// Loads a stream through the shared engine. +/// +/// # Safety +/// +/// `store` must be the original owner cell passed to [`es_open`]. `request` +/// must reference a readable buffer. `out_events` must be writable, and +/// `out_error` must be null or writable. The request buffer may alias either +/// output. `store`, `out_events`, and `out_error` must all be distinct. +pub unsafe extern "C" fn es_load_stream( + store: *mut *mut EsStore, + request: *const EsBuf, + out_events: *mut EsBuf, + out_error: *mut EsBuf, +) -> i32 { + let request: Result<(u8, String, String, Option), AbiError> = decode(request); + clear_buffer_output(out_events); + let lease = match EsStore::acquire(store) { + Ok(lease) => lease, + Err(error) => return write_error(out_error, error), + }; + ffi_call(Some(&lease.store), out_error, || { + if out_events.is_null() { + return Err(AbiError::State("out_events is null")); + } + let (version, aggregate_type, aggregate_id, after) = request?; + require_version(version)?; + let after = after + .map(usize::try_from) + .transpose() + .map_err(AbiError::InputInteger)?; + let stream = StreamIdentity::new(aggregate_type, aggregate_id); + let query_limit = NonZeroUsize::new(MAX_LIST_ITEMS + 1) + .ok_or(AbiError::State("invalid list query limit"))?; + let events = lease + .inner + .runtime + .block_on( + lease.inner.engine.load_opaque_events_page_bounded( + &stream, + after, + query_limit, + NonZeroUsize::new(MAX_STORED_EVENT_PAGE_BYTES) + .ok_or(AbiError::State("invalid stored page byte limit"))?, + ), + ) + .map_err(AbiError::from)?; + if events.len() > MAX_LIST_ITEMS { + return Err(AbiError::ResourceLimit { + resource: "list_items", + observed: events.len(), + limit: MAX_LIST_ITEMS, + }); + } + unsafe { out_events.write(encode_event_page(events)?) }; + Ok(()) + }) +} + +#[unsafe(no_mangle)] +/// Reads the current stream version, where zero means no events. +/// +/// # Safety +/// +/// `store` must be the original owner cell passed to [`es_open`]. `request` +/// must reference a readable buffer. `out_version` must be writable, and +/// `out_error` must be null or writable. The request storage may alias either +/// output. `store`, `out_version`, and `out_error` must all be distinct. +pub unsafe extern "C" fn es_current_version( + store: *mut *mut EsStore, + request: *const EsBuf, + out_version: *mut u64, + out_error: *mut EsBuf, +) -> i32 { + let request: Result<(u8, String, String), AbiError> = decode(request); + if !out_version.is_null() { + unsafe { out_version.write(0) }; + } + let lease = match EsStore::acquire(store) { + Ok(lease) => lease, + Err(error) => return write_error(out_error, error), + }; + ffi_call(Some(&lease.store), out_error, || { + if out_version.is_null() { + return Err(AbiError::State("out_version is null")); + } + let (version, aggregate_type, aggregate_id) = request?; + require_version(version)?; + let stream = StreamIdentity::new(aggregate_type, aggregate_id); + let version = lease + .inner + .runtime + .block_on(lease.inner.engine.current_version(&stream)) + .map_err(AbiError::from)?; + let version = u64::try_from(version).map_err(AbiError::StorageInteger)?; + unsafe { out_version.write(version) }; + Ok(()) + }) +} + +#[unsafe(no_mangle)] +/// Atomically appends events at the requested expected version. +/// +/// # Safety +/// +/// `store` must be the original owner cell passed to [`es_open`]. `request` +/// must reference a readable buffer. `out_error` must be null or writable. The +/// owner cell and `out_error` must be distinct. The request buffer may alias +/// `out_error`; it is decoded before the output is written. +pub unsafe extern "C" fn es_commit( + store: *mut *mut EsStore, + request: *const EsBuf, + out_error: *mut EsBuf, +) -> i32 { + let lease = match EsStore::acquire(store) { + Ok(lease) => lease, + Err(error) => return write_error(out_error, error), + }; + ffi_call(Some(&lease.store), out_error, || { + let (version, aggregate_type, aggregate_id, expected, events): CommitWire = + decode_validated(request, |wire: &CommitWire| { + if wire.4.observed <= MAX_COMMIT_EVENTS { + return Ok(()); + } + Err(AbiError::ResourceLimit { + resource: "commit_events", + observed: wire.4.observed, + limit: MAX_COMMIT_EVENTS, + }) + })?; + require_version(version)?; + require_error_text_limit(aggregate_type.len())?; + require_error_text_limit(aggregate_id.len())?; + let CommitEvents { + values: events, + observed: _, + } = events; + if events.is_empty() { + return Err(AbiError::MalformedInput); + } + let expected_sequence = usize::try_from(expected).map_err(AbiError::InputInteger)?; + expected_sequence + .checked_add(events.len()) + .ok_or(AbiError::MalformedInput)?; + let proposed = events + .into_iter() + .map(|(event_type, event_version, OpaqueBytes(payload))| { + require_payload_limit(payload.len())?; + Ok(OpaqueProposedEvent::new(event_type, event_version, payload)) + }) + .collect::, AbiError>>()?; + let stream = StreamIdentity::new(aggregate_type.clone(), aggregate_id.clone()); + match lease + .inner + .runtime + .block_on(lease.inner.engine.commit_opaque(OpaqueCommitRequest::new( + stream.clone(), + expected_sequence, + &proposed, + ))) { + Ok(()) => Ok(()), + Err(EngineError::OptimisticLock) => { + let actual = lease + .inner + .runtime + .block_on(lease.inner.engine.current_version(&stream)) + .map_err(AbiError::from)?; + let actual = u64::try_from(actual).map_err(AbiError::StorageInteger)?; + Err(AbiError::Conflict { + aggregate_type, + aggregate_id, + expected, + actual, + }) + } + Err(error) => Err(AbiError::from(error)), + } + }) +} + +#[unsafe(no_mangle)] +/// Closes a store returned by [`es_open`]. A null owner or handle is a no-op. /// /// # Safety /// @@ -132,8 +449,7 @@ pub unsafe extern "C" fn es_close(store: *mut *mut EsStore) -> i32 { } #[unsafe(no_mangle)] -/// Releases an engine-owned output buffer. A null owner or null buffer is a -/// no-op. +/// Releases an engine-owned output buffer. A null owner or buffer is a no-op. /// /// # Safety /// @@ -157,15 +473,26 @@ pub unsafe extern "C" fn es_buf_free(buffer: *mut EsBuf) { } struct OpenOptions { - path: String, + connect_options: SqliteConnectOptions, busy_timeout: Duration, pool_size: u32, runtime_threads: usize, } +struct ParsedSqliteOptions { + connect_options: SqliteConnectOptions, + memory_cache: Option, +} + +#[derive(Clone, Copy, Eq, PartialEq)] +enum SqliteCacheMode { + Shared, + Private, +} + struct StoreInner { - _runtime: tokio::runtime::Runtime, - _engine: Engine, + runtime: tokio::runtime::Runtime, + engine: Engine, } struct StoreEntry { @@ -173,16 +500,15 @@ struct StoreEntry { store: Arc, } -#[cfg(test)] struct StoreLease { store: Arc, - _inner: Arc, + inner: Arc, } enum StoreState { Open { active_calls: usize }, Closing { active_calls: usize }, - Closed(Result<(), AbiError>), + Closed(CloseOutcome), } #[derive(Clone, Copy)] @@ -191,6 +517,57 @@ enum CloseRole { Join, } +#[derive(Clone, Copy)] +enum CloseOutcome { + Ok, + Panic, +} + +#[derive(Debug, thiserror::Error)] +enum AbiError { + #[error("malformed input")] + MalformedInput, + #[error("malformed CBOR input")] + CborDecode(#[source] ciborium::de::Error), + #[error("malformed SQLite options")] + SqliteOptions(#[source] sqlx::Error), + #[error("malformed integer input")] + InputInteger(#[source] TryFromIntError), + #[error("non-canonical CBOR input")] + CanonicalCbor(#[source] ciborium::ser::Error), + #[error("optimistic concurrency conflict")] + Conflict { + aggregate_type: String, + aggregate_id: String, + expected: u64, + actual: u64, + }, + #[error("engine storage failure")] + Engine(#[source] EngineError), + #[error("SQLite storage failure")] + Sql(#[source] sqlx::Error), + #[error("SQLite migration failure")] + Migration(#[source] SqliteJobError), + #[error("integer storage failure")] + StorageInteger(#[source] TryFromIntError), + #[error("response CBOR encoding failure")] + CborEncode(#[source] ciborium::ser::Error), + #[error("response I/O failure")] + ResponseIo(#[source] io::Error), + #[error("runtime initialization failure")] + Runtime(#[source] io::Error), + #[error("{0}")] + State(&'static str), + #[error("resource {resource} observed {observed} above limit {limit}")] + ResourceLimit { + resource: &'static str, + observed: usize, + limit: usize, + }, + #[error("engine panic")] + Panic, +} + fn store_registry() -> &'static Mutex> { STORE_REGISTRY.get_or_init(|| Mutex::new(HashMap::new())) } @@ -269,21 +646,7 @@ fn retire_store(owner_address: usize, store: &Arc) { } } -#[derive(Clone, Copy)] -enum AbiError { - MalformedInput, - Storage, - State(&'static str), - ResourceLimit { - resource: &'static str, - observed: usize, - limit: usize, - }, - Panic, -} - impl EsStore { - #[cfg(test)] fn acquire(owner: *mut *mut Self) -> Result { if owner.is_null() { return Err(AbiError::State("store owner is null")); @@ -319,10 +682,7 @@ impl EsStore { .ok_or(AbiError::State("active call count overflow"))?; drop(state); drop(registry); - Ok(StoreLease { - store, - _inner: inner, - }) + Ok(StoreLease { store, inner }) } fn close(&self, role: CloseRole) -> Result<(), AbiError> { @@ -333,7 +693,8 @@ impl EsStore { .unwrap_or_else(std::sync::PoisonError::into_inner); loop { match &*state { - StoreState::Closed(result) => return *result, + StoreState::Closed(CloseOutcome::Ok) => return Ok(()), + StoreState::Closed(CloseOutcome::Panic) => return Err(AbiError::Panic), StoreState::Open { .. } | StoreState::Closing { .. } => { state = self .state_changed @@ -361,38 +722,32 @@ impl EsStore { .lock() .unwrap_or_else(std::sync::PoisonError::into_inner) .take(); - let result = catch_unwind(AssertUnwindSafe(|| drop(inner))).map_err(|_| AbiError::Panic); + let result = catch_unwind(AssertUnwindSafe(|| { + #[cfg(test)] + assert!( + !self.panic_on_close.load(Ordering::Acquire), + "forced store teardown panic" + ); + + drop(inner); + })) + .map_err(|_| AbiError::Panic); + let outcome = if result.is_ok() { + CloseOutcome::Ok + } else { + CloseOutcome::Panic + }; let mut state = self .state .lock() .unwrap_or_else(std::sync::PoisonError::into_inner); - *state = StoreState::Closed(result); + *state = StoreState::Closed(outcome); drop(state); self.state_changed.notify_all(); result } } -#[cfg(test)] -impl StoreLease { - fn wait_until_closing(&self) { - let mut state = self - .store - .state - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner); - while matches!(&*state, StoreState::Open { .. }) { - state = self - .store - .state_changed - .wait(state) - .unwrap_or_else(std::sync::PoisonError::into_inner); - } - drop(state); - } -} - -#[cfg(test)] impl Drop for StoreLease { fn drop(&mut self) { let mut state = self @@ -422,19 +777,67 @@ impl Drop for StoreLease { } } +#[cfg(test)] +impl StoreLease { + fn wait_until_closing(&self) { + let mut state = self + .store + .state + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + while matches!(&*state, StoreState::Open { .. }) { + state = self + .store + .state_changed + .wait(state) + .unwrap_or_else(std::sync::PoisonError::into_inner); + } + drop(state); + } +} + +impl From for AbiError { + fn from(error: EngineError) -> Self { + match error { + EngineError::EventPageTooLarge { observed, limit } => Self::ResourceLimit { + resource: "stored_event_page", + observed, + limit, + }, + other => Self::Engine(other), + } + } +} + impl AbiError { const fn code(&self) -> i32 { match self { - Self::MalformedInput => ES_ERR_DECODE, - Self::Storage => ES_ERR_STORAGE, - Self::State(_) => ES_ERR_STATE, + Self::MalformedInput + | Self::CborDecode(_) + | Self::SqliteOptions(_) + | Self::InputInteger(_) + | Self::CanonicalCbor(_) => ES_ERR_DECODE, + Self::Conflict { .. } => ES_ERR_CONFLICT, + Self::Engine(_) | Self::Sql(_) | Self::Migration(_) | Self::StorageInteger(_) => { + ES_ERR_STORAGE + } + Self::CborEncode(_) | Self::ResponseIo(_) | Self::Runtime(_) | Self::State(_) => { + ES_ERR_STATE + } Self::ResourceLimit { .. } => ES_ERR_RESOURCE_LIMIT, Self::Panic => ES_ERR_PANIC, } } } -fn decode_open_options(buffer: *const EsBuf) -> Result { +fn decode(buffer: *const EsBuf) -> Result { + decode_validated(buffer, |_| Ok(())) +} + +fn decode_validated( + buffer: *const EsBuf, + validate: impl FnOnce(&T) -> Result<(), AbiError>, +) -> Result { let Some(buffer) = (unsafe { buffer.as_ref() }) else { return Err(AbiError::MalformedInput); }; @@ -449,26 +852,177 @@ fn decode_open_options(buffer: *const EsBuf) -> Result { }); } let bytes = unsafe { std::slice::from_raw_parts(buffer.ptr, buffer.len) }; - let decoded: (u8, String, u64, u32, u32) = + let decoded = ciborium::de::from_reader_with_recursion_limit(Cursor::new(bytes), MAX_CBOR_DEPTH) - .map_err(|_| AbiError::MalformedInput)?; + .map_err(AbiError::CborDecode)?; + validate(&decoded)?; let mut canonical = Vec::new(); - ciborium::into_writer(&decoded, &mut canonical).map_err(|_| AbiError::MalformedInput)?; + ciborium::into_writer(&decoded, &mut canonical).map_err(AbiError::CanonicalCbor)?; if canonical != bytes { return Err(AbiError::MalformedInput); } + Ok(decoded) +} + +struct BoundedResponseWriter { + bytes: Vec, + overflow_observed: Option, +} + +impl BoundedResponseWriter { + const fn new() -> Self { + Self { + bytes: Vec::new(), + overflow_observed: None, + } + } + + fn into_buffer(self) -> EsBuf { + owned_buffer(self.bytes) + } + + fn io_error(&self, source: io::Error) -> AbiError { + let Some(observed) = self.overflow_observed else { + return AbiError::ResponseIo(source); + }; + AbiError::ResourceLimit { + resource: "encoded_response_buffer", + observed, + limit: MAX_RESPONSE_BYTES, + } + } + + fn cbor_error(&self, source: ciborium::ser::Error) -> AbiError { + let Some(observed) = self.overflow_observed else { + return AbiError::CborEncode(source); + }; + AbiError::ResourceLimit { + resource: "encoded_response_buffer", + observed, + limit: MAX_RESPONSE_BYTES, + } + } +} + +impl Write for BoundedResponseWriter { + fn write(&mut self, buffer: &[u8]) -> io::Result { + let observed = self + .bytes + .len() + .checked_add(buffer.len()) + .ok_or_else(|| io::Error::other("response size overflow"))?; + if observed > MAX_RESPONSE_BYTES { + self.overflow_observed = Some(observed); + return Err(io::Error::other("response byte limit exceeded")); + } + self.bytes.extend_from_slice(buffer); + Ok(buffer.len()) + } + + fn flush(&mut self) -> io::Result<()> { + Ok(()) + } +} + +fn encode_event_page(events: Vec) -> Result { + let mut writer = BoundedResponseWriter::new(); + writer + .write_all(&[0x82, 0x01]) + .map_err(|source| writer.io_error(source))?; + write_array_header(&mut writer, events.len())?; + for event in events { + let wire = event_to_wire(event)?; + if let Err(source) = ciborium::into_writer(&wire, &mut writer) { + return Err(writer.cbor_error(source)); + } + } + Ok(writer.into_buffer()) +} + +fn write_array_header(writer: &mut BoundedResponseWriter, length: usize) -> Result<(), AbiError> { + let encoded = match length { + 0..=23 => vec![0x80 | u8::try_from(length).map_err(AbiError::StorageInteger)?], + 24..=255 => vec![ + 0x98, + u8::try_from(length).map_err(AbiError::StorageInteger)?, + ], + 256..=65_535 => { + let length = u16::try_from(length).map_err(AbiError::StorageInteger)?; + let mut encoded = vec![0x99]; + encoded.extend_from_slice(&length.to_be_bytes()); + encoded + } + _ => return Err(AbiError::State("event page length exceeds ABI limit")), + }; + writer + .write_all(&encoded) + .map_err(|source| writer.io_error(source)) +} + +fn owned_buffer(bytes: Vec) -> EsBuf { + let boxed = bytes.into_boxed_slice(); + EsBuf { + len: boxed.len(), + ptr: Box::into_raw(boxed).cast::(), + } +} + +const fn require_version(version: u8) -> Result<(), AbiError> { + if version == 1 { + Ok(()) + } else { + Err(AbiError::MalformedInput) + } +} + +fn event_to_wire(event: OpaqueStoredEvent) -> Result { + let (sequence, event_type, event_version, payload) = event.into_parts(); + require_payload_limit(payload.len())?; + Ok(( + u64::try_from(sequence).map_err(AbiError::StorageInteger)?, + event_type, + event_version, + OpaqueBytes(payload), + )) +} + +const fn require_payload_limit(observed: usize) -> Result<(), AbiError> { + if observed <= MAX_PAYLOAD_BYTES { + return Ok(()); + } + Err(AbiError::ResourceLimit { + resource: "opaque_domain_payload", + observed, + limit: MAX_PAYLOAD_BYTES, + }) +} + +const fn require_error_text_limit(observed: usize) -> Result<(), AbiError> { + if observed <= MAX_ERROR_TEXT_BYTES { + return Ok(()); + } + Err(AbiError::ResourceLimit { + resource: "error_detail_text", + observed, + limit: MAX_ERROR_TEXT_BYTES, + }) +} + +fn decode_open_options(buffer: *const EsBuf) -> Result { let (version, path, busy_timeout_ms, pool_size, runtime_threads): (u8, String, u64, u32, u32) = - decoded; - if version != 1 - || pool_size == 0 - || runtime_threads == 0 - || (path == "sqlite::memory:" && pool_size != 1) - { + decode(buffer)?; + if version != 1 || pool_size == 0 || runtime_threads == 0 { + return Err(AbiError::MalformedInput); + } + + let parsed_sqlite = parse_sqlite_options(&path)?; + if parsed_sqlite.memory_cache == Some(SqliteCacheMode::Private) { return Err(AbiError::MalformedInput); } - let runtime_threads = usize::try_from(runtime_threads).map_err(|_| AbiError::MalformedInput)?; + + let runtime_threads = usize::try_from(runtime_threads).map_err(AbiError::InputInteger)?; let runtime_thread_limit = - usize::try_from(MAX_RUNTIME_THREADS).map_err(|_| AbiError::MalformedInput)?; + usize::try_from(MAX_RUNTIME_THREADS).map_err(AbiError::InputInteger)?; if runtime_threads > runtime_thread_limit { return Err(AbiError::ResourceLimit { resource: "runtime_threads", @@ -477,24 +1031,56 @@ fn decode_open_options(buffer: *const EsBuf) -> Result { }); } Ok(OpenOptions { - path, + connect_options: parsed_sqlite.connect_options, busy_timeout: Duration::from_millis(busy_timeout_ms), pool_size, runtime_threads, }) } +fn parse_sqlite_options(path: &str) -> Result { + let connect_options = SqliteConnectOptions::from_str(path).map_err(AbiError::SqliteOptions)?; + let path = path + .trim_start_matches("sqlite://") + .trim_start_matches("sqlite:"); + let mut database_and_parameters = path.splitn(2, '?'); + let database = database_and_parameters.next().unwrap_or_default(); + let parameters = database_and_parameters.next().unwrap_or_default(); + let database_is_memory = database == ":memory:"; + let initial_cache = if database_is_memory { + SqliteCacheMode::Shared + } else { + SqliteCacheMode::Private + }; + let (in_memory, cache_mode) = form_urlencoded::parse(parameters.as_bytes()).fold( + (database_is_memory, initial_cache), + |(in_memory, cache_mode), (key, value)| match (key.as_ref(), value.as_ref()) { + ("mode", "memory") => (true, SqliteCacheMode::Shared), + ("cache", "private") => (in_memory, SqliteCacheMode::Private), + ("cache", "shared") => (in_memory, SqliteCacheMode::Shared), + _ => (in_memory, cache_mode), + }, + ); + + Ok(ParsedSqliteOptions { + connect_options, + memory_cache: in_memory.then_some(cache_mode), + }) +} + fn ffi_call( store: Option<&EsStore>, out_error: *mut EsBuf, call: impl FnOnce() -> Result<(), AbiError>, ) -> i32 { - clear_buffer_output(out_error); if store.is_some_and(|store| store.poisoned.load(Ordering::Acquire)) { return write_error(out_error, AbiError::State("store is poisoned")); } match catch_unwind(AssertUnwindSafe(call)) { - Ok(Ok(())) => ES_OK, + Ok(Ok(())) => { + clear_buffer_output(out_error); + ES_OK + } Ok(Err(error)) => write_error(out_error, error), Err(_) => { if let Some(store) = store { @@ -506,10 +1092,33 @@ fn ffi_call( } fn write_error(out_error: *mut EsBuf, error: AbiError) -> i32 { + clear_buffer_output(out_error); let class = error.code(); let detail = match error { - AbiError::MalformedInput => ciborium::Value::Text("malformed input".to_string()), - AbiError::Storage => ciborium::Value::Text("storage failure".to_string()), + AbiError::MalformedInput + | AbiError::CborDecode(_) + | AbiError::SqliteOptions(_) + | AbiError::InputInteger(_) + | AbiError::CanonicalCbor(_) => ciborium::Value::Text("malformed input".to_string()), + AbiError::Conflict { + aggregate_type, + aggregate_id, + expected, + actual, + } => ciborium::Value::Array(vec![ + ciborium::Value::Text(aggregate_type), + ciborium::Value::Text(aggregate_id), + ciborium::Value::Integer(expected.into()), + ciborium::Value::Integer(actual.into()), + ]), + AbiError::Engine(_) + | AbiError::Sql(_) + | AbiError::Migration(_) + | AbiError::StorageInteger(_) => ciborium::Value::Text("storage failure".to_string()), + AbiError::CborEncode(_) | AbiError::ResponseIo(_) => { + ciborium::Value::Text("response encoding failed".to_string()) + } + AbiError::Runtime(_) => ciborium::Value::Text("runtime initialization failed".to_string()), AbiError::State(detail) => ciborium::Value::Text(detail.to_string()), AbiError::ResourceLimit { resource, @@ -524,13 +1133,10 @@ fn write_error(out_error: *mut EsBuf, error: AbiError) -> i32 { }; if !out_error.is_null() { let mut bytes = Vec::new(); - if ciborium::into_writer(&(1_u8, class, detail), &mut bytes).is_ok() { - let boxed = bytes.into_boxed_slice(); - let buffer = EsBuf { - len: boxed.len(), - ptr: Box::into_raw(boxed).cast::(), - }; - unsafe { out_error.write(buffer) }; + if ciborium::into_writer(&(1_u8, class, detail), &mut bytes).is_ok() + && bytes.len() <= MAX_RESPONSE_BYTES + { + unsafe { out_error.write(owned_buffer(bytes)) }; } } class @@ -560,69 +1166,124 @@ mod tests { } } - fn encoded_open_options(runtime_threads: u32) -> Vec { + fn encode_request(value: &impl Serialize) -> Vec { let mut encoded = Vec::new(); - ciborium::into_writer( - &(1_u8, "sqlite::memory:", 5_000_u64, 1_u32, runtime_threads), - &mut encoded, - ) - .unwrap(); + ciborium::into_writer(value, &mut encoded).unwrap(); encoded } - fn open_options() -> Vec { - encoded_open_options(1) + fn caller_buffer(bytes: &mut [u8]) -> EsBuf { + EsBuf { + ptr: bytes.as_mut_ptr(), + len: bytes.len(), + } } - fn decode_error(buffer: EsBuf) -> (u8, i32, Value) { + fn open_store(store: &mut *mut EsStore) { + let mut encoded = encode_request(&(1_u8, "sqlite::memory:", 5_000_u64, 1_u32, 1_u32)); + let mut error = empty_buffer(); + let options = caller_buffer(&mut encoded); + + assert_eq!( + unsafe { + es_open( + &raw const options, + std::ptr::from_mut(store), + &raw mut error, + ) + }, + ES_OK + ); + assert!(!(*store).is_null()); + assert!(error.ptr.is_null()); + } + + fn decode_error(buffer: &EsBuf) -> (u8, i32, Value) { let bytes = unsafe { std::slice::from_raw_parts(buffer.ptr, buffer.len) }; ciborium::from_reader(Cursor::new(bytes)).unwrap() } - #[test] - fn opens_migrates_and_closes_an_in_memory_store() { - let mut encoded = open_options(); - let mut store = ptr::null_mut(); + fn assert_stream_version( + store: &mut *mut EsStore, + aggregate_type: &str, + aggregate_id: &str, + expected: u64, + ) { + let mut request = encode_request(&(1_u8, aggregate_type, aggregate_id)); + let request_buffer = caller_buffer(&mut request); + let mut actual = u64::MAX; let mut error = empty_buffer(); - let options = EsBuf { - ptr: encoded.as_mut_ptr(), - len: encoded.len(), - }; assert_eq!( - unsafe { es_open(&raw const options, &raw mut store, &raw mut error) }, + unsafe { + es_current_version( + store, + &raw const request_buffer, + &raw mut actual, + &raw mut error, + ) + }, ES_OK ); - assert!(!store.is_null()); - assert_eq!(unsafe { es_close(&raw mut store) }, ES_OK); - assert!(store.is_null()); - assert!(error.ptr.is_null()); + assert_eq!(actual, expected); + unsafe { es_buf_free(&raw mut error) }; } - #[test] - fn open_clears_the_store_output_before_decoding() { - let mut malformed = [0xff]; - let mut store = std::ptr::NonNull::::dangling().as_ptr(); + fn assert_stream_text_limit_rejected( + store: &mut *mut EsStore, + aggregate_type: &str, + aggregate_id: &str, + ) { + let mut request = encode_request(&( + 1_u8, + aggregate_type, + aggregate_id, + 0_u64, + vec![("Created", "1.0", OpaqueBytes(Vec::new()))], + )); + let request_buffer = caller_buffer(&mut request); let mut error = empty_buffer(); - let options = EsBuf { - ptr: malformed.as_mut_ptr(), - len: malformed.len(), - }; - let result = unsafe { es_open(&raw const options, &raw mut store, &raw mut error) }; + assert_eq!( + unsafe { es_commit(store, &raw const request_buffer, &raw mut error) }, + ES_ERR_RESOURCE_LIMIT + ); + assert_eq!( + decode_error(&error), + ( + 1, + ES_ERR_RESOURCE_LIMIT, + Value::Array(vec![ + Value::Text("error_detail_text".to_string()), + Value::Integer((MAX_ERROR_TEXT_BYTES + 1).into()), + Value::Integer(MAX_ERROR_TEXT_BYTES.into()), + ]), + ) + ); + unsafe { es_buf_free(&raw mut error) }; + assert_stream_version(store, aggregate_type, aggregate_id, 0); + } - assert_eq!(result, ES_ERR_DECODE); + #[test] + fn opens_migrates_and_closes_an_in_memory_store() { + let mut store = ptr::null_mut(); + open_store(&mut store); + + assert_eq!(unsafe { es_close(&raw mut store) }, ES_OK); assert!(store.is_null()); - unsafe { es_buf_free(&raw mut error) }; + assert_eq!(unsafe { es_close(&raw mut store) }, ES_OK); } #[test] fn open_options_accept_the_maximum_runtime_thread_count() { - let mut encoded = encoded_open_options(MAX_RUNTIME_THREADS); - let buffer = EsBuf { - ptr: encoded.as_mut_ptr(), - len: encoded.len(), - }; + let mut encoded = encode_request(&( + 1_u8, + "sqlite::memory:", + 5_000_u64, + 1_u32, + MAX_RUNTIME_THREADS, + )); + let buffer = caller_buffer(&mut encoded); let Ok(options) = decode_open_options(&raw const buffer) else { panic!("maximum runtime thread count must be accepted"); @@ -636,11 +1297,14 @@ mod tests { #[test] fn open_options_reject_the_first_runtime_thread_count_above_the_limit() { - let mut encoded = encoded_open_options(MAX_RUNTIME_THREADS + 1); - let buffer = EsBuf { - ptr: encoded.as_mut_ptr(), - len: encoded.len(), - }; + let mut encoded = encode_request(&( + 1_u8, + "sqlite::memory:", + 5_000_u64, + 1_u32, + MAX_RUNTIME_THREADS + 1, + )); + let buffer = caller_buffer(&mut encoded); let result = decode_open_options(&raw const buffer); @@ -656,41 +1320,454 @@ mod tests { } #[test] - fn successful_open_clears_a_stale_error_output() { - let mut encoded = open_options(); + fn open_options_reject_each_invalid_field() { + fn assert_malformed(options: (u8, &'static str, u64, u32, u32)) { + let mut encoded = encode_request(&options); + let buffer = caller_buffer(&mut encoded); + + assert!(matches!( + decode_open_options(&raw const buffer), + Err(AbiError::MalformedInput) + )); + } + + assert_malformed((2, "sqlite::memory:", 5_000, 1, 1)); + assert_malformed((1, "sqlite::memory:", 5_000, 0, 1)); + assert_malformed((1, "sqlite::memory:", 5_000, 1, 0)); + } + + #[test] + fn open_options_accept_supported_shared_cache_in_memory_urls() { + for path in [ + "sqlite::memory:", + "sqlite://?mode=memory", + "sqlite://:memory:", + "sqlite://?mode=memory&cache=shared", + "sqlite://?cache=private&mode=memory", + ] { + let mut encoded = encode_request(&(1_u8, path, 5_000_u64, 2_u32, 1_u32)); + let buffer = caller_buffer(&mut encoded); + + assert!( + decode_open_options(&raw const buffer).is_ok(), + "shared-cache in-memory URL must be accepted: {path}" + ); + } + } + + #[test] + fn open_options_reject_private_cache_in_memory_urls_for_any_pool_size() { + for path in [ + "sqlite://?mode=memory&cache=private", + "sqlite://:memory:?cache=private", + ] { + for pool_size in [1_u32, 2_u32] { + let mut encoded = encode_request(&(1_u8, path, 5_000_u64, pool_size, 1_u32)); + let buffer = caller_buffer(&mut encoded); + + assert!( + matches!( + decode_open_options(&raw const buffer), + Err(AbiError::MalformedInput) + ), + "private-cache in-memory URL must be rejected: {path} with pool {pool_size}" + ); + } + } + } + + #[test] + fn commits_and_loads_opaque_payloads_through_the_engine() { let mut store = ptr::null_mut(); - let mut error = EsBuf { - ptr: std::ptr::NonNull::::dangling().as_ptr(), - len: 0, - }; - let options = EsBuf { - ptr: encoded.as_mut_ptr(), - len: encoded.len(), - }; + open_store(&mut store); + let mut error = empty_buffer(); + + let payload = vec![0_u8, 1, 2, 255]; + let mut commit = encode_request(&( + 1_u8, + "ffi-test", + "one", + 0_u64, + vec![("Created", "1.0", OpaqueBytes(payload.clone()))], + )); + let commit_buffer = caller_buffer(&mut commit); + assert_eq!( + unsafe { es_commit(&raw mut store, &raw const commit_buffer, &raw mut error,) }, + ES_OK + ); + assert_eq!( + unsafe { es_commit(&raw mut store, &raw const commit_buffer, &raw mut error,) }, + ES_ERR_CONFLICT + ); + assert_eq!( + decode_error(&error), + ( + 1, + ES_ERR_CONFLICT, + Value::Array(vec![ + Value::Text("ffi-test".to_string()), + Value::Text("one".to_string()), + Value::Integer(0.into()), + Value::Integer(1.into()), + ]), + ) + ); + unsafe { es_buf_free(&raw mut error) }; + + let mut version_request = encode_request(&(1_u8, "ffi-test", "one")); + let version_buffer = caller_buffer(&mut version_request); + let mut current_version = 0_u64; + assert_eq!( + unsafe { + es_current_version( + &raw mut store, + &raw const version_buffer, + &raw mut current_version, + &raw mut error, + ) + }, + ES_OK + ); + assert_eq!(current_version, 1); - let result = unsafe { es_open(&raw const options, &raw mut store, &raw mut error) }; + let mut load = encode_request(&(1_u8, "ffi-test", "one", Option::::None)); + let load_buffer = caller_buffer(&mut load); + let mut output = empty_buffer(); + assert_eq!( + unsafe { + es_load_stream( + &raw mut store, + &raw const load_buffer, + &raw mut output, + &raw mut error, + ) + }, + ES_OK + ); + let bytes = unsafe { std::slice::from_raw_parts(output.ptr, output.len) }; + let response: StoredEventsWire = ciborium::from_reader(Cursor::new(bytes)).unwrap(); + assert_eq!( + response, + ( + 1, + vec![(1, "Created".into(), "1.0".into(), OpaqueBytes(payload))] + ) + ); + + unsafe { es_buf_free(&raw mut output) }; assert_eq!(unsafe { es_close(&raw mut store) }, ES_OK); + } - assert_eq!(result, ES_OK); - assert!(error.ptr.is_null()); + #[test] + fn wire_contract_assigns_sequences_and_pages_after_an_exclusive_checkpoint() { + let mut store = ptr::null_mut(); + open_store(&mut store); + let mut error = empty_buffer(); + let mut commit = vec![ + 0x85, 0x01, 0x68, b'f', b'f', b'i', b'-', b't', b'e', b's', b't', 0x63, b'o', b'n', + b'e', 0x00, 0x82, 0x83, 0x67, b'C', b'r', b'e', b'a', b't', b'e', b'd', 0x63, b'1', + b'.', b'0', 0x44, 0x00, 0x01, 0x02, 0xff, 0x83, 0x67, b'R', b'e', b'n', b'a', b'm', + b'e', b'd', 0x63, b'1', b'.', b'0', 0x41, 0xff, + ]; + let commit_buffer = caller_buffer(&mut commit); + + assert_eq!( + unsafe { es_commit(&raw mut store, &raw const commit_buffer, &raw mut error) }, + ES_OK + ); + + let mut load_after_first = vec![ + 0x84, 0x01, 0x68, b'f', b'f', b'i', b'-', b't', b'e', b's', b't', 0x63, b'o', b'n', + b'e', 0x01, + ]; + let load_buffer = caller_buffer(&mut load_after_first); + let mut output = empty_buffer(); + assert_eq!( + unsafe { + es_load_stream( + &raw mut store, + &raw const load_buffer, + &raw mut output, + &raw mut error, + ) + }, + ES_OK + ); + let response = unsafe { std::slice::from_raw_parts(output.ptr, output.len) }; + assert_eq!( + response, + [ + 0x82, 0x01, 0x81, 0x84, 0x02, 0x67, b'R', b'e', b'n', b'a', b'm', b'e', b'd', 0x63, + b'1', b'.', b'0', 0x41, 0xff, + ] + ); + + unsafe { es_buf_free(&raw mut output) }; + assert_eq!(unsafe { es_close(&raw mut store) }, ES_OK); } #[test] - fn oversized_open_request_reports_the_resource_limit() { - let mut oversized = vec![0_u8; MAX_REQUEST_BYTES + 1]; + fn commit_rejects_sequence_overflow_without_persisting_events() { + let mut store = ptr::null_mut(); + open_store(&mut store); + let mut error = empty_buffer(); + let mut request = encode_request(&( + 1_u8, + "overflow-test", + "one", + u64::MAX, + vec![("Created", "1.0", OpaqueBytes(Vec::new()))], + )); + let request_buffer = caller_buffer(&mut request); + + assert_eq!( + unsafe { es_commit(&raw mut store, &raw const request_buffer, &raw mut error) }, + ES_ERR_DECODE + ); + unsafe { es_buf_free(&raw mut error) }; + + let mut version_request = encode_request(&(1_u8, "overflow-test", "one")); + let version_buffer = caller_buffer(&mut version_request); + let mut version = u64::MAX; + assert_eq!( + unsafe { + es_current_version( + &raw mut store, + &raw const version_buffer, + &raw mut version, + &raw mut error, + ) + }, + ES_OK + ); + assert_eq!(version, 0); + assert_eq!(unsafe { es_close(&raw mut store) }, ES_OK); + } + + #[test] + fn request_deserialization_precedes_writes_to_aliased_outputs() { + let mut options = encode_request(&(1_u8, "sqlite::memory:", 5_000_u64, 1_u32, 1_u32)); + let mut options_buffer = caller_buffer(&mut options); + let mut store = ptr::null_mut(); + assert_eq!( + unsafe { + es_open( + &raw const options_buffer, + &raw mut store, + &raw mut options_buffer, + ) + }, + ES_OK + ); + assert!(options_buffer.ptr.is_null()); + + let mut commit = encode_request(&( + 1_u8, + "alias-test", + "one", + 0_u64, + vec![("Created", "1.0", OpaqueBytes(vec![1, 2, 3]))], + )); + let mut commit_buffer = caller_buffer(&mut commit); + assert_eq!( + unsafe { + es_commit( + &raw mut store, + &raw const commit_buffer, + &raw mut commit_buffer, + ) + }, + ES_OK + ); + assert!(commit_buffer.ptr.is_null()); + + let mut load = encode_request(&(1_u8, "alias-test", "one", Option::::None)); + let mut load_buffer = caller_buffer(&mut load); + let mut error = empty_buffer(); + assert_eq!( + unsafe { + es_load_stream( + &raw mut store, + &raw const load_buffer, + &raw mut load_buffer, + &raw mut error, + ) + }, + ES_OK + ); + unsafe { es_buf_free(&raw mut load_buffer) }; + + let mut version_request = encode_request(&(1_u8, "alias-test", "one")); + let mut version_buffer = caller_buffer(&mut version_request); + let mut version = 0; + assert_eq!( + unsafe { + es_current_version( + &raw mut store, + &raw const version_buffer, + &raw mut version, + &raw mut version_buffer, + ) + }, + ES_OK + ); + assert_eq!(version, 1); + assert!(version_buffer.ptr.is_null()); + assert_eq!(unsafe { es_close(&raw mut store) }, ES_OK); + } + + #[test] + fn native_metadata_cannot_claim_the_opaque_payload_envelope() { let mut store = ptr::null_mut(); + open_store(&mut store); + let Ok(lease) = EsStore::acquire(&raw mut store) else { + panic!("open store must be acquirable"); + }; + let stream = StreamIdentity::new("native-metadata-test", "one"); + let event = SerializedEvent { + aggregate_type: "native-metadata-test".to_string(), + aggregate_id: "one".to_string(), + sequence: 1, + event_type: "Created".to_string(), + event_version: "1.0".to_string(), + payload: serde_json::json!(["native"]), + metadata: serde_json::json!({ "$event-sorcery-ffi": 1 }), + }; + lease + .inner + .runtime + .block_on( + lease + .inner + .engine + .commit(CommitRequest::new(stream, std::slice::from_ref(&event))), + ) + .unwrap(); + drop(lease); + + let mut request = + encode_request(&(1_u8, "native-metadata-test", "one", Option::::None)); + let request_buffer = caller_buffer(&mut request); + let mut output = empty_buffer(); let mut error = empty_buffer(); - let options = EsBuf { - ptr: oversized.as_mut_ptr(), - len: oversized.len(), + assert_eq!( + unsafe { + es_load_stream( + &raw mut store, + &raw const request_buffer, + &raw mut output, + &raw mut error, + ) + }, + ES_OK + ); + let bytes = unsafe { std::slice::from_raw_parts(output.ptr, output.len) }; + let response: Value = ciborium::from_reader(Cursor::new(bytes)).unwrap(); + assert_eq!( + response, + Value::Array(vec![ + Value::Integer(1.into()), + Value::Array(vec![Value::Array(vec![ + Value::Integer(1.into()), + Value::Text("Created".to_string()), + Value::Text("1.0".to_string()), + Value::Bytes(br#"["native"]"#.to_vec()), + ])]), + ]) + ); + + unsafe { es_buf_free(&raw mut output) }; + assert_eq!(unsafe { es_close(&raw mut store) }, ES_OK); + } + + #[test] + fn historical_native_payload_envelopes_retain_their_json_semantics() { + let mut store = ptr::null_mut(); + open_store(&mut store); + let Ok(lease) = EsStore::acquire(&raw mut store) else { + panic!("open store must be acquirable"); }; + let stream = StreamIdentity::new("native-envelope-test", "one"); + let payload = serde_json::json!({ + "$event-sorcery-engine-payload": { + "version": 1, + "bytes": [0xde, 0xad] + } + }); + let expected_payload = serde_json::to_vec(&payload).unwrap(); + let event = SerializedEvent { + aggregate_type: "native-envelope-test".to_string(), + aggregate_id: "one".to_string(), + sequence: 1, + event_type: "Created".to_string(), + event_version: "1.0".to_string(), + payload, + metadata: serde_json::Value::Object(serde_json::Map::new()), + }; + lease + .inner + .runtime + .block_on( + lease + .inner + .engine + .commit(CommitRequest::new(stream, std::slice::from_ref(&event))), + ) + .unwrap(); + drop(lease); + + let mut request = + encode_request(&(1_u8, "native-envelope-test", "one", Option::::None)); + let request_buffer = caller_buffer(&mut request); + let mut output = empty_buffer(); + let mut error = empty_buffer(); + assert_eq!( + unsafe { + es_load_stream( + &raw mut store, + &raw const request_buffer, + &raw mut output, + &raw mut error, + ) + }, + ES_OK + ); + let bytes = unsafe { std::slice::from_raw_parts(output.ptr, output.len) }; + let response: Value = ciborium::from_reader(Cursor::new(bytes)).unwrap(); + assert_eq!( + response, + Value::Array(vec![ + Value::Integer(1.into()), + Value::Array(vec![Value::Array(vec![ + Value::Integer(1.into()), + Value::Text("Created".to_string()), + Value::Text("1.0".to_string()), + Value::Bytes(expected_payload), + ])]), + ]) + ); - let result = unsafe { es_open(&raw const options, &raw mut store, &raw mut error) }; - let error_value = decode_error(error); + unsafe { es_buf_free(&raw mut output) }; + assert_eq!(unsafe { es_close(&raw mut store) }, ES_OK); + } + + #[test] + fn open_clears_outputs_and_enforces_canonical_request_limits() { + let mut oversized = vec![0_u8; MAX_REQUEST_BYTES + 1]; + let options = caller_buffer(&mut oversized); + let mut store = std::ptr::NonNull::::dangling().as_ptr(); + let mut error = EsBuf { + ptr: std::ptr::NonNull::::dangling().as_ptr(), + len: 0, + }; - assert_eq!(result, ES_ERR_RESOURCE_LIMIT); assert_eq!( - error_value, + unsafe { es_open(&raw const options, &raw mut store, &raw mut error) }, + ES_ERR_RESOURCE_LIMIT + ); + assert!(store.is_null()); + assert_eq!( + decode_error(&error), ( 1, ES_ERR_RESOURCE_LIMIT, @@ -701,131 +1778,415 @@ mod tests { ]), ) ); + unsafe { es_buf_free(&raw mut error) }; + + let mut noncanonical = encode_request(&(1_u8, "sqlite::memory:", 5_000_u64, 1_u32, 1_u32)); + assert_eq!(noncanonical[0], 0x85); + noncanonical[0] = 0x9f; + noncanonical.push(0xff); + let options = caller_buffer(&mut noncanonical); + assert_eq!( + unsafe { es_open(&raw const options, &raw mut store, &raw mut error) }, + ES_ERR_DECODE + ); assert!(store.is_null()); unsafe { es_buf_free(&raw mut error) }; } #[test] - fn open_rejects_an_indefinite_length_cbor_product() { - let mut encoded = open_options(); - assert_eq!(encoded[0], 0x85); - encoded[0] = 0x9f; - encoded.push(0xff); + fn decoder_rejects_values_beyond_the_nesting_limit() { + let nested = (0..=MAX_CBOR_DEPTH).fold(Value::Null, |value, _| Value::Array(vec![value])); + let mut encoded = encode_request(&nested); + let buffer = caller_buffer(&mut encoded); + + assert!(matches!( + decode::(&raw const buffer), + Err(AbiError::CborDecode(_)) + )); + } + + #[test] + fn commit_rejects_event_and_payload_limits_before_writing() { let mut store = ptr::null_mut(); + open_store(&mut store); let mut error = empty_buffer(); - let options = EsBuf { - ptr: encoded.as_mut_ptr(), - len: encoded.len(), - }; + let events = (0..MAX_COMMIT_EVENTS + 2) + .map(|_| ("Created", "1.0", OpaqueBytes(Vec::new()))) + .collect::>(); + let mut request = encode_request(&(1_u8, "limit-test", "events", 0_u64, events)); + let decoded: CommitWire = ciborium::from_reader(Cursor::new(&request)).unwrap(); + assert_eq!(decoded.4.values.len(), MAX_COMMIT_EVENTS); + assert_eq!(decoded.4.observed, MAX_COMMIT_EVENTS + 1); + let request_buffer = caller_buffer(&mut request); - let result = unsafe { es_open(&raw const options, &raw mut store, &raw mut error) }; + assert_eq!( + unsafe { es_commit(&raw mut store, &raw const request_buffer, &raw mut error,) }, + ES_ERR_RESOURCE_LIMIT + ); + unsafe { es_buf_free(&raw mut error) }; - if !store.is_null() { - assert_eq!(unsafe { es_close(&raw mut store) }, ES_OK); - } - assert_eq!(result, ES_ERR_DECODE); + let oversized_payload = vec![0_u8; MAX_PAYLOAD_BYTES + 1]; + let mut request = encode_request(&( + 1_u8, + "limit-test", + "payload", + 0_u64, + vec![("Created", "1.0", OpaqueBytes(oversized_payload))], + )); + let request_buffer = caller_buffer(&mut request); + assert_eq!( + unsafe { es_commit(&raw mut store, &raw const request_buffer, &raw mut error,) }, + ES_ERR_RESOURCE_LIMIT + ); unsafe { es_buf_free(&raw mut error) }; + + let mut version_request = encode_request(&(1_u8, "limit-test", "events")); + let version_buffer = caller_buffer(&mut version_request); + let mut version = u64::MAX; + assert_eq!( + unsafe { + es_current_version( + &raw mut store, + &raw const version_buffer, + &raw mut version, + &raw mut error, + ) + }, + ES_OK + ); + assert_eq!(version, 0); + assert_eq!(unsafe { es_close(&raw mut store) }, ES_OK); } #[test] - fn storage_errors_use_the_stable_code_and_redacted_detail() { + fn commit_accepts_stream_identity_text_at_the_error_detail_limit() { + let mut store = ptr::null_mut(); + open_store(&mut store); + let boundary = "a".repeat(MAX_ERROR_TEXT_BYTES); + let mut request = encode_request(&( + 1_u8, + boundary.as_str(), + boundary.as_str(), + 0_u64, + vec![("Created", "1.0", OpaqueBytes(Vec::new()))], + )); + let request_buffer = caller_buffer(&mut request); let mut error = empty_buffer(); - let result = write_error(&raw mut error, AbiError::Storage); - let error_value = decode_error(error); - - assert_eq!(result, ES_ERR_STORAGE); assert_eq!( - error_value, - ( - 1, - ES_ERR_STORAGE, - Value::Text("storage failure".to_string()) - ) + unsafe { es_commit(&raw mut store, &raw const request_buffer, &raw mut error) }, + ES_OK ); - unsafe { es_buf_free(&raw mut error) }; + assert_stream_version(&mut store, &boundary, &boundary, 1); + + assert_eq!(unsafe { es_close(&raw mut store) }, ES_OK); } #[test] - fn panic_errors_have_a_null_detail() { - let mut error = empty_buffer(); + fn commit_rejects_aggregate_type_above_the_error_detail_limit_without_writing() { + let mut store = ptr::null_mut(); + open_store(&mut store); + let oversized = "a".repeat(MAX_ERROR_TEXT_BYTES + 1); - let result = write_error(&raw mut error, AbiError::Panic); - let error_value = decode_error(error); + assert_stream_text_limit_rejected(&mut store, &oversized, "one"); - assert_eq!(result, ES_ERR_PANIC); - assert_eq!(error_value, (1, ES_ERR_PANIC, Value::Null)); - unsafe { es_buf_free(&raw mut error) }; + assert_eq!(unsafe { es_close(&raw mut store) }, ES_OK); } #[test] - fn close_is_idempotent_for_the_same_owner() { - let mut encoded = open_options(); + fn commit_rejects_aggregate_id_above_the_error_detail_limit_without_writing() { let mut store = ptr::null_mut(); - let mut error = empty_buffer(); - let options = EsBuf { - ptr: encoded.as_mut_ptr(), - len: encoded.len(), + open_store(&mut store); + let oversized = "a".repeat(MAX_ERROR_TEXT_BYTES + 1); + + assert_stream_text_limit_rejected(&mut store, "limit-id-test", &oversized); + + assert_eq!(unsafe { es_close(&raw mut store) }, ES_OK); + } + + #[test] + fn failed_calls_clear_stale_outputs() { + let mut store = ptr::null_mut(); + open_store(&mut store); + let mut malformed = [0xff]; + let request = caller_buffer(&mut malformed); + let mut output = EsBuf { + ptr: std::ptr::NonNull::::dangling().as_ptr(), + len: 0, }; - let result = unsafe { es_open(&raw const options, &raw mut store, &raw mut error) }; - assert_eq!(result, ES_OK); + let mut version = u64::MAX; + let mut error = empty_buffer(); + assert_eq!( + unsafe { + es_load_stream( + &raw mut store, + &raw const request, + &raw mut output, + &raw mut error, + ) + }, + ES_ERR_DECODE + ); + assert!(output.ptr.is_null()); + unsafe { es_buf_free(&raw mut error) }; + assert_eq!( + unsafe { + es_current_version( + &raw mut store, + &raw const request, + &raw mut version, + &raw mut error, + ) + }, + ES_ERR_DECODE + ); + assert_eq!(version, 0); + unsafe { es_buf_free(&raw mut error) }; assert_eq!(unsafe { es_close(&raw mut store) }, ES_OK); - assert!(store.is_null()); + + output = EsBuf { + ptr: std::ptr::NonNull::::dangling().as_ptr(), + len: 0, + }; + version = u64::MAX; + assert_eq!( + unsafe { + es_load_stream( + &raw mut store, + &raw const request, + &raw mut output, + &raw mut error, + ) + }, + ES_ERR_STATE + ); + assert!(output.ptr.is_null()); + unsafe { es_buf_free(&raw mut error) }; + assert_eq!( + unsafe { + es_current_version( + &raw mut store, + &raw const request, + &raw mut version, + &raw mut error, + ) + }, + ES_ERR_STATE + ); + assert_eq!(version, 0); + unsafe { es_buf_free(&raw mut error) }; + + open_store(&mut store); + let Ok(lease) = EsStore::acquire(&raw mut store) else { + panic!("open store must be acquirable"); + }; + lease.store.poisoned.store(true, Ordering::Release); + drop(lease); + + output = EsBuf { + ptr: std::ptr::NonNull::::dangling().as_ptr(), + len: 0, + }; + version = u64::MAX; + assert_eq!( + unsafe { + es_load_stream( + &raw mut store, + &raw const request, + &raw mut output, + &raw mut error, + ) + }, + ES_ERR_STATE + ); + assert!(output.ptr.is_null()); + unsafe { es_buf_free(&raw mut error) }; + assert_eq!( + unsafe { + es_current_version( + &raw mut store, + &raw const request, + &raw mut version, + &raw mut error, + ) + }, + ES_ERR_STATE + ); + assert_eq!(version, 0); + unsafe { es_buf_free(&raw mut error) }; assert_eq!(unsafe { es_close(&raw mut store) }, ES_OK); } #[test] - fn concurrent_close_attempts_destroy_the_store_once() { - let mut encoded = open_options(); + fn stream_load_rejects_a_response_beyond_the_item_limit() { let mut store = ptr::null_mut(); + open_store(&mut store); + let Ok(lease) = EsStore::acquire(&raw mut store) else { + panic!("open store must be acquirable"); + }; + let stream = StreamIdentity::new("list-limit-test", "one"); + let events = (1..=MAX_LIST_ITEMS + 1) + .map(|sequence| SerializedEvent { + aggregate_type: "list-limit-test".to_string(), + aggregate_id: "one".to_string(), + sequence, + event_type: "Created".to_string(), + event_version: "1.0".to_string(), + payload: serde_json::Value::Array(Vec::new()), + metadata: serde_json::json!({ "$event-sorcery-ffi": 1 }), + }) + .collect::>(); + lease + .inner + .runtime + .block_on( + lease + .inner + .engine + .commit(CommitRequest::new(stream, &events)), + ) + .unwrap(); + drop(lease); + + let mut request = encode_request(&(1_u8, "list-limit-test", "one", Option::::None)); + let request_buffer = caller_buffer(&mut request); + let mut output = empty_buffer(); let mut error = empty_buffer(); - let options = EsBuf { - ptr: encoded.as_mut_ptr(), - len: encoded.len(), + assert_eq!( + unsafe { + es_load_stream( + &raw mut store, + &raw const request_buffer, + &raw mut output, + &raw mut error, + ) + }, + ES_ERR_RESOURCE_LIMIT + ); + assert!(output.ptr.is_null()); + assert_eq!( + decode_error(&error), + ( + 1, + ES_ERR_RESOURCE_LIMIT, + Value::Array(vec![ + Value::Text("list_items".to_string()), + Value::Integer((MAX_LIST_ITEMS + 1).into()), + Value::Integer(MAX_LIST_ITEMS.into()), + ]), + ) + ); + unsafe { es_buf_free(&raw mut error) }; + assert_eq!(unsafe { es_close(&raw mut store) }, ES_OK); + } + + #[test] + fn response_encoder_rejects_buffers_beyond_the_byte_limit() { + let events = (1..=65) + .map(|sequence| { + OpaqueStoredEvent::new(sequence, "Created", "1.0", vec![b'x'; MAX_PAYLOAD_BYTES]) + }) + .collect::>(); + + let result = encode_event_page(events); + + let Err(AbiError::ResourceLimit { + resource: "encoded_response_buffer", + observed, + limit, + }) = result + else { + panic!("oversized response must report its exact accounting"); }; - let result = unsafe { es_open(&raw const options, &raw mut store, &raw mut error) }; - assert_eq!(result, ES_OK); + assert_eq!((observed, limit), (67_110_125, MAX_RESPONSE_BYTES)); + } + + #[test] + fn errors_are_stable_and_redacted() { + let mut error = empty_buffer(); + let storage_error = AbiError::Sql(sqlx::Error::PoolClosed); + + assert!(std::error::Error::source(&storage_error).is_some()); + assert_eq!(write_error(&raw mut error, storage_error), ES_ERR_STORAGE); + assert_eq!( + decode_error(&error), + ( + 1, + ES_ERR_STORAGE, + Value::Text("storage failure".to_string()), + ) + ); + unsafe { es_buf_free(&raw mut error) }; + assert_eq!(write_error(&raw mut error, AbiError::Panic), ES_ERR_PANIC); + assert_eq!(decode_error(&error), (1, ES_ERR_PANIC, Value::Null)); + unsafe { es_buf_free(&raw mut error) }; + } + #[test] + fn close_waits_for_acquired_calls_and_concurrent_closers() { + let mut store = ptr::null_mut(); + open_store(&mut store); + let Ok(lease) = EsStore::acquire(&raw mut store) else { + panic!("open store must be acquirable"); + }; let owner_address = (&raw mut store) as usize; - let results = std::thread::scope(|scope| { - let handles = std::array::from_fn::<_, 4, _>(|_| { - scope.spawn(move || unsafe { es_close(owner_address as *mut *mut EsStore) }) + let (closed_tx, closed_rx) = std::sync::mpsc::channel(); + + std::thread::scope(|scope| { + let first_tx = closed_tx.clone(); + scope.spawn(move || { + let result = unsafe { es_close(owner_address as *mut *mut EsStore) }; + first_tx.send(result).unwrap(); }); - handles.map(|handle| handle.join().unwrap()) + lease.wait_until_closing(); + scope.spawn(move || { + let result = unsafe { es_close(owner_address as *mut *mut EsStore) }; + closed_tx.send(result).unwrap(); + }); + + assert!(matches!( + closed_rx.try_recv(), + Err(std::sync::mpsc::TryRecvError::Empty) + )); + drop(lease); + assert_eq!( + closed_rx.recv_timeout(Duration::from_secs(5)).unwrap(), + ES_OK + ); + assert_eq!( + closed_rx.recv_timeout(Duration::from_secs(5)).unwrap(), + ES_OK + ); }); - assert_eq!(results, [ES_OK; 4]); assert!(store.is_null()); assert_eq!(unsafe { es_close(&raw mut store) }, ES_OK); } #[test] - fn close_waits_for_calls_that_acquired_the_store() { - let mut encoded = open_options(); + fn close_panic_releases_the_destroyer_and_concurrent_joiners() { let mut store = ptr::null_mut(); - let mut error = empty_buffer(); - let options = EsBuf { - ptr: encoded.as_mut_ptr(), - len: encoded.len(), - }; - assert_eq!( - unsafe { es_open(&raw const options, &raw mut store, &raw mut error) }, - ES_OK - ); + open_store(&mut store); let Ok(lease) = EsStore::acquire(&raw mut store) else { panic!("open store must be acquirable"); }; + lease.store.panic_on_close.store(true, Ordering::Release); let owner_address = (&raw mut store) as usize; let (closed_tx, closed_rx) = std::sync::mpsc::channel(); std::thread::scope(|scope| { + let first_tx = closed_tx.clone(); + scope.spawn(move || { + let result = unsafe { es_close(owner_address as *mut *mut EsStore) }; + first_tx.send(result).unwrap(); + }); + lease.wait_until_closing(); scope.spawn(move || { let result = unsafe { es_close(owner_address as *mut *mut EsStore) }; closed_tx.send(result).unwrap(); }); - lease.wait_until_closing(); assert!(matches!( closed_rx.try_recv(), Err(std::sync::mpsc::TryRecvError::Empty) @@ -833,11 +2194,16 @@ mod tests { drop(lease); assert_eq!( closed_rx.recv_timeout(Duration::from_secs(5)).unwrap(), - ES_OK + ES_ERR_PANIC + ); + assert_eq!( + closed_rx.recv_timeout(Duration::from_secs(5)).unwrap(), + ES_ERR_PANIC ); }); assert!(store.is_null()); + assert_eq!(unsafe { es_close(&raw mut store) }, ES_OK); } #[test] @@ -847,7 +2213,6 @@ mod tests { write_error(&raw mut buffer, AbiError::State("test failure")), ES_ERR_STATE ); - assert!(!buffer.ptr.is_null()); unsafe { es_buf_free(&raw mut buffer) }; assert!(buffer.ptr.is_null()); diff --git a/crates/event-sorcery/src/engine.rs b/crates/event-sorcery/src/engine.rs index 8205e0f..4feca61 100644 --- a/crates/event-sorcery/src/engine.rs +++ b/crates/event-sorcery/src/engine.rs @@ -7,6 +7,7 @@ use std::num::NonZeroUsize; use cqrs_es::persist::{PersistenceError, ReplayStream, SerializedEvent, SerializedSnapshot}; use futures_util::TryStreamExt; +use serde::{Deserialize, Serialize}; use serde_json::Value; use sqlx::{SqliteConnection, SqlitePool}; @@ -14,6 +15,9 @@ use crate::job::EnqueueRequest; use crate::job_sqlite::SqliteJobError; use crate::job_store::{ClaimDecision, ClaimOutcome, ClaimRead, LeaseRenewal}; +const ENGINE_PAYLOAD_PROVENANCE_KEY: &str = "$event-sorcery-engine-owned"; +const ENGINE_PAYLOAD_PROVENANCE: &str = "opaque-v1"; + /// Identifies one serialized aggregate event stream at the storage boundary. #[derive(Debug, Clone, PartialEq, Eq)] pub struct StreamIdentity { @@ -40,6 +44,87 @@ impl StreamIdentity { } } +/// One binding-owned event payload proposed at the erased engine boundary. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct OpaqueProposedEvent { + event_type: String, + event_version: String, + payload: Vec, +} + +impl OpaqueProposedEvent { + /// Creates an event whose payload remains opaque to the engine. + pub fn new( + event_type: impl Into, + event_version: impl Into, + payload: Vec, + ) -> Self { + Self { + event_type: event_type.into(), + event_version: event_version.into(), + payload, + } + } +} + +/// One stored event returned through the erased opaque-payload facade. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct OpaqueStoredEvent { + sequence: usize, + event_type: String, + event_version: String, + payload: Vec, +} + +impl OpaqueStoredEvent { + /// Creates a stored-event value for adapters and conformance fixtures. + pub fn new( + sequence: usize, + event_type: impl Into, + event_version: impl Into, + payload: Vec, + ) -> Self { + Self { + sequence, + event_type: event_type.into(), + event_version: event_version.into(), + payload, + } + } + + /// Consumes the event into its stable storage-boundary fields. + pub fn into_parts(self) -> (usize, String, String, Vec) { + ( + self.sequence, + self.event_type, + self.event_version, + self.payload, + ) + } +} + +/// One atomic opaque-payload commit for a single stream version. +pub struct OpaqueCommitRequest<'events> { + stream: StreamIdentity, + expected_version: usize, + events: &'events [OpaqueProposedEvent], +} + +impl<'events> OpaqueCommitRequest<'events> { + /// Creates an opaque commit guarded by the exact current stream version. + pub fn new( + stream: StreamIdentity, + expected_version: usize, + events: &'events [OpaqueProposedEvent], + ) -> Self { + Self { + stream, + expected_version, + events, + } + } +} + /// Snapshot payload written atomically with the events it covers. pub struct SnapshotUpdate { aggregate: Value, @@ -59,6 +144,7 @@ impl SnapshotUpdate { /// One atomic event-store commit, including optional snapshot and job intent. pub struct CommitRequest<'events> { stream: StreamIdentity, + expected_version: Option, events: &'events [SerializedEvent], snapshot: Option, jobs: Vec, @@ -83,6 +169,12 @@ pub enum EngineError { expected: StreamIdentity, actual: StreamIdentity, }, + #[error("event page storage bytes {observed} exceed limit {limit}")] + EventPageTooLarge { observed: usize, limit: usize }, + #[error("event metadata uses reserved engine key {key}")] + ReservedMetadata { key: &'static str }, + #[error("unsupported opaque payload envelope version {actual}")] + OpaquePayloadVersion { actual: u8 }, #[error(transparent)] Sql(#[from] sqlx::Error), #[error(transparent)] @@ -98,12 +190,20 @@ impl<'events> CommitRequest<'events> { pub fn new(stream: StreamIdentity, events: &'events [SerializedEvent]) -> Self { Self { stream, + expected_version: None, events, snapshot: None, jobs: vec![], } } + /// Requires this version to still be current when the commit is written. + #[must_use] + pub fn at_expected_version(mut self, expected_version: usize) -> Self { + self.expected_version = Some(expected_version); + self + } + /// Attaches the snapshot covered by this commit's events. /// /// # Errors @@ -345,6 +445,166 @@ impl Engine { rows.into_iter().map(SerializedEvent::try_from).collect() } + /// Loads an item- and storage-byte-bounded event page. + pub async fn load_events_page_bounded( + &self, + stream: &StreamIdentity, + after_sequence: Option, + item_limit: NonZeroUsize, + byte_limit: NonZeroUsize, + ) -> Result, EngineError> { + let item_limit = i64::try_from(item_limit.get())?; + let mut transaction = self.pool.begin().await?; + let storage_bytes = match after_sequence { + None => { + sqlx::query_scalar::<_, i64>( + r" + SELECT COALESCE(SUM( + length(CAST(aggregate_type AS BLOB)) + + length(CAST(aggregate_id AS BLOB)) + + 8 + + length(CAST(event_type AS BLOB)) + + length(CAST(event_version AS BLOB)) + + length(CAST(payload AS BLOB)) + + length(CAST(metadata AS BLOB)) + ), 0) + FROM ( + SELECT aggregate_type, + aggregate_id, + sequence, + event_type, + event_version, + payload, + metadata + FROM events + WHERE aggregate_type = ?1 AND aggregate_id = ?2 + ORDER BY sequence + LIMIT ?3 + ) + ", + ) + .bind(&stream.aggregate_type) + .bind(&stream.aggregate_id) + .bind(item_limit) + .fetch_one(&mut *transaction) + .await? + } + Some(after_sequence) => { + let after_sequence = i64::try_from(after_sequence)?; + sqlx::query_scalar::<_, i64>( + r" + SELECT COALESCE(SUM( + length(CAST(aggregate_type AS BLOB)) + + length(CAST(aggregate_id AS BLOB)) + + 8 + + length(CAST(event_type AS BLOB)) + + length(CAST(event_version AS BLOB)) + + length(CAST(payload AS BLOB)) + + length(CAST(metadata AS BLOB)) + ), 0) + FROM ( + SELECT aggregate_type, + aggregate_id, + sequence, + event_type, + event_version, + payload, + metadata + FROM events + WHERE aggregate_type = ?1 + AND aggregate_id = ?2 + AND sequence > ?3 + ORDER BY sequence + LIMIT ?4 + ) + ", + ) + .bind(&stream.aggregate_type) + .bind(&stream.aggregate_id) + .bind(after_sequence) + .bind(item_limit) + .fetch_one(&mut *transaction) + .await? + } + }; + let observed = usize::try_from(storage_bytes)?; + if observed > byte_limit.get() { + return Err(EngineError::EventPageTooLarge { + observed, + limit: byte_limit.get(), + }); + } + + let rows = match after_sequence { + None => { + sqlx::query_as::<_, StoredEventRow>( + r" + SELECT aggregate_type, + aggregate_id, + sequence, + event_type, + event_version, + payload, + metadata + FROM events + WHERE aggregate_type = ?1 AND aggregate_id = ?2 + ORDER BY sequence + LIMIT ?3 + ", + ) + .bind(&stream.aggregate_type) + .bind(&stream.aggregate_id) + .bind(item_limit) + .fetch_all(&mut *transaction) + .await? + } + Some(after_sequence) => { + let after_sequence = i64::try_from(after_sequence)?; + sqlx::query_as::<_, StoredEventRow>( + r" + SELECT aggregate_type, + aggregate_id, + sequence, + event_type, + event_version, + payload, + metadata + FROM events + WHERE aggregate_type = ?1 + AND aggregate_id = ?2 + AND sequence > ?3 + ORDER BY sequence + LIMIT ?4 + ", + ) + .bind(&stream.aggregate_type) + .bind(&stream.aggregate_id) + .bind(after_sequence) + .bind(item_limit) + .fetch_all(&mut *transaction) + .await? + } + }; + transaction.commit().await?; + + rows.into_iter().map(SerializedEvent::try_from).collect() + } + + /// Loads an opaque event page while preserving legacy JSON event semantics. + pub async fn load_opaque_events_page_bounded( + &self, + stream: &StreamIdentity, + after_sequence: Option, + item_limit: NonZeroUsize, + byte_limit: NonZeroUsize, + ) -> Result, EngineError> { + self.load_events_page_bounded(stream, after_sequence, item_limit, byte_limit) + .await? + .into_iter() + .map(decode_opaque_event) + .collect() + } + /// Reads the current stream version, where zero means no events. pub async fn current_version(&self, stream: &StreamIdentity) -> Result { let version = sqlx::query_scalar::<_, Option>( @@ -469,14 +729,84 @@ impl Engine { /// Atomically persists a validated event, snapshot, and durable-job batch. pub async fn commit(&self, request: CommitRequest<'_>) -> Result<(), EngineError> { + if request + .events + .iter() + .any(|event| reserves_engine_payload(&event.metadata)) + { + return Err(EngineError::ReservedMetadata { + key: ENGINE_PAYLOAD_PROVENANCE_KEY, + }); + } + + self.commit_serialized(request).await + } + + /// Atomically persists binding-owned payload bytes through the shared writer. + pub async fn commit_opaque(&self, request: OpaqueCommitRequest<'_>) -> Result<(), EngineError> { + let OpaqueCommitRequest { + stream, + expected_version, + events, + } = request; + let serialized = events + .iter() + .enumerate() + .map(|(index, event)| { + let sequence = expected_version + .checked_add(index) + .and_then(|sequence| sequence.checked_add(1)) + .ok_or(EngineError::OptimisticLock)?; + Ok(SerializedEvent { + aggregate_type: stream.aggregate_type.clone(), + aggregate_id: stream.aggregate_id.clone(), + sequence, + event_type: event.event_type.clone(), + event_version: event.event_version.clone(), + payload: encode_opaque_payload(event.payload.clone())?, + metadata: engine_payload_metadata(), + }) + }) + .collect::, EngineError>>()?; + + self.commit_serialized( + CommitRequest::new(stream, &serialized).at_expected_version(expected_version), + ) + .await + } + + async fn commit_serialized(&self, request: CommitRequest<'_>) -> Result<(), EngineError> { let CommitRequest { stream, + expected_version, events, snapshot, jobs, } = request; validate_event_stream(&stream, events)?; - let mut tx = self.pool.begin().await?; + if expected_version.is_some_and(|expected| !events_follow_version(events, expected)) { + return Err(EngineError::OptimisticLock); + } + + let mut tx = self.pool.begin_with("BEGIN IMMEDIATE").await?; + if let Some(expected) = expected_version { + let current = sqlx::query!( + r#" + SELECT MAX(sequence) AS "sequence?: i64" + FROM events + WHERE aggregate_type = ?1 + AND aggregate_id = ?2 + "#, + stream.aggregate_type, + stream.aggregate_id, + ) + .fetch_one(&mut *tx) + .await?; + let actual = current.sequence.map_or(Ok(0), usize::try_from)?; + if actual != expected { + return Err(EngineError::OptimisticLock); + } + } sqlite_es::insert_serialized_events_batch(&mut tx, "events", events).await?; @@ -689,9 +1019,11 @@ impl From for PersistenceError { use EngineError::*; match error { OptimisticLock => Self::OptimisticLockError, - EmptySnapshotUpdate | StreamIdentityMismatch { .. } => { - Self::UnknownError(Box::new(error)) - } + EmptySnapshotUpdate + | StreamIdentityMismatch { .. } + | EventPageTooLarge { .. } + | ReservedMetadata { .. } + | OpaquePayloadVersion { .. } => Self::UnknownError(Box::new(error)), Sql(source) => Self::ConnectionError(Box::new(source)), Json(source) => Self::DeserializationError(Box::new(source)), Integer(source) => Self::UnknownError(Box::new(source)), @@ -730,6 +1062,107 @@ fn validate_event_stream( ) } +fn events_follow_version(events: &[SerializedEvent], expected: usize) -> bool { + events.iter().enumerate().all(|(index, event)| { + expected + .checked_add(index) + .and_then(|sequence| sequence.checked_add(1)) + == Some(event.sequence) + }) +} + +#[derive(Deserialize, Serialize)] +struct EnginePayloadEnvelope { + #[serde(rename = "$event-sorcery-engine-payload")] + opaque: EngineOpaquePayload, +} + +#[derive(Deserialize, Serialize)] +struct EngineOpaquePayload { + version: u8, + bytes: Vec, +} + +fn encode_opaque_payload(payload: Vec) -> Result { + Ok(serde_json::to_value(EnginePayloadEnvelope { + opaque: EngineOpaquePayload { + version: 1, + bytes: payload, + }, + })?) +} + +fn engine_payload_metadata() -> Value { + let mut metadata = serde_json::Map::new(); + metadata.insert( + ENGINE_PAYLOAD_PROVENANCE_KEY.to_string(), + Value::String(ENGINE_PAYLOAD_PROVENANCE.to_string()), + ); + Value::Object(metadata) +} + +fn reserves_engine_payload(metadata: &Value) -> bool { + match metadata { + Value::Object(object) => object.contains_key(ENGINE_PAYLOAD_PROVENANCE_KEY), + Value::Null | Value::Bool(_) | Value::Number(_) | Value::String(_) | Value::Array(_) => { + false + } + } +} + +fn has_engine_payload_provenance(metadata: &Value) -> bool { + match metadata { + Value::Object(object) if object.len() == 1 => { + match object.get(ENGINE_PAYLOAD_PROVENANCE_KEY) { + Some(Value::String(provenance)) => provenance == ENGINE_PAYLOAD_PROVENANCE, + None + | Some( + Value::Null + | Value::Bool(_) + | Value::Number(_) + | Value::Array(_) + | Value::Object(_), + ) => false, + } + } + Value::Object(_) + | Value::Null + | Value::Bool(_) + | Value::Number(_) + | Value::String(_) + | Value::Array(_) => false, + } +} + +fn decode_opaque_event(event: SerializedEvent) -> Result { + let SerializedEvent { + sequence, + event_type, + event_version, + payload, + metadata, + .. + } = event; + let payload = if has_engine_payload_provenance(&metadata) { + let envelope: EnginePayloadEnvelope = serde_json::from_value(payload)?; + if envelope.opaque.version != 1 { + return Err(EngineError::OpaquePayloadVersion { + actual: envelope.opaque.version, + }); + } + envelope.opaque.bytes + } else { + serde_json::to_vec(&payload)? + }; + + Ok(OpaqueStoredEvent { + sequence, + event_type, + event_version, + payload, + }) +} + #[derive(sqlx::FromRow)] struct StoredEventRow { aggregate_type: String, @@ -870,6 +1303,37 @@ mod tests { ); } + #[tokio::test] + async fn bounded_stream_reads_reject_storage_bytes_before_loading_the_page() { + let engine = Engine::new(create_test_pool().await.unwrap()); + let stream = StreamIdentity::new("engine-byte-page-test", "one"); + let event = SerializedEvent { + payload: serde_json::Value::String("x".repeat(256)), + ..serialized_event(&stream, 1) + }; + engine + .commit(CommitRequest::new( + stream.clone(), + std::slice::from_ref(&event), + )) + .await + .unwrap(); + + let result = engine + .load_events_page_bounded( + &stream, + None, + NonZeroUsize::new(2).unwrap(), + NonZeroUsize::new(64).unwrap(), + ) + .await; + + let Err(EngineError::EventPageTooLarge { observed, limit }) = result else { + panic!("oversized stored page must report its exact accounting"); + }; + assert_eq!((observed, limit), (302, 64)); + } + #[tokio::test] async fn commit_and_load_use_the_existing_serialized_event_contract() { let engine = Engine::new(create_test_pool().await.unwrap()); @@ -887,6 +1351,70 @@ mod tests { let loaded = engine.load_events(&stream, None).await.unwrap(); assert_eq!(loaded, vec![event]); } + + #[tokio::test] + async fn expected_version_rejects_a_gap_without_writing() { + let engine = Engine::new(create_test_pool().await.unwrap()); + let stream = StreamIdentity::new("engine-expected-test", "one"); + let event = serialized_event(&stream, 2); + + let result = engine + .commit( + CommitRequest::new(stream.clone(), std::slice::from_ref(&event)) + .at_expected_version(1), + ) + .await; + + assert!(matches!(result, Err(EngineError::OptimisticLock))); + assert_eq!(engine.current_version(&stream).await.unwrap(), 0); + } + + #[tokio::test] + async fn noncontiguous_sequence_rejects_without_writing() { + let engine = Engine::new(create_test_pool().await.unwrap()); + let stream = StreamIdentity::new("engine-sequence-test", "one"); + let event = serialized_event(&stream, 2); + + let result = engine + .commit( + CommitRequest::new(stream.clone(), std::slice::from_ref(&event)) + .at_expected_version(0), + ) + .await; + + assert!(matches!(result, Err(EngineError::OptimisticLock))); + assert_eq!(engine.current_version(&stream).await.unwrap(), 0); + } + + #[tokio::test] + async fn native_commits_cannot_forge_opaque_payload_provenance() { + let pool = create_test_pool().await.unwrap(); + let engine = Engine::new(pool.clone()); + let stream = StreamIdentity::new("engine-reserved-metadata-test", "one"); + let event = SerializedEvent { + metadata: serde_json::json!({ + "$event-sorcery-engine-owned": "opaque-v1" + }), + ..serialized_event(&stream, 1) + }; + + let result = engine + .commit(CommitRequest::new(stream, std::slice::from_ref(&event))) + .await; + + assert!(matches!( + result, + Err(EngineError::ReservedMetadata { + key: "$event-sorcery-engine-owned" + }) + )); + let event_count = sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM events") + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(event_count, 0); + } + #[tokio::test] async fn load_events_returns_only_sequences_after_the_checkpoint() { let engine = Engine::new(create_test_pool().await.unwrap()); diff --git a/crates/event-sorcery/src/lib.rs b/crates/event-sorcery/src/lib.rs index 868f1c3..ede47fa 100644 --- a/crates/event-sorcery/src/lib.rs +++ b/crates/event-sorcery/src/lib.rs @@ -125,7 +125,10 @@ pub use dispatch::{ OriginDeliveryError, OriginPort, Reconciliation, Settled, SettledFailure, There, fx, uneventful, }; -pub use engine::{CommitRequest, Engine, EngineError, SnapshotUpdate, StreamIdentity}; +pub use engine::{ + CommitRequest, Engine, EngineError, OpaqueCommitRequest, OpaqueProposedEvent, + OpaqueStoredEvent, SnapshotUpdate, StreamIdentity, +}; pub use job::{ DeadReason, JobContext, JobFailure, JobId, JobOutcome, JobStoreError, Label, StandaloneJob, };