Nats Data Stream by tamingchaos · Pull Request #1966 · 0xPolygon/cdk-erigon

tamingchaos · 2025-07-17T01:22:26Z

This pull request introduces support for NATS-based datastreaming in the zkevm module, allowing the system to use NATS as a backend for data streaming in addition to the existing TCP-based solution. It adds new configuration flags and options, updates the client and server initialization logic to support NATS, and refactors related code to use more flexible interfaces. Several dependencies are also updated to support the new functionality.

NATS datastreaming integration:

Added new configuration flags and struct fields (L2NatsUrl, DataStreamNatsHost, DataStreamNatsPort) to enable and configure NATS-based datastream endpoints (cmd/utils/flags.go, eth/ethconfig/config_zkevm.go, turbo/cli/default_flags.go, turbo/cli/flags_zkevm.go) [1] [2] [3] [4] [5].
Introduced a natsManager in the Ethereum struct, initialized and started in the backend, with proper shutdown logic (eth/backend.go) [1] [2] [3].
Updated datastream server and client initialization to use NATS if configured, falling back to the standard HTTP client otherwise (eth/backend.go, zk/datastream/client/stream_client.go) [1] [2] [3] [4].

Refactoring and abstraction improvements:

Refactored the datastream client and server to use interface types, improving flexibility and enabling NATS integration (eth/backend.go, turbo/stages/zk_stages.go, zk/datastream/client/stream_client.go) [1] [2] [3] [4].
Split NewClient into NewClient (returns interface) and NewStreamClient (returns concrete type), and updated tests to use the correct constructor (zk/datastream/client/stream_client.go, zk/datastream/client/stream_client_test.go) [1] [2] [3] [4] [5] [6] [7] [8].

Dependency updates:

Updated and added dependencies in go.mod to support NATS and other improvements [1] [2] [3] [4] [5].

Minor improvements:

Improved logging and prewarm flag behavior in datastream server initialization (eth/backend.go).

IvanBelyakoff

I did my best to look through the files. Amazing job!
Given the amount of tests, I believe it is much better tested than what we had!

* Initial plan * Update zkevm-data-streamer to v0.2.11 and Go to 1.25 Co-authored-by: tamingchaos <6331188+tamingchaos@users.noreply.github.com> * Remove accidentally committed binary Co-authored-by: tamingchaos <6331188+tamingchaos@users.noreply.github.com> --------- Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com> Co-authored-by: tamingchaos <6331188+tamingchaos@users.noreply.github.com>

This commit introduces a suite of tests for NATS JetStream, covering various functionalities including: - Basic server operations and message publishing - Server-client communication setup - Bidirectional message exchange - Cluster replication and fault tolerance - Message ordering verification - Persistence of messages across server restarts These tests ensure the reliability and performance of NATS as an embedded service within the Erigon datastreaming infrastructure.

…d JetStream replication

…treaming - Implement NATS client with full DatastreamClient interface compatibility - Add state machines for reliable L2 block processing and message ordering - Support bookmark-based stream positioning via NATS KV store - Include comprehensive test suite with e2e, behavior, and interface tests - Add configuration for NATS connection (host, port, TLS support) - Document state machines and architecture with Mermaid diagrams - Integrate NATS client into stage_batches with factory pattern

… metadata management

Copilot

Pull Request Overview

This PR introduces NATS JetStream as an alternative datastreaming infrastructure for Erigon's zkEVM implementation, moving away from the file-based datastreamer approach. The changes include:

Implementation of a NATS-based datastream client and server that wraps the existing file-based implementation
Comprehensive test suite for NATS functionality including cluster configurations, persistence, and various topologies
Integration of NATS metadata management using JetStream KV store for tracking stream state
Factory pattern for creating NATS-enabled stream servers

Reviewed Changes

Copilot reviewed 64 out of 65 changed files in this pull request and generated 3 comments.

Show a summary per file

File	Description
zk/stages/test_utils.go	Adds ExecutePerFile stub method to TestDatastreamClient interface
zk/stages/stage_batches_test.go	Updates type reference from local DatastreamClient to types.DatastreamClient
zk/stages/stage_batches_datastream.go	Moves isReading check earlier in StartRead and updates type references
zk/stages/stage_batches.go	Extracts DatastreamClient interface to types package and updates imports
zk/datastream/types/client.go	Defines DatastreamClient interface with ExecutePerFile method
zk/datastream/test/nats/*.go	Comprehensive test suite for NATS functionality
zk/datastream/natsstream/*.go	NATS implementation files including client, server, metadata management
zk/datastream/server/*.go	Updates factory interface signature and adds version parameter comment
zk/datastream/test/data_stream_compare/test_datastream_compare.go	Renames NewClient to NewStreamClient
turbo/cli/default_flags.go	Adds new command-line flags for NATS configuration

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

Copilot · 2025-11-06T10:45:09Z

+	if r.isReading.Load() {
+		return fmt.Errorf("tried starting datastream client runner thread while another is running")
+	}
+
 	if diffBlock > client.DefaultEntryChannelSize {


The reordering of the isReading check is good, but there's a race condition: between checking isReading (line 29) and setting it later (line 38 in the original, not shown but implied), another goroutine could start. Consider using atomic.CompareAndSwap to atomically check and set the flag to prevent this race condition.

Copilot · 2025-11-06T10:45:10Z


 func (f *ZkEVMDataStreamServerFactory) CreateStreamServer(port uint16, systemID uint64, streamType datastreamer.StreamType, fileName string, writeTimeout time.Duration, inactivityTimeout time.Duration, inactivityCheckInterval time.Duration, cfg *dslog.Config) (StreamServer, error) {
-	// after we moved to protobuff encoding we no longer need to support multiple versions.
+	// The version parameter from the interface is ignored as we've hardcoded the version to 3


The comment states that the version parameter is ignored, but the method signature in the interface (line 57 of interfaces.go) doesn't include a version parameter anymore. This comment appears to be outdated and should be updated or removed to avoid confusion.

Copilot · 2025-11-06T10:45:10Z

+func (n *NATSStreamServer) GetDataBetweenBookmarks(bookmarkFrom, bookmarkTo []byte) ([]byte, error) {
+	// TODO: This method appears to be unused and should be removed from the interface
+	// after the original datastreamer is retired. We're not implementing it for NATS.
+	panic("GetDataBetweenBookmarks is not implemented for NATS and appears to be unused. " +
+		"This method should be removed from the interface after datastreamer is fully retired.")
+}


Using panic for unimplemented methods in production code is problematic. If this method is called, it will crash the application. Consider returning an error like fmt.Errorf('GetDataBetweenBookmarks is not implemented for NATS') instead, which allows callers to handle the situation gracefully.

Resolves three issues identified by GitHub Copilot PR review on PR 0xPolygon#1966. ## 1. Fix race condition in DatastreamClientRunner.StartRead **File:** zk/stages/stage_batches_datastream.go **Issue:** TOCTOU (time-of-check-to-time-of-use) race condition between checking isReading flag and setting it inside goroutine. **Fix:** Use atomic.CompareAndSwap to atomically check and set the flag: - Changed from Load() + Store() pattern to CompareAndSwap(false, true) - Removed redundant Store(true) from inside goroutine - Ensures only one goroutine can start reading at a time **Impact:** Prevents potential race where multiple goroutines could start reading simultaneously. ## 2. Update outdated comment about version parameter **File:** zk/datastream/server/data_stream_server.go **Issue:** Comment referenced non-existent "version parameter from interface" that was removed in earlier refactoring. **Fix:** Updated comment to explain why version 3 is hardcoded: - Before: "The version parameter from the interface is ignored..." - After: "Version 3 is hardcoded for protobuf encoding support" **Impact:** Improves code clarity and removes confusion. ## 3. Replace panic with error return **File:** zk/datastream/natsstream/stream_server.go **Issue:** GetDataBetweenBookmarks() used panic() for unimplemented method, which would crash the application if called. **Fix:** Return proper error instead of panic: - Changed: panic("...") - To: return nil, fmt.Errorf("...") **Impact:** Allows callers to handle unimplemented method gracefully instead of crashing. Method is marked for removal after datastreamer retirement.

tamingchaos force-pushed the nats-research branch from 5e001ec to ff6bce2 Compare August 12, 2025 14:47

IvanBelyakoff previously approved these changes Aug 14, 2025

View reviewed changes

tamingchaos force-pushed the nats-research branch from ff6bce2 to 5fac91c Compare September 1, 2025 12:25

tamingchaos dismissed IvanBelyakoff’s stale review via c34a6b4 September 8, 2025 17:41

tamingchaos force-pushed the nats-research branch from 5fac91c to c34a6b4 Compare September 8, 2025 17:41

tamingchaos force-pushed the nats-research branch from c34a6b4 to b22334e Compare October 18, 2025 13:54

Copilot AI and others added 12 commits October 31, 2025 11:13

feat(nats): implement embedded NATS server management

be04cbf

feat(nats): created shims to test publishing data to NATS

88eea64

feat(tests): add comprehensive tests for NATS cluster, leaf nodes, an…

083fab6

…d JetStream replication

feat(nats): updated server implementation to meet DS parity

10c72c0

First cut full feature complete Nats

93ebb95

test fixes

0b2f9e9

ensure Nats is enabled before initialising

2eb67d1

add consolidated nats documentation

525487c

feat(migration): implement TCP to NATS datastream migration tool with…

48d4aa8

… metadata management

tamingchaos force-pushed the nats-research branch from 59a1355 to 48d4aa8 Compare October 31, 2025 11:43

tamingchaos added 2 commits November 3, 2025 17:42

fix: ensure NATS manager is not nil before stopping

c91e9b8

chore: update zkevm-data-streamer dependency to v0.2.13

6a093fc

tamingchaos marked this pull request as ready for review November 6, 2025 10:35

Copilot AI review requested due to automatic review settings November 6, 2025 10:35

tamingchaos requested review from eduadiez, hexoscott, mandrigin and revitteth as code owners November 6, 2025 10:35

Copilot AI reviewed Nov 6, 2025

View reviewed changes

tamingchaos added 2 commits November 6, 2025 17:50

Merge branch 'zkevm' into nats-research

39cc46f

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

Nats Data Stream#1966

Nats Data Stream#1966
tamingchaos wants to merge 16 commits into
0xPolygon:zkevmfrom
tamingchaos:nats-research

tamingchaos commented Jul 17, 2025 •

edited

Loading

Uh oh!

IvanBelyakoff left a comment

Uh oh!

Copilot AI left a comment

Uh oh!

Copilot AI Nov 6, 2025

Uh oh!

Copilot AI Nov 6, 2025

Uh oh!

Copilot AI Nov 6, 2025

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

Uh oh!

Conversation

tamingchaos commented Jul 17, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

IvanBelyakoff left a comment

Choose a reason for hiding this comment

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull Request Overview

Reviewed Changes

Uh oh!

Copilot AI Nov 6, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI Nov 6, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI Nov 6, 2025

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

tamingchaos commented Jul 17, 2025 •

edited

Loading