Update/v0.7.3

.gitignore

@@ -31,3 +31,4 @@ coverage.out
 db/db-packr.go
 db/packrd/
 packrd/
+nohup.out

.goreleaser.yaml

@@ -1,4 +1,6 @@
 # .goreleaser.yaml
+version: 2
+
 builds:
 - main: ./cmd/
   goos:

aggregator/aggregator.go

@@ -136,6 +136,12 @@ func (a *Aggregator) Start(ctx context.Context) error {
 		return fmt.Errorf("failed to initialize proofs cache %w", err)
 	}
 
+	for !a.isSynced(ctx, nil) {
+		log.Info("Waiting for synchronizer to sync...")
+		time.Sleep(a.cfg.RetryTime.Duration)
+		continue
+	}
+
 	address := fmt.Sprintf("%s:%d", a.cfg.Host, a.cfg.Port)
 	lis, err := net.Listen("tcp", address)
 	if err != nil {
@@ -227,11 +233,6 @@ func (a *Aggregator) Channel(stream prover.AggregatorService_ChannelServer) erro
 			_, err = a.tryBuildFinalProof(ctx, prover, nil)
 			if err != nil {
 				log.Errorf("Error checking proofs to verify: %v", err)
-
-				if errors.Is(err, context.Canceled) {
-					// the context was canceled, just continue, the loop will stop in the <-ctx.Done() case
-					continue
-				}
 			}
 
 			proofGenerated, err := a.tryAggregateProofs(ctx, prover)
@@ -495,9 +496,10 @@ func (a *Aggregator) tryBuildFinalProof(ctx context.Context, prover proverInterf
 	}
 	log.Debug("Send final proof time reached")
 
-	if err = a.waitForSynchronizerToSyncUp(ctx, nil); err != nil {
-		log.Warn("waiting for the synchronizer to sync up was canceled", err)
-		return false, err
+	for !a.isSynced(ctx, nil) {
+		log.Info("Waiting for synchronizer to sync...")
+		time.Sleep(a.cfg.RetryTime.Duration)
+		continue
 	}
 
 	var lastVerifiedBatchNum uint64
@@ -1073,41 +1075,41 @@ func (a *Aggregator) resetVerifyProofTime() {
 
 // isSynced checks if the state is synchronized with L1. If a batch number is
 // provided, it makes sure that the state is synced with that batch.
-func (a *Aggregator) isSynced(ctx context.Context, batchNum *uint64) (bool, error) {
+func (a *Aggregator) isSynced(ctx context.Context, batchNum *uint64) bool {
 	// get latest verified batch as seen by the synchronizer
 	lastVerifiedBatch, err := a.State.GetLastVerifiedBatch(ctx, nil)
 	if err == state.ErrNotFound {
-		return false, nil
+		return false
 	}
 	if err != nil {
 		log.Warnf("Failed to get last consolidated batch: %v", err)
-		return false, err
+		return false
 	}
 
 	if lastVerifiedBatch == nil {
-		return false, nil
+		return false
 	}
 
 	if batchNum != nil && lastVerifiedBatch.BatchNumber < *batchNum {
 		log.Infof("Waiting for the state to be synced, lastVerifiedBatchNum: %d, waiting for batch: %d", lastVerifiedBatch.BatchNumber, batchNum)
-		return false, nil
+		return false
 	}
 
 	// latest verified batch in L1
 	lastVerifiedEthBatchNum, err := a.Ethman.GetLatestVerifiedBatchNum()
 	if err != nil {
 		log.Warnf("Failed to get last eth batch, err: %v", err)
-		return false, err
+		return false
 	}
 
 	// check if L2 is synced with L1
 	if lastVerifiedBatch.BatchNumber < lastVerifiedEthBatchNum {
 		log.Infof("Waiting for the state to be synced, lastVerifiedBatchNum: %d, lastVerifiedEthBatchNum: %d, waiting for batch",
 			lastVerifiedBatch.BatchNumber, lastVerifiedEthBatchNum)
-		return false, nil
+		return false
 	}
 
-	return true, nil
+	return true
 }
 
 func (a *Aggregator) buildInputProver(ctx context.Context, batchToVerify *state.Batch) (*prover.InputProver, error) {
@@ -1301,9 +1303,9 @@ func (a *Aggregator) handleMonitoredTxResult(result ethtxmanager.MonitoredTxResu
 
 	// wait for the synchronizer to catch up the verified batches
 	log.Debug("A final proof has been sent, waiting for the network to be synced")
-	if err := a.waitForSynchronizerToSyncUp(a.ctx, &proofBatchNumberFinal); err != nil {
-		log.Warn("waiting for the synchronizer to sync up was canceled", err)
-		return
+	for !a.isSynced(a.ctx, &proofBatchNumberFinal) {
+		log.Info("Waiting for synchronizer to sync...")
+		time.Sleep(a.cfg.RetryTime.Duration)
 	}
 
 	// network is synced with the final proof, we can safely delete all recursive
@@ -1314,24 +1316,6 @@ func (a *Aggregator) handleMonitoredTxResult(result ethtxmanager.MonitoredTxResu
 	}
 }
 
-func (a *Aggregator) waitForSynchronizerToSyncUp(ctx context.Context, batchNum *uint64) error {
-	for {
-		log.Info("waiting for the synchronizer to sync...")
-		synced, err := a.isSynced(ctx, batchNum)
-		if err != nil && errors.Is(err, context.Canceled) {
-			// if context is canceled, stop the loop, since it will never
-			// be able to execute properly and break in this case, and we will be stuck in it forever
-			return err
-		}
-
-		if synced {
-			return nil
-		}
-
-		time.Sleep(a.cfg.RetryTime.Duration)
-	}
-}
-
 func buildMonitoredTxID(batchNumber, batchNumberFinal uint64) string {
 	return fmt.Sprintf(monitoredIDFormat, batchNumber, batchNumberFinal)
 }

aggregator/aggregator_test.go

@@ -1450,79 +1450,8 @@ func TestIsSynced(t *testing.T) {
 				tc.setup(m, &a)
 			}
 
-			synced, _ := a.isSynced(a.ctx, tc.batchNum)
+			synced := a.isSynced(a.ctx, tc.batchNum)
 			assert.Equal(tc.synced, synced)
 		})
 	}
 }
-
-func TestWaitForSynchronizerToSyncUp(t *testing.T) {
-	t.Parallel()
-
-	cfg := Config{}
-	batchNum := uint64(42)
-	testCases := []struct {
-		name     string
-		setup    func(mox, *Aggregator)
-		batchNum *uint64
-		synced   bool
-	}{
-		{
-			name:     "state context canceled",
-			synced:   false,
-			batchNum: &batchNum,
-			setup: func(m mox, a *Aggregator) {
-				m.stateMock.On("GetLastVerifiedBatch", mock.Anything, nil).Return(nil, context.Canceled).Once()
-			},
-		},
-		{
-			name:     "ok after multiple iterations",
-			synced:   true,
-			batchNum: &batchNum,
-			setup: func(m mox, a *Aggregator) {
-				latestVerifiedBatch := state.VerifiedBatch{BatchNumber: batchNum}
-				m.stateMock.On("GetLastVerifiedBatch", mock.Anything, nil).Return(nil, nil).Once()
-				m.stateMock.On("GetLastVerifiedBatch", mock.Anything, nil).Return(&latestVerifiedBatch, nil).Once()
-				m.etherman.On("GetLatestVerifiedBatchNum").Return(batchNum, nil).Once()
-			},
-		},
-		{
-			name:     "ok with batch number",
-			synced:   true,
-			batchNum: &batchNum,
-			setup: func(m mox, a *Aggregator) {
-				latestVerifiedBatch := state.VerifiedBatch{BatchNumber: batchNum}
-				m.stateMock.On("GetLastVerifiedBatch", mock.Anything, nil).Return(&latestVerifiedBatch, nil).Once()
-				m.etherman.On("GetLatestVerifiedBatchNum").Return(batchNum, nil).Once()
-			},
-		},
-	}
-	for _, tc := range testCases {
-		t.Run(tc.name, func(t *testing.T) {
-			stateMock := mocks.NewStateMock(t)
-			ethTxManager := mocks.NewEthTxManager(t)
-			etherman := mocks.NewEtherman(t)
-			proverMock := mocks.NewProverMock(t)
-			a, err := New(cfg, stateMock, ethTxManager, etherman, nil, nil)
-			require.NoError(t, err)
-			aggregatorCtx := context.WithValue(context.Background(), "owner", "aggregator") //nolint:staticcheck
-			a.ctx, a.exit = context.WithCancel(aggregatorCtx)
-			m := mox{
-				stateMock:    stateMock,
-				ethTxManager: ethTxManager,
-				etherman:     etherman,
-				proverMock:   proverMock,
-			}
-			if tc.setup != nil {
-				tc.setup(m, &a)
-			}
-
-			err = a.waitForSynchronizerToSyncUp(a.ctx, tc.batchNum)
-			if tc.synced {
-				assert.NoError(t, err)
-			} else {
-				assert.Error(t, err)
-			}
-		})
-	}
-}

cmd/run.go

@@ -753,7 +753,8 @@ func forkIDIntervals(ctx context.Context, st *state.State, etherman *etherman.Cl
 		if err != nil && !errors.Is(err, state.ErrStateNotSynchronized) {
 			return []state.ForkIDInterval{}, fmt.Errorf("error checking lastL1BlockSynced. Error: %v", err)
 		}
-		if lastBlock != nil {
+		// If lastBlock is below genesisBlock means state.ErrStateNotSynchronized (haven't started yet the sync process, is doing pregenesis sync)
+		if lastBlock != nil && lastBlock.BlockNumber >= genesisBlockNumber {
 			log.Info("Getting forkIDs intervals. Please wait...")
 			// Read Fork ID FROM POE SC
 			forkIntervals, err := etherman.GetForks(ctx, genesisBlockNumber, lastBlock.BlockNumber)
@@ -793,32 +794,14 @@ func forkIDIntervals(ctx context.Context, st *state.State, etherman *etherman.Cl
 			}
 			forkIDIntervals = forkIntervals
 		} else {
-			log.Debug("Getting all forkIDs")
-
-			// Get last L1 block number
-			bn, err := etherman.GetLatestBlockNumber(ctx)
-			if err != nil {
-				return []state.ForkIDInterval{}, fmt.Errorf("error getting latest block number. Error: %v", err)
-			}
-
-			// Get all forkIDs since genesis
-			forkIntervals, err := etherman.GetForks(ctx, genesisBlockNumber, bn)
+			log.Debug("Getting initial forkID")
+			forkIntervals, err := etherman.GetForks(ctx, genesisBlockNumber, genesisBlockNumber)
 			if err != nil {
 				return []state.ForkIDInterval{}, fmt.Errorf("error getting forks. Please check the configuration. Error: %v", err)
 			} else if len(forkIntervals) == 0 {
 				return []state.ForkIDInterval{}, fmt.Errorf("error: no forkID received. It should receive at least one, please check the configuration...")
 			}
 			forkIDIntervals = forkIntervals
-
-			log.Debugf("Retrieved %d forkIDs", len(forkIDIntervals))
-
-			log.Debug("Adding forkIDs to db and memory")
-			for _, forkID := range forkIDIntervals {
-				err = st.AddForkIDInterval(ctx, forkID, nil)
-				if err != nil {
-					log.Fatal("error adding forkID to db. Error: ", err)
-				}
-			}
 		}
 	}
 	return forkIDIntervals, nil

config/config_test.go

@@ -177,6 +177,14 @@ func Test_Defaults(t *testing.T) {
 			path:          "Sequencer.StreamServer.WriteTimeout",
 			expectedValue: types.NewDuration(5 * time.Second),
 		},
+		{
+			path:          "Sequencer.StreamServer.InactivityTimeout",
+			expectedValue: types.NewDuration(120 * time.Second),
+		},
+		{
+			path:          "Sequencer.StreamServer.InactivityCheckInterval",
+			expectedValue: types.NewDuration(5 * time.Second),
+		},
 		{
 			path:          "Sequencer.StreamServer.Enabled",
 			expectedValue: false,

config/default.go

@@ -165,6 +165,8 @@ StateConsistencyCheckInterval = "5s"
 		Filename = ""
 		Version = 0
 		WriteTimeout = "5s"
+		InactivityTimeout = "120s"
+		InactivityCheckInterval = "5s"
 		Enabled = false
 
 [SequenceSender]