Repository: ethereum-optimism/supersim

Issue:
Support websocket on supersim L2s
Support websocket for op chain a and op chain b.
Only the L1 currently supports websocket

Hints from issue comments:
@fainashalts: Hi @OscBacon I've added this issue to our backlog and we'll keep you posted! Thanks for bringing it to our attention.@OscBacon: Hey @fainashalts any chance you could prioritize this more please?
My project is part of the superchain incubator and this is forcing us for now to stay on three anvil chains since we can't test subscribing to events.
Since supersim is using anvil under the hood it should be a matter of properly exposing both http and ws RPCs@fainashalts: @OscBacon yes sorry I wasn't more clear! our team will talk about this later today and figure out priority. 😄 backlog is just a pit stop. thank you for clarifying the need, very helpful for our discussion.@tremarkley: @OscBacon I believe that websockets should work for all the supersim chains. Do you mind sending me some steps I can follow to reproduce this issue?@OscBacon: Hey @tremarkley for sure!
```
supersim
# In another shell
wscat -c localhost:9546
```

`wscat` is a nifty too to send out requests to a WS server, like `cat`
Install with `npm i -g wscat`

For comparison, run `wscat -c localhost:8545` to connect to the L1.

As you will see, expected behavior:
```
wscat -c localhost:8545
Connected (press CTRL+C to quit)
```

but with the L2:
```
wscat -c localhost:9546
error: Unexpected server response: 400
```@tremarkley: @OscBacon this fix is included on supersim version 0.1.0-alpha.48

Solution patch:
diff --git a/anvil/anvil.go b/anvil/anvil.go
index dd9909b..309bc6a 100644
--- a/anvil/anvil.go
+++ b/anvil/anvil.go
@@ -234,7 +234,7 @@ func (a *Anvil) Start(ctx context.Context) error {
 		return ctx.Err()
 	}
 
-	rpcClient, err := rpc.Dial(a.wsEndpoint())
+	rpcClient, err := rpc.Dial(a.WSEndpoint())
 	if err != nil {
 		return fmt.Errorf("failed to create RPC client: %w", err)
 	}
@@ -266,7 +266,7 @@ func (a *Anvil) Endpoint() string {
 	return fmt.Sprintf("http://%s:%d", a.cfg.Host, a.cfg.Port)
 }
 
-func (a *Anvil) wsEndpoint() string {
+func (a *Anvil) WSEndpoint() string {
 	return fmt.Sprintf("ws://%s:%d", a.cfg.Host, a.cfg.Port)
 }
 
diff --git a/config/chain.go b/config/chain.go
index 2413aec..f19650b 100644
--- a/config/chain.go
+++ b/config/chain.go
@@ -106,6 +106,7 @@ type TraceCallResult struct {
 type Chain interface {
 	// Properties
 	Endpoint() string
+	WSEndpoint() string
 	LogPath() string
 	Config() *ChainConfig
 	EthClient() *ethclient.Client
diff --git a/opsimulator/opsimulator.go b/opsimulator/opsimulator.go
index 9ec9ef6..7dce295 100644
--- a/opsimulator/opsimulator.go
+++ b/opsimulator/opsimulator.go
@@ -11,6 +11,7 @@ import (
 	"net"
 	"net/http"
 	"strconv"
+	"sync"
 	"sync/atomic"
 
 	ophttp "github.com/ethereum-optimism/optimism/op-service/httputil"
@@ -29,6 +30,7 @@ import (
 	"github.com/ethereum/go-ethereum/ethclient"
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/rpc"
+	"github.com/gorilla/websocket"
 )
 
 const (
@@ -58,6 +60,12 @@ type OpSimulator struct {
 	ethClient *ethclient.Client
 
 	stopped atomic.Bool
+
+	// WebSocket configuration
+	upgrader      websocket.Upgrader
+	subscriptions map[string]*websocket.Conn
+	subsMutex     sync.Mutex
+	clientConnMu  sync.Mutex
 }
 
 // OpSimulator wraps around the l2 chain. By embedding `Chain`, it also implements the same inteface
@@ -83,12 +91,32 @@ func New(log log.Logger, closeApp context.CancelCauseFunc, port uint64, host str
 		},
 
 		peers: peers,
+
+		upgrader: websocket.Upgrader{
+			ReadBufferSize:  1024,
+			WriteBufferSize: 1024,
+			CheckOrigin: func(r *http.Request) bool {
+				// Allow all origins for development
+				return true
+			},
+		},
 	}
 }
 
 func (opSim *OpSimulator) Start(ctx context.Context) error {
 	mux := http.NewServeMux()
-	mux.Handle("/", corsHandler(opSim.handler(ctx)))
+
+	// Use a single handler for both HTTP and WebSocket
+	mux.Handle("/", corsHandler(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		// Check if this is a WebSocket request
+		if websocket.IsWebSocketUpgrade(r) {
+			opSim.handleWebSocket(w, r)
+			return
+		}
+
+		// Otherwise handle as HTTP request
+		opSim.handler(ctx)(w, r)
+	})))
 
 	hs, err := ophttp.StartHTTPServer(net.JoinHostPort(opSim.host, fmt.Sprintf("%d", opSim.port)), mux)
 	if err != nil {
@@ -292,6 +320,55 @@ func (opSim *OpSimulator) startBackgroundTasks() {
 	})
 }
 
+func (opSim *OpSimulator) superviseEthRequest(ctx context.Context, msg *jsonRpcMessage) *jsonRpcMessage {
+	if msg.Method == "eth_sendRawTransaction" {
+		var params []hexutil.Bytes
+		if err := json.Unmarshal(msg.Params, &params); err != nil {
+			opSim.log.Error("bad params sent to eth_sendRawTransaction", "err", err)
+			return msg.errorResponse(err)
+		}
+		if len(params) != 1 {
+			opSim.log.Error("eth_sendRawTransaction request has invalid number of params")
+			return msg.errorResponse(&jsonError{Code: InvalidParams, Message: "invalid request params"})
+		}
+
+		tx := new(types.Transaction)
+		if err := tx.UnmarshalBinary(params[0]); err != nil {
+			opSim.log.Error("failed to decode transaction data", "err", err)
+			return msg.errorResponse(err)
+		}
+
+		txHash := tx.Hash()
+
+		// Deposits should not be directly sendable to the L2
+		if tx.IsDepositTx() {
+			opSim.log.Error("rejecting deposit tx", "hash", txHash.String())
+			return msg.errorResponse(&jsonError{Code: InvalidParams, Message: "cannot process deposit tx"})
+		}
+
+		// Simulate the tx.
+		logs, err := opSim.SimulatedLogs(ctx, tx)
+		if err != nil {
+			opSim.log.Warn("failed to simulate transaction!!!", "err", err, "hash", txHash)
+			return msg.errorResponse(&jsonError{Code: InvalidParams, Message: "failed tx simulation"})
+		} else {
+			if err := opSim.checkInteropInvariants(ctx, logs); err != nil {
+				opSim.log.Error("unable to statisfy interop invariants within transaction", "err", err, "hash", txHash)
+				return msg.errorResponse(&jsonError{Code: InvalidParams, Message: err.Error()})
+			}
+		}
+	}
+
+	// NOTE: This fans out the batch request into individual requests. To match expected behavior, this
+	// should filter out messages that are invalid and reconstruct a single batch request to forward
+	response, jsonErr := forwardRPCRequest(ctx, opSim.Chain.EthClient().Client(), msg)
+	if jsonErr != nil {
+		return msg.errorResponse(jsonErr)
+	}
+
+	return response
+}
+
 func (opSim *OpSimulator) handler(ctx context.Context) http.HandlerFunc {
 	return func(w http.ResponseWriter, r *http.Request) {
 
@@ -307,60 +384,9 @@ func (opSim *OpSimulator) handler(ctx context.Context) http.HandlerFunc {
 			return
 		}
 
-		rpcClient := opSim.Chain.EthClient().Client()
 		batchRes := make([]*jsonRpcMessage, len(msgs))
 		for i, msg := range msgs {
-			if msg.Method == "eth_sendRawTransaction" {
-				var params []hexutil.Bytes
-				if err := json.Unmarshal(msg.Params, &params); err != nil {
-					opSim.log.Error("bad params sent to eth_sendRawTransaction", "err", err)
-					batchRes[i] = msg.errorResponse(err)
-					continue
-				}
-				if len(params) != 1 {
-					opSim.log.Error("eth_sendRawTransaction request has invalid number of params")
-					batchRes[i] = msg.errorResponse(&jsonError{Code: InvalidParams, Message: "invalid request params"})
-					continue
-				}
-
-				tx := new(types.Transaction)
-				if err := tx.UnmarshalBinary(params[0]); err != nil {
-					opSim.log.Error("failed to decode transaction data", "err", err)
-					batchRes[i] = msg.errorResponse(err)
-					continue
-				}
-
-				txHash := tx.Hash()
-
-				// Deposits should not be directly sendable to the L2
-				if tx.IsDepositTx() {
-					opSim.log.Error("rejecting deposit tx", "hash", txHash.String())
-					batchRes[i] = msg.errorResponse(&jsonError{Code: InvalidParams, Message: "cannot process deposit tx"})
-					continue
-				}
-
-				// Simulate the tx.
-				logs, err := opSim.SimulatedLogs(ctx, tx)
-				if err != nil {
-					opSim.log.Warn("failed to simulate transaction!!!", "err", err, "hash", txHash)
-					batchRes[i] = msg.errorResponse(&jsonError{Code: InvalidParams, Message: "failed tx simulation"})
-					continue
-				} else {
-					if err := opSim.checkInteropInvariants(ctx, logs); err != nil {
-						opSim.log.Error("unable to statisfy interop invariants within transaction", "err", err, "hash", txHash)
-						batchRes[i] = msg.errorResponse(&jsonError{Code: InvalidParams, Message: err.Error()})
-						continue
-					}
-				}
-			}
-
-			// NOTE: This fans out the batch request into individual requests. To match expected behavior, this
-			// should filter out messages that are invalid and reconstruct a single batch request to forward
-			var jsonErr *jsonError
-			batchRes[i], jsonErr = forwardRPCRequest(ctx, rpcClient, msg)
-			if jsonErr != nil {
-				batchRes[i] = msg.errorResponse(jsonErr)
-			}
+			batchRes[i] = opSim.superviseEthRequest(ctx, msg)
 		}
 
 		var encdata []byte
@@ -498,6 +524,11 @@ func (opSim *OpSimulator) Endpoint() string {
 	return fmt.Sprintf("http://%s:%d", opSim.host, opSim.port)
 }
 
+// Overridden such that the correct port is used
+func (opSim *OpSimulator) WSEndpoint() string {
+	return fmt.Sprintf("ws://%s:%d", opSim.host, opSim.port)
+}
+
 func corsHandler(next http.Handler) http.Handler {
 	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		w.Header().Set("Access-Control-Allow-Origin", "*")
@@ -512,3 +543,204 @@ func corsHandler(next http.Handler) http.Handler {
 		next.ServeHTTP(w, r)
 	})
 }
+
+func (opSim *OpSimulator) handleWebSocket(w http.ResponseWriter, r *http.Request) {
+	conn, err := opSim.upgrader.Upgrade(w, r, nil)
+	if err != nil {
+		opSim.log.Error("Failed to upgrade connection to WebSocket", "error", err)
+		return
+	}
+	defer conn.Close()
+
+	// Create a context for managing the connection lifecycle
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	// Channel to signal when we're done
+	done := make(chan struct{})
+
+	// Start a goroutine to handle messages from the client
+	go func() {
+		defer close(done)
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			default:
+				messageType, message, err := conn.ReadMessage()
+				if err != nil {
+					if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway, websocket.CloseAbnormalClosure) {
+						opSim.log.Error("WebSocket error", "error", err)
+					}
+					return
+				}
+				// Process the message in a separate goroutine to avoid blocking
+				go func(msgType int, msg []byte) {
+					if err := opSim.processWSRequest(ctx, msgType, msg, conn); err != nil {
+						opSim.log.Error("Failed to process WS RPC request", "error", err)
+					}
+				}(messageType, message)
+			}
+		}
+	}()
+
+	// Wait for the connection to be closed
+	<-done
+}
+
+func (opSim *OpSimulator) processWSRequest(ctx context.Context, clientRequestMessageType int, clientRequestMessage []byte, clientConn *websocket.Conn) error {
+	var msgs []*jsonRpcMessage
+	var isBatchRequest bool
+	var singleMsg jsonRpcMessage
+	if err := json.Unmarshal(clientRequestMessage, &singleMsg); err == nil {
+		msgs = []*jsonRpcMessage{&singleMsg}
+	} else {
+		// If that fails, try parsing as a batch
+		if err := json.Unmarshal(clientRequestMessage, &msgs); err != nil {
+			return fmt.Errorf("failed to parse JSON-RPC request: %w", err)
+		}
+		isBatchRequest = true
+	}
+
+	batchRes := make([]*jsonRpcMessage, len(msgs))
+	for i, msg := range msgs {
+		if msg.Method == "eth_subscribe" {
+			chainConn, resp, err := websocket.DefaultDialer.Dial(opSim.Chain.WSEndpoint(), nil)
+			if err != nil {
+				batchRes[i] = msg.errorResponse(err)
+				continue
+			}
+			if resp != nil {
+				defer resp.Body.Close()
+			}
+
+			msgBytes, err := json.Marshal(msg)
+			if err != nil {
+				batchRes[i] = msg.errorResponse(err)
+				continue
+			}
+			if err := chainConn.WriteMessage(clientRequestMessageType, msgBytes); err != nil {
+				batchRes[i] = msg.errorResponse(err)
+				continue
+			}
+
+			// Read the subscription ID
+			_, responseMessage, err := chainConn.ReadMessage()
+			if err != nil {
+				batchRes[i] = msg.errorResponse(err)
+				continue
+			}
+
+			var subResponse struct {
+				ID     int    `json:"id"`
+				Result string `json:"result"`
+			}
+			if err := json.Unmarshal(responseMessage, &subResponse); err != nil {
+				batchRes[i] = msg.errorResponse(err)
+				continue
+			}
+
+			opSim.subsMutex.Lock()
+			if opSim.subscriptions == nil {
+				opSim.subscriptions = make(map[string]*websocket.Conn)
+			}
+			opSim.subscriptions[subResponse.Result] = chainConn
+			opSim.subsMutex.Unlock()
+
+			var jsonRpcResponse *jsonRpcMessage
+			if err := json.Unmarshal(responseMessage, &jsonRpcResponse); err != nil {
+				batchRes[i] = msg.errorResponse(err)
+				continue
+			}
+			batchRes[i] = jsonRpcResponse
+
+			// Start a goroutine to forward messages from the subscription
+			go func() {
+				defer func() {
+					opSim.subsMutex.Lock()
+					delete(opSim.subscriptions, subResponse.Result)
+					opSim.subsMutex.Unlock()
+					chainConn.Close()
+				}()
+
+				for {
+					messageType, message, err := chainConn.ReadMessage()
+					if err != nil {
+						if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway, websocket.CloseAbnormalClosure) {
+							opSim.log.Error("WS connection closed unexpectedly", "error", err)
+						}
+						return
+					}
+					opSim.clientConnMu.Lock()
+					if err := clientConn.WriteMessage(messageType, message); err != nil {
+						opSim.clientConnMu.Unlock()
+						opSim.log.Error("Failed to write WS Message", "error", err)
+						return
+					}
+					opSim.clientConnMu.Unlock()
+				}
+			}()
+
+			continue
+		}
+
+		if msg.Method == "eth_unsubscribe" {
+			var params []string
+			if err := json.Unmarshal(msg.Params, &params); err != nil {
+				batchRes[i] = msg.errorResponse(err)
+				continue
+			}
+			if len(params) != 1 {
+				batchRes[i] = msg.errorResponse(&jsonError{Code: InvalidParams, Message: "invalid request params"})
+				continue
+			}
+			subscriptionID := params[0]
+			subConn, exists := opSim.subscriptions[subscriptionID]
+			if exists {
+				defer subConn.Close()
+				// Send unsubscribe request
+				if err := subConn.WriteMessage(clientRequestMessageType, clientRequestMessage); err != nil {
+					batchRes[i] = msg.errorResponse(err)
+					continue
+				}
+				opSim.subsMutex.Lock()
+				delete(opSim.subscriptions, subscriptionID)
+				opSim.subsMutex.Unlock()
+
+				// Read the response
+				_, response, err := subConn.ReadMessage()
+				if err != nil {
+					batchRes[i] = msg.errorResponse(err)
+					continue
+				}
+				var jsonRpcResponse *jsonRpcMessage
+				if err := json.Unmarshal(response, &jsonRpcResponse); err != nil {
+					batchRes[i] = msg.errorResponse(err)
+					continue
+				}
+				batchRes[i] = jsonRpcResponse
+
+				continue
+			}
+			// if subscription does not exist, then pass through the request
+		}
+
+		batchRes[i] = opSim.superviseEthRequest(ctx, msg)
+	}
+
+	opSim.clientConnMu.Lock()
+	defer opSim.clientConnMu.Unlock()
+	if isBatchRequest {
+		batchResponseJSON, err := json.Marshal(batchRes)
+		if err != nil {
+			return fmt.Errorf("failed to marshal batch response: %w", err)
+		}
+		return clientConn.WriteMessage(clientRequestMessageType, batchResponseJSON)
+	} else {
+		responseJSON, err := json.Marshal(batchRes[0])
+		if err != nil {
+			return fmt.Errorf("failed to marshal response: %w", err)
+		}
+		return clientConn.WriteMessage(clientRequestMessageType, responseJSON)
+	}
+}
diff --git a/orchestrator/orchestrator.go b/orchestrator/orchestrator.go
index b6ae1ce..69e4257 100644
--- a/orchestrator/orchestrator.go
+++ b/orchestrator/orchestrator.go
@@ -241,6 +241,13 @@ func (o *Orchestrator) Endpoint(chainId uint64) string {
 	return o.l2OpSims[chainId].Endpoint()
 }
 
+func (o *Orchestrator) WSEndpoint(chainId uint64) string {
+	if o.l1Chain.Config().ChainID == chainId {
+		return o.l1Chain.WSEndpoint()
+	}
+	return o.l2OpSims[chainId].WSEndpoint()
+}
+
 func (o *Orchestrator) ConfigAsString() string {
 	var b strings.Builder
 	l1Cfg := o.l1Chain.Config()



Test patch:
diff --git a/supersim_test.go b/supersim_test.go
index ea1e1a2..b885758 100644
--- a/supersim_test.go
+++ b/supersim_test.go
@@ -33,6 +33,7 @@ import (
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"golang.org/x/net/websocket"
 )
 
 const (
@@ -232,6 +233,36 @@ func TestStartup(t *testing.T) {
 	l1Client.Close()
 }
 
+func TestWsStartup(t *testing.T) {
+	t.Parallel()
+	testSuite := createTestSuite(t)
+
+	l2Chains := testSuite.Supersim.Orchestrator.L2Chains()
+	require.True(t, len(l2Chains) > 0)
+
+	// test that all chains can be queried
+	for _, chain := range l2Chains {
+		l2Client, err := rpc.DialWebsocket(context.Background(), chain.WSEndpoint(), "")
+		require.NoError(t, err)
+
+		var chainId gethmath.HexOrDecimal64
+		require.NoError(t, l2Client.CallContext(context.Background(), &chainId, "eth_chainId"))
+		require.Equal(t, chain.Config().ChainID, uint64(chainId))
+
+		l2Client.Close()
+	}
+
+	// test that l1 anvil can be queried
+	l1Client, err := rpc.DialWebsocket(context.Background(), testSuite.Supersim.Orchestrator.L1Chain().WSEndpoint(), "")
+	require.NoError(t, err)
+
+	var chainId gethmath.HexOrDecimal64
+	require.NoError(t, l1Client.CallContext(context.Background(), &chainId, "eth_chainId"))
+	require.Equal(t, testSuite.Supersim.Orchestrator.L1Chain().Config().ChainID, uint64(chainId))
+
+	l1Client.Close()
+}
+
 func TestL2Count(t *testing.T) {
 	t.Parallel()
 	testSuite := createTestSuite(t, func(cfg *config.CLIConfig) *config.CLIConfig {
@@ -429,6 +460,27 @@ func TestBatchJsonRpcRequests(t *testing.T) {
 	}
 }
 
+func TestBatchJsonRpcRequestsWs(t *testing.T) {
+	t.Parallel()
+	testSuite := createTestSuite(t)
+
+	for _, chain := range testSuite.Supersim.Orchestrator.L2Chains() {
+		client, err := ethclient.Dial(testSuite.Supersim.Orchestrator.WSEndpoint(chain.Config().ChainID))
+		require.NoError(t, err)
+		defer client.Close()
+
+		elems := []rpc.BatchElem{{Method: "eth_chainId", Result: new(hexutil.Uint64)}, {Method: "eth_blockNumber", Result: new(hexutil.Uint64)}}
+		require.NoError(t, client.Client().BatchCall(elems))
+
+		require.Nil(t, elems[0].Error)
+		require.Nil(t, elems[1].Error)
+
+		require.NotZero(t, uint64(*(elems[0].Result).(*hexutil.Uint64)))
+		// TODO: fix later, this occasionally fails when we set anvil on block-time 2
+		// require.NotZero(t, uint64(*(elems[1].Result).(*hexutil.Uint64)))
+	}
+}
+
 func TestBatchJsonRpcRequestErrorHandling(t *testing.T) {
 	t.Parallel()
 	testSuite := createInteropTestSuite(t)
@@ -549,6 +601,70 @@ func TestInteropInvariantCheckSucceeds(t *testing.T) {
 	require.True(t, receipt.Status == 1, "initiating message transaction failed")
 }
 
+func TestInteropInvariantCheckSucceedsWs(t *testing.T) {
+	t.Parallel()
+	testSuite := createInteropTestSuite(t)
+
+	sourceClient, err := ethclient.Dial(testSuite.Supersim.Orchestrator.WSEndpoint(testSuite.Supersim.NetworkConfig.L2Configs[0].ChainID))
+	require.NoError(t, err)
+	defer sourceClient.Close()
+
+	destClient, err := ethclient.Dial(testSuite.Supersim.Orchestrator.WSEndpoint(testSuite.Supersim.NetworkConfig.L2Configs[1].ChainID))
+	require.NoError(t, err)
+	defer destClient.Close()
+
+	privateKey, err := testSuite.DevKeys.Secret(devkeys.UserKey(0))
+	require.NoError(t, err)
+
+	l2ToL2CrossDomainMessenger, err := bindings.NewL2ToL2CrossDomainMessenger(predeploys.L2toL2CrossDomainMessengerAddr, sourceClient)
+	require.NoError(t, err)
+
+	// Create initiating message using L2ToL2CrossDomainMessenger
+	origin := predeploys.L2toL2CrossDomainMessengerAddr
+	parsedSchemaRegistryAbi, _ := abi.JSON(strings.NewReader(opbindings.SchemaRegistryABI))
+	data, err := parsedSchemaRegistryAbi.Pack("register", "uint256 value", common.HexToAddress("0x0000000000000000000000000000000000000000"), false)
+	require.NoError(t, err)
+
+	sourceTransactor, err := bind.NewKeyedTransactorWithChainID(privateKey, testSuite.SourceChainID)
+	require.NoError(t, err)
+	tx, err := l2ToL2CrossDomainMessenger.SendMessage(sourceTransactor, testSuite.DestChainID, predeploys.SchemaRegistryAddr, data)
+	require.NoError(t, err)
+
+	initiatingMessageTxReceipt, err := bind.WaitMined(context.Background(), sourceClient, tx)
+	require.NoError(t, err)
+	require.True(t, initiatingMessageTxReceipt.Status == 1, "initiating message transaction failed")
+
+	// progress forward one block before sending tx
+	err = testSuite.DestEthClient.Client().CallContext(context.Background(), nil, "anvil_mine", uint64(3), uint64(2))
+	require.NoError(t, err)
+	err = sourceClient.Client().CallContext(context.Background(), nil, "anvil_mine", uint64(3), uint64(2))
+	require.NoError(t, err)
+
+	l2tol2CDM, err := bindings.NewL2ToL2CrossDomainMessengerTransactor(predeploys.L2toL2CrossDomainMessengerAddr, destClient)
+	require.NoError(t, err)
+	initiatingMessageBlockHeader, err := sourceClient.HeaderByNumber(context.Background(), initiatingMessageTxReceipt.BlockNumber)
+	require.NoError(t, err)
+	initiatingMessageLog := initiatingMessageTxReceipt.Logs[0]
+	identifier := bindings.Identifier{
+		Origin:      origin,
+		BlockNumber: initiatingMessageTxReceipt.BlockNumber,
+		LogIndex:    big.NewInt(0),
+		Timestamp:   new(big.Int).SetUint64(initiatingMessageBlockHeader.Time),
+		ChainId:     testSuite.SourceChainID,
+	}
+	transactor, err := bind.NewKeyedTransactorWithChainID(privateKey, testSuite.DestChainID)
+	require.NoError(t, err)
+	transactor.AccessList = interop.MessageAccessList(&identifier, interop.ExecutingMessagePayloadBytes(initiatingMessageLog))
+
+	// Should succeed
+	tx, err = l2tol2CDM.RelayMessage(transactor, identifier, interop.ExecutingMessagePayloadBytes(initiatingMessageLog))
+	require.NoError(t, err)
+
+	receipt, err := bind.WaitMined(context.Background(), testSuite.DestEthClient, tx)
+	require.NoError(t, err)
+	require.True(t, receipt.Status == 1, "initiating message transaction failed")
+}
+
 func TestInteropInvariantCheckFailsBadLogIndex(t *testing.T) {
 	t.Parallel()
 	testSuite := createInteropTestSuite(t)
@@ -603,6 +719,68 @@ func TestInteropInvariantCheckFailsBadLogIndex(t *testing.T) {
 	require.Error(t, err)
 }
 
+func TestInteropInvariantCheckFailsBadLogIndexWs(t *testing.T) {
+	t.Parallel()
+	testSuite := createInteropTestSuite(t)
+
+	sourceClient, err := ethclient.Dial(testSuite.Supersim.Orchestrator.WSEndpoint(testSuite.Supersim.NetworkConfig.L2Configs[0].ChainID))
+	require.NoError(t, err)
+	defer sourceClient.Close()
+
+	destClient, err := ethclient.Dial(testSuite.Supersim.Orchestrator.WSEndpoint(testSuite.Supersim.NetworkConfig.L2Configs[1].ChainID))
+	require.NoError(t, err)
+	defer destClient.Close()
+
+	privateKey, err := testSuite.DevKeys.Secret(devkeys.UserKey(0))
+	require.NoError(t, err)
+
+	l2ToL2CrossDomainMessenger, err := bindings.NewL2ToL2CrossDomainMessenger(predeploys.L2toL2CrossDomainMessengerAddr, sourceClient)
+	require.NoError(t, err)
+
+	// Create initiating message using L2ToL2CrossDomainMessenger
+	origin := predeploys.L2toL2CrossDomainMessengerAddr
+	parsedSchemaRegistryAbi, _ := abi.JSON(strings.NewReader(opbindings.SchemaRegistryABI))
+	data, err := parsedSchemaRegistryAbi.Pack("register", "uint256 value", common.HexToAddress("0x0000000000000000000000000000000000000000"), false)
+	require.NoError(t, err)
+
+	sourceTransactor, err := bind.NewKeyedTransactorWithChainID(privateKey, testSuite.SourceChainID)
+	require.NoError(t, err)
+	tx, err := l2ToL2CrossDomainMessenger.SendMessage(sourceTransactor, testSuite.DestChainID, predeploys.SchemaRegistryAddr, data)
+	require.NoError(t, err)
+
+	initiatingMessageTxReceipt, err := bind.WaitMined(context.Background(), sourceClient, tx)
+	require.NoError(t, err)
+	require.True(t, initiatingMessageTxReceipt.Status == 1, "initiating message transaction failed")
+
+	// progress forward one block before sending tx
+	err = destClient.Client().CallContext(context.Background(), nil, "anvil_mine", uint64(3), uint64(2))
+	require.NoError(t, err)
+	err = sourceClient.Client().CallContext(context.Background(), nil, "anvil_mine", uint64(3), uint64(2))
+	require.NoError(t, err)
+
+	crossL2Inbox, err := bindings.NewCrossL2Inbox(predeploys.CrossL2InboxAddr, destClient)
+	require.NoError(t, err)
+	initiatingMessageBlockHeader, err := sourceClient.HeaderByNumber(context.Background(), initiatingMessageTxReceipt.BlockNumber)
+	require.NoError(t, err)
+
+	initiatingMessageLog := initiatingMessageTxReceipt.Logs[0]
+	identifier := bindings.Identifier{
+		Origin:      origin,
+		BlockNumber: initiatingMessageTxReceipt.BlockNumber,
+		LogIndex:    big.NewInt(5), // Wrong index
+		Timestamp:   new(big.Int).SetUint64(initiatingMessageBlockHeader.Time),
+		ChainId:     testSuite.SourceChainID,
+	}
+	transactor, err := bind.NewKeyedTransactorWithChainID(privateKey, testSuite.DestChainID)
+	require.NoError(t, err)
+
+	transactor.AccessList = interop.MessageAccessList(&identifier, interop.ExecutingMessagePayloadBytes(initiatingMessageLog))
+
+	// Should fail because the log index is incorrect
+	_, err = crossL2Inbox.ValidateMessage(transactor, identifier, crypto.Keccak256Hash(interop.ExecutingMessagePayloadBytes(initiatingMessageLog)))
+	require.Error(t, err)
+}
+
 func TestInteropInvariantCheckBadBlockNumber(t *testing.T) {
 	t.Parallel()
 	testSuite := createInteropTestSuite(t)
@@ -1068,3 +1246,59 @@ func TestAdminGetL2ToL2MessageByMsgHash(t *testing.T) {
 	assert.Equal(t, tx.To().String(), message.Target.String())
 	assert.Equal(t, tx.To().String(), message.Sender.String())
 }
+
+func TestEthSubscribeNewHeads(t *testing.T) {
+	t.Parallel()
+	testSuite := createTestSuite(t)
+
+	// Connect to the WebSocket endpoint
+	conn, err := websocket.Dial(testSuite.Supersim.Orchestrator.WSEndpoint(testSuite.Supersim.NetworkConfig.L2Configs[0].ChainID), "", "http://localhost")
+	require.NoError(t, err)
+	defer conn.Close()
+
+	// Subscribe to newHeads
+	subscribeMsg := map[string]interface{}{
+		"jsonrpc": "2.0",
+		"id":      1,
+		"method":  "eth_subscribe",
+		"params":  []string{"newHeads"},
+	}
+	err = websocket.JSON.Send(conn, subscribeMsg)
+	require.NoError(t, err)
+
+	// Read the subscription response
+	var subscribeResponse map[string]interface{}
+	err = websocket.JSON.Receive(conn, &subscribeResponse)
+	require.NoError(t, err)
+	require.NotNil(t, subscribeResponse["result"])
+	subscriptionID := subscribeResponse["result"].(string)
+	require.NotEmpty(t, subscriptionID)
+
+	// Mine a new block to trigger the subscription
+	err = testSuite.Supersim.Orchestrator.L2Chains()[0].EthClient().Client().CallContext(context.Background(), nil, "anvil_mine", uint64(1))
+	require.NoError(t, err)
+
+	// Read the notification
+	var notification map[string]interface{}
+	err = websocket.JSON.Receive(conn, &notification)
+	require.NoError(t, err)
+	require.Equal(t, "eth_subscription", notification["method"])
+	require.Equal(t, subscriptionID, notification["params"].(map[string]interface{})["subscription"])
+	require.NotNil(t, notification["params"].(map[string]interface{})["result"])
+
+	// Unsubscribe
+	unsubscribeMsg := map[string]interface{}{
+		"jsonrpc": "2.0",
+		"id":      2,
+		"method":  "eth_unsubscribe",
+		"params":  []string{subscriptionID},
+	}
+	err = websocket.JSON.Send(conn, unsubscribeMsg)
+	require.NoError(t, err)
+
+	// Read the unsubscribe response
+	var unsubscribeResponse map[string]interface{}
+	err = websocket.JSON.Receive(conn, &unsubscribeResponse)
+	require.NoError(t, err)
+	require.True(t, unsubscribeResponse["result"].(bool))
+}
diff --git a/testutils/mockchain.go b/testutils/mockchain.go
index ef9bfce..a67857f 100644
--- a/testutils/mockchain.go
+++ b/testutils/mockchain.go
@@ -32,6 +32,10 @@ func (c *MockChain) Endpoint() string {
 	return "http://localhost:8545"
 }
 
+func (c *MockChain) WSEndpoint() string {
+	return "ws://localhost:8545"
+}
+
 func (c *MockChain) LogPath() string {
 	return "var/chain/log"
 }