Store U+0000 in strings on PostgreSQL and MongoDB

crates/app/src/cmd_migrate.rs

@@ -16,6 +16,8 @@ use extenddb_config as config;
 use extenddb_storage::error::StorageError;
 use extenddb_storage::management_store::OpError;
 
+use crate::migrate_connection::with_migrate_application_name;
+
 /// Exit status of a preview (`migrate` without `--yes`) that found work to do.
 /// Distinct from 1 so a script can tell "pending" from "failed".
 pub const EXIT_PENDING: u8 = 3;
@@ -114,7 +116,8 @@ pub async fn run(args: MigrateArgs) -> anyhow::Result<u8> {
     // Serialize concurrent migrators, such as several replicas starting at
     // once, so they don't race each other applying schema changes.
     bootstrap.acquire_migration_lock().await.map_err(op_err)?;
-    let result = match CatalogResourceTagStep::open(app_config.storage.connection_config()).await {
+    let catalog_connection = with_migrate_application_name(app_config.storage.connection_config());
+    let result = match CatalogResourceTagStep::open(&catalog_connection).await {
         Ok(rtag) => apply_migrations(bootstrap.as_ref(), &args, &rtag).await,
         Err(e) => Err(e),
     };
@@ -302,6 +305,15 @@ async fn apply_migrations(
         return Ok(EXIT_PENDING);
     }
 
+    // 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().await.map_err(op_err)?;
     }

crates/app/src/lib.rs

@@ -36,6 +36,7 @@ mod cmd_verify;
 mod init_helpers;
 mod manage_http;
 mod manage_types;
+mod migrate_connection;
 mod serve_helpers;
 mod util;
 

crates/app/src/migrate_connection.rs

@@ -0,0 +1,86 @@
+// SPDX-License-Identifier: Apache-2.0
+
+//! The `application_name` the migrate command sets on the PostgreSQL sessions
+//! it opens through the configured connection string.
+
+/// `application_name` the migrate command carries on every PostgreSQL session
+/// it opens. The exclusive-access check for `004_escape_control_chars`
+/// (`extenddb-storage-postgres`, `migrations::MIGRATE_APPLICATION_NAME`) counts
+/// every session on the catalog and data databases except those with this
+/// name. The bootstrapper sets it on its own pools; the catalog stores opened
+/// by `CatalogResourceTagStep` connect through the configured connection
+/// string and would otherwise be counted as foreign sessions, which makes
+/// `migrate --yes` refuse every upgrade from a release without that migration.
+const MIGRATE_APPLICATION_NAME: &str = "extenddb-migrate";
+
+/// `connection` with its `application_name` set to `extenddb-migrate` when it is
+/// a PostgreSQL URL. An `application_name` already in the query string is
+/// replaced: the exclusive-access check skips only sessions with this exact
+/// name, so a URL that keeps an operator's name would make `migrate --yes`
+/// count its own sessions as foreign and refuse. Other backends' connection
+/// strings are returned unchanged.
+pub(crate) fn with_migrate_application_name(connection: &str) -> String {
+    let is_postgres =
+        connection.starts_with("postgres://") || connection.starts_with("postgresql://");
+    if !is_postgres {
+        return connection.to_owned();
+    }
+    let (base, query) = connection.split_once('?').unwrap_or((connection, ""));
+    let mut params: Vec<&str> = query
+        .split('&')
+        .filter(|p| !p.is_empty() && !p.starts_with("application_name="))
+        .collect();
+    let ours = format!("application_name={MIGRATE_APPLICATION_NAME}");
+    params.push(&ours);
+    format!("{base}?{}", params.join("&"))
+}
+
+#[cfg(test)]
+mod application_name_tests {
+    use super::with_migrate_application_name;
+
+    #[test]
+    fn a_postgres_url_without_a_query_string_gets_the_name() {
+        assert_eq!(
+            with_migrate_application_name("postgresql://app:pw@db:5432/extenddb_catalog"),
+            "postgresql://app:pw@db:5432/extenddb_catalog?application_name=extenddb-migrate"
+        );
+        assert_eq!(
+            with_migrate_application_name("postgres://db/cat"),
+            "postgres://db/cat?application_name=extenddb-migrate"
+        );
+    }
+
+    #[test]
+    fn a_postgres_url_with_a_query_string_gets_the_name_appended() {
+        assert_eq!(
+            with_migrate_application_name("postgresql://db/cat?sslmode=require"),
+            "postgresql://db/cat?sslmode=require&application_name=extenddb-migrate"
+        );
+    }
+
+    #[test]
+    fn an_existing_application_name_is_replaced() {
+        assert_eq!(
+            with_migrate_application_name("postgresql://db/cat?application_name=ops"),
+            "postgresql://db/cat?application_name=extenddb-migrate"
+        );
+        assert_eq!(
+            with_migrate_application_name(
+                "postgresql://db/cat?sslmode=require&application_name=ops&connect_timeout=5"
+            ),
+            "postgresql://db/cat?sslmode=require&connect_timeout=5&application_name=extenddb-migrate"
+        );
+    }
+
+    #[test]
+    fn other_backends_are_unchanged() {
+        for given in [
+            "/var/lib/extenddb/extenddb.sqlite",
+            "mongodb://127.0.0.1:27017/?replicaSet=rs0",
+            "cassandra://127.0.0.1:9042",
+        ] {
+            assert_eq!(with_migrate_application_name(given), given);
+        }
+    }
+}

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