feat(node/test): improve connection drop tests

crates/node/rpc/src/ws.rs

@@ -71,6 +71,7 @@ impl WsServer for WsRPC {
             .await
             .map(|state| *state)
         {
+            info!(target: "rpc::ws", "Sending safe head update: {:?}", new_state.sync_state.safe_head());
             current_safe_head = new_state.sync_state.safe_head();
             Self::send_state_update(&sink, current_safe_head).await?;
         }

mise.toml

@@ -1,3 +1,3 @@
 [tools]
 
-"ubi:kurtosis-tech/kurtosis-cli-release-artifacts[exe=kurtosis]" = "1.10.3"
+"ubi:kurtosis-tech/kurtosis-cli-release-artifacts[exe=kurtosis]" = "1.8.1"

tests/devnets/first-kona-conductor.yaml

@@ -56,6 +56,7 @@ ethereum_package:
   participants:
   - el_type: geth
     cl_type: teku
+    cl_image: consensys/teku:25.7.1
   network_params:
     preset: minimal
     genesis_delay: 5

tests/devnets/large-kona-sequencer.yaml

@@ -94,6 +94,7 @@ ethereum_package:
   participants:
   - el_type: geth
     cl_type: teku
+    cl_image: consensys/teku:25.7.1
   network_params:
     preset: minimal
     genesis_delay: 5

tests/devnets/large-kona.yaml

@@ -84,6 +84,7 @@ ethereum_package:
   participants:
   - el_type: geth
     cl_type: teku
+    cl_image: consensys/teku:25.7.1
   network_params:
     preset: minimal
     genesis_delay: 5

tests/devnets/preinterop-supervisor.yaml

@@ -67,6 +67,7 @@ ethereum_package:
   participants:
     - el_type: geth
       cl_type: teku
+      cl_image: consensys/teku:25.7.1
   network_params:
     preset: minimal
     genesis_delay: 5

tests/devnets/simple-kona-conductor.yaml

@@ -43,6 +43,7 @@ ethereum_package:
   participants:
   - el_type: geth
     cl_type: teku
+    cl_image: consensys/teku:25.7.1
   network_params:
     preset: minimal
     genesis_delay: 5

tests/devnets/simple-kona-geth.yaml

@@ -42,6 +42,7 @@ ethereum_package:
   participants:
   - el_type: geth
     cl_type: teku
+    cl_image: consensys/teku:25.7.1
   network_params:
     preset: minimal
     genesis_delay: 5

tests/devnets/simple-kona-sequencer.yaml

@@ -38,6 +38,7 @@ ethereum_package:
   participants:
   - el_type: geth
     cl_type: teku
+    cl_image: consensys/teku:25.7.1
   network_params:
     preset: minimal
     genesis_delay: 5

tests/devnets/simple-kona.yaml

@@ -42,6 +42,7 @@ ethereum_package:
   participants:
   - el_type: geth
     cl_type: teku
+    cl_image: consensys/teku:25.7.1
   network_params:
     preset: minimal
     genesis_delay: 5

tests/devnets/simple-supervisor.yaml

@@ -74,6 +74,7 @@ ethereum_package:
   participants:
     - el_type: geth
       cl_type: teku
+      cl_image: consensys/teku:25.7.1
   network_params:
     preset: minimal
     genesis_delay: 5

tests/node/common/engine_test.go

@@ -36,7 +36,7 @@ func TestEngine(gt *testing.T) {
 					done <- struct{}{}
 				}()
 
-				queue <- GetDevWS(t, node, "engine_queue_size", done)
+				queue <- node_utils.GetDevWS(t, node, "engine_queue_size", done)
 			}()
 
 			q := <-queue

tests/node/common/sync_ws_test.go

@@ -34,9 +34,9 @@ func TestSyncUnsafeBecomesSafe(gt *testing.T) {
 		go func(node *dsl.L2CLNode) {
 			defer wg.Done()
 
-			unsafeBlocks := GetKonaWs(t, node, "unsafe_head", time.After(SECS_WAIT_FOR_UNSAFE_HEAD*time.Second))
+			unsafeBlocks := node_utils.GetKonaWs(t, node, "unsafe_head", time.After(SECS_WAIT_FOR_UNSAFE_HEAD*time.Second))
 
-			safeBlocks := GetKonaWs(t, node, "safe_head", time.After(SECS_WAIT_FOR_SAFE_HEAD*time.Second))
+			safeBlocks := node_utils.GetKonaWs(t, node, "safe_head", time.After(SECS_WAIT_FOR_SAFE_HEAD*time.Second))
 
 			require.GreaterOrEqual(t, len(unsafeBlocks), 1, "we didn't receive enough unsafe gossip blocks!")
 			require.GreaterOrEqual(t, len(safeBlocks), 1, "we didn't receive enough safe gossip blocks!")
@@ -80,7 +80,7 @@ func TestSyncUnsafe(gt *testing.T) {
 		go func(node *dsl.L2CLNode) {
 			defer wg.Done()
 
-			output := GetKonaWs(t, node, "unsafe_head", time.After(10*time.Second))
+			output := node_utils.GetKonaWs(t, node, "unsafe_head", time.After(10*time.Second))
 
 			// For each block, we check that the block is actually in the chain of the other nodes.
 			// That should always be the case unless there is a reorg or a long sync.
@@ -125,7 +125,7 @@ func TestSyncSafe(gt *testing.T) {
 			defer wg.Done()
 			clName := node.Escape().ID().Key()
 
-			output := GetKonaWs(t, node, "safe_head", time.After(10*time.Second))
+			output := node_utils.GetKonaWs(t, node, "safe_head", time.After(10*time.Second))
 
 			// For each block, we check that the block is actually in the chain of the other nodes.
 			// That should always be the case unless there is a reorg or a long sync.
@@ -170,7 +170,7 @@ func TestSyncFinalized(gt *testing.T) {
 			defer wg.Done()
 			clName := node.Escape().ID().Key()
 
-			output := GetKonaWs(t, node, "finalized_head", time.After(4*time.Minute))
+			output := node_utils.GetKonaWs(t, node, "finalized_head", time.After(4*time.Minute))
 
 			// We should check that we received at least 2 finalized blocks within 4 minutes!
 			require.Greater(t, len(output), 1, "we didn't receive enough finalized gossip blocks!")

tests/node/restart/conn_drop_test.go

@@ -7,6 +7,7 @@ import (
 
 	"github.com/ethereum-optimism/optimism/op-devstack/devtest"
 	"github.com/ethereum-optimism/optimism/op-devstack/dsl"
+	"github.com/ethereum-optimism/optimism/op-service/eth"
 	"github.com/ethereum-optimism/optimism/op-service/retry"
 	"github.com/ethereum-optimism/optimism/op-supervisor/supervisor/types"
 	node_utils "github.com/op-rs/kona/node/utils"
@@ -32,17 +33,53 @@ func TestConnDropSync(gt *testing.T) {
 		node.DisconnectPeer(&sequencer)
 
 		// Ensure that the node is no longer connected to the sequencer
+		t.Logf("node %s is disconnected from sequencer %s", clName, sequencer.Escape().ID().Key())
 		seqPeers := sequencer.Peers()
 		for _, peer := range seqPeers.Peers {
 			t.Require().NotEqual(peer.PeerID, node.PeerInfo().PeerID, "expected node %s to be disconnected from sequencer %s", clName, sequencer.Escape().ID().Key())
 		}
 
+		peers := node.Peers()
+		for _, peer := range peers.Peers {
+			t.Require().NotEqual(peer.PeerID, sequencer.PeerInfo().PeerID, "expected node %s to be disconnected from sequencer %s", clName, sequencer.Escape().ID().Key())
+		}
+
+		currentUnsafeHead := node.ChainSyncStatus(node.ChainID(), types.LocalUnsafe)
+
+		endSignal := make(chan struct{})
+
+		safeHeads := node_utils.GetKonaWsAsync(t, &node, "safe_head", endSignal)
+		unsafeHeads := node_utils.GetKonaWsAsync(t, &node, "unsafe_head", endSignal)
+
+		// Ensures that....
+		// - the node's safe head is advancing and eventually catches up with the unsafe head
+		// - the node's unsafe head is NOT advancing during this time
+		check := func() error {
+		outer_loop:
+			for {
+				select {
+				case safeHead := <-safeHeads:
+					t.Logf("node %s safe head is advancing", clName)
+					if safeHead.Number >= currentUnsafeHead.Number {
+						t.Logf("node %s safe head caught up with unsafe head", clName)
+						break outer_loop
+					}
+				case unsafeHead := <-unsafeHeads:
+					return fmt.Errorf("node %s unsafe head is advancing: %d", clName, unsafeHead.Number)
+				}
+			}
+
+			endSignal <- struct{}{}
+
+			return nil
+		}
+
 		// Check that...
 		// - the node's safe head is advancing
 		// - the node's unsafe head is advancing (through consolidation)
 		// - the node's safe head's number is catching up with the unsafe head's number
 		// - the node's unsafe head is strictly lagging behind the sequencer's unsafe head
-		postDisconnectCheckFuns = append(postDisconnectCheckFuns, node.AdvancedFn(types.LocalSafe, 50, 200), node.AdvancedFn(types.LocalUnsafe, 50, 200), SafeUnsafeMatchedFn(t, node, 50))
+		postDisconnectCheckFuns = append(postDisconnectCheckFuns, node.AdvancedFn(types.LocalSafe, 50, 200), node.AdvancedFn(types.LocalUnsafe, 50, 200), check)
 	}
 
 	postDisconnectCheckFuns = append(postDisconnectCheckFuns, sequencer.AdvancedFn(types.LocalUnsafe, 50, 200))
@@ -72,28 +109,50 @@ func TestConnDropSync(gt *testing.T) {
 		t.Require().True(found, "expected node %s to be connected to reference node %s", clName, sequencer.Escape().ID().Key())
 
 		// Check that the node is resyncing with the unsafe head network
-		postReconnectCheckFuns = append(postReconnectCheckFuns, node.MatchedFn(&sequencer, types.LocalSafe, 50), node.MatchedFn(&sequencer, types.LocalUnsafe, 50))
+		postReconnectCheckFuns = append(postReconnectCheckFuns, MatchedWithinRange(t, node, sequencer, 3, types.LocalSafe, 50), node.AdvancedFn(types.LocalUnsafe, 50, 100), MatchedWithinRange(t, node, sequencer, 3, types.LocalUnsafe, 100))
 	}
 
 	dsl.CheckAll(t, postReconnectCheckFuns...)
 }
 
-// MatchedFn returns a lambda that checks the baseNode head with given safety level is matched with the refNode chain sync status provider
-// Composable with other lambdas to wait in parallel
-func SafeUnsafeMatchedFn(t devtest.T, clNode dsl.L2CLNode, attempts int) dsl.CheckFunc {
+func MatchedWithinRange(t devtest.T, baseNode, refNode dsl.L2CLNode, delta uint64, lvl types.SafetyLevel, attempts int) dsl.CheckFunc {
 	logger := t.Logger()
-	chainID := clNode.ChainID()
+	chainID := baseNode.ChainID()
+
 	return func() error {
+		base := baseNode.ChainSyncStatus(chainID, lvl)
+		ref := refNode.ChainSyncStatus(chainID, lvl)
+		logger.Info("Expecting node to match with reference", "base", base.Number, "ref", ref.Number)
 		return retry.Do0(t.Ctx(), attempts, &retry.FixedStrategy{Dur: 2 * time.Second},
 			func() error {
-				base := clNode.ChainSyncStatus(chainID, types.LocalSafe)
-				ref := clNode.ChainSyncStatus(chainID, types.LocalUnsafe)
-				if ref.Hash == base.Hash && ref.Number == base.Number {
-					logger.Info("Node safe and unsafe heads matched", "ref", ref.Number, "base", base.Number)
+				base = baseNode.ChainSyncStatus(chainID, lvl)
+				ref = refNode.ChainSyncStatus(chainID, lvl)
+				if ref.Number <= base.Number+delta || ref.Number >= base.Number-delta {
+					logger.Info("Node matched", "ref_id", refNode, "base_id", baseNode, "ref", ref.Number, "base", base.Number, "delta", delta)
+
+					// We get the same block from the head and tail node
+					var headNode dsl.L2CLNode
+					var tailNode eth.BlockID
+					if ref.Number > base.Number {
+						headNode = refNode
+						tailNode = base
+					} else {
+						headNode = baseNode
+						tailNode = ref
+					}
+
+					baseBlock, err := headNode.Escape().RollupAPI().OutputAtBlock(t.Ctx(), tailNode.Number)
+					if err != nil {
+						return err
+					}
+
+					t.Require().Equal(baseBlock.BlockRef.Number, tailNode.Number, "expected block number to match")
+					t.Require().Equal(baseBlock.BlockRef.Hash, tailNode.Hash, "expected block hash to match")
+
 					return nil
 				}
-				logger.Info("Node safe and unsafe heads not matched", "safe", base.Number, "unsafe", ref.Number, "ref", ref.Hash, "base", base.Hash)
-				return fmt.Errorf("expected safe and unsafe heads to match")
+				logger.Info("Node sync status", "base", base.Number, "ref", ref.Number)
+				return fmt.Errorf("expected head to match: %s", lvl)
 			})
 	}
 }