shift the max value logic to health monitor

Signed-off-by: Yashvardhan Kukreja <[email protected]>

Author: yashvardhan-kukreja

op-conductor/health/monitor.go

@@ -65,8 +65,9 @@ func NewSequencerHealthMonitor(log log.Logger, metrics metrics.Metricer, interva
 		}
 	}
 	if rollupBoostToleratePartialHealthinessToleranceLimit != 0 {
+		hm.rollupBoostPartialHealthinessToleranceLimit = rollupBoostToleratePartialHealthinessToleranceLimit
 		var err error
-		hm.timeTolerantRollupBoostPartialHealthinessMgr, err = NewTimeBoundedRotatingCounter(rollupBoostToleratePartialHealthinessToleranceIntervalSeconds, rollupBoostToleratePartialHealthinessToleranceLimit)
+		hm.rollupBoostPartialHealthinessToleranceCounter, err = NewTimeBoundedRotatingCounter(rollupBoostToleratePartialHealthinessToleranceIntervalSeconds)
 		if err != nil {
 			panic(fmt.Errorf("failed to setup health monitor: %w", err))
 		}
@@ -100,12 +101,13 @@ type SequencerHealthMonitor struct {
 
 	timeProviderFn func() uint64
 
-	node                                         dial.RollupClientInterface
-	p2p                                          apis.P2PClient
-	supervisor                                   SupervisorHealthAPI
-	rb                                           client.RollupBoostClient
-	elP2p                                        *ElP2pHealthMonitor
-	timeTolerantRollupBoostPartialHealthinessMgr *timeBoundedRotatingCounter
+	node                                          dial.RollupClientInterface
+	p2p                                           apis.P2PClient
+	supervisor                                    SupervisorHealthAPI
+	rb                                            client.RollupBoostClient
+	elP2p                                         *ElP2pHealthMonitor
+	rollupBoostPartialHealthinessToleranceLimit   uint64
+	rollupBoostPartialHealthinessToleranceCounter *timeBoundedRotatingCounter
 }
 
 var _ HealthMonitor = (*SequencerHealthMonitor)(nil)
@@ -296,11 +298,10 @@ func (hm *SequencerHealthMonitor) checkRollupBoost(ctx context.Context) error {
 	case client.HealthStatusHealthy:
 		return nil
 	case client.HealthStatusPartial:
-		if hm.timeTolerantRollupBoostPartialHealthinessMgr != nil {
-			if _, err := hm.timeTolerantRollupBoostPartialHealthinessMgr.Increment(); err == nil {
-				hm.log.Warn("[Tolerating Failure] Rollup boost is partial failure, builder is down but fallback execution client is up", "err", ErrRollupBoostPartiallyHealthy)
-				return nil
-			}
+		if hm.rollupBoostPartialHealthinessToleranceCounter != nil && hm.rollupBoostPartialHealthinessToleranceCounter.CurrentValue() < hm.rollupBoostPartialHealthinessToleranceLimit {
+			latestValue := hm.rollupBoostPartialHealthinessToleranceCounter.Increment()
+			hm.log.Debug("Rollup-boost partial unhealthiness failure tolerated", "currentValue", latestValue, "limit", hm.rollupBoostPartialHealthinessToleranceLimit)
+			return nil
 		}
 		hm.log.Error("Rollup boost is partial failure, builder is down but fallback execution client is up", "err", ErrRollupBoostPartiallyHealthy)
 		return ErrRollupBoostPartiallyHealthy

op-conductor/health/monitor_test.go

@@ -473,57 +473,52 @@ func (s *HealthMonitorTestSuite) TestRollupBoostPartialStatusWithTolerance() {
 	rb.EXPECT().Healthcheck(mock.Anything).Return(client.HealthStatusPartial, nil)
 
 	toleranceLimit := uint64(2)
-	toleranceIntervalSeconds := uint64(3)
+	toleranceIntervalSeconds := uint64(6)
 
-	timeBoundedRotatingCounter, err := NewTimeBoundedRotatingCounter(toleranceIntervalSeconds, toleranceLimit)
+	timeBoundedRotatingCounter, err := NewTimeBoundedRotatingCounter(toleranceIntervalSeconds)
 	s.Nil(err)
 
+	tp := &timeProvider{now: 1758792282}
+
 	// Start monitor with all dependencies as well as tolerance of 2 rollup-boost partial unhealthiness per 3s period
 	monitor := s.SetupMonitorWithRollupBoost(now, 60, 60, rc, pc, rb, nil, func(shm *SequencerHealthMonitor) {
-		tp := &timeProvider{now: 1758792282}
 		timeBoundedRotatingCounter.timeProviderFn = tp.Now
 
-		// pollute the cache of timeBoundRotatingCounter with 999 elements so as to later test the lazy cleanup
-		// note: the 1000th element will be added by the first healthchecl run
+		// pollute the cache of timeBoundRotatingCounter with 998 elements so as to later test the lazy cleanup
+		// note: the 999th and 1000th element will be added by the first healthcheck run
 		for i := 0; i < 999; i++ {
 			timeBoundedRotatingCounter.temporalCache[int64(i)] = uint64(1)
 		}
 
-		shm.timeTolerantRollupBoostPartialHealthinessMgr = timeBoundedRotatingCounter
+		shm.rollupBoostPartialHealthinessToleranceCounter = timeBoundedRotatingCounter
+		shm.rollupBoostPartialHealthinessToleranceLimit = toleranceLimit
 	})
 
 	healthUpdateCh := monitor.Subscribe()
 
-	// first error is tolerated (time t+1)
-	healthStatus := <-healthUpdateCh
-	s.Nil(healthStatus)
-	s.Len(timeBoundedRotatingCounter.temporalCache, 1000) // lazy cleanup of the cache not done yet as it's within the bounds
+	s.Eventually(func() bool {
+		return len(timeBoundedRotatingCounter.temporalCache) == 1000
+	}, time.Second*3, time.Second*1)
 
-	// second error is tolerated as well (time t+2)
-	healthStatus = <-healthUpdateCh
-	s.Nil(healthStatus)
-	s.Len(timeBoundedRotatingCounter.temporalCache, 1000) // no change of the cache until the next reset
+	firstHealthStatus := <-healthUpdateCh
+	secondHealthStatus := <-healthUpdateCh
+	thirdHealthStatus := <-healthUpdateCh
 
-	// third error isn't tolerated (time t+3)
-	healthFailure := <-healthUpdateCh
-	s.Equal(ErrRollupBoostPartiallyHealthy, healthFailure)
+	s.Nil(firstHealthStatus)
+	s.Nil(secondHealthStatus)
+	s.Equal(ErrRollupBoostPartiallyHealthy, thirdHealthStatus)
 
-	// by now, because of three healthchecks, three seconds have been simulated to pass (by the timeProviderFn)
+	tp.Now() // simulate another second passing
+	// by now, because of three healthchecks, six seconds (CurrentValue + Increment + CurrentValue + Increment + CurrentValue + tp.Now()) have been simulated to pass (by the timeProviderFn)
 	// this should reset the time bound counter, thereby allowing partial unhealthiness failures to be tolerated again
 
-	// first error after the reset is tolerated (time t+4)
-	healthStatus = <-healthUpdateCh
-	s.Nil(healthStatus)
-	s.Len(timeBoundedRotatingCounter.temporalCache, 1) // lazy cleanup of the cache done and it's left with only the current value
-
-	// second error after the reset is tolerated as well (time t+5)
-	healthStatus = <-healthUpdateCh
-	s.Nil(healthStatus)
-	s.Len(timeBoundedRotatingCounter.temporalCache, 1) // no change to the cache until the next reset
+	fourthHealthStatus := <-healthUpdateCh
+	fifthHealthStatus := <-healthUpdateCh
+	sixthHealthStatus := <-healthUpdateCh
 
-	// third error after the reset isn't tolerated (time t+6)
-	healthFailure = <-healthUpdateCh
-	s.Equal(ErrRollupBoostPartiallyHealthy, healthFailure)
+	s.Nil(fourthHealthStatus)
+	s.Nil(fifthHealthStatus)
+	s.Equal(ErrRollupBoostPartiallyHealthy, sixthHealthStatus)
 
 	s.NoError(monitor.Stop())
 }

op-conductor/health/timeboundcounter.go

@@ -5,31 +5,29 @@ import (
 	"sync"
 )
 
-// this is a type of counter which keeps on incrementing until its reset interval is hit
+// this is a type of counter which keeps on incrementing until its reset interval is hit after which it resets to 0
 // this can be used to track time-based rate-limit, error counts, etc.
 type timeBoundedRotatingCounter struct {
 	resetIntervalSeconds uint64
-	maxValue             uint64
 	timeProviderFn       func() uint64
 
 	mut           *sync.RWMutex
 	temporalCache map[int64]uint64
 }
 
-func NewTimeBoundedRotatingCounter(resetIntervalSeconds, maxValue uint64) (*timeBoundedRotatingCounter, error) {
+func NewTimeBoundedRotatingCounter(resetIntervalSeconds uint64) (*timeBoundedRotatingCounter, error) {
 	if resetIntervalSeconds == 0 {
 		return nil, fmt.Errorf("reset interval seconds must be more than 0")
 	}
 	return &timeBoundedRotatingCounter{
 		resetIntervalSeconds: resetIntervalSeconds,
-		maxValue:             maxValue,
 		mut:                  &sync.RWMutex{},
 		temporalCache:        map[int64]uint64{},
 		timeProviderFn:       currentTimeProvider,
 	}, nil
 }
 
-func (t *timeBoundedRotatingCounter) Increment() (uint64, error) {
+func (t *timeBoundedRotatingCounter) Increment() uint64 {
 	// let's take `resetIntervalSeconds` as 60s
 	// truncatedTimestamp is current timestamp rounded off by 60s (resetIntervalSeconds)
 	// thereby generating a value which stays same until the next 60s helping track and incrementing the counter corresponding to it for the next 60s
@@ -47,11 +45,8 @@ func (t *timeBoundedRotatingCounter) Increment() (uint64, error) {
 		}
 	}()
 
-	if t.maxValue == 0 || t.temporalCache[truncatedTimestamp] < t.maxValue {
-		t.temporalCache[truncatedTimestamp]++
-		return t.temporalCache[truncatedTimestamp], nil
-	}
-	return t.maxValue, fmt.Errorf("counter at its max value, please wait %ds for it to be reset", (t.resetIntervalSeconds - (currentTsSeconds % t.resetIntervalSeconds)))
+	t.temporalCache[truncatedTimestamp]++
+	return t.temporalCache[truncatedTimestamp]
 }
 
 func (t *timeBoundedRotatingCounter) CurrentValue() uint64 {

op-conductor/health/timeboundcounter_test.go

@@ -10,13 +10,13 @@ import (
 func TestTimeBoundedRotatingCounterSetup(t *testing.T) {
 	t.Parallel()
 	t.Run("fail with 0 interval seconds value", func(t *testing.T) {
-		counter, err := NewTimeBoundedRotatingCounter(0, 0)
+		counter, err := NewTimeBoundedRotatingCounter(0)
 		require.Error(t, err)
 		require.Nil(t, counter)
 	})
 
-	t.Run("succeed with 0 max value", func(t *testing.T) {
-		counter, err := NewTimeBoundedRotatingCounter(2, 0)
+	t.Run("succeed with non-zero interval seconds value", func(t *testing.T) {
+		counter, err := NewTimeBoundedRotatingCounter(2)
 		require.NoError(t, err)
 		require.NotNil(t, counter)
 	})
@@ -26,8 +26,8 @@ func TestTimeBoundedRotatingCounterIncrement(t *testing.T) {
 
 	mockTimeProvider := &timeProvider{now: 0} // every access to .Now() will increment its value simulating a one-second time passing
 
-	maxValue, resetInterval := uint64(2), uint64(6)
-	counter, err := NewTimeBoundedRotatingCounter(resetInterval, maxValue)
+	resetInterval := uint64(6)
+	counter, err := NewTimeBoundedRotatingCounter(resetInterval)
 	require.NoError(t, err)
 	require.NotNil(t, counter)
 	counter.timeProviderFn = mockTimeProvider.Now
@@ -36,44 +36,38 @@ func TestTimeBoundedRotatingCounterIncrement(t *testing.T) {
 	require.Equal(t, uint64(0), counter.CurrentValue())
 	require.Equal(t, int(mockTimeProvider.now), 1)
 
-	newValue, err := counter.Increment()
-	require.NoError(t, err)
+	newValue := counter.Increment()
 	require.Equal(t, uint64(1), newValue)
 	require.Equal(t, int(mockTimeProvider.now), 2)
 	require.Equal(t, uint64(1), counter.CurrentValue())
 	require.Equal(t, int(mockTimeProvider.now), 3)
 
-	newValue, err = counter.Increment()
-	require.NoError(t, err)
+	newValue = counter.Increment()
 	require.Equal(t, uint64(2), newValue)
 	require.Equal(t, int(mockTimeProvider.now), 4)
 	require.Equal(t, uint64(2), counter.CurrentValue())
 	require.Equal(t, int(mockTimeProvider.now), 5)
 
-	newValue, err = counter.Increment()
-	require.Error(t, err)
-	require.Equal(t, uint64(2), newValue)
+	newValue = counter.Increment()
+	require.Equal(t, uint64(3), newValue)
 	require.Equal(t, int(mockTimeProvider.now), 6)
 	require.Equal(t, uint64(0), counter.CurrentValue()) // the next second counter rotates returning 0 as the current value
 	require.Equal(t, int(mockTimeProvider.now), 7)
 
-	newValue, err = counter.Increment()
-	require.NoError(t, err)
+	newValue = counter.Increment()
 	require.Equal(t, uint64(1), newValue)
 	require.Equal(t, int(mockTimeProvider.now), 8)
 	require.Equal(t, uint64(1), counter.CurrentValue())
 	require.Equal(t, int(mockTimeProvider.now), 9)
 
-	newValue, err = counter.Increment()
-	require.NoError(t, err)
+	newValue = counter.Increment()
 	require.Equal(t, uint64(2), newValue)
 	require.Equal(t, int(mockTimeProvider.now), 10)
 	require.Equal(t, uint64(2), counter.CurrentValue())
 	require.Equal(t, int(mockTimeProvider.now), 11)
 
-	newValue, err = counter.Increment()
-	require.Error(t, err)
-	require.Equal(t, uint64(2), newValue)
+	newValue = counter.Increment()
+	require.Equal(t, uint64(3), newValue)
 	require.Equal(t, int(mockTimeProvider.now), 12)
 	require.Equal(t, uint64(0), counter.CurrentValue()) // the next second counter rotates returning 0 as the current value
 	require.Equal(t, int(mockTimeProvider.now), 13)
@@ -85,7 +79,7 @@ func TestTimeBoundedRotatingCounterIncrement(t *testing.T) {
 func TestTimeBoundedRotatingCounterConcurrentAccess(t *testing.T) {
 	mockTimeProvider := &timeProvider{now: 0}
 
-	counter, err := NewTimeBoundedRotatingCounter(1, 9)
+	counter, err := NewTimeBoundedRotatingCounter(1)
 	require.NoError(t, err)
 	require.NotNil(t, counter)
 	counter.timeProviderFn = mockTimeProvider.Now
@@ -95,8 +89,7 @@ func TestTimeBoundedRotatingCounterConcurrentAccess(t *testing.T) {
 
 	write := func() {
 		defer wg.Done()
-		_, err := counter.Increment()
-		require.NoError(t, err) // considering the max value is 9, the increment should never fail
+		counter.Increment()
 	}
 	read := func() {
 		defer wg.Done()
@@ -116,21 +109,19 @@ func TestTimeBoundedRotatingCounterLazyCleanup(t *testing.T) {
 
 	// a counter with a reset interval of 2 ensuring every two-seconds the counter's cache would track a new key:value
 	// we'll trigger the 2-second increment by calling .Increment() and .CurrentValue() because both under the hood, would call .Now() of the mockTimeProvider
-	counter, err := NewTimeBoundedRotatingCounter(2, 9)
+	counter, err := NewTimeBoundedRotatingCounter(2)
 	require.NoError(t, err)
 	require.NotNil(t, counter)
 	counter.timeProviderFn = mockTimeProvider.Now
 
 	for i := 0; i < 1000; i++ {
-		_, err := counter.Increment() // trigger a 1-second time increase
-		require.NoError(t, err)
+		counter.Increment()    // trigger a 1-second time increase
 		counter.CurrentValue() // trigger another 1-second time increase, causing the counter interval to reset ensuring next Increment would write a new key in the cache
 	}
 
 	require.Equal(t, 1000, len(counter.temporalCache))
 
 	// 1001th increment should trigger the lazy cleanup this time
-	_, err = counter.Increment()
-	require.NoError(t, err)
+	counter.Increment()
 	require.Equal(t, 1, len(counter.temporalCache))
 }