diff --git a/CHANGELOG.md b/CHANGELOG.md index 57a11834..8468314a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -394,6 +394,21 @@ 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. 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 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..d0829569 100644 --- a/pkg/daemon/daemon.go +++ b/pkg/daemon/daemon.go @@ -479,6 +479,13 @@ type Daemon struct { 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 @@ -664,6 +671,8 @@ func New(cfg Config) *Daemon { tunnels: NewTunnelManager(), ports: NewPortManager(), stopCh: make(chan struct{}), + finLinger: defaultFinLinger, + finHoldIdle: defaultFinHoldIdle, hsPoll: newHandshakePollSched(), synTokens: cfg.synRateLimit(), synLastFill: time.Now(), @@ -1726,8 +1735,19 @@ func (d *Daemon) doStop() { conn.Mu.Lock() st := conn.State seq := conn.SendSeq + held, lingerTimer := conn.heldFIN, conn.finLingerTimer + conn.heldFIN, conn.finLingerTimer = nil, nil conn.Mu.Unlock() - if st == StateEstablished { + 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) + } else if st == StateEstablished { // Send FIN fin := &protocol.Packet{ Version: protocol.Version, @@ -3341,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 @@ -3594,55 +3634,45 @@ 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() + // 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 { + 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 { + d.sendDelayedACK(conn) + return } + d.acceptFIN(conn, pkt.Seq) return } @@ -3654,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{}{ @@ -3712,6 +3743,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 { @@ -3734,6 +3767,26 @@ 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 + if conn.finHoldTimer != nil { + conn.finHoldTimer.Stop() + conn.finHoldTimer = nil + } + } else if conn.finPending { + conn.finHeard = time.Now() + } + 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 +3827,118 @@ 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.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() + 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) +} + +// 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 + 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 @@ -5072,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 @@ -5091,35 +5277,17 @@ func (d *Daemon) CloseConnection(conn *Connection) { "conn_id": connID, }) } - if st == StateEstablished { - finData := []byte{0} // 1-byte sentinel so retxEntry has non-zero length - fin := &protocol.Packet{ - Version: protocol.Version, - Flags: protocol.FlagFIN, - Protocol: protocol.ProtoStream, - Src: localAddr, - Dst: remoteAddr, - SrcPort: localPort, - DstPort: remotePort, - Seq: sendSeq, + if fin != nil { + if hold { + // The last ACK may have been handled since the check above. + d.releaseHeldFINIfAcked(conn) + } else { + d.sendFIN(conn, fin) } - 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 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 @@ -5138,6 +5306,62 @@ 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() +} + +// 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, timer := conn.heldFIN, conn.finLingerTimer + conn.heldFIN, conn.finLingerTimer = nil, nil + st := conn.State + conn.Mu.Unlock() + 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. @@ -6303,6 +6527,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 e9a66419..ea4c91ae 100644 --- a/pkg/daemon/ports.go +++ b/pkg/daemon/ports.go @@ -287,6 +287,13 @@ 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; 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 @@ -336,6 +343,15 @@ 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), 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 new file mode 100644 index 00000000..9527b628 --- /dev/null +++ b/pkg/daemon/zz_fin_in_order_test.go @@ -0,0 +1,487 @@ +// 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")) + // 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)) + + 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") + } + 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")) + + 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) + } +} + +// 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) + } +} + +// 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 = time.Second + 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, 3*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 < 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) + } + 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) + } +} + +// 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. +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 +// 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) + + 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): + 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 42dbd92e..06d6dacf 100644 --- a/pkg/daemon/zz_large_write_pieces_test.go +++ b/pkg/daemon/zz_large_write_pieces_test.go @@ -247,11 +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)) } - 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 comes directly after the tail. It is held until the data is + // acknowledged (releaseHeldFIN), never sent here. + conn.Mu.Lock() + held, next := conn.heldFIN, conn.SendSeq + conn.Mu.Unlock() + 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 0fde1ab1..d0339f28 100644 --- a/pkg/daemon/zz_segment_fits_packet_test.go +++ b/pkg/daemon/zz_segment_fits_packet_test.go @@ -280,10 +280,15 @@ 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: 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) } 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)