From 772d6012790b7b0508e9adb234d19ff884216dae Mon Sep 17 00:00:00 2001 From: samliok Date: Tue, 4 Aug 2026 17:02:26 -0400 Subject: [PATCH 1/3] Make the testVM pause gate resumable pause/resume flipped an atomic that BuildBlock and WaitForPendingBlock read once on entry before blocking on ctx.Done() alone, so resume could not release a caller already parked in the gate. pause now creates a channel that resume closes, and both methods select on it alongside ctx.Done(). --- instance_test.go | 102 +++++++++++++++++++++++++++++++++++++++++++---- 1 file changed, 94 insertions(+), 8 deletions(-) diff --git a/instance_test.go b/instance_test.go index cfada105..44cc3468 100644 --- a/instance_test.go +++ b/instance_test.go @@ -368,6 +368,59 @@ func TestInstanceRestartAcrossEpochs(t *testing.T) { waitForNumBlocks(t, storage, storage.NumBlocks()+2) } +// TestTestVMResumeWakesPausedCallers asserts that resume() releases callers already +// parked in the testVM pause gate, while paused still produces no block. The gate +// sampled the paused flag once and then waited only on ctx.Done(), so a caller that +// entered before resume() stayed parked until its context was cancelled, which never +// happens while the round it is stalling cannot advance. +func TestTestVMResumeWakesPausedCallers(t *testing.T) { + vm := newTestVM() + vm.pause() + + waited := make(chan struct{}) + go func() { + vm.WaitForPendingBlock(t.Context()) + close(waited) + }() + + type buildResult struct { + blk avalanchego.VMBlock + err error + } + built := make(chan buildResult, 1) + go func() { + blk, err := vm.BuildBlock(t.Context(), 0) + built <- buildResult{blk: blk, err: err} + }() + + // Long enough for both callers to enter the gate, and for the unpaused + // WaitForPendingBlock path (100ms) to have returned had the gate not held. + time.Sleep(200 * time.Millisecond) + select { + case <-waited: + require.Fail(t, "WaitForPendingBlock returned while paused") + case r := <-built: + require.Fail(t, "BuildBlock returned while paused", "block %v err %v", r.blk, r.err) + default: + } + + vm.resume() + + select { + case <-waited: + case <-time.After(3 * time.Second): + require.Fail(t, "WaitForPendingBlock did not return after resume") + } + + select { + case r := <-built: + require.NoError(t, r.err) + require.NotNil(t, r.blk) + case <-time.After(3 * time.Second): + require.Fail(t, "BuildBlock did not return after resume") + } +} + func TestParseBlockSizeMatchesBytes(t *testing.T) { // Case 1: Bytes() first, Size() second, size returns the cached length. pb := &ParsedBlock{ @@ -674,7 +727,13 @@ type testVM struct { // machinery, which builds its block once the inner build times out, still runs — // so pausing before an epoch change leaves the sealing block at the tip with // nothing built on top. Lets a test pin the chain tip without touching storage. - paused atomic.Bool + // + // resume closes resumed, which releases callers already parked in the gate. A + // caller must wake on resume and not only on ctx cancellation: the epoch cancels + // that context on a round transition, and a round waiting for a block to build + // cannot transition, so the parked caller would never be released. + lock sync.Mutex + resumed chan struct{} // non-nil while paused } func newTestVM() *testVM { @@ -683,13 +742,37 @@ func newTestVM() *testVM { return vm } -func (vm *testVM) pause() { vm.paused.Store(true) } -func (vm *testVM) resume() { vm.paused.Store(false) } +func (vm *testVM) pause() { + vm.lock.Lock() + defer vm.lock.Unlock() + if vm.resumed == nil { + vm.resumed = make(chan struct{}) + } +} + +func (vm *testVM) resume() { + vm.lock.Lock() + defer vm.lock.Unlock() + if vm.resumed != nil { + close(vm.resumed) + vm.resumed = nil + } +} + +// pauseGate returns the channel the next resume closes, or nil if not paused. +func (vm *testVM) pauseGate() <-chan struct{} { + vm.lock.Lock() + defer vm.lock.Unlock() + return vm.resumed +} func (vm *testVM) BuildBlock(ctx context.Context, _ uint64) (avalanchego.VMBlock, error) { - if vm.paused.Load() { - <-ctx.Done() // let the caller's impatient build time out - return nil, ctx.Err() + if gate := vm.pauseGate(); gate != nil { + select { + case <-gate: + case <-ctx.Done(): // let the caller's impatient build time out + return nil, ctx.Err() + } } h := vm.nextHeight.Add(1) - 1 payload := make([]byte, 8) @@ -698,8 +781,11 @@ func (vm *testVM) BuildBlock(ctx context.Context, _ uint64) (avalanchego.VMBlock } func (vm *testVM) WaitForPendingBlock(ctx context.Context) { - if vm.paused.Load() { - <-ctx.Done() // no pending block while paused + if gate := vm.pauseGate(); gate != nil { + select { + case <-gate: // resumed, a block is pending again + case <-ctx.Done(): // no pending block while paused + } return } select { From c438f62f2e2bfa986768b585734b12a61945f488 Mon Sep 17 00:00:00 2001 From: samliok Date: Tue, 4 Aug 2026 17:52:11 -0400 Subject: [PATCH 2/3] fix TestNetworkSimpleFuzz: sweep the rounds map when replication commits a block createFinalizedBlockVerificationTask indexed only the block it verified, so a finalization stored while its sequence was ahead of the storage was never revisited once replication caught up. The node then held a finalized block it could not commit, and any peer that answered with that sequence drove processFinalizedBlock into storeFinalization on a round that already had a finalization, whose error reaches HandleMessage. Index from the round instead, matching persistFinalization. --- simplex/epoch.go | 4 ++- simplex/replication_test.go | 69 +++++++++++++++++++++++++++++++++++++ 2 files changed, 72 insertions(+), 1 deletion(-) diff --git a/simplex/epoch.go b/simplex/epoch.go index 2fcccfa8..b244164f 100644 --- a/simplex/epoch.go +++ b/simplex/epoch.go @@ -2232,7 +2232,9 @@ func (e *Epoch) createFinalizedBlockVerificationTask(block common.Block, finaliz zap.Uint64("seq", md.Seq), zap.Stringer("digest", md.Digest)) - if err := e.indexFinalization(verifiedBlock, *finalization); err != nil { + // Sweep from this round, not just this block. Later rounds may already hold a + // finalization stored while their sequence was ahead of the storage. + if err := e.indexFinalizations(md.Round); err != nil { e.haltedError = err e.Logger.Error("Failed to index finalization", zap.Error(err)) return md.Digest diff --git a/simplex/replication_test.go b/simplex/replication_test.go index a9050341..d2e35847 100644 --- a/simplex/replication_test.go +++ b/simplex/replication_test.go @@ -1940,3 +1940,72 @@ func TestLeaderStartsRoundAfterReplicatedQuorumRound(t *testing.T) { }) } } + +// setupStrandedFinalization returns a node whose storage holds seq 0 and whose WAL holds the +// block and finalization for round 2 (seq 2), which recovery restores into the rounds map +// without indexing. Only seq 1 is genuinely missing. +func setupStrandedFinalization(t *testing.T, nodes []common.NodeID) (*simplex.Epoch, *InMemStorage, []common.VerifiedFinalizedBlock) { + ctx := context.Background() + blocks := createBlocks(t, nodes, 3) + + conf, wal, storage := DefaultTestNodeEpochConfig(t, nodes[3], NewNoopComm(nodes), testutil.NewTestBlockBuilder()) + conf.ReplicationEnabled = true + require.NoError(t, storage.Index(ctx, blocks[0].VerifiedBlock, blocks[0].Finalization)) + + third := blocks[2].VerifiedBlock + thirdBytes, err := third.Bytes() + require.NoError(t, err) + blockRecord, err := common.BlockRecord(third.BlockHeader(), thirdBytes) + require.NoError(t, err) + require.NoError(t, wal.Append(blockRecord)) + _, finalizationRecord := NewFinalizationRecord(t, &TestSignatureAggregator{N: len(nodes)}, third, nodes) + require.NoError(t, wal.Append(finalizationRecord)) + + e, err := simplex.NewEpoch(conf) + require.NoError(t, err) + t.Cleanup(e.Stop) + require.NoError(t, e.Start()) + require.Equal(t, uint64(1), storage.NumBlocks()) + + return e, storage, blocks +} + +func replicateSeq(block common.VerifiedFinalizedBlock) *common.Message { + return &common.Message{ReplicationResponse: &common.ReplicationResponse{ + Data: []common.QuorumRound{{ + Block: block.VerifiedBlock.(common.Block), + Finalization: &block.Finalization, + }}, + }} +} + +// TestReplicationIndexesStrandedFinalization asserts a node commits a finalization it already +// holds once storage reaches its sequence. The replicated commit path indexed only its own +// block (epoch.go:2235), so a finalization stored while it was ahead of storage was never +// revisited and the node stalled holding everything it needed. +func TestReplicationIndexesStrandedFinalization(t *testing.T) { + nodes := []common.NodeID{{1}, {2}, {3}, {4}} + e, storage, blocks := setupStrandedFinalization(t, nodes) + + // filling the one real gap should commit seq 1 and then seq 2 from the rounds map + require.NoError(t, e.HandleMessage(replicateSeq(blocks[1]), nodes[1])) + storage.WaitForBlockCommit(1) + require.Eventually(t, func() bool { return storage.NumBlocks() == 3 }, + 5*time.Second, 10*time.Millisecond, + "seq 2 never indexed, though the node holds its block and finalization") +} + +// TestReplicationRedeliversStoredFinalization asserts a replication response carrying a +// finalization the round already holds is not an error. processFinalizedBlock propagates +// storeFinalization's "already has a finalization" error (epoch.go:2009), which reached +// HandleMessage whenever a peer answered with a sequence the node had stranded. +func TestReplicationRedeliversStoredFinalization(t *testing.T) { + nodes := []common.NodeID{{1}, {2}, {3}, {4}} + e, storage, blocks := setupStrandedFinalization(t, nodes) + + require.NoError(t, e.HandleMessage(replicateSeq(blocks[1]), nodes[1])) + storage.WaitForBlockCommit(1) + + // a peer answering with seq 2, whose finalization round 2 already holds, must not fail + require.NoError(t, e.HandleMessage(replicateSeq(blocks[2]), nodes[1])) +} From a49e49446fe3abea694ab7441a38b9ed70adbc2d Mon Sep 17 00:00:00 2001 From: samliok Date: Tue, 4 Aug 2026 18:03:08 -0400 Subject: [PATCH 3/3] Revert "fix TestNetworkSimpleFuzz: sweep the rounds map when replication commits a block" This reverts commit c438f62f2e2bfa986768b585734b12a61945f488. --- simplex/epoch.go | 4 +-- simplex/replication_test.go | 69 ------------------------------------- 2 files changed, 1 insertion(+), 72 deletions(-) diff --git a/simplex/epoch.go b/simplex/epoch.go index b244164f..2fcccfa8 100644 --- a/simplex/epoch.go +++ b/simplex/epoch.go @@ -2232,9 +2232,7 @@ func (e *Epoch) createFinalizedBlockVerificationTask(block common.Block, finaliz zap.Uint64("seq", md.Seq), zap.Stringer("digest", md.Digest)) - // Sweep from this round, not just this block. Later rounds may already hold a - // finalization stored while their sequence was ahead of the storage. - if err := e.indexFinalizations(md.Round); err != nil { + if err := e.indexFinalization(verifiedBlock, *finalization); err != nil { e.haltedError = err e.Logger.Error("Failed to index finalization", zap.Error(err)) return md.Digest diff --git a/simplex/replication_test.go b/simplex/replication_test.go index d2e35847..a9050341 100644 --- a/simplex/replication_test.go +++ b/simplex/replication_test.go @@ -1940,72 +1940,3 @@ func TestLeaderStartsRoundAfterReplicatedQuorumRound(t *testing.T) { }) } } - -// setupStrandedFinalization returns a node whose storage holds seq 0 and whose WAL holds the -// block and finalization for round 2 (seq 2), which recovery restores into the rounds map -// without indexing. Only seq 1 is genuinely missing. -func setupStrandedFinalization(t *testing.T, nodes []common.NodeID) (*simplex.Epoch, *InMemStorage, []common.VerifiedFinalizedBlock) { - ctx := context.Background() - blocks := createBlocks(t, nodes, 3) - - conf, wal, storage := DefaultTestNodeEpochConfig(t, nodes[3], NewNoopComm(nodes), testutil.NewTestBlockBuilder()) - conf.ReplicationEnabled = true - require.NoError(t, storage.Index(ctx, blocks[0].VerifiedBlock, blocks[0].Finalization)) - - third := blocks[2].VerifiedBlock - thirdBytes, err := third.Bytes() - require.NoError(t, err) - blockRecord, err := common.BlockRecord(third.BlockHeader(), thirdBytes) - require.NoError(t, err) - require.NoError(t, wal.Append(blockRecord)) - _, finalizationRecord := NewFinalizationRecord(t, &TestSignatureAggregator{N: len(nodes)}, third, nodes) - require.NoError(t, wal.Append(finalizationRecord)) - - e, err := simplex.NewEpoch(conf) - require.NoError(t, err) - t.Cleanup(e.Stop) - require.NoError(t, e.Start()) - require.Equal(t, uint64(1), storage.NumBlocks()) - - return e, storage, blocks -} - -func replicateSeq(block common.VerifiedFinalizedBlock) *common.Message { - return &common.Message{ReplicationResponse: &common.ReplicationResponse{ - Data: []common.QuorumRound{{ - Block: block.VerifiedBlock.(common.Block), - Finalization: &block.Finalization, - }}, - }} -} - -// TestReplicationIndexesStrandedFinalization asserts a node commits a finalization it already -// holds once storage reaches its sequence. The replicated commit path indexed only its own -// block (epoch.go:2235), so a finalization stored while it was ahead of storage was never -// revisited and the node stalled holding everything it needed. -func TestReplicationIndexesStrandedFinalization(t *testing.T) { - nodes := []common.NodeID{{1}, {2}, {3}, {4}} - e, storage, blocks := setupStrandedFinalization(t, nodes) - - // filling the one real gap should commit seq 1 and then seq 2 from the rounds map - require.NoError(t, e.HandleMessage(replicateSeq(blocks[1]), nodes[1])) - storage.WaitForBlockCommit(1) - require.Eventually(t, func() bool { return storage.NumBlocks() == 3 }, - 5*time.Second, 10*time.Millisecond, - "seq 2 never indexed, though the node holds its block and finalization") -} - -// TestReplicationRedeliversStoredFinalization asserts a replication response carrying a -// finalization the round already holds is not an error. processFinalizedBlock propagates -// storeFinalization's "already has a finalization" error (epoch.go:2009), which reached -// HandleMessage whenever a peer answered with a sequence the node had stranded. -func TestReplicationRedeliversStoredFinalization(t *testing.T) { - nodes := []common.NodeID{{1}, {2}, {3}, {4}} - e, storage, blocks := setupStrandedFinalization(t, nodes) - - require.NoError(t, e.HandleMessage(replicateSeq(blocks[1]), nodes[1])) - storage.WaitForBlockCommit(1) - - // a peer answering with seq 2, whose finalization round 2 already holds, must not fail - require.NoError(t, e.HandleMessage(replicateSeq(blocks[2]), nodes[1])) -}