test(sync): demultiplex sync frames by type in interop tests

The sync channel is full-duplex, so Origin can interleave
server-initiated frames (e.g. a DeltaPush echoing another peer's
write, or a keepalive) ahead of the response a test is waiting for.
Loop and match on msg_type instead of assuming the next frame on the
socket is always the expected reply.

Author: farhan-syah

nodedb-lite/tests/sync_interop_delta_ack.rs

@@ -35,13 +35,26 @@ async fn push_and_recv(
         .await
         .expect("send DeltaPush");
 
-    let resp = tokio::time::timeout(Duration::from_secs(10), ws.next())
-        .await
-        .expect("timeout waiting for delta response")
-        .expect("stream closed before response")
-        .expect("WebSocket read error");
-
-    SyncFrame::from_bytes(resp.into_data().as_ref()).expect("decode response frame")
+    // The sync channel is full-duplex: Origin may interleave its own
+    // server-initiated frames (e.g. a `DeltaPush` echoing another peer's
+    // write, or a keepalive `Ping`) ahead of the ack for the frame we just
+    // sent. A real client demultiplexes by frame type rather than assuming
+    // the next frame is its response, so skip anything that is not a
+    // terminal ack for our push.
+    loop {
+        let resp = tokio::time::timeout(Duration::from_secs(10), ws.next())
+            .await
+            .expect("timeout waiting for delta response")
+            .expect("stream closed before response")
+            .expect("WebSocket read error");
+
+        let frame =
+            SyncFrame::from_bytes(resp.into_data().as_ref()).expect("decode response frame");
+        match frame.msg_type {
+            SyncMessageType::DeltaAck | SyncMessageType::DeltaReject => return frame,
+            _ => continue,
+        }
+    }
 }
 
 // ── tests ─────────────────────────────────────────────────────────────────────

nodedb-lite/tests/sync_interop_live.rs

@@ -436,14 +436,22 @@ async fn live_shape_snapshot_with_wal_lsn() {
     .await
     .expect("send");
 
-    let resp = tokio::time::timeout(Duration::from_secs(5), ws.next())
-        .await
-        .expect("timeout")
-        .expect("closed")
-        .expect("error");
+    // Full-duplex channel: Origin may interleave a server-initiated frame
+    // (e.g. a `DeltaPush` for a concurrent write) ahead of the `ShapeSnapshot`
+    // answering our subscribe. Demultiplex by frame type rather than assuming
+    // the next frame is the snapshot.
+    let frame = loop {
+        let resp = tokio::time::timeout(Duration::from_secs(5), ws.next())
+            .await
+            .expect("timeout")
+            .expect("closed")
+            .expect("error");
 
-    let frame = SyncFrame::from_bytes(resp.into_data().as_ref()).expect("decode");
-    assert_eq!(frame.msg_type, SyncMessageType::ShapeSnapshot);
+        let frame = SyncFrame::from_bytes(resp.into_data().as_ref()).expect("decode");
+        if frame.msg_type == SyncMessageType::ShapeSnapshot {
+            break frame;
+        }
+    };
 
     let snapshot: ShapeSnapshotMsg = frame.decode_body().expect("decode");
     assert_eq!(snapshot.shape_id, "lsn-live-shape");

nodedb-lite/tests/sync_interop_resync.rs

@@ -47,23 +47,24 @@ async fn subscribe_shape(
         .await
         .expect("send ShapeSubscribe");
 
-    let resp = tokio::time::timeout(Duration::from_secs(10), ws.next())
-        .await
-        .expect("timeout waiting for ShapeSnapshot")
-        .expect("stream closed before snapshot")
-        .expect("WebSocket read error");
-
-    let frame = SyncFrame::from_bytes(resp.into_data().as_ref()).expect("decode frame");
-    assert_eq!(
-        frame.msg_type,
-        SyncMessageType::ShapeSnapshot,
-        "expected ShapeSnapshot, got {:?}",
-        frame.msg_type
-    );
+    // The channel is full-duplex: Origin may interleave a server-initiated
+    // frame (e.g. a `DeltaPush` for a concurrent write, or a keepalive) ahead
+    // of the `ShapeSnapshot` that answers our subscribe. Demultiplex by frame
+    // type rather than assuming the next frame is the snapshot.
+    loop {
+        let resp = tokio::time::timeout(Duration::from_secs(10), ws.next())
+            .await
+            .expect("timeout waiting for ShapeSnapshot")
+            .expect("stream closed before snapshot")
+            .expect("WebSocket read error");
 
-    frame
-        .decode_body::<ShapeSnapshotMsg>()
-        .expect("decode ShapeSnapshotMsg")
+        let frame = SyncFrame::from_bytes(resp.into_data().as_ref()).expect("decode frame");
+        if frame.msg_type == SyncMessageType::ShapeSnapshot {
+            return frame
+                .decode_body::<ShapeSnapshotMsg>()
+                .expect("decode ShapeSnapshotMsg");
+        }
+    }
 }
 
 // ── tests ─────────────────────────────────────────────────────────────────────