fix(storage-postgres): store attribute names containing U+0000 in the catalog

.github/workflows/integration.yml

@@ -516,6 +516,7 @@ jobs:
           cargo test --release -p extenddb-storage-postgres
           --test migrations_fresh --test migrations_released --test migrations_crash
           --test migrations_golden --test migrations_guards --test migrations_runs
+          --test migrations_name_escape
 
       # The daemonized server logs to syslog; dump it so server-side failures
       # are diagnosable from the job log.

CONTRIBUTING.md

@@ -341,27 +341,27 @@ SQLite, PostgreSQL, Cassandra and MongoDB migrate through this framework.
 
 ### Example: a nullable column on `tables`
 
-Migration 7, `table_note` (7 being the next free id at the time of writing;
-migration 6 is `backup_catalog_index`), adds a nullable `note TEXT` column to
+Migration 8, `table_note` (8 being the next free id at the time of writing;
+migration 7 is `attribute_name_escape`), 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 = 7, "table_note",
+    TableNote = 8, "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/0007_table_note.sql`:
+`crates/storage-postgres/migrations/0008_table_note.sql`:
 
 ```sql
 -- Copyright 2026 ExtendDB contributors
 -- SPDX-License-Identifier: Apache-2.0
 --
--- Migration 7, table_note, PostgreSQL catalog database: a nullable note
+-- Migration 8, 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;
@@ -376,7 +376,7 @@ In `crates/storage-postgres/src/migrations/mod.rs`, the arm in `migration()`:
             bodies: vec![file_body!(
                 StoreKind::Catalog,
                 "migrations",
-                "0007_table_note.sql"
+                "0008_table_note.sql"
             )],
         },
 ```
@@ -389,19 +389,20 @@ and the id in the transactional arm of `runs_in()`:
         | MigrationId::SortKeyDefinitions
         | MigrationId::DropContinuousBackups
         | MigrationId::BackupCatalogIndex
+        | MigrationId::AttributeNameEscape
         | MigrationId::TableNote => RunsIn::Transaction,
 ```
 
 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/m0007_table_note.rs`:
+`crates/storage-sqlite/src/migrations/m0008_table_note.rs`:
 
 ```rust
 // Copyright 2026 ExtendDB contributors
 // SPDX-License-Identifier: Apache-2.0
 
-//! Migration 7, `table_note`, SQLite: a nullable `note` column on `tables`.
+//! Migration 8, `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
@@ -413,10 +414,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!("m0007_table_note.rs");
+pub(crate) const SOURCE: &str = include_str!("m0008_table_note.rs");
 
 /// Recorded as this migration's checksum. Changes only with a new migration.
-const FINGERPRINT: &str = "0007_table_note/sqlite/1";
+const FINGERPRINT: &str = "0008_table_note/sqlite/1";
 
 pub(crate) struct TableNote;
 
@@ -455,7 +456,7 @@ beside the others:
 ```rust
 #[rustfmt::skip]
 #[allow(clippy::all, deprecated)]
-mod m0007_table_note;
+mod m0008_table_note;
 ```
 
 and the arm in `migration()`:
@@ -466,8 +467,8 @@ and the arm in `migration()`:
             convergent: true,
             bodies: vec![StoreBody::code(
                 StoreKind::Catalog,
-                &m0007_table_note::TableNote,
-                m0007_table_note::SOURCE,
+                &m0008_table_note::TableNote,
+                m0008_table_note::SOURCE,
             )],
         },
 ```
@@ -479,8 +480,8 @@ rows of `tables`:
 
 ```rust
 #[tokio::test]
-async fn migration_7_adds_the_note_column_to_tables() {
-    let db = temp_db("m7");
+async fn migration_8_adds_the_note_column_to_tables() {
+    let db = temp_db("m8");
     load_dump(&db, DEV_0_1_14).await;
     run(&db).await.expect("migrate");
     let mut text = dump(
@@ -490,12 +491,12 @@ async fn migration_7_adds_the_note_column_to_tables() {
     )
     .await;
     text.push_str(&dump(&db, "SELECT table_name, note FROM tables ORDER BY table_name").await);
-    golden("0007_table_note", &text);
+    golden("0008_table_note", &text);
 }
 ```
 
-`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
+`EXTENDDB_BLESS_EXPECTED=1 cargo test -p extenddb-storage-sqlite --lib migration_8`
+writes `crates/storage-sqlite/tests/fixtures/golden/0008_table_note.txt`, one
 tab-separated line per row under each query:
 
 ```
@@ -510,15 +511,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 7 refuse a catalog that has it, and a rollback restores the
+without migration 8 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 7, and on a catalog at
-migration 6 `extenddb migrate` lists migration 7 as pending and exits 3.
+entry. `extenddb migrate --list` then shows migration 8, and on a catalog at
+migration 7 `extenddb migrate` lists migration 8 as pending and exits 3.
 
 ### Running the migration tests
 

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

@@ -150,6 +150,11 @@ pub fn migration(id: MigrationId) -> Migration<CassandraStores> {
                 m0006_backup_catalog_index::SOURCE,
             )],
         },
+        MigrationId::AttributeNameEscape => Migration::NotApplicable(
+            "Cassandra stores the documents that carry attribute names as JSON text in text \
+             columns, where U+0000 is the six-character escape \\u0000, so every name reads back \
+             as it was written",
+        ),
     }
 }
 

crates/storage-cassandra/tests/direct/backup_engine.rs

@@ -4164,12 +4164,17 @@ async fn test_a_migration_6_that_fails_in_its_backfill_stays_pending_and_resumes
     let (engine, account_id, _drop_keyspaces) = isolated_engine("mf", true).await;
     let catalog = engine.catalog_keyspace();
     let prefix = prefix_of(&engine);
+    // Migration 6 is pending, and so is every later migration this binary
+    // carries.
     let behind_at_5 = |status: CatalogStatus| match status {
         CatalogStatus::Behind { version, pending } => {
             version == Some(MigrationId::DropContinuousBackups.number())
                 && pending
                     .iter()
-                    .all(|p| p.id == MigrationId::BackupCatalogIndex.number())
+                    .any(|p| p.id == MigrationId::BackupCatalogIndex.number())
+                && pending
+                    .iter()
+                    .all(|p| p.id >= MigrationId::BackupCatalogIndex.number())
         }
         _ => false,
     };

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

@@ -160,6 +160,10 @@ pub fn migration(id: MigrationId) -> Migration<MongoStores> {
                 m0006_backup_catalog_index::SOURCE,
             )],
         },
+        MigrationId::AttributeNameEscape => Migration::NotApplicable(
+            "MongoDB stores the catalog's attribute names as BSON string values, which carry their \
+             length and so can hold U+0000, and every name reads back as it was written",
+        ),
     }
 }
 

crates/storage-mongodb/tests/migrations_fresh.rs

@@ -322,6 +322,7 @@ async fn a_fresh_store_gets_every_migration_and_the_expected_definitions() {
             (4, d.target.catalog.clone()),
             (5, d.target.catalog.clone()),
             (6, d.target.catalog.clone()),
+            (7, d.target.catalog.clone()),
         ]
     );
     let not_applicable: Vec<u32> = report
@@ -330,7 +331,7 @@ async fn a_fresh_store_gets_every_migration_and_the_expected_definitions() {
         .filter(|s| s.outcome == Outcome::NotApplicable)
         .map(|s| s.id)
         .collect();
-    assert_eq!(not_applicable, [3, 4], "{report:?}");
+    assert_eq!(not_applicable, [3, 4, 7], "{report:?}");
     assert_eq!(
         d.status().await,
         CatalogStatus::Current {
@@ -348,6 +349,7 @@ async fn a_fresh_store_gets_every_migration_and_the_expected_definitions() {
             (4, Outcome::NotApplicable),
             (5, Outcome::Applied),
             (6, Outcome::Applied),
+            (7, Outcome::NotApplicable),
         ]
     );
     assert!(
@@ -442,8 +444,8 @@ async fn a_second_run_applies_nothing_and_reconverge_keeps_the_rows() {
 
     let reconverged = d.reconverge().await.expect("reconverge");
     // Migration 1's steps in the catalog and the data database, and
-    // migrations 5's and 6's in the catalog: 2 is not convergent, and 3 and 4
-    // do not apply on MongoDB.
+    // migrations 5's and 6's in the catalog: 2 is not convergent, and 3, 4 and
+    // 7 do not apply on MongoDB.
     let ran: Vec<(u32, String)> = reconverged
         .applied
         .iter()
@@ -533,19 +535,19 @@ async fn a_changed_checksum_a_newer_catalog_and_an_older_release_are_reported()
         doc! { "_id": id, "name": "later", "kind": kind, "checksum": "c", "outcome": "applied",
         "notes": "", "applied_at": bson::DateTime::now(), "applied_by": "9.9.9" }
     };
-    ledger.insert_one(row(7, "breaking")).await.expect("insert");
+    ledger.insert_one(row(8, "breaking")).await.expect("insert");
     let status = d.status().await;
     assert_eq!(status.label(), "AheadIncompatible");
     assert_eq!(status.exit_code(), exit::AHEAD_INCOMPATIBLE);
     ledger
-        .delete_one(doc! { "_id": 7_i64 })
+        .delete_one(doc! { "_id": 8_i64 })
         .await
         .expect("delete");
-    ledger.insert_one(row(7, "additive")).await.expect("insert");
+    ledger.insert_one(row(8, "additive")).await.expect("insert");
     assert_eq!(d.status().await.label(), "AheadCompatible");
     assert!(d.migrate().await.expect("migrate").applied.is_empty());
     ledger
-        .delete_one(doc! { "_id": 7_i64 })
+        .delete_one(doc! { "_id": 8_i64 })
         .await
         .expect("delete");
 
@@ -626,6 +628,6 @@ async fn a_ledger_document_this_binary_cannot_read_stops_the_status() {
         .expect_err("an error");
     migrations.close().await;
     assert!(error.to_string().contains("\"sideways\""), "{error}");
-    assert_eq!(MigrationId::LATEST.number(), 6);
+    assert_eq!(MigrationId::LATEST.number(), 7);
     d.drop_databases().await;
 }

crates/storage-mongodb/tests/migrations_locks.rs

@@ -53,7 +53,7 @@ async fn two_migrators_started_together_apply_once_and_the_other_finds_nothing()
     applied.sort_unstable();
     assert_eq!(
         applied,
-        [0, 7],
+        [0, 8],
         "one applies every step and the other finds nothing"
     );
     assert_eq!(

crates/storage-mongodb/tests/migrations_released.rs

@@ -161,7 +161,7 @@ async fn every_released_catalog_upgrades_to_the_definitions_of_a_fresh_install()
             panic!("{state} is not behind");
         };
         assert_eq!(version, None, "{state} has no ledger");
-        assert_eq!(pending.len(), 7, "{state}");
+        assert_eq!(pending.len(), 8, "{state}");
         let report = d.migrate().await.expect("migrate");
         assert_eq!(
             (report.from, report.to),

crates/storage-postgres/src/backup_engine.rs

@@ -11,9 +11,9 @@ use std::sync::Arc;
 use std::time::Duration;
 
 use extenddb_core::types::{
-    BackupDescription, BackupDetails, BackupSummary, GsiInput, Item, KeySchemaElement, LsiInput,
-    Projection, RestoreSummary, ScalarAttributeType, SourceTableDetails, StreamSpecification,
-    TableDescription,
+    AttributeDefinition, BackupDescription, BackupDetails, BackupSummary, GsiInput, Item,
+    KeySchemaElement, LsiInput, Projection, RestoreSummary, ScalarAttributeType,
+    SourceTableDetails, StreamSpecification, TableDescription,
 };
 use extenddb_storage::backup_catalog::{
     AbandonedLegacyBackup, JOB_HEARTBEAT_INTERVAL, JobBound, limit_exceeded_message, new_generation,
@@ -31,6 +31,7 @@ use extenddb_storage::{BackupRestoreSource, JobPermit, LegacyBackupReader, Resto
 use futures::future::BoxFuture;
 
 use crate::PostgresEngine;
+use crate::catalog_json::{escape_json, from_catalog_json, to_catalog_json, unescape_json};
 use crate::create_table::RestoreRegistration;
 use crate::data::{
     all_sort_key_info, data_table_name, index_table_name, insert_index_rows_multi,
@@ -366,9 +367,9 @@ impl PostgresEngine {
         let mut lsis = Vec::new();
         let mut gsi_ids = Vec::new();
         for (index_name, index_id, index_type, ks, proj, ipt) in rows {
-            let key_schema: Vec<KeySchemaElement> = serde_json::from_value(ks)
+            let key_schema: Vec<KeySchemaElement> = from_catalog_json(ks)
                 .map_err(|e| StorageError::Internal(format!("index key schema: {e}")))?;
-            let projection: Projection = serde_json::from_value(proj)
+            let projection: Projection = from_catalog_json(proj)
                 .map_err(|e| StorageError::Internal(format!("index projection: {e}")))?;
             if index_type == "LSI" {
                 lsis.push(LsiInput {
@@ -462,9 +463,9 @@ impl PostgresEngine {
         for (index_id, ks, proj) in index_rows {
             indexes.push(RestoreIndex {
                 table: index_table_name(&index_id),
-                key_schema: serde_json::from_value(ks)
+                key_schema: from_catalog_json(ks)
                     .map_err(|e| StorageError::Internal(format!("index key schema: {e}")))?,
-                projection: serde_json::from_value(proj)
+                projection: from_catalog_json(proj)
                     .map_err(|e| StorageError::Internal(format!("index projection: {e}")))?,
             });
         }
@@ -1133,7 +1134,7 @@ impl LegacyBackupReader for PostgresEngine {
             // uses, which CreateTable would refuse on the service; drop those.
             if omitted {
                 let mut used: std::collections::HashSet<String> =
-                    serde_json::from_value::<Vec<KeySchemaElement>>(key_schema.clone())
+                    from_catalog_json::<Vec<KeySchemaElement>>(key_schema.clone())
                         .map_err(|e| StorageError::Internal(format!("key schema: {e}")))?
                         .into_iter()
                         .map(|k| k.attribute_name)
@@ -1151,15 +1152,17 @@ impl LegacyBackupReader for PostgresEngine {
                 {
                     used.insert(k.attribute_name.clone());
                 }
-                if let Some(defs) = attr_defs.as_array_mut() {
-                    defs.retain(|d| {
-                        d.get("AttributeName")
-                            .and_then(serde_json::Value::as_str)
-                            .is_some_and(|n| used.contains(n))
-                    });
-                }
+                // The names are compared decoded: the stored column holds the
+                // escaped form (`catalog_json`), and the index names above
+                // were decoded with the captured definition.
+                let mut defs: Vec<AttributeDefinition> = from_catalog_json(attr_defs)
+                    .map_err(|e| StorageError::Internal(format!("attribute definitions: {e}")))?;
+                defs.retain(|d| used.contains(&d.attribute_name));
+                attr_defs = to_catalog_json(&defs)?;
             }
-            let definition = definition.to_json()?;
+            // The definition carries attribute names, so it is stored in the
+            // form of the catalog columns it was read from.
+            let definition = escape_json(definition.to_json()?);
 
             // All catalog-side writes are one transaction, so a crash cannot
             // leave a backup AVAILABLE with partial items. The items are read
@@ -1314,7 +1317,7 @@ impl LegacyBackupReader for PostgresEngine {
             ) = row;
 
             let key_schema: Vec<extenddb_core::types::KeySchemaElement> =
-                serde_json::from_value(ks_json)
+                from_catalog_json(ks_json)
                     .map_err(|e| StorageError::Internal(format!("Parse key schema: {e}")))?;
 
             Ok(BackupDescription {
@@ -1558,7 +1561,7 @@ impl LegacyBackupReader for PostgresEngine {
             };
 
             let definition = definition
-                .map(|d| BackupTableDefinition::from_json(d, &backup_arn))
+                .map(|d| BackupTableDefinition::from_json(unescape_json(d), &backup_arn))
                 .transpose()?;
             // A backup taken before catalog 0.0.4 has no definition; its
             // vector indexes are known only from the `backups.vector_indexes`
@@ -1594,9 +1597,9 @@ impl LegacyBackupReader for PostgresEngine {
                 source_table_name,
                 source_table_arn,
                 source_table_id,
-                key_schema: serde_json::from_value(ks_json)
+                key_schema: from_catalog_json(ks_json)
                     .map_err(|e| StorageError::Internal(format!("Parse key schema: {e}")))?,
-                attribute_definitions: serde_json::from_value(ad_json)
+                attribute_definitions: from_catalog_json(ad_json)
                     .map_err(|e| StorageError::Internal(format!("Parse attr defs: {e}")))?,
                 billing_mode,
                 provisioned_throughput,