Fix issues #188–#193

Cargo.toml

@@ -164,7 +164,7 @@ chrono = { version = "0.4", features = ["std"], default-features = false }
 
 # MessagePack
 rmpv = "1"
-zerompk = { version = "0.5", features = ["std", "derive"] }
+zerompk = { version = "0.6", features = ["std", "derive"] }
 
 # Bitmap
 roaring = "0.11"

nodedb-client/src/native/connection/mod.rs

@@ -356,9 +356,10 @@ impl NativeConnection {
         op: OpCode,
         fields: TextFields,
     ) -> NodeDbResult<NativeResponse> {
+        let req_seq = self.next_seq();
         let req = NativeRequest {
             op,
-            seq: self.next_seq(),
+            seq: req_seq,
             fields: RequestFields::Text(fields),
         };
 
@@ -374,9 +375,9 @@ impl NativeConnection {
         self.stream.flush().await.map_err(io_err)?;
 
         let mut combined_rows: Vec<Vec<nodedb_types::Value>> = Vec::new();
-        let mut final_resp: Option<NativeResponse> = None;
+        let mut partial_columns: Option<Vec<String>> = None;
 
-        loop {
+        let final_resp = loop {
             let mut len_buf = [0u8; FRAME_HEADER_LEN];
             self.stream.read_exact(&mut len_buf).await.map_err(io_err)?;
             let resp_len = u32::from_be_bytes(len_buf);
@@ -396,30 +397,51 @@ impl NativeConnection {
                 NodeDbError::serialization("msgpack", format!("response decode: {e}"))
             })?;
 
+            if resp.seq != req_seq {
+                // A fan-out query for a preceding request can leave stale
+                // trailing frames on the wire after that request's terminal
+                // frame was already returned to its caller. Discard any
+                // frame that doesn't belong to this request rather than
+                // misattributing it — never surface another request's rows
+                // or status as this request's response.
+                tracing::warn!(
+                    expected_seq = req_seq,
+                    got_seq = resp.seq,
+                    "native connection: discarding stale response frame"
+                );
+                continue;
+            }
+
             if resp.status == ResponseStatus::Partial {
+                if partial_columns.is_none() {
+                    partial_columns = resp.columns;
+                }
                 if let Some(rows) = resp.rows {
                     combined_rows.extend(rows);
                 }
-                if final_resp.is_none() {
-                    final_resp = Some(NativeResponse { rows: None, ..resp });
-                }
-            } else {
-                if combined_rows.is_empty() {
-                    final_resp = Some(resp);
-                } else {
-                    if let Some(ref rows) = resp.rows {
-                        combined_rows.extend(rows.iter().cloned());
-                    }
-                    let mut merged = final_resp.unwrap_or(resp);
-                    merged.rows = Some(combined_rows);
-                    merged.status = ResponseStatus::Ok;
-                    final_resp = Some(merged);
-                }
-                break;
+                continue;
             }
-        }
 
-        final_resp.ok_or_else(|| NodeDbError::internal("no final response received"))
+            // The terminal frame owns status and all terminal metadata. In
+            // particular, never turn a stream error into success merely
+            // because earlier partial rows were received.
+            if resp.status == ResponseStatus::Error {
+                break resp;
+            }
+            let mut terminal = resp;
+            if let Some(rows) = terminal.rows.take() {
+                combined_rows.extend(rows);
+            }
+            if !combined_rows.is_empty() {
+                terminal.rows = Some(combined_rows);
+            }
+            if terminal.columns.is_none() {
+                terminal.columns = partial_columns;
+            }
+            break terminal;
+        };
+
+        Ok(final_resp)
     }
 }
 

nodedb-cluster-tests/tests/descriptor_versioning_cross_node.rs

@@ -19,7 +19,7 @@ mod common;
 
 use std::time::Duration;
 
-use common::cluster_harness::{TestCluster, wait_for};
+use common::cluster_harness::{TestCluster, TestClusterNode, wait_for};
 
 const TENANT: u64 = 1;
 
@@ -144,6 +144,49 @@ async fn alter_collection_bumps_version_monotonically() {
     cluster.shutdown().await;
 }
 
+#[tokio::test(flavor = "multi_thread", worker_threads = 6)]
+async fn concurrent_cross_node_updates_allocate_distinct_descriptor_versions() {
+    let cluster = TestCluster::spawn_three().await.expect("3-node cluster");
+    cluster
+        .exec_ddl_on_any_leader("CREATE USER owner_a WITH PASSWORD 'pw' ROLE READWRITE")
+        .await
+        .expect("create owner_a");
+    cluster
+        .exec_ddl_on_any_leader("CREATE USER owner_b WITH PASSWORD 'pw' ROLE READWRITE")
+        .await
+        .expect("create owner_b");
+    cluster
+        .exec_ddl_on_any_leader("CREATE COLLECTION concurrently_owned")
+        .await
+        .expect("create collection");
+
+    let update_a = cluster.nodes[0]
+        .client
+        .simple_query("ALTER COLLECTION concurrently_owned OWNER TO owner_a");
+    let update_b = cluster.nodes[1]
+        .client
+        .simple_query("ALTER COLLECTION concurrently_owned OWNER TO owner_b");
+    let (result_a, result_b) = tokio::join!(update_a, update_b);
+    result_a.expect("node 0 concurrent owner update");
+    result_b.expect("node 1 concurrent owner update");
+
+    wait_for(
+        "both concurrent updates apply as distinct versions on every node",
+        Duration::from_secs(15),
+        Duration::from_millis(50),
+        || {
+            cluster.nodes.iter().all(|node| {
+                node.collection_descriptor(TENANT, "concurrently_owned")
+                    .map(|stamp| stamp.0)
+                    == Some(3)
+            })
+        },
+    )
+    .await;
+
+    cluster.shutdown().await;
+}
+
 #[tokio::test(flavor = "multi_thread", worker_threads = 6)]
 async fn distinct_collections_get_independent_versions() {
     let cluster = TestCluster::spawn_three().await.expect("3-node cluster");
@@ -187,3 +230,113 @@ async fn distinct_collections_get_independent_versions() {
 
     cluster.shutdown().await;
 }
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 6)]
+async fn historical_descriptor_entries_replay_without_regressing_the_latest_version() {
+    let data_dir = tempfile::tempdir().expect("tempdir");
+    let data_path = data_dir.path().to_path_buf();
+    let node = TestClusterNode::spawn_single_node_calvin_on_path(4, data_path.clone())
+        .await
+        .expect("spawn single-node metadata group");
+    wait_for(
+        "single-node sequencer leader elected",
+        Duration::from_secs(10),
+        Duration::from_millis(50),
+        || node.sequencer_leader() == node.node_id,
+    )
+    .await;
+    wait_for(
+        "single-node metadata leader elected",
+        Duration::from_secs(10),
+        Duration::from_millis(50),
+        || node.shared.is_metadata_leader(),
+    )
+    .await;
+
+    node.client
+        .simple_query(
+            "CREATE COLLECTION replay_graph (id TEXT PRIMARY KEY, name TEXT) \
+             WITH (engine='document_strict')",
+        )
+        .await
+        .expect("create graph-bearing collection");
+    wait_for(
+        "collection descriptor reaches version 1",
+        Duration::from_secs(10),
+        Duration::from_millis(50),
+        || {
+            node.collection_descriptor(TENANT, "replay_graph")
+                .map(|v| v.0)
+                == Some(1)
+        },
+    )
+    .await;
+
+    node.client
+        .simple_query(
+            "GRAPH INSERT EDGE IN replay_graph FROM 'a' TO 'b' \
+             TYPE 'knows' PROPERTIES '{}'",
+        )
+        .await
+        .expect("insert edge and mark collection edge-bearing");
+    wait_for(
+        "edge-bearing descriptor reaches version 2",
+        Duration::from_secs(10),
+        Duration::from_millis(50),
+        || {
+            node.collection_descriptor(TENANT, "replay_graph")
+                .map(|v| v.0)
+                == Some(2)
+        },
+    )
+    .await;
+
+    node.graceful_shutdown_wal_only().await;
+    let node = TestClusterNode::spawn_single_node_calvin_on_path(4, data_path)
+        .await
+        .expect("restart against the persisted catalog and full metadata log");
+    wait_for(
+        "single-node sequencer leader re-elected after restart",
+        Duration::from_secs(10),
+        Duration::from_millis(50),
+        || node.sequencer_leader() == node.node_id,
+    )
+    .await;
+    wait_for(
+        "single-node metadata leader re-elected after restart",
+        Duration::from_secs(10),
+        Duration::from_millis(50),
+        || node.shared.is_metadata_leader(),
+    )
+    .await;
+
+    wait_for(
+        "latest collection descriptor remains visible after replay",
+        Duration::from_secs(10),
+        Duration::from_millis(50),
+        || {
+            node.collection_descriptor(TENANT, "replay_graph")
+                .map(|v| v.0)
+                == Some(2)
+        },
+    )
+    .await;
+
+    node.client
+        .simple_query(
+            "CREATE COLLECTION ddl_after_descriptor_replay \
+             (id TEXT PRIMARY KEY) WITH (engine='document_strict')",
+        )
+        .await
+        .expect(
+            "historical metadata replay must advance its watermark so later DDL remains usable",
+        );
+    assert_eq!(
+        node.collection_descriptor(TENANT, "replay_graph")
+            .map(|version| version.0),
+        Some(2),
+        "replaying historical version 1 must not overwrite the persisted latest version 2"
+    );
+
+    node.shutdown().await;
+}