tcp: do not leak SACKed-then-ACKed segments into Outstanding outside recovery

pkg/tcpip/transport/tcp/snd.go

@@ -184,6 +184,17 @@ type sender struct {
 	// RFC3522 Section 3.2.
 	retransmitTS uint32
 
+	// sackedOutsideRecovery is the number of packets that were newly
+	// SACKed while no loss recovery was in progress. SetPipe() only
+	// accounts SACKed segments into Outstanding during loss recovery, and
+	// the cumulative-ACK path skips previously-SACKed segments assuming
+	// SetPipe() accounted for them; packets counted here are the ones
+	// that assumption misses, and their count is removed from Outstanding
+	// when they are cumulatively ACKed. Reset whenever Outstanding is
+	// recomputed or reset wholesale (recovery entry, RTO, full window
+	// drain). See gvisor.dev/issue/14092.
+	sackedOutsideRecovery int
+
 	// startCork start corking the segments.
 	startCork bool
 
@@ -631,6 +642,7 @@ func (s *sender) retransmitTimerExpired() tcpip.Error {
 	// We'll keep on transmitting (or retransmitting) as we get acks for
 	// the data we transmit.
 	s.Outstanding = 0
+	s.sackedOutsideRecovery = 0
 
 	// Expunge all SACK information as per https://tools.ietf.org/html/rfc6675#section-5.1
 	//
@@ -1137,6 +1149,8 @@ func (s *sender) enterRecovery() {
 	// the 3 duplicate ACKs and are now not in flight.
 	s.SndCwnd = s.Ssthresh + 3
 	s.SackedOut = 0
+	// SetPipe() owns all SACK accounting for the duration of recovery.
+	s.sackedOutsideRecovery = 0
 	s.DupAckCount = 0
 	s.FastRecovery.First = s.SndUna
 	s.FastRecovery.Last = s.SndNxt - 1
@@ -1556,8 +1570,27 @@ func (s *sender) handleRcvdSegment(rcvdSeg *segment) {
 			// which have start/end before sndUna and are used to
 			// indicate spurious retransmissions.
 			if rcvdSeg.ackNumber.LessThan(sb.Start) && s.SndUna.LessThan(sb.Start) && sb.End.LessThanEq(s.SndNxt) && !s.ep.scoreboard.IsSACKED(sb) {
+				// Track segments that this block newly covers while
+				// no loss recovery is in progress: SetPipe() will
+				// not account for them (it is a no-op outside
+				// recovery), so the cumulative-ACK path must, or
+				// their packet count leaks into Outstanding
+				// permanently. See gvisor.dev/issue/14092.
+				var newlyCovered []*segment
+				if !s.FastRecovery.Active {
+					for seg := s.writeList.Front(); seg != nil && seg.sequenceNumber.LessThan(sb.End); seg = seg.Next() {
+						if seg.xmitCount != 0 && !s.ep.scoreboard.IsSACKED(seg.sackBlock()) {
+							newlyCovered = append(newlyCovered, seg)
+						}
+					}
+				}
 				s.ep.scoreboard.Insert(sb)
 				rcvdSeg.hasNewSACKInfo = true
+				for _, seg := range newlyCovered {
+					if s.ep.scoreboard.IsSACKED(seg.sackBlock()) {
+						s.sackedOutsideRecovery += s.pCount(seg, s.MaxPayloadSize)
+					}
+				}
 			}
 		}
 
@@ -1694,11 +1727,21 @@ func (s *sender) handleRcvdSegment(rcvdSeg *segment) {
 
 			// If SACK is enabled then only reduce outstanding if
 			// the segment was not previously SACKED as these have
-			// already been accounted for in SetPipe().
+			// already been accounted for in SetPipe() (during loss
+			// recovery) or in sackedOutsideRecovery below (outside
+			// of it).
 			if !s.ep.SACKPermitted || !s.ep.scoreboard.IsSACKED(seg.sackBlock()) {
 				s.Outstanding -= s.pCount(seg, s.MaxPayloadSize)
 			} else {
 				s.SackedOut -= s.pCount(seg, s.MaxPayloadSize)
+				// SetPipe() only accounts for SACKed segments
+				// during loss recovery; segments SACKed outside
+				// of it are recorded in sackedOutsideRecovery
+				// and must be removed from Outstanding here.
+				if n := min(s.pCount(seg, s.MaxPayloadSize), s.sackedOutsideRecovery); n > 0 {
+					s.Outstanding -= n
+					s.sackedOutsideRecovery -= n
+				}
 			}
 			seg.DecRef()
 			ackLeft -= datalen
@@ -1744,6 +1787,7 @@ func (s *sender) handleRcvdSegment(rcvdSeg *segment) {
 		// RFC 6298 Rule 5.3
 		if s.SndUna == s.SndNxt {
 			s.Outstanding = 0
+			s.sackedOutsideRecovery = 0
 			// Reset firstRetransmittedSegXmitTime to the zero value.
 			s.firstRetransmittedSegXmitTime = tcpip.MonotonicTime{}
 			s.resendTimer.disable()

pkg/tcpip/transport/tcp/test/e2e/tcp_sack_test.go

@@ -947,6 +947,56 @@ func TestNoSpuriousRecoveryWithDSACK(t *testing.T) {
 	verifySpuriousRecoveryMetric(t, c, 0 /* numSpuriousRecovery */, 0 /* numSpuriousRTO */)
 }
 
+// TestSACKedThenAckedSegmentLeavesOutstanding is a regression test for
+// gvisor.dev/issue/14092: a segment SACKed outside of loss recovery and then
+// cumulatively ACKed must be removed from Outstanding. SetPipe() only
+// accounts for SACKed segments during recovery, so before the fix the
+// segment's packet count leaked into Outstanding permanently, throttling
+// long-lived connections on reordering-but-lossless paths.
+func TestSACKedThenAckedSegmentLeavesOutstanding(t *testing.T) {
+	for _, enableRACK := range []bool{true, false} {
+		t.Run(fmt.Sprintf("enableRACK: %v", enableRACK), func(t *testing.T) {
+			const numPackets = 4
+			var c *context.Context
+			outstanding := make(chan int, 1)
+			probe := func(state *tcp.TCPEndpointState) {
+				// Match the cumulative ACK of the first two
+				// segments; the other two are still in flight.
+				if state.Sender.SndUna == c.IRS.Add(1+2*maxPayload) {
+					select {
+					case outstanding <- state.Sender.Outstanding:
+					default:
+					}
+				}
+			}
+			c = context.NewWithProbe(t, uint32(mtu), probe)
+			defer c.Cleanup()
+
+			e2e.SendAndReceiveWithSACK(t, c, maxPayload, numPackets, enableRACK)
+
+			seq := seqnum.Value(context.TestInitialSequenceNumber).Add(1)
+			// SACK the second segment. A single duplicate ACK does
+			// not trigger loss recovery: this is benign reordering,
+			// not loss, so SetPipe() never runs to account for the
+			// SACKed segment.
+			c.SendAckWithSACK(seq, 0, []header.SACKBlock{
+				{Start: c.IRS.Add(1 + maxPayload), End: c.IRS.Add(1 + 2*maxPayload)},
+			})
+			// Cumulatively ACK the first two segments, covering the
+			// SACKed one.
+			c.SendAck(seq, 2*maxPayload)
+
+			if got, want := <-outstanding, 2; got != want {
+				t.Errorf("got Sender.Outstanding = %d after a cumulative ACK covering a SACKed segment, want %d", got, want)
+			}
+
+			// ACK the remaining data so the endpoint quiesces for
+			// the leak check.
+			c.SendAck(seq, numPackets*maxPayload)
+		})
+	}
+}
+
 func TestMain(m *testing.M) {
 	refs.SetLeakMode(refs.LeaksPanic)
 	code := m.Run()