[Musyg]

This is #112 by @frontierlabs-admin, rebased on current main. The two commits keep their author. Fixes #111.

#112 no longer applies to main. The rebase adapts it to:

• AtomicWaker (Replace TCP read and write waker mutexes with AtomicWaker #113): wakers use register/wake;
• TcpFlags (Make TCP flag checks more readable #115): the FIN arm matches ACK|FIN and ACK|PSH|FIN by name;
• the deferred shutdown signal (Wake deferred TCP shutdown after the final ACK #114): a shutdown in CloseWait waits on it, and the main loop now also raises it in CloseWait once nothing is in flight.

The behaviour is otherwise the one described in #112. Its 16 regression tests are included and pass, with the existing ones (72 unit tests). fmt and clippy are clean.

This is the first part of #118, split as you asked. The next ones (ACK validation, FIN retransmission and crossed closes, close after drop, Linux qualification) will follow one at a time.

[chatgpt-codex-connector]

Codex Review Summary: ❌ Failed

This comment shows the latest Codex review activity on this pull request.

Review	Status	Commit	Review trigger
📝 Code Review	❌ Failed 2026-11-09T19:08:32.697992Z We couldn’t complete this review. Comment @codex review to try again.	4f1e7b3	PR opened

ℹ️ About Codex in GitHub

Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you

• Open a pull request for review
• Mark a draft as ready
• Comment "@codex review" or "@codex security review".

Codex reacts with 👀 while any review is running, comments if it has suggestions, and reacts with 👍 once all reviews finish with no findings.

[chatgpt-codex-connector]

You have reached your Codex usage limits for code reviews. You can see your limits in the Codex usage dashboard.
To continue using code reviews, add credits to your account and enable them for code reviews in your settings.

[SajjadPourali]

Please check these tests

--- a/src/stream/tcp.rs
+++ b/src/stream/tcp.rs
@@ -1837,8 +1837,48 @@
             .expect("no end of stream after the peer's FIN")
             .unwrap();
         data
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn regression_delayed_response_after_peer_fin() {
+        let (mut stream, mut up_rx, local_seq) = established().await;
+        stream.stream_sender().send(peer_segment(1001, local_seq, ACK | FIN, u16::MAX, &[])).unwrap();
+        written(&mut up_rx).await;
+        pass(TcpConfig::default().close_wait_timeout + Duration::from_millis(100)).await;
+        let result = stream.write_all(b"delayed response").await;
+        assert!(result.is_ok(), "half-closed stream rejected delayed response: {result:?}");
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn regression_drop_retransmits_unacknowledged_response() {
+        let (mut stream, mut up_rx, local_seq) = established().await;
+        stream.stream_sender().send(peer_segment(1001, local_seq, ACK | FIN, u16::MAX, &[])).unwrap();
+        written(&mut up_rx).await;
+        stream.write_all(b"response").await.unwrap();
+        let original = up_rx.recv().await.unwrap();
+        assert_eq!(original.payload.as_deref(), Some(&b"response"[..]));
+        drop(stream);
+        written(&mut up_rx).await;
+        pass(RTO + Duration::from_millis(100)).await;
+        let packets = written(&mut up_rx).await;
+        assert!(packets.iter().any(|p| p.payload.as_deref() == Some(&b"response"[..])),
+            "unacknowledged response was not retransmitted after drop");
     }

+    #[tokio::test(start_paused = true)]
+    async fn regression_fin_retry_preserves_sequence_number() {
+        let (mut stream, mut up_rx, local_seq) = established().await;
+        stream.stream_sender().send(peer_segment(1001, local_seq, ACK | FIN, u16::MAX, &[])).unwrap();
+        written(&mut up_rx).await;
+        let mut cx = Context::from_waker(Waker::noop());
+        assert!(Pin::new(&mut stream).poll_shutdown(&mut cx).is_pending());
+        let original = fins_written(&mut up_rx).await;
+        assert_eq!(original.len(), 1);
+        pass(TcpConfig::default().last_ack_timeout).await;
+        let retry = fins_written(&mut up_rx).await;
+        assert_eq!(retry, original, "FIN retransmission must reuse the original sequence number");
+    }
+
     /// RFC 9293 §3.6: a FIN closes only the peer's direction. The stack acknowledges it and keeps
     /// sending, and its own FIN waits for the application to shut down, which takes it to LastAck.
     #[tokio::test(start_paused = true)]

[SajjadPourali]

Could we keep the write side open until the application shuts it down, or make this timeout opt-in? A client can send a request and half-close, then wait more than five seconds for the server to prepare a response. In that case this timer sends our FIN before the application writes the response. Could we add a test for a delayed response

[SajjadPourali]

What should happen if response data is still in flight here? This sends FIN and then aborts the task that retransmits unacknowledged data. If a response segment was lost, the peer can be left waiting for the missing bytes. The abort behavior already exists on main, but this new path also sends FIN; could we cover this case or track it separately?

[SajjadPourali]

The initial FIN increments tcb.seq, so this retry appears to use the sequence number after the original FIN. Should the retry use the original FIN sequence instead? This logic is inherited from main, but the new shutdown path calls it. The retransmission test could assert that both FINs have the same sequence number.

[Musyg]

Thanks for the review and the tests. Pushed 7c804fd and 6c0a4e0:

1. CloseWait no longer has a timer. After the peer's FIN, the write side stays open until the application shuts it down or drops the stream. close_wait_timeout stays in TcpConfig so the API does not change, documented as unused (a #[deprecated] would need a minor release). Your delayed-response test is included, plus one that waits 30 s before replying.
2. task_last_ack now resends our FIN at its own sequence number. Your test is included; it fails before the change (102 instead of 101).
3. Drop with data in flight: as you noted, this is inherited from main, so I kept it out of this PR. It is fixed in part 4 of the split (close after drop), where the session task outlives the stream until the data and the FIN are acknowledged. Your test is included there.

The CI run on 6c0a4e0 is waiting for your approval. cargo semver-checks passes locally.