feat(mongodb): make transaction read concern configurable

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
     }
 }

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<mongodb::options::ReadConcern, String> {
+    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

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)

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::<config::MongoStorageConfig>()
+        .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, &region, max_connections)
-            .await
-            .map_err(|e| BackendError::ConnectionFailed {
-                backend: "mongodb".to_string(),
-                details: e.to_string(),
-            })?;
+        let engine = MongoEngine::new(
+            &connection_string,
+            &region,
+            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<String, (bool, std::time::Instant)>,
+    /// 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<Self, StorageError> {
         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

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:

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<ServerComponents, BackendError>> {

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<Box<dyn Future<Output = Result<ServerComponents, BackendError>> + 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<ServerComponents, BackendError> {

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.

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: