mirror of
https://github.com/luxfi/consensus.git
synced 2026-09-24 00:26:58 +00:00
fix: BuildBlock death spiral — don't decrement on error, force preference after Accept failure
Q-01: BuildBlock errors were silently swallowed and the pendingBuildBlocks counter decremented, causing permanent block production halt. Now the counter is retained on error for retry on next tick. Q-02: SetPreference failures after Accept caused VM/consensus state divergence. All Accept paths now check SetPreference errors and call ForcePreference to keep consensus aligned with the accepted block.
This commit is contained in:
@@ -613,8 +613,8 @@ func TestTransitiveNotifyPendingTxsWithVMError(t *testing.T) {
|
||||
err := engine.Notify(ctx, msg)
|
||||
require.NoError(err) // Errors are swallowed, not fatal
|
||||
|
||||
// Pending builds should be cleared after attempt
|
||||
require.Zero(engine.PendingBuildBlocks())
|
||||
// Pending builds must NOT be cleared on error — retained for retry (Q-01 fix)
|
||||
require.Equal(1, engine.PendingBuildBlocks())
|
||||
}
|
||||
|
||||
// TestTransitiveNotifyStateSyncDone tests Notify with StateSyncDone message
|
||||
|
||||
@@ -282,6 +282,18 @@ func (c *ChainConsensus) SyncState(lastAcceptedID ids.ID, height uint64) {
|
||||
}
|
||||
}
|
||||
|
||||
// ForcePreference forces the consensus preferred tip to the given block.
|
||||
// This is a recovery mechanism used when SetPreference fails after Accept —
|
||||
// without it, the VM and consensus engine disagree on the chain tip, causing
|
||||
// a state divergence death spiral.
|
||||
func (c *ChainConsensus) ForcePreference(blockID ids.ID) {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
c.finalizedTip = blockID
|
||||
// Ensure the forced block is a tip
|
||||
c.tips[blockID] = true
|
||||
}
|
||||
|
||||
// GetFinalizedTip returns the current finalized tip block ID.
|
||||
// This is useful for debugging and health checks.
|
||||
func (c *ChainConsensus) GetFinalizedTip() ids.ID {
|
||||
|
||||
+60
-4
@@ -11,6 +11,7 @@ import (
|
||||
"github.com/luxfi/consensus/engine"
|
||||
"github.com/luxfi/consensus/engine/chain/block"
|
||||
"github.com/luxfi/ids"
|
||||
"github.com/luxfi/log"
|
||||
)
|
||||
|
||||
// -----------------------------------------------------------------------------
|
||||
@@ -182,6 +183,13 @@ func WithSlashing(detector *slashing.Detector, db *slashing.DB) Option {
|
||||
}
|
||||
}
|
||||
|
||||
// WithLogger sets the engine logger.
|
||||
func WithLogger(l log.Logger) Option {
|
||||
return func(t *Transitive) {
|
||||
t.log = l
|
||||
}
|
||||
}
|
||||
|
||||
// WithVoteBuffers sets channel buffer sizes.
|
||||
func WithVoteBuffers(requests, votes int) Option {
|
||||
return func(t *Transitive) {
|
||||
@@ -247,6 +255,9 @@ type Transitive struct {
|
||||
// Slashing: equivocation detection (optional, nil disables)
|
||||
slashingDetector *slashing.Detector
|
||||
slashingDB *slashing.DB
|
||||
|
||||
// Logger for consensus events (nil-safe: uses log.Noop() if unset)
|
||||
log log.Logger
|
||||
}
|
||||
|
||||
// New creates an engine with default parameters.
|
||||
@@ -298,6 +309,10 @@ func NewWithConfig(cfg Config, opts ...Option) *Transitive {
|
||||
opt(t)
|
||||
}
|
||||
|
||||
if t.log == nil {
|
||||
t.log = log.Noop()
|
||||
}
|
||||
|
||||
return t
|
||||
}
|
||||
|
||||
@@ -747,7 +762,14 @@ func (t *Transitive) processPendingBlocks() {
|
||||
action.vmBlock.Accept(ctx)
|
||||
}
|
||||
if vm != nil {
|
||||
vm.SetPreference(ctx, action.blockID)
|
||||
if err := vm.SetPreference(ctx, action.blockID); err != nil {
|
||||
t.log.Crit("SetPreference failed after Accept — forcing preference",
|
||||
"blockID", action.blockID,
|
||||
"error", err)
|
||||
// Force consensus to match the accepted block so the
|
||||
// engine and VM don't diverge on preferred tip.
|
||||
t.consensus.ForcePreference(action.blockID)
|
||||
}
|
||||
}
|
||||
} else {
|
||||
if action.vmBlock != nil {
|
||||
@@ -863,6 +885,19 @@ func (t *Transitive) tryFinalizeBlock(ctx context.Context, blockID ids.ID) {
|
||||
_ = pending.VMBlock.Accept(ctx)
|
||||
}
|
||||
|
||||
// SetPreference must follow Accept to keep VM and consensus in sync.
|
||||
t.mu.RLock()
|
||||
vm := t.vm
|
||||
t.mu.RUnlock()
|
||||
if vm != nil {
|
||||
if err := vm.SetPreference(ctx, blockID); err != nil {
|
||||
t.log.Crit("SetPreference failed after Accept — forcing preference",
|
||||
"blockID", blockID,
|
||||
"error", err)
|
||||
t.consensus.ForcePreference(blockID)
|
||||
}
|
||||
}
|
||||
|
||||
// Pipeline: block finalized → immediately build next
|
||||
t.signalPipeline()
|
||||
}
|
||||
@@ -888,10 +923,22 @@ func (t *Transitive) DrainAccepted(ctx context.Context) {
|
||||
}
|
||||
t.mu.Unlock()
|
||||
|
||||
t.mu.RLock()
|
||||
vm := t.vm
|
||||
t.mu.RUnlock()
|
||||
|
||||
for _, a := range toAccept {
|
||||
if a.blk != nil {
|
||||
_ = a.blk.Accept(ctx)
|
||||
}
|
||||
if vm != nil {
|
||||
if err := vm.SetPreference(ctx, a.id); err != nil {
|
||||
t.log.Crit("SetPreference failed after Accept — forcing preference",
|
||||
"blockID", a.id,
|
||||
"error", err)
|
||||
t.consensus.ForcePreference(a.id)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if len(toAccept) > 0 {
|
||||
@@ -905,12 +952,16 @@ func (t *Transitive) buildBlocksLocked(ctx context.Context) error {
|
||||
}
|
||||
|
||||
for t.pendingBuildBlocks > 0 {
|
||||
t.pendingBuildBlocks--
|
||||
|
||||
vmBlock, err := t.vm.BuildBlock(ctx)
|
||||
if err != nil {
|
||||
t.log.Error("BuildBlock failed, will retry next tick",
|
||||
"error", err,
|
||||
"pendingBuildBlocks", t.pendingBuildBlocks)
|
||||
// Do NOT decrement pendingBuildBlocks — the request is still
|
||||
// outstanding and will be retried on the next Notify or pipeline tick.
|
||||
return nil
|
||||
}
|
||||
t.pendingBuildBlocks--
|
||||
|
||||
t.blocksBuilt++
|
||||
|
||||
@@ -956,7 +1007,12 @@ func (t *Transitive) buildBlocksLocked(ctx context.Context) error {
|
||||
t.mu.Unlock()
|
||||
_ = vmBlock.Accept(ctx)
|
||||
if t.vm != nil {
|
||||
_ = t.vm.SetPreference(ctx, vmBlock.ID())
|
||||
if err := t.vm.SetPreference(ctx, vmBlock.ID()); err != nil {
|
||||
t.log.Crit("SetPreference failed after Accept — forcing preference",
|
||||
"blockID", vmBlock.ID(),
|
||||
"error", err)
|
||||
t.consensus.ForcePreference(vmBlock.ID())
|
||||
}
|
||||
}
|
||||
t.mu.Lock()
|
||||
} else {
|
||||
|
||||
@@ -1394,3 +1394,206 @@ func TestSlashing_SlashingDBAccessor(t *testing.T) {
|
||||
t.Fatal("SlashingDB() should return nil when slashing disabled")
|
||||
}
|
||||
}
|
||||
|
||||
// =============================================================================
|
||||
// Q-01: BuildBlock error must NOT decrement pendingBuildBlocks
|
||||
// =============================================================================
|
||||
|
||||
// failingBuildVM fails BuildBlock N times then succeeds.
|
||||
type failingBuildVM struct {
|
||||
failCount int
|
||||
callCount int
|
||||
blocks []*mockBlock
|
||||
preferenceID ids.ID
|
||||
setPreferenceErr error
|
||||
}
|
||||
|
||||
func (m *failingBuildVM) BuildBlock(_ context.Context) (block.Block, error) {
|
||||
m.callCount++
|
||||
if m.callCount <= m.failCount {
|
||||
return nil, errors.New("transient BuildBlock error")
|
||||
}
|
||||
blk := &mockBlock{
|
||||
id: ids.GenerateTestID(),
|
||||
height: uint64(m.callCount),
|
||||
}
|
||||
m.blocks = append(m.blocks, blk)
|
||||
return blk, nil
|
||||
}
|
||||
|
||||
func (m *failingBuildVM) GetBlock(_ context.Context, id ids.ID) (block.Block, error) {
|
||||
for _, b := range m.blocks {
|
||||
if b.id == id {
|
||||
return b, nil
|
||||
}
|
||||
}
|
||||
return nil, errors.New("not found")
|
||||
}
|
||||
|
||||
func (m *failingBuildVM) ParseBlock(_ context.Context, b []byte) (block.Block, error) {
|
||||
return &mockBlock{bytes: b}, nil
|
||||
}
|
||||
|
||||
func (m *failingBuildVM) LastAccepted(_ context.Context) (ids.ID, error) {
|
||||
return m.preferenceID, nil
|
||||
}
|
||||
|
||||
func (m *failingBuildVM) SetPreference(_ context.Context, id ids.ID) error {
|
||||
if m.setPreferenceErr != nil {
|
||||
return m.setPreferenceErr
|
||||
}
|
||||
m.preferenceID = id
|
||||
return nil
|
||||
}
|
||||
|
||||
// TestBuildBlockError_DoesNotDecrementPending verifies Q-01: BuildBlock error
|
||||
// must not decrement pendingBuildBlocks so the build is retried.
|
||||
func TestBuildBlockError_DoesNotDecrementPending(t *testing.T) {
|
||||
vm := &failingBuildVM{failCount: 1}
|
||||
eng := New(WithVM(vm))
|
||||
|
||||
if err := eng.Start(context.Background(), true); err != nil {
|
||||
t.Fatalf("Start: %v", err)
|
||||
}
|
||||
defer eng.Stop(context.Background())
|
||||
|
||||
// Send PendingTxs — BuildBlock will fail on first attempt
|
||||
if err := eng.Notify(context.Background(), Message{Type: PendingTxs}); err != nil {
|
||||
t.Fatalf("Notify: %v", err)
|
||||
}
|
||||
|
||||
// BuildBlock was called once (and failed)
|
||||
if vm.callCount != 1 {
|
||||
t.Fatalf("expected 1 BuildBlock call, got %d", vm.callCount)
|
||||
}
|
||||
|
||||
// pendingBuildBlocks must still be 1 — NOT decremented on error
|
||||
if got := eng.PendingBuildBlocks(); got != 1 {
|
||||
t.Fatalf("expected pendingBuildBlocks=1 after error, got %d", got)
|
||||
}
|
||||
}
|
||||
|
||||
// TestBuildBlockError_RetriesOnNextNotify verifies that after a BuildBlock
|
||||
// error, the next Notify successfully builds the block (retry).
|
||||
func TestBuildBlockError_RetriesOnNextNotify(t *testing.T) {
|
||||
vm := &failingBuildVM{failCount: 1}
|
||||
eng := New(WithVM(vm))
|
||||
|
||||
if err := eng.Start(context.Background(), true); err != nil {
|
||||
t.Fatalf("Start: %v", err)
|
||||
}
|
||||
defer eng.Stop(context.Background())
|
||||
|
||||
// First Notify: BuildBlock fails, counter stays at 1
|
||||
eng.Notify(context.Background(), Message{Type: PendingTxs})
|
||||
if vm.callCount != 1 {
|
||||
t.Fatalf("expected 1 call after first Notify, got %d", vm.callCount)
|
||||
}
|
||||
if eng.PendingBuildBlocks() != 1 {
|
||||
t.Fatalf("expected pendingBuildBlocks=1 after error, got %d", eng.PendingBuildBlocks())
|
||||
}
|
||||
|
||||
// Second Notify: adds another pending + retries. failCount=1 means second
|
||||
// call succeeds. The loop processes both pending (the retained one + new one).
|
||||
eng.Notify(context.Background(), Message{Type: PendingTxs})
|
||||
|
||||
// Second call succeeded, third call is the second pending item.
|
||||
// Call 2 succeeded (first pending consumed), call 3 succeeded (second pending consumed).
|
||||
if vm.callCount < 2 {
|
||||
t.Fatalf("expected at least 2 BuildBlock calls after retry, got %d", vm.callCount)
|
||||
}
|
||||
|
||||
// All pending should be consumed now
|
||||
if got := eng.PendingBuildBlocks(); got != 0 {
|
||||
t.Fatalf("expected pendingBuildBlocks=0 after successful retry, got %d", got)
|
||||
}
|
||||
}
|
||||
|
||||
// TestBuildBlockError_RepeatedFailures_NoPermanentHalt verifies that repeated
|
||||
// BuildBlock failures don't permanently halt block production. Each Notify
|
||||
// retains the counter, so future Notifys always attempt to build.
|
||||
func TestBuildBlockError_RepeatedFailures_NoPermanentHalt(t *testing.T) {
|
||||
vm := &failingBuildVM{failCount: 100} // fails 100 times
|
||||
eng := New(WithVM(vm))
|
||||
|
||||
if err := eng.Start(context.Background(), true); err != nil {
|
||||
t.Fatalf("Start: %v", err)
|
||||
}
|
||||
defer eng.Stop(context.Background())
|
||||
|
||||
// Send 5 notifications — all will fail but counter grows
|
||||
for i := 0; i < 5; i++ {
|
||||
eng.Notify(context.Background(), Message{Type: PendingTxs})
|
||||
}
|
||||
|
||||
// pendingBuildBlocks should be 5 — none consumed because all failed
|
||||
if got := eng.PendingBuildBlocks(); got != 5 {
|
||||
t.Fatalf("expected pendingBuildBlocks=5 after 5 failures, got %d", got)
|
||||
}
|
||||
|
||||
// Now make BuildBlock succeed (lower failCount below callCount)
|
||||
vm.failCount = vm.callCount // next call will succeed
|
||||
|
||||
// Next Notify adds one more pending (total 6) and retries — all 6 should build
|
||||
eng.Notify(context.Background(), Message{Type: PendingTxs})
|
||||
|
||||
if got := eng.PendingBuildBlocks(); got != 0 {
|
||||
t.Fatalf("expected pendingBuildBlocks=0 after recovery, got %d", got)
|
||||
}
|
||||
|
||||
if len(vm.blocks) < 6 {
|
||||
t.Fatalf("expected at least 6 blocks built after recovery, got %d", len(vm.blocks))
|
||||
}
|
||||
}
|
||||
|
||||
// =============================================================================
|
||||
// Q-02: SetPreference failure after Accept must force preference
|
||||
// =============================================================================
|
||||
|
||||
// TestSetPreferenceFailure_ForcesPreference verifies that when SetPreference
|
||||
// fails after Accept, the consensus engine forces its preference to match the
|
||||
// accepted block, preventing state divergence.
|
||||
func TestSetPreferenceFailure_ForcesPreference(t *testing.T) {
|
||||
vm := &failingBuildVM{
|
||||
failCount: 0, // BuildBlock always succeeds
|
||||
setPreferenceErr: errors.New("SetPreference broken"),
|
||||
}
|
||||
|
||||
// K=1 mode: self-vote causes immediate acceptance in buildBlocksLocked
|
||||
eng := New(
|
||||
WithVM(vm),
|
||||
WithParams(config.Parameters{
|
||||
K: 1,
|
||||
AlphaPreference: 1,
|
||||
AlphaConfidence: 1,
|
||||
Beta: 1,
|
||||
}),
|
||||
)
|
||||
|
||||
if err := eng.Start(context.Background(), true); err != nil {
|
||||
t.Fatalf("Start: %v", err)
|
||||
}
|
||||
defer eng.Stop(context.Background())
|
||||
|
||||
// Trigger block build — will be immediately accepted (K=1),
|
||||
// then SetPreference will fail, and ForcePreference must be called.
|
||||
eng.Notify(context.Background(), Message{Type: PendingTxs})
|
||||
|
||||
// Block was built and accepted
|
||||
if len(vm.blocks) != 1 {
|
||||
t.Fatalf("expected 1 block built, got %d", len(vm.blocks))
|
||||
}
|
||||
|
||||
acceptedBlockID := vm.blocks[0].id
|
||||
|
||||
// Consensus preference must match the accepted block despite SetPreference error
|
||||
if pref := eng.Preference(); pref != acceptedBlockID {
|
||||
t.Fatalf("expected preference=%s after forced preference, got %s",
|
||||
acceptedBlockID, pref)
|
||||
}
|
||||
|
||||
// Verify consensus internal state agrees
|
||||
if tip := eng.consensus.GetFinalizedTip(); tip != acceptedBlockID {
|
||||
t.Fatalf("expected finalizedTip=%s, got %s", acceptedBlockID, tip)
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user