From 18d7927dd2d6f9fc1bd9b5d722d726380290d290 Mon Sep 17 00:00:00 2001 From: Adrian Dobrita Date: Tue, 9 Jun 2026 13:23:26 +0300 Subject: [PATCH 1/4] port request SOE meta watchdog --- epochStart/shardchain/trigger.go | 102 +++++- epochStart/shardchain/triggerRegistry_test.go | 2 + epochStart/shardchain/trigger_test.go | 332 ++++++++++++++++++ process/interface.go | 1 + 4 files changed, 430 insertions(+), 7 deletions(-) diff --git a/epochStart/shardchain/trigger.go b/epochStart/shardchain/trigger.go index ebab0414fab..757eee09db2 100644 --- a/epochStart/shardchain/trigger.go +++ b/epochStart/shardchain/trigger.go @@ -19,12 +19,13 @@ import ( "github.com/multiversx/mx-chain-core-go/display" "github.com/multiversx/mx-chain-core-go/hashing" "github.com/multiversx/mx-chain-core-go/marshal" + logger "github.com/multiversx/mx-chain-logger-go" + "github.com/multiversx/mx-chain-go/common" "github.com/multiversx/mx-chain-go/dataRetriever" "github.com/multiversx/mx-chain-go/epochStart" "github.com/multiversx/mx-chain-go/process" "github.com/multiversx/mx-chain-go/storage" - logger "github.com/multiversx/mx-chain-logger-go" ) var log = logger.GetOrCreate("epochStart/shardchain") @@ -38,6 +39,8 @@ var _ closing.Closer = (*trigger)(nil) // sleepTime defines the time in milliseconds between each iteration made in requestMissingMiniBlocks method const sleepTime = 1 * time.Second +const numRoundsWithoutReceivedMetaBlocks = 5 + // ArgsShardEpochStartTrigger struct { defines the arguments needed for new start of epoch trigger type ArgsShardEpochStartTrigger struct { Marshalizer marshal.Marshalizer @@ -72,11 +75,12 @@ type trigger struct { epochStartShardHeader data.HeaderHandler epochStartMeta data.HeaderHandler - mutTrigger sync.RWMutex - mapHashHdr map[string]data.HeaderHandler - mapNonceHashes map[uint64][]string - mapEpochStartHdrs map[string]data.HeaderHandler - mapFinalizedEpochs map[uint32]string + mutTrigger sync.RWMutex + mapHashHdr map[string]data.HeaderHandler + mapNonceHashes map[uint64][]string + mapEpochStartHdrs map[string]data.HeaderHandler + mapFinalizedEpochs map[uint32]string + mapPreparedEpochStartHdrs map[string]struct{} headersPool dataRetriever.HeadersPool proofsPool dataRetriever.ProofsPool @@ -115,6 +119,8 @@ type trigger struct { mutMissingValidatorsInfo sync.RWMutex cancelFunc func() + chanMetaBlockReceived chan struct{} + extraDelayForRequestBlockInfo time.Duration } @@ -273,6 +279,7 @@ func NewEpochStartTrigger(args *ArgsShardEpochStartTrigger) (*trigger, error) { roundHandler: args.RoundHandler, enableEpochsHandler: args.EnableEpochsHandler, extraDelayForRequestBlockInfo: args.ExtraDelayForRequestBlockInfo, + chanMetaBlockReceived: make(chan struct{}, 1), } t.headersPool.RegisterHandler(t.receivedMetaBlock) @@ -285,11 +292,13 @@ func NewEpochStartTrigger(args *ArgsShardEpochStartTrigger) (*trigger, error) { t.mapMissingMiniBlocks = make(map[string]uint32) t.mapMissingValidatorsInfo = make(map[string]uint32) + t.mapPreparedEpochStartHdrs = make(map[string]struct{}) var ctx context.Context ctx, t.cancelFunc = context.WithCancel(context.Background()) go t.requestMissingMiniBlocks(ctx) go t.requestMissingValidatorsInfo(ctx) + go t.watchdogRequestEpochStartMetaBlock(ctx) return t, nil } @@ -333,6 +342,16 @@ func (t *trigger) requestMissingMiniBlocks(ctx context.Context) { t.mutMissingMiniBlocks.RLock() if len(t.mapMissingMiniBlocks) == 0 { t.mutMissingMiniBlocks.RUnlock() + + t.mutTrigger.Lock() + if t.isEpochStart { + t.mutTrigger.Unlock() + continue + } + + t.updateTriggerFromMeta() + t.mutTrigger.Unlock() + continue } @@ -588,6 +607,11 @@ func (t *trigger) receivedMetaBlock(headerHandler data.HeaderHandler, metaBlockH return } + select { + case t.chanMetaBlockReceived <- struct{}{}: + default: + } + log.Debug("received meta header in trigger", "header hash", metaBlockHash) if t.enableEpochsHandler.IsFlagEnabledInEpoch(common.AndromedaFlag, headerHandler.GetEpoch()) { proof, err := t.proofsPool.GetProof(headerHandler.GetShardID(), metaBlockHash) @@ -880,7 +904,10 @@ func (t *trigger) checkIfTriggerCanBeActivated(hash string, metaHdr data.HeaderH } } - t.epochStartNotifier.NotifyAllPrepare(metaHdr, blockBody) + if _, alreadyPrepared := t.mapPreparedEpochStartHdrs[hash]; !alreadyPrepared { + t.epochStartNotifier.NotifyAllPrepare(metaHdr, blockBody) + t.mapPreparedEpochStartHdrs[hash] = struct{}{} + } isMetaHdrFinal, finalityAttestingRound := t.isMetaBlockFinal(hash, metaHdr) return isMetaHdrFinal, finalityAttestingRound @@ -1095,6 +1122,7 @@ func (t *trigger) SetProcessed(header data.HeaderHandler, _ data.BodyHandler) { t.mapNonceHashes = make(map[uint64][]string) t.mapEpochStartHdrs = make(map[string]data.HeaderHandler) t.mapFinalizedEpochs = make(map[uint32]string) + t.mapPreparedEpochStartHdrs = make(map[string]struct{}) t.saveCurrentState(header.GetRound()) @@ -1253,6 +1281,66 @@ func (t *trigger) saveCurrentState(round uint64) { } } +func (t *trigger) computeWatchdogTimeout() time.Duration { + timeout := t.roundHandler.TimeDuration() * numRoundsWithoutReceivedMetaBlocks + if timeout <= 0 { + return 0 + } + return timeout +} + +func (t *trigger) watchdogRequestEpochStartMetaBlock(ctx context.Context) { + watchdogTimeout := t.computeWatchdogTimeout() + if watchdogTimeout == 0 { + return + } + + timer := time.NewTimer(watchdogTimeout) + defer timer.Stop() + + for { + select { + case <-ctx.Done(): + log.Debug("watchdogRequestEpochStartMetaBlock: trigger's go routine is stopping...") + return + case <-t.chanMetaBlockReceived: + timer.Reset(t.resetWatchdogTimeout(watchdogTimeout)) + case <-timer.C: + t.handleWatchdogTimeout() + timer.Reset(t.resetWatchdogTimeout(watchdogTimeout)) + } + } +} + +func (t *trigger) resetWatchdogTimeout(fallback time.Duration) time.Duration { + timeout := t.computeWatchdogTimeout() + if timeout == 0 { + return fallback + } + return timeout +} + +func (t *trigger) handleWatchdogTimeout() { + t.mutTrigger.RLock() + epoch := t.epoch + isEpochStart := t.isEpochStart + t.mutTrigger.RUnlock() + + if isEpochStart { + return + } + + if !t.enableEpochsHandler.IsFlagEnabledInEpoch(common.AndromedaFlag, epoch) { + return + } + + log.Debug("watchdog: no metablock received for too long, requesting epoch start metablock", + "current epoch", epoch, + "requesting epoch", epoch+1, + ) + go t.requestHandler.RequestStartOfEpochMetaBlock(epoch + 1) +} + // Close will close the endless running go routine func (t *trigger) Close() error { if t.cancelFunc != nil { diff --git a/epochStart/shardchain/triggerRegistry_test.go b/epochStart/shardchain/triggerRegistry_test.go index 970f48f6a73..68a09f9c1bf 100644 --- a/epochStart/shardchain/triggerRegistry_test.go +++ b/epochStart/shardchain/triggerRegistry_test.go @@ -57,6 +57,8 @@ func cloneTrigger(t *trigger) *trigger { rt.mapFinalizedEpochs = t.mapFinalizedEpochs rt.roundHandler = t.roundHandler rt.enableEpochsHandler = t.enableEpochsHandler + rt.chanMetaBlockReceived = t.chanMetaBlockReceived + rt.mapPreparedEpochStartHdrs = t.mapPreparedEpochStartHdrs return rt } diff --git a/epochStart/shardchain/trigger_test.go b/epochStart/shardchain/trigger_test.go index 34dd10d97be..6d6ff836243 100644 --- a/epochStart/shardchain/trigger_test.go +++ b/epochStart/shardchain/trigger_test.go @@ -1134,3 +1134,335 @@ func TestTrigger_ReceivedProof(t *testing.T) { require.True(t, wasCalled) }) } + +func TestTrigger_WatchdogRequestEpochStartMetaBlock(t *testing.T) { + t.Parallel() + + t.Run("fires after timeout", func(t *testing.T) { + t.Parallel() + + var requestedEpoch atomic.Uint32 + var called atomic.Int32 + args := createMockShardEpochStartTriggerArguments() + args.RoundHandler = &mock.RoundHandlerStub{ + TimeDurationCalled: func() time.Duration { + return 10 * time.Millisecond + }, + } + args.Epoch = 5 + args.RequestHandler = &testscommon.RequestHandlerStub{ + RequestStartOfEpochMetaBlockCalled: func(epoch uint32) { + requestedEpoch.Store(epoch) + called.Add(1) + }, + } + args.EnableEpochsHandler = &enableEpochsHandlerMock.EnableEpochsHandlerStub{ + IsFlagEnabledInEpochCalled: func(flag core.EnableEpochFlag, epoch uint32) bool { + return flag == common.AndromedaFlag + }, + } + + et, err := NewEpochStartTrigger(args) + require.Nil(t, err) + defer func() { + _ = et.Close() + }() + + time.Sleep(200 * time.Millisecond) + + require.Greater(t, called.Load(), int32(0)) + require.Equal(t, uint32(6), requestedEpoch.Load()) + }) + + t.Run("resets timer on any metablock reception", func(t *testing.T) { + t.Parallel() + + var called atomic.Int32 + args := createMockShardEpochStartTriggerArguments() + args.RoundHandler = &mock.RoundHandlerStub{ + TimeDurationCalled: func() time.Duration { + return 30 * time.Millisecond + }, + } + args.RequestHandler = &testscommon.RequestHandlerStub{ + RequestStartOfEpochMetaBlockCalled: func(epoch uint32) { + called.Add(1) + }, + } + args.EnableEpochsHandler = &enableEpochsHandlerMock.EnableEpochsHandlerStub{ + IsFlagEnabledInEpochCalled: func(flag core.EnableEpochFlag, epoch uint32) bool { + return flag == common.AndromedaFlag + }, + } + + et, err := NewEpochStartTrigger(args) + require.Nil(t, err) + defer func() { + _ = et.Close() + }() + + for i := 0; i < 10; i++ { + select { + case et.chanMetaBlockReceived <- struct{}{}: + default: + } + time.Sleep(20 * time.Millisecond) + } + + require.Equal(t, int32(0), called.Load()) + }) + + t.Run("skips when epoch start already detected", func(t *testing.T) { + t.Parallel() + + var called atomic.Int32 + args := createMockShardEpochStartTriggerArguments() + args.RoundHandler = &mock.RoundHandlerStub{ + TimeDurationCalled: func() time.Duration { + return 10 * time.Millisecond + }, + IndexCalled: func() int64 { + return 100 + }, + } + args.RequestHandler = &testscommon.RequestHandlerStub{ + RequestStartOfEpochMetaBlockCalled: func(epoch uint32) { + called.Add(1) + }, + } + args.EnableEpochsHandler = &enableEpochsHandlerMock.EnableEpochsHandlerStub{ + IsFlagEnabledInEpochCalled: func(flag core.EnableEpochFlag, epoch uint32) bool { + return flag == common.AndromedaFlag + }, + } + + et, err := NewEpochStartTrigger(args) + require.Nil(t, err) + defer func() { + _ = et.Close() + }() + + et.mutTrigger.Lock() + et.isEpochStart = true + et.mutTrigger.Unlock() + + time.Sleep(200 * time.Millisecond) + + require.Equal(t, int32(0), called.Load()) + }) + + t.Run("skips when Andromeda disabled", func(t *testing.T) { + t.Parallel() + + var called atomic.Int32 + args := createMockShardEpochStartTriggerArguments() + args.RoundHandler = &mock.RoundHandlerStub{ + TimeDurationCalled: func() time.Duration { + return 10 * time.Millisecond + }, + IndexCalled: func() int64 { + return 100 + }, + } + args.RequestHandler = &testscommon.RequestHandlerStub{ + RequestStartOfEpochMetaBlockCalled: func(epoch uint32) { + called.Add(1) + }, + } + args.EnableEpochsHandler = &enableEpochsHandlerMock.EnableEpochsHandlerStub{ + IsFlagEnabledInEpochCalled: func(flag core.EnableEpochFlag, epoch uint32) bool { + return false + }, + } + + et, err := NewEpochStartTrigger(args) + require.Nil(t, err) + defer func() { + _ = et.Close() + }() + + time.Sleep(200 * time.Millisecond) + + require.Equal(t, int32(0), called.Load()) + }) + + t.Run("stops on context cancellation", func(t *testing.T) { + t.Parallel() + + var called atomic.Int32 + args := createMockShardEpochStartTriggerArguments() + args.RoundHandler = &mock.RoundHandlerStub{ + TimeDurationCalled: func() time.Duration { + return 10 * time.Millisecond + }, + IndexCalled: func() int64 { + return 100 + }, + } + args.RequestHandler = &testscommon.RequestHandlerStub{ + RequestStartOfEpochMetaBlockCalled: func(epoch uint32) { + called.Add(1) + }, + } + args.EnableEpochsHandler = &enableEpochsHandlerMock.EnableEpochsHandlerStub{ + IsFlagEnabledInEpochCalled: func(flag core.EnableEpochFlag, epoch uint32) bool { + return flag == common.AndromedaFlag + }, + } + + et, err := NewEpochStartTrigger(args) + require.Nil(t, err) + + err = et.Close() + require.Nil(t, err) + + calledBefore := called.Load() + time.Sleep(200 * time.Millisecond) + + require.Equal(t, calledBefore, called.Load()) + }) + + t.Run("does not start when TimeDuration is zero", func(t *testing.T) { + t.Parallel() + + var called atomic.Int32 + args := createMockShardEpochStartTriggerArguments() + args.RoundHandler = &mock.RoundHandlerStub{ + TimeDurationCalled: func() time.Duration { + return 0 + }, + } + args.RequestHandler = &testscommon.RequestHandlerStub{ + RequestStartOfEpochMetaBlockCalled: func(epoch uint32) { + called.Add(1) + }, + } + args.EnableEpochsHandler = &enableEpochsHandlerMock.EnableEpochsHandlerStub{ + IsFlagEnabledInEpochCalled: func(flag core.EnableEpochFlag, epoch uint32) bool { + return flag == common.AndromedaFlag + }, + } + + et, err := NewEpochStartTrigger(args) + require.Nil(t, err) + defer func() { + _ = et.Close() + }() + + time.Sleep(100 * time.Millisecond) + + require.Equal(t, int32(0), called.Load()) + }) + + t.Run("receivedMetaBlock signals watchdog", func(t *testing.T) { + t.Parallel() + + args := createMockShardEpochStartTriggerArguments() + // zero TimeDuration prevents the watchdog goroutine from starting + args.RoundHandler = &mock.RoundHandlerStub{ + TimeDurationCalled: func() time.Duration { + return 0 + }, + } + et, err := NewEpochStartTrigger(args) + require.Nil(t, err) + defer func() { + _ = et.Close() + }() + + et.receivedMetaBlock(&block.MetaBlock{ + Nonce: 10, + Round: 42, + }, []byte("hash")) + + select { + case <-et.chanMetaBlockReceived: + // expected + default: + require.Fail(t, "channel should have been signaled") + } + }) + + t.Run("receivedMetaBlock requests proof when missing in Andromeda", func(t *testing.T) { + t.Parallel() + + var proofRequested atomic.Int32 + var requestedHashMut sync.Mutex + var requestedHash []byte + args := createMockShardEpochStartTriggerArguments() + args.Epoch = 5 + args.EnableEpochsHandler = &enableEpochsHandlerMock.EnableEpochsHandlerStub{ + IsFlagEnabledInEpochCalled: func(flag core.EnableEpochFlag, epoch uint32) bool { + return flag == common.AndromedaFlag + }, + } + args.RequestHandler = &testscommon.RequestHandlerStub{ + RequestEquivalentProofByHashCalled: func(headerShard uint32, headerHash []byte) { + requestedHashMut.Lock() + requestedHash = headerHash + requestedHashMut.Unlock() + proofRequested.Add(1) + }, + } + args.DataPool = &dataRetrieverMock.PoolsHolderStub{ + HeadersCalled: func() dataRetriever.HeadersPool { + return &mock.HeadersCacherStub{} + }, + MiniBlocksCalled: func() storage.Cacher { + return cache.NewCacherStub() + }, + CurrEpochValidatorInfoCalled: func() dataRetriever.ValidatorInfoCacher { + return &vic.ValidatorInfoCacherStub{} + }, + ProofsCalled: func() dataRetriever.ProofsPool { + return &dataRetrieverMock.ProofsPoolMock{ + GetProofCalled: func(_ uint32, _ []byte) (data.HeaderProofHandler, error) { + return nil, errors.New("proof not found") + }, + } + }, + } + + et, err := NewEpochStartTrigger(args) + require.Nil(t, err) + defer func() { + _ = et.Close() + }() + + metaBlockHash := []byte("metablock-hash") + et.receivedMetaBlock(&block.MetaBlock{ + Nonce: 10, + Round: 42, + Epoch: 6, + EpochStart: block.EpochStart{LastFinalizedHeaders: []block.EpochStartShardData{{}}}, + }, metaBlockHash) + + time.Sleep(50 * time.Millisecond) + + require.Equal(t, int32(1), proofRequested.Load()) + requestedHashMut.Lock() + require.Equal(t, metaBlockHash, requestedHash) + requestedHashMut.Unlock() + + proofRequested.Store(0) + et.receivedMetaBlock(&block.MetaBlock{ + Nonce: 11, + Round: 43, + Epoch: 6, + }, []byte("regular-metablock-hash")) + + time.Sleep(50 * time.Millisecond) + require.Equal(t, int32(0), proofRequested.Load()) + + proofRequested.Store(0) + et.receivedMetaBlock(&block.MetaBlock{ + Nonce: 5, + Round: 30, + Epoch: 5, + EpochStart: block.EpochStart{LastFinalizedHeaders: []block.EpochStartShardData{{}}}, + }, []byte("old-epoch-start-hash")) + + time.Sleep(50 * time.Millisecond) + require.Equal(t, int32(0), proofRequested.Load()) + }) +} diff --git a/process/interface.go b/process/interface.go index e394f4c0862..ff5f924ae1a 100644 --- a/process/interface.go +++ b/process/interface.go @@ -1072,6 +1072,7 @@ type RoundTimeDurationHandler interface { // RoundHandler defines the actions which should be handled by a round implementation type RoundHandler interface { Index() int64 + TimeDuration() time.Duration IsInterfaceNil() bool } From 55c16e01a1e1d43c8df958054e1d99b803e2e441 Mon Sep 17 00:00:00 2001 From: Adrian Dobrita Date: Tue, 9 Jun 2026 13:30:14 +0300 Subject: [PATCH 2/4] remove Andromeda request gate --- epochStart/shardchain/trigger.go | 4 ---- epochStart/shardchain/trigger_test.go | 8 ++++++-- 2 files changed, 6 insertions(+), 6 deletions(-) diff --git a/epochStart/shardchain/trigger.go b/epochStart/shardchain/trigger.go index 757eee09db2..c588a70068a 100644 --- a/epochStart/shardchain/trigger.go +++ b/epochStart/shardchain/trigger.go @@ -1330,10 +1330,6 @@ func (t *trigger) handleWatchdogTimeout() { return } - if !t.enableEpochsHandler.IsFlagEnabledInEpoch(common.AndromedaFlag, epoch) { - return - } - log.Debug("watchdog: no metablock received for too long, requesting epoch start metablock", "current epoch", epoch, "requesting epoch", epoch+1, diff --git a/epochStart/shardchain/trigger_test.go b/epochStart/shardchain/trigger_test.go index 6d6ff836243..c277d53452d 100644 --- a/epochStart/shardchain/trigger_test.go +++ b/epochStart/shardchain/trigger_test.go @@ -1251,9 +1251,10 @@ func TestTrigger_WatchdogRequestEpochStartMetaBlock(t *testing.T) { require.Equal(t, int32(0), called.Load()) }) - t.Run("skips when Andromeda disabled", func(t *testing.T) { + t.Run("fires even when Andromeda disabled", func(t *testing.T) { t.Parallel() + var requestedEpoch atomic.Uint32 var called atomic.Int32 args := createMockShardEpochStartTriggerArguments() args.RoundHandler = &mock.RoundHandlerStub{ @@ -1264,8 +1265,10 @@ func TestTrigger_WatchdogRequestEpochStartMetaBlock(t *testing.T) { return 100 }, } + args.Epoch = 5 args.RequestHandler = &testscommon.RequestHandlerStub{ RequestStartOfEpochMetaBlockCalled: func(epoch uint32) { + requestedEpoch.Store(epoch) called.Add(1) }, } @@ -1283,7 +1286,8 @@ func TestTrigger_WatchdogRequestEpochStartMetaBlock(t *testing.T) { time.Sleep(200 * time.Millisecond) - require.Equal(t, int32(0), called.Load()) + require.Greater(t, called.Load(), int32(0)) + require.Equal(t, uint32(6), requestedEpoch.Load()) }) t.Run("stops on context cancellation", func(t *testing.T) { From 682e1ff1a867a7ad0e127792854f0a1a0c61f662 Mon Sep 17 00:00:00 2001 From: Adrian Dobrita Date: Tue, 9 Jun 2026 16:22:44 +0300 Subject: [PATCH 3/4] add meta sync watchdog for epoch start block --- factory/consensus/consensusComponents.go | 1 + integrationTests/testSyncNode.go | 1 + process/errors.go | 3 + process/sync/argBootstrapper.go | 2 + process/sync/export_test.go | 11 ++ process/sync/metablock.go | 99 ++++++++++++++ process/sync/metablock_test.go | 165 +++++++++++++++++++++++ 7 files changed, 282 insertions(+) diff --git a/factory/consensus/consensusComponents.go b/factory/consensus/consensusComponents.go index d7db14cccff..ed810e324ce 100644 --- a/factory/consensus/consensusComponents.go +++ b/factory/consensus/consensusComponents.go @@ -657,6 +657,7 @@ func (ccf *consensusComponentsFactory) createMetaChainBootstrapper() (process.Bo EpochBootstrapper: ccf.processComponents.EpochStartTrigger(), ValidatorAccountsDB: ccf.stateComponents.PeerAccounts(), ValidatorStatisticsDBSyncer: validatorAccountsDBSyncer, + Watchdog: ccf.coreComponents.Watchdog(), } return sync.NewMetaBootstrap(argsMetaBootstrapper) diff --git a/integrationTests/testSyncNode.go b/integrationTests/testSyncNode.go index 5f1a212892d..d7e3bb34665 100644 --- a/integrationTests/testSyncNode.go +++ b/integrationTests/testSyncNode.go @@ -246,6 +246,7 @@ func (tpn *TestProcessorNode) createMetaChainBootstrapper() (TestBootstrapper, e EpochBootstrapper: tpn.EpochStartTrigger, ValidatorAccountsDB: tpn.PeerState, ValidatorStatisticsDBSyncer: &mock.AccountsDBSyncerStub{}, + Watchdog: &testscommon.WatchdogMock{}, } bootstrap, err := sync.NewMetaBootstrap(argsMetaBootstrapper) diff --git a/process/errors.go b/process/errors.go index 3612db8f68c..bf627ddc706 100644 --- a/process/errors.go +++ b/process/errors.go @@ -1191,6 +1191,9 @@ var ErrNilESDTGlobalSettingsHandler = errors.New("nil esdt global settings handl // ErrNilEnableEpochsHandler signals that a nil enable epochs handler has been provided var ErrNilEnableEpochsHandler = errors.New("nil enable epochs handler") +// ErrNilWatchdog signals that a nil watchdog has been provided +var ErrNilWatchdog = errors.New("nil watchdog") + // ErrNilEpochChangeGracePeriodHandler signals that a nil epoch change grace period handler has been provided var ErrNilEpochChangeGracePeriodHandler = errors.New("nil epoch change grace period handler") diff --git a/process/sync/argBootstrapper.go b/process/sync/argBootstrapper.go index 587ecedd258..c01760c07f3 100644 --- a/process/sync/argBootstrapper.go +++ b/process/sync/argBootstrapper.go @@ -8,6 +8,7 @@ import ( "github.com/multiversx/mx-chain-core-go/data/typeConverters" "github.com/multiversx/mx-chain-core-go/hashing" "github.com/multiversx/mx-chain-core-go/marshal" + "github.com/multiversx/mx-chain-go/common" "github.com/multiversx/mx-chain-go/consensus" "github.com/multiversx/mx-chain-go/dataRetriever" @@ -65,4 +66,5 @@ type ArgMetaBootstrapper struct { EpochBootstrapper process.EpochBootstrapper ValidatorStatisticsDBSyncer process.AccountsDBSyncer ValidatorAccountsDB state.AccountsAdapter + Watchdog core.WatchdogTimer } diff --git a/process/sync/export_test.go b/process/sync/export_test.go index f8f172b733e..bb90b38659c 100644 --- a/process/sync/export_test.go +++ b/process/sync/export_test.go @@ -6,6 +6,7 @@ import ( "github.com/multiversx/mx-chain-core-go/core" "github.com/multiversx/mx-chain-core-go/data" "github.com/multiversx/mx-chain-core-go/data/block" + "github.com/multiversx/mx-chain-go/common" "github.com/multiversx/mx-chain-go/process" ) @@ -35,6 +36,16 @@ func (boot *MetaBootstrap) ReceivedProof(header data.HeaderProofHandler) { boot.processReceivedProof(header) } +// RequestEpochStartBlockIfStuck - +func (boot *MetaBootstrap) RequestEpochStartBlockIfStuck() { + boot.requestEpochStartBlockIfStuck() +} + +// SetWatchdogLastNonce - +func (boot *MetaBootstrap) SetWatchdogLastNonce(nonce uint64) { + boot.watchdogLastNonce = nonce +} + // SetRcvHdrNonce - func (boot *MetaBootstrap) SetRcvHdrNonce() { boot.chRcvHdrNonce <- true diff --git a/process/sync/metablock.go b/process/sync/metablock.go index b821bfe4d3c..65a4a08922d 100644 --- a/process/sync/metablock.go +++ b/process/sync/metablock.go @@ -8,6 +8,8 @@ import ( "github.com/multiversx/mx-chain-core-go/core/check" "github.com/multiversx/mx-chain-core-go/data" "github.com/multiversx/mx-chain-core-go/data/block" + + "github.com/multiversx/mx-chain-go/common" "github.com/multiversx/mx-chain-go/dataRetriever" "github.com/multiversx/mx-chain-go/process" "github.com/multiversx/mx-chain-go/state" @@ -15,12 +17,21 @@ import ( "github.com/multiversx/mx-chain-go/trie/storageMarker" ) +const ( + numRoundsWithoutCommittedBlock = 5 + metaSyncEpochStartWatchdogID = "metaSyncEpochStartWatchdog" +) + // MetaBootstrap implements the bootstrap mechanism type MetaBootstrap struct { *baseBootstrap epochBootstrapper process.EpochBootstrapper validatorStatisticsDBSyncer process.AccountsDBSyncer validatorAccountsDB state.AccountsAdapter + + watchdog core.WatchdogTimer + watchdogCtx context.Context + watchdogLastNonce uint64 } // NewMetaBootstrap creates a new Bootstrap object @@ -46,6 +57,9 @@ func NewMetaBootstrap(arguments ArgMetaBootstrapper) (*MetaBootstrap, error) { if check.IfNil(arguments.ValidatorAccountsDB) { return nil, process.ErrNilPeerAccountsAdapter } + if check.IfNil(arguments.Watchdog) { + return nil, process.ErrNilWatchdog + } err := checkBaseBootstrapParameters(arguments.ArgBaseBootstrapper) if err != nil { @@ -94,6 +108,7 @@ func NewMetaBootstrap(arguments ArgMetaBootstrapper) (*MetaBootstrap, error) { epochBootstrapper: arguments.EpochBootstrapper, validatorStatisticsDBSyncer: arguments.ValidatorStatisticsDBSyncer, validatorAccountsDB: arguments.ValidatorAccountsDB, + watchdog: arguments.Watchdog, } base.blockBootstrapper = &boot @@ -152,11 +167,91 @@ func (boot *MetaBootstrap) StartSyncingBlocks() error { var ctx context.Context ctx, boot.cancelFunc = context.WithCancel(context.Background()) + boot.watchdogCtx = ctx go boot.syncBlocks(ctx) + boot.armEpochStartWatchdog() + return nil } +func (boot *MetaBootstrap) armEpochStartWatchdog() { + if boot.watchdogCtx == nil || boot.watchdogCtx.Err() != nil { + return + } + + timeout := boot.roundHandler.TimeDuration() * numRoundsWithoutCommittedBlock + if timeout <= 0 { + return + } + + // capture the baseline nonce now so the alarm measures progress over a single interval + boot.watchdogLastNonce = boot.currentBlockNonce() + boot.watchdog.Set(boot.epochStartWatchdogCallback, timeout, metaSyncEpochStartWatchdogID) +} + +func (boot *MetaBootstrap) currentBlockNonce() uint64 { + currentHeader := boot.chainHandler.GetCurrentBlockHeader() + if check.IfNil(currentHeader) { + return 0 + } + + return currentHeader.GetNonce() +} + +func (boot *MetaBootstrap) epochStartWatchdogCallback(_ string) { + if boot.watchdogCtx == nil || boot.watchdogCtx.Err() != nil { + return + } + defer boot.armEpochStartWatchdog() + + boot.requestEpochStartBlockIfStuck() +} + +func (boot *MetaBootstrap) requestEpochStartBlockIfStuck() { + currentHeader := boot.chainHandler.GetCurrentBlockHeader() + if check.IfNil(currentHeader) { + return + } + + currentNonce := currentHeader.GetNonce() + if currentNonce != boot.watchdogLastNonce { + return + } + + currentEpoch := currentHeader.GetEpoch() + targetEpoch := currentEpoch + 1 + if !boot.enableEpochsHandler.IsFlagEnabledInEpoch(common.AndromedaFlag, targetEpoch) { + return + } + + targetNonce := currentNonce + 1 + + header, headerHash, err := process.GetMetaHeaderFromPoolWithNonce(targetNonce, boot.headers) + if err == nil && !check.IfNil(header) { + if boot.proofs.HasProof(core.MetachainShardId, headerHash) { + return + } + + log.Debug("epoch start watchdog: header present without proof, requesting proof by hash", + "nonce", targetNonce, + "epoch", header.GetEpoch(), + "hash", headerHash, + ) + boot.requestHandler.SetEpoch(header.GetEpoch()) + boot.requestHandler.RequestEquivalentProofByHash(core.MetachainShardId, headerHash) + return + } + + log.Debug("epoch start watchdog: stuck without epoch change metablock, requesting header and proof", + "nonce", targetNonce, + "epoch", targetEpoch, + ) + boot.requestHandler.SetEpoch(targetEpoch) + boot.requestHandler.RequestStartOfEpochMetaBlock(targetEpoch) + boot.requestHandler.RequestEquivalentProofByNonce(core.MetachainShardId, targetNonce) +} + func (boot *MetaBootstrap) setLastEpochStartRound() { hdr := boot.chainHandler.GetCurrentBlockHeader() if check.IfNil(hdr) || hdr.GetEpoch() < 1 { @@ -222,6 +317,10 @@ func (boot *MetaBootstrap) Close() error { return nil } + if !check.IfNil(boot.watchdog) { + boot.watchdog.Stop(metaSyncEpochStartWatchdogID) + } + return boot.baseBootstrap.Close() } diff --git a/process/sync/metablock_test.go b/process/sync/metablock_test.go index 8649b23cca1..a4438f722f3 100644 --- a/process/sync/metablock_test.go +++ b/process/sync/metablock_test.go @@ -96,11 +96,164 @@ func CreateMetaBootstrapMockArguments() sync.ArgMetaBootstrapper { EpochBootstrapper: &mock.EpochStartTriggerStub{}, ValidatorAccountsDB: &stateMock.AccountsStub{}, ValidatorStatisticsDBSyncer: &mock.AccountsDBSyncerStub{}, + Watchdog: &testscommon.WatchdogMock{}, } return argsMetaBootstrapper } +func TestMetaBootstrap_RequestEpochStartBlockIfStuck(t *testing.T) { + t.Parallel() + + const currentNonce = uint64(100) + const currentEpoch = uint32(0) + + newArgs := func() (sync.ArgMetaBootstrapper, *headerRequestsRecorder) { + recorder := &headerRequestsRecorder{} + args := CreateMetaBootstrapMockArguments() + args.ChainHandler = &testscommon.ChainHandlerStub{ + GetCurrentBlockHeaderCalled: func() data.HeaderHandler { + return &block.MetaBlock{Nonce: currentNonce, Epoch: currentEpoch} + }, + } + args.RequestHandler = &testscommon.RequestHandlerStub{ + RequestStartOfEpochMetaBlockCalled: func(epoch uint32) { + recorder.startOfEpochCalls++ + recorder.startOfEpochArg = epoch + }, + RequestEquivalentProofByHashCalled: func(headerShard uint32, headerHash []byte) { + recorder.proofByHashCalls++ + recorder.proofByHashArg = headerHash + }, + RequestEquivalentProofByNonceCalled: func(headerShard uint32, headerNonce uint64) { + recorder.proofByNonceCalls++ + recorder.proofByNonceArg = headerNonce + }, + } + args.EnableEpochsHandler = &enableEpochsHandlerMock.EnableEpochsHandlerStub{ + IsFlagEnabledInEpochCalled: func(flag core.EnableEpochFlag, epoch uint32) bool { + return flag == common.AndromedaFlag + }, + } + return args, recorder + } + + t.Run("not stuck (nonce advanced since last arm) does not request anything", func(t *testing.T) { + t.Parallel() + + args, recorder := newArgs() + boot, err := sync.NewMetaBootstrap(args) + require.Nil(t, err) + + // baseline is behind the current nonce -> progress was made during the interval + boot.SetWatchdogLastNonce(currentNonce - 1) + boot.RequestEpochStartBlockIfStuck() + require.Zero(t, recorder.total()) + }) + + t.Run("stuck, header absent, requests start of epoch block and proof by nonce", func(t *testing.T) { + t.Parallel() + + args, recorder := newArgs() + boot, err := sync.NewMetaBootstrap(args) + require.Nil(t, err) + + boot.SetWatchdogLastNonce(currentNonce) + boot.RequestEpochStartBlockIfStuck() + + require.Equal(t, 1, recorder.startOfEpochCalls) + require.Equal(t, currentEpoch+1, recorder.startOfEpochArg) + require.Equal(t, 1, recorder.proofByNonceCalls) + require.Equal(t, currentNonce+1, recorder.proofByNonceArg) + require.Equal(t, 0, recorder.proofByHashCalls) + }) + + t.Run("stuck, header present without proof, requests proof by hash", func(t *testing.T) { + t.Parallel() + + expectedHash := []byte("epoch-change-hash") + args, recorder := newArgs() + args.PoolsHolder = poolsWithMetaHeader(currentNonce+1, currentEpoch+1, expectedHash, false) + boot, err := sync.NewMetaBootstrap(args) + require.Nil(t, err) + + boot.SetWatchdogLastNonce(currentNonce) + boot.RequestEpochStartBlockIfStuck() + + require.Equal(t, 1, recorder.proofByHashCalls) + require.Equal(t, expectedHash, recorder.proofByHashArg) + require.Equal(t, 0, recorder.startOfEpochCalls) + require.Equal(t, 0, recorder.proofByNonceCalls) + }) + + t.Run("stuck, header present with proof, does not request anything", func(t *testing.T) { + t.Parallel() + + args, recorder := newArgs() + args.PoolsHolder = poolsWithMetaHeader(currentNonce+1, currentEpoch+1, []byte("hash"), true) + boot, err := sync.NewMetaBootstrap(args) + require.Nil(t, err) + + boot.SetWatchdogLastNonce(currentNonce) + boot.RequestEpochStartBlockIfStuck() + + require.Zero(t, recorder.total()) + }) + + t.Run("stuck but andromeda not enabled does not request anything", func(t *testing.T) { + t.Parallel() + + args, recorder := newArgs() + args.EnableEpochsHandler = &enableEpochsHandlerMock.EnableEpochsHandlerStub{ + IsFlagEnabledInEpochCalled: func(flag core.EnableEpochFlag, epoch uint32) bool { + return false + }, + } + boot, err := sync.NewMetaBootstrap(args) + require.Nil(t, err) + + boot.SetWatchdogLastNonce(currentNonce) + boot.RequestEpochStartBlockIfStuck() + + require.Zero(t, recorder.total()) + }) +} + +type headerRequestsRecorder struct { + startOfEpochCalls int + startOfEpochArg uint32 + proofByHashCalls int + proofByHashArg []byte + proofByNonceCalls int + proofByNonceArg uint64 +} + +func (r *headerRequestsRecorder) total() int { + return r.startOfEpochCalls + r.proofByHashCalls + r.proofByNonceCalls +} + +func poolsWithMetaHeader(nonce uint64, epoch uint32, hash []byte, hasProof bool) *dataRetrieverMock.PoolsHolderStub { + pools := createMockPools() + pools.HeadersCalled = func() dataRetriever.HeadersPool { + return &mock.HeadersCacherStub{ + GetHeaderByNonceAndShardIdCalled: func(hdrNonce uint64, shardId uint32) ([]data.HeaderHandler, [][]byte, error) { + if hdrNonce != nonce { + return nil, nil, errors.New("not found") + } + return []data.HeaderHandler{&block.MetaBlock{Nonce: nonce, Epoch: epoch}}, [][]byte{hash}, nil + }, + } + } + pools.ProofsCalled = func() dataRetriever.ProofsPool { + return &dataRetrieverMock.ProofsPoolMock{ + HasProofCalled: func(shardID uint32, headerHash []byte) bool { + return hasProof + }, + } + } + return pools +} + // ------- NewMetaBootstrap func TestNewMetaBootstrap_NilPoolsHolderShouldErr(t *testing.T) { @@ -115,6 +268,18 @@ func TestNewMetaBootstrap_NilPoolsHolderShouldErr(t *testing.T) { assert.Equal(t, process.ErrNilPoolsHolder, err) } +func TestNewMetaBootstrap_NilWatchdogShouldErr(t *testing.T) { + t.Parallel() + + args := CreateMetaBootstrapMockArguments() + args.Watchdog = nil + + bs, err := sync.NewMetaBootstrap(args) + + assert.True(t, check.IfNil(bs)) + assert.Equal(t, process.ErrNilWatchdog, err) +} + func TestNewMetaBootstrap_NilValidatorDBShouldErr(t *testing.T) { t.Parallel() From 2064a96fbc7f6e9183e1a379221af0c76edfa9e1 Mon Sep 17 00:00:00 2001 From: Adrian Dobrita Date: Wed, 10 Jun 2026 11:18:44 +0300 Subject: [PATCH 4/4] fix after review --- epochStart/shardchain/trigger.go | 14 ++++++++++++-- 1 file changed, 12 insertions(+), 2 deletions(-) diff --git a/epochStart/shardchain/trigger.go b/epochStart/shardchain/trigger.go index c588a70068a..471842f04a3 100644 --- a/epochStart/shardchain/trigger.go +++ b/epochStart/shardchain/trigger.go @@ -1298,16 +1298,26 @@ func (t *trigger) watchdogRequestEpochStartMetaBlock(ctx context.Context) { timer := time.NewTimer(watchdogTimeout) defer timer.Stop() + resetTimer := func(d time.Duration) { + if !timer.Stop() { + select { + case <-timer.C: + default: + } + } + timer.Reset(d) + } + for { select { case <-ctx.Done(): log.Debug("watchdogRequestEpochStartMetaBlock: trigger's go routine is stopping...") return case <-t.chanMetaBlockReceived: - timer.Reset(t.resetWatchdogTimeout(watchdogTimeout)) + resetTimer(t.resetWatchdogTimeout(watchdogTimeout)) case <-timer.C: t.handleWatchdogTimeout() - timer.Reset(t.resetWatchdogTimeout(watchdogTimeout)) + resetTimer(t.resetWatchdogTimeout(watchdogTimeout)) } } }