Harden sync push retries and stabilize auto-sync acceptance coverage

Author: lucidprogrammer

Cargo.lock

@@ -29,6 +29,8 @@ time.workspace = true
 [dev-dependencies]
 contextdb-server = { path = "../contextdb-server", version = "0.2" }
 tokio = { version = "1", features = ["macros", "rt-multi-thread"] }
+futures-util = "0.3"
+async-nats = "0.46"
 testcontainers = "0.27"
 tempfile = "3"
 criterion = { version = "0.8", features = ["html_reports"] }

crates/contextdb-engine/Cargo.toml

@@ -265,17 +265,42 @@ impl SyncClient {
 
                     match tokio::time::timeout(PUSH_REQUEST_TIMEOUT, inbox_sub.next()).await {
                         Ok(Some(msg)) => {
-                            if msg.status == Some(async_nats::StatusCode::NO_RESPONDERS)
-                                && attempt < 4
-                            {
-                                tracing::debug!(attempt, "push got no responders, retrying");
-                                continue;
+                            if let Some(status) = msg.status {
+                                if status == async_nats::StatusCode::NO_RESPONDERS && attempt < 4 {
+                                    tracing::debug!(attempt, "push got no responders, retrying");
+                                    continue;
+                                }
+                                if attempt < 4 {
+                                    tracing::debug!(
+                                        attempt,
+                                        ?status,
+                                        "push got status reply, retrying"
+                                    );
+                                    continue;
+                                }
+                                return Err(Error::SyncError(format!(
+                                    "push failed with NATS status reply: {status:?}"
+                                )));
                             }
-                            let envelope = decode(&msg.payload)
-                                .map_err(|e| Error::SyncError(e.to_string()))?;
-                            let response: PushResponse =
-                                rmp_serde::from_slice(&envelope.payload)
-                                    .map_err(|e| Error::SyncError(e.to_string()))?;
+
+                            let envelope = match decode(&msg.payload) {
+                                Ok(envelope) => envelope,
+                                Err(err) if attempt < 4 => {
+                                    tracing::debug!(attempt, error = %err, "push got malformed reply envelope, retrying");
+                                    continue;
+                                }
+                                Err(err) => return Err(Error::SyncError(err.to_string())),
+                            };
+                            let response: PushResponse = match rmp_serde::from_slice(
+                                &envelope.payload,
+                            ) {
+                                Ok(response) => response,
+                                Err(err) if attempt < 4 => {
+                                    tracing::debug!(attempt, error = %err, "push got malformed reply payload, retrying");
+                                    continue;
+                                }
+                                Err(err) => return Err(Error::SyncError(err.to_string())),
+                            };
                             if let Some(err) = response.error {
                                 return Err(Error::SyncError(err));
                             }

crates/contextdb-server/src/sync_client.rs

@@ -1,5 +1,6 @@
 use contextdb_engine::Database;
 use contextdb_engine::sync_types::{ConflictPolicies, ConflictPolicy};
+use contextdb_server::protocol::{MessageType, PushResponse, WireApplyResult, encode};
 use contextdb_server::{SyncClient, SyncServer};
 use std::collections::HashMap;
 use std::sync::Arc;
@@ -114,6 +115,70 @@ async fn sync_round_trip_smoke() {
     let _ = client.pull(&policies).await;
 }
 
+#[tokio::test]
+async fn sync_00b_push_retries_malformed_reply_before_succeeding() {
+    use contextdb_core::Value;
+    use futures_util::StreamExt;
+    use uuid::Uuid;
+
+    let nats = start_nats().await;
+    let edge = Arc::new(Database::open_memory());
+    let empty = HashMap::new();
+    edge.execute("CREATE TABLE t (id UUID PRIMARY KEY, v TEXT)", &empty)
+        .unwrap();
+
+    let responder = async_nats::connect(&nats.nats_url).await.unwrap();
+    let mut sub = responder
+        .subscribe(contextdb_server::subjects::push_subject("malformed-reply"))
+        .await
+        .unwrap();
+
+    tokio::spawn(async move {
+        let mut attempt = 0u32;
+        while let Some(msg) = sub.next().await {
+            attempt += 1;
+            if let Some(reply) = msg.reply {
+                let payload = if attempt == 1 {
+                    vec![0x91]
+                } else {
+                    encode(
+                        MessageType::PushResponse,
+                        &PushResponse {
+                            result: Some(WireApplyResult {
+                                applied_rows: 1,
+                                skipped_rows: 0,
+                                conflicts: Vec::new(),
+                                new_lsn: 2,
+                            }),
+                            error: None,
+                        },
+                    )
+                    .unwrap()
+                };
+                responder.publish(reply, payload.into()).await.unwrap();
+                if attempt >= 2 {
+                    break;
+                }
+            }
+        }
+    });
+
+    let client = SyncClient::new(edge.clone(), &nats.nats_url, "malformed-reply");
+    let id = Uuid::new_v4();
+    let mut p = HashMap::new();
+    p.insert("id".to_string(), Value::Uuid(id));
+    p.insert("v".to_string(), Value::Text("retry".into()));
+    edge.execute("INSERT INTO t (id, v) VALUES ($id, $v)", &p)
+        .unwrap();
+
+    let result = client
+        .push()
+        .await
+        .expect("push should retry malformed reply");
+    assert_eq!(result.applied_rows, 1);
+    assert!(client.push_watermark() > 0, "push watermark should advance");
+}
+
 /// I connected the sync server to a NATS account that denies its sync subscriptions,
 /// and the server task stayed alive instead of panicking during bootstrap.
 #[tokio::test]

crates/contextdb-server/tests/sync_integration.rs

@@ -2,6 +2,9 @@
 
 use contextdb_core::{Direction, Error, Value, VersionedRow};
 use contextdb_engine::{Database, QueryResult};
+use contextdb_server::protocol::{MessageType, PullRequest, PullResponse, decode, encode};
+use contextdb_server::subjects::pull_subject;
+use futures_util::StreamExt;
 use std::collections::HashMap;
 use std::fs::File;
 use std::io::{ErrorKind, Read, Write};
@@ -297,6 +300,71 @@ pub(crate) async fn start_nats() -> NatsFixture {
     }
 }
 
+pub(crate) async fn wait_for_sync_server_ready(
+    nats_url: &str,
+    tenant_id: &str,
+    timeout: Duration,
+) -> bool {
+    let start = Instant::now();
+    while start.elapsed() < timeout {
+        let client = match async_nats::connect(nats_url).await {
+            Ok(client) => client,
+            Err(_) => {
+                tokio::time::sleep(Duration::from_millis(100)).await;
+                continue;
+            }
+        };
+
+        let inbox = client.new_inbox();
+        let mut inbox_sub = match client.subscribe(inbox.clone()).await {
+            Ok(sub) => sub,
+            Err(_) => {
+                tokio::time::sleep(Duration::from_millis(100)).await;
+                continue;
+            }
+        };
+
+        let payload = match encode(
+            MessageType::PullRequest,
+            &PullRequest {
+                since_lsn: 0,
+                max_entries: Some(1),
+            },
+        ) {
+            Ok(payload) => payload,
+            Err(_) => return false,
+        };
+
+        if client
+            .publish_with_reply(pull_subject(tenant_id), inbox.clone(), payload.into())
+            .await
+            .is_err()
+        {
+            tokio::time::sleep(Duration::from_millis(100)).await;
+            continue;
+        }
+
+        let response = tokio::time::timeout(Duration::from_millis(500), inbox_sub.next()).await;
+        match response {
+            Ok(Some(msg)) if msg.status == Some(async_nats::StatusCode::NO_RESPONDERS) => {}
+            Ok(Some(msg)) if msg.status.is_some() => {}
+            Ok(Some(msg)) => {
+                if let Ok(envelope) = decode(&msg.payload)
+                    && matches!(envelope.message_type, MessageType::PullResponse)
+                    && rmp_serde::from_slice::<PullResponse>(&envelope.payload).is_ok()
+                {
+                    return true;
+                }
+            }
+            Ok(None) | Err(_) => {}
+        }
+
+        tokio::time::sleep(Duration::from_millis(100)).await;
+    }
+
+    false
+}
+
 pub(crate) fn wait_until(timeout: Duration, mut predicate: impl FnMut() -> bool) -> bool {
     let start = Instant::now();
     while start.elapsed() < timeout {

tests/acceptance/common.rs

@@ -721,6 +721,10 @@ async fn f12c_auto_sync_pushes_deletes() {
     let ws_url = &nats.ws_url;
     let tenant = "f12c_auto_sync_pushes_deletes";
     let mut server = spawn_server(&server_path, tenant, nats_url);
+    assert!(
+        wait_for_sync_server_ready(nats_url, tenant, Duration::from_secs(15)).await,
+        "sync server should be ready before f12c setup push begins"
+    );
 
     let mut child = spawn_cli(&edge_path, &["--tenant-id", tenant, "--nats-url", ws_url]);
 
@@ -735,23 +739,31 @@ async fn f12c_auto_sync_pushes_deletes() {
          DELETE FROM sensors WHERE id = '00000000-0000-0000-0000-000000000002'\n",
     );
 
-    // Wait for DELETE to propagate — server should have 1 row, not 2
-    // Each checker uses a unique path to avoid stale local data from previous pulls
-    let mut checker_idx = 0u32;
-    let mut last_stdout = String::new();
+    let checker_path = edge_path.with_file_name("f12c-checker.db");
+    let checker_setup = run_cli_script(
+        &checker_path,
+        &["--tenant-id", tenant, "--nats-url", ws_url],
+        "CREATE TABLE sensors (id UUID PRIMARY KEY, name TEXT)\n.quit\n",
+    );
+    assert!(
+        checker_setup.status.success(),
+        "checker edge setup must succeed"
+    );
+
+    // Wait for DELETE to propagate while the writer CLI is still running.
+    // Reuse a single checker edge so the probe stays end-to-end without creating a new
+    // edge process and schema bootstrap on every polling iteration.
+    let mut last_checker_stdout = String::new();
     let found = wait_until(Duration::from_secs(30), || {
-        checker_idx += 1;
-        let fresh_path = edge_path.with_file_name(format!("f12c-checker-{checker_idx}.db"));
         let check = run_cli_script(
-            &fresh_path,
+            &checker_path,
             &["--tenant-id", tenant, "--nats-url", ws_url],
-            "CREATE TABLE sensors (id UUID PRIMARY KEY, name TEXT)\n\
-             .sync pull\n\
+            ".sync pull\n\
              SELECT count(*) FROM sensors\n\
              .quit\n",
         );
         let stdout = output_string(&check.stdout);
-        last_stdout = stdout.clone();
+        last_checker_stdout = stdout.clone();
         stdout.contains("| 1")
     });
 
@@ -761,7 +773,7 @@ async fn f12c_auto_sync_pushes_deletes() {
 
     assert!(
         found,
-        "DELETE must auto-sync to server while CLI is still running; child stdout={}; child stderr={}; last checker stdout={last_stdout}",
+        "DELETE must auto-sync to server while CLI is still running; child stdout={}; child stderr={}; last checker stdout={last_checker_stdout}",
         output_string(&output.stdout),
         output_string(&output.stderr),
     );