Make peer_connection_listener persistent

[sanity]

Summary

While debugging the flaky PUT/connectivity tests we instrumented the transport layer and found that peer_connection_listener (the task that owns each UDP socket) returns after handing off a single inbound packet. The main event loop immediately re-spawns a new listener for that connection, but if anything in the select! chain fails to push the returned future back into the stream, that connection stops draining inbound packets entirely. That’s exactly what we’re seeing: the intermediate node logs that it successfully queued and sent PutForward, yet the requester never logs any inbound message for that transaction—the listener simply isn’t polled again.

This “spawn-per-packet” architecture couples inbound progress to the main event loop’s scheduling and is brittle under load. A more robust design is the standard 1:1 model: one long-lived async task per peer connection that continuously drives both the outbound mpsc and inbound datagrams, forwarding inbound packets via a channel to the event loop. Only connection lifecycle events should be sent back through the priority select. This decouples transport I/O from the higher-level routing logic and eliminates the possibility that a bookkeeping hiccup starves a connection.

Evidence

• In /tmp/connectivity_trace_run10.log we see the intermediate node (v6MWKgqHeCHyzLfJ) log Sending outbound message… PutForward and the peer_connection_listener log that it wrote the packet to socket 127.0.0.1:38147. However, the requester (v6MWKgqK3rzUh1F6) never logs a matching Received message… PutForward, the PUT result never reaches the operation manager, and the test times out.
• There are no [CONN_LIFECYCLE] … closed logs for that socket, so the connection itself is still open; it simply isn’t polled after the first packet.
• The direct-ack “shortcut” we added earlier doesn’t fire either, because the final hop never processes the PutForward message, so no SuccessfulPut is generated at all.

Proposal

1. Refactor peer_connection_listener into a persistent task (one per connection) that loops forever, driving both outbound mpsc and inbound UDP reads. When it receives data it sends a ConnEvent::InboundMessage into a new conn_events channel that the event loop polls alongside the existing sources.
2. Update priority_select::SelectResult and process_select_result to consume these ConnEvents instead of whole PeerConnectionInbound structs, eliminating the need to re-spawn futures on every packet.
3. Keep connection lifecycle handling (drop, errors, etc.) inside that task; it can emit ConnEvent::ClosedChannel when necessary so the event loop cleans up state.
4. Once that’s in place, remove the direct-ack workaround (tracked in issue Clean up PUT direct-ack shortcut #2077) so we rely solely on the proper upstream response path.

This aligns the transport code with a more conventional, maintainable architecture and prevents individual connections from being starved because the select! loop lost track of their listener future.

[github-actions]

🤖 Auto-labeled

Applied labels:

• T-bug (99% confidence)
• P-high (90% confidence)
• A-networking (98% confidence)

Reasoning: The report describes incorrect behavior: peer_connection_listener exits after one packet causing dropped inbound messages and flaky PUT/connectivity tests, which is a bug. It affects core transport reliability and tests, so priority is high. The fix requires a non-trivial refactor of the transport/select loop (moderate effort). This is squarely a networking-area issue. The submitter proposes an architectural change, so a design discussion is likely warranted.

Previous labels: none

If these labels are incorrect, please update them. This helps improve the auto-labeling system.

[iduartgomez]

Please before even thinking about refactoring point exactly what changes are you planning to do so it can be reviewed. PeerConnections are already independent tasks and idk what spawn per packet refers to (sounds wrong).

[sanity]

@iduartgomez Here is the concrete change set, plus why the failure shows up on this branch:

1. Add a dedicated conn_events channel. Each peer-connection task stays running and feeds ConnEvent::{InboundMessage, ClosedChannel} into that channel instead of returning a PeerConnectionInbound bundle. No more respawning a listener after every packet.
2. Update priority_select::SelectResult and process_select_result to consume these ConnEvent messages directly. The current PeerConnection variant disappears; ConnEvent::OutboundMessage remains the path for outgoing traffic.
3. When a connection is established (inbound or outbound) we spawn exactly one async task for it. That task continuously drives the outbound mpsc receiver and the UDP socket, forwarding inbound packets via conn_events and emitting ConnEvent::ClosedChannel if the connection fails so the event loop can prune state and notify handshake.

Background: the existing design dates back to when the event loop expected to own each connection. Transport now hands off a PeerConnectionInbound (socket + channel + first message) and asks the loop to spawn a fresh listener immediately. Under heavier concurrency that handoff is fragile—if the select loop delays that re-spawn, the connection stops draining packets and we lose responses.

Why this surfaced only here: on main the flaky PUT/connectivity tests mostly run along a single hop, so they rarely expose the transport issue. In this branch we changed test_put_contract_three_hop_returns_response to deliberately pick the farthest peer as the client and the nearest peer as storage, and the freenet_test macro now pushes small networks toward a fully connected mesh. That makes the test exercise the full three-hop PUT routing every run, reliably triggering the latent transport bug. The architecture fix above addresses the underlying problem rather than masking it with the direct-ack shortcut.

[iduartgomez]

For clarification:

1. Each message is not a single packet, IDK where this is coming from. It may or may not fit a single packet (if you are not sending a contract it likely does).
2. Having a persistent task consuming the inbound messages or processing outbound messages sounds ok, but it just moves the problem elsewhere... If now the channel sending back the messages is blocked in the select , that is gonna provoke problems anyway, unless some sort of buffering is added, and even then, it will at some point spill over if there are problems that block consumption from that task products.

The change sounds ok, but does not seem to address the core issue which was alluded to initially here:

but if anything in the select! chain fails to push the returned future back into the stream, that connection stops draining inbound packets entirely

The futures should always be pushed back with priority since is the hot path. Why is not being pushed? That event loop should be blocking on anything and should move quickly.

[sanity]

@iduartgomez Thanks for the clarifications; here’s a straight response to each point.

• The “per packet” wording was sloppy on my part. The code in p2p_protoc spawns peer_connection_listener, and that future blocks until PeerConnection::recv produces a full NetMessage (which may span multiple UDP datagrams). Once it returns that NetMessage, the select loop re-spawns the listener. So the unit is a NetMessage, not a single datagram.

• In the failing run (/tmp/connectivity_trace_run10.log), peer v6MWKgqHeCHyzLfJ logged Sending outbound message … PutForward(01K9ST1F9TQPB508DSJYV5D801) and the matching [CONN_LIFECYCLE] peer_connection_listener received outbound work / Sending message to peer lines. The destination v6MWKgqK3rzUh1F6 never logged Received message … PutForward or the corresponding process_message entry, even though later messages (RequestSub, ReturnSub) do arrive on that same socket. That’s the gap the deterministic three-hop test is exposing: the socket stays open, but that specific NetMessage never surfaces.

• The future is still being requeued—the trace continues to show “PrioritySelect: peer_connections READY.” The issue isn’t that the select loop fails to push the future back, it’s that even after the listener is re-spawned the NetMessage we just transmitted never reappears on the inbound path. With the current architecture there’s nowhere between PeerConnection::recv and the event loop where we can see whether the decoded message was queued or dropped, which makes it hard to debug.

• Moving to a persistent task plus a conn_events channel doesn’t solve the bug by itself, and it would still need backpressure. The benefit is that all inbound/outbound work for a connection lives in that task, and the dispatcher consumes a single stream of ConnEvents. If the dispatcher stalls we see the queue backing up instead of silently losing the next NetMessage because the listener wasn’t reinserted fast enough. It also gives us a single place to instrument when we continue digging into PeerConnection::recv.

• This only showed up in this branch because we changed test_put_contract_three_hop_returns_response and the freenet_test macro to force gateway → intermediate → client routing on every run. On main the test usually stays on one hop, so the same transport issue rarely occurs—here it reproduces consistently.

I’m not claiming the architectural change fixes the missing NetMessage outright; it removes the brittle handoff so we can actually observe whether PeerConnection::recv produced the message and where it went next. I’m still planning to dig into the reassembly path after that, but I wanted to lay out the transport adjustments first so we’re not chasing both problems at once.

[iduartgomez]

Sounds good, let's do the change and see what else we find.

That said, the future being pushed back, means that messages wouldn't be lost cause messages are not lost when send over the send halv of the channel and eventually would be consumed.

The problem may be within the transport layer itself then.

[sanity]

@iduartgomez Before I rip apart the transport/event-loop glue, I’d like to take the smallest possible step that still gets us data on the failure. Here’s what I’m proposing instead:

1. Keep the current peer_connection_listener structure for now and add explicit instrumentation inside PeerConnection::recv, the packet reassembly paths, and handle_peer_connection_msg. That will tell us whether the gateway actually decodes the PutForward NetMessage and where it disappears (e.g., dropped before it reaches the dispatcher).
2. Use that logging to chase the concrete loss case. If we can prove the NetMessage is being decoded but not enqueued, we’ll fix that directly without touching the event loop. If the data points squarely at the listener handoff, then we’ll go ahead with the persistent-task refactor, but with evidence that it’s necessary.

This keeps the CI goal in sight and avoids a large, risky refactor unless the instrumentation shows it’s required. If that direction sounds reasonable, I’ll start by adding the targeted logs and rerunning the flaky tests so we have clearer data.

[iduartgomez]

Yes, that sounds a better approach overall.