Merge pull request #310 from ava-labs/timeout-dont-include-get

dont include Get messages in request latency calculation

Author: StephenButtolph

snow/networking/router/chain_router.go

@@ -161,7 +161,7 @@ func (cr *ChainRouter) RegisterRequest(
 		cr.log.Error("expected message type to be one of GetMsg, PullQueryMsg, PushQueryMsg, GetAcceptedFrontierMsg, GetAcceptedMsg but got %s", msgType)
 		return
 	}
-	cr.timeoutManager.RegisterRequest(validatorID, chainID, uniqueRequestID, timeoutHandler)
+	cr.timeoutManager.RegisterRequest(validatorID, chainID, msgType, uniqueRequestID, timeoutHandler)
 }
 
 // Shutdown shuts down this router

snow/networking/timeout/manager.go

@@ -62,6 +62,7 @@ func (m *Manager) RegisterChain(ctx *snow.Context, namespace string) error {
 func (m *Manager) RegisterRequest(
 	validatorID ids.ShortID,
 	chainID ids.ID,
+	msgType constants.MsgType,
 	uniqueRequestID ids.ID,
 	timeoutHandler func(),
 ) (time.Time, bool) {
@@ -70,7 +71,7 @@ func (m *Manager) RegisterRequest(
 		m.benchlistMgr.RegisterFailure(chainID, validatorID)
 		timeoutHandler()
 	}
-	return m.tm.Put(uniqueRequestID, newTimeoutHandler), true
+	return m.tm.Put(uniqueRequestID, msgType, newTimeoutHandler), true
 }
 
 // RegisterResponse registers that we received a response from [validatorID]

snow/networking/timeout/manager_test.go

@@ -36,7 +36,7 @@ func TestManagerFire(t *testing.T) {
 	wg := sync.WaitGroup{}
 	wg.Add(1)
 
-	manager.RegisterRequest(ids.ShortID{}, ids.ID{}, ids.GenerateTestID(), wg.Done)
+	manager.RegisterRequest(ids.ShortID{}, ids.ID{}, constants.PullQueryMsg, ids.GenerateTestID(), wg.Done)
 
 	wg.Wait()
 }
@@ -64,11 +64,11 @@ func TestManagerCancel(t *testing.T) {
 	fired := new(bool)
 
 	id := ids.GenerateTestID()
-	manager.RegisterRequest(ids.ShortID{}, ids.ID{}, id, func() { *fired = true })
+	manager.RegisterRequest(ids.ShortID{}, ids.ID{}, constants.PullQueryMsg, id, func() { *fired = true })
 
 	manager.RegisterResponse(ids.ShortID{}, ids.ID{}, id, constants.GetMsg, 1*time.Second)
 
-	manager.RegisterRequest(ids.ShortID{}, ids.ID{}, ids.GenerateTestID(), wg.Done)
+	manager.RegisterRequest(ids.ShortID{}, ids.ID{}, constants.PullQueryMsg, ids.GenerateTestID(), wg.Done)
 
 	wg.Wait()
 

utils/timer/adaptive_timeout_manager.go

@@ -13,6 +13,7 @@ import (
 	"github.com/prometheus/client_golang/prometheus"
 
 	"github.com/ava-labs/avalanchego/ids"
+	"github.com/ava-labs/avalanchego/utils/constants"
 	"github.com/ava-labs/avalanchego/utils/math"
 	"github.com/ava-labs/avalanchego/utils/wrappers"
 )
@@ -22,11 +23,12 @@ var (
 )
 
 type adaptiveTimeout struct {
-	index    int           // Index in the wait queue
-	id       ids.ID        // Unique ID of this timeout
-	handler  func()        // Function to execute if timed out
-	duration time.Duration // How long this timeout was set for
-	deadline time.Time     // When this timeout should be fired
+	index    int               // Index in the wait queue
+	id       ids.ID            // Unique ID of this timeout
+	handler  func()            // Function to execute if timed out
+	duration time.Duration     // How long this timeout was set for
+	deadline time.Time         // When this timeout should be fired
+	msgType  constants.MsgType // Type of this outstanding request
 }
 
 // A timeoutQueue implements heap.Interface and holds adaptiveTimeouts.
@@ -78,6 +80,7 @@ type AdaptiveTimeoutManager struct {
 	// Tells the time. Can be faked for testing.
 	clock                            Clock
 	networkTimeoutMetric, avgLatency prometheus.Gauge
+	numTimeouts                      prometheus.Counter
 	// Averages the response time from all peers
 	averager math.Averager
 	// Timeout is [timeoutCoefficient] * average response time
@@ -103,6 +106,11 @@ func (tm *AdaptiveTimeoutManager) Initialize(config *AdaptiveTimeoutConfig) erro
 		Name:      "avg_network_latency",
 		Help:      "Average network latency in nanoseconds",
 	})
+	tm.numTimeouts = prometheus.NewCounter(prometheus.CounterOpts{
+		Namespace: config.MetricsNamespace,
+		Name:      "request_timeouts",
+		Help:      "Number of timed out requests",
+	})
 
 	switch {
 	case config.InitialTimeout > config.MaximumTimeout:
@@ -126,6 +134,7 @@ func (tm *AdaptiveTimeoutManager) Initialize(config *AdaptiveTimeoutConfig) erro
 	errs := &wrappers.Errs{}
 	errs.Add(config.Registerer.Register(tm.networkTimeoutMetric))
 	errs.Add(config.Registerer.Register(tm.avgLatency))
+	errs.Add(config.Registerer.Register(tm.numTimeouts))
 	return errs.Err
 }
 
@@ -145,14 +154,14 @@ func (tm *AdaptiveTimeoutManager) Stop() { tm.timer.Stop() }
 // Put registers a timeout for [id]. If the timeout occurs, [timeoutHandler] is called.
 // Returns the time at which the timeout will fire if it is not first
 // removed by calling [tm.Remove].
-func (tm *AdaptiveTimeoutManager) Put(id ids.ID, timeoutHandler func()) time.Time {
+func (tm *AdaptiveTimeoutManager) Put(id ids.ID, msgType constants.MsgType, timeoutHandler func()) time.Time {
 	tm.lock.Lock()
 	defer tm.lock.Unlock()
-	return tm.put(id, timeoutHandler)
+	return tm.put(id, msgType, timeoutHandler)
 }
 
 // Assumes [tm.lock] is held
-func (tm *AdaptiveTimeoutManager) put(id ids.ID, handler func()) time.Time {
+func (tm *AdaptiveTimeoutManager) put(id ids.ID, msgType constants.MsgType, handler func()) time.Time {
 	currentTime := tm.clock.Time()
 	tm.remove(id, currentTime)
 
@@ -161,6 +170,7 @@ func (tm *AdaptiveTimeoutManager) put(id ids.ID, handler func()) time.Time {
 		handler:  handler,
 		duration: tm.currentTimeout,
 		deadline: currentTime.Add(tm.currentTimeout),
+		msgType:  msgType,
 	}
 	tm.timeoutMap[id] = timeout
 	heap.Push(&tm.timeoutQueue, timeout)
@@ -185,9 +195,14 @@ func (tm *AdaptiveTimeoutManager) remove(id ids.ID, now time.Time) {
 	}
 
 	// Observe the response time to update average network response time
-	timeoutRegisteredAt := timeout.deadline.Add(-1 * timeout.duration)
-	latency := now.Sub(timeoutRegisteredAt)
-	tm.observeLatencyAndUpdateTimeout(latency, now)
+	// Don't include Get requests in calculation, since an adversary
+	// can cause you to issue a Get request and then cause it to timeout,
+	// increasing your timeout.
+	if timeout.msgType != constants.GetMsg {
+		timeoutRegisteredAt := timeout.deadline.Add(-1 * timeout.duration)
+		latency := now.Sub(timeoutRegisteredAt)
+		tm.observeLatencyAndUpdateTimeout(latency, now)
+	}
 
 	// Remove the timeout from the map
 	delete(tm.timeoutMap, id)
@@ -213,6 +228,7 @@ func (tm *AdaptiveTimeoutManager) timeout() {
 		if timeoutHandler == nil {
 			break
 		}
+		tm.numTimeouts.Inc()
 
 		// Don't execute a callback with a lock held
 		tm.lock.Unlock()

utils/timer/adaptive_timeout_manager_test.go

@@ -135,6 +135,7 @@ func TestAdaptiveTimeoutManager(t *testing.T) {
 	timeoutZeroCalled := utils.AtomicBool{}
 	tm.Put(
 		id0,
+		constants.PullQueryMsg,
 		func() { timeoutZeroCalled.SetValue(true) },
 	)
 
@@ -171,6 +172,7 @@ func TestAdaptiveTimeoutManager(t *testing.T) {
 	// This should overwrite the first Put for id0
 	tm.Put(
 		id0,
+		constants.PullQueryMsg,
 		func() { wg.Done() },
 	)
 
@@ -208,10 +210,12 @@ func TestAdaptiveTimeoutManager(t *testing.T) {
 	wg.Add(2)
 	tm.Put(
 		id1,
+		constants.PullQueryMsg,
 		func() { wg.Done() },
 	)
 	tm.Put(
 		id2,
+		constants.PullQueryMsg,
 		func() { wg.Done() },
 	)
 
@@ -270,14 +274,14 @@ func TestAdaptiveTimeoutManager2(t *testing.T) {
 
 		numSuccessful--
 		if numSuccessful > 0 {
-			tm.Put(ids.ID{byte(numSuccessful)}, *callback)
+			tm.Put(ids.ID{byte(numSuccessful)}, constants.PullQueryMsg, *callback)
 		}
 		if numSuccessful >= 0 {
 			wg.Done()
 		}
 		if numSuccessful%2 == 0 {
 			tm.Remove(ids.ID{byte(numSuccessful)})
-			tm.Put(ids.ID{byte(numSuccessful)}, *callback)
+			tm.Put(ids.ID{byte(numSuccessful)}, constants.PullQueryMsg, *callback)
 		}
 	}
 	(*callback)()