refactor: Block.WaitUntilExecuted instead of synchronous enqueue

Author: ARR4N

block.go

@@ -43,16 +43,20 @@ func (vm *VM) AcceptBlock(ctx context.Context, b *blocks.Block) error {
 	vm.last.settled.Store(b.LastSettled())
 	vm.last.accepted.Store(b)
 
+	// This MUST NOT happen before the database and [VM.last] are updated to
+	// reflect that the block has been accepted.
+	if err := vm.exec.EnqueueAccepted(ctx, b); err != nil {
+		return err
+	}
 	// When the chain is bootstrapping, avalanchego expects to be able to call
 	// `Verify` and `Accept` in a loop over blocks. Reporting an error during
 	// either `Verify` or `Accept` is considered FATAL during this process.
 	// Therefore, we must ensure that avalanchego does not get too far ahead of
 	// the execution thread and FATAL during block verification.
-	execSynchronous := vm.consensusState.Get() == snow.Bootstrapping
-	// This MUST NOT happen before the database and [VM.last] are updated to
-	// reflect that the block has been accepted.
-	if err := vm.exec.EnqueueAccepted(ctx, b, execSynchronous); err != nil {
-		return err
+	if vm.consensusState.Get() == snow.Bootstrapping {
+		if err := b.WaitUntilExecuted(ctx); err != nil {
+			return fmt.Errorf("waiting for block %d to execute: %v", b.Height(), err)
+		}
 	}
 
 	vm.logger().Debug(

blocks/block.go

@@ -29,6 +29,9 @@ type Block struct {
 	ancestry  atomic.Pointer[ancestry]
 	execution atomic.Pointer[executionResults]
 
+	executed chan struct{} // closed after `execution` is set
+	settled  chan struct{} // closed after `ancestry` is cleared
+
 	log logging.Logger
 }
 
@@ -43,7 +46,9 @@ func InMemoryBlockCount() uint64 {
 // New constructs a new Block.
 func New(eth *types.Block, parent, lastSettled *Block, log logging.Logger) (*Block, error) {
 	b := &Block{
-		Block: eth,
+		Block:    eth,
+		executed: make(chan struct{}),
+		settled:  make(chan struct{}),
 	}
 	// TODO(arr4n) change to runtime.AddCleanup after the Go version has been
 	// bumped to >=1.24.0.

blocks/execution.go

@@ -1,6 +1,7 @@
 package blocks
 
 import (
+	"context"
 	"fmt"
 	"slices"
 	"time"
@@ -93,9 +94,21 @@ func (b *Block) markExecuted(e *executionResults) error {
 		b.log.Error("Block re-marked as executed")
 		return fmt.Errorf("block %d re-marked as executed", b.Height())
 	}
+	close(b.executed)
 	return nil
 }
 
+// WaitUntilExecuted blocks until either [Block.MarkExecuted] is called or the
+// [context.Context] is cancelled.
+func (b *Block) WaitUntilExecuted(ctx context.Context) error {
+	select {
+	case <-ctx.Done():
+		return ctx.Err()
+	case <-b.executed:
+		return nil
+	}
+}
+
 // Executed reports whether [Block.MarkExecuted] has been called and returned
 // without error.
 func (b *Block) Executed() bool {

blocks/settlement.go

@@ -1,6 +1,9 @@
 package blocks
 
-import "slices"
+import (
+	"context"
+	"slices"
+)
 
 type ancestry struct {
 	parent, lastSettled *Block
@@ -17,11 +20,23 @@ func (b *Block) MarkSettled() {
 		b.log.Fatal("Block re-settled")
 	}
 	if b.ancestry.CompareAndSwap(a, nil) {
+		close(b.settled)
 		return
 	}
 	b.log.Fatal("Block ancestry changed")
 }
 
+// WaitUntilSettled blocks until either [Block.MarkSettled] is called or the
+// [context.Context] is cancelled.
+func (b *Block) WaitUntilSettled(ctx context.Context) error {
+	select {
+	case <-ctx.Done():
+		return ctx.Err()
+	case <-b.settled:
+		return nil
+	}
+}
+
 // ParentBlock returns the block's parent unless [Block.MarkSettled] has been
 // called, in which case it returns nil.
 func (b *Block) ParentBlock() *Block {

sae_test.go

@@ -398,11 +398,7 @@ func TestBasicE2E(t *testing.T) {
 
 		t.Run("subscriptions", func(t *testing.T) {
 			t.Run("head_events", func(t *testing.T) {
-				for n := len(acceptedBlocks); !acceptedBlocks[n-1].Executed(); {
-					// There's no need to block on execution in production so
-					// making the executed bool a channel is overkill.
-					runtime.Gosched()
-				}
+				require.NoError(t, acceptedBlocks[len(acceptedBlocks)-1].WaitUntilExecuted(ctx))
 
 				headSub.Unsubscribe()
 				require.NoErrorf(t, <-headSub.Err(), "receive on %T.Err()", headSub)

saexec/execution.go

@@ -23,42 +23,24 @@ import (
 	"github.com/ava-labs/strevm/queue"
 )
 
-// Enqueue pushes a new block to the FIFO queue. It is non-blocking unless the
-// `synchronous` argument is true, in which case it returns when either the
-// [context.Context] is cancelled or the block has been executed.
-//
-// The `synchronous` argument SHOULD be true i.f.f. the chain is bootstrapping.
-func (e *Executor) EnqueueAccepted(ctx context.Context, block *blocks.Block, synchronous bool) error {
-	err := e.queue.UseThenSignal(ctx, func(q *queue.FIFO[*blocks.Block]) error {
+// Enqueue pushes a new block to the FIFO queue.
+func (e *Executor) EnqueueAccepted(ctx context.Context, block *blocks.Block) error {
+	return e.queue.UseThenSignal(ctx, func(q *queue.FIFO[*blocks.Block]) error {
 		q.Push(block)
-		e.queueCleared.Block()
 		return nil
 	})
-	if err != nil || !synchronous {
-		return err
-	}
-	return e.queueCleared.Wait(ctx)
 }
 
 func (e *Executor) processQueue() {
 	ctx := e.quitCtx()
 
 	for {
-		type pop struct {
-			block      *blocks.Block
-			emptyAfter bool
-		}
-
-		popped, err := sink.FromMonitor(ctx, e.queue,
+		block, err := sink.FromMonitor(ctx, e.queue,
 			func(q *queue.FIFO[*blocks.Block]) bool {
 				return q.Len() > 0
 			},
-			func(q *queue.FIFO[*blocks.Block]) (pop, error) {
-				b := q.Pop()
-				return pop{
-					block:      b,
-					emptyAfter: q.Len() == 0,
-				}, nil
+			func(q *queue.FIFO[*blocks.Block]) (*blocks.Block, error) {
+				return q.Pop(), nil
 			},
 		)
 		if errors.Is(err, context.Canceled) {
@@ -72,7 +54,6 @@ func (e *Executor) processQueue() {
 			return
 		}
 
-		block := popped.block
 		switch err := e.execute(ctx, block); {
 		case errors.Is(err, context.Canceled):
 			return
@@ -86,13 +67,6 @@ func (e *Executor) processQueue() {
 			)
 			return
 		}
-
-		// This may race with a concurrent call to [VM.AcceptBlock], but that is
-		// documented and also acceptable as we only ever Wait() inside
-		// [VM.AcceptBlock].
-		if popped.emptyAfter {
-			e.queueCleared.Open()
-		}
 	}
 }
 

saexec/saexec.go

@@ -37,12 +37,6 @@ type Executor struct {
 	queue        sink.Monitor[*queue.FIFO[*blocks.Block]]
 	lastExecuted *atomic.Pointer[blocks.Block] // shared with VM
 
-	// During bootstrapping we need to know when the queue has been executed to
-	// avoid verification returning an error due to waiting on the execution
-	// stream for settlement. This [sink.Gate] is only valid in the absence of
-	// concurrent calls to [Executor.EnqueueAccepted].
-	queueCleared sink.Gate
-
 	headEvents  event.FeedOf[core.ChainHeadEvent]
 	chainEvents event.FeedOf[core.ChainEvent]
 	logEvents   event.FeedOf[[]*types.Log]
@@ -119,7 +113,6 @@ func (e *Executor) init() error {
 func (e *Executor) Run(quit <-chan struct{}, ready chan<- struct{}) {
 	e.quit = quit
 	e.queue = sink.NewMonitor(new(queue.FIFO[*blocks.Block]))
-	e.queueCleared = sink.NewGate()
 	e.spawn(e.processQueue)
 
 	close(ready)