fix(fork): proposer fork after syncing with multiple forks

block/fork.go

@@ -39,11 +39,6 @@ func (m *Manager) MonitorForkUpdateLoop(ctx context.Context) error {
 
 // checkForkUpdate checks if the hub has a fork update
 func (m *Manager) checkForkUpdate(msg string) error {
-	// if instruction exists no need to check for fork update
-	if types.InstructionExists(m.RootDir) {
-		return nil
-	}
-
 	rollapp, err := m.SLClient.GetRollapp()
 	if err != nil {
 		return err
@@ -54,23 +49,28 @@ func (m *Manager) checkForkUpdate(msg string) error {
 		actualRevision   = m.State.GetRevision()
 		expectedRevision = rollapp.GetRevisionForHeight(nextHeight)
 	)
-	if shouldStopNode(expectedRevision, nextHeight, actualRevision) {
-		err = m.createInstruction(expectedRevision)
+
+	if shouldStopNode(expectedRevision, nextHeight, actualRevision, m.RunMode) {
+		instruction, err := m.createInstruction(expectedRevision)
 		if err != nil {
 			return err
 		}
 
+		err = types.PersistInstructionToDisk(m.RootDir, instruction)
+		if err != nil {
+			return err
+		}
 		m.freezeNode(fmt.Errorf("%s  local_block_height: %d rollapp_revision_start_height: %d local_revision: %d rollapp_revision: %d", msg, m.State.Height(), expectedRevision.StartHeight, actualRevision, expectedRevision.Number))
 	}
 
 	return nil
 }
 
 // createInstruction writes file to disk with fork information
-func (m *Manager) createInstruction(expectedRevision types.Revision) error {
+func (m *Manager) createInstruction(expectedRevision types.Revision) (types.Instruction, error) {
 	obsoleteDrs, err := m.SLClient.GetObsoleteDrs()
 	if err != nil {
-		return err
+		return types.Instruction{}, err
 	}
 
 	instruction := types.Instruction{
@@ -79,12 +79,7 @@ func (m *Manager) createInstruction(expectedRevision types.Revision) error {
 		FaultyDRS:           obsoleteDrs,
 	}
 
-	err = types.PersistInstructionToDisk(m.RootDir, instruction)
-	if err != nil {
-		return err
-	}
-
-	return nil
+	return instruction, nil
 }
 
 // shouldStopNode determines if a rollapp node should be stopped based on revision criteria.
@@ -96,7 +91,11 @@ func shouldStopNode(
 	expectedRevision types.Revision,
 	nextHeight uint64,
 	actualRevisionNumber uint64,
+	nodeMode uint,
 ) bool {
+	if nodeMode == RunModeFullNode {
+		return nextHeight > expectedRevision.StartHeight && actualRevisionNumber < expectedRevision.Number
+	}
 	return nextHeight >= expectedRevision.StartHeight && actualRevisionNumber < expectedRevision.Number
 }
 
@@ -250,41 +249,57 @@ func (m *Manager) updateStateWhenFork() error {
 	return nil
 }
 
-// forkFromInstruction checks if fork is needed, reading instruction file, and performs fork actions
-func (m *Manager) forkFromInstruction() error {
-	// if instruction file exists proposer needs to do fork actions (if settlement height is higher than revision height it is considered fork already happened and no need to do anything)
-	instruction, forkNeeded := m.forkNeeded()
-	if !forkNeeded {
-		return nil
+// checkRevisionAndFork checks if fork is needed after syncing, and performs fork actions
+func (m *Manager) checkRevisionAndFork() error {
+	// it is checked again whether the node is the active proposer, since this could have changed after syncing.
+	amIProposerOnSL, err := m.AmIProposerOnSL()
+	if err != nil {
+		return fmt.Errorf("am i proposer on SL: %w", err)
+	}
+	if !amIProposerOnSL {
+		return fmt.Errorf("the node is no longer the proposer. please restart.")
+	}
+
+	// update sequencer in case it changed after syncing
+	err = m.UpdateProposerFromSL()
+	if err != nil {
+		return err
+	}
+
+	// get the revision for the current height to check against local state
+	rollapp, err := m.SLClient.GetRollapp()
+	if err != nil {
+		return err
 	}
-	if m.RunMode == RunModeProposer {
-		// it is checked again whether the node is the active proposer, since this could have changed after syncing.
-		amIProposerOnSL, err := m.AmIProposerOnSL()
+	expectedRevision := rollapp.GetRevisionForHeight(m.State.NextHeight())
+
+	// create fork batch in case it has not been submitted yet
+	if m.LastSettlementHeight.Load() < expectedRevision.StartHeight {
+		instruction, err := m.createInstruction(expectedRevision)
 		if err != nil {
-			return fmt.Errorf("am i proposer on SL: %w", err)
-		}
-		if !amIProposerOnSL {
-			return fmt.Errorf("the node is no longer the proposer. please restart.")
+			return err
 		}
 		// update revision with revision after fork
 		m.State.SetRevision(instruction.Revision)
-		// update sequencer in case it changed after syncing
-		err = m.UpdateProposerFromSL()
+		// create and submit fork batch
+		err = m.doFork(instruction)
 		if err != nil {
 			return err
 		}
-		// create fork batch in case it has not been submitted yet
-		if m.LastSettlementHeight.Load() < instruction.RevisionStartHeight {
-			err := m.doFork(instruction)
-			if err != nil {
-				return err
-			}
-		}
 	}
-	// remove instruction file after fork to avoid enter loop again
-	err := types.DeleteInstructionFromDisk(m.RootDir)
-	if err != nil {
-		return fmt.Errorf("deleting instruction file: %w", err)
+
+	// this cannot happen. it means the revision number obtained is not the same or the next revision. unable to fork.
+	if expectedRevision.Number != m.State.GetRevision() {
+		panic("Inconsistent expected revision number from Hub. Unable to fork")
 	}
+
+	// remove instruction file after fork to avoid enter fork loop again
+	if _, instructionExists := m.forkNeeded(); instructionExists {
+		err := types.DeleteInstructionFromDisk(m.RootDir)
+		if err != nil {
+			return fmt.Errorf("deleting instruction file: %w", err)
+		}
+	}
+
 	return nil
 }

block/fork_test.go

@@ -22,6 +22,7 @@ func TestShouldStopNode(t *testing.T) {
 		rollapp  *types.Rollapp
 		block    *types.Block
 		height   uint64
+		runMode  uint
 		expected bool
 	}{
 		{
@@ -40,6 +41,7 @@ func TestShouldStopNode(t *testing.T) {
 				},
 			},
 			height:   150,
+			runMode:  RunModeFullNode,
 			expected: true,
 		},
 		{
@@ -58,6 +60,7 @@ func TestShouldStopNode(t *testing.T) {
 				},
 			},
 			height:   50,
+			runMode:  RunModeFullNode,
 			expected: false,
 		},
 		{
@@ -76,14 +79,15 @@ func TestShouldStopNode(t *testing.T) {
 				},
 			},
 			height:   150,
+			runMode:  RunModeFullNode,
 			expected: false,
 		},
 	}
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			expectedRevision := tt.rollapp.GetRevisionForHeight(tt.height)
-			result := shouldStopNode(expectedRevision, tt.height, tt.block.Header.Version.App)
+			result := shouldStopNode(expectedRevision, tt.height, tt.block.Header.Version.App, tt.runMode)
 			assert.Equal(t, tt.expected, result)
 		})
 	}
@@ -249,7 +253,7 @@ func TestCreateInstruction(t *testing.T) {
 
 			manager.SLClient = mockSL
 			expectedRevision := tt.rollapp.GetRevisionForHeight(tt.block.Header.Height)
-			err := manager.createInstruction(expectedRevision)
+			_, err := manager.createInstruction(expectedRevision)
 			if tt.expectedError {
 				assert.Error(t, err)
 			} else {

block/manager.go

@@ -246,9 +246,14 @@ func (m *Manager) Start(ctx context.Context) error {
 	if err != nil {
 		return fmt.Errorf("am i proposer on SL: %w", err)
 	}
-	amIProposer := amIProposerOnSL || m.AmIProposerOnRollapp()
 
-	m.logger.Info("starting block manager", "mode", map[bool]string{true: "proposer", false: "full node"}[amIProposer])
+	if amIProposerOnSL || m.AmIProposerOnRollapp() {
+		m.RunMode = RunModeProposer
+	} else {
+		m.RunMode = RunModeFullNode
+	}
+
+	m.logger.Info("starting block manager", "mode", map[bool]string{true: "proposer", false: "full node"}[m.RunMode == RunModeProposer])
 
 	// update local state from latest state in settlement
 	err = m.updateFromLastSettlementState()
@@ -287,7 +292,7 @@ func (m *Manager) Start(ctx context.Context) error {
 	})
 
 	// run based on the node role
-	if !amIProposer {
+	if m.RunMode == RunModeFullNode {
 		return m.runAsFullNode(ctx, eg)
 	}
 

block/manager_test.go

@@ -131,7 +131,8 @@ func TestInitialState(t *testing.T) {
 	}
 }
 
-// TestProduceOnlyAfterSynced should test that we are resuming publishing blocks after we are synced
+//	should test that we are resuming publishing blocks after we are synced
+//
 // 1. Submit a batch and desync the manager
 // 2. Fail to produce blocks
 // 2. Sync the manager

block/modes.go

@@ -7,6 +7,7 @@ import (
 
 	"github.com/dymensionxyz/dymint/p2p"
 	"github.com/dymensionxyz/dymint/settlement"
+	"github.com/dymensionxyz/dymint/types"
 	uerrors "github.com/dymensionxyz/dymint/utils/errors"
 	uevent "github.com/dymensionxyz/dymint/utils/event"
 	"golang.org/x/sync/errgroup"
@@ -22,7 +23,6 @@ const (
 // setFraudHandler sets the fraud handler for the block manager.
 func (m *Manager) runAsFullNode(ctx context.Context, eg *errgroup.Group) error {
 	m.logger.Info("starting block manager", "mode", "full node")
-	m.RunMode = RunModeFullNode
 	// update latest finalized height
 	err := m.updateLastFinalizedHeightFromSettlement()
 	if err != nil {
@@ -36,18 +36,19 @@ func (m *Manager) runAsFullNode(ctx context.Context, eg *errgroup.Group) error {
 
 	m.subscribeFullNodeEvents(ctx)
 
-	// forkFromInstruction deletes fork instruction file for full nodes
-	err = m.forkFromInstruction()
-	if err != nil {
-		return err
+	if _, instructionExists := m.forkNeeded(); instructionExists {
+		// remove instruction file after fork to avoid enter fork loop again
+		err := types.DeleteInstructionFromDisk(m.RootDir)
+		if err != nil {
+			return fmt.Errorf("deleting instruction file: %w", err)
+		}
 	}
 
 	return nil
 }
 
 func (m *Manager) runAsProposer(ctx context.Context, eg *errgroup.Group) error {
 	m.logger.Info("starting block manager", "mode", "proposer")
-	m.RunMode = RunModeProposer
 	// Subscribe to batch events, to update last submitted height in case batch confirmation was lost. This could happen if the sequencer crash/restarted just after submitting a batch to the settlement and by the time we query the last batch, this batch wasn't accepted yet.
 	go uevent.MustSubscribe(ctx, m.Pubsub, "updateSubmittedHeightLoop", settlement.EventQueryNewSettlementBatchAccepted, m.UpdateLastSubmittedHeight, m.logger)
 	// Subscribe to P2P received blocks events (used for P2P syncing).
@@ -60,8 +61,8 @@ func (m *Manager) runAsProposer(ctx context.Context, eg *errgroup.Group) error {
 	// Sequencer must wait till node is synced till last submittedHeight, in case it is not
 	m.waitForSettlementSyncing()
 
-	// forkFromInstruction executes fork if necessary
-	err := m.forkFromInstruction()
+	// checkRevisionAndFork executes fork if necessary
+	err := m.checkRevisionAndFork()
 	if err != nil {
 		return err
 	}

testutil/types.go

@@ -24,7 +24,7 @@ const (
 	// BlockVersion is the default block version for testing
 	BlockVersion = 1
 	// AppVersion is the default app version for testing
-	AppVersion = 2
+	AppVersion = 0
 
 	SettlementAccountPrefix = "dym"
 )