diff --git a/crates/config/src/lib.rs b/crates/config/src/lib.rs index ff53419c..ee1c5181 100644 --- a/crates/config/src/lib.rs +++ b/crates/config/src/lib.rs @@ -147,7 +147,7 @@ impl StorageConfig { } /// Get a reference to the underlying trait object for factory calls. - pub fn as_trait(&self) -> &dyn extenddb_storage::config::StorageConfig { + pub fn as_trait(&self) -> &(dyn extenddb_storage::config::StorageConfig + 'static) { &*self.config } } diff --git a/crates/storage-mongodb/src/config.rs b/crates/storage-mongodb/src/config.rs index bb878029..81294f51 100644 --- a/crates/storage-mongodb/src/config.rs +++ b/crates/storage-mongodb/src/config.rs @@ -21,6 +21,22 @@ pub struct MongoStorageConfig { /// Maximum concurrent connections for catalog/management operations #[serde(default = "default_max_catalog_connections")] pub max_catalog_connections: u32, + /// Read concern used for the multi-document transactions that back + /// conditional writes, `TransactWriteItems`, `TransactGetItems`, and the + /// idempotency-token path (RFC-0003). Defaults to `"snapshot"`, matching + /// real MongoDB's strongest isolation level. + /// + /// Some MongoDB-wire-compatible backends (e.g. DocumentDB) do not + /// implement `readConcern: snapshot` and reject transactions that + /// request it with `CommandNotSupported` (error code 115). Setting this + /// to `"majority"` or `"local"` lets the backend run against such + /// targets, at the cost of the stronger isolation snapshot reads + /// provide: concurrent transactions may observe a slightly different + /// view of the data than they would under snapshot isolation. Only + /// change this if the target deployment's MongoDB-compatible server + /// does not support snapshot reads. + #[serde(default = "default_transaction_read_concern")] + pub transaction_read_concern: String, } fn default_max_connections() -> u32 { @@ -31,6 +47,34 @@ fn default_max_catalog_connections() -> u32 { 20 } +fn default_transaction_read_concern() -> String { + "snapshot".to_owned() +} + +/// Parse [`MongoStorageConfig::transaction_read_concern`] into a driver +/// [`mongodb::options::ReadConcern`]. +/// +/// Accepts the standard MongoDB read concern levels usable inside a +/// transaction (`snapshot`, `majority`, `local`, `linearizable`, +/// `available`), case-insensitively. Any other value is rejected at startup +/// rather than silently passed through to the driver, so a typo surfaces as +/// a clear configuration error instead of an opaque runtime failure. +pub fn parse_transaction_read_concern( + value: &str, +) -> Result { + match value.to_ascii_lowercase().as_str() { + "snapshot" => Ok(mongodb::options::ReadConcern::snapshot()), + "majority" => Ok(mongodb::options::ReadConcern::majority()), + "local" => Ok(mongodb::options::ReadConcern::local()), + "linearizable" => Ok(mongodb::options::ReadConcern::linearizable()), + "available" => Ok(mongodb::options::ReadConcern::available()), + other => Err(format!( + "invalid storage.mongodb.transaction_read_concern {other:?}: expected one of \ + \"snapshot\", \"majority\", \"local\", \"linearizable\", \"available\"" + )), + } +} + impl extenddb_storage::config::StorageConfig for MongoStorageConfig { fn connection_config(&self) -> &str { &self.connection_string diff --git a/crates/storage-mongodb/src/data_engine.rs b/crates/storage-mongodb/src/data_engine.rs index 4a6f4570..66ef6218 100644 --- a/crates/storage-mongodb/src/data_engine.rs +++ b/crates/storage-mongodb/src/data_engine.rs @@ -296,7 +296,7 @@ impl MongoEngine { .map_err(|e| StorageError::Internal(e.to_string()))?; let tx_options = mongodb::options::TransactionOptions::builder() - .read_concern(mongodb::options::ReadConcern::snapshot()) + .read_concern(self.transaction_read_concern()) .write_concern( mongodb::options::WriteConcern::builder() .w(mongodb::options::Acknowledgment::Majority) @@ -508,7 +508,7 @@ impl MongoEngine { .map_err(|e| StorageError::Internal(e.to_string()))?; let tx_options = mongodb::options::TransactionOptions::builder() - .read_concern(mongodb::options::ReadConcern::snapshot()) + .read_concern(self.transaction_read_concern()) .write_concern( mongodb::options::WriteConcern::builder() .w(mongodb::options::Acknowledgment::Majority) @@ -737,7 +737,7 @@ impl MongoEngine { .map_err(|e| StorageError::Internal(e.to_string()))?; let tx_options = mongodb::options::TransactionOptions::builder() - .read_concern(mongodb::options::ReadConcern::snapshot()) + .read_concern(self.transaction_read_concern()) .write_concern( mongodb::options::WriteConcern::builder() .w(mongodb::options::Acknowledgment::Majority) @@ -1993,7 +1993,7 @@ impl MongoEngine { .map_err(|e| StorageError::Internal(e.to_string()))?; let tx_options = mongodb::options::TransactionOptions::builder() - .read_concern(mongodb::options::ReadConcern::snapshot()) + .read_concern(self.transaction_read_concern()) .build(); session @@ -2039,7 +2039,7 @@ impl MongoEngine { .map_err(|e| StorageError::Internal(e.to_string()))?; let tx_options = mongodb::options::TransactionOptions::builder() - .read_concern(mongodb::options::ReadConcern::snapshot()) + .read_concern(self.transaction_read_concern()) .write_concern( mongodb::options::WriteConcern::builder() .w(mongodb::options::Acknowledgment::Majority) diff --git a/crates/storage-mongodb/src/lib.rs b/crates/storage-mongodb/src/lib.rs index 455efa33..84599a9b 100644 --- a/crates/storage-mongodb/src/lib.rs +++ b/crates/storage-mongodb/src/lib.rs @@ -79,7 +79,7 @@ impl ServerRuntimeHooks for MongoRuntimeHooks { /// Build the assembled server components for the mongo backend (`serve`). fn server_components_factory( - config: &dyn extenddb_storage::config::StorageConfig, + config: &(dyn extenddb_storage::config::StorageConfig + 'static), region: &str, // MongoDB bootstrap needs operator input (databases, admin credentials), so // `bootstrap_if_uninitialized` is not honored here: an uninitialized catalog @@ -91,14 +91,33 @@ fn server_components_factory( let connection_string = config.connection_config().to_string(); let max_connections = config.max_connections(); let region = region.to_string(); + // Backend-specific settings (transaction_read_concern) aren't exposed on + // the generic StorageConfig trait, so downcast to the concrete mongo + // config. This factory is only ever invoked with a MongoStorageConfig + // (registered together in `register()` below), so the downcast cannot + // fail in practice; fall back to the field's own default rather than + // panicking if it ever did. + let raw_read_concern = config + .as_any() + .downcast_ref::() + .map(|c| c.transaction_read_concern.clone()) + .unwrap_or_else(|| "snapshot".to_string()); Box::pin(async move { + let tx_read_concern = config::parse_transaction_read_concern(&raw_read_concern) + .map_err(BackendError::InitializationFailed)?; + // Create MongoEngine - let engine = MongoEngine::new(&connection_string, ®ion, max_connections) - .await - .map_err(|e| BackendError::ConnectionFailed { - backend: "mongodb".to_string(), - details: e.to_string(), - })?; + let engine = MongoEngine::new( + &connection_string, + ®ion, + max_connections, + tx_read_concern, + ) + .await + .map_err(|e| BackendError::ConnectionFailed { + backend: "mongodb".to_string(), + details: e.to_string(), + })?; let engine = Arc::new(engine); @@ -230,6 +249,11 @@ pub struct MongoEngine { /// so GSI additions/removals on other ExtendDB instances converge within /// the TTL window. gsi_cache: dashmap::DashMap, + /// Read concern applied to every multi-document transaction this engine + /// opens. Defaults to `snapshot` (real MongoDB); configurable via + /// [`crate::config::MongoStorageConfig::transaction_read_concern`] for + /// MongoDB-wire-compatible backends that don't support snapshot reads. + tx_read_concern: mongodb::options::ReadConcern, } /// Build a MongoDB client from a connection string, applying the shared @@ -290,6 +314,7 @@ impl MongoEngine { connection_string: &str, region: &str, max_connections: u32, + tx_read_concern: mongodb::options::ReadConcern, ) -> Result { let client = connect_guarded(connection_string, Some(max_connections), true).await?; @@ -302,9 +327,17 @@ impl MongoEngine { data_db, region: region.to_owned(), gsi_cache: dashmap::DashMap::new(), + tx_read_concern, }) } + /// Read concern to use for a newly-opened multi-document transaction. + /// See [`MongoEngine::tx_read_concern`] / `transaction_read_concern` + /// config for why this isn't always `snapshot`. + pub(crate) fn transaction_read_concern(&self) -> mongodb::options::ReadConcern { + self.tx_read_concern.clone() + } + /// Look up a fresh GSI-cache entry for `table_id`. /// /// Returns `Some(has_gsi)` when a cache entry exists and is younger than diff --git a/crates/storage-postgres/src/lib.rs b/crates/storage-postgres/src/lib.rs index d792fc6a..0d7a51c0 100755 --- a/crates/storage-postgres/src/lib.rs +++ b/crates/storage-postgres/src/lib.rs @@ -619,7 +619,7 @@ impl ServerRuntimeHooks for PostgresRuntimeHooks { /// Build server components for the Postgres backend (registered in [`register`]). fn server_components_factory( - config: &dyn extenddb_storage::config::StorageConfig, + config: &(dyn extenddb_storage::config::StorageConfig + 'static), region: &str, // PostgreSQL bootstrap needs operator input (databases, roles, admin // credentials), so `bootstrap_if_uninitialized` is not honored here: diff --git a/crates/storage-sqlite/src/lib.rs b/crates/storage-sqlite/src/lib.rs index c6ae5f92..54ffc0aa 100644 --- a/crates/storage-sqlite/src/lib.rs +++ b/crates/storage-sqlite/src/lib.rs @@ -180,7 +180,7 @@ use crate::hooks::SqliteRuntimeHooks; const MAX_ITEM_SIZE_BYTES: usize = 400_000; fn sqlite_server_components_factory( - config: &dyn extenddb_storage::config::StorageConfig, + config: &(dyn extenddb_storage::config::StorageConfig + 'static), region: &str, options: extenddb_storage::server_components::ServerComponentsOptions, ) -> futures::future::BoxFuture<'static, Result> { diff --git a/crates/storage/src/server_components.rs b/crates/storage/src/server_components.rs index 0755b4e6..480776c4 100644 --- a/crates/storage/src/server_components.rs +++ b/crates/storage/src/server_components.rs @@ -101,9 +101,17 @@ pub struct ServerComponentsOptions { /// /// Takes a `StorageConfig` trait object, region string, and options; returns /// a Future that resolves to `ServerComponents` or `BackendError`. +/// +/// The trait object is bound `+ 'static` (rather than the reference's own +/// elided lifetime) so factories can use [`StorageConfig::as_any`] to +/// downcast to a concrete backend config and read backend-specific settings +/// (e.g. the mongo backend's `transaction_read_concern`) — `Any::downcast_ref` +/// requires the pointee to be provably `'static`, which an unannotated `&dyn +/// StorageConfig` does not guarantee even though every real implementor owns +/// its data. pub type ServerComponentsFactory = fn( - &dyn StorageConfig, + &(dyn StorageConfig + 'static), &str, ServerComponentsOptions, ) -> Pin> + Send>>; @@ -114,7 +122,7 @@ pub type ServerComponentsFactory = /// [`set_backend`](crate::set_backend). Returns `BackendNotInstalled` if no /// backend has been installed. pub async fn create_server_components( - config: &dyn StorageConfig, + config: &(dyn StorageConfig + 'static), region: &str, options: ServerComponentsOptions, ) -> Result { diff --git a/docs/design/13-storage-mongodb.md b/docs/design/13-storage-mongodb.md index 9835314e..299a5ff0 100644 --- a/docs/design/13-storage-mongodb.md +++ b/docs/design/13-storage-mongodb.md @@ -13,6 +13,18 @@ transactions on replica sets). **Minimum MongoDB version:** 7.0 (multi-document transactions, snapshot reads). +**Transaction read concern:** `storage.mongodb.transaction_read_concern` +(default `"snapshot"`) controls the read concern applied to every +multi-document transaction this backend opens (conditional writes, +`TransactWriteItems`, `TransactGetItems`, idempotency-token checks). Real +MongoDB 7.0+ supports `"snapshot"` and it is the strongest isolation level, +so it remains the default. Some MongoDB-wire-compatible servers (e.g. +DocumentDB) do not implement `readConcern: snapshot` and reject transactions +that request it with `CommandNotSupported` (error code 115); set this to +`"majority"` or `"local"` to run against such targets. Doing so weakens +isolation between concurrent transactions relative to `"snapshot"` — see +§5.1 for what that trades away. + **Read preference:** `primary` only. `MongoEngine::new` rejects connection strings that request `secondary`, `secondaryPreferred`, `primaryPreferred`, or `nearest` — DynamoDB's `ConsistentRead=true` contract requires linearizable reads, which only @@ -427,7 +439,8 @@ drops the collection. `PutItem`, `DeleteItem`, and `UpdateItem` — when they carry a `ConditionExpression`, a `StreamCapture`, or write to a table with GSIs — run inside a MongoDB client session bound to a multi-document transaction -with snapshot read concern and majority write concern. Within the session: +with the configured `transaction_read_concern` (default `"snapshot"`) and +majority write concern. Within the session: 1. `find_one` the current document. 2. Evaluate the DynamoDB condition in Rust @@ -1101,6 +1114,7 @@ insert per write, inside the base write's session. `(pk, sk_?, base_pk, base_sk_?)` for index queries. `GetRecords` uses the compound `(shard_id, sequence_number)` index. -**TransactWriteItems.** Multi-collection ACID transaction with -snapshot read concern and majority write concern; up to 100 operations +**TransactWriteItems.** Multi-collection ACID transaction with the +configured `transaction_read_concern` (default `"snapshot"`) and +majority write concern; up to 100 operations per the DDB spec. Retried on transient conflicts with jittered backoff. diff --git a/extenddb.sample.toml b/extenddb.sample.toml index eaf6cbe2..f493bffe 100755 --- a/extenddb.sample.toml +++ b/extenddb.sample.toml @@ -87,6 +87,17 @@ # MongoDB connection string. Requires a replica set (even single-node). # connection_string = "mongodb://localhost:27017/?replicaSet=rs0" # max_pool_size = 20 # Maximum concurrent connections to MongoDB. +# transaction_read_concern = "snapshot" + # Read concern for the multi-document transactions + # backing conditional writes, TransactWriteItems, + # TransactGetItems, and idempotency tokens. Defaults + # to "snapshot" (real MongoDB's strongest isolation). + # Some MongoDB-wire-compatible backends (e.g. + # DocumentDB) reject "snapshot" transactions with + # CommandNotSupported (115); set this to "majority" + # or "local" to run against those targets. Weakens + # isolation between concurrent transactions — only + # change this if the target server requires it. [auth] # provider = "builtin" # Auth provider: