keyvm, mpcvm: one latch, not one per chain

Both had grown the same thing: a depth-one channel, a non-blocking send guarded
by a select with an empty default, and a WaitForEvent selecting over it and the
context. That is what vm.Latch is, so they use it. Four copies of the pattern
become none, and the comment explaining why the send is non-blocking lives with
the primitive rather than beside each caller.

Co-authored-by: Hanzo Dev <[email protected]>

keyvm/vm.go

@@ -105,7 +105,7 @@ type VM struct {
 	versdb   *versiondb.Database
 	state    database.Database // == versdb; buffered writes commit per block
 	toEngine chan<- vmcore.Message
-	notify   chan struct{}
+	work     vmcore.Latch
 
 	networkID uint32
 	clock     mockable.Clock
@@ -157,7 +157,6 @@ func (vm *VM) Initialize(ctx context.Context, init vmcore.Init) error {
 	vm.versdb = versiondb.New(init.DB)
 	vm.state = vm.versdb
 	vm.toEngine = init.ToEngine
-	vm.notify = make(chan struct{}, 1)
 
 	if init.Runtime != nil {
 		if logger, ok := init.Runtime.Log.(log.Logger); ok {
@@ -470,21 +469,13 @@ func (vm *VM) SubmitTx(tx *Transaction) (ids.ID, error) {
 	vm.mempool = append(vm.mempool, tx)
 	vm.mempoolLock.Unlock()
 
-	select {
-	case vm.notify <- struct{}{}:
-	default:
-	}
+	vm.work.Signal()
 	return tx.ID(), nil
 }
 
 // WaitForEvent blocks until there are pending transactions or the VM stops.
 func (vm *VM) WaitForEvent(ctx context.Context) (vmcore.Message, error) {
-	select {
-	case <-ctx.Done():
-		return vmcore.Message{}, ctx.Err()
-	case <-vm.notify:
-		return vmcore.Message{Type: vmcore.PendingTxs}, nil
-	}
+	return vm.work.WaitForEvent(ctx)
 }
 
 // BuildBlock drains the mempool into a new block extending the last accepted
@@ -538,10 +529,7 @@ func (vm *VM) requeue(txs []*Transaction) {
 	vm.mempoolLock.Lock()
 	vm.mempool = append(txs, vm.mempool...)
 	vm.mempoolLock.Unlock()
-	select {
-	case vm.notify <- struct{}{}:
-	default:
-	}
+	vm.work.Signal()
 }
 
 // dropFromMempool removes accepted transactions from the mempool by ID.

mpcvm/custody.go

@@ -504,12 +504,8 @@ func (vm *VM) stage(op *Operation) {
 	vm.order = append(vm.order, op.CeremonyID)
 	vm.mu.Unlock()
 
-	// Tell the engine there is work. Non-blocking: a full channel means the
-	// engine already has a pending build notification, which is all this is.
-	select {
-	case vm.buildRequests <- struct{}{}:
-	default:
-	}
+	// Tell the engine there is work.
+	vm.work.Signal()
 }
 
 // drain returns the staged operations in staging order, ready for a block.

mpcvm/vm.go

@@ -11,7 +11,6 @@
 package mpcvm
 
 import (
-	"sort"
 	"context"
 	"crypto/sha256"
 	"encoding/binary"
@@ -20,6 +19,7 @@ import (
 	"errors"
 	"fmt"
 	"net/http"
+	"sort"
 	"sync"
 	"time"
 
@@ -151,9 +151,9 @@ type VM struct {
 	// Completed ceremonies awaiting inclusion in a block, and their staging
 	// order. Every participant stages the same operations, so whichever node
 	// proposes produces a block the others can verify.
-	inflight      map[string]*Operation
-	order         []string
-	buildRequests chan struct{}
+	inflight map[string]*Operation
+	order    []string
+	work     vmcore.Latch
 
 	// localMesh is the in-process ceremony transport used when the VM has no
 	// p2p sender (single-process dev and tests). Nil on a networked node.
@@ -210,7 +210,6 @@ func (vm *VM) Initialize(
 	vm.pendingBlocks = make(map[ids.ID]*Block)
 	vm.shares = make(map[string]*heldShare)
 	vm.inflight = make(map[string]*Operation)
-	vm.buildRequests = make(chan struct{}, 1)
 	vm.sessionRouters = make(map[string]*gossipRouter)
 	vm.pendingBySession = make(map[string][]*protocol.Message)
 	vm.dailySigningCount = make(map[string]uint64)
@@ -951,12 +950,7 @@ func (vm *VM) NewHTTPHandler(ctx context.Context) (http.Handler, error) {
 // completed ceremony never reaches a block unless some other chain happened to
 // wake the builder.
 func (vm *VM) WaitForEvent(ctx context.Context) (vmcore.Message, error) {
-	select {
-	case <-vm.buildRequests:
-		return vmcore.Message{Type: vmcore.PendingTxs}, nil
-	case <-ctx.Done():
-		return vmcore.Message{}, ctx.Err()
-	}
+	return vm.work.WaitForEvent(ctx)
 }
 
 // Helper methods