add e2e block production monitor

cmd/zetae2e/local/local.go

@@ -190,6 +190,10 @@ func localE2ETest(cmd *cobra.Command, _ []string) {
 		noError(deployerRunner.FundEmissionsPool())
 	}
 
+	// monitor block production to ensure we fail fast if there are consensus failures
+	// this is not run in an errgroup since only returning an error will not exit immedately
+	go monitorBlockProductionExit(ctx, conf)
+
 	// wait for keygen to be completed
 	// if setup is skipped, we assume that the keygen is already completed
 	if !skipSetup {

cmd/zetae2e/local/monitor_block_production.go

@@ -0,0 +1,55 @@
+package local
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"time"
+
+	rpchttp "github.com/cometbft/cometbft/rpc/client/http"
+	cometbft_types "github.com/cometbft/cometbft/types"
+
+	"github.com/zeta-chain/node/e2e/config"
+)
+
+// monitorBlockProductionExit calls monitorBlockProduction and exits upon any error
+func monitorBlockProductionExit(ctx context.Context, conf config.Config) {
+	err := monitorBlockProduction(ctx, conf)
+	if err != nil {
+		fmt.Printf("❌ block monitor: %v\n", err)
+		os.Exit(2)
+	}
+}
+
+// monitorBlockProduction subscribes to new block events to monitor if blocks are being produced
+// at least every four seconds
+func monitorBlockProduction(ctx context.Context, conf config.Config) error {
+	rpcClient, err := rpchttp.New(conf.RPCs.ZetaCoreRPC, "/websocket")
+	if err != nil {
+		return fmt.Errorf("new zetacore rpc: %w", err)
+	}
+
+	err = rpcClient.WSEvents.Start()
+	if err != nil {
+		return fmt.Errorf("start ws events: %w", err)
+	}
+	blockEventChan, err := rpcClient.WSEvents.Subscribe(ctx, "", "tm.event='NewBlock'")
+	if err != nil {
+		return fmt.Errorf("subscribe: %w", err)
+	}
+	latestNewBlockEvent := cometbft_types.EventDataNewBlock{}
+	for {
+		select {
+		case event := <-blockEventChan:
+			newBlockEvent, ok := event.Data.(cometbft_types.EventDataNewBlock)
+			if !ok {
+				return fmt.Errorf("expecting new block event, got %T", event.Data)
+			}
+			latestNewBlockEvent = newBlockEvent
+		case <-time.After(4 * time.Second):
+			return fmt.Errorf("timed out waiting for new block (last block %d)", latestNewBlockEvent.Block.Height)
+		case <-ctx.Done():
+			return nil
+		}
+	}
+}