Expand isSequenceScheduled to include running and queued tasks

common/block_scheduler.go

@@ -30,8 +30,10 @@ type BlockDependencyManager struct {
 	scheduler Scheduler
 
 	dependencies []*TaskWithDependents
-	maxDeps      uint64
-	closed       atomic.Bool
+	// scheduledSeqs holds every seq with a task pending, queued or running.
+	scheduledSeqs map[uint64]struct{}
+	maxDeps       uint64
+	closed        atomic.Bool
 }
 
 type TaskWithDependents struct {
@@ -52,9 +54,10 @@ func (t *TaskWithDependents) String() string {
 
 func NewBlockVerificationScheduler(logger Logger, maxDeps uint64, scheduler Scheduler) *BlockDependencyManager {
 	b := &BlockDependencyManager{
-		logger:    logger,
-		maxDeps:   maxDeps,
-		scheduler: scheduler,
+		logger:        logger,
+		maxDeps:       maxDeps,
+		scheduler:     scheduler,
+		scheduledSeqs: make(map[uint64]struct{}),
 	}
 
 	b.logger.Debug("Created BlockVerificationScheduler", zap.Uint64("maxDeps", maxDeps))
@@ -121,17 +124,13 @@ func (bs *BlockDependencyManager) ExecuteEmptyRoundDependents(emptyRound uint64)
 	bs.dependencies = remainingDeps
 }
 
+// IsSequenceScheduled reports whether a task for seq is pending on dependencies, queued or running.
 func (bs *BlockDependencyManager) IsSequenceScheduled(seq uint64) bool {
 	bs.lock.Lock()
 	defer bs.lock.Unlock()
 
-	for _, dep := range bs.dependencies {
-		if dep.blockSeq == seq {
-			return true
-		}
-	}
-
-	return false
+	_, ok := bs.scheduledSeqs[seq]
+	return ok
 }
 
 func (bs *BlockDependencyManager) ScheduleTaskWithDependencies(task Task, blockSeq uint64, prev *Digest, emptyRounds []uint64) error {
@@ -144,6 +143,9 @@ func (bs *BlockDependencyManager) ScheduleTaskWithDependencies(task Task, blockS
 
 	wrappedTask := func() Digest {
 		id := task()
+		bs.lock.Lock()
+		delete(bs.scheduledSeqs, blockSeq)
+		bs.lock.Unlock()
 		bs.ExecuteBlockDependents(id)
 		return id
 	}
@@ -154,6 +156,8 @@ func (bs *BlockDependencyManager) ScheduleTaskWithDependencies(task Task, blockS
 		return fmt.Errorf("%w: %d pending verifications (max %d)", ErrTooManyPendingVerifications, totalSize, bs.maxDeps)
 	}
 
+	bs.scheduledSeqs[blockSeq] = struct{}{}
+
 	if prev == nil && len(emptyRounds) == 0 {
 		bs.logger.Debug("Scheduling block verification task with no dependencies", zap.Uint64("blockSeq", blockSeq))
 		bs.scheduler.Schedule(wrappedTask)
@@ -185,6 +189,7 @@ func (bs *BlockDependencyManager) RemoveOldTasks(seq uint64) {
 	for _, taskWithDeps := range bs.dependencies {
 		if taskWithDeps.blockSeq <= seq {
 			bs.logger.Debug("Removing block verification task as its block seq is less than or equal to finalized seq", zap.Uint64("blockSeq", taskWithDeps.blockSeq), zap.Uint64("finalizedSeq", seq))
+			delete(bs.scheduledSeqs, taskWithDeps.blockSeq)
 			continue
 		}
 		remainingDeps = append(remainingDeps, taskWithDeps)

common/block_scheduler_test.go

@@ -223,6 +223,30 @@ func TestBlockVerificationScheduler(t *testing.T) {
 		waitReceive(t, done2)
 	})
 
+	t.Run("IsSequenceScheduled covers queued and running tasks", func(t *testing.T) {
+		scheduler := NewScheduler(noopLogger{}, defaultMaxDeps)
+		bvs := NewBlockVerificationScheduler(noopLogger{}, defaultMaxDeps, scheduler)
+		defer bvs.Close()
+
+		started := make(chan struct{}, 1)
+		release := make(chan struct{})
+		task := func() Digest {
+			started <- struct{}{}
+			<-release
+			return makeDigest(t)
+		}
+
+		require.False(t, bvs.IsSequenceScheduled(3))
+		require.NoError(t, bvs.ScheduleTaskWithDependencies(task, 3, nil, nil))
+
+		// Running with no dependencies still counts as scheduled.
+		waitReceive(t, started)
+		require.True(t, bvs.IsSequenceScheduled(3))
+
+		close(release)
+		require.Eventually(t, func() bool { return !bvs.IsSequenceScheduled(3) }, defaultWaitDuration, 10*time.Millisecond)
+	})
+
 	t.Run("RemoveOldTasks removes tasks with blockSeq <= finalized seq", func(t *testing.T) {
 		scheduler := NewScheduler(noopLogger{}, defaultMaxDeps)
 		bvs := NewBlockVerificationScheduler(noopLogger{}, defaultMaxDeps, scheduler)
@@ -257,6 +281,8 @@ func TestBlockVerificationScheduler(t *testing.T) {
 
 		// Finalize up to seq=6 — this should remove the old task (seq=5) but keep the new one (seq=8).
 		bvs.RemoveOldTasks(6)
+		require.False(t, bvs.IsSequenceScheduled(oldSeq))
+		require.True(t, bvs.IsSequenceScheduled(newSeq))
 
 		// Now resolve the dependency round. Only the "new" task should execute.
 		bvs.ExecuteEmptyRoundDependents(depRound)

nonvalidator/non_validator_test.go

@@ -9,6 +9,7 @@ import (
 	"fmt"
 	"slices"
 	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -1251,3 +1252,54 @@ func TestNonValidatorDropsQuorumRoundPastSequenceWindow(t *testing.T) {
 		"indexed a block that was past the sequence window when it was received",
 	)
 }
+
+// TestNonValidatorIgnoresReplayedQuorumRoundWhileVerifying asserts that replaying the finalized
+// quorum round for nextSeqToCommit while its verification task is in flight does not schedule
+// duplicate tasks.
+func TestNonValidatorIgnoresReplayedQuorumRoundWhileVerifying(t *testing.T) {
+	tc := newSeededChain(t, testNodes, 2)
+	storage := tc.CloneUntil(2)
+
+	block, finalization, err := tc.Retrieve(2)
+	require.NoError(t, err)
+
+	// Hold the first verification so the replays arrive while its task is still running.
+	release := make(chan struct{})
+	var verifications atomic.Int32
+	tb := block.(*testutil.TestBlock)
+	tb.VerificationError = errors.New("verification failed")
+	tb.OnVerify = func() {
+		if verifications.Add(1) == 1 {
+			<-release
+		}
+	}
+
+	nv, err := NewNonValidator(
+		Config{
+			Storage:                    storage,
+			Comm:                       testutil.NewNoopComm(testNodes.NodeIDs()),
+			Logger:                     testutil.MakeLogger(t, 1),
+			SignatureAggregatorCreator: tc.signatureAggregatorCreator,
+			MaxSequenceWindow:          5,
+			ID:                         common.NodeID{16},
+			StartTime:                  time.Now(),
+		},
+	)
+	require.NoError(t, err)
+	nv.Start()
+	defer nv.Stop()
+
+	// Send many of the same responses that
+	for range 10 {
+		require.NoError(t, nv.HandleMessage(
+			&common.Message{ReplicationResponse: &common.ReplicationResponse{
+				Data: []common.QuorumRound{{Block: tb, Finalization: &finalization}},
+			}},
+			common.NodeID{42},
+		))
+	}
+	close(release)
+
+	require.Never(t, func() bool { return verifications.Load() > 1 }, time.Second, 10*time.Millisecond,
+		"a replayed quorum round scheduled a duplicate verification task")
+}