Store U+0000 in strings on PostgreSQL and MongoDB

CONTRIBUTING.md

@@ -341,26 +341,26 @@ SQLite, PostgreSQL, Cassandra and MongoDB migrate through this framework.
 
 ### Example: a nullable column on `tables`
 
-Migration 6, `table_note`, adds a nullable `note TEXT` column to the catalog's
+Migration 7, `table_note`, adds a nullable `note TEXT` column to the catalog's
 `tables` table on SQLite and PostgreSQL.
 
 In `crates/storage/src/migrations/registry.rs`, at the end of the
 `migration_ids!` list:
 
 ```rust
     /// Adds a nullable `note` column to the `tables` catalog table.
-    TableNote = 6, "table_note",
+    TableNote = 7, "table_note",
         "add a nullable note column to the tables catalog";
 ```
 
 PostgreSQL has `ADD COLUMN IF NOT EXISTS`, so its body is a file,
-`crates/storage-postgres/migrations/0006_table_note.sql`:
+`crates/storage-postgres/migrations/0007_table_note.sql`:
 
 ```sql
 -- Copyright 2026 ExtendDB contributors
 -- SPDX-License-Identifier: Apache-2.0
 --
--- Migration 6, table_note, PostgreSQL catalog database: a nullable note
+-- Migration 7, table_note, PostgreSQL catalog database: a nullable note
 -- column on tables. IF NOT EXISTS makes a second run change nothing.
 
 ALTER TABLE tables ADD COLUMN IF NOT EXISTS note TEXT;
@@ -375,7 +375,7 @@ In `crates/storage-postgres/src/migrations/mod.rs`, the arm in `migration()`:
             bodies: vec![file_body!(
                 StoreKind::Catalog,
                 "migrations",
-                "0006_table_note.sql"
+                "0007_table_note.sql"
             )],
         },
 ```
@@ -393,13 +393,13 @@ and the id in the transactional arm of `runs_in()`:
 SQLite has no `ADD COLUMN IF NOT EXISTS`, and a file with a plain
 `ADD COLUMN` fails when it runs a second time, so the SQLite body is code
 that checks for the column first.
-`crates/storage-sqlite/src/migrations/m0006_table_note.rs`:
+`crates/storage-sqlite/src/migrations/m0007_table_note.rs`:
 
 ```rust
 // Copyright 2026 ExtendDB contributors
 // SPDX-License-Identifier: Apache-2.0
 
-//! Migration 6, `table_note`, SQLite: a nullable `note` column on `tables`.
+//! Migration 7, `table_note`, SQLite: a nullable `note` column on `tables`.
 //!
 //! SQLite has no `ADD COLUMN IF NOT EXISTS`, so the body reads the table's
 //! columns first and adds the column only when it is missing. A second run
@@ -411,10 +411,10 @@ use extenddb_storage::migrations::{CodeBody, Done, MigrationError};
 use super::SqliteStores;
 
 /// This file, whose SHA-256 the checksum test pins.
-pub(crate) const SOURCE: &str = include_str!("m0006_table_note.rs");
+pub(crate) const SOURCE: &str = include_str!("m0007_table_note.rs");
 
 /// Recorded as this migration's checksum. Changes only with a new migration.
-const FINGERPRINT: &str = "0006_table_note/sqlite/1";
+const FINGERPRINT: &str = "0007_table_note/sqlite/1";
 
 pub(crate) struct TableNote;
 
@@ -453,7 +453,7 @@ beside the others:
 ```rust
 #[rustfmt::skip]
 #[allow(clippy::all, deprecated)]
-mod m0006_table_note;
+mod m0007_table_note;
 ```
 
 and the arm in `migration()`:
@@ -464,8 +464,8 @@ and the arm in `migration()`:
             convergent: true,
             bodies: vec![StoreBody::code(
                 StoreKind::Catalog,
-                &m0006_table_note::TableNote,
-                m0006_table_note::SOURCE,
+                &m0007_table_note::TableNote,
+                m0007_table_note::SOURCE,
             )],
         },
 ```
@@ -477,8 +477,8 @@ rows of `tables`:
 
 ```rust
 #[tokio::test]
-async fn migration_6_adds_the_note_column_to_tables() {
-    let db = temp_db("m6");
+async fn migration_7_adds_the_note_column_to_tables() {
+    let db = temp_db("m7");
     load_dump(&db, DEV_0_1_14).await;
     run(&db).await.expect("migrate");
     let mut text = dump(
@@ -488,12 +488,12 @@ async fn migration_6_adds_the_note_column_to_tables() {
     )
     .await;
     text.push_str(&dump(&db, "SELECT table_name, note FROM tables ORDER BY table_name").await);
-    golden("0006_table_note", &text);
+    golden("0007_table_note", &text);
 }
 ```
 
-`EXTENDDB_BLESS_EXPECTED=1 cargo test -p extenddb-storage-sqlite --lib migration_6`
-writes `crates/storage-sqlite/tests/fixtures/golden/0006_table_note.txt`, one
+`EXTENDDB_BLESS_EXPECTED=1 cargo test -p extenddb-storage-sqlite --lib migration_7`
+writes `crates/storage-sqlite/tests/fixtures/golden/0007_table_note.txt`, one
 tab-separated line per row under each query:
 
 ```
@@ -508,15 +508,15 @@ The PostgreSQL body is a file, so it has no golden test.
 The example declares `Kind::Breaking`. A nullable column can be additive by
 the rule in step 4, but declaring it additive needs the old-binary job, which
 this example does not add. As a breaking migration it makes a 1.x binary
-without migration 6 refuse a catalog that has it, and a rollback restores the
+without migration 7 refuse a catalog that has it, and a rollback restores the
 backup.
 
 The same PR changes the code that reads and writes `tables.note` in each
 backend's catalog store, and commits the `docs/schema/sqlite.txt` and
 `docs/schema/postgres.txt` that the integration jobs regenerate. It changes
 no per-table DDL, because `tables` is a catalog table, and adds no `RELEASED`
-entry. `extenddb migrate --list` then shows migration 6, and on a catalog at
-migration 5 `extenddb migrate` lists migration 6 as pending and exits 3.
+entry. `extenddb migrate --list` then shows migration 7, and on a catalog at
+migration 6 `extenddb migrate` lists migration 7 as pending and exits 3.
 
 ### Running the migration tests
 

crates/storage-cassandra/src/migrations/mod.rs

@@ -127,6 +127,10 @@ pub fn migration(id: MigrationId) -> Migration<CassandraStores> {
                 "0005_drop_continuous_backups.cql"
             )],
         },
+        MigrationId::EscapeControlChars => Migration::NotApplicable(
+            "Cassandra stores U+0000 in strings as written, so its stored form of strings does \
+             not change",
+        ),
     }
 }
 

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

@@ -15,7 +15,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 {
@@ -34,6 +36,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 }`
@@ -44,10 +93,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();
 
@@ -194,9 +242,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)
@@ -256,8 +302,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}")))?;
@@ -436,6 +481,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