diff --git a/driver/network/local/local.go b/driver/network/local/local.go index 28362253..c1a77a08 100644 --- a/driver/network/local/local.go +++ b/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) diff --git a/driver/network/local/local_test.go b/driver/network/local/local_test.go index 293809b3..68551ffe 100644 --- a/driver/network/local/local_test.go +++ b/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) + } +} diff --git a/driver/node/opera.go b/driver/node/opera.go index 7e5390bd..b30688d7 100644 --- a/driver/node/opera.go +++ b/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() } diff --git a/driver/node/opera_test.go b/driver/node/opera_test.go index 807582c0..78c04d31 100644 --- a/driver/node/opera_test.go +++ b/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 +} diff --git a/driver/rpc/rpc.go b/driver/rpc/rpc.go index c81f5d0f..eaefdb8d 100644 --- a/driver/rpc/rpc.go +++ b/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 diff --git a/driver/rpc/rpc_test.go b/driver/rpc/rpc_test.go index b7be99d4..34161807 100644 --- a/driver/rpc/rpc_test.go +++ b/driver/rpc/rpc_test.go @@ -111,3 +111,48 @@ func TestRpcClientImpl_WaitTransactionReceipt_Error(t *testing.T) { t.Fatalf("expected error %v, got %v", injectedError, err) } } + +func TestRpcClientImpl_WaitTransactionReceipt_TransientRetry(t *testing.T) { + t.Parallel() + + ctrl := gomock.NewController(t) + + mock := NewMockrpcClient(ctrl) + client := Impl{ + rpcClient: mock, + txReceiptTimeout: 30 * time.Second, + } + + injectedResult := map[string]any{ + "cumulativeGasUsed": "0x0", + "logsBloom": "0x" + strings.Repeat("00", 256), + "logs": []map[string]any{}, + "transactionHash": "0x" + strings.Repeat("00", 32), + "gasUsed": "0x0", + } + + call := 0 + mock.EXPECT(). + Call(gomock.Any(), "eth_getTransactionReceipt", gomock.Any()). + DoAndReturn(func(result interface{}, method string, args ...interface{}) error { + call++ + if call <= 2 { + return errors.New("Post \"http://0.0.0.0:41851\": dial tcp 0.0.0.0:41851: connect: connection refused") + } + resultPtr, ok := result.(*map[string]any) + if !ok { + t.Fatalf("result type is not *map[string]any") + } + *resultPtr = injectedResult + return nil + }). + Times(3) + + receipt, err := client.WaitTransactionReceipt(common.Hash{}) + if err != nil { + t.Fatalf("expected no error after transient retries, got %v", err) + } + if receipt == nil { + t.Fatal("expected non-nil receipt") + } +}