From a724d1efc006c6943c02490bbf82d01a0cb885dd Mon Sep 17 00:00:00 2001 From: tac0turtle Date: Tue, 1 Sep 2026 11:42:04 +0200 Subject: [PATCH 1/2] fix(node): fail closed during sequencer recovery --- docs/learn/config.md | 2 +- node/failover.go | 109 ++++++++---- node/failover_test.go | 173 ++++++++++++++++++++ node/sequencer_recovery_integration_test.go | 23 +-- pkg/config/config.go | 6 +- pkg/sync/sync_service.go | 19 ++- pkg/sync/sync_service_test.go | 34 ++++ 7 files changed, 307 insertions(+), 59 deletions(-) create mode 100644 node/failover_test.go diff --git a/docs/learn/config.md b/docs/learn/config.md index b66fbcfcc1..948fdede0d 100644 --- a/docs/learn/config.md +++ b/docs/learn/config.md @@ -338,7 +338,7 @@ _Constant:_ `FlagScrapeInterval` ### Catchup Timeout **Description:** -When set to a non-zero duration, the aggregator syncs from DA and P2P before producing blocks. The value specifies how long to wait for P2P catchup after DA sync completes. Requires aggregator mode. Mutually exclusive with Raft consensus. +When set to a non-zero duration, the aggregator recovers before producing blocks. If `p2p.peers` is configured, both header and data sync must initialize from P2P and the node must reach the highest observed P2P height; failure to establish continuity before this timeout is fatal. With no configured peers, recovery waits for the DA head and does not apply a P2P deadline. The default `0` disables recovery catchup. Requires aggregator mode. Mutually exclusive with Raft consensus. **YAML:** diff --git a/node/failover.go b/node/failover.go index 1ae1134b4d..f226669dcf 100644 --- a/node/failover.go +++ b/node/failover.go @@ -5,6 +5,7 @@ import ( "errors" "fmt" "net/http" + "strings" "sync/atomic" "time" @@ -39,8 +40,12 @@ type failoverState struct { // catchup fields — used when the aggregator needs to sync before producing catchupEnabled bool catchupTimeout time.Duration + p2pRecovery bool daBlockTime time.Duration store store.Store + + // catchupStatusFn is overridden by focused readiness tests. + catchupStatusFn func(context.Context) (catchupStatus, error) } func newSyncMode( @@ -182,6 +187,7 @@ func setupFailoverState( store: rktStore, catchupEnabled: catchupEnabled, catchupTimeout: nodeConfig.Node.CatchupTimeout.Duration, + p2pRecovery: strings.TrimSpace(nodeConfig.P2P.Peers) != "", daBlockTime: nodeConfig.DA.BlockTime.Duration, }, nil } @@ -293,10 +299,10 @@ func (f *failoverState) Run(pCtx context.Context) (multiErr error) { return wg.Wait() } -// runCatchupPhase starts the catchup syncer, waits until DA head is reached and P2P -// is caught up, then stops the syncer so the executor can take over. +// runCatchupPhase starts the catchup syncer, waits until the configured recovery +// sources are caught up, then stops the syncer so the executor can take over. func (f *failoverState) runCatchupPhase(ctx context.Context) error { - f.logger.Info().Msg("catchup: syncing from DA and P2P before producing blocks") + f.logger.Info().Msg("catchup: syncing from configured recovery sources before producing blocks") if err := f.bc.Syncer.Start(ctx); err != nil { return fmt.Errorf("catchup syncer start: %w", err) @@ -315,7 +321,56 @@ func (f *failoverState) runCatchupPhase(ctx context.Context) error { return nil } -// waitForCatchup polls DA and P2P catchup status until both sources indicate the node is caught up. +type catchupStatus struct { + storeHeight uint64 + headerHeight uint64 + dataHeight uint64 + headerP2PReady bool + dataP2PReady bool + daCaughtUp bool + pendingEvents int +} + +func (s catchupStatus) ready(p2pRecovery bool) bool { + if !s.daCaughtUp || s.pendingEvents != 0 { + return false + } + if !p2pRecovery { + return true + } + return s.p2pReady() +} + +func (s catchupStatus) p2pReady() bool { + return s.headerP2PReady && s.dataP2PReady && + s.storeHeight >= max(s.headerHeight, s.dataHeight) +} + +func (f *failoverState) catchupStatus(ctx context.Context) (catchupStatus, error) { + if f.catchupStatusFn != nil { + return f.catchupStatusFn(ctx) + } + + storeHeight, err := f.store.Height(ctx) + if err != nil { + return catchupStatus{}, err + } + status := catchupStatus{ + storeHeight: storeHeight, + headerHeight: f.headerSyncService.Store().Height(), + dataHeight: f.dataSyncService.Store().Height(), + headerP2PReady: f.headerSyncService.P2PInitialized(), + dataP2PReady: f.dataSyncService.P2PInitialized(), + } + if f.bc.Syncer != nil { + status.daCaughtUp = f.bc.Syncer.HasReachedDAHead() + status.pendingEvents = f.bc.Syncer.PendingCount() + } + return status, nil +} + +// waitForCatchup polls DA and, when peers are configured, P2P catchup status +// until all required sources indicate the node is caught up. func (f *failoverState) waitForCatchup(ctx context.Context) (bool, error) { pollInterval := f.daBlockTime if pollInterval <= 0 { @@ -326,46 +381,44 @@ func (f *failoverState) waitForCatchup(ctx context.Context) (bool, error) { defer ticker.Stop() var timeoutCh <-chan time.Time - if f.catchupTimeout > 0 { + if f.p2pRecovery && f.catchupTimeout > 0 { f.logger.Debug().Dur("p2p_timeout", f.catchupTimeout).Msg("P2P catchup timeout configured") timeoutCh = time.After(f.catchupTimeout) } else { - f.logger.Debug().Msg("P2P catchup timeout disabled, relying on DA only") + f.logger.Debug().Msg("configured P2P recovery not required, relying on DA only") } - ignoreP2P := false for { select { case <-ctx.Done(): - return false, nil + return false, ctx.Err() case <-timeoutCh: - f.logger.Info().Msg("catchup: P2P timeout reached, ignoring P2P status") - ignoreP2P = true - timeoutCh = nil + status, err := f.catchupStatus(ctx) + if err != nil { + return false, fmt.Errorf("P2P recovery timed out after %s and failed to read recovery heights: %w", f.catchupTimeout, err) + } + if status.p2pReady() { + timeoutCh = nil + continue + } + return false, fmt.Errorf( + "P2P recovery timed out after %s before continuity was established (store height %d, header height %d, data height %d)", + f.catchupTimeout, status.storeHeight, status.headerHeight, status.dataHeight, + ) case <-ticker.C: - daCaughtUp := f.bc.Syncer != nil && f.bc.Syncer.HasReachedDAHead() - - storeHeight, err := f.store.Height(ctx) + status, err := f.catchupStatus(ctx) if err != nil { f.logger.Warn().Err(err).Msg("failed to get store height during catchup") continue } - - maxP2PHeight := max( - f.headerSyncService.Store().Height(), - f.dataSyncService.Store().Height(), - ) - - p2pCaughtUp := ignoreP2P || (maxP2PHeight > 0 && storeHeight >= maxP2PHeight) - if !ignoreP2P && f.catchupTimeout == 0 && maxP2PHeight == 0 { - p2pCaughtUp = true + if f.p2pRecovery && status.p2pReady() { + timeoutCh = nil } - - pipelineDrained := f.bc.Syncer == nil || f.bc.Syncer.PendingCount() == 0 - if daCaughtUp && p2pCaughtUp && pipelineDrained { + if status.ready(f.p2pRecovery) { f.logger.Info(). - Uint64("store_height", storeHeight). - Uint64("max_p2p_height", maxP2PHeight). + Uint64("store_height", status.storeHeight). + Uint64("header_height", status.headerHeight). + Uint64("data_height", status.dataHeight). Msg("catchup: fully caught up") return true, nil } diff --git a/node/failover_test.go b/node/failover_test.go new file mode 100644 index 0000000000..8c0aafbe0a --- /dev/null +++ b/node/failover_test.go @@ -0,0 +1,173 @@ +package node + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/rs/zerolog" + "github.com/stretchr/testify/require" +) + +func TestCatchupStatusReady(t *testing.T) { + tests := []struct { + name string + p2pRecovery bool + status catchupStatus + want bool + }{ + { + name: "DA only ready", + status: catchupStatus{ + daCaughtUp: true, + }, + want: true, + }, + { + name: "configured peers not initialized", + p2pRecovery: true, + status: catchupStatus{ + daCaughtUp: true, + }, + }, + { + name: "only header P2P initialized", + p2pRecovery: true, + status: catchupStatus{ + daCaughtUp: true, + headerP2PReady: true, + }, + }, + { + name: "only data P2P initialized", + p2pRecovery: true, + status: catchupStatus{ + daCaughtUp: true, + dataP2PReady: true, + }, + }, + { + name: "observed header height ahead of store", + p2pRecovery: true, + status: catchupStatus{ + storeHeight: 9, + headerHeight: 10, + dataHeight: 9, + headerP2PReady: true, + dataP2PReady: true, + daCaughtUp: true, + }, + }, + { + name: "observed data height ahead of store", + p2pRecovery: true, + status: catchupStatus{ + storeHeight: 9, + headerHeight: 9, + dataHeight: 10, + headerP2PReady: true, + dataP2PReady: true, + daCaughtUp: true, + }, + }, + { + name: "pending catchup events", + p2pRecovery: true, + status: catchupStatus{ + storeHeight: 10, + headerHeight: 10, + dataHeight: 10, + headerP2PReady: true, + dataP2PReady: true, + daCaughtUp: true, + pendingEvents: 1, + }, + }, + { + name: "combined DA and P2P ready", + p2pRecovery: true, + status: catchupStatus{ + storeHeight: 11, + headerHeight: 10, + dataHeight: 11, + headerP2PReady: true, + dataP2PReady: true, + daCaughtUp: true, + }, + want: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + require.Equal(t, tt.want, tt.status.ready(tt.p2pRecovery)) + }) + } +} + +func TestWaitForCatchupP2PTimeoutFailsClosed(t *testing.T) { + f := &failoverState{ + logger: zerolog.Nop(), + p2pRecovery: true, + catchupTimeout: 20 * time.Millisecond, + daBlockTime: time.Millisecond, + catchupStatusFn: func(context.Context) (catchupStatus, error) { + return catchupStatus{ + storeHeight: 7, + headerHeight: 9, + dataHeight: 8, + }, nil + }, + } + + caughtUp, err := f.waitForCatchup(t.Context()) + require.False(t, caughtUp) + require.ErrorContains(t, err, "P2P recovery timed out") + require.ErrorContains(t, err, "store height 7") + require.ErrorContains(t, err, "header height 9") + require.ErrorContains(t, err, "data height 8") +} + +func TestWaitForCatchupContextCancellation(t *testing.T) { + ctx, cancel := context.WithCancelCause(t.Context()) + wantErr := errors.New("operator canceled recovery") + cancel(wantErr) + + f := &failoverState{ + logger: zerolog.Nop(), + p2pRecovery: true, + catchupTimeout: time.Hour, + daBlockTime: time.Hour, + } + + caughtUp, err := f.waitForCatchup(ctx) + require.False(t, caughtUp) + require.ErrorIs(t, err, context.Canceled) + require.Equal(t, wantErr, context.Cause(ctx)) +} + +func TestWaitForCatchupP2PBudgetDoesNotLimitDARecovery(t *testing.T) { + statusCalls := 0 + f := &failoverState{ + logger: zerolog.Nop(), + p2pRecovery: true, + catchupTimeout: 10 * time.Millisecond, + daBlockTime: time.Millisecond, + catchupStatusFn: func(context.Context) (catchupStatus, error) { + statusCalls++ + return catchupStatus{ + storeHeight: 10, + headerHeight: 10, + dataHeight: 10, + headerP2PReady: true, + dataP2PReady: true, + daCaughtUp: statusCalls > 15, + }, nil + }, + } + + caughtUp, err := f.waitForCatchup(t.Context()) + require.NoError(t, err) + require.True(t, caughtUp) +} diff --git a/node/sequencer_recovery_integration_test.go b/node/sequencer_recovery_integration_test.go index 03e4972416..25ce55f101 100644 --- a/node/sequencer_recovery_integration_test.go +++ b/node/sequencer_recovery_integration_test.go @@ -3,7 +3,6 @@ package node import ( - "bytes" "context" "errors" "fmt" @@ -196,27 +195,7 @@ func TestSequencerRecoveryFromP2P(t *testing.T) { "recovery node should catch up via P2P and produce new blocks") requireEmptyChan(t, errChan) - // If the recovery node synced from P2P (got the original blocks), - // verify the hashes match. If P2P didn't connect in time and the - // node produced its own chain, we skip the hash assertion since - // the recovery still succeeded (just without P2P data). - recHeight, err := recoveryNode.Store.Height(t.Context()) - require.NoError(t, err) - if recHeight >= fnHeight { - allMatch := true - for h, expHash := range originalHashes { - header, _, err := recoveryNode.Store.GetBlockData(t.Context(), h) - if err != nil || !bytes.Equal(header.Hash(), expHash) { - allMatch = false - break - } - } - if allMatch { - t.Log("recovery node synced original blocks from P2P — all hashes verified") - } else { - t.Log("recovery node produced its own blocks (P2P sync was not completed in time)") - } - } + assertBlockHashesMatch(t, recoveryNode, originalHashes) // Shutdown recCancel() diff --git a/pkg/config/config.go b/pkg/config/config.go index 74aab704c3..f9b5ef1783 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -54,7 +54,7 @@ const ( FlagReadinessMaxBlocksBehind = FlagPrefixEvnode + "node.readiness_max_blocks_behind" // FlagScrapeInterval is a flag for specifying the reaper scrape interval FlagScrapeInterval = FlagPrefixEvnode + "node.scrape_interval" - // FlagCatchupTimeout is a flag for waiting for P2P catchup before starting block production + // FlagCatchupTimeout enables recovery catchup and limits configured P2P recovery. FlagCatchupTimeout = FlagPrefixEvnode + "node.catchup_timeout" // FlagClearCache is a flag for clearing the cache FlagClearCache = FlagPrefixEvnode + "clear_cache" @@ -314,7 +314,7 @@ type NodeConfig struct { LazyMode bool `mapstructure:"lazy_mode" yaml:"lazy_mode" comment:"Enables lazy aggregation mode, where blocks are only produced when transactions are available or after LazyBlockTime. Optimizes resources by avoiding empty block creation during periods of inactivity."` LazyBlockInterval DurationWrapper `mapstructure:"lazy_block_interval" yaml:"lazy_block_interval" comment:"Maximum interval between blocks in lazy aggregation mode (LazyAggregator). Ensures blocks are produced periodically even without transactions to keep the chain active. Generally larger than BlockTime."` ScrapeInterval DurationWrapper `mapstructure:"scrape_interval" yaml:"scrape_interval" comment:"Interval at which the reaper polls the execution layer for new transactions. Lower values reduce transaction detection latency but increase RPC load. Examples: \"250ms\", \"500ms\", \"1s\"."` - CatchupTimeout DurationWrapper `mapstructure:"catchup_timeout" yaml:"catchup_timeout" comment:"When set, the aggregator syncs from DA and P2P before producing blocks. Value specifies time to wait for P2P catchup. Requires aggregator mode."` + CatchupTimeout DurationWrapper `mapstructure:"catchup_timeout" yaml:"catchup_timeout" comment:"When set, the aggregator recovers before producing blocks. With configured P2P peers, failure to establish continuity before this timeout is fatal. Requires aggregator mode."` // Readiness / health configuration ReadinessWindowSeconds uint64 `mapstructure:"readiness_window_seconds" yaml:"readiness_window_seconds" comment:"Time window in seconds used to calculate ReadinessMaxBlocksBehind based on block time. Default: 15 seconds."` @@ -628,7 +628,7 @@ func AddFlags(cmd *cobra.Command) { cmd.Flags().Uint64(FlagReadinessWindowSeconds, def.Node.ReadinessWindowSeconds, "time window in seconds for calculating readiness threshold based on block time (default: 15s)") cmd.Flags().Uint64(FlagReadinessMaxBlocksBehind, def.Node.ReadinessMaxBlocksBehind, "how many blocks behind best-known head the node can be and still be considered ready (0 = must be at head)") cmd.Flags().Duration(FlagScrapeInterval, def.Node.ScrapeInterval.Duration, "interval at which the reaper polls the execution layer for new transactions") - cmd.Flags().Duration(FlagCatchupTimeout, def.Node.CatchupTimeout.Duration, "sync from DA and P2P before producing blocks. Value specifies time to wait for P2P catchup. Requires aggregator mode.") + cmd.Flags().Duration(FlagCatchupTimeout, def.Node.CatchupTimeout.Duration, "recover before producing blocks; configured P2P recovery is fatal if continuity is not established before this timeout. Requires aggregator mode.") // Data Availability configuration flags cmd.Flags().String(FlagDAAddress, def.DA.Address, "DA address (host:port)") diff --git a/pkg/sync/sync_service.go b/pkg/sync/sync_service.go index e469c47430..532b8194e3 100644 --- a/pkg/sync/sync_service.go +++ b/pkg/sync/sync_service.go @@ -67,6 +67,7 @@ type SyncService[H store.EntityWithDAHint[H]] struct { topicSubscription header.Subscription[H] storeInitialized atomic.Bool + p2pInitialized atomic.Bool } // NewDataSyncService returns a new DataSyncService. @@ -123,6 +124,13 @@ func (syncService *SyncService[H]) Store() header.Store[H] { return syncService.store } +// P2PInitialized reports whether the service successfully initialized its +// store and syncer from a P2P peer during startup. Store initialization through +// DA retrieval or block publishing does not satisfy this condition. +func (syncService *SyncService[H]) P2PInitialized() bool { + return syncService.p2pInitialized.Load() +} + // WriteToStoreAndBroadcast broadcasts provided header or block to P2P network. func (syncService *SyncService[H]) WriteToStoreAndBroadcast(ctx context.Context, headerOrData H, opts ...pubsub.PubOpt) error { if syncService.genesis.InitialHeight == 0 { @@ -401,12 +409,13 @@ func (syncService *SyncService[H]) initFromP2PWithRetry(ctx context.Context, pee if _, err := syncService.startSyncer(ctx); err != nil { return false, err } + syncService.p2pInitialized.Store(true) return true, nil } - // block with exponential backoff until initialization succeeds, context is canceled, or timeout. - // If timeout is reached, we return nil to allow startup to continue - DA sync will - // provide headers and WriteToStoreAndBroadcast will lazily initialize the store/syncer. + // Block with exponential backoff until initialization succeeds, the context is + // canceled, or the service timeout expires. The caller decides whether P2P + // initialization is mandatory before proceeding. backoff := 1 * time.Second maxBackoff := 10 * time.Second @@ -430,7 +439,7 @@ func (syncService *SyncService[H]) initFromP2PWithRetry(ctx context.Context, pee case <-timeoutTimer.C: syncService.logger.Warn(). Dur("timeout", p2pInitTimeout). - Msg("P2P header sync initialization timed out, deferring to DA sync") + Msg("P2P header sync initialization timed out") return nil case <-retryTimer.C: } @@ -516,7 +525,7 @@ func (syncService *SyncService[H]) getNetworkID(network string) string { func (syncService *SyncService[H]) getPeerIDs() []peer.ID { peerIDs := syncService.p2p.PeerIDs() - if !syncService.conf.Node.Aggregator { + if !syncService.conf.Node.Aggregator || syncService.conf.Node.CatchupTimeout.Duration > 0 { peerIDs = append(peerIDs, getPeers(syncService.conf.P2P.Peers, syncService.logger)...) } return peerIDs diff --git a/pkg/sync/sync_service_test.go b/pkg/sync/sync_service_test.go index a7d87584d5..4de4071418 100644 --- a/pkg/sync/sync_service_test.go +++ b/pkg/sync/sync_service_test.go @@ -15,7 +15,11 @@ import ( goheadersync "github.com/celestiaorg/go-header/sync" "github.com/ipfs/go-datastore" "github.com/ipfs/go-datastore/sync" + pubsub "github.com/libp2p/go-libp2p-pubsub" "github.com/libp2p/go-libp2p/core/crypto" + "github.com/libp2p/go-libp2p/core/host" + "github.com/libp2p/go-libp2p/core/peer" + "github.com/libp2p/go-libp2p/p2p/net/conngater" mocknet "github.com/libp2p/go-libp2p/p2p/net/mock" "github.com/multiformats/go-multiaddr" "github.com/rs/zerolog" @@ -78,6 +82,35 @@ type verifierCapturingP2PDataSubscriber struct { verifier func(context.Context, *types.P2PData) error } +type peerListP2PClient struct { + peerIDs []peer.ID +} + +func (c peerListP2PClient) PubSub() *pubsub.PubSub { return nil } +func (c peerListP2PClient) Info() (string, string, string, error) { return "", "", "", nil } +func (c peerListP2PClient) Host() host.Host { return nil } +func (c peerListP2PClient) ConnectionGater() *conngater.BasicConnectionGater { return nil } +func (c peerListP2PClient) PeerIDs() []peer.ID { return c.peerIDs } + +func TestCatchupAggregatorIncludesConfiguredPeerIDs(t *testing.T) { + configuredKey, _, err := crypto.GenerateEd25519Key(cryptoRand.Reader) + require.NoError(t, err) + configuredID, err := peer.IDFromPrivateKey(configuredKey) + require.NoError(t, err) + + conf := config.DefaultConfig() + conf.Node.Aggregator = true + conf.Node.CatchupTimeout = config.DurationWrapper{Duration: time.Second} + conf.P2P.Peers = "/ip4/127.0.0.1/tcp/7676/p2p/" + configuredID.String() + + svc := &SyncService[*types.P2PData]{ + conf: conf, + p2p: peerListP2PClient{}, + logger: zerolog.Nop(), + } + require.Equal(t, []peer.ID{configuredID}, svc.getPeerIDs()) +} + func (s *verifierCapturingP2PDataSubscriber) SetVerifier( verifier func(context.Context, *types.P2PData) error, ) error { @@ -211,6 +244,7 @@ func TestHeaderSyncServiceStartForPublishingWithPeers(t *testing.T) { require.NoError(t, svc.WriteToStoreAndBroadcast(ctx, &types.P2PSignedHeader{SignedHeader: signedHeader})) require.True(t, svc.storeInitialized.Load()) + require.False(t, svc.P2PInitialized(), "publishing must not count as P2P initialization") } func TestHeaderSyncServiceRestart(t *testing.T) { From ee13756dc59ba605708e186b597ecfc3de8e4f20 Mon Sep 17 00:00:00 2001 From: tac0turtle Date: Tue, 1 Sep 2026 14:55:08 +0200 Subject: [PATCH 2/2] fix(sync): retry P2P init throughout catchup recovery Do not abandon P2P initialization after the 30s Start timeout when catchup recovery requires continuity. Keep retrying in the background so P2PInitialized can still flip during waitForCatchup, and include readiness flags in the timeout error. --- node/failover.go | 14 ++-- node/failover_test.go | 22 ++++++ pkg/sync/sync_service.go | 125 +++++++++++++++++++++++----------- pkg/sync/sync_service_test.go | 63 +++++++++++++++++ 4 files changed, 179 insertions(+), 45 deletions(-) diff --git a/node/failover.go b/node/failover.go index f226669dcf..594501b3d3 100644 --- a/node/failover.go +++ b/node/failover.go @@ -44,7 +44,6 @@ type failoverState struct { daBlockTime time.Duration store store.Store - // catchupStatusFn is overridden by focused readiness tests. catchupStatusFn func(context.Context) (catchupStatus, error) } @@ -346,6 +345,14 @@ func (s catchupStatus) p2pReady() bool { s.storeHeight >= max(s.headerHeight, s.dataHeight) } +func (s catchupStatus) continuityTimeoutError(timeout time.Duration) error { + return fmt.Errorf( + "P2P recovery timed out after %s before continuity was established (store height %d, header height %d, data height %d, header P2P ready %t, data P2P ready %t, DA caught up %t, pending events %d)", + timeout, s.storeHeight, s.headerHeight, s.dataHeight, + s.headerP2PReady, s.dataP2PReady, s.daCaughtUp, s.pendingEvents, + ) +} + func (f *failoverState) catchupStatus(ctx context.Context) (catchupStatus, error) { if f.catchupStatusFn != nil { return f.catchupStatusFn(ctx) @@ -401,10 +408,7 @@ func (f *failoverState) waitForCatchup(ctx context.Context) (bool, error) { timeoutCh = nil continue } - return false, fmt.Errorf( - "P2P recovery timed out after %s before continuity was established (store height %d, header height %d, data height %d)", - f.catchupTimeout, status.storeHeight, status.headerHeight, status.dataHeight, - ) + return false, status.continuityTimeoutError(f.catchupTimeout) case <-ticker.C: status, err := f.catchupStatus(ctx) if err != nil { diff --git a/node/failover_test.go b/node/failover_test.go index 8c0aafbe0a..8377c93c92 100644 --- a/node/failover_test.go +++ b/node/failover_test.go @@ -127,6 +127,28 @@ func TestWaitForCatchupP2PTimeoutFailsClosed(t *testing.T) { require.ErrorContains(t, err, "store height 7") require.ErrorContains(t, err, "header height 9") require.ErrorContains(t, err, "data height 8") + require.ErrorContains(t, err, "header P2P ready false") + require.ErrorContains(t, err, "data P2P ready false") + require.ErrorContains(t, err, "DA caught up false") + require.ErrorContains(t, err, "pending events 0") +} + +func TestCatchupStatusContinuityTimeoutErrorIncludesReadiness(t *testing.T) { + err := catchupStatus{ + storeHeight: 1, + headerHeight: 2, + dataHeight: 3, + headerP2PReady: true, + daCaughtUp: true, + pendingEvents: 4, + }.continuityTimeoutError(time.Second) + require.ErrorContains(t, err, "store height 1") + require.ErrorContains(t, err, "header height 2") + require.ErrorContains(t, err, "data height 3") + require.ErrorContains(t, err, "header P2P ready true") + require.ErrorContains(t, err, "data P2P ready false") + require.ErrorContains(t, err, "DA caught up true") + require.ErrorContains(t, err, "pending events 4") } func TestWaitForCatchupContextCancellation(t *testing.T) { diff --git a/pkg/sync/sync_service.go b/pkg/sync/sync_service.go index 532b8194e3..622f9ac8fc 100644 --- a/pkg/sync/sync_service.go +++ b/pkg/sync/sync_service.go @@ -29,6 +29,10 @@ type syncType string const ( headerSync syncType = "headerSync" dataSync syncType = "dataSync" + + // p2pInitTimeout is how long non-recovery startup waits for a peer to serve + // genesis before deferring store initialization to DA or the first produced block. + p2pInitTimeout = 30 * time.Second ) // HeaderSyncService is the P2P Sync Service for headers. @@ -214,6 +218,7 @@ func (syncService *SyncService[H]) Start(ctx context.Context) error { return fmt.Errorf("failed to start subscriber: %w", err) } + syncService.continueP2PInitIfNeeded(ctx, peerIDs) return nil } @@ -370,6 +375,28 @@ func (s *SyncService[H]) Height() uint64 { return s.store.Height() } +// keepRetryingP2PInit is true when catchup recovery requires P2P continuity, +// so Start must not abandon initialization after p2pInitTimeout. +func (syncService *SyncService[H]) keepRetryingP2PInit() bool { + return syncService.conf.Node.Aggregator && + syncService.conf.Node.CatchupTimeout.Duration > 0 && + strings.TrimSpace(syncService.conf.P2P.Peers) != "" +} + +// continueP2PInitIfNeeded keeps trying P2P initialization after the subscriber is +// up so P2PInitialized can still become true during catchup. +func (syncService *SyncService[H]) continueP2PInitIfNeeded(ctx context.Context, peerIDs []peer.ID) { + if !syncService.keepRetryingP2PInit() || syncService.P2PInitialized() || len(peerIDs) == 0 { + return + } + + go func() { + if err := syncService.retryInitFromP2P(ctx, 0); err != nil && ctx.Err() == nil { + syncService.logger.Warn().Err(err).Msg("background P2P initialization stopped") + } + }() +} + // initFromP2PWithRetry initializes the syncer from P2P with a retry mechanism. // It inspects the local store to determine the first height to request: // - when the store already contains items, it reuses the latest height as the starting point; @@ -379,66 +406,85 @@ func (syncService *SyncService[H]) initFromP2PWithRetry(ctx context.Context, pee return nil } - tryInit := func(ctx context.Context) (bool, error) { - var ( - trusted H - err error - heightToQuery uint64 - ) - - head, headErr := syncService.store.Head(ctx) - switch { - case errors.Is(headErr, header.ErrNotFound), errors.Is(headErr, header.ErrEmptyStore): - heightToQuery = syncService.genesis.InitialHeight - case headErr != nil: - return false, fmt.Errorf("failed to inspect local store head: %w", headErr) - default: - heightToQuery = head.Height() + if syncService.keepRetryingP2PInit() { + ok, err := syncService.tryInitFromP2P(ctx) + if ok { + return nil } - - if trusted, err = syncService.ex.GetByHeight(ctx, heightToQuery); err != nil { - return false, fmt.Errorf("failed to fetch height %d from peers: %w", heightToQuery, err) + if ctx.Err() != nil { + return ctx.Err() } + syncService.logger.Info().Err(err).Msg("P2P initialization pending; continuing in background until context is canceled") + return nil + } - if syncService.storeInitialized.CompareAndSwap(false, true) { - if _, err := syncService.initStore(ctx, trusted); err != nil { - syncService.storeInitialized.Store(false) - return false, fmt.Errorf("failed to initialize the store: %w", err) - } - } - if _, err := syncService.startSyncer(ctx); err != nil { - return false, err + return syncService.retryInitFromP2P(ctx, p2pInitTimeout) +} + +func (syncService *SyncService[H]) tryInitFromP2P(ctx context.Context) (bool, error) { + var ( + trusted H + err error + heightToQuery uint64 + ) + + head, headErr := syncService.store.Head(ctx) + switch { + case errors.Is(headErr, header.ErrNotFound), errors.Is(headErr, header.ErrEmptyStore): + heightToQuery = syncService.genesis.InitialHeight + case headErr != nil: + return false, fmt.Errorf("failed to inspect local store head: %w", headErr) + default: + heightToQuery = head.Height() + } + + if trusted, err = syncService.ex.GetByHeight(ctx, heightToQuery); err != nil { + return false, fmt.Errorf("failed to fetch height %d from peers: %w", heightToQuery, err) + } + + if syncService.storeInitialized.CompareAndSwap(false, true) { + if _, err := syncService.initStore(ctx, trusted); err != nil { + syncService.storeInitialized.Store(false) + return false, fmt.Errorf("failed to initialize the store: %w", err) } - syncService.p2pInitialized.Store(true) - return true, nil } + if _, err := syncService.startSyncer(ctx); err != nil { + return false, err + } + syncService.p2pInitialized.Store(true) + return true, nil +} - // Block with exponential backoff until initialization succeeds, the context is - // canceled, or the service timeout expires. The caller decides whether P2P - // initialization is mandatory before proceeding. +// retryInitFromP2P retries tryInitFromP2P with exponential backoff. A zero +// giveUpAfter retries until ctx is canceled so catchup can still observe P2PInitialized. +func (syncService *SyncService[H]) retryInitFromP2P(ctx context.Context, giveUpAfter time.Duration) error { backoff := 1 * time.Second maxBackoff := 10 * time.Second - p2pInitTimeout := 30 * time.Second - timeoutTimer := time.NewTimer(p2pInitTimeout) - defer timeoutTimer.Stop() - retryTimer := time.NewTimer(backoff) - defer retryTimer.Stop() + var timeoutCh <-chan time.Time + if giveUpAfter > 0 { + timeoutTimer := time.NewTimer(giveUpAfter) + defer timeoutTimer.Stop() + timeoutCh = timeoutTimer.C + } for { - ok, err := tryInit(ctx) + ok, err := syncService.tryInitFromP2P(ctx) if ok { return nil } syncService.logger.Info().Err(err).Dur("retry_in", backoff).Msg("headers not yet available from peers, waiting to initialize header sync") + retryTimer := time.NewTimer(backoff) select { case <-ctx.Done(): + retryTimer.Stop() return ctx.Err() - case <-timeoutTimer.C: + case <-timeoutCh: + retryTimer.Stop() syncService.logger.Warn(). - Dur("timeout", p2pInitTimeout). + Dur("timeout", giveUpAfter). Msg("P2P header sync initialization timed out") return nil case <-retryTimer.C: @@ -447,7 +493,6 @@ func (syncService *SyncService[H]) initFromP2PWithRetry(ctx context.Context, pee if backoff > maxBackoff { backoff = maxBackoff } - retryTimer.Reset(backoff) } } diff --git a/pkg/sync/sync_service_test.go b/pkg/sync/sync_service_test.go index 4de4071418..08e54bd453 100644 --- a/pkg/sync/sync_service_test.go +++ b/pkg/sync/sync_service_test.go @@ -111,6 +111,69 @@ func TestCatchupAggregatorIncludesConfiguredPeerIDs(t *testing.T) { require.Equal(t, []peer.ID{configuredID}, svc.getPeerIDs()) } +func TestKeepRetryingP2PInit(t *testing.T) { + configuredKey, _, err := crypto.GenerateEd25519Key(cryptoRand.Reader) + require.NoError(t, err) + configuredID, err := peer.IDFromPrivateKey(configuredKey) + require.NoError(t, err) + peerAddr := "/ip4/127.0.0.1/tcp/7676/p2p/" + configuredID.String() + + tests := []struct { + name string + mutate func(*config.Config) + want bool + }{ + { + name: "catchup aggregator with peers", + mutate: func(conf *config.Config) { + conf.Node.Aggregator = true + conf.Node.CatchupTimeout = config.DurationWrapper{Duration: time.Minute} + conf.P2P.Peers = peerAddr + }, + want: true, + }, + { + name: "catchup disabled", + mutate: func(conf *config.Config) { + conf.Node.Aggregator = true + conf.P2P.Peers = peerAddr + }, + }, + { + name: "no configured peers", + mutate: func(conf *config.Config) { + conf.Node.Aggregator = true + conf.Node.CatchupTimeout = config.DurationWrapper{Duration: time.Minute} + }, + }, + { + name: "full node with catchup config", + mutate: func(conf *config.Config) { + conf.Node.CatchupTimeout = config.DurationWrapper{Duration: time.Minute} + conf.P2P.Peers = peerAddr + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + conf := config.DefaultConfig() + tt.mutate(&conf) + svc := &SyncService[*types.P2PData]{conf: conf} + require.Equal(t, tt.want, svc.keepRetryingP2PInit()) + }) + } +} + +func TestContinueP2PInitIfNeededNoopsWithoutCatchupRequirement(t *testing.T) { + svc := &SyncService[*types.P2PData]{ + conf: config.DefaultConfig(), + logger: zerolog.Nop(), + } + // Would panic in retryInitFromP2P if a goroutine were started with a nil store. + svc.continueP2PInitIfNeeded(t.Context(), []peer.ID{"12D3KooWCatchupPeer"}) +} + func (s *verifierCapturingP2PDataSubscriber) SetVerifier( verifier func(context.Context, *types.P2PData) error, ) error {