test: increase test coverage (block manager)

.gitignore

@@ -11,3 +11,4 @@ types/pb/tendermint
 build
 .aider*
 .DS_Store
+coverage.out

block/aggregation.go

@@ -0,0 +1,103 @@
+package block
+
+import (
+	"context"
+	"time"
+)
+
+// AggregationLoop is responsible for aggregating transactions into rollup-blocks.
+func (m *Manager) AggregationLoop(ctx context.Context) {
+	initialHeight := m.genesis.InitialHeight //nolint:gosec
+	height, err := m.store.Height(ctx)
+	if err != nil {
+ m.logger.Error("error while getting store height", "error", err)
+ return
+	}
+	var delay time.Duration
+
+	// TODO(tzdybal): double-check when https://github.com/celestiaorg/rollmint/issues/699 is resolved
+	if height < initialHeight {
+ delay = time.Until(m.genesis.GenesisDAStartTime.Add(m.config.Node.BlockTime.Duration))
+	} else {
+ lastBlockTime := m.getLastBlockTime()
+ delay = time.Until(lastBlockTime.Add(m.config.Node.BlockTime.Duration))
+	}
+
+	if delay > 0 {
+ m.logger.Info("Waiting to produce block", "delay", delay)
+ time.Sleep(delay)
+	}
+
+	// blockTimer is used to signal when to build a block based on the
+	// rollup block time. A timer is used so that the time to build a block
+	// can be taken into account.
+	blockTimer := time.NewTimer(0)
+	defer blockTimer.Stop()
+
+	// Lazy Aggregator mode.
+	// In Lazy Aggregator mode, blocks are built only when there are
+	// transactions or every LazyBlockTime.
+	if m.config.Node.LazyAggregator {
+ m.lazyAggregationLoop(ctx, blockTimer)
+ return
+	}
+
+	m.normalAggregationLoop(ctx, blockTimer)
+}
+
+func (m *Manager) lazyAggregationLoop(ctx context.Context, blockTimer *time.Timer) {
+	// lazyTimer triggers block publication even during inactivity
+	lazyTimer := time.NewTimer(0)
+	defer lazyTimer.Stop()
+
+	for {
+ select {
+ case <-ctx.Done():
+ return
+
+ case <-lazyTimer.C:
+ case <-blockTimer.C:
+ }
+
+ // Reset the start time
+ start := time.Now()
+
+ // Attempt to publish the block regardless of activity
+ if err := m.publishBlock(ctx); err != nil && ctx.Err() == nil {
+ m.logger.Error("error while publishing block", "error", err)
+ }
+
+ // Reset both timers for the next aggregation window
+ lazyTimer.Reset(getRemainingSleep(start, m.config.Node.LazyBlockTime.Duration))
+ blockTimer.Reset(getRemainingSleep(start, m.config.Node.BlockTime.Duration))
+	}
+}
+
+func (m *Manager) normalAggregationLoop(ctx context.Context, blockTimer *time.Timer) {
+	for {
+ select {
+ case <-ctx.Done():
+ return
+ case <-blockTimer.C:
+ // Define the start time for the block production period
+ start := time.Now()
+
+ if err := m.publishBlock(ctx); err != nil && ctx.Err() == nil {
+ m.logger.Error("error while publishing block", "error", err)
+ }
+ // Reset the blockTimer to signal the next block production
+ // period based on the block time.
+ blockTimer.Reset(getRemainingSleep(start, m.config.Node.BlockTime.Duration))
+ }
+	}
+}
+
+func getRemainingSleep(start time.Time, interval time.Duration) time.Duration {
+	elapsed := time.Since(start)
+
+	if elapsed < interval {
+ return interval - elapsed
+	}
+
+	return time.Millisecond
+}

block/aggregation_test.go

@@ -0,0 +1,210 @@
+package block
+
+import (
+	"context"
+	"errors"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"cosmossdk.io/log"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
+	"github.com/stretchr/testify/require"
+
+	"github.com/rollkit/rollkit/pkg/cache"
+	"github.com/rollkit/rollkit/pkg/config"
+	genesispkg "github.com/rollkit/rollkit/pkg/genesis"
+	"github.com/rollkit/rollkit/test/mocks"
+	"github.com/rollkit/rollkit/types"
+)
+
+// TestAggregationLoop_Normal_BasicInterval verifies the basic time interval logic of the normal aggregation loop.
+func TestAggregationLoop_Normal_BasicInterval(t *testing.T) {
+	assert := assert.New(t)
+	require := require.New(t)
+
+	blockTime := 50 * time.Millisecond
+	waitTime := blockTime*4 + blockTime/2
+
+	mockStore := mocks.NewStore(t)
+	mockStore.On("Height", mock.Anything).Return(uint64(1), nil).Maybe()
+	mockStore.On("GetState", mock.Anything).Return(types.State{LastBlockTime: time.Now().Add(-blockTime)}, nil).Maybe()
+
+	mockExec := mocks.NewExecutor(t)
+	mockSeq := mocks.NewSequencer(t)
+	mockDAC := mocks.NewClient(t)
+	logger := log.NewTestLogger(t)
+
+	m := &Manager{
+ store: mockStore,
+ exec: mockExec,
+ sequencer: mockSeq,
+ dalc: mockDAC,
+ logger: logger,
+ config: config.Config{
+ Node: config.NodeConfig{
+ BlockTime: config.DurationWrapper{Duration: blockTime},
+ LazyAggregator: false,
+ },
+ DA: config.DAConfig{
+ BlockTime: config.DurationWrapper{Duration: 1 * time.Second},
+ },
+ },
+ genesis: genesispkg.Genesis{
+ InitialHeight: 1,
+ },
+ lastState: types.State{
+ LastBlockTime: time.Now().Add(-blockTime),
+ },
+ lastStateMtx: &sync.RWMutex{},
+ metrics: NopMetrics(),
+ headerCache: cache.NewCache[types.SignedHeader](),
+ dataCache: cache.NewCache[types.Data](),
+	}
+
+	var publishTimes []time.Time
+	var publishLock sync.Mutex
+	mockPublishBlock := func(ctx context.Context) error {
+ publishLock.Lock()
+ defer publishLock.Unlock()
+ publishTimes = append(publishTimes, time.Now())
+ m.logger.Debug("Mock publishBlock called", "time", publishTimes[len(publishTimes)-1])
+ return nil
+	}
+	m.publishBlock = mockPublishBlock
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	var wg sync.WaitGroup
+	wg.Add(1)
+	go func() {
+ defer wg.Done()
+ m.AggregationLoop(ctx)
+ m.logger.Info("AggregationLoop exited")
+	}()
+
+	m.logger.Info("Waiting for blocks...", "duration", waitTime)
+	time.Sleep(waitTime)
+
+	m.logger.Info("Cancelling context")
+	cancel()
+	m.logger.Info("Waiting for WaitGroup")
+	wg.Wait()
+	m.logger.Info("WaitGroup finished")
+
+	publishLock.Lock()
+	defer publishLock.Unlock()
+
+	m.logger.Info("Recorded publish times", "count", len(publishTimes), "times", publishTimes)
+
+	expectedCallsLow := int(waitTime/blockTime) - 1
+	expectedCallsHigh := int(waitTime/blockTime) + 1
+	require.GreaterOrEqualf(len(publishTimes), expectedCallsLow, "Expected at least %d calls, got %d", expectedCallsLow, len(publishTimes))
+	require.LessOrEqualf(len(publishTimes), expectedCallsHigh, "Expected at most %d calls, got %d", expectedCallsHigh, len(publishTimes))
+
+	if len(publishTimes) > 1 {
+ for i := 1; i < len(publishTimes); i++ {
+ interval := publishTimes[i].Sub(publishTimes[i-1])
+ m.logger.Debug("Checking interval", "index", i, "interval", interval)
+ tolerance := blockTime / 2
+ assert.True(WithinDuration(t, blockTime, interval, tolerance), "Interval %d (%v) not within tolerance (%v) of blockTime (%v)", i, interval, tolerance, blockTime)
+ }
+	}
+}
+
+// TestAggregationLoop_Normal_PublishBlockError verifies the loop continues after publishBlock returns an error.
+func TestAggregationLoop_Normal_PublishBlockError(t *testing.T) {
+	assert := assert.New(t)
+	require := require.New(t)
+
+	blockTime := 50 * time.Millisecond
+	waitTime := blockTime*4 + blockTime/2
+	tolerance := blockTime / 2
+
+	mockStore := mocks.NewStore(t)
+	mockStore.On("Height", mock.Anything).Return(uint64(1), nil).Maybe()
+	mockStore.On("GetState", mock.Anything).Return(types.State{LastBlockTime: time.Now().Add(-blockTime)}, nil).Maybe()
+
+	mockExec := mocks.NewExecutor(t)
+	mockSeq := mocks.NewSequencer(t)
+	mockDAC := mocks.NewClient(t)
+
+	mockLogger := log.NewTestLogger(t)
+
+	// Create a basic Manager instance
+	m := &Manager{
+ store: mockStore,
+ exec: mockExec,
+ sequencer: mockSeq,
+ dalc: mockDAC,
+ logger: mockLogger,
+ config: config.Config{
+ Node: config.NodeConfig{
+ BlockTime: config.DurationWrapper{Duration: blockTime},
+ LazyAggregator: false,
+ },
+ DA: config.DAConfig{
+ BlockTime: config.DurationWrapper{Duration: 1 * time.Second},
+ },
+ },
+ genesis: genesispkg.Genesis{
+ InitialHeight: 1,
+ },
+ lastState: types.State{
+ LastBlockTime: time.Now().Add(-blockTime),
+ },
+ lastStateMtx: &sync.RWMutex{},
+ metrics: NopMetrics(),
+ headerCache: cache.NewCache[types.SignedHeader](),
+ dataCache: cache.NewCache[types.Data](),
+	}
+
+	var publishCalls atomic.Int64
+	var publishTimes []time.Time
+	var publishLock sync.Mutex
+	expectedErr := errors.New("failed to publish block")
+
+	mockPublishBlock := func(ctx context.Context) error {
+ callNum := publishCalls.Add(1)
+ publishLock.Lock()
+ publishTimes = append(publishTimes, time.Now())
+ publishLock.Unlock()
+
+ if callNum == 1 {
+ m.logger.Debug("Mock publishBlock returning error", "call", callNum)
+ return expectedErr
+ }
+ m.logger.Debug("Mock publishBlock returning nil", "call", callNum)
+ return nil
+	}
+	m.publishBlock = mockPublishBlock
+
+	ctx, cancel := context.WithCancel(context.Background())
+	var wg sync.WaitGroup
+	wg.Add(1)
+	go func() {
+ defer wg.Done()
+ m.AggregationLoop(ctx)
+ m.logger.Info("AggregationLoop exited")
+	}()
+
+	time.Sleep(waitTime)
+
+	cancel()
+	wg.Wait()
+
+	publishLock.Lock()
+	defer publishLock.Unlock()
+
+	calls := publishCalls.Load()
+	assert.GreaterOrEqualf(calls, int64(4), "publishBlock should have been called multiple times (around 4), but was called %d times", calls)
+	assert.LessOrEqualf(calls, int64(5), "publishBlock should have been called multiple times (around 4-5), but was called %d times", calls)
+
+	require.GreaterOrEqual(len(publishTimes), 3, "Need at least 3 timestamps to check intervals after error")
+	for i := 2; i < len(publishTimes); i++ {
+ interval := publishTimes[i].Sub(publishTimes[i-1])
+ WithinDuration(t, blockTime, interval, tolerance)
+	}
+}