From d24ed5c98111c8d48bd450cab51ffe5fa336f672 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Fabr=C3=ADcio=20Bracht?= Date: Sat, 1 Aug 2026 23:14:34 -0300 Subject: [PATCH 1/2] merge indexed fields instead of replacing on repeat add_index --- CHANGELOG.md | 6 + Cargo.lock | 6 +- crates/mqdb-agent/Cargo.toml | 2 +- crates/mqdb-agent/src/database/query.rs | 39 +++++++ crates/mqdb-agent/src/database/schema_ops.rs | 6 +- crates/mqdb-cli/Cargo.toml | 2 +- crates/mqdb-core/Cargo.toml | 2 +- crates/mqdb-core/src/index.rs | 114 ++++++++++++++++++- 8 files changed, 167 insertions(+), 10 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 79ee2bc3..6bafd1e6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,12 @@ All notable changes to this project will be documented in this file. Each entry lists the date and the crate versions that were released. +## 2026-08-01 — mqdb-cli 0.8.25, mqdb-core 0.7.8, mqdb-agent 0.8.18 + +### Fixed + +- **Indexing a second field on an entity no longer de-registers the first (agent mode).** `IndexManager::add_index` replaced an entity's index definition instead of merging it, and the agent persisted only the newly-added fields. So registering a second indexed/unique field on the same entity (e.g. a `email` unique constraint followed by a `username` unique constraint) silently dropped the first field from the index registry: its `idx/{entity}/{field}/…` entries were orphaned, equality/range filters on it quietly fell back to full scans, and rows created afterward were never indexed on it. `add_index` now merges fields into the existing definition (an entity's index is the union of every field ever indexed on it, since entries are stored per field), and the agent persists the merged definition (computed, persisted, and committed before the in-memory registry is updated, so a failed commit leaves memory and disk consistent). This also removes the blocker for a future narrowed stale-index self-heal (#90). Migration note: a database created before this fix persisted only the last-registered field, so any earlier field lost to the old bug stays de-registered on upgrade — re-run `add index` (or re-declare the unique constraint) for that field to re-register and reindex it; its orphaned entries are otherwise inert and never produce wrong results. + ## 2026-07-31 — mqdb-cli 0.8.24, mqdb-agent 0.8.17 ### Fixed diff --git a/Cargo.lock b/Cargo.lock index 2bdbc33a..585d0d90 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1420,7 +1420,7 @@ dependencies = [ [[package]] name = "mqdb-agent" -version = "0.8.17" +version = "0.8.18" dependencies = [ "arc-swap", "argon2", @@ -1455,7 +1455,7 @@ dependencies = [ [[package]] name = "mqdb-cli" -version = "0.8.24" +version = "0.8.25" dependencies = [ "base64", "bebytes", @@ -1515,7 +1515,7 @@ dependencies = [ [[package]] name = "mqdb-core" -version = "0.7.7" +version = "0.7.8" dependencies = [ "arc-swap", "bebytes", diff --git a/crates/mqdb-agent/Cargo.toml b/crates/mqdb-agent/Cargo.toml index a944cdd9..23edead1 100644 --- a/crates/mqdb-agent/Cargo.toml +++ b/crates/mqdb-agent/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "mqdb-agent" -version = "0.8.17" +version = "0.8.18" edition.workspace = true license = "Apache-2.0" authors.workspace = true diff --git a/crates/mqdb-agent/src/database/query.rs b/crates/mqdb-agent/src/database/query.rs index 77efe1f1..aa856283 100644 --- a/crates/mqdb-agent/src/database/query.rs +++ b/crates/mqdb-agent/src/database/query.rs @@ -474,6 +474,45 @@ mod stale_index_tests { (tmp, db) } + #[tokio::test] + async fn second_add_index_keeps_first_field_indexed() { + let (_tmp, db) = test_db().await; + db.add_index("users".to_string(), vec!["email".to_string()]) + .await + .unwrap(); + db.add_index("users".to_string(), vec!["username".to_string()]) + .await + .unwrap(); + + db.create( + "users".to_string(), + json!({ "id": "u1", "email": "a@x.com", "username": "alice" }), + None, + None, + None, + &ScopeConfig::default(), + ) + .await + .unwrap(); + + // A row created after BOTH add_index calls must be indexed on the FIRST + // field too — under the old replace-not-merge bug, `email` was de-registered + // and only `username` would be indexed for new rows. + let email_value = keys::encode_value_for_index(&json!("a@x.com")).unwrap(); + let email_key = keys::encode_index_key("users", "email", &email_value, "u1"); + assert!( + db.storage.get(&email_key).unwrap().is_some(), + "first indexed field must remain indexed after a second add_index" + ); + + let username_value = keys::encode_value_for_index(&json!("alice")).unwrap(); + let username_key = keys::encode_index_key("users", "username", &username_value, "u1"); + assert!( + db.storage.get(&username_key).unwrap().is_some(), + "second indexed field must be indexed" + ); + } + #[tokio::test] async fn list_purges_stale_index_entry_for_missing_row() { let (_tmp, db) = test_db().await; diff --git a/crates/mqdb-agent/src/database/schema_ops.rs b/crates/mqdb-agent/src/database/schema_ops.rs index 752baffb..b0748877 100644 --- a/crates/mqdb-agent/src/database/schema_ops.rs +++ b/crates/mqdb-agent/src/database/schema_ops.rs @@ -42,12 +42,12 @@ impl Database { drop(schema_registry); { - let definition = mqdb_core::index::IndexDefinition::new(entity.clone(), fields); let mut batch = self.storage.batch(); let mut manager = self.index_manager.write().await; - manager.persist_index(&mut batch, &definition)?; + let merged = manager.merged_definition(&entity, fields); + manager.persist_index(&mut batch, &merged)?; batch.commit()?; - manager.add_index(definition); + manager.add_index(merged); drop(manager); } diff --git a/crates/mqdb-cli/Cargo.toml b/crates/mqdb-cli/Cargo.toml index ace04f8c..c97f7ac2 100644 --- a/crates/mqdb-cli/Cargo.toml +++ b/crates/mqdb-cli/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "mqdb-cli" -version = "0.8.24" +version = "0.8.25" publish = false edition.workspace = true license = "AGPL-3.0-only" diff --git a/crates/mqdb-core/Cargo.toml b/crates/mqdb-core/Cargo.toml index 74421385..7439fdec 100644 --- a/crates/mqdb-core/Cargo.toml +++ b/crates/mqdb-core/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "mqdb-core" -version = "0.7.7" +version = "0.7.8" edition.workspace = true license = "Apache-2.0" authors.workspace = true diff --git a/crates/mqdb-core/src/index.rs b/crates/mqdb-core/src/index.rs index e2fb1e42..d6e76b08 100644 --- a/crates/mqdb-core/src/index.rs +++ b/crates/mqdb-core/src/index.rs @@ -33,8 +33,24 @@ impl IndexManager { } } + /// Register indexed fields for an entity, **merging** into any existing + /// definition rather than replacing it. An entity's index definition is the + /// union of every field ever indexed on it (index entries are stored per + /// field), so registering a second index/unique field must not de-register + /// the first. pub fn add_index(&mut self, definition: IndexDefinition) { - self.indexes.insert(definition.entity.clone(), definition); + match self.indexes.get_mut(&definition.entity) { + Some(existing) => { + for field in definition.fields { + if !existing.fields.contains(&field) { + existing.fields.push(field); + } + } + } + None => { + self.indexes.insert(definition.entity.clone(), definition); + } + } } #[allow(clippy::must_use_candidate)] @@ -115,6 +131,26 @@ impl IndexManager { Ok(()) } + /// Compute the merged index definition for `entity` — the union of its + /// currently registered fields and `new_fields` — **without** mutating + /// in-memory state. Callers persist and commit this before applying + /// [`add_index`](Self::add_index), so a failed commit leaves the in-memory + /// registry and disk consistent (both without the new fields). + #[must_use] + pub fn merged_definition(&self, entity: &str, new_fields: Vec) -> IndexDefinition { + let mut fields = self + .indexes + .get(entity) + .map(|d| d.fields.clone()) + .unwrap_or_default(); + for field in new_fields { + if !fields.contains(&field) { + fields.push(field); + } + } + IndexDefinition::new(entity.to_string(), fields) + } + /// # Errors /// Returns an error if reading or deserializing index definitions fails. pub fn load_indexes(&mut self, storage: &Storage) -> Result<()> { @@ -235,6 +271,82 @@ mod tests { assert!(!mgr.is_field_indexed("posts", "age")); } + #[test] + fn add_index_merges_fields_instead_of_replacing() { + let mut mgr = IndexManager::new(); + mgr.add_index(IndexDefinition::new("users".into(), vec!["email".into()])); + mgr.add_index(IndexDefinition::new( + "users".into(), + vec!["username".into()], + )); + + assert!( + mgr.is_field_indexed("users", "email"), + "first field must survive" + ); + assert!( + mgr.is_field_indexed("users", "username"), + "second field must be added" + ); + assert_eq!( + mgr.get_indexed_fields("users").unwrap(), + &vec!["email".to_string(), "username".to_string()] + ); + + // idempotent: re-adding an existing field does not duplicate it + mgr.add_index(IndexDefinition::new("users".into(), vec!["email".into()])); + assert_eq!(mgr.get_indexed_fields("users").unwrap().len(), 2); + } + + #[test] + fn merged_definition_unions_without_mutating() { + let mut mgr = IndexManager::new(); + mgr.add_index(IndexDefinition::new("users".into(), vec!["email".into()])); + + let merged = mgr.merged_definition("users", vec!["username".into()]); + assert_eq!( + merged.fields, + vec!["email".to_string(), "username".to_string()] + ); + // must NOT mutate in-memory state (persist/commit happens first) + assert_eq!( + mgr.get_indexed_fields("users").unwrap(), + &vec!["email".to_string()] + ); + // dedups against existing fields + assert_eq!( + mgr.merged_definition("users", vec!["email".into()]).fields, + vec!["email".to_string()] + ); + } + + #[test] + fn persist_merged_then_reload_has_all_fields() { + // Mirrors the agent's safe order: compute merged -> persist -> commit -> apply. + let storage = Storage::memory(); + let mut mgr = IndexManager::new(); + + let d1 = mgr.merged_definition("users", vec!["email".into()]); + let mut b1 = storage.batch(); + mgr.persist_index(&mut b1, &d1).unwrap(); + b1.commit().unwrap(); + mgr.add_index(d1); + + let d2 = mgr.merged_definition("users", vec!["username".into()]); + let mut b2 = storage.batch(); + mgr.persist_index(&mut b2, &d2).unwrap(); + b2.commit().unwrap(); + mgr.add_index(d2); + + let mut loaded = IndexManager::new(); + loaded.load_indexes(&storage).unwrap(); + assert_eq!( + loaded.get_indexed_fields("users").unwrap(), + &vec!["email".to_string(), "username".to_string()], + "restart must reload every indexed field, not just the last" + ); + } + fn setup_indexed_entities(ages: &[i64]) -> (Storage, IndexManager) { let storage = Storage::memory(); let mut mgr = IndexManager::new(); From 18b497d056d4a43b6e58197b5777d2750bcd8919 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Fabr=C3=ADcio=20Bracht?= Date: Sun, 2 Aug 2026 15:01:10 -0300 Subject: [PATCH 2/2] guard index-merge commit ordering with a fault-injection test and soften migration note --- CHANGELOG.md | 2 +- crates/mqdb-agent/src/database/schema_ops.rs | 157 +++++++++++++++++++ crates/mqdb-wasm/Cargo.lock | 2 +- 3 files changed, 159 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 6bafd1e6..e35ad45f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,7 +8,7 @@ Each entry lists the date and the crate versions that were released. ### Fixed -- **Indexing a second field on an entity no longer de-registers the first (agent mode).** `IndexManager::add_index` replaced an entity's index definition instead of merging it, and the agent persisted only the newly-added fields. So registering a second indexed/unique field on the same entity (e.g. a `email` unique constraint followed by a `username` unique constraint) silently dropped the first field from the index registry: its `idx/{entity}/{field}/…` entries were orphaned, equality/range filters on it quietly fell back to full scans, and rows created afterward were never indexed on it. `add_index` now merges fields into the existing definition (an entity's index is the union of every field ever indexed on it, since entries are stored per field), and the agent persists the merged definition (computed, persisted, and committed before the in-memory registry is updated, so a failed commit leaves memory and disk consistent). This also removes the blocker for a future narrowed stale-index self-heal (#90). Migration note: a database created before this fix persisted only the last-registered field, so any earlier field lost to the old bug stays de-registered on upgrade — re-run `add index` (or re-declare the unique constraint) for that field to re-register and reindex it; its orphaned entries are otherwise inert and never produce wrong results. +- **Indexing a second field on an entity no longer de-registers the first (agent mode).** `IndexManager::add_index` replaced an entity's index definition instead of merging it, and the agent persisted only the newly-added fields. So registering a second indexed/unique field on the same entity (e.g. a `email` unique constraint followed by a `username` unique constraint) silently dropped the first field from the index registry: its `idx/{entity}/{field}/…` entries were orphaned, equality/range filters on it quietly fell back to full scans, and rows created afterward were never indexed on it. `add_index` now merges fields into the existing definition (an entity's index is the union of every field ever indexed on it, since entries are stored per field), and the agent persists the merged definition (computed, persisted, and committed before the in-memory registry is updated, so a failed commit leaves memory and disk consistent). This also removes the blocker for a future narrowed stale-index self-heal (#90). Migration note: a database created before this fix persisted only the last-registered field, so an earlier field lost to the old bug stays de-registered on upgrade — re-declare it via `add index` / the unique constraint to re-register and reindex. Re-declaring backfills current values but does not purge entries orphaned during the de-registered window; if the field's value was updated in that window a stale entry can linger and produce a false positive on a lookup of the old value, so a full index rebuild for the entity is needed to fully clean such entries. ## 2026-07-31 — mqdb-cli 0.8.24, mqdb-agent 0.8.17 diff --git a/crates/mqdb-agent/src/database/schema_ops.rs b/crates/mqdb-agent/src/database/schema_ops.rs index b0748877..bcb04ea9 100644 --- a/crates/mqdb-agent/src/database/schema_ops.rs +++ b/crates/mqdb-agent/src/database/schema_ops.rs @@ -280,3 +280,160 @@ impl Database { Ok(()) } } + +#[cfg(test)] +mod tests { + use super::Database; + use mqdb_core::config::DatabaseConfig; + use mqdb_core::error::{Error, Result}; + use mqdb_core::keys; + use mqdb_core::storage::{BatchOperations, MemoryBackend, StorageBackend}; + use mqdb_core::types::ScopeConfig; + use serde_json::json; + use std::path::PathBuf; + use std::sync::Arc; + use std::sync::atomic::{AtomicBool, Ordering}; + + /// Wraps an in-memory backend but can be told to fail every `commit()`. + struct FailingBackend { + inner: MemoryBackend, + fail_commit: Arc, + } + + impl StorageBackend for FailingBackend { + fn get(&self, key: &[u8]) -> Result>> { + self.inner.get(key) + } + fn insert(&self, key: &[u8], value: &[u8]) -> Result<()> { + self.inner.insert(key, value) + } + fn remove(&self, key: &[u8]) -> Result<()> { + self.inner.remove(key) + } + fn prefix_scan(&self, prefix: &[u8]) -> Result, Vec)>> { + self.inner.prefix_scan(prefix) + } + fn prefix_count(&self, prefix: &[u8]) -> Result { + self.inner.prefix_count(prefix) + } + fn prefix_scan_keys(&self, prefix: &[u8]) -> Result>> { + self.inner.prefix_scan_keys(prefix) + } + fn prefix_scan_batch( + &self, + prefix: &[u8], + batch_size: usize, + after_key: Option<&[u8]>, + ) -> Result, Vec)>> { + self.inner.prefix_scan_batch(prefix, batch_size, after_key) + } + fn range_scan(&self, start: &[u8], end: &[u8]) -> Result, Vec)>> { + self.inner.range_scan(start, end) + } + fn batch(&self) -> Box { + Box::new(FailingBatch { + inner: self.inner.batch(), + fail_commit: Arc::clone(&self.fail_commit), + }) + } + fn flush(&self) -> Result<()> { + self.inner.flush() + } + } + + struct FailingBatch { + inner: Box, + fail_commit: Arc, + } + + impl BatchOperations for FailingBatch { + fn insert(&mut self, key: Vec, value: Vec) { + self.inner.insert(key, value); + } + fn remove(&mut self, key: Vec) { + self.inner.remove(key); + } + fn expect_value(&mut self, key: Vec, expected_value: Vec) { + self.inner.expect_value(key, expected_value); + } + fn expect_absent(&mut self, key: Vec) { + self.inner.expect_absent(key); + } + fn commit(self: Box) -> Result<()> { + if self.fail_commit.load(Ordering::SeqCst) { + return Err(Error::Internal("injected commit failure".to_string())); + } + self.inner.commit() + } + } + + // Guards the persist-before-mutate ordering: because the in-memory merge is + // applied only AFTER a successful commit, a commit failure during add_index + // must leave the index registry and disk consistent (the new field unindexed). + #[tokio::test] + async fn add_index_commit_failure_leaves_registry_consistent() { + let fail = Arc::new(AtomicBool::new(false)); + let backend = Arc::new(FailingBackend { + inner: MemoryBackend::new(), + fail_commit: Arc::clone(&fail), + }); + let config = DatabaseConfig::new(PathBuf::from("mem")).without_background_tasks(); + let db = Database::open_with_backend(backend, config).await.unwrap(); + + db.add_index("users".to_string(), vec!["email".to_string()]) + .await + .unwrap(); + + fail.store(true, Ordering::SeqCst); + let result = db + .add_index("users".to_string(), vec!["username".to_string()]) + .await; + assert!(result.is_err(), "add_index must fail when its commit fails"); + fail.store(false, Ordering::SeqCst); + + db.create( + "users".to_string(), + json!({ "id": "u1", "email": "a@x.com", "username": "alice" }), + None, + None, + None, + &ScopeConfig::default(), + ) + .await + .unwrap(); + + let email_value = keys::encode_value_for_index(&json!("a@x.com")).unwrap(); + let email_key = keys::encode_index_key("users", "email", &email_value, "u1"); + assert!( + db.storage.get(&email_key).unwrap().is_some(), + "email must remain indexed after the failed add_index" + ); + let username_value = keys::encode_value_for_index(&json!("alice")).unwrap(); + let username_key = keys::encode_index_key("users", "username", &username_value, "u1"); + assert!( + db.storage.get(&username_key).unwrap().is_none(), + "username must not be indexed after a failed add_index commit" + ); + + // Recoverable: a later successful add_index registers and indexes it. + db.add_index("users".to_string(), vec!["username".to_string()]) + .await + .unwrap(); + db.create( + "users".to_string(), + json!({ "id": "u2", "email": "b@x.com", "username": "bob" }), + None, + None, + None, + &ScopeConfig::default(), + ) + .await + .unwrap(); + let bob_value = keys::encode_value_for_index(&json!("bob")).unwrap(); + let bob_key = keys::encode_index_key("users", "username", &bob_value, "u2"); + assert!( + db.storage.get(&bob_key).unwrap().is_some(), + "username must be indexed after a successful re-add" + ); + } +} diff --git a/crates/mqdb-wasm/Cargo.lock b/crates/mqdb-wasm/Cargo.lock index 980a6836..f021d3be 100644 --- a/crates/mqdb-wasm/Cargo.lock +++ b/crates/mqdb-wasm/Cargo.lock @@ -342,7 +342,7 @@ dependencies = [ [[package]] name = "mqdb-core" -version = "0.7.7" +version = "0.7.8" dependencies = [ "arc-swap", "bebytes",