test(node/sync-e2e): adding e2e safe head sync tests

crates/protocol/protocol/src/block.rs

@@ -23,12 +23,10 @@ pub struct BlockInfo {
     /// The block hash
     pub hash: B256,
     /// The block number
-    #[cfg_attr(feature = "serde", serde(with = "alloy_serde::quantity"))]
     pub number: u64,
     /// The parent block hash
     pub parent_hash: B256,
     /// The block timestamp
-    #[cfg_attr(feature = "serde", serde(with = "alloy_serde::quantity"))]
     pub timestamp: u64,
 }
 
@@ -81,10 +79,7 @@ pub struct L2BlockInfo {
     #[cfg_attr(feature = "serde", serde(rename = "l1origin", alias = "l1Origin"))]
     pub l1_origin: BlockNumHash,
     /// The sequence number of the L2 block
-    #[cfg_attr(
-        feature = "serde",
-        serde(with = "alloy_serde::quantity", rename = "sequenceNumber", alias = "seqNum")
-    )]
+    #[cfg_attr(feature = "serde", serde(rename = "sequenceNumber", alias = "seqNum"))]
     pub seq_num: u64,
 }
 
@@ -463,9 +458,9 @@ mod tests {
 
         let json = r#"{
             "hash": "0x0101010101010101010101010101010101010101010101010101010101010101",
-            "number": "0x1",
+            "number": 1,
             "parentHash": "0x0202020202020202020202020202020202020202020202020202020202020202",
-            "timestamp": "0x1"
+            "timestamp": 1
         }"#;
 
         let deserialized: BlockInfo = serde_json::from_str(json).unwrap();
@@ -518,14 +513,14 @@ mod tests {
 
         let json = r#"{
             "hash": "0x0101010101010101010101010101010101010101010101010101010101010101",
-            "number": "0x1",
+            "number": 1,
             "parentHash": "0x0202020202020202020202020202020202020202020202020202020202020202",
-            "timestamp": "0x1",
+            "timestamp": 1,
             "l1origin": {
                 "hash": "0x0303030303030303030303030303030303030303030303030303030303030303",
                 "number": 2
             },
-            "sequenceNumber": "0x3"
+            "sequenceNumber": 3
         }"#;
 
         let deserialized: L2BlockInfo = serde_json::from_str(json).unwrap();

tests/Justfile

@@ -46,4 +46,4 @@ isolate_test DEVNET_ENV_URL:
 
 test-e2e DEVNET COMMIT_TAG="": 
     just deploy "{{SOURCE}}/devnets/{{DEVNET}}.yaml" {{COMMIT_TAG}}
-    just isolate_test "kt://{{DEVNET}}-devnet"
+    just isolate_test "{{SOURCE}}/devnets/specs/{{DEVNET}}-devnet.json"

tests/devnets/large-kona.yaml

@@ -1,17 +1,18 @@
-# A larger network configuration for kurtosis (https://github.com/ethpandaops/optimism-package)
+# A large network configuration for kurtosis (https://github.com/ethpandaops/optimism-package)
+# Spins up a large EL/CL network.
 
 optimism_package:
   chains:
     # Chain with more nodes
     - participants:
       - el_type: op-geth
         cl_type: op-node
-        count: 1
+        count: 2
       - el_type: op-reth
         cl_type: kona-node
         # Note: we use the local image for now. This allows us to run the tests in CI pipelines without publishing new docker images every time.
         cl_image: "kona-node:local"
-        count: 2
+        count: 4
       network_params:
         network: "kurtosis"
         network_id: "2151908"

tests/go.mod

@@ -5,6 +5,8 @@ go 1.24.3
 // We're using the "develop" branch of the Optimism repo to include the latest changes to the `devnet-sdk` package.
 require github.com/ethereum-optimism/optimism v1.13.3-0.20250520004549-7962d43f57e6
 
+require github.com/stretchr/testify v1.10.0 
+
 require (
 	github.com/Azure/go-ansiterm v0.0.0-20250102033503-faa5f7b0171c // indirect
 	github.com/BurntSushi/toml v1.5.0 // indirect
@@ -129,7 +131,6 @@ require (
 	github.com/spaolacci/murmur3 v1.1.0 // indirect
 	github.com/spf13/afero v1.12.0 // indirect
 	github.com/stretchr/objx v0.5.2 // indirect
-	github.com/stretchr/testify v1.10.0 // indirect
 	github.com/supranational/blst v0.3.14 // indirect
 	github.com/syndtr/goleveldb v1.0.1-0.20220614013038-64ee5596c38a // indirect
 	github.com/tklauser/go-sysconf v0.3.14 // indirect

tests/node/mod.go

@@ -7,6 +7,11 @@ import (
 	"io"
 	"net/http"
 	"time"
+
+	"github.com/ethereum-optimism/optimism/devnet-sdk/testing/systest"
+	"github.com/ethereum-optimism/optimism/op-service/eth"
+	"github.com/gorilla/websocket"
+	"github.com/stretchr/testify/require"
 )
 
 // --- Generic RPC request/response types -------------------------------------
@@ -36,7 +41,7 @@ const (
 	DEFAULT_TIMEOUT = 10 * time.Second
 )
 
-func SendRPCRequest(addr string, method string, params ...any) (rpcResponse, error) {
+func SendRPCRequest[T any](addr string, method string, resOutput *T, params ...any) error {
 	// 1. Build the payload.
 	s := rpcRequest{
 		JSONRPC: "2.0",
@@ -48,7 +53,7 @@ func SendRPCRequest(addr string, method string, params ...any) (rpcResponse, err
 	// 1. Marshal the request.
 	payload, err := json.Marshal(s)
 	if err != nil {
-		return (rpcResponse{}), err
+		return err
 	}
 
 	// 2. Configure an HTTP client with sensible timeouts.
@@ -63,29 +68,92 @@ func SendRPCRequest(addr string, method string, params ...any) (rpcResponse, err
 	// 4. Build the HTTP request.
 	req, err := http.NewRequestWithContext(ctx, http.MethodPost,
 		addr, bytes.NewReader(payload))
+
 	if err != nil {
-		return (rpcResponse{}), err
+		return err
 	}
 	req.Header.Set("Content-Type", "application/json")
 
 	// 5. Send the request.
 	resp, err := client.Do(req)
 	if err != nil {
-		return (rpcResponse{}), err
+		return err
 	}
 	defer resp.Body.Close()
 
 	// 6. Read and decode the response.
 	respBytes, err := io.ReadAll(resp.Body)
 	if err != nil {
-		return (rpcResponse{}), err
+		return err
 	}
 
 	var rpcResp rpcResponse
 	if err := json.Unmarshal(respBytes, &rpcResp); err != nil {
-		return (rpcResponse{}), err
+		return err
+	}
+
+	err = json.Unmarshal(rpcResp.Result, resOutput)
+
+	if err != nil {
+		return err
+	}
+
+	return nil
+
+}
+
+func GetKonaWS[T any](t systest.T, wsRPC string, method string, runUntil <-chan T) []eth.L2BlockRef {
+	conn, _, err := websocket.DefaultDialer.DialContext(t.Context(), wsRPC, nil)
+	require.NoError(t, err, "dial: %v", err)
+	defer conn.Close()
+
+	// 1. send the *_subscribe request
+	require.NoError(t, conn.WriteJSON(rpcRequest{
+		JSONRPC: "2.0",
+		ID:      1,
+		Method:  "ws_subscribe_" + method,
+		Params:  nil,
+	}), "subscribe: %v", err)
+
+	// 2. read the ack – blocking read just once
+	var a rpcResponse
+	require.NoError(t, conn.ReadJSON(&a), "ack: %v", err)
+	t.Log("subscribed to websocket - id=", string(a.Result))
+
+	output := make([]eth.L2BlockRef, 0)
+
+	// 3. start a goroutine that keeps reading pushes
+outer_loop:
+	for {
+		select {
+		case <-runUntil:
+			// Clean‑up if necessary, then exit
+			t.Log(method, "subscriber", "stopping: runUntil condition met")
+			break outer_loop
+		case <-t.Context().Done():
+			// Clean‑up if necessary, then exit
+			t.Log("unsafe head subscriber", "stopping: context cancelled")
+			break outer_loop
+		default:
+			var msg json.RawMessage
+			require.NoError(t, conn.ReadJSON(&msg), "read: %v", err)
+
+			var p push
+			require.NoError(t, json.Unmarshal(msg, &p), "decode: %v", err)
+
+			t.Log("received websocket message - ", p.Params.Result)
+			output = append(output, p.Params.Result)
+		}
 	}
 
-	return rpcResp, nil
+	require.NoError(t, conn.WriteJSON(rpcRequest{
+		JSONRPC: "2.0",
+		ID:      2,
+		Method:  "ws_unsubscribe_" + method,
+		Params:  []interface{}{a.Result},
+	}), "unsubscribe: %v", err)
+
+	t.Log("gracefully closed websocket connection")
 
+	return output
 }

tests/node/node_test.go

@@ -10,14 +10,15 @@ import (
 // Contains general system tests for the p2p connectivity of the node.
 // This assumes there is at least two L2 chains. The second chain is used to test a larger network.
 func TestSystemNodeP2p(t *testing.T) {
-	// Check that the node has at least 1 peer that is connected to its topics when there is more than 1 peer in the network.
 	systest.SystemTest(t,
-		peerCount(1, 1),
+		allPeersInNetwork(),
 		validators.HasSufficientL2Nodes(0, 2),
 	)
 
+	// Check that the node has at least 1 peer that is connected to its topics when there is more than 1 peer in the network.
 	systest.SystemTest(t,
-		allPeersInNetwork(),
+		peerCount(1, 1),
+		validators.HasSufficientL2Nodes(0, 2),
 	)
 
 	// Check that the node has at least 2 peers that are connected to its topics when there is more than 3 peers in the network initially.
@@ -27,4 +28,17 @@ func TestSystemNodeP2p(t *testing.T) {
 		validators.HasSufficientL2Nodes(0, 3),
 	)
 
+	// Check that the node has at least 4 peers that are connected to its topics when there is more than 9 peers in the network initially.
+	// We put a lower bound on the number of connected peers to account for network instability.
+	systest.SystemTest(t,
+		peerCount(5, 3),
+		validators.HasSufficientL2Nodes(0, 6),
+	)
+}
+
+func TestSystemNodeSync(t *testing.T) {
+	systest.SystemTest(t,
+		syncSafe(),
+		// TODO(@theochap): we should add a custom validator that checks that there is at least one peer that supports the `kona` protocol.
+	)
 }

tests/node/p2p.go

@@ -1,8 +1,6 @@
 package node
 
 import (
-	"encoding/json"
-
 	"github.com/ethereum-optimism/optimism/devnet-sdk/system"
 	"github.com/ethereum-optimism/optimism/devnet-sdk/testing/systest"
 	"github.com/ethereum-optimism/optimism/op-service/apis"
@@ -19,20 +17,11 @@ func peerCount(minPeersKnown uint, minPeersConnected uint) systest.SystemTestFun
 				clRPC := node.CLRPC()
 				clName := node.CLName()
 
-				rpcResp, err := SendRPCRequest(clRPC, "opp2p_peerStats")
-
-				if err != nil {
-					t.Errorf("failed to send RPC request to node %s: %s", clName, err)
-				} else if rpcResp.Error != nil {
-					t.Errorf("received RPC error from node %s: %s", clName, rpcResp.Error)
-				}
-
 				peerStats := apis.PeerStats{}
-
-				err = json.Unmarshal(rpcResp.Result, &peerStats)
+				err := SendRPCRequest(clRPC, "opp2p_peerStats", &peerStats)
 
 				if err != nil {
-					t.Errorf("failed to unmarshal result: %s", err)
+					t.Errorf("failed to send RPC request to node %s: %s", clName, err)
 				}
 
 				require.GreaterOrEqual(t, peerStats.Known, minPeersKnown, "node %s has not enough known peers", clName)
@@ -62,20 +51,11 @@ func allPeersInNetwork() systest.SystemTestFunc {
 				clRPC := node.CLRPC()
 				clName := node.CLName()
 
-				rpcResp, err := SendRPCRequest(clRPC, "opp2p_self")
-
-				if err != nil {
-					t.Errorf("failed to send RPC request to node %s: %s", clName, err)
-				} else if rpcResp.Error != nil {
-					t.Errorf("received RPC error from node %s: %s", clName, rpcResp.Error)
-				}
-
 				peerInfo := apis.PeerInfo{}
-
-				err = json.Unmarshal(rpcResp.Result, &peerInfo)
+				err := SendRPCRequest(clRPC, "opp2p_self", &peerInfo)
 
 				if err != nil {
-					t.Errorf("failed to unmarshal result: %s", err)
+					t.Errorf("failed to send RPC request to node %s: %s", clName, err)
 				}
 
 				peerIds[peerInfo.PeerID.String()] = true
@@ -86,20 +66,11 @@ func allPeersInNetwork() systest.SystemTestFunc {
 				clRPC := node.CLRPC()
 				clName := node.CLName()
 
-				rpcResp, err := SendRPCRequest(clRPC, "opp2p_peers", true)
-
-				if err != nil {
-					t.Errorf("failed to send RPC request to node %s: %s", clName, err)
-				} else if rpcResp.Error != nil {
-					t.Errorf("received RPC error from node %s: %s", clName, rpcResp.Error)
-				}
-
 				peerDump := apis.PeerDump{}
-
-				err = json.Unmarshal(rpcResp.Result, &peerDump)
+				err := SendRPCRequest(clRPC, "opp2p_peers", &peerDump, true)
 
 				if err != nil {
-					t.Errorf("failed to unmarshal result: %s", err)
+					t.Errorf("failed to send RPC request to node %s: %s", clName, err)
 				}
 
 				for _, peer := range peerDump.Peers {