Store U+0000 in strings on PostgreSQL and MongoDB

crates/app/src/cmd_migrate.rs

@@ -170,6 +170,15 @@ async fn apply_migrations(
         );
     }
 
+    // A migration that rewrites rows in place needs every other session off
+    // the databases it touches. Check before the first write, so a refusal
+    // leaves the catalog version where it was and an older binary can still be
+    // restarted.
+    bootstrap
+        .check_exclusive_access_for_pending_migrations()
+        .await
+        .map_err(|e| anyhow::anyhow!("{e:?}"))?;
+
     if catalog_pending {
         bootstrap
             .run_catalog_migrations()

crates/storage-mongodb/src/data/mod.rs

@@ -13,7 +13,9 @@ use extenddb_core::types::{
     AttributeDefinition, AttributeValue, Item, KeySchemaElement, ScalarAttributeType,
 };
 use extenddb_storage::error::StorageError;
-use extenddb_storage::util::{composite_pk_to_text, encode_netstring_composite, sk_info};
+use extenddb_storage::util::{
+    composite_pk_to_text, encode_netstring_composite, escape_json_keys, sk_info, unescape_json_keys,
+};
 
 /// Returns the `MongoDB` collection name for a `DynamoDB` table.
 pub fn data_collection_name(table_id: &str) -> String {
@@ -32,6 +34,53 @@ pub fn composite_id(pk_text: &str, sk_text: &str) -> String {
     encode_netstring_composite(&[pk_text.to_owned(), sk_text.to_owned()])
 }
 
+/// Convert an item (base item or projected index item) to the BSON stored
+/// under `item_data`.
+///
+/// BSON field names are C strings and cannot contain U+0000, while `DynamoDB`
+/// accepts it in attribute names and map keys. Every object key in the item's
+/// JSON form is run through the shared order-preserving escape (U+0000 and
+/// U+0001 become two-character sequences, see
+/// `extenddb_storage::util::escape_control`) before the BSON conversion, so
+/// the stored document never carries a NUL field name. String values are left
+/// alone; BSON strings are length-prefixed and hold U+0000 verbatim.
+///
+/// Every conversion of an item to `item_data` BSON must go through this
+/// function or [`json_to_item_data`]; a direct `bson::to_bson` on item JSON
+/// fails in the driver at write time when a name contains U+0000.
+/// [`item_data_to_json`] reverses the escape on read.
+pub(crate) fn item_to_item_data(item: &Item) -> Result<bson::Bson, StorageError> {
+    let item_json =
+        serde_json::to_value(item).map_err(|e| StorageError::Internal(e.to_string()))?;
+    json_to_item_data(item_json)
+}
+
+/// JSON-tree variant of [`item_to_item_data`], for payloads that carry items
+/// nested inside another structure (stream records with key/image maps).
+pub(crate) fn json_to_item_data(json: serde_json::Value) -> Result<bson::Bson, StorageError> {
+    let escaped = escape_json_keys(json);
+    bson::to_bson(&escaped).map_err(|e| StorageError::Internal(e.to_string()))
+}
+
+/// Reverse of [`json_to_item_data`]: BSON read from storage back to JSON with
+/// the object-key escape undone.
+///
+/// Documents written by a release before the escape hold their field names
+/// raw, and there is no migration and no startup check for them: unlike
+/// `PostgreSQL`, nothing here can fail to store (BSON strings already hold
+/// U+0000, and a raw field name reads back through the driver), so the only
+/// effect is on a legacy name that happens to contain U+0001 immediately
+/// followed by U+0001 or U+0002, which this decoder reads as U+0000 or
+/// U+0001. A lone U+0001 and every string value are unchanged. That exposure
+/// was judged too narrow to make every `MongoDB` upgrade acknowledge a gate
+/// with nothing to rewrite; the upgrade manual's "Data Migrations" section
+/// states the limitation for operators.
+pub(crate) fn item_data_to_json(value: &bson::Bson) -> Result<serde_json::Value, StorageError> {
+    let json: serde_json::Value = bson::from_bson(value.clone())
+        .map_err(|e| StorageError::Internal(format!("BSON to JSON conversion error: {e}")))?;
+    Ok(unescape_json_keys(json))
+}
+
 /// Convert a `DynamoDB` Item to a `MongoDB` BSON document for storage.
 ///
 /// Document structure: `{ _id, pk, sk_s/sk_n/sk_b, item_data }`
@@ -42,10 +91,9 @@ pub fn item_to_document(
 ) -> Result<Document, StorageError> {
     let pk_text = composite_pk_to_text(item, key_schema)?;
 
-    // Serialize the full item as item_data
-    let item_json =
-        serde_json::to_value(item).map_err(|e| StorageError::Internal(e.to_string()))?;
-    let item_bson = bson::to_bson(&item_json).map_err(|e| StorageError::Internal(e.to_string()))?;
+    // Serialize the full item as item_data (object keys escaped, see
+    // item_to_item_data).
+    let item_bson = item_to_item_data(item)?;
 
     let mut doc = Document::new();
 
@@ -192,9 +240,7 @@ pub fn index_document(
         insert_typed_sk(&mut doc, &field, sk_type, sk_value)?;
     }
 
-    let item_json =
-        serde_json::to_value(projected).map_err(|e| StorageError::Internal(e.to_string()))?;
-    let item_bson = bson::to_bson(&item_json).map_err(|e| StorageError::Internal(e.to_string()))?;
+    let item_bson = item_to_item_data(projected)?;
     doc.insert("item_data", item_bson);
 
     Ok(doc)
@@ -254,8 +300,7 @@ pub fn document_to_item(doc: &Document) -> Result<Item, StorageError> {
         .get("item_data")
         .ok_or_else(|| StorageError::Internal("Document missing item_data field".to_string()))?;
 
-    let json_value: serde_json::Value = bson::from_bson(item_data.clone())
-        .map_err(|e| StorageError::Internal(format!("BSON to JSON conversion error: {e}")))?;
+    let json_value = item_data_to_json(item_data)?;
 
     let item: Item = serde_json::from_value(json_value)
         .map_err(|e| StorageError::Internal(format!("JSON to Item conversion error: {e}")))?;
@@ -434,6 +479,89 @@ mod tests {
         assert!(matches!(err, StorageError::Validation(_)));
     }
 
+    /// Walk a BSON tree and assert no document field name contains U+0000.
+    fn assert_no_nul_field_names(value: &bson::Bson) {
+        match value {
+            bson::Bson::Document(doc) => {
+                for (k, v) in doc {
+                    assert!(!k.contains('\u{0}'), "field name contains NUL: {k:?}");
+                    assert_no_nul_field_names(v);
+                }
+            }
+            bson::Bson::Array(items) => {
+                for item in items {
+                    assert_no_nul_field_names(item);
+                }
+            }
+            _ => {}
+        }
+    }
+
+    #[test]
+    fn item_document_round_trips_control_characters_in_names() {
+        use std::collections::BTreeMap;
+
+        let (schema, attrs) = schema_pk_str_sk_num();
+
+        let mut inner = BTreeMap::new();
+        inner.insert(
+            "deep\u{0}key".to_owned(),
+            AttributeValue::S("v\u{0}".to_owned()),
+        );
+        let mut mid = BTreeMap::new();
+        mid.insert("mid\u{1}key".to_owned(), AttributeValue::M(inner));
+        let mut in_list = BTreeMap::new();
+        in_list.insert(
+            "list\u{0}\u{1}key".to_owned(),
+            AttributeValue::S("w".to_owned()),
+        );
+
+        let mut item = Item::new();
+        item.insert("pk".to_owned(), AttributeValue::S("p".to_owned()));
+        item.insert("sk".to_owned(), AttributeValue::N("1".to_owned()));
+        item.insert(
+            "top\u{0}\u{1}name".to_owned(),
+            AttributeValue::S("x".to_owned()),
+        );
+        item.insert("map".to_owned(), AttributeValue::M(mid));
+        item.insert(
+            "list".to_owned(),
+            AttributeValue::L(vec![AttributeValue::M(in_list)]),
+        );
+
+        let doc = item_to_document(&item, &schema, &attrs).unwrap();
+        assert_no_nul_field_names(&bson::Bson::Document(doc.clone()));
+
+        let back = document_to_item(&doc).unwrap();
+        assert_eq!(back, item);
+    }
+
+    #[test]
+    fn plain_item_document_stores_raw_field_names() {
+        use std::collections::BTreeMap;
+
+        let (schema, attrs) = schema_pk_str_sk_num();
+
+        let mut inner = BTreeMap::new();
+        inner.insert("k".to_owned(), AttributeValue::S("v".to_owned()));
+        let mut item = Item::new();
+        item.insert("pk".to_owned(), AttributeValue::S("p".to_owned()));
+        item.insert("sk".to_owned(), AttributeValue::N("1".to_owned()));
+        item.insert("attr".to_owned(), AttributeValue::M(inner));
+
+        let doc = item_to_document(&item, &schema, &attrs).unwrap();
+        let item_data = doc.get_document("item_data").unwrap();
+        let names: Vec<&str> = item_data.keys().map(String::as_str).collect();
+        assert_eq!(names, vec!["attr", "pk", "sk"]);
+        let attr_m = item_data
+            .get_document("attr")
+            .unwrap()
+            .get_document("M")
+            .unwrap();
+        let map_keys: Vec<&str> = attr_m.keys().map(String::as_str).collect();
+        assert_eq!(map_keys, vec!["k"]);
+    }
+
     #[test]
     fn composite_id_disambiguates_delimiter_in_pk_or_sk() {
         // Two items whose naive "{pk}#{sk}" strings would collide must

crates/storage-mongodb/src/data_engine.rs

@@ -18,7 +18,7 @@ use extenddb_core::types::{
 };
 use extenddb_storage::error::StorageError;
 use extenddb_storage::util::{
-    composite_pk_to_text, encode_netstring_composite, pk_to_text, sk_info,
+    composite_pk_to_text, encode_netstring_composite, needs_escape, pk_to_text, sk_info,
 };
 use extenddb_storage::{
     DataEngine, IdempotencyKey, ItemPairResult, QueryResult, StreamCapture, TransactGetOp,
@@ -29,16 +29,17 @@ use crate::MongoEngine;
 use crate::condition::condition_to_filter;
 use crate::data::{
     binary_sk_to_hex, composite_id, data_collection_name, document_to_item, index_document,
-    index_entry_filter, item_to_document, pk_filter, sk_field_name, sk_suffix,
+    index_entry_filter, item_to_document, json_to_item_data, pk_filter, sk_field_name, sk_suffix,
 };
 use crate::pushdown::{Pushable, is_pushable};
 
 use extenddb_core::types::{Projection, ProjectionType};
 
 /// Resolve a single-component update path for the native MongoDB fast path.
 /// Literal attribute names containing dots or beginning with `$` cannot be
-/// represented safely by MongoDB's ordinary update paths, so they must use
-/// the Rust/session path.
+/// represented safely by MongoDB's ordinary update paths, and names
+/// containing U+0000 or U+0001 differ from the stored (escaped) field name,
+/// so they must use the Rust/session path.
 fn native_attribute_name(path: &[PathElement], maps: &ExpressionMaps) -> Option<String> {
     if path.len() != 1 {
         return None;
@@ -47,7 +48,8 @@ fn native_attribute_name(path: &[PathElement], maps: &ExpressionMaps) -> Option<
         return None;
     };
     let attr_name = resolve_name_ref(raw_name, maps).ok()?;
-    (!attr_name.contains('.') && !attr_name.starts_with('$')).then(|| attr_name.into_owned())
+    (!attr_name.contains('.') && !attr_name.starts_with('$') && !needs_escape(&attr_name))
+        .then(|| attr_name.into_owned())
 }
 
 #[derive(Clone, Copy)]
@@ -1654,8 +1656,11 @@ impl MongoEngine {
                         _ => return None, // complex expressions (if_not_exists, list_append, arithmetic)
                     };
                     let field = format!("item_data.{attr_name}");
+                    // The value lands inside item_data, so its nested map
+                    // keys must be in stored (escaped) form, matching what
+                    // item_to_document writes.
                     let val_json = serde_json::to_value(val).ok()?;
-                    let val_bson = bson::to_bson(&val_json).ok()?;
+                    let val_bson = json_to_item_data(val_json).ok()?;
                     set_doc.insert(field, val_bson);
                 }
                 UpdateAction::Remove { path } => {
@@ -2036,8 +2041,10 @@ impl MongoEngine {
 
         let record_json =
             serde_json::to_value(&record).map_err(|e| StorageError::Internal(e.to_string()))?;
-        let record_bson =
-            bson::to_bson(&record_json).map_err(|e| StorageError::Internal(e.to_string()))?;
+        // The record's key and image maps carry attribute names as object
+        // keys, which become BSON field names; convert through the same
+        // escaping helper as item_data.
+        let record_bson = json_to_item_data(record_json)?;
 
         // key_info already carries table_id — no need to re-read the catalog
         // just to resolve it, and re-reading inside the session against the
@@ -4388,6 +4395,21 @@ transaction_read_concern = "majority""#,
         assert_eq!(native_attribute_name(&path, &maps), None);
     }
 
+    #[test]
+    fn native_update_rejects_control_character_literal_attribute_names() {
+        // Stored field names carry the escaped form of U+0000 and U+0001,
+        // so a native update path built from the raw name would target a
+        // different field; these names must use the Rust/session path.
+        for name in ["a\u{0}b", "a\u{1}b"] {
+            let mut names = std::collections::HashMap::new();
+            names.insert("name".to_owned(), name.to_owned());
+            let maps = ExpressionMaps::new(names, std::collections::HashMap::new());
+            let path = vec![PathElement::Attribute("#name".to_owned())];
+
+            assert_eq!(native_attribute_name(&path, &maps), None, "{name:?}");
+        }
+    }
+
     #[test]
     fn native_update_keeps_simple_attribute_names_on_fast_path() {
         let maps = ExpressionMaps::new(

crates/storage-mongodb/src/metadata_engine.rs

@@ -18,6 +18,7 @@ use extenddb_storage::error::StorageError;
 
 use crate::MongoEngine;
 use crate::data::{data_collection_name, document_to_item};
+use extenddb_storage::util::escape_control;
 
 fn extract_id_fields(doc: &Document) -> (String, String) {
     let id = doc.get_document("_id").ok();
@@ -35,13 +36,15 @@ fn extract_id_fields(doc: &Document) -> (String, String) {
 /// Build the expression that reads a DynamoDB attribute from `item_data` as a
 /// literal field name. MongoDB normally interprets dots in a query field path
 /// as nested-document traversal, but DynamoDB permits dots in attribute names.
+/// The stored field name is the escaped form of the attribute name (see
+/// `extenddb_storage::util::escape_control`), so the lookup targets that form.
 fn literal_ttl_value_expression(ttl_attribute: &str) -> Document {
     doc! {
         "$getField": {
             "field": "N",
             "input": {
                 "$getField": {
-                    "field": ttl_attribute,
+                    "field": escape_control(ttl_attribute).as_ref(),
                     "input": "$item_data",
                 }
             }
@@ -69,7 +72,7 @@ fn literal_ttl_epoch_expression(ttl_attribute: &str) -> Document {
 /// also performs the expiry comparison in MongoDB, so non-expired low-`_id`
 /// documents cannot starve expired documents later in the collection.
 fn ttl_candidate_query(ttl_attribute: &str, now_epoch: i64) -> (Document, Document) {
-    let ttl_field = format!("item_data.{ttl_attribute}.N");
+    let ttl_field = format!("item_data.{}.N", escape_control(ttl_attribute));
     if ttl_attribute.contains('.') {
         let ttl_epoch = literal_ttl_epoch_expression(ttl_attribute);
         (
@@ -374,8 +377,12 @@ impl MetadataEngine for MongoEngine {
             let coll_name = data_collection_name(table_id);
             let data_coll = self.data_db.collection::<Document>(&coll_name);
 
-            let index_name = format!("idx_ttl_{ttl_attribute}");
-            let index_key = format!("item_data.{ttl_attribute}.N");
+            // The stored field name is the escaped attribute name, and the
+            // escape also keeps U+0000 out of the index name. Names without
+            // U+0000 or U+0001 escape to themselves, so existing index names
+            // do not change.
+            let index_name = format!("idx_ttl_{}", escape_control(&ttl_attribute));
+            let index_key = format!("item_data.{}.N", escape_control(&ttl_attribute));
 
             let index = IndexModel::builder()
                 .keys(doc! { &index_key: 1 })
@@ -433,7 +440,9 @@ impl MetadataEngine for MongoEngine {
                 .get_str("table_id")
                 .map_err(|_| StorageError::Internal("missing table_id".to_string()))?;
 
-            let index_name = format!("idx_ttl_{ttl_attribute}");
+            // Same escaped form as create_ttl_index so the drop targets the
+            // index that was created.
+            let index_name = format!("idx_ttl_{}", escape_control(&ttl_attribute));
 
             let coll_name = data_collection_name(table_id);
             let data_coll = self.data_db.collection::<Document>(&coll_name);
@@ -513,9 +522,11 @@ impl MetadataEngine for MongoEngine {
                 if items.len() >= limit {
                     break;
                 }
-                // Parse the TTL value and check if expired
+                // Parse the TTL value and check if expired. The stored field
+                // name is the escaped attribute name.
                 if let Ok(item_data) = doc.get_document("item_data")
-                    && let Ok(ttl_obj) = item_data.get_document(&ttl_attribute)
+                    && let Ok(ttl_obj) =
+                        item_data.get_document(escape_control(&ttl_attribute).as_ref())
                     && let Ok(n_str) = ttl_obj.get_str("N")
                     && let Ok(ttl_val) = n_str.parse::<i64>()
                     && ttl_val >= 1