Feat/tcp connecotr hardening#179
Conversation
…rt and connection handling improvements
thaodt
left a comment
thaodt
left a comment
There was a problem hiding this comment.
Approved with some small changes then we're good to go.
| let mut listener_a = TcpListener::new(server_stack, 7001u16, buf(), buf()); | ||
| let mut listener_b = TcpListener::new(server_stack, 7002u16, buf(), buf()); | ||
| let dialer_a = TcpDialer::new(client_stack, endpoint(7001), buf(), buf()); | ||
| let dialer_b = TcpDialer::new(client_stack, endpoint(7002), buf(), buf()); |
There was a problem hiding this comment.
Lets make this test exercise the pooled-listener path it is intended to cover.
This creates two unrelated TcpListener<1> values on different ports and never calls TcpListener<N>::with_buffers or TcpServer<N>.
The test therefore passes even if same-port fan-out or pooled worker creation is broken and no other in-tree test exercises with_buffers, please use one pooled N=2 endpoint with two clients.
There was a problem hiding this comment.
Reworked it to one pooled TcpListener::<2>::with_buffers(...) on a single port with two clients, so it now fails if same-port fan-out or pooled-socket creation regresses.
Reaching the N>1 pool from an integration test needed a public accept — the Listener impl is N=1-only and the pooled path is otherwise reachable only through TcpServer<N> + the session engine, which this transport-level smoke deliberately avoids. So I added TcpListener::<N>::accept_on(index) and de-duped the accept body it now shares with the N=1 impl and the server workers. Bonus: with_buffers is drivable directly instead of only through TcpServer.
Please let me know what you think!
There was a problem hiding this comment.
This addresses my original concern.
One non-blocking wording nit: see #179 (comment)
…prove time scaling in tests
…atchdog The loopback harness polls two non-terminating embassy-net stacks in the background, so a transport regression or a dropped crossover packet would leave `block_on` pending until the outer CI timeout instead of failing the test. Race the foreground/background `select` against a 20s wall-clock watchdog that re-arms its waker each poll (so the deadline is observed even when the stacks would otherwise park) and panics with a clear message. Addresses review feedback on #179. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
…ed N=2 path The Embassy loopback concurrency test used two unrelated `TcpListener<1>` on different ports, so it never touched `with_buffers` construction or same-port fan-out. Rework it to stand up one `TcpListener::<2>::with_buffers(...)` on a single port and dial it with two clients. Reaching the N>1 pool from an integration test required a public accept: the `Listener` impl is `N=1`-only and the pooled path (`into_server_futures`/ `serve_socket_slot`) was private, reachable only through `TcpServer<N>` + the AimX session engine. Add `TcpListener::<N>::accept_on(index)` and factor the accept body — previously duplicated between the `N=1` `Listener` impl and the server workers — into a shared `accept_on_slot`. Side benefit: `with_buffers` can now be driven directly instead of only through `TcpServer`. Addresses review feedback on #179. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
| /// Each client lands on a distinct pooled slot; a broken pool (only one socket | ||
| /// accepting, or both racing to the same slot) would hang the second session and | ||
| /// trip the watchdog. (Wiring the pool into the AimX session engine via | ||
| /// `TcpServer<N>` is covered elsewhere; this stays at the transport layer.) |
There was a problem hiding this comment.
nit: it says TcpServer<N> is "covered elsewhere", but I could not find another in-tree test of the embassy TcpServer<N> worker path. It would be more accurate to say that path is intentionally outside this transport-level smoke.
| pub fn accept_on( | ||
| &self, | ||
| index: usize, | ||
| ) -> impl Future<Output = TransportResult<Box<dyn Connection>>> + Send + '_ { |
There was a problem hiding this comment.
When a checked-out slot has 2 separate tasks waiting on accept_on(index), both call poll_take, but TcpSocketSlot stores only one Option<Waker>, the 2nd registration overwrites the first, so after the socket returns only one task wakes while the other can remain pending forever.
Because this new method takes &self, safe callers can create this state, so enforce one waiter per index or support multiple waiters.
| let mut socket = poll_fn(|cx| slot.poll_take(cx)).await; | ||
| socket.abort(); | ||
| match socket.accept(local_endpoint).await { |
There was a problem hiding this comment.
When an accept_on future is dropped after taking the socket but while TcpSocket::accept is pending - for e.g., when a timeout or shutdown select wins, the async local drops the TcpSocket instead of calling slot.put. The slot remains None, so every later accept on that index waits forever, you may want to retain the socket in a drop guard or otherwise return it on cancellation.
2 participants
Changes
accept()errors.embassy-net'sTcpSocket::accept()can fail synchronously (a port-0local_endpointrejected asInvalidPort), and theErrarms ofserve_socket_slotand theListener<1>compat path had no await point — a tight loop with zero yields that starves the cooperative executor. Both nowembassy_futures::yield_now().await, degrading a static misconfig to a debuggable warn-loop. New optionalembassy-futuresdep._test-embassy-loopback). The socket pool is welded to a concreteembassy_net::tcp::TcpSocket, so its recycling and waker handoff were review-only. Now exercised over two realembassy-netstacks wired by an in-memory driver-channel crossover: recycle → re-accept, concurrent accept slots, and dialer redial after a failed connect / dropped link. Uses a wall-clock host time driver (the frozen stub clock stalls the delayed-ACK timer and hangs the firstflush()). Kept offembassy-runtimeand out of[dev-dependencies]; wired intomake check.TcpSocketSlotSAFETY comment (dialer/worker/listener-owned, not just "dialer-owned") and documented the intentional double-abort()on the recycle path.aimdb-tcp-connector/CHANGELOG.md(initial) + global changelog index link.