simulators/eth2/engine: Fix Execution Finality Verification

simulators/eth2/engine/running_testnet.go

@@ -21,6 +21,7 @@ import (
 	"github.com/protolambda/zrnt/eth2/beacon/common"
 	"github.com/protolambda/zrnt/eth2/beacon/phase0"
 	"github.com/protolambda/zrnt/eth2/util/math"
+	"github.com/protolambda/ztyp/tree"
 )
 
 var MAX_PARTICIPATION_SCORE = 7
@@ -250,6 +251,133 @@ func (t *Testnet) WaitForFinality(ctx context.Context, timeoutSlots common.Slot)
 	}
 }
 
+// WaitForExecutionFinality blocks until a beacon client reaches finality
+// and the finality checkpoint contains an execution payload,
+// or timeoutSlots have passed, whichever happens first.
+func (t *Testnet) WaitForExecutionFinality(ctx context.Context, timeoutSlots common.Slot) (common.Checkpoint, error) {
+	genesis := t.GenesisTime()
+	slotDuration := time.Duration(t.spec.SECONDS_PER_SLOT) * time.Second
+	timer := time.NewTicker(slotDuration)
+	runningBeacons := t.VerificationNodes().BeaconClients().Running()
+	done := make(chan common.Checkpoint, len(runningBeacons))
+	var timeout <-chan time.Time
+	if timeoutSlots > 0 {
+		timeout = t.SlotsTimeout(timeoutSlots)
+	} else {
+		timeout = make(<-chan time.Time)
+	}
+	for {
+		select {
+		case <-ctx.Done():
+			return common.Checkpoint{}, fmt.Errorf("context called")
+		case <-timeout:
+			return common.Checkpoint{}, fmt.Errorf("Timeout")
+		case finalized := <-done:
+			return finalized, nil
+		case tim := <-timer.C:
+			// start polling after first slot of genesis
+			if tim.Before(genesis.Add(slotDuration)) {
+				t.Logf("Time till genesis: %s", genesis.Sub(tim))
+				continue
+			}
+
+			// new slot, log and check status of all beacon nodes
+			type res struct {
+				idx int
+				msg string
+				err error
+			}
+			var (
+				wg sync.WaitGroup
+				ch = make(chan res, len(runningBeacons))
+			)
+			for i, b := range runningBeacons {
+				wg.Add(1)
+				go func(ctx context.Context, i int, b *BeaconClient, ch chan res) {
+					defer wg.Done()
+					ctx, cancel := context.WithTimeout(ctx, time.Second*5)
+					defer cancel()
+
+					var (
+						slot      common.Slot
+						head      string
+						justified string
+						finalized string
+					)
+
+					var headInfo eth2api.BeaconBlockHeaderAndInfo
+					if exists, err := beaconapi.BlockHeader(ctx, b.API, eth2api.BlockHead, &headInfo); err != nil {
+						ch <- res{err: fmt.Errorf("beacon %d: failed to poll head: %v", i, err)}
+						return
+					} else if !exists {
+						ch <- res{err: fmt.Errorf("beacon %d: no head block", i)}
+						return
+					}
+
+					var checkpoints eth2api.FinalityCheckpoints
+					if exists, err := beaconapi.FinalityCheckpoints(ctx, b.API, eth2api.StateIdRoot(headInfo.Header.Message.StateRoot), &checkpoints); err != nil || !exists {
+						if exists, err = beaconapi.FinalityCheckpoints(ctx, b.API, eth2api.StateIdSlot(headInfo.Header.Message.Slot), &checkpoints); err != nil {
+							ch <- res{err: fmt.Errorf("beacon %d: failed to poll finality checkpoint: %v", i, err)}
+							return
+						} else if !exists {
+							ch <- res{err: fmt.Errorf("beacon %d: Expected state for head block", i)}
+							return
+						}
+					}
+
+					slot = headInfo.Header.Message.Slot
+					head = shorten(headInfo.Root.String())
+					justified = shorten(checkpoints.CurrentJustified.String())
+					finalized = shorten(checkpoints.Finalized.String())
+
+					var (
+						execution    tree.Root
+						executionStr = "0x0000..0000"
+					)
+
+					if (checkpoints.Finalized != common.Checkpoint{}) {
+						var versionedBlock eth2api.VersionedSignedBeaconBlock
+						if exists, err := beaconapi.BlockV2(ctx, b.API, eth2api.BlockIdRoot(checkpoints.Finalized.Root), &versionedBlock); err != nil {
+							ch <- res{err: fmt.Errorf("beacon %d: failed to retrieve block: %v", i, err)}
+							return
+						} else if !exists {
+							ch <- res{err: fmt.Errorf("beacon %d: block not found", i)}
+							return
+						}
+
+						switch versionedBlock.Version {
+						case "bellatrix":
+							block := versionedBlock.Data.(*bellatrix.SignedBeaconBlock)
+							execution = block.Message.Body.ExecutionPayload.BlockHash
+							executionStr = shorten(execution.String())
+						}
+					}
+
+					ch <- res{i, fmt.Sprintf("beacon %d: slot=%d, head=%s, finalized_exec_payload=%s, justified=%s, finalized=%s", i, slot, head, executionStr, justified, finalized), nil}
+
+					if (execution != tree.Root{}) {
+						done <- checkpoints.Finalized
+					}
+				}(ctx, i, b, ch)
+			}
+			wg.Wait()
+			close(ch)
+
+			// print out logs in ascending idx order
+			sorted := make([]string, len(runningBeacons))
+			for out := range ch {
+				if out.err != nil {
+					return common.Checkpoint{}, out.err
+				}
+				sorted[out.idx] = out.msg
+			}
+			for _, msg := range sorted {
+				t.Logf(msg)
+			}
+		}
+	}
+}
+
 // Waits for the current epoch to be finalized, or timeoutSlots have passed, whichever happens first.
 func (t *Testnet) WaitForCurrentEpochFinalization(ctx context.Context, timeoutSlots common.Slot) (common.Checkpoint, error) {
 	genesis := t.GenesisTime()

simulators/eth2/engine/scenarios.go

@@ -148,9 +148,9 @@ func BlockLatestSafeFinalized(t *hivesim.T, env *testEnv, n node) {
 
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
-	_, err := testnet.WaitForFinality(ctx, testnet.spec.SLOTS_PER_EPOCH*beacon.Slot(EPOCHS_TO_FINALITY+1))
+	_, err := testnet.WaitForExecutionFinality(ctx, testnet.spec.SLOTS_PER_EPOCH*beacon.Slot(EPOCHS_TO_FINALITY+2))
 	if err != nil {
-		t.Fatalf("FAIL: Waiting for finality: %v", err)
+		t.Fatalf("FAIL: Waiting for execution finality: %v", err)
 	}
 	if err := VerifyELBlockLabels(testnet, ctx); err != nil {
 		t.Fatalf("FAIL: Verifying EL block labels: %v", err)