From 434756e4f2b495ddd8d4cafbcab7cbfbef0a674b Mon Sep 17 00:00:00 2001
From: Meta Vens <rwvens@hotmail.com>
Date: Wed, 30 Sep 2026 18:39:32 +0200
Subject: [PATCH] Odin test: pipelined puts with a lost datagram are both
 written

Adapted from Soul's S-6 probe. Two puts go out on one session through a
relay that drops the first reliable datagram. The transport resends it and
delivers both in order; Odin accepts both, so both are on disk and the file
matches the working set. The probe's other finding, that the transport
acknowledges the second before the first is resent, is the transport's
defect (tracked by Self) and is not pinned here.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 crates/odin-daemon/src/main.rs | 109 +++++++++++++++++++++++++++++++++
 1 file changed, 109 insertions(+)

diff --git a/crates/odin-daemon/src/main.rs b/crates/odin-daemon/src/main.rs
index 5c2b03f..509e499 100644
--- a/crates/odin-daemon/src/main.rs
+++ b/crates/odin-daemon/src/main.rs
@@ -3401,6 +3401,115 @@ mod tests {
         Ok(())
     }
 
+    /// Two puts pipelined on one session with the first datagram lost on the
+    /// way: the transport resends it and delivers both in order, and Odin
+    /// accepts, and so writes, both. The transport may acknowledge the second
+    /// before the first is resent (see `accept_raw_document`); what holds is
+    /// that neither put is lost and both are on disk once delivered.
+    #[test]
+    fn pipelined_puts_with_a_lost_datagram_are_both_written() -> Result<()> {
+        let mut odin = activated_odin()?;
+        odin.pass()?;
+        let server = odin.server.local_addr()?;
+        let acknowledged = odin.serve_while(move || -> Result<bool> {
+            // A relay that drops the client's first reliable data datagram.
+            let relay_socket = UdpSocket::bind("127.0.0.1:0")?;
+            relay_socket.set_read_timeout(Some(Duration::from_millis(5)))?;
+            let relay_addr = relay_socket.local_addr()?;
+            let stop = Arc::new(AtomicBool::new(false));
+            let relay = thread::spawn({
+                let stop = stop.clone();
+                move || -> Result<()> {
+                    let mut client = None;
+                    let mut dropped = false;
+                    let mut buffer = vec![0u8; 65_535];
+                    while !stop.load(Ordering::Relaxed) {
+                        let Ok((length, from)) = relay_socket.recv_from(&mut buffer) else {
+                            continue;
+                        };
+                        let datagram = &buffer[..length];
+                        if from == server {
+                            if let Some(client) = client {
+                                relay_socket.send_to(datagram, client)?;
+                            }
+                            continue;
+                        }
+                        client = Some(from);
+                        let reliable_data = cultnet_rs::decode_rudp_packet(datagram).is_ok_and(
+                            |packet| {
+                                packet.packet_type == cultnet_rs::CultNetRudpPacketType::Data
+                                    && packet.reliable
+                            },
+                        );
+                        if reliable_data && !dropped {
+                            dropped = true;
+                            continue;
+                        }
+                        relay_socket.send_to(datagram, server)?;
+                    }
+                    Ok(())
+                }
+            });
+            let socket = UdpSocket::bind("127.0.0.1:0")?;
+            socket.set_read_timeout(Some(Duration::from_millis(5)))?;
+            let mut transport = CultNetRudpSocketTransportConnection::new(
+                CultNetRudpSocketTransportOptions::client("pipeliner", socket, relay_addr, 0x6100_0001),
+            )?;
+            transport.connect(Vec::new())?;
+            let deadline = Instant::now() + Duration::from_secs(3);
+            while !transport.connected() {
+                let _ = transport.receive_once()?;
+                transport.poll_resends()?;
+                ensure!(Instant::now() < deadline, "timed out connecting");
+            }
+            let put = |key: &str| -> Result<Vec<u8>> {
+                let message = CultNetMessage::DocumentPutRaw {
+                    message_id: format!("put-{key}"),
+                    document: CultNetRawDocumentRecord {
+                        schema_id: "ghostlight.doc.v1".into(),
+                        record_key: key.into(),
+                        stored_at: rfc3339_millis(unix_millis()?)?,
+                        payload_encoding: CultNetRawPayloadEncoding::Messagepack,
+                        payload: vec![7; 64],
+                        source_runtime_id: None,
+                        source_agent_id: None,
+                        source_role: None,
+                        tags: None,
+                    },
+                };
+                Ok(encode_cultnet_message_to_vec(&message, CultNetWireContract::CultNetSchemaV0)?)
+            };
+            let receipts = [
+                transport.send_reliable("schema", put("first")?)?,
+                transport.send_reliable("schema", put("second")?)?,
+            ];
+            let deadline = Instant::now() + Duration::from_secs(5);
+            let all_acknowledged = |transport: &CultNetRudpSocketTransportConnection| {
+                receipts.iter().all(|receipt| {
+                    transport.reliable_send_status(receipt)
+                        == CultNetRudpReliableSendStatus::Acknowledged
+                })
+            };
+            while !all_acknowledged(&transport) && Instant::now() < deadline {
+                let _ = transport.receive_once()?;
+                transport.poll_resends()?;
+            }
+            stop.store(true, Ordering::Relaxed);
+            relay.join().expect("relay thread panicked")?;
+            Ok(all_acknowledged(&transport))
+        })??;
+        assert!(acknowledged, "both puts were delivered");
+        let written: Vec<String> = odin
+            .file_records()?
+            .into_iter()
+            .filter(|entry| entry.r#type == "ghostlight.doc")
+            .map(|entry| entry.key)
+            .collect();
+        assert_eq!(written, vec!["first".to_owned(), "second".to_owned()]);
+        assert_eq!(odin.file_records()?, odin.working_set());
+        Ok(())
+    }
+
     /// A put whose write is not made is refused, so it is never accepted:
     /// with the write failing, and with the lease lost.
     #[test]