Refactor online nodes.

driver/network/local/local.go

@@ -26,6 +26,7 @@ import (
 	"path/filepath"
 	"sync"
 	"sync/atomic"
+	"time"
 
 	"github.com/0xsoniclabs/norma/driver"
 	"github.com/0xsoniclabs/norma/driver/docker"
@@ -194,6 +195,7 @@ func (n *LocalNetwork) addNodeIntoNetwork(node *node.OperaNode) error {
 	if err != nil {
 		return fmt.Errorf("failed to get node id; %v", err)
 	}
+	slog.Info("Adding node into network", "node", node.GetLabel(), "id", id)
 	for _, other := range n.nodes {
 		if err = other.AddPeer(id); err != nil {
 			return fmt.Errorf("failed to add peer; %v", err)
@@ -203,6 +205,46 @@ func (n *LocalNetwork) addNodeIntoNetwork(node *node.OperaNode) error {
 	return nil
 }
 
+// WaitForBlockProduction waits until all nodes in the network have logged at
+// least one "New block" line. The provided context controls the overall deadline.
+func (n *LocalNetwork) WaitForBlockProduction(ctx context.Context) error {
+	n.nodesMutex.Lock()
+	nodes := make([]*node.OperaNode, 0, len(n.nodes))
+	for _, nd := range n.nodes {
+		nodes = append(nodes, nd)
+	}
+	n.nodesMutex.Unlock()
+
+	slog.Info("waiting for block production on all nodes", "count", len(nodes))
+
+	type result struct {
+		label string
+		err   error
+	}
+	results := make(chan result, len(nodes))
+
+	for _, nd := range nodes {
+		go func(nd *node.OperaNode) {
+			err := nd.CheckBlockProducing(ctx)
+			results <- result{nd.GetLabel(), err}
+		}(nd)
+	}
+
+	var errs []error
+	for range nodes {
+		r := <-results
+		if r.err != nil {
+			errs = append(errs, fmt.Errorf("node %s: %w", r.label, r.err))
+		}
+	}
+	if len(errs) > 0 {
+		slog.Error("block production check failed for some nodes", "failures", len(errs))
+	} else {
+		slog.Info("all nodes confirmed block production", "count", len(nodes))
+	}
+	return errors.Join(errs...)
+}
+
 // createNode is an internal version of CreateNode enabling the creation
 // of validator and non-validator nodes in the network.
 func (n *LocalNetwork) createNode(ctx context.Context, nodeConfig *node.OperaNodeConfig) (*node.OperaNode, error) {
@@ -213,6 +255,19 @@ func (n *LocalNetwork) createNode(ctx context.Context, nodeConfig *node.OperaNod
 	if err := n.addNodeIntoNetwork(node); err != nil {
 		return nil, fmt.Errorf("failed to connect node; %w", err)
 	}
+
+	// Wait for the newly added node to produce a block.
+	blockCtx, blockCancel := context.WithTimeout(ctx, 50*time.Second)
+	defer blockCancel()
+	if err := node.CheckBlockProducing(blockCtx); err != nil {
+		return nil, fmt.Errorf("node %s failed block production check: %w", node.GetLabel(), err)
+	}
+
+	// Wait until all nodes in the network are producing blocks.
+	if err := n.WaitForBlockProduction(blockCtx); err != nil {
+		return nil, fmt.Errorf("network failed block production check: %w", err)
+	}
+
 	n.listenerMutex.Lock()
 	for listener := range n.listeners {
 		listener.AfterNodeCreation(node)

driver/network/local/local_test.go

@@ -801,3 +801,28 @@ func TestLocalNetwork_MountDataDir_Can_Be_Reused(t *testing.T) {
 			temp, currVisitedDirs)
 	}
 }
+
+func TestWaitForBlockProduction_AllNodesProduceBlocks(t *testing.T) {
+	t.Parallel()
+	config := driver.NetworkConfig{
+		Validators: driver.NewDefaultTestValidators(t.Name(), 2),
+	}
+	ctx, cancel := context.WithTimeout(t.Context(), 5*time.Minute)
+	defer cancel()
+
+	net, err := NewLocalNetwork(ctx, &config)
+	if err != nil {
+		t.Fatalf("failed to create local network: %v", err)
+	}
+	t.Cleanup(func() {
+		_ = net.Shutdown()
+	})
+
+	// WaitForBlockProduction should succeed since validators are already
+	// producing blocks after network startup.
+	blockCtx, blockCancel := context.WithTimeout(ctx, 30*time.Second)
+	defer blockCancel()
+	if err := net.WaitForBlockProduction(blockCtx); err != nil {
+		t.Fatalf("WaitForBlockProduction failed: %v", err)
+	}
+}

driver/node/opera.go

@@ -22,10 +22,12 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"log/slog"
 	"os"
 	"path/filepath"
 	"regexp"
 	"slices"
+	"strconv"
 	"strings"
 	"sync"
 	"time"
@@ -211,6 +213,7 @@ func StartOperaDockerNode(ctx context.Context, client *docker.Client, dn *docker
 		genesisJSONPath = *config.GenesisJsonPath
 	}
 
+	slog.Info("Starting Opera node", "label", config.Label, "image", image, "validatorId", validatorId, "genesisJsonPath", genesisJSONPath)
 	host, err := network.RetryReturn(ctx, network.DefaultRetryAttempts, 1*time.Second, func() (*docker.Container, error) {
 		ports, err := network.GetFreePorts(len(operaServices.Services()))
 		if err != nil {
@@ -293,6 +296,7 @@ func StartOperaDockerNode(ctx context.Context, client *docker.Client, dn *docker
 	if err != nil {
 		return nil, err
 	}
+	slog.Info("Opera node started", "label", config.Label, "hostname", host.Hostname())
 
 	// Use a private copy of the config to avoid modifying the original.
 	nodeConfig := *config
@@ -313,6 +317,7 @@ func StartOperaDockerNode(ctx context.Context, client *docker.Client, dn *docker
 		tempDirs:  tempDirs,
 	}
 
+	slog.Info("Waiting for Opera node to be ready", "label", config.Label, "hostname", host.Hostname())
 	// Wait until the OperaNode inside the Container is ready.
 	err = network.Retry(ctx, network.DefaultRetryAttempts, 1*time.Second, func() error {
 		if err := node.host.CheckRunning(); err != nil {
@@ -407,6 +412,79 @@ func (n *OperaNode) StreamLog() (io.ReadCloser, error) {
 	return n.host.StreamLog()
 }
 
+// CheckBlockProducing verifies that the node is producing blocks by streaming
+// the container logs and waiting until at least 2 "New block" lines with
+// strictly increasing block indices are observed. The provided context controls
+// the deadline; if it expires before the condition is met, an error is returned.
+func (n *OperaNode) CheckBlockProducing(ctx context.Context) error {
+	slog.Info("checking block production", "node", n.GetLabel())
+	reader, err := n.StreamLog()
+	if err != nil {
+		return fmt.Errorf("failed to stream log for block production check: %w", err)
+	}
+	defer func() {
+		if err := reader.Close(); err != nil {
+			slog.Error("failed to close log stream", "node", n.GetLabel(), "error", err)
+		}
+	}()
+
+	done := make(chan struct{})
+	var scanErr error
+	go func() {
+		defer close(done)
+		scanner := bufio.NewScanner(reader)
+		var lastIndex int64 = -1
+		seen := 0
+		for scanner.Scan() {
+			line := scanner.Text()
+			if !strings.Contains(line, "New block") {
+				continue
+			}
+			idx, err := parseBlockIndex(line)
+			if err != nil {
+				continue
+			}
+			if idx > lastIndex {
+				slog.Debug("observed block", "node", n.GetLabel(), "index", idx, "seen", seen+1)
+				lastIndex = idx
+				seen++
+			}
+			if seen >= 2 {
+				return
+			}
+		}
+		if err := scanner.Err(); err != nil {
+			scanErr = err
+		} else {
+			scanErr = fmt.Errorf("log stream ended without observing 2 increasing blocks (saw %d)", seen)
+		}
+	}()
+
+	select {
+	case <-done:
+		slog.Info("block production confirmed", "node", n.GetLabel())
+		return scanErr
+	case <-ctx.Done():
+		slog.Warn("block production check timed out", "node", n.GetLabel())
+		if err := reader.Close(); err != nil {
+			slog.Error("failed to close log stream", "node", n.GetLabel(), "error", err)
+		}
+		return fmt.Errorf("node %s did not produce 2 increasing blocks before timeout: %w", n.GetLabel(), ctx.Err())
+	}
+}
+
+// blockIndexReg matches the block index field in a "New block" log line.
+var blockIndexReg = regexp.MustCompile(`index=(\d+)`)
+
+// parseBlockIndex extracts the block index from a "New block" log line.
+func parseBlockIndex(line string) (int64, error) {
+	m := blockIndexReg.FindStringSubmatch(line)
+	if len(m) < 2 {
+		return 0, fmt.Errorf("no block index found in line")
+	}
+	return strconv.ParseInt(m[1], 10, 64)
+}
+
 func (n *OperaNode) Stop() error {
 	return n.host.Stop()
 }

driver/node/opera_test.go

@@ -18,6 +18,7 @@ package node
 
 import (
 	"bufio"
+	"context"
 	"fmt"
 	"io"
 	"net/http"
@@ -295,3 +296,111 @@ func TestClient_Stop_Graceful(t *testing.T) {
 		t.Errorf("container did not stop gracefully")
 	}
 }
+
+func TestCheckBlockProducing_SucceedsWithTwoIncreasingBlocks(t *testing.T) {
+	t.Parallel()
+	logContent := "INFO [05-04|09:34:15.537] Starting node\n" +
+		"INFO [05-04|09:34:16.000] New block index=1 gas_used=0 txs=0 t=1ms\n" +
+		"INFO [05-04|09:34:17.000] New block index=2 gas_used=0 txs=0 t=1ms\n"
+
+	node := &OperaNode{
+		host:   &fakeLogHost{logContent: logContent},
+		config: &OperaNodeConfig{Label: "test-node"},
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	if err := node.CheckBlockProducing(ctx); err != nil {
+		t.Fatalf("expected no error, got: %v", err)
+	}
+}
+
+func TestCheckBlockProducing_FailsWithOnlyOneBlock(t *testing.T) {
+	t.Parallel()
+	logContent := "INFO [05-04|09:34:15.537] Starting node\n" +
+		"INFO [05-04|09:34:16.000] New block index=1 gas_used=0 txs=0 t=1ms\n"
+
+	node := &OperaNode{
+		host:   &fakeLogHost{logContent: logContent},
+		config: &OperaNodeConfig{Label: "test-node"},
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	err := node.CheckBlockProducing(ctx)
+	if err == nil {
+		t.Fatal("expected error, got nil")
+	}
+	if !strings.Contains(err.Error(), "without observing 2 increasing blocks") {
+		t.Fatalf("unexpected error message: %v", err)
+	}
+}
+
+func TestCheckBlockProducing_FailsWhenNoBlockBeforeTimeout(t *testing.T) {
+	t.Parallel()
+
+	node := &OperaNode{
+		host:   &blockingLogHost{},
+		config: &OperaNodeConfig{Label: "test-node"},
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
+	defer cancel()
+
+	err := node.CheckBlockProducing(ctx)
+	if err == nil {
+		t.Fatal("expected error, got nil")
+	}
+	if !strings.Contains(err.Error(), "did not produce 2 increasing blocks before timeout") {
+		t.Fatalf("unexpected error message: %v", err)
+	}
+}
+
+func TestCheckBlockProducing_FailsWhenLogStreamEndsWithoutBlock(t *testing.T) {
+	t.Parallel()
+	logContent := "INFO Starting node\nINFO IPC endpoint opened\n"
+
+	node := &OperaNode{
+		host:   &fakeLogHost{logContent: logContent},
+		config: &OperaNodeConfig{Label: "test-node"},
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	err := node.CheckBlockProducing(ctx)
+	if err == nil {
+		t.Fatal("expected error, got nil")
+	}
+	if !strings.Contains(err.Error(), "without observing 2 increasing blocks") {
+		t.Fatalf("unexpected error message: %v", err)
+	}
+}
+
+// fakeLogHost is a test Host implementation that returns pre-configured log content.
+type fakeLogHost struct {
+	cleanupHostStub
+	logContent string
+}
+
+func (h *fakeLogHost) StreamLog() (io.ReadCloser, error) {
+	return io.NopCloser(strings.NewReader(h.logContent)), nil
+}
+
+// blockingLogHost is a test Host implementation whose StreamLog blocks until
+// the reader is closed (simulating a running container with no block output).
+type blockingLogHost struct {
+	cleanupHostStub
+}
+
+func (h *blockingLogHost) StreamLog() (io.ReadCloser, error) {
+	r, w := io.Pipe()
+	// Write some non-block log lines, then keep the pipe open.
+	go func() {
+		_, _ = w.Write([]byte("INFO Starting node\nINFO IPC endpoint opened\n"))
+		// Never close w — the reader blocks until r.Close() is called.
+	}()
+	return r, nil
+}

driver/rpc/rpc.go

@@ -22,6 +22,7 @@ import (
 	"errors"
 	"fmt"
 	"math/big"
+	"strings"
 	"time"
 
 	"github.com/0xsoniclabs/sonic/api/sonicapi"
@@ -112,7 +113,7 @@ func (r Impl) WaitTransactionReceipt(txHash common.Hash) (*types.Receipt, error)
 	delay := time.Millisecond
 	for time.Since(begin) < r.txReceiptTimeout {
 		receipt, err := r.transactionReceipt(txHash)
-		if errors.Is(err, ethereum.NotFound) {
+		if errors.Is(err, ethereum.NotFound) || isTransientError(err) {
 			time.Sleep(delay)
 			delay = 2 * delay
 			if delay > maxDelay {
@@ -210,6 +211,19 @@ func (r Impl) transactionReceipt(txHash common.Hash) (*types.Receipt, error) {
 	return receipt, nil
 }
 
+// isTransientError returns true if the error is a transient network error that
+// should be retried (e.g. connection refused, connection reset, EOF).
+func isTransientError(err error) bool {
+	if err == nil {
+		return false
+	}
+	msg := err.Error()
+	return strings.Contains(msg, "connection refused") ||
+		strings.Contains(msg, "connection reset") ||
+		strings.Contains(msg, "EOF") ||
+		strings.Contains(msg, "broken pipe")
+}
+
 // TxPoolStatus represents the presence of a transaction in the transaction pool.
 type TxPoolStatus int