From 48f36eb7bc9f853342f932eb2f4c06875f7bd8cf Mon Sep 17 00:00:00 2001 From: Teo Calin Date: Thu, 8 Oct 2026 12:04:03 +0300 Subject: [PATCH 1/4] daemon: a FIN waits for the data sent before it A FIN that overtook a lost segment was acted on at once: the receiver closed the stream with the segment missing and answered with a FIN-ACK, and that FIN-ACK made the sender drop the segment from its retransmit queue. Under loss a writer that closed right after its last write lost the end of it: 4 of 30 pub/sub events at 5% loss never reached the broker while the publisher reported success. Every version behaves this way. Receiver: a FIN ahead of undelivered data is remembered, and an immediate ACK with SACK asks for the gap; the FIN is handled (receive side closed, TIME_WAIT, FIN-ACK) once delivery reaches it. A side that has already closed its own receive side still takes the peer's FIN at once. Sender: CloseConnection holds the FIN until its data is acknowledged, at most finLinger (5 s), in a goroutine, so receivers that have not upgraded get the whole write too. FIN placement and the FIN retransmit rules are unchanged. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 12 + pkg/daemon/daemon.go | 226 +++++++++++++----- pkg/daemon/ports.go | 5 + pkg/daemon/zz_fin_in_order_test.go | 279 ++++++++++++++++++++++ pkg/daemon/zz_large_write_pieces_test.go | 12 +- pkg/daemon/zz_segment_fits_packet_test.go | 5 +- pkg/daemon/zz_streampacket_test.go | 3 + 7 files changed, 471 insertions(+), 71 deletions(-) create mode 100644 pkg/daemon/zz_fin_in_order_test.go diff --git a/CHANGELOG.md b/CHANGELOG.md index 57a11834..10361d80 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -394,6 +394,18 @@ Detailed per-release notes are on the `appstore upgrade` does not wait. ### Fixed +- **A write followed at once by a close no longer loses its end under + loss.** A FIN that overtook a lost segment was acted on at once: the + receiver closed the stream with the segment missing, and its FIN-ACK made + the sender drop the segment it was about to resend. Every version did + this. In the lab, 4 of 30 pub/sub events at 5% loss never reached the + broker while `pilotctl publish` reported success; one-shot sends and any + client that closes right after writing were exposed the same way. A + receiver now handles the FIN only once the data before it has arrived, + and asks for what is missing meanwhile. A closing sender also holds its + FIN until its data is acknowledged (5 s at most, without blocking the + caller), so the end of a write survives loss on receivers that have not + upgraded too. - **A send error no longer strands the end of a stream write.** When the tunnel refused a segment of a write's last piece (or of any write up to 56 segments), the daemon returned the error and left the rest of the write diff --git a/pkg/daemon/daemon.go b/pkg/daemon/daemon.go index a258c472..ecd38d0c 100644 --- a/pkg/daemon/daemon.go +++ b/pkg/daemon/daemon.go @@ -475,6 +475,10 @@ type Daemon struct { // (see dialConnectionLocked): dialOKSeq counts completed dials, and a // dial compares it with the count it saw when it started. A counter, // not lastDialOKNano, so a wall-clock step cannot hide timeouts. + // finLinger bounds the wait for a closing connection's data to be + // acknowledged before its FIN goes out (sendFINAfterData). + finLinger time.Duration + lastDialOKNano atomic.Int64 dialOKSeq atomic.Uint64 consecutiveDialTimeouts atomic.Uint64 @@ -664,6 +668,7 @@ func New(cfg Config) *Daemon { tunnels: NewTunnelManager(), ports: NewPortManager(), stopCh: make(chan struct{}), + finLinger: defaultFinLinger, hsPoll: newHandshakePollSched(), synTokens: cfg.synRateLimit(), synLastFill: time.Now(), @@ -3594,55 +3599,34 @@ func (d *Daemon) handleStreamPacket(pkt *protocol.Packet) { // FIN — remote close (or FIN-ACK acknowledging our FIN) if pkt.HasFlag(protocol.FlagFIN) { conn := d.ports.FindConnection(pkt.DstPort, pkt.Src, pkt.SrcPort) - if conn != nil { - conn.CloseRecvBuf() - conn.Mu.Lock() - wasFinWait := conn.State == StateFinWait - wasTimeWait := conn.State == StateTimeWait - conn.State = StateTimeWait - // Only refresh LastActivity on the FIRST FIN. A peer's FIN-ACK - // reply has FlagFIN set too; if we refresh on every FIN we keep - // idleSweepLoop from reaping this conn, which combined with the - // guarded send below would have created a packet-amplification - // loop (storm) at line rate. - if !wasTimeWait { - conn.LastActivity = time.Now() - } - conn.KeepaliveUnacked = 0 - sendSeq := conn.SendSeq - conn.Mu.Unlock() - // If we were in FIN_WAIT, this is a FIN-ACK — clear retx buffer - if wasFinWait { - conn.RetxMu.Lock() - conn.Unacked = nil - conn.RetxMu.Unlock() - } - if !wasTimeWait { - d.publishEvent("conn.fin", map[string]interface{}{ - "remote_addr": pkt.Src.String(), "remote_port": pkt.SrcPort, - "local_port": pkt.DstPort, "conn_id": conn.ID, - }) - } - // Connection will be reaped by idleSweepLoop after TimeWaitDuration - - // Send FIN-ACK once, only on the FIRST FIN. Subsequent FINs are - // our peer's FIN-ACK reply (or duplicates); echoing them back - // triggers a storm because FIN-ACK has FlagFIN set. - if !wasTimeWait { - finack := &protocol.Packet{ - Version: protocol.Version, - Flags: protocol.FlagFIN | protocol.FlagACK, - Protocol: protocol.ProtoStream, - Src: conn.LocalAddr, - Dst: pkt.Src, - SrcPort: pkt.DstPort, - DstPort: pkt.SrcPort, - Seq: sendSeq, - Ack: pkt.Seq + 1, - } - d.tunnels.Send(pkt.Src.Node, finack) - } + if conn == nil { + return + } + // A FIN is the end of the peer's stream: it may only be acted on + // once everything sent before it has been delivered. Acting on a FIN + // that overtook a lost segment closed the stream with that data + // missing — and the FIN-ACK made the sender drop the segment it was + // about to resend — so a writer that closed right after its last + // write lost the end of it under loss (pub/sub events, one-shot + // sends). Hold the FIN until delivery reaches it, and ask for the + // missing data meanwhile. + // Only while this side still delivers data: once it has closed its + // own receive side (FIN_WAIT), the data would be discarded anyway. + conn.Mu.Lock() + receiving := conn.State == StateEstablished + conn.Mu.Unlock() + conn.RecvMu.Lock() + early := receiving && !conn.RecvClosed && seqAfter(pkt.Seq, conn.ExpectedSeq) + if early { + conn.finPending = true + conn.finSeq = pkt.Seq } + conn.RecvMu.Unlock() + if early { + d.sendDelayedACK(conn) + return + } + d.acceptFIN(conn, pkt.Seq) return } @@ -3734,6 +3718,20 @@ func (d *Daemon) handleStreamPacket(pkt *protocol.Packet) { conn.RecvAck = cumAck conn.Mu.Unlock() + // A FIN that arrived ahead of this data is handled now that + // delivery has reached it. + conn.RecvMu.Lock() + finReady := conn.finPending && !seqAfter(conn.finSeq, cumAck) + finSeq := conn.finSeq + if finReady { + conn.finPending = false + } + conn.RecvMu.Unlock() + if finReady { + d.acceptFIN(conn, finSeq) + return + } + // Check if we have out-of-order data — ACK immediately with SACK conn.RecvMu.Lock() hasOOO := len(conn.OOOBuf) > 0 @@ -3774,6 +3772,61 @@ func (d *Daemon) handleStreamPacket(pkt *protocol.Packet) { } } +// acceptFIN handles the peer's FIN once every byte sent before it has been +// delivered (see the FIN branch of handleStreamPacket): it closes the +// receive side, moves to TIME_WAIT and answers with a FIN-ACK. +func (d *Daemon) acceptFIN(conn *Connection, finSeq uint32) { + conn.CloseRecvBuf() + conn.Mu.Lock() + wasFinWait := conn.State == StateFinWait + wasTimeWait := conn.State == StateTimeWait + conn.State = StateTimeWait + // Only refresh LastActivity on the FIRST FIN. A peer's FIN-ACK + // reply has FlagFIN set too; if we refresh on every FIN we keep + // idleSweepLoop from reaping this conn, which combined with the + // guarded send below would have created a packet-amplification + // loop (storm) at line rate. + if !wasTimeWait { + conn.LastActivity = time.Now() + } + conn.KeepaliveUnacked = 0 + sendSeq := conn.SendSeq + localAddr, localPort := conn.LocalAddr, conn.LocalPort + remoteAddr, remotePort := conn.RemoteAddr, conn.RemotePort + conn.Mu.Unlock() + // If we were in FIN_WAIT, this is a FIN-ACK — clear retx buffer + if wasFinWait { + conn.RetxMu.Lock() + conn.Unacked = nil + conn.RetxMu.Unlock() + } + if !wasTimeWait { + d.publishEvent("conn.fin", map[string]interface{}{ + "remote_addr": remoteAddr.String(), "remote_port": remotePort, + "local_port": localPort, "conn_id": conn.ID, + }) + } + // Connection will be reaped by idleSweepLoop after TimeWaitDuration + + // Send FIN-ACK once, only on the FIRST FIN. Subsequent FINs are + // our peer's FIN-ACK reply (or duplicates); echoing them back + // triggers a storm because FIN-ACK has FlagFIN set. + if !wasTimeWait { + finack := &protocol.Packet{ + Version: protocol.Version, + Flags: protocol.FlagFIN | protocol.FlagACK, + Protocol: protocol.ProtoStream, + Src: localAddr, + Dst: remoteAddr, + SrcPort: localPort, + DstPort: remotePort, + Seq: sendSeq, + Ack: finSeq + 1, + } + d.tunnels.Send(remoteAddr.Node, finack) + } +} + // sendDelayedACK sends a cumulative ACK for a connection, including SACK blocks if needed. func (d *Daemon) sendDelayedACK(conn *Connection) { // Reset delayed ACK state @@ -5092,7 +5145,6 @@ func (d *Daemon) CloseConnection(conn *Connection) { }) } if st == StateEstablished { - finData := []byte{0} // 1-byte sentinel so retxEntry has non-zero length fin := &protocol.Packet{ Version: protocol.Version, Flags: protocol.FlagFIN, @@ -5103,22 +5155,16 @@ func (d *Daemon) CloseConnection(conn *Connection) { DstPort: remotePort, Seq: sendSeq, } - d.tunnels.Send(remoteAddr.Node, fin) - // Track FIN in retransmission buffer so the retxLoop retries it. - // Use isFIN=true so retransmitUnacked can distinguish the FIN entry - // from regular data entries (which must be retransmitted as data even - // when the connection is in StateFinWait). - conn.RetxMu.Lock() - now := time.Now() - conn.Unacked = append(conn.Unacked, &retxEntry{ - data: finData, - seq: sendSeq, - sentAt: now, - origSentAt: now, - attempts: 1, - isFIN: true, - }) - conn.RetxMu.Unlock() + // A receiver up to v1.17.0 acts on a FIN that overtook a lost + // segment, closing the stream without it. Send the FIN once the + // data is acknowledged (finLinger at most), so the end of a write + // that is followed at once by a close survives loss on such + // receivers too. The caller does not wait. + if d.unackedHasData(conn) { + go d.sendFINAfterData(conn, fin) + } else { + d.sendFIN(conn, fin) + } } conn.CloseRecvBuf() // P1-003: stop a pending delayed-ACK timer so it doesn't fire after @@ -5138,6 +5184,56 @@ func (d *Daemon) CloseConnection(conn *Connection) { conn.Mu.Unlock() } +// defaultFinLinger bounds how long a closing connection waits for its data +// to be acknowledged before it sends its FIN anyway (Daemon.finLinger). +const defaultFinLinger = 5 * time.Second + +// sendFIN puts the FIN on the wire and tracks it for retransmission. +func (d *Daemon) sendFIN(conn *Connection, fin *protocol.Packet) { + d.tunnels.Send(fin.Dst.Node, fin) + // Track FIN in retransmission buffer so the retxLoop retries it. + // Use isFIN=true so retransmitUnacked can distinguish the FIN entry + // from regular data entries (which must be retransmitted as data even + // when the connection is in StateFinWait). + conn.RetxMu.Lock() + now := time.Now() + conn.Unacked = append(conn.Unacked, &retxEntry{ + data: []byte{0}, // 1-byte sentinel so retxEntry has non-zero length + seq: fin.Seq, + sentAt: now, + origSentAt: now, + attempts: 1, + isFIN: true, + }) + conn.RetxMu.Unlock() +} + +// sendFINAfterData sends the FIN once every data segment is acknowledged, +// or after finLinger. It gives up if the connection goes away, the daemon +// stops, or the peer's own FIN closed the connection meanwhile (its +// FIN-ACK then stands for ours). +func (d *Daemon) sendFINAfterData(conn *Connection, fin *protocol.Packet) { + deadline := time.Now().Add(d.finLinger) + tick := time.NewTicker(20 * time.Millisecond) + defer tick.Stop() + for d.unackedHasData(conn) && time.Now().Before(deadline) { + select { + case <-conn.RetxStop: + return + case <-d.stopCh: + return + case <-tick.C: + } + } + conn.Mu.Lock() + st := conn.State + conn.Mu.Unlock() + if st != StateFinWait { + return + } + d.sendFIN(conn, fin) +} + // SendDatagram sends an unreliable unicast packet. Broadcast addresses // (Node=0xFFFFFFFF) are rejected — use BroadcastDatagram, which requires // an admin token. diff --git a/pkg/daemon/ports.go b/pkg/daemon/ports.go index e9a66419..667815fe 100644 --- a/pkg/daemon/ports.go +++ b/pkg/daemon/ports.go @@ -336,6 +336,11 @@ type Connection struct { RecvMu sync.Mutex ExpectedSeq uint32 // next in-order seq expected OOOBuf []*recvSegment // out-of-order buffer + // finPending is set when the peer's FIN arrived ahead of data it sent + // before it; finSeq is the FIN's sequence number. The FIN is handled + // once delivery reaches finSeq (see Daemon.acceptFIN). Guarded by RecvMu. + finPending bool + finSeq uint32 // Delayed ACK AckMu sync.Mutex // protects PendingACKs, ACKTimer and QuickACKs PendingACKs int // count of unacked received segments diff --git a/pkg/daemon/zz_fin_in_order_test.go b/pkg/daemon/zz_fin_in_order_test.go new file mode 100644 index 00000000..fb8f7b7f --- /dev/null +++ b/pkg/daemon/zz_fin_in_order_test.go @@ -0,0 +1,279 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +package daemon + +import ( + "bytes" + "net" + "sync" + "testing" + "time" + + "github.com/pilot-protocol/common/protocol" +) + +// finTestConn is an established connection on d to peerNode, ports 55555 +// (local) and 443 (remote), expecting the peer's data at seq 1000. +func finTestConn(t *testing.T, d *Daemon, peerNode uint32) *Connection { + t.Helper() + conn := d.ports.NewConnection(55555, protocol.Addr{Network: 0, Node: peerNode}, 443) + conn.LocalAddr = protocol.Addr{Network: 0, Node: d.NodeID()} + conn.Mu.Lock() + conn.State = StateEstablished + conn.SendSeq = 42 + conn.Mu.Unlock() + conn.RecvMu.Lock() + conn.ExpectedSeq = 1000 + conn.RecvMu.Unlock() + return conn +} + +func finTestData(peerNode, self uint32, seq uint32, payload string) *protocol.Packet { + pkt := streamPacket(protocol.FlagACK, peerNode, self, 443, 55555, seq, 42) + pkt.Payload = []byte(payload) + return pkt +} + +// A FIN that overtook a lost segment used to close the stream at once, so +// the reader saw EOF with the segment missing, and the FIN-ACK made the +// sender drop it. Under loss, a writer that closed right after its last +// write lost the end of it: in the lab, 4 in 30 pub/sub events at 5% loss +// never reached the broker while the publisher reported success. The FIN +// now waits for the data before it. +func TestEarlyFINIsHeldUntilTheDataBeforeItArrives(t *testing.T) { + t.Parallel() + d, peerNode, peerConn := setupDaemonWithPeer(t, Config{Public: true}) + d.setNodeID_testhelper(0xABCD00F1) + conn := finTestConn(t, d, peerNode) + + d.handleStreamPacket(finTestData(peerNode, d.NodeID(), 1000, "abcde")) + // [1005,1010) is lost; the FIN at 1010 arrives first. + d.handleStreamPacket(streamPacket(protocol.FlagFIN, peerNode, d.NodeID(), 443, 55555, 1010, 0)) + + conn.Mu.Lock() + st := conn.State + conn.Mu.Unlock() + if st != StateEstablished { + t.Fatalf("state = %v after a FIN ahead of missing data, want it still established", st) + } + conn.RecvMu.Lock() + closed := conn.RecvClosed + conn.RecvMu.Unlock() + if closed { + t.Fatal("the receive side was closed with data still missing") + } + for { + pkt := readPacket(t, peerConn, 300*time.Millisecond) + if pkt == nil { + t.Fatal("no ACK asking for the missing data") + } + if pkt.HasFlag(protocol.FlagFIN) { + t.Fatal("a FIN-ACK went out with data still missing: the sender would drop it") + } + if pkt.Ack == 1005 { + break // the ACK that asks for [1005,1010) + } + } + + d.handleStreamPacket(finTestData(peerNode, d.NodeID(), 1005, "fghij")) + + var got []byte + for b := range conn.RecvBuf { + got = append(got, b...) + } + if string(got) != "abcdefghij" { + t.Fatalf("reader got %q, want every byte before EOF", got) + } + var finack *protocol.Packet + for finack == nil { + pkt := readPacket(t, peerConn, time.Second) + if pkt == nil { + t.Fatal("no FIN-ACK once the data had arrived") + } + if pkt.HasFlag(protocol.FlagFIN) { + finack = pkt + } + } + if !finack.HasFlag(protocol.FlagACK) || finack.Ack != 1011 { + t.Fatalf("FIN-ACK flags=%d ack=%d, want FIN|ACK acknowledging 1011", finack.Flags, finack.Ack) + } + conn.Mu.Lock() + st = conn.State + conn.Mu.Unlock() + if st != StateTimeWait { + t.Fatalf("state = %v after the FIN was handled, want TIME_WAIT", st) + } +} + +// A connection this side has already closed discards the peer's data, so +// the peer's FIN is handled at once even if data before it is missing. +func TestFINOnAClosingConnectionIsAcceptedAtOnce(t *testing.T) { + t.Parallel() + d, peerNode, peerConn := setupDaemonWithPeer(t, Config{Public: true}) + d.setNodeID_testhelper(0xABCD00F2) + conn := finTestConn(t, d, peerNode) + conn.Mu.Lock() + conn.State = StateFinWait + conn.Mu.Unlock() + + d.handleStreamPacket(streamPacket(protocol.FlagFIN|protocol.FlagACK, peerNode, d.NodeID(), 443, 55555, 1010, 43)) + + pkt := readPacket(t, peerConn, 500*time.Millisecond) + if pkt == nil || !pkt.HasFlag(protocol.FlagFIN) { + t.Fatalf("got %v, want a FIN-ACK at once", pkt) + } + conn.Mu.Lock() + st := conn.State + conn.Mu.Unlock() + if st != StateTimeWait { + t.Fatalf("state = %v, want TIME_WAIT", st) + } +} + +// Closing with data unacknowledged sends the FIN once the data is +// acknowledged, and after finLinger at the latest. +func TestCloseSendsTheFINAfterTheLingerIfNothingIsAcknowledged(t *testing.T) { + t.Parallel() + d, pc, conn := newSendDataFixture(t, 0xD9D90003) + d.finLinger = 300 * time.Millisecond + conn.NoDelay = true + if err := d.SendData(conn, []byte("unacknowledged")); err != nil { + t.Fatalf("SendData: %v", err) + } + readOneSegment(t, pc, 2*time.Second) // the data + start := time.Now() + d.CloseConnection(conn) + fin := readOneSegment(t, pc, 2*time.Second) + if fin.Flags&protocol.FlagFIN == 0 { + t.Fatalf("next packet flags %#x, want the FIN", fin.Flags) + } + if waited := time.Since(start); waited < 250*time.Millisecond { + t.Fatalf("FIN sent after %s with the data unacknowledged, want it held for the linger", waited) + } +} + +// End to end over two daemons: a writer that closes right after its last +// write, with one segment's first transmission lost, still gets every byte +// to the reader before EOF. finLinger 0 sends the FIN at once, as senders up +// to v1.17.0 do, so only the receiver's rule protects the data. +func TestWriteThenCloseSurvivesALostSegment(t *testing.T) { + if testing.Short() { + t.Skip("multi-second transfer") + } + for _, tc := range []struct { + name string + linger time.Duration + }{ + {"FIN at once (old senders)", 0}, + {"FIN after the data is acknowledged", defaultFinLinger}, + } { + t.Run(tc.name, func(t *testing.T) { + got, want := runWriteThenClose(t, tc.linger) + if !bytes.Equal(got, want) { + t.Fatalf("reader got %d of %d bytes before EOF", len(got), len(want)) + } + }) + } +} + +func runWriteThenClose(t *testing.T, linger time.Duration) (got, want []byte) { + const idA, idB uint32 = 0xA0000F01, 0xB0000F01 + const portA, portB uint16 = 5100, 6100 + const isnA uint32 = 200000 + + A := New(Config{Public: true}) + B := New(Config{Public: true}) + A.finLinger = linger + for _, d := range []*Daemon{A, B} { + if err := d.tunnels.Listen("127.0.0.1:0"); err != nil { + t.Fatal(err) + } + dd := d + t.Cleanup(func() { dd.tunnels.Close() }) + } + A.setNodeID_testhelper(idA) + B.setNodeID_testhelper(idB) + pa := parkedSock(t) // A → B traffic lands here + pb := parkedSock(t) // B → A traffic lands here + A.AddTunnelPeer(idB, pa.LocalAddr().(*net.UDPAddr)) + B.AddTunnelPeer(idA, pb.LocalAddr().(*net.UDPAddr)) + + cA := A.ports.NewConnection(portA, protocol.Addr{Node: idB}, portB) + cA.LocalAddr = protocol.Addr{Node: idA} + cA.State = StateEstablished + cA.SendSeq = isnA + cA.RecvAck = 1 + cA.NoDelay = true + + cB := B.ports.NewConnection(portB, protocol.Addr{Node: idA}, portA) + cB.LocalAddr = protocol.Addr{Node: idB} + cB.State = StateEstablished + cB.SendSeq = 1 + cB.RecvAck = isnA + cB.ExpectedSeq = isnA + + A.startRetxLoop(cA) + t.Cleanup(func() { A.ports.RemoveConnection(cA.ID) }) + + lostSeq := isnA + SendSegmentSize // the second segment's first send + stop := make(chan struct{}) + var wg sync.WaitGroup + wg.Add(2) + go func() { // A → B, with one segment lost once + defer wg.Done() + dropped := false + for { + select { + case <-stop: + return + default: + } + pkt := parkedRead(pa) + if pkt == nil { + continue + } + if len(pkt.Payload) > 0 && pkt.Seq == lostSeq && !dropped { + dropped = true + continue + } + B.handleStreamPacket(pkt) + } + }() + go func() { // B → A + defer wg.Done() + for { + select { + case <-stop: + return + default: + } + if pkt := parkedRead(pb); pkt != nil { + A.handleStreamPacket(pkt) + } + } + }() + t.Cleanup(func() { close(stop); wg.Wait() }) + + want = make([]byte, 3*SendSegmentSize+100) + for i := range want { + want[i] = byte(i*13 + i/SendSegmentSize) + } + if err := A.SendData(cA, want); err != nil { + t.Fatalf("SendData: %v", err) + } + A.CloseConnection(cA) + + done := make(chan struct{}) + go func() { + defer close(done) + for b := range cB.RecvBuf { + got = append(got, b...) + } + }() + select { + case <-done: + case <-time.After(15 * time.Second): + t.Fatalf("reader still open after 15s with %d of %d bytes", len(got), len(want)) + } + return got, want +} diff --git a/pkg/daemon/zz_large_write_pieces_test.go b/pkg/daemon/zz_large_write_pieces_test.go index 42dbd92e..cd2fefd7 100644 --- a/pkg/daemon/zz_large_write_pieces_test.go +++ b/pkg/daemon/zz_large_write_pieces_test.go @@ -247,11 +247,13 @@ func TestCloseCommitsTheWholeTailDespiteSendErrors(t *testing.T) { if got := trackedStream(t, conn); !bytes.Equal(got, tail) { t.Fatalf("%d bytes of a %d-byte tail given sequence numbers before the FIN", len(got), len(tail)) } - conn.RetxMu.Lock() - defer conn.RetxMu.Unlock() - fin := conn.Unacked[len(conn.Unacked)-1] - if want := uint32(1000 + len(tail)); !fin.isFIN || fin.seq != want { - t.Fatalf("last tracked entry isFIN=%v seq=%d, want the FIN at %d", fin.isFIN, fin.seq, want) + // The FIN's sequence number comes directly after the tail. It is sent + // once the data is acknowledged (sendFINAfterData), never here. + conn.Mu.Lock() + next := conn.SendSeq + conn.Mu.Unlock() + if want := uint32(1000 + len(tail) + 1); next != want { + t.Fatalf("SendSeq after close = %d, want %d: the FIN reserved directly after the tail", next, want) } } diff --git a/pkg/daemon/zz_segment_fits_packet_test.go b/pkg/daemon/zz_segment_fits_packet_test.go index 0fde1ab1..ffde3e95 100644 --- a/pkg/daemon/zz_segment_fits_packet_test.go +++ b/pkg/daemon/zz_segment_fits_packet_test.go @@ -280,10 +280,13 @@ func TestCloseSendsHeldTailBeforeFIN(t *testing.T) { } d.CloseConnection(conn) - pkts := readSegments(t, pc, 2) + pkts := readSegments(t, pc, 1) if string(pkts[0].Payload) != "last words" { t.Fatalf("first packet after close carries %q, want the held write", pkts[0].Payload) } + // The FIN waits for the data to be acknowledged (sendFINAfterData). + conn.ProcessAck(pkts[0].Seq+uint32(len(pkts[0].Payload)), false) + pkts = append(pkts, readSegments(t, pc, 1)...) if pkts[1].Flags&protocol.FlagFIN == 0 { t.Fatalf("second packet after close has flags %#x, want FIN", pkts[1].Flags) } diff --git a/pkg/daemon/zz_streampacket_test.go b/pkg/daemon/zz_streampacket_test.go index c791caf7..f644915b 100644 --- a/pkg/daemon/zz_streampacket_test.go +++ b/pkg/daemon/zz_streampacket_test.go @@ -251,6 +251,9 @@ func TestHandleStreamFINEstablishedClosesAndSendsFINACK(t *testing.T) { conn.State = StateEstablished conn.SendSeq = 42 conn.Mu.Unlock() + conn.RecvMu.Lock() + conn.ExpectedSeq = 9000 // everything before the FIN delivered + conn.RecvMu.Unlock() fin := streamPacket(protocol.FlagFIN, peerNode, d.NodeID(), 443, 55555, 9000, 0) d.handleStreamPacket(fin) From 1383c6a4287b9c9812db895ca32525ea3c364563 Mon Sep 17 00:00:00 2001 From: Teo Calin Date: Thu, 8 Oct 2026 12:21:29 +0300 Subject: [PATCH 2/4] daemon: shutdown sends a FIN that was waiting for its data A connection closed with data unacknowledged holds its FIN until the data is acknowledged. Stopping the daemon in that window dropped the FIN: the waiting goroutine returned on stop, and shutdown sends FINs only to established connections. The peer then kept the connection open until keepalive gave up on it (TestConnectionAfterPeerRestart: 5 of 5 failed at its 60 s budget). The held FIN now lives on the connection, and whichever of the two takes it sends it, once. Co-Authored-By: Claude Opus 5.5 --- pkg/daemon/daemon.go | 27 +++++++++++++++++------- pkg/daemon/ports.go | 4 ++++ pkg/daemon/zz_fin_in_order_test.go | 34 ++++++++++++++++++++++++++++++ 3 files changed, 57 insertions(+), 8 deletions(-) diff --git a/pkg/daemon/daemon.go b/pkg/daemon/daemon.go index ecd38d0c..bdc03a53 100644 --- a/pkg/daemon/daemon.go +++ b/pkg/daemon/daemon.go @@ -1731,7 +1731,13 @@ func (d *Daemon) doStop() { conn.Mu.Lock() st := conn.State seq := conn.SendSeq + held := conn.heldFIN + conn.heldFIN = nil conn.Mu.Unlock() + if held != nil && st == StateFinWait { + // Closed with data still unacknowledged: its FIN was waiting. + d.tunnels.Send(conn.RemoteAddr.Node, held) + } if st == StateEstablished { // Send FIN fin := &protocol.Packet{ @@ -5160,8 +5166,11 @@ func (d *Daemon) CloseConnection(conn *Connection) { // data is acknowledged (finLinger at most), so the end of a write // that is followed at once by a close survives loss on such // receivers too. The caller does not wait. - if d.unackedHasData(conn) { - go d.sendFINAfterData(conn, fin) + if d.unackedHasData(conn) && !d.stopping() { + conn.Mu.Lock() + conn.heldFIN = fin + conn.Mu.Unlock() + go d.sendFINAfterData(conn) } else { d.sendFIN(conn, fin) } @@ -5208,11 +5217,11 @@ func (d *Daemon) sendFIN(conn *Connection, fin *protocol.Packet) { conn.RetxMu.Unlock() } -// sendFINAfterData sends the FIN once every data segment is acknowledged, -// or after finLinger. It gives up if the connection goes away, the daemon -// stops, or the peer's own FIN closed the connection meanwhile (its -// FIN-ACK then stands for ours). -func (d *Daemon) sendFINAfterData(conn *Connection, fin *protocol.Packet) { +// sendFINAfterData sends the held FIN once every data segment is +// acknowledged, or after finLinger. It gives up if the connection goes +// away or the peer's own FIN closed it meanwhile (its FIN-ACK then stands +// for ours). When the daemon stops, shutdown sends the FIN instead. +func (d *Daemon) sendFINAfterData(conn *Connection) { deadline := time.Now().Add(d.finLinger) tick := time.NewTicker(20 * time.Millisecond) defer tick.Stop() @@ -5226,9 +5235,11 @@ func (d *Daemon) sendFINAfterData(conn *Connection, fin *protocol.Packet) { } } conn.Mu.Lock() + fin := conn.heldFIN + conn.heldFIN = nil st := conn.State conn.Mu.Unlock() - if st != StateFinWait { + if fin == nil || st != StateFinWait { return } d.sendFIN(conn, fin) diff --git a/pkg/daemon/ports.go b/pkg/daemon/ports.go index 667815fe..259b827f 100644 --- a/pkg/daemon/ports.go +++ b/pkg/daemon/ports.go @@ -287,6 +287,10 @@ type Connection struct { RemotePort uint16 State ConnState LastActivity time.Time // updated on send/recv + // heldFIN is this side's FIN while it waits for the data before it to + // be acknowledged (Daemon.sendFINAfterData). Whoever takes it sends it: + // that goroutine, or shutdown. Guarded by Mu. + heldFIN *protocol.Packet // Reliable delivery SendSeq uint32 RecvAck uint32 diff --git a/pkg/daemon/zz_fin_in_order_test.go b/pkg/daemon/zz_fin_in_order_test.go index fb8f7b7f..53547469 100644 --- a/pkg/daemon/zz_fin_in_order_test.go +++ b/pkg/daemon/zz_fin_in_order_test.go @@ -152,6 +152,40 @@ func TestCloseSendsTheFINAfterTheLingerIfNothingIsAcknowledged(t *testing.T) { } } +// A FIN still waiting for its data when the daemon stops goes out at +// shutdown, as it did when it was sent at close: otherwise the peer is left +// with an open connection until keepalive gives up on it. +func TestStopSendsAFINThatWasWaitingForData(t *testing.T) { + t.Parallel() + d, pc, conn := newSendDataFixture(t, 0xD9D90004) + conn.NoDelay = true + if err := d.SendData(conn, []byte("unacknowledged")); err != nil { + t.Fatalf("SendData: %v", err) + } + readOneSegment(t, pc, 2*time.Second) // the data + d.CloseConnection(conn) + conn.Mu.Lock() + held := conn.heldFIN != nil + conn.Mu.Unlock() + if !held { + t.Fatal("the FIN went out with the data unacknowledged; nothing to test") + } + stopped := make(chan struct{}) + go func() { d.Stop(); close(stopped) }() + fin := readOneSegment(t, pc, 2*time.Second) + if fin.Flags&protocol.FlagFIN == 0 { + t.Fatalf("next packet flags %#x, want the FIN", fin.Flags) + } + if fin.Seq != 1000+uint32(len("unacknowledged")) { + t.Fatalf("FIN seq %d, want %d", fin.Seq, 1000+len("unacknowledged")) + } + <-stopped + pc.SetReadDeadline(time.Now().Add(200 * time.Millisecond)) + if n, _, err := pc.ReadFromUDP(make([]byte, 65535)); err == nil { + t.Fatalf("a second packet (%d bytes) after the FIN: it must go out once", n) + } +} + // End to end over two daemons: a writer that closes right after its last // write, with one segment's first transmission lost, still gets every byte // to the reader before EOF. finLinger 0 sends the FIN at once, as senders up From 43e401d84aff5139ca1b2ec89ea284d346864a0a Mon Sep 17 00:00:00 2001 From: Teo Calin Date: Thu, 8 Oct 2026 13:19:23 +0300 Subject: [PATCH 3/4] daemon: held FINs go out on the ACK, never get lost, and expire when the peer goes silent Review of the FIN hold found: - The held FIN could be dropped: the goroutine that sends it started before CloseConnection set FIN_WAIT, and a goroutine that found the data already acknowledged read ESTABLISHED and gave up. 3 of 300 runs of the write-then-close test lost the FIN. A held FIN is now dropped only once the connection is in TIME_WAIT or CLOSED. - A FIN held for data the sender never resends (a daemon stopped or restarted with a segment lost, any version) stayed held until keepalive gave up, about two minutes. It is now handled once nothing has come from the peer for 15 s; a sender still there resends at most 10 s apart (RTOMax). - Each closing connection polled every 20 ms for up to 5 s: 2,000 closes at once cost half a core. There is no goroutine now: the ACK path sends the held FIN when the last data is acknowledged, a timer sends it at the linger, and the reaper and shutdown send it if still held. - Shutdown could send a second FIN, at SendSeq, after a held one, and a receiver would hold that one for good. It now sends the held FIN only. The peer FIN's first sequence number is kept. Tests: the early-FIN test reads only what the FIN caused, the end-to-end test's reader no longer races its failure message, and the tail test checks the held FIN's sequence number again. New tests for the ACK-path release, the silent-peer expiry (and data restarting its wait) and a half-done close at shutdown each fail with their fix reverted. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 7 +- pkg/daemon/daemon.go | 134 ++++++++++++++++------ pkg/daemon/ports.go | 19 ++- pkg/daemon/zz_fin_in_order_test.go | 130 +++++++++++++++++++-- pkg/daemon/zz_large_write_pieces_test.go | 14 ++- pkg/daemon/zz_segment_fits_packet_test.go | 4 +- 6 files changed, 250 insertions(+), 58 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 10361d80..8468314a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -402,8 +402,11 @@ Detailed per-release notes are on the broker while `pilotctl publish` reported success; one-shot sends and any client that closes right after writing were exposed the same way. A receiver now handles the FIN only once the data before it has arrived, - and asks for what is missing meanwhile. A closing sender also holds its - FIN until its data is acknowledged (5 s at most, without blocking the + and asks for what is missing meanwhile. If the sender stops resending it + (a daemon stopped or restarted with a segment lost), the receiver closes + the stream without it 15 s after it last heard from the sender, rather + than waiting about two minutes for keepalive. A closing sender also holds + its FIN until its data is acknowledged (5 s at most, without blocking the caller), so the end of a write survives loss on receivers that have not upgraded too. - **A send error no longer strands the end of a stream write.** When the diff --git a/pkg/daemon/daemon.go b/pkg/daemon/daemon.go index bdc03a53..0b5e6f67 100644 --- a/pkg/daemon/daemon.go +++ b/pkg/daemon/daemon.go @@ -475,14 +475,17 @@ type Daemon struct { // (see dialConnectionLocked): dialOKSeq counts completed dials, and a // dial compares it with the count it saw when it started. A counter, // not lastDialOKNano, so a wall-clock step cannot hide timeouts. - // finLinger bounds the wait for a closing connection's data to be - // acknowledged before its FIN goes out (sendFINAfterData). - finLinger time.Duration - lastDialOKNano atomic.Int64 dialOKSeq atomic.Uint64 consecutiveDialTimeouts atomic.Uint64 + // finLinger bounds the wait for a closing connection's data to be + // acknowledged before its FIN goes out anyway (releaseHeldFIN). + // finHoldIdle bounds how long a FIN from the peer is held for missing + // data once nothing more comes from the peer (expireHeldFIN). + finLinger time.Duration + finHoldIdle time.Duration + startTime time.Time stopCh chan struct{} // closed on Stop() to signal goroutines beaconSelection *beaconSelectionState // multi-beacon discovery state @@ -669,6 +672,7 @@ func New(cfg Config) *Daemon { ports: NewPortManager(), stopCh: make(chan struct{}), finLinger: defaultFinLinger, + finHoldIdle: defaultFinHoldIdle, hsPoll: newHandshakePollSched(), synTokens: cfg.synRateLimit(), synLastFill: time.Now(), @@ -1731,14 +1735,19 @@ func (d *Daemon) doStop() { conn.Mu.Lock() st := conn.State seq := conn.SendSeq - held := conn.heldFIN - conn.heldFIN = nil + held, lingerTimer := conn.heldFIN, conn.finLingerTimer + conn.heldFIN, conn.finLingerTimer = nil, nil conn.Mu.Unlock() - if held != nil && st == StateFinWait { + if lingerTimer != nil { + lingerTimer.Stop() + } + if held != nil && st != StateTimeWait && st != StateClosed { // Closed with data still unacknowledged: its FIN was waiting. + // It may still read ESTABLISHED, its close half done; the + // FIN's sequence number is already reserved, so send that one + // and no second FIN after it. d.tunnels.Send(conn.RemoteAddr.Node, held) - } - if st == StateEstablished { + } else if st == StateEstablished { // Send FIN fin := &protocol.Packet{ Version: protocol.Version, @@ -3621,11 +3630,22 @@ func (d *Daemon) handleStreamPacket(pkt *protocol.Packet) { conn.Mu.Lock() receiving := conn.State == StateEstablished conn.Mu.Unlock() + // A sender that stops without resending the missing data (a daemon + // stopped or restarted with a segment lost, of any version) would + // leave the FIN held for good: once nothing has come from the peer + // for finHoldIdle, the FIN is handled anyway (expireHeldFIN). conn.RecvMu.Lock() early := receiving && !conn.RecvClosed && seqAfter(pkt.Seq, conn.ExpectedSeq) if early { - conn.finPending = true - conn.finSeq = pkt.Seq + if !conn.finPending { + // The first copy's sequence number; a later one, or a + // retransmission, does not move it. + conn.finPending = true + conn.finSeq = pkt.Seq + c := conn + conn.finHoldTimer = time.AfterFunc(d.finHoldIdle, func() { d.expireHeldFIN(c) }) + } + conn.finHeard = time.Now() } conn.RecvMu.Unlock() if early { @@ -3702,6 +3722,8 @@ func (d *Daemon) handleStreamPacket(pkt *protocol.Packet) { } conn.ProcessAck(pkt.Ack, isPureACK) } + // A FIN held for the data this ACK acknowledged goes out now. + d.releaseHeldFINIfAcked(conn) // Check if payload is SACK info (not user data) if isSACK { @@ -3731,6 +3753,12 @@ func (d *Daemon) handleStreamPacket(pkt *protocol.Packet) { finSeq := conn.finSeq if finReady { conn.finPending = false + if conn.finHoldTimer != nil { + conn.finHoldTimer.Stop() + conn.finHoldTimer = nil + } + } else if conn.finPending { + conn.finHeard = time.Now() } conn.RecvMu.Unlock() if finReady { @@ -3778,6 +3806,38 @@ func (d *Daemon) handleStreamPacket(pkt *protocol.Packet) { } } +// defaultFinHoldIdle is how long a FIN held for missing data waits after the +// last packet from the peer (Daemon.finHoldIdle). A sender that is still +// there resends the missing data with its retransmission timeout, at most +// 10 s (RTOMax) apart, so it is always heard from sooner. +const defaultFinHoldIdle = 15 * time.Second + +// expireHeldFIN handles a FIN held for missing data once nothing has come +// from the peer for finHoldIdle: the sender has stopped resending (it was +// stopped or restarted, or gave up). The reader gets EOF without the +// missing data, as every version up to v1.17.0 gave it at once. Runs on the +// hold timer; a FIN that delivery reached first is not handled again. +func (d *Daemon) expireHeldFIN(conn *Connection) { + conn.RecvMu.Lock() + if !conn.finPending { + conn.RecvMu.Unlock() + return + } + if wait := d.finHoldIdle - time.Since(conn.finHeard); wait > 0 { + conn.finHoldTimer.Reset(wait) + conn.RecvMu.Unlock() + return + } + conn.finPending = false + conn.finHoldTimer = nil + finSeq, missing := conn.finSeq, conn.finSeq-conn.ExpectedSeq + conn.RecvMu.Unlock() + slog.Warn("peer closed with data missing and stopped resending it — closing the stream without it", + "conn_id", conn.ID, "remote_addr", conn.RemoteAddr, "remote_port", conn.RemotePort, + "missing_bytes", missing, "silent_for", d.finHoldIdle) + d.acceptFIN(conn, finSeq) +} + // acceptFIN handles the peer's FIN once every byte sent before it has been // delivered (see the FIN branch of handleStreamPacket): it closes the // receive side, moves to TIME_WAIT and answers with a FIN-ACK. @@ -5167,10 +5227,13 @@ func (d *Daemon) CloseConnection(conn *Connection) { // that is followed at once by a close survives loss on such // receivers too. The caller does not wait. if d.unackedHasData(conn) && !d.stopping() { + c := conn conn.Mu.Lock() conn.heldFIN = fin + conn.finLingerTimer = time.AfterFunc(d.finLinger, func() { d.releaseHeldFIN(c) }) conn.Mu.Unlock() - go d.sendFINAfterData(conn) + // The last ACK may have been handled since the check above. + d.releaseHeldFINIfAcked(conn) } else { d.sendFIN(conn, fin) } @@ -5217,34 +5280,38 @@ func (d *Daemon) sendFIN(conn *Connection, fin *protocol.Packet) { conn.RetxMu.Unlock() } -// sendFINAfterData sends the held FIN once every data segment is -// acknowledged, or after finLinger. It gives up if the connection goes -// away or the peer's own FIN closed it meanwhile (its FIN-ACK then stands -// for ours). When the daemon stops, shutdown sends the FIN instead. -func (d *Daemon) sendFINAfterData(conn *Connection) { - deadline := time.Now().Add(d.finLinger) - tick := time.NewTicker(20 * time.Millisecond) - defer tick.Stop() - for d.unackedHasData(conn) && time.Now().Before(deadline) { - select { - case <-conn.RetxStop: - return - case <-d.stopCh: - return - case <-tick.C: - } - } +// releaseHeldFIN sends this side's held FIN, if it still holds one: when +// the ACK path sees the data before it acknowledged, when finLinger runs +// out, when the reaper removes the connection, or at shutdown. Whichever +// comes first sends it; the others find nothing held. A connection already +// in TIME_WAIT or CLOSED sends nothing: the peer's FIN closed it, and the +// FIN-ACK this side sent then stands for its own FIN. +func (d *Daemon) releaseHeldFIN(conn *Connection) { conn.Mu.Lock() - fin := conn.heldFIN - conn.heldFIN = nil + fin, timer := conn.heldFIN, conn.finLingerTimer + conn.heldFIN, conn.finLingerTimer = nil, nil st := conn.State conn.Mu.Unlock() - if fin == nil || st != StateFinWait { + if timer != nil { + timer.Stop() + } + if fin == nil || st == StateTimeWait || st == StateClosed { return } d.sendFIN(conn, fin) } +// releaseHeldFINIfAcked sends the held FIN once no data is unacknowledged. +// Called from the ACK path. +func (d *Daemon) releaseHeldFINIfAcked(conn *Connection) { + conn.Mu.Lock() + held := conn.heldFIN != nil + conn.Mu.Unlock() + if held && !d.unackedHasData(conn) { + d.releaseHeldFIN(conn) + } +} + // SendDatagram sends an unreliable unicast packet. Broadcast addresses // (Node=0xFFFFFFFF) are rejected — use BroadcastDatagram, which requires // an admin token. @@ -6410,6 +6477,9 @@ func (d *Daemon) idleSweepLoop() { // Clean up stale non-established connections (CLOSED, FIN_WAIT, etc.) stale := d.ports.StaleConnections(timeWaitDur) for _, conn := range stale { + // With a time-wait shorter than finLinger, a closing + // connection can be reaped while it holds its FIN. + d.releaseHeldFIN(conn) conn.CloseRecvBuf() d.ports.RemoveConnection(conn.ID) } diff --git a/pkg/daemon/ports.go b/pkg/daemon/ports.go index 259b827f..ea4c91ae 100644 --- a/pkg/daemon/ports.go +++ b/pkg/daemon/ports.go @@ -288,9 +288,12 @@ type Connection struct { State ConnState LastActivity time.Time // updated on send/recv // heldFIN is this side's FIN while it waits for the data before it to - // be acknowledged (Daemon.sendFINAfterData). Whoever takes it sends it: - // that goroutine, or shutdown. Guarded by Mu. - heldFIN *protocol.Packet + // be acknowledged; finLingerTimer sends it after Daemon.finLinger + // regardless. Whoever takes it sends it (Daemon.releaseHeldFIN): the ACK + // that acknowledges the last data, the timer, the reaper or shutdown. + // Guarded by Mu. + heldFIN *protocol.Packet + finLingerTimer *time.Timer // Reliable delivery SendSeq uint32 RecvAck uint32 @@ -342,9 +345,13 @@ type Connection struct { OOOBuf []*recvSegment // out-of-order buffer // finPending is set when the peer's FIN arrived ahead of data it sent // before it; finSeq is the FIN's sequence number. The FIN is handled - // once delivery reaches finSeq (see Daemon.acceptFIN). Guarded by RecvMu. - finPending bool - finSeq uint32 + // once delivery reaches finSeq (see Daemon.acceptFIN), or by + // finHoldTimer once nothing has come from the peer since finHeard for + // Daemon.finHoldIdle (Daemon.expireHeldFIN). Guarded by RecvMu. + finPending bool + finSeq uint32 + finHeard time.Time + finHoldTimer *time.Timer // Delayed ACK AckMu sync.Mutex // protects PendingACKs, ACKTimer and QuickACKs PendingACKs int // count of unacked received segments diff --git a/pkg/daemon/zz_fin_in_order_test.go b/pkg/daemon/zz_fin_in_order_test.go index 53547469..4fa208fb 100644 --- a/pkg/daemon/zz_fin_in_order_test.go +++ b/pkg/daemon/zz_fin_in_order_test.go @@ -47,6 +47,10 @@ func TestEarlyFINIsHeldUntilTheDataBeforeItArrives(t *testing.T) { conn := finTestConn(t, d, peerNode) d.handleStreamPacket(finTestData(peerNode, d.NodeID(), 1000, "abcde")) + // Drain what the first segment caused, so the packets checked below are + // the FIN's. + for readPacket(t, peerConn, 150*time.Millisecond) != nil { + } // [1005,1010) is lost; the FIN at 1010 arrives first. d.handleStreamPacket(streamPacket(protocol.FlagFIN, peerNode, d.NodeID(), 443, 55555, 1010, 0)) @@ -62,17 +66,12 @@ func TestEarlyFINIsHeldUntilTheDataBeforeItArrives(t *testing.T) { if closed { t.Fatal("the receive side was closed with data still missing") } - for { - pkt := readPacket(t, peerConn, 300*time.Millisecond) - if pkt == nil { - t.Fatal("no ACK asking for the missing data") - } - if pkt.HasFlag(protocol.FlagFIN) { - t.Fatal("a FIN-ACK went out with data still missing: the sender would drop it") - } - if pkt.Ack == 1005 { - break // the ACK that asks for [1005,1010) - } + pkt := readPacket(t, peerConn, 300*time.Millisecond) + if pkt == nil { + t.Fatal("no ACK in reply to the early FIN") + } + if pkt.HasFlag(protocol.FlagFIN) || pkt.Ack != 1005 { + t.Fatalf("reply to the early FIN: flags=%d ack=%d, want a plain ACK asking for 1005 (a FIN-ACK makes the sender drop the missing data)", pkt.Flags, pkt.Ack) } d.handleStreamPacket(finTestData(peerNode, d.NodeID(), 1005, "fghij")) @@ -186,6 +185,107 @@ func TestStopSendsAFINThatWasWaitingForData(t *testing.T) { } } +// The held FIN goes out from the ACK path as soon as the data before it is +// acknowledged, not on a poll or at the linger. +func TestHeldFINGoesOutWithTheACKOfTheLastData(t *testing.T) { + t.Parallel() + const peer = 0xD9D90005 + d, pc, conn := newSendDataFixture(t, peer) + d.setNodeID_testhelper(0x22222222) + conn.NoDelay = true + if err := d.SendData(conn, []byte("payload")); err != nil { + t.Fatalf("SendData: %v", err) + } + data := readOneSegment(t, pc, 2*time.Second) + d.CloseConnection(conn) + conn.Mu.Lock() + held := conn.heldFIN != nil + conn.Mu.Unlock() + if !held { + t.Fatal("the FIN went out with the data unacknowledged; nothing to test") + } + start := time.Now() + ack := streamPacket(protocol.FlagACK, peer, d.NodeID(), 80, 2000, 1, data.Seq+uint32(len(data.Payload))) + ack.Window = 64 + d.handleStreamPacket(ack) + fin := readOneSegment(t, pc, 2*time.Second) + if fin.Flags&protocol.FlagFIN == 0 || fin.Seq != data.Seq+uint32(len(data.Payload)) { + t.Fatalf("after the ACK: flags %#x seq %d, want the FIN at %d", fin.Flags, fin.Seq, data.Seq+uint32(len(data.Payload))) + } + if waited := time.Since(start); waited > 500*time.Millisecond { + t.Fatalf("FIN sent %s after the ACK, want it at once", waited) + } +} + +// A FIN held for data the peer never resends (it was stopped or restarted +// with a segment lost) is handled once nothing has come from the peer for +// finHoldIdle: the reader gets EOF instead of waiting for keepalive to give +// up two minutes later. Data still arriving holds it longer. +func TestHeldPeerFINIsHandledOnceThePeerFallsSilent(t *testing.T) { + t.Parallel() + d, peerNode, peerConn := setupDaemonWithPeer(t, Config{Public: true}) + d.setNodeID_testhelper(0xABCD00F3) + d.finHoldIdle = 400 * time.Millisecond + conn := finTestConn(t, d, peerNode) + + d.handleStreamPacket(finTestData(peerNode, d.NodeID(), 1000, "abcde")) + start := time.Now() + d.handleStreamPacket(streamPacket(protocol.FlagFIN, peerNode, d.NodeID(), 443, 55555, 1020, 0)) + // The peer is still there for a while: an out-of-order segment after + // 250 ms restarts the wait. + time.Sleep(250 * time.Millisecond) + d.handleStreamPacket(finTestData(peerNode, d.NodeID(), 1010, "klmno")) + + var finack *protocol.Packet + for finack == nil { + pkt := readPacket(t, peerConn, 2*time.Second) + if pkt == nil { + t.Fatal("the held FIN was never handled after the peer fell silent") + } + if pkt.HasFlag(protocol.FlagFIN) { + finack = pkt + } + } + if waited := time.Since(start); waited < 600*time.Millisecond { + t.Fatalf("FIN handled %s after it arrived, want the wait restarted by the data at 250 ms (>= 650 ms)", waited) + } + if finack.Ack != 1021 { + t.Fatalf("FIN-ACK ack=%d, want 1021", finack.Ack) + } + var got []byte + for b := range conn.RecvBuf { + got = append(got, b...) + } + if string(got) != "abcde" { + t.Fatalf("reader got %q before EOF, want the in-order data only", got) + } +} + +// Shutdown can catch a close half done: the FIN held, the state still +// ESTABLISHED. It sends that FIN, at its reserved sequence number, and no +// second one after it, which a receiver would hold for good. +func TestStopSendsOnlyTheHeldFINOfAHalfDoneClose(t *testing.T) { + t.Parallel() + d, pc, conn := newSendDataFixture(t, 0xD9D90006) + conn.Mu.Lock() + conn.heldFIN = &protocol.Packet{ + Version: protocol.Version, Flags: protocol.FlagFIN, Protocol: protocol.ProtoStream, + Src: conn.LocalAddr, Dst: conn.RemoteAddr, SrcPort: conn.LocalPort, DstPort: conn.RemotePort, + Seq: 1000, + } + conn.SendSeq = 1001 // reserved, as CloseConnection does + conn.Mu.Unlock() + d.Stop() + fin := readOneSegment(t, pc, 2*time.Second) + if fin.Flags&protocol.FlagFIN == 0 || fin.Seq != 1000 { + t.Fatalf("at stop: flags %#x seq %d, want the held FIN at 1000", fin.Flags, fin.Seq) + } + pc.SetReadDeadline(time.Now().Add(200 * time.Millisecond)) + if n, _, err := pc.ReadFromUDP(make([]byte, 65535)); err == nil { + t.Fatalf("a second packet (%d bytes) after the held FIN", n) + } +} + // End to end over two daemons: a writer that closes right after its last // write, with one segment's first transmission lost, still gets every byte // to the reader before EOF. finLinger 0 sends the FIN at once, as senders up @@ -297,17 +397,23 @@ func runWriteThenClose(t *testing.T, linger time.Duration) (got, want []byte) { } A.CloseConnection(cA) + var mu sync.Mutex done := make(chan struct{}) go func() { defer close(done) for b := range cB.RecvBuf { + mu.Lock() got = append(got, b...) + mu.Unlock() } }() select { case <-done: case <-time.After(15 * time.Second): - t.Fatalf("reader still open after 15s with %d of %d bytes", len(got), len(want)) + mu.Lock() + n := len(got) + mu.Unlock() + t.Fatalf("reader still open after 15s with %d of %d bytes", n, len(want)) } return got, want } diff --git a/pkg/daemon/zz_large_write_pieces_test.go b/pkg/daemon/zz_large_write_pieces_test.go index cd2fefd7..06d6dacf 100644 --- a/pkg/daemon/zz_large_write_pieces_test.go +++ b/pkg/daemon/zz_large_write_pieces_test.go @@ -247,13 +247,17 @@ func TestCloseCommitsTheWholeTailDespiteSendErrors(t *testing.T) { if got := trackedStream(t, conn); !bytes.Equal(got, tail) { t.Fatalf("%d bytes of a %d-byte tail given sequence numbers before the FIN", len(got), len(tail)) } - // The FIN's sequence number comes directly after the tail. It is sent - // once the data is acknowledged (sendFINAfterData), never here. + // The FIN comes directly after the tail. It is held until the data is + // acknowledged (releaseHeldFIN), never sent here. conn.Mu.Lock() - next := conn.SendSeq + held, next := conn.heldFIN, conn.SendSeq conn.Mu.Unlock() - if want := uint32(1000 + len(tail) + 1); next != want { - t.Fatalf("SendSeq after close = %d, want %d: the FIN reserved directly after the tail", next, want) + want := uint32(1000 + len(tail)) + if held == nil || held.Seq != want || held.Flags&protocol.FlagFIN == 0 { + t.Fatalf("held FIN after close = %+v, want a FIN at seq %d, directly after the tail", held, want) + } + if next != want+1 { + t.Fatalf("SendSeq after close = %d, want %d: the FIN's sequence number reserved", next, want+1) } } diff --git a/pkg/daemon/zz_segment_fits_packet_test.go b/pkg/daemon/zz_segment_fits_packet_test.go index ffde3e95..d0339f28 100644 --- a/pkg/daemon/zz_segment_fits_packet_test.go +++ b/pkg/daemon/zz_segment_fits_packet_test.go @@ -284,8 +284,10 @@ func TestCloseSendsHeldTailBeforeFIN(t *testing.T) { if string(pkts[0].Payload) != "last words" { t.Fatalf("first packet after close carries %q, want the held write", pkts[0].Payload) } - // The FIN waits for the data to be acknowledged (sendFINAfterData). + // The FIN waits for the data to be acknowledged: the ACK path sends it + // (releaseHeldFIN). conn.ProcessAck(pkts[0].Seq+uint32(len(pkts[0].Payload)), false) + d.releaseHeldFINIfAcked(conn) pkts = append(pkts, readSegments(t, pc, 1)...) if pkts[1].Flags&protocol.FlagFIN == 0 { t.Fatalf("second packet after close has flags %#x, want FIN", pkts[1].Flags) From 8f3ade43eae1d554f09301ad14e8a4e0b2572be9 Mon Sep 17 00:00:00 2001 From: Teo Calin Date: Thu, 8 Oct 2026 13:40:48 +0300 Subject: [PATCH 4/4] daemon: a SYN while the peer's FIN is held is a restarted peer; the hold ends with the connection Second review of the FIN hold: - A peer that restarts and dials the same port again (its ephemeral ports start over) matched the old connection while its FIN was held, got the old SYN-ACK back, and had its data acknowledged as duplicates and never delivered. A peer that sent a FIN is past its SYN, so that SYN now ends the old connection, quietly (a FIN-ACK would reach the new one), and opens a new connection. - The hold timer outlived the hold: after an RST, a local close, the reaper or shutdown, it later set TIME_WAIT on the removed connection and sent a FIN-ACK that could close a new connection on the same ports. The hold now ends with the connection (stopFINHold), and the timer acts only on an established connection. - CloseConnection reserves the FIN's sequence number and holds the FIN in one critical section, so shutdown cannot see the reservation without the held FIN and send a second FIN past it. - The silent-peer test has wider margins (1 s hold, data at 250 ms). Co-Authored-By: Claude Opus 5.5 --- pkg/daemon/daemon.go | 100 +++++++++++++++++++++-------- pkg/daemon/zz_fin_in_order_test.go | 76 ++++++++++++++++++++-- 2 files changed, 147 insertions(+), 29 deletions(-) diff --git a/pkg/daemon/daemon.go b/pkg/daemon/daemon.go index 0b5e6f67..d0829569 100644 --- a/pkg/daemon/daemon.go +++ b/pkg/daemon/daemon.go @@ -3361,6 +3361,26 @@ func (d *Daemon) handleStreamPacket(pkt *protocol.Packet) { // them as retransmits resends a stale SYN-ACK seq that the new client rejects, // blocking the new connection for up to TimeWaitDuration + IdleSweepInterval. if existing := d.ports.FindConnection(pkt.DstPort, pkt.Src, pkt.SrcPort); existing != nil { + // A peer that sent a FIN is past its SYN, so a SYN on the same + // ports while that FIN is held for missing data is a new + // connection from a restarted peer (its ephemeral ports start + // over), not a resend. Answered as the old connection, its data + // would be acknowledged as duplicates and never delivered. End + // the old one here, quietly: its data is not coming, and a + // FIN-ACK would reach the new connection. + existing.RecvMu.Lock() + restarted := existing.finPending + existing.RecvMu.Unlock() + if restarted { + slog.Info("new connection on the ports of one whose FIN was held — the peer restarted; closing the old one", + "conn_id", existing.ID, "remote_addr", pkt.Src, "remote_port", pkt.SrcPort) + existing.Mu.Lock() + existing.State = StateClosed + existing.Mu.Unlock() + d.stopFINHold(existing) + existing.CloseRecvBuf() + d.ports.RemoveConnection(existing.ID) + } existing.Mu.Lock() st := existing.State eAck := existing.RecvAck @@ -3664,6 +3684,7 @@ func (d *Daemon) handleStreamPacket(pkt *protocol.Packet) { conn.State = StateClosed conn.Mu.Unlock() conn.signalDial() // a refused dial ends now, not at the next backstop tick + d.stopFINHold(conn) conn.CloseRecvBuf() d.ports.RemoveConnection(conn.ID) d.publishEvent("conn.rst", map[string]interface{}{ @@ -3818,11 +3839,23 @@ const defaultFinHoldIdle = 15 * time.Second // missing data, as every version up to v1.17.0 gave it at once. Runs on the // hold timer; a FIN that delivery reached first is not handled again. func (d *Daemon) expireHeldFIN(conn *Connection) { + conn.Mu.Lock() + open := conn.State == StateEstablished + conn.Mu.Unlock() conn.RecvMu.Lock() if !conn.finPending { conn.RecvMu.Unlock() return } + if !open { + // The connection ended some other way (RST, a local close, the + // reaper, shutdown): nothing to answer, and a FIN-ACK now could + // reach a new connection on the same ports. + conn.finPending = false + conn.finHoldTimer = nil + conn.RecvMu.Unlock() + return + } if wait := d.finHoldIdle - time.Since(conn.finHeard); wait > 0 { conn.finHoldTimer.Reset(wait) conn.RecvMu.Unlock() @@ -3838,10 +3871,23 @@ func (d *Daemon) expireHeldFIN(conn *Connection) { d.acceptFIN(conn, finSeq) } +// stopFINHold ends the hold on the peer's FIN, if any, and its timer: the +// connection ended some other way, or the FIN is being handled now. +func (d *Daemon) stopFINHold(conn *Connection) { + conn.RecvMu.Lock() + conn.finPending = false + if conn.finHoldTimer != nil { + conn.finHoldTimer.Stop() + conn.finHoldTimer = nil + } + conn.RecvMu.Unlock() +} + // acceptFIN handles the peer's FIN once every byte sent before it has been // delivered (see the FIN branch of handleStreamPacket): it closes the // receive side, moves to TIME_WAIT and answers with a FIN-ACK. func (d *Daemon) acceptFIN(conn *Connection, finSeq uint32) { + d.stopFINHold(conn) conn.CloseRecvBuf() conn.Mu.Lock() wasFinWait := conn.State == StateFinWait @@ -5191,14 +5237,35 @@ func (d *Daemon) CloseConnection(conn *Connection) { // segment with the same seq as the FIN sentinel (iter 24 fix, same // pattern as the sendSegment pre-increment fix in iter 23). d.sendHeldTailBeforeClose(conn) + // A receiver up to v1.17.0 acts on a FIN that overtook a lost segment, + // closing the stream without it. With data unacknowledged the FIN is + // held and sent once the data is acknowledged (finLinger at most; + // releaseHeldFIN), so the end of a write followed at once by a close + // survives loss on such receivers too. The caller does not wait. The + // FIN's sequence number is reserved and the FIN held in one critical + // section, so shutdown never sees the one without the other. + hold := d.unackedHasData(conn) && !d.stopping() conn.Mu.Lock() st := conn.State - sendSeq := conn.SendSeq + var fin *protocol.Packet if st == StateEstablished { + fin = &protocol.Packet{ + Version: protocol.Version, + Flags: protocol.FlagFIN, + Protocol: protocol.ProtoStream, + Src: conn.LocalAddr, + Dst: conn.RemoteAddr, + SrcPort: conn.LocalPort, + DstPort: conn.RemotePort, + Seq: conn.SendSeq, + } conn.SendSeq++ // reserve FIN seq atomically with the read + if hold { + c := conn + conn.heldFIN = fin + conn.finLingerTimer = time.AfterFunc(d.finLinger, func() { d.releaseHeldFIN(c) }) + } } - localAddr := conn.LocalAddr - localPort := conn.LocalPort remoteAddr := conn.RemoteAddr remotePort := conn.RemotePort connID := conn.ID @@ -5210,34 +5277,17 @@ func (d *Daemon) CloseConnection(conn *Connection) { "conn_id": connID, }) } - if st == StateEstablished { - fin := &protocol.Packet{ - Version: protocol.Version, - Flags: protocol.FlagFIN, - Protocol: protocol.ProtoStream, - Src: localAddr, - Dst: remoteAddr, - SrcPort: localPort, - DstPort: remotePort, - Seq: sendSeq, - } - // A receiver up to v1.17.0 acts on a FIN that overtook a lost - // segment, closing the stream without it. Send the FIN once the - // data is acknowledged (finLinger at most), so the end of a write - // that is followed at once by a close survives loss on such - // receivers too. The caller does not wait. - if d.unackedHasData(conn) && !d.stopping() { - c := conn - conn.Mu.Lock() - conn.heldFIN = fin - conn.finLingerTimer = time.AfterFunc(d.finLinger, func() { d.releaseHeldFIN(c) }) - conn.Mu.Unlock() + if fin != nil { + if hold { // The last ACK may have been handled since the check above. d.releaseHeldFINIfAcked(conn) } else { d.sendFIN(conn, fin) } } + // A FIN from the peer held for missing data is moot: this side no + // longer reads. + d.stopFINHold(conn) conn.CloseRecvBuf() // P1-003: stop a pending delayed-ACK timer so it doesn't fire after // the connection is gone and queue an ACK for a dead peer. ACKTimer diff --git a/pkg/daemon/zz_fin_in_order_test.go b/pkg/daemon/zz_fin_in_order_test.go index 4fa208fb..9527b628 100644 --- a/pkg/daemon/zz_fin_in_order_test.go +++ b/pkg/daemon/zz_fin_in_order_test.go @@ -225,7 +225,7 @@ func TestHeldPeerFINIsHandledOnceThePeerFallsSilent(t *testing.T) { t.Parallel() d, peerNode, peerConn := setupDaemonWithPeer(t, Config{Public: true}) d.setNodeID_testhelper(0xABCD00F3) - d.finHoldIdle = 400 * time.Millisecond + d.finHoldIdle = time.Second conn := finTestConn(t, d, peerNode) d.handleStreamPacket(finTestData(peerNode, d.NodeID(), 1000, "abcde")) @@ -238,7 +238,7 @@ func TestHeldPeerFINIsHandledOnceThePeerFallsSilent(t *testing.T) { var finack *protocol.Packet for finack == nil { - pkt := readPacket(t, peerConn, 2*time.Second) + pkt := readPacket(t, peerConn, 3*time.Second) if pkt == nil { t.Fatal("the held FIN was never handled after the peer fell silent") } @@ -246,8 +246,8 @@ func TestHeldPeerFINIsHandledOnceThePeerFallsSilent(t *testing.T) { finack = pkt } } - if waited := time.Since(start); waited < 600*time.Millisecond { - t.Fatalf("FIN handled %s after it arrived, want the wait restarted by the data at 250 ms (>= 650 ms)", waited) + if waited := time.Since(start); waited < 1200*time.Millisecond { + t.Fatalf("FIN handled %s after it arrived, want the wait restarted by the data at 250 ms (>= 1.25 s)", waited) } if finack.Ack != 1021 { t.Fatalf("FIN-ACK ack=%d, want 1021", finack.Ack) @@ -261,6 +261,74 @@ func TestHeldPeerFINIsHandledOnceThePeerFallsSilent(t *testing.T) { } } +// A SYN on the ports of a connection whose FIN is held is a restarted +// peer dialling again (its ephemeral ports start over), not a resend: a +// peer that sent a FIN is past its SYN. The old connection ends, quietly, +// and the SYN opens a new one. Answered as the old connection, the new +// one's data was acknowledged as duplicates and never delivered. +func TestSYNOnAConnectionWithAHeldFINIsANewConnection(t *testing.T) { + t.Parallel() + d, peerNode, peerConn := setupDaemonWithPeer(t, Config{Public: true}) + d.setNodeID_testhelper(0xABCD00F4) + if _, err := d.ports.Bind(55555); err != nil { + t.Fatal(err) + } + old := finTestConn(t, d, peerNode) + d.handleStreamPacket(finTestData(peerNode, d.NodeID(), 1000, "abcde")) + d.handleStreamPacket(streamPacket(protocol.FlagFIN, peerNode, d.NodeID(), 443, 55555, 1010, 0)) + for readPacket(t, peerConn, 150*time.Millisecond) != nil { + } + + d.handleStreamPacket(streamPacket(protocol.FlagSYN, peerNode, d.NodeID(), 443, 55555, 7000, 0)) + + var got []byte + for b := range old.RecvBuf { + got = append(got, b...) + } + if string(got) != "abcde" { + t.Fatalf("old reader got %q before EOF, want the in-order data", got) + } + cur := d.ports.FindConnection(55555, protocol.Addr{Node: peerNode}, 443) + if cur == nil || cur.ID == old.ID { + t.Fatalf("connection on the ports after the SYN = %v, want a new one", cur) + } + for { + pkt := readPacket(t, peerConn, time.Second) + if pkt == nil { + t.Fatal("no SYN-ACK for the new connection") + } + if pkt.HasFlag(protocol.FlagFIN) { + t.Fatalf("a FIN (flags %d) went to the peer: it would close its new connection", pkt.Flags) + } + if pkt.HasFlag(protocol.FlagSYN) && pkt.HasFlag(protocol.FlagACK) { + if pkt.Ack != 7001 { + t.Fatalf("SYN-ACK ack=%d, want 7001: answered as the old connection", pkt.Ack) + } + break + } + } +} + +// A connection that ends some other way while the peer's FIN is held (here +// an RST) stops the hold: no FIN-ACK goes out for it later, where it could +// reach a new connection on the same ports. +func TestHoldEndsWithTheConnection(t *testing.T) { + t.Parallel() + d, peerNode, peerConn := setupDaemonWithPeer(t, Config{Public: true}) + d.setNodeID_testhelper(0xABCD00F5) + d.finHoldIdle = 200 * time.Millisecond + finTestConn(t, d, peerNode) + d.handleStreamPacket(finTestData(peerNode, d.NodeID(), 1000, "abcde")) + d.handleStreamPacket(streamPacket(protocol.FlagFIN, peerNode, d.NodeID(), 443, 55555, 1010, 0)) + d.handleStreamPacket(streamPacket(protocol.FlagRST, peerNode, d.NodeID(), 443, 55555, 0, 0)) + deadline := time.Now().Add(700 * time.Millisecond) + for time.Now().Before(deadline) { + if pkt := readPacket(t, peerConn, 100*time.Millisecond); pkt != nil && pkt.HasFlag(protocol.FlagFIN) { + t.Fatalf("FIN (flags %d) sent for a connection the peer reset", pkt.Flags) + } + } +} + // Shutdown can catch a close half done: the FIN held, the state still // ESTABLISHED. It sends that FIN, at its reserved sequence number, and no // second one after it, which a receiver would hold for good.