v2.2.0 fix(ec): enable time based out of sync tolerance

beacon/goclient/goclient.go

@@ -118,6 +118,7 @@
 
 	syncDistanceTolerance phase0.Slot
 	nodeSyncingFn         func(ctx context.Context, opts *api.NodeSyncingOpts) (*api.Response[*apiv1.SyncState], error)
+	lastHealthy           time.Time
 
 	operatorDataStore operatordatastore.OperatorDataStore
 
@@ -233,6 +234,19 @@
 
 var errSyncing = errors.New("syncing")
 
+const unhealthyDurationTolerance = 1 * time.Minute
+
+// Checks if the error occurred within the tolerance window
+// and overrides it if true
+func (gc *GoClient) checkIsWithinTolerance(err error) error {
+	unhealthyDuration := time.Since(gc.lastHealthy)
+	if unhealthyDuration < unhealthyDurationTolerance {
+		return nil
+	}
+
+	return err
+}
+
 // Healthy returns if beacon node is currently healthy: responds to requests, not in the syncing state, not optimistic
 // (for optimistic see https://github.com/ethereum/consensus-specs/blob/dev/sync/optimistic.md#block-production).
 func (gc *GoClient) Healthy(ctx context.Context) error {
@@ -244,36 +258,51 @@
 		)
 		// TODO: get rid of global variable, pass metrics to goClient
 		recordBeaconClientStatus(ctx, statusUnknown, gc.client.Address())
-		return fmt.Errorf("failed to obtain node syncing status: %w", err)
+		err = fmt.Errorf("failed to obtain node syncing status: %w", err)
+		return gc.checkIsWithinTolerance(err)
 	}
+
 	if nodeSyncingResp == nil {
 		gc.log.Error(clNilResponseErrMsg,
 			zap.String("api", "NodeSyncing"),
 		)
 		recordBeaconClientStatus(ctx, statusUnknown, gc.client.Address())
-		return fmt.Errorf("node syncing response is nil")
+		err = fmt.Errorf("node syncing response is nil")
+		return gc.checkIsWithinTolerance(err)
 	}
 	if nodeSyncingResp.Data == nil {
 		gc.log.Error(clNilResponseDataErrMsg,
 			zap.String("api", "NodeSyncing"),
 		)
 		recordBeaconClientStatus(ctx, statusUnknown, gc.client.Address())
-		return fmt.Errorf("node syncing data is nil")
+		err = fmt.Errorf("node syncing data is nil")
+		return gc.checkIsWithinTolerance(err)
 	}
 	syncState := nodeSyncingResp.Data
 	recordBeaconClientStatus(ctx, statusSyncing, gc.client.Address())
 	recordSyncDistance(ctx, syncState.SyncDistance, gc.client.Address())
 
 	// TODO: also check if syncState.ElOffline when github.com/attestantio/go-eth2-client supports it
-	if syncState.IsSyncing && syncState.SyncDistance > gc.syncDistanceTolerance {
-		gc.log.Error("Consensus client is not synced")
-		return errSyncing
+	if syncState.IsSyncing {
+		// check within the allowed distance
+		if syncState.SyncDistance > gc.syncDistanceTolerance {
+			gc.log.Error("Consensus client is not synced")
+			return errSyncing
+		}
+		unhealthyDuration := time.Since(gc.lastHealthy)
+		// within the distance but out of time
+		if unhealthyDuration > unhealthyDurationTolerance {
+			return fmt.Errorf("not synced for too long (%d): %w", unhealthyDuration, errSyncing)
+		}
 	}
+
 	if syncState.IsOptimistic {
 		gc.log.Error("Consensus client is in optimistic mode")
 		return fmt.Errorf("optimistic")
 	}
 
+	gc.lastHealthy = time.Now()
+
 	recordBeaconClientStatus(ctx, statusSynced, gc.client.Address())
 
 	return nil

beacon/goclient/goclient_test.go

@@ -3,6 +3,7 @@ package goclient
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"net/http"
 	"net/http/httptest"
@@ -18,6 +19,7 @@ import (
 	"github.com/ssvlabs/ssv/operator/slotticker"
 	"github.com/ssvlabs/ssv/protocol/v2/blockchain/beacon"
 	registrystorage "github.com/ssvlabs/ssv/registry/storage"
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"go.uber.org/zap"
 )
@@ -68,6 +70,66 @@ func TestHealthy(t *testing.T) {
 		err = client.Healthy(ctx)
 		require.NoError(t, err)
 	})
+
+	t.Run("within distance/time allowance", func(t *testing.T) {
+		lh := time.Now().Add(-59 * time.Second)
+		client.lastHealthy = lh
+		client.nodeSyncingFn = func(ctx context.Context, opts *api.NodeSyncingOpts) (*api.Response[*v1.SyncState], error) {
+			r := new(api.Response[*v1.SyncState])
+			r.Data = &v1.SyncState{
+				IsSyncing: true,
+			}
+			return r, nil
+		}
+
+		err = client.Healthy(ctx)
+		require.NoError(t, err)
+		assert.True(t, client.lastHealthy.After(lh))
+	})
+
+	t.Run("outside time allowance", func(t *testing.T) {
+		lh := time.Now().Add(-time.Minute)
+		client.lastHealthy = lh
+		client.nodeSyncingFn = func(ctx context.Context, opts *api.NodeSyncingOpts) (*api.Response[*v1.SyncState], error) {
+			r := new(api.Response[*v1.SyncState])
+			r.Data = &v1.SyncState{
+				IsSyncing: true,
+			}
+			return r, nil
+		}
+
+		err = client.Healthy(ctx)
+		require.ErrorIs(t, err, errSyncing)
+		assert.True(t, client.lastHealthy == lh)
+	})
+
+	t.Run("sync error overriden if within time limits", func(t *testing.T) {
+		client.lastHealthy = time.Now().Add(-59 * time.Second)
+		client.nodeSyncingFn = func(ctx context.Context, opts *api.NodeSyncingOpts) (*api.Response[*v1.SyncState], error) {
+			return nil, errors.New("some err")
+		}
+
+		err = client.Healthy(ctx)
+		require.NoError(t, err)
+	})
+	t.Run("sync nil response err overriden if within time limits", func(t *testing.T) {
+		client.lastHealthy = time.Now().Add(-59 * time.Second)
+		client.nodeSyncingFn = func(ctx context.Context, opts *api.NodeSyncingOpts) (*api.Response[*v1.SyncState], error) {
+			return nil, nil
+		}
+
+		err = client.Healthy(ctx)
+		require.NoError(t, err)
+	})
+	t.Run("sync nil response data err overriden if within time ", func(t *testing.T) {
+		client.lastHealthy = time.Now().Add(-59 * time.Second)
+		client.nodeSyncingFn = func(ctx context.Context, opts *api.NodeSyncingOpts) (*api.Response[*v1.SyncState], error) {
+			return new(api.Response[*v1.SyncState]), nil
+		}
+
+		err = client.Healthy(ctx)
+		require.NoError(t, err)
+	})
 }
 
 func TestTimeouts(t *testing.T) {

eth/executionclient/execution_client.go

@@ -49,12 +49,15 @@ type ExecutionClient struct {
 
 	syncDistanceTolerance uint64
 	syncProgressFn        func(context.Context) (*ethereum.SyncProgress, error)
+	lastHealthy           time.Time
 
 	// variables
 	client *ethclient.Client
 	closed chan struct{}
 }
 
+const unhealthyDurationTolerance = 1 * time.Minute
+
 // New creates a new instance of ExecutionClient.
 func New(ctx context.Context, nodeAddr string, contractAddr ethcommon.Address, opts ...Option) (*ExecutionClient, error) {
 	client := &ExecutionClient{
@@ -67,6 +70,7 @@ func New(ctx context.Context, nodeAddr string, contractAddr ethcommon.Address, o
 		reconnectionMaxInterval:     DefaultReconnectionMaxInterval,
 		logBatchSize:                DefaultHistoricalLogsBatchSize, // TODO Make batch of logs adaptive depending on "websocket: read limit"
 		closed:                      make(chan struct{}),
+		lastHealthy:                 time.Now(),
 	}
 	for _, opt := range opts {
 		opt(client)
@@ -256,7 +260,14 @@ func (ec *ExecutionClient) Healthy(ctx context.Context) error {
 		ec.logger.Error(elResponseErrMsg,
 			zap.String("method", "eth_syncing"),
 			zap.Error(err))
-		return err
+
+		unhealthyDuration := time.Since(ec.lastHealthy)
+		if unhealthyDuration < unhealthyDurationTolerance {
+			// override error if we're in the tolerance window
+			return nil
+		}
+
+		return errors.Join(fmt.Errorf("check sync progress: %w", err), errSyncing)
 	}
 	recordRequestDuration(ctx, ec.nodeAddr, time.Since(start))
 
@@ -271,6 +282,13 @@ func (ec *ExecutionClient) Healthy(ctx context.Context) error {
 		if syncDistance > ec.syncDistanceTolerance {
 			return fmt.Errorf("sync distance exceeds tolerance (%d): %w", syncDistance, errSyncing)
 		}
+	} else {
+		ec.lastHealthy = time.Now()
+	}
+
+	unhealthyDuration := time.Since(ec.lastHealthy)
+	if unhealthyDuration > unhealthyDurationTolerance {
+		return fmt.Errorf("not synced for too long (%d): %w", unhealthyDuration, errSyncing)
 	}
 
 	recordExecutionClientStatus(ctx, statusReady, ec.nodeAddr)

eth/executionclient/execution_client_test.go

@@ -2,6 +2,7 @@ package executionclient
 
 import (
 	"context"
+	"errors"
 	"math/big"
 	"net/http/httptest"
 	"strings"
@@ -605,7 +606,7 @@ func TestSimSSV(t *testing.T) {
 	require.NoError(t, sim.Close())
 }
 
-func TestSyncProgress(t *testing.T) {
+func TestHealthy(t *testing.T) {
 	const testTimeout = 1 * time.Second
 	ctx, cancel := context.WithTimeout(context.Background(), testTimeout)
 	defer cancel()
@@ -643,7 +644,7 @@ func TestSyncProgress(t *testing.T) {
 	err = client.Healthy(ctx)
 	require.NoError(t, err)
 
-	t.Run("out of sync", func(t *testing.T) {
+	t.Run("distance out of sync", func(t *testing.T) {
 		client.syncProgressFn = func(context.Context) (*ethereum.SyncProgress, error) {
 			p := new(ethereum.SyncProgress)
 			p.CurrentBlock = 5
@@ -655,7 +656,7 @@ func TestSyncProgress(t *testing.T) {
 		require.ErrorIs(t, err, errSyncing)
 	})
 
-	t.Run("within tolerable limits", func(t *testing.T) {
+	t.Run("sync distance within tolerable limits", func(t *testing.T) {
 		client, err := New(ctx, addr, contractAddr, WithSyncDistanceTolerance(2))
 		require.NoError(t, err)
 
@@ -669,6 +670,62 @@ func TestSyncProgress(t *testing.T) {
 		err = client.Healthy(ctx)
 		require.NoError(t, err)
 	})
+
+	t.Run("overrides error if within time tolerance", func(t *testing.T) {
+		client, err := New(ctx, addr, contractAddr)
+		require.NoError(t, err)
+
+		client.syncProgressFn = func(context.Context) (*ethereum.SyncProgress, error) {
+			return nil, errors.New("connection refused")
+		}
+
+		err = client.Healthy(ctx)
+		require.NoError(t, err)
+	})
+
+	t.Run("propagates error if outside of time tolerance", func(t *testing.T) {
+		client, err := New(ctx, addr, contractAddr)
+		require.NoError(t, err)
+
+		client.syncProgressFn = func(context.Context) (*ethereum.SyncProgress, error) {
+			return nil, errors.New("connection refused")
+		}
+
+		client.lastHealthy = time.Now().Add(-61 * time.Second)
+		err = client.Healthy(ctx)
+		require.ErrorIs(t, err, errSyncing)
+	})
+
+	t.Run("within block distance but outside of time tolerance", func(t *testing.T) {
+		client, err := New(ctx, addr, contractAddr)
+		require.NoError(t, err)
+
+		client.syncProgressFn = func(context.Context) (*ethereum.SyncProgress, error) {
+			p := new(ethereum.SyncProgress)
+			return p, nil
+		}
+
+		client.lastHealthy = time.Now().Add(-61 * time.Second)
+		err = client.Healthy(ctx)
+		require.ErrorIs(t, err, errSyncing)
+	})
+
+	t.Run("overwrites checkpoint on OK response", func(t *testing.T) {
+		client, err := New(ctx, addr, contractAddr)
+		require.NoError(t, err)
+
+		old := client.lastHealthy
+
+		client.syncProgressFn = func(context.Context) (*ethereum.SyncProgress, error) {
+			time.Sleep(time.Millisecond)
+			return nil, nil
+		}
+
+		err = client.Healthy(ctx)
+		require.NoError(t, err)
+
+		require.True(t, client.lastHealthy.After(old))
+	})
 }
 
 func httpToWebSocketURL(url string) string {