loss.go

     1  // Copyright 2023 The Go Authors. All rights reserved.
     2  // Use of this source code is governed by a BSD-style
     3  // license that can be found in the LICENSE file.
     4  
     5  package quic
     6  
     7  import (
     8  	"context"
     9  	"log/slog"
    10  	"math"
    11  	"time"
    12  )
    13  
    14  type lossState struct {
    15  	side connSide
    16  
    17  	// True when the handshake is confirmed.
    18  	// https://www.rfc-editor.org/rfc/rfc9001#section-4.1.2
    19  	handshakeConfirmed bool
    20  
    21  	// Peer's max_ack_delay transport parameter.
    22  	// https://www.rfc-editor.org/rfc/rfc9000.html#section-18.2-4.28.1
    23  	maxAckDelay time.Duration
    24  
    25  	// Time of the next event: PTO expiration (if ptoTimerArmed is true),
    26  	// or loss detection.
    27  	// The connection must call lossState.advance when the timer expires.
    28  	timer time.Time
    29  
    30  	// True when the PTO timer is set.
    31  	ptoTimerArmed bool
    32  
    33  	// True when the PTO timer has expired and a probe packet has not yet been sent.
    34  	ptoExpired bool
    35  
    36  	// Count of PTO expirations since the lack received acknowledgement.
    37  	// https://www.rfc-editor.org/rfc/rfc9002#section-6.2.1-9
    38  	ptoBackoffCount int
    39  
    40  	// Anti-amplification limit: Three times the amount of data received from
    41  	// the peer, less the amount of data sent.
    42  	//
    43  	// Set to antiAmplificationUnlimited (MaxInt) to disable the limit.
    44  	// The limit is always disabled for clients, and for servers after the
    45  	// peer's address is validated.
    46  	//
    47  	// Anti-amplification is per-address; this will need to change if/when we
    48  	// support address migration.
    49  	//
    50  	// https://www.rfc-editor.org/rfc/rfc9000#section-8-2
    51  	antiAmplificationLimit int
    52  
    53  	// Count of non-ack-eliciting packets (ACKs) sent since the last ack-eliciting one.
    54  	consecutiveNonAckElicitingPackets int
    55  
    56  	rtt   rttState
    57  	pacer pacerState
    58  	cc    *ccReno
    59  
    60  	// Per-space loss detection state.
    61  	spaces [numberSpaceCount]struct {
    62  		sentPacketList
    63  		maxAcked         packetNumber
    64  		lastAckEliciting packetNumber
    65  	}
    66  
    67  	// Temporary state used when processing an ACK frame.
    68  	ackFrameRTT                  time.Duration // RTT from latest packet in frame
    69  	ackFrameContainsAckEliciting bool          // newly acks an ack-eliciting packet?
    70  }
    71  
    72  const antiAmplificationUnlimited = math.MaxInt
    73  
    74  func (c *lossState) init(side connSide, maxDatagramSize int, now time.Time) {
    75  	c.side = side
    76  	if side == clientSide {
    77  		// Clients don't have an anti-amplification limit.
    78  		c.antiAmplificationLimit = antiAmplificationUnlimited
    79  	}
    80  	c.rtt.init()
    81  	c.cc = newReno(maxDatagramSize)
    82  	c.pacer.init(now, c.cc.congestionWindow, timerGranularity)
    83  
    84  	// Peer's assumed max_ack_delay, prior to receiving transport parameters.
    85  	// https://www.rfc-editor.org/rfc/rfc9000#section-18.2
    86  	c.maxAckDelay = 25 * time.Millisecond
    87  
    88  	for space := range c.spaces {
    89  		c.spaces[space].maxAcked = -1
    90  		c.spaces[space].lastAckEliciting = -1
    91  	}
    92  }
    93  
    94  // setMaxAckDelay sets the max_ack_delay transport parameter received from the peer.
    95  func (c *lossState) setMaxAckDelay(d time.Duration) {
    96  	if d >= (1<<14)*time.Millisecond {
    97  		// Values of 2^14 or greater are invalid.
    98  		// https://www.rfc-editor.org/rfc/rfc9000.html#section-18.2-4.28.1
    99  		return
   100  	}
   101  	c.maxAckDelay = d
   102  }
   103  
   104  // confirmHandshake indicates the handshake has been confirmed.
   105  func (c *lossState) confirmHandshake() {
   106  	c.handshakeConfirmed = true
   107  }
   108  
   109  // validateClientAddress disables the anti-amplification limit after
   110  // a server validates a client's address.
   111  func (c *lossState) validateClientAddress() {
   112  	c.antiAmplificationLimit = antiAmplificationUnlimited
   113  }
   114  
   115  // minDatagramSize is the minimum datagram size permitted by
   116  // anti-amplification protection.
   117  //
   118  // Defining a minimum size avoids the case where, say, anti-amplification
   119  // technically allows us to send a 1-byte datagram, but no such datagram
   120  // can be constructed.
   121  const minPacketSize = 128
   122  
   123  type ccLimit int
   124  
   125  const (
   126  	ccOK      = ccLimit(iota) // OK to send
   127  	ccBlocked                 // sending blocked by anti-amplification
   128  	ccLimited                 // sending blocked by congestion control
   129  	ccPaced                   // sending allowed by congestion, but delayed by pacer
   130  )
   131  
   132  // sendLimit reports whether sending is possible at this time.
   133  // When sending is pacing limited, it returns the next time a packet may be sent.
   134  func (c *lossState) sendLimit(now time.Time) (limit ccLimit, next time.Time) {
   135  	if c.antiAmplificationLimit < minPacketSize {
   136  		// When at the anti-amplification limit, we may not send anything.
   137  		return ccBlocked, time.Time{}
   138  	}
   139  	if c.ptoExpired {
   140  		// On PTO expiry, send a probe.
   141  		return ccOK, time.Time{}
   142  	}
   143  	if !c.cc.canSend() {
   144  		// Congestion control blocks sending.
   145  		return ccLimited, time.Time{}
   146  	}
   147  	if c.cc.bytesInFlight == 0 {
   148  		// If no bytes are in flight, send packet unpaced.
   149  		return ccOK, time.Time{}
   150  	}
   151  	canSend, next := c.pacer.canSend(now)
   152  	if !canSend {
   153  		// Pacer blocks sending.
   154  		return ccPaced, next
   155  	}
   156  	return ccOK, time.Time{}
   157  }
   158  
   159  // maxSendSize reports the maximum datagram size that may be sent.
   160  func (c *lossState) maxSendSize() int {
   161  	return min(c.antiAmplificationLimit, c.cc.maxDatagramSize)
   162  }
   163  
   164  // advance is called when time passes.
   165  // The lossf function is called for each packet newly detected as lost.
   166  func (c *lossState) advance(now time.Time, lossf func(numberSpace, *sentPacket, packetFate)) {
   167  	c.pacer.advance(now, c.cc.congestionWindow, c.rtt.smoothedRTT)
   168  	if c.ptoTimerArmed && !c.timer.IsZero() && !c.timer.After(now) {
   169  		c.ptoExpired = true
   170  		c.timer = time.Time{}
   171  		c.ptoBackoffCount++
   172  	}
   173  	c.detectLoss(now, lossf)
   174  }
   175  
   176  // nextNumber returns the next packet number to use in a space.
   177  func (c *lossState) nextNumber(space numberSpace) packetNumber {
   178  	return c.spaces[space].nextNum
   179  }
   180  
   181  // skipNumber skips a packet number as a defense against optimistic ACK attacks.
   182  func (c *lossState) skipNumber(now time.Time, space numberSpace) {
   183  	sent := newSentPacket()
   184  	sent.num = c.spaces[space].nextNum
   185  	sent.time = now
   186  	sent.state = sentPacketUnsent
   187  	c.spaces[space].add(sent)
   188  }
   189  
   190  // packetSent records a sent packet.
   191  func (c *lossState) packetSent(now time.Time, log *slog.Logger, space numberSpace, sent *sentPacket) {
   192  	sent.time = now
   193  	c.spaces[space].add(sent)
   194  	size := sent.size
   195  	if c.antiAmplificationLimit != antiAmplificationUnlimited {
   196  		c.antiAmplificationLimit = max(0, c.antiAmplificationLimit-size)
   197  	}
   198  	if sent.inFlight {
   199  		c.cc.packetSent(now, log, space, sent)
   200  		c.pacer.packetSent(now, size, c.cc.congestionWindow, c.rtt.smoothedRTT)
   201  		if sent.ackEliciting {
   202  			c.spaces[space].lastAckEliciting = sent.num
   203  			c.ptoExpired = false // reset expired PTO timer after sending probe
   204  		}
   205  		c.scheduleTimer(now)
   206  		if logEnabled(log, QLogLevelPacket) {
   207  			logBytesInFlight(log, c.cc.bytesInFlight)
   208  		}
   209  	}
   210  	if sent.ackEliciting {
   211  		c.consecutiveNonAckElicitingPackets = 0
   212  	} else {
   213  		c.consecutiveNonAckElicitingPackets++
   214  	}
   215  }
   216  
   217  // datagramReceived records a datagram (not packet!) received from the peer.
   218  func (c *lossState) datagramReceived(now time.Time, size int) {
   219  	if c.antiAmplificationLimit != antiAmplificationUnlimited {
   220  		c.antiAmplificationLimit += 3 * size
   221  		// Reset the PTO timer, possibly to a point in the past, in which
   222  		// case the caller should execute it immediately.
   223  		// https://www.rfc-editor.org/rfc/rfc9002.html#section-6.2.2.1-2
   224  		c.scheduleTimer(now)
   225  		if c.ptoTimerArmed && !c.timer.IsZero() && !c.timer.After(now) {
   226  			c.ptoExpired = true
   227  			c.timer = time.Time{}
   228  		}
   229  	}
   230  }
   231  
   232  // receiveAckStart starts processing an ACK frame.
   233  // Call receiveAckRange for each range in the frame.
   234  // Call receiveAckFrameEnd after all ranges are processed.
   235  func (c *lossState) receiveAckStart() {
   236  	c.ackFrameContainsAckEliciting = false
   237  	c.ackFrameRTT = -1
   238  }
   239  
   240  // receiveAckRange processes a range within an ACK frame.
   241  // The ackf function is called for each newly-acknowledged packet.
   242  func (c *lossState) receiveAckRange(now time.Time, space numberSpace, rangeIndex int, start, end packetNumber, ackf func(numberSpace, *sentPacket, packetFate)) error {
   243  	// Limit our range to the intersection of the ACK range and
   244  	// the in-flight packets we have state for.
   245  	if s := c.spaces[space].start(); start < s {
   246  		start = s
   247  	}
   248  	if e := c.spaces[space].end(); end > e {
   249  		return localTransportError{
   250  			code:   errProtocolViolation,
   251  			reason: "acknowledgement for unsent packet",
   252  		}
   253  	}
   254  	if start >= end {
   255  		return nil
   256  	}
   257  	if rangeIndex == 0 {
   258  		// If the latest packet in the ACK frame is newly-acked,
   259  		// record the RTT in c.ackFrameRTT.
   260  		sent := c.spaces[space].num(end - 1)
   261  		if sent.state == sentPacketSent {
   262  			c.ackFrameRTT = max(0, now.Sub(sent.time))
   263  		}
   264  	}
   265  	for pnum := start; pnum < end; pnum++ {
   266  		sent := c.spaces[space].num(pnum)
   267  		if sent.state == sentPacketUnsent {
   268  			return localTransportError{
   269  				code:   errProtocolViolation,
   270  				reason: "acknowledgement for unsent packet",
   271  			}
   272  		}
   273  		if sent.state != sentPacketSent {
   274  			continue
   275  		}
   276  		// This is a newly-acknowledged packet.
   277  		if pnum > c.spaces[space].maxAcked {
   278  			c.spaces[space].maxAcked = pnum
   279  		}
   280  		sent.state = sentPacketAcked
   281  		c.cc.packetAcked(now, sent)
   282  		ackf(space, sent, packetAcked)
   283  		if sent.ackEliciting {
   284  			c.ackFrameContainsAckEliciting = true
   285  		}
   286  	}
   287  	return nil
   288  }
   289  
   290  // receiveAckEnd finishes processing an ack frame.
   291  // The lossf function is called for each packet newly detected as lost.
   292  func (c *lossState) receiveAckEnd(now time.Time, log *slog.Logger, space numberSpace, ackDelay time.Duration, lossf func(numberSpace, *sentPacket, packetFate)) {
   293  	c.spaces[space].sentPacketList.clean()
   294  	// Update the RTT sample when the largest acknowledged packet in the ACK frame
   295  	// is newly acknowledged, and at least one newly acknowledged packet is ack-eliciting.
   296  	// https://www.rfc-editor.org/rfc/rfc9002.html#section-5.1-2.2
   297  	if c.ackFrameRTT >= 0 && c.ackFrameContainsAckEliciting {
   298  		c.rtt.updateSample(now, c.handshakeConfirmed, space, c.ackFrameRTT, ackDelay, c.maxAckDelay)
   299  	}
   300  	// Reset the PTO backoff.
   301  	// Exception: A client does not reset the backoff on acks for Initial packets.
   302  	// https://www.rfc-editor.org/rfc/rfc9002.html#section-6.2.1-9
   303  	if !(c.side == clientSide && space == initialSpace) {
   304  		c.ptoBackoffCount = 0
   305  	}
   306  	// If the client has set a PTO timer with no packets in flight
   307  	// we want to restart that timer now. Clearing c.timer does this.
   308  	// https://www.rfc-editor.org/rfc/rfc9002.html#section-6.2.2.1-3
   309  	c.timer = time.Time{}
   310  	c.detectLoss(now, lossf)
   311  	c.cc.packetBatchEnd(now, log, space, &c.rtt, c.maxAckDelay)
   312  
   313  	if logEnabled(log, QLogLevelPacket) {
   314  		var ssthresh slog.Attr
   315  		if c.cc.slowStartThreshold != math.MaxInt {
   316  			ssthresh = slog.Int("ssthresh", c.cc.slowStartThreshold)
   317  		}
   318  		log.LogAttrs(context.Background(), QLogLevelPacket,
   319  			"recovery:metrics_updated",
   320  			slog.Duration("min_rtt", c.rtt.minRTT),
   321  			slog.Duration("smoothed_rtt", c.rtt.smoothedRTT),
   322  			slog.Duration("latest_rtt", c.rtt.latestRTT),
   323  			slog.Duration("rtt_variance", c.rtt.rttvar),
   324  			slog.Int("congestion_window", c.cc.congestionWindow),
   325  			slog.Int("bytes_in_flight", c.cc.bytesInFlight),
   326  			ssthresh,
   327  		)
   328  	}
   329  }
   330  
   331  // discardPackets declares that packets within a number space will not be delivered
   332  // and that data contained in them should be resent.
   333  // For example, after receiving a Retry packet we discard already-sent Initial packets.
   334  func (c *lossState) discardPackets(space numberSpace, log *slog.Logger, lossf func(numberSpace, *sentPacket, packetFate)) {
   335  	for i := 0; i < c.spaces[space].size; i++ {
   336  		sent := c.spaces[space].nth(i)
   337  		if sent.state != sentPacketSent {
   338  			// This should not be possible, since we only discard packets
   339  			// in spaces which have never received an ack, but check anyway.
   340  			continue
   341  		}
   342  		sent.state = sentPacketLost
   343  		c.cc.packetDiscarded(sent)
   344  		lossf(numberSpace(space), sent, packetLost)
   345  	}
   346  	c.spaces[space].clean()
   347  	if logEnabled(log, QLogLevelPacket) {
   348  		logBytesInFlight(log, c.cc.bytesInFlight)
   349  	}
   350  }
   351  
   352  // discardKeys is called when dropping packet protection keys for a number space.
   353  func (c *lossState) discardKeys(now time.Time, log *slog.Logger, space numberSpace) {
   354  	// https://www.rfc-editor.org/rfc/rfc9002.html#section-6.4
   355  	for i := 0; i < c.spaces[space].size; i++ {
   356  		sent := c.spaces[space].nth(i)
   357  		if sent.state != sentPacketSent {
   358  			continue
   359  		}
   360  		c.cc.packetDiscarded(sent)
   361  	}
   362  	c.spaces[space].discard()
   363  	c.spaces[space].maxAcked = -1
   364  	c.spaces[space].lastAckEliciting = -1
   365  	c.scheduleTimer(now)
   366  	if logEnabled(log, QLogLevelPacket) {
   367  		logBytesInFlight(log, c.cc.bytesInFlight)
   368  	}
   369  }
   370  
   371  func (c *lossState) lossDuration() time.Duration {
   372  	// https://www.rfc-editor.org/rfc/rfc9002.html#section-6.1.2
   373  	return max((9*max(c.rtt.smoothedRTT, c.rtt.latestRTT))/8, timerGranularity)
   374  }
   375  
   376  func (c *lossState) detectLoss(now time.Time, lossf func(numberSpace, *sentPacket, packetFate)) {
   377  	// https://www.rfc-editor.org/rfc/rfc9002.html#section-6.1.1-1
   378  	const lossThreshold = 3
   379  
   380  	lossTime := now.Add(-c.lossDuration())
   381  	for space := numberSpace(0); space < numberSpaceCount; space++ {
   382  		for i := 0; i < c.spaces[space].size; i++ {
   383  			sent := c.spaces[space].nth(i)
   384  			if sent.state != sentPacketSent {
   385  				continue
   386  			}
   387  			// RFC 9002 Section 6.1 states that a packet is only declared lost if it
   388  			// is "in flight", which excludes packets that contain only ACK frames.
   389  			// However, we need some way to determine when to drop state for ACK-only
   390  			// packets, and the loss algorithm in Appendix A handles loss detection of
   391  			// not-in-flight packets identically to all others, so we do the same here.
   392  			switch {
   393  			case c.spaces[space].maxAcked-sent.num >= lossThreshold:
   394  				// Packet threshold
   395  				// https://www.rfc-editor.org/rfc/rfc9002.html#section-6.1.1
   396  				fallthrough
   397  			case sent.num <= c.spaces[space].maxAcked && !sent.time.After(lossTime):
   398  				// Time threshold
   399  				// https://www.rfc-editor.org/rfc/rfc9002.html#section-6.1.2
   400  				sent.state = sentPacketLost
   401  				lossf(space, sent, packetLost)
   402  				if sent.inFlight {
   403  					c.cc.packetLost(now, space, sent, &c.rtt)
   404  				}
   405  			}
   406  			if sent.state != sentPacketLost {
   407  				break
   408  			}
   409  		}
   410  		c.spaces[space].clean()
   411  	}
   412  	c.scheduleTimer(now)
   413  }
   414  
   415  // scheduleTimer sets the loss or PTO timer.
   416  //
   417  // The connection is responsible for arranging for advance to be called after
   418  // the timer expires.
   419  //
   420  // The timer may be set to a point in the past, in which advance should be called
   421  // immediately. We don't do this here, because executing the timer can cause
   422  // packet loss events, and it's simpler for the connection if loss events only
   423  // occur when advancing time.
   424  func (c *lossState) scheduleTimer(now time.Time) {
   425  	c.ptoTimerArmed = false
   426  
   427  	// Loss timer for sent packets.
   428  	// The loss timer is only started once a later packet has been acknowledged,
   429  	// and takes precedence over the PTO timer.
   430  	// https://www.rfc-editor.org/rfc/rfc9002.html#section-6.1.2
   431  	var oldestPotentiallyLost time.Time
   432  	for space := numberSpace(0); space < numberSpaceCount; space++ {
   433  		if c.spaces[space].size > 0 && c.spaces[space].start() <= c.spaces[space].maxAcked {
   434  			firstTime := c.spaces[space].nth(0).time
   435  			if oldestPotentiallyLost.IsZero() || firstTime.Before(oldestPotentiallyLost) {
   436  				oldestPotentiallyLost = firstTime
   437  			}
   438  		}
   439  	}
   440  	if !oldestPotentiallyLost.IsZero() {
   441  		c.timer = oldestPotentiallyLost.Add(c.lossDuration())
   442  		return
   443  	}
   444  
   445  	// PTO timer.
   446  	if c.ptoExpired {
   447  		// PTO timer has expired, don't restart it until we send a probe.
   448  		c.timer = time.Time{}
   449  		return
   450  	}
   451  	if c.antiAmplificationLimit >= 0 && c.antiAmplificationLimit < minPacketSize {
   452  		// Server is at its anti-amplification limit and can't send any more data.
   453  		// https://www.rfc-editor.org/rfc/rfc9002.html#section-6.2.2.1-1
   454  		c.timer = time.Time{}
   455  		return
   456  	}
   457  	// Timer starts at the most recently sent ack-eliciting packet.
   458  	// Prior to confirming the handshake, we consider the Initial and Handshake
   459  	// number spaces; after, we consider only Application Data.
   460  	var last time.Time
   461  	if !c.handshakeConfirmed {
   462  		for space := initialSpace; space <= handshakeSpace; space++ {
   463  			sent := c.spaces[space].num(c.spaces[space].lastAckEliciting)
   464  			if sent == nil {
   465  				continue
   466  			}
   467  			if last.IsZero() || last.After(sent.time) {
   468  				last = sent.time
   469  			}
   470  		}
   471  	} else {
   472  		sent := c.spaces[appDataSpace].num(c.spaces[appDataSpace].lastAckEliciting)
   473  		if sent != nil {
   474  			last = sent.time
   475  		}
   476  	}
   477  	if last.IsZero() &&
   478  		c.side == clientSide &&
   479  		c.spaces[handshakeSpace].maxAcked < 0 &&
   480  		!c.handshakeConfirmed {
   481  		// The client must always set a PTO timer prior to receiving an ack for a
   482  		// handshake packet or the handshake being confirmed.
   483  		// https://www.rfc-editor.org/rfc/rfc9002.html#section-6.2.2.1
   484  		if !c.timer.IsZero() {
   485  			// If c.timer is non-zero here, we've already set the PTO timer and
   486  			// should leave it as-is rather than moving it forward.
   487  			c.ptoTimerArmed = true
   488  			return
   489  		}
   490  		last = now
   491  	} else if last.IsZero() {
   492  		c.timer = time.Time{}
   493  		return
   494  	}
   495  	c.timer = last.Add(c.ptoPeriod())
   496  	c.ptoTimerArmed = true
   497  }
   498  
   499  func (c *lossState) ptoPeriod() time.Duration {
   500  	// https://www.rfc-editor.org/rfc/rfc9002.html#section-6.2.1
   501  	return c.ptoBasePeriod() << c.ptoBackoffCount
   502  }
   503  
   504  func (c *lossState) ptoBasePeriod() time.Duration {
   505  	// https://www.rfc-editor.org/rfc/rfc9002.html#section-6.2.1
   506  	pto := c.rtt.smoothedRTT + max(4*c.rtt.rttvar, timerGranularity)
   507  	if c.handshakeConfirmed {
   508  		// The max_ack_delay is the maximum amount of time the peer might delay sending
   509  		// an ack to us. We only take it into account for the Application Data space.
   510  		// https://www.rfc-editor.org/rfc/rfc9002.html#section-6.2.1-4
   511  		pto += c.maxAckDelay
   512  	}
   513  	return pto
   514  }
   515  
   516  func logBytesInFlight(log *slog.Logger, bytesInFlight int) {
   517  	log.LogAttrs(context.Background(), QLogLevelPacket,
   518  		"recovery:metrics_updated",
   519  		slog.Int("bytes_in_flight", bytesInFlight),
   520  	)
   521  }
   522
View as plain text