fix

.github/workflows/ci_zkevm.yml

@@ -116,10 +116,22 @@ jobs:
         working-directory: ./kurtosis-cdk
         run: kurtosis run --enclave cdk-v1 --args-file params.yml --image-download always .
 
-      - name: Setup CPU monitoring script
+      - name: Run process with CPU monitoring
+        working-directory: ./cdk-erigon
         run: |
-          bash ./cdk-erigon/.github/scripts/cpu_monitor.sh &
-          echo $! > /tmp/monitor.pid
+          # Start monitoring in background
+          bash ./github/scripts/cpu_monitor.sh &
+          monitor_pid=$!
+
+          # Wait for 30 seconds
+          sleep 30
+
+          # Stop monitoring and get analysis
+          kill -TERM $monitor_pid
+          wait $monitor_pid || {
+            echo "CPU usage exceeded threshold!"
+            exit 1
+          }
 
       - name: Monitor verified batches
         working-directory: ./kurtosis-cdk
@@ -162,23 +174,6 @@ jobs:
           EOF
           docker run --network=host  --volume "./tmp/:/config/" --env BRIDGE_TEST_CONFIG_FILE=/config/test.toml bridge-e2e-realnetwork-erc20
 
-      # Analysis step that runs at the end
-      - name: Analyze CPU usage
-        run: |
-          if [ -f /tmp/monitor.pid ]; then
-            monitor_pid=$(cat /tmp/monitor.pid)
-            # Send SIGTERM to trigger the cleanup and analysis
-            kill -TERM $monitor_pid
-            # Wait for the analysis to complete
-            wait $monitor_pid || {
-              echo "CPU usage exceeded threshold!"
-              exit 1
-            }
-          else
-            echo "Monitor PID file not found!"
-            exit 1
-          fi
-
       - name: Upload logs
         uses: actions/upload-artifact@v3
         with:

zk/stages/stage_batches_datastream.go

@@ -65,7 +65,6 @@ func (r *DatastreamClientRunner) StartRead() error {
 			case <-timeout.C:
 				log.Debug(fmt.Sprintf("[%s] Timeout waiting for stream to start", r.logPrefix))
 			}
-			timeout.Stop()
 		}
 	}()
 

zk/stages/test_utils.go

@@ -10,9 +10,9 @@ type TestDatastreamClient struct {
 	fullL2Blocks          []types.FullL2Block
 	gerUpdates            []types.GerUpdate
 	lastWrittenTimeAtomic atomic.Int64
-	streamingAtomic       atomic.Bool
 	progress              atomic.Uint64
 	entriesChan           chan interface{}
+	streamChan            chan interface{}
 	errChan               chan error
 	isStarted             bool
 }
@@ -23,8 +23,11 @@ func NewTestDatastreamClient(fullL2Blocks []types.FullL2Block, gerUpdates []type
 		gerUpdates:   gerUpdates,
 		entriesChan:  make(chan interface{}, 1000),
 		errChan:      make(chan error, 100),
+		streamChan:   make(chan interface{}, 1),
 	}
 
+	client.streamChan <- true
+
 	return client
 }
 
@@ -33,8 +36,7 @@ func (c *TestDatastreamClient) EnsureConnected() (bool, error) {
 }
 
 func (c *TestDatastreamClient) ReadAllEntriesToChannel() error {
-	c.streamingAtomic.Store(true)
-	defer c.streamingAtomic.Swap(false)
+	defer func() { c.streamChan <- true }()
 
 	for i := range c.fullL2Blocks {
 		c.entriesChan <- &c.fullL2Blocks[i]
@@ -76,7 +78,7 @@ func (c *TestDatastreamClient) GetLastWrittenTimeAtomic() *atomic.Int64 {
 }
 
 func (c *TestDatastreamClient) GetStreamChan() *chan interface{} {
-	return &c.entriesChan
+	return &c.streamChan
 }
 
 func (c *TestDatastreamClient) GetProgressAtomic() *atomic.Uint64 {