fix: P0 correctness bugs (halt detection, hangs, panic, docs)

README.md

@@ -8,7 +8,7 @@ Data Availability metrics and monitoring tool for evstack using Celestia DA. Thi
 - Automatic verification of header and data blobs on Celestia
 - Retry logic with exponential backoff for pending DA submissions
 - Prometheus metrics for tracking unverified block ranges
-- Support for both streaming and one-shot block verification modes
+- Streaming block verification of every new block
 - Account balance monitoring via Celestia consensus RPC with automatic failover
 
 ## Quick Start
@@ -38,13 +38,14 @@ The monitor command streams EVM block headers and verifies DA submission on Cele
 ```
 
 
-### Enable Prometheus Metrics
+### Prometheus Metrics Endpoint
+
+The Prometheus metrics server always runs. Use `--port` to change the port:
 
 ```bash
 ./ev-metrics monitor \
   --header-namespace collect_testnet_header \
   --data-namespace collect_testnet_data \
-  --enable-metrics \
   --port 2112
 ```
 
@@ -66,7 +67,6 @@ Metrics will be available at `http://localhost:2112/metrics`
 - `--celestia-token`: Celestia authentication token (optional), sent as a `Bearer` token on all Celestia requests, including block timestamp lookups. Required when the celestia-node has auth enabled (the default).
 - `--duration`: Duration in seconds to stream (0 = infinite)
 - `--chain-id`: Chain identifier for metrics labels (default: "testnet")
-- `--enable-metrics`: Enable Prometheus metrics HTTP server (default: false)
 - `--port`: HTTP server port for metrics (default: 2112)
 - `--jsonrpc-scrape-interval`: JSON-RPC health check scrape interval in seconds (default: 10)
 - `--reference-node`: Reference node RPC endpoint URL (sequencer) for drift monitoring
@@ -76,8 +76,7 @@ Metrics will be available at `http://localhost:2112/metrics`
 - `--balance.consensus-rpc-urls`: Comma-separated consensus RPC URLs for balance queries (required if balance.addresses is set)
 - `--balance.scrape-interval`: Balance check scrape interval in seconds (default: 30)
 - `--verifier.workers`: Number of concurrent workers for block verification (default: 50)
-- `--verbose`: Enable verbose logging (default: false)
-- `--debug`: Enable debug logging for submission details (default: false, can also be set via `EVMETRICS_DEBUG=true` environment variable)
+- `--verbose`: Enable verbose (debug level) logging (default: false)
 
 ### Example with Custom Endpoints
 
@@ -100,7 +99,6 @@ Enable JSON-RPC request duration monitoring by providing the `--evm-rpc-url` fla
   --data-namespace collect_testnet_data \
   --evm-ws-url "ws://localhost:8546" \
   --evm-rpc-url "http://localhost:8545" \
-  --enable-metrics \
   --jsonrpc-scrape-interval 10
 ```
 
@@ -116,15 +114,14 @@ Monitor native token balances for one or more Celestia addresses:
   --data-namespace collect_testnet_data \
   --balance.addresses "celestia1abc...,celestia1def..." \
   --balance.consensus-rpc-urls "https://rpc.celestia.org,https://rpc-mocha.pops.one" \
-  --balance.scrape-interval 30 \
-  --enable-metrics
+  --balance.scrape-interval 30
 ```
 
 This will query account balances every 30 seconds with automatic failover between RPC endpoints.
 
 ## Prometheus Metrics
 
-When metrics are enabled, the following metrics are exposed:
+The following metrics are exposed:
 
 ### `ev_metrics_unsubmitted_block_range_start`
 - **Type**: Gauge
@@ -194,7 +191,7 @@ When metrics are enabled, the following metrics are exposed:
 ### `ev_metrics_block_time_slo_seconds`
 - **Type**: Gauge
 - **Labels**: `chain_id`, `quantile`
-- **Description**: SLO thresholds for block time
+- **Description**: SLO thresholds for block time (not yet emitted, see #44)
 - **Values**:
   - `0.5`: 2.0s
   - `0.9`: 3.0s
@@ -204,13 +201,13 @@ When metrics are enabled, the following metrics are exposed:
 ### `ev_metrics_block_receive_delay_seconds`
 - **Type**: Histogram
 - **Labels**: `chain_id`
-- **Description**: Delay between block creation and reception with histogram buckets
+- **Description**: Delay between block creation and reception with histogram buckets (not yet emitted, see #44)
 - **Buckets**: 0.1, 0.25, 0.5, 1.0, 2.0, 3.0, 5.0, 10.0, 15.0, 30.0, 60.0 seconds
 
 ### `ev_metrics_block_receive_delay_slo_seconds`
 - **Type**: Gauge
 - **Labels**: `chain_id`, `quantile`
-- **Description**: SLO thresholds for block receive delay
+- **Description**: SLO thresholds for block receive delay (not yet emitted, see #44)
 - **Values**:
   - `0.5`: 1.0s
   - `0.9`: 3.0s
@@ -277,11 +274,14 @@ When `--balance.addresses` and `--balance.consensus-rpc-urls` are provided:
 
 ## Debug Logging
 
-Debug logging provides detailed visibility into DA submission process. Enable with `--debug` flag or `EVMETRICS_DEBUG=true` environment variable.
+There are two independent debug logging mechanisms:
+
+- `--verbose` flag: sets the application log level to debug.
+- `EVMETRICS_DEBUG=true` environment variable: logs DA submission details from the metrics package.
 
-**Use cases**: Troubleshoot submission failures, verify submission flow, diagnose Celestia RPC issues.
+**Use cases**: Troubleshoot submission failures, verify submission flow.
 
-**Logs include**: Successful/failed submissions, DA height updates, submission timing.
+**`EVMETRICS_DEBUG` logs include**: Successful/failed submission attempts and DA height updates.
 
 **Example**:
 ```bash

cmd/monitor.go

@@ -92,7 +92,7 @@ func NewMonitorCmd() *cobra.Command {
 	cmd.Flags().StringVar(&flags.dataNS, flagDataNS, "", "Data namespace (my_app_data_namespace)")
 	cmd.Flags().IntVar(&flags.duration, flagDuration, 0, "Duration in seconds to stream (0 = infinite)")
 	cmd.Flags().BoolVar(&flags.verbose, flagVerbose, false, "Enable verbose logging")
-	cmd.Flags().IntVar(&flags.port, flagPort, 2112, "HTTP server port for metrics (only used if --enable-metrics is set)")
+	cmd.Flags().IntVar(&flags.port, flagPort, 2112, "HTTP server port for metrics")
 	cmd.Flags().StringVar(&flags.chainID, flagChain, "testnet", "chainID identifier for metrics labels")
 	cmd.Flags().StringVar(&flags.referenceNode, flagReferenceNode, "", "Reference node RPC endpoint URL (sequencer) for drift monitoring")
 	cmd.Flags().StringVar(&flags.fullNodes, flagFullNodes, "", "Comma-separated list of full node RPC endpoint URLs for drift monitoring")
@@ -248,8 +248,8 @@ func monitorAndExportMetrics(_ *cobra.Command, _ []string) error {
 
 	err = metrics.StartServer(ctx, metricsPath, flags.port, logger, exporters...)
 
-	// context.Canceled is expected during graceful shutdown, not an error
-	if err != nil && !errors.Is(err, context.Canceled) {
+	// context.Canceled (shutdown) and context.DeadlineExceeded (--duration) are clean exits
+	if err != nil && !errors.Is(err, context.Canceled) && !errors.Is(err, context.DeadlineExceeded) {
 		return err
 	}
 

pkg/exporters/drift/drift.go

@@ -30,11 +30,12 @@ type exporter struct {
 	fullNodes       []string
 	pollingInterval int
 	logger          zerolog.Logger
+	lastHeight      uint64
 }
 
 // ExportMetrics continuously checks block heights of a reference node and multiple full nodes,
 // recording the heights and their drift into the provided metrics instance at specified intervals.
-func (e exporter) ExportMetrics(ctx context.Context, m *metrics.Metrics) error {
+func (e *exporter) ExportMetrics(ctx context.Context, m *metrics.Metrics) error {
 	ticker := time.NewTicker(time.Duration(e.pollingInterval) * time.Second)
 	defer ticker.Stop()
 
@@ -63,7 +64,10 @@ func (e exporter) ExportMetrics(ctx context.Context, m *metrics.Metrics) error {
 			m.RecordReferenceBlockHeight(e.chainID, e.referenceNode, refHeight)
 			// Also update time since last block metric when we see new blocks via polling
 			// This ensures the metric stays accurate even if WebSocket subscription fails
-			m.UpdateLastBlockTime(e.chainID, time.Now())
+			if refHeight > e.lastHeight {
+				m.UpdateLastBlockTime(e.chainID, time.Now())
+				e.lastHeight = refHeight
+			}
 			e.logger.Info().Uint64("height", refHeight).Str("endpoint", e.referenceNode).Msg("recorded reference node height")
 
 			// get each full node height and calculate drift
@@ -80,10 +84,6 @@ func (e exporter) ExportMetrics(ctx context.Context, m *metrics.Metrics) error {
 				m.RecordCurrentBlockHeight(e.chainID, fullNode, currentHeight)
 				m.RecordBlockHeightDrift(e.chainID, fullNode, refHeight, currentHeight)
 
-				// Also update time since last block metric when we see new blocks via polling
-				// This ensures the metric stays accurate even if WebSocket subscription fails
-				m.UpdateLastBlockTime(e.chainID, time.Now())
-
 				drift := int64(refHeight) - int64(currentHeight)
 				e.logger.Info().
 					Uint64("ref_height", refHeight).

pkg/exporters/drift/drift_test.go

@@ -0,0 +1,82 @@
+package drift
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/evstack/ev-metrics/pkg/metrics"
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/rs/zerolog"
+	"github.com/stretchr/testify/require"
+)
+
+func newBlockNumberServer(t *testing.T, height *atomic.Uint64) *httptest.Server {
+	t.Helper()
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		var req struct {
+			ID json.RawMessage `json:"id"`
+		}
+		if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
+			http.Error(w, err.Error(), http.StatusBadRequest)
+			return
+		}
+		w.Header().Set("Content-Type", "application/json")
+		_, _ = fmt.Fprintf(w, `{"jsonrpc":"2.0","id":%s,"result":"0x%x"}`, req.ID, height.Load())
+	}))
+	t.Cleanup(srv.Close)
+	return srv
+}
+
+func TestExportMetrics_TimeSinceLastBlockOnlyResetsWhenHeightAdvances(t *testing.T) {
+	var height atomic.Uint64
+	height.Store(10)
+	srv := newBlockNumberServer(t, &height)
+
+	reg := prometheus.NewRegistry()
+	m := metrics.NewWithRegistry("test", reg)
+	e := NewMetricsExporter("chain", srv.URL, nil, 1, zerolog.Nop())
+
+	ctx, cancel := context.WithCancel(context.Background())
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		_ = e.ExportMetrics(ctx, m)
+	}()
+	t.Cleanup(func() {
+		cancel()
+		<-done
+	})
+
+	timeSince := func() float64 {
+		m.UpdateTimeSinceLastBlock()
+		families, err := reg.Gather()
+		require.NoError(t, err)
+		for _, mf := range families {
+			if mf.GetName() == "test_time_since_last_block_seconds" {
+				return mf.GetMetric()[0].GetGauge().GetValue()
+			}
+		}
+		t.Fatal("time_since_last_block_seconds not found")
+		return 0
+	}
+
+	// polls at ~1s (height 10, first seen) and ~2s (height unchanged)
+	time.Sleep(2500 * time.Millisecond)
+	require.GreaterOrEqual(t, timeSince(), 1.2, "constant height must not reset the gauge")
+
+	// poll at ~3s sees a lower height
+	height.Store(5)
+	time.Sleep(time.Second)
+	require.GreaterOrEqual(t, timeSince(), 2.2, "decreasing height must not reset the gauge")
+
+	// poll at ~4s sees a new height
+	height.Store(11)
+	time.Sleep(time.Second)
+	require.Less(t, timeSince(), 1.0, "advancing height must reset the gauge")
+}

pkg/exporters/verifier/verifier.go

@@ -86,7 +86,6 @@ func (e *exporter) ExportMetrics(ctx context.Context, m *metrics.Metrics) error
 	// It captures sub by reference so reassignment during reconnection is reflected.
 	shutdown := func() {
 		sub.Unsubscribe()
-		close(blockQueue)
 		workerGroup.Wait()
 	}
 
@@ -108,7 +107,6 @@ func (e *exporter) ExportMetrics(ctx context.Context, m *metrics.Metrics) error
 			newSub := e.reconnectSubscription(ctx, headers)
 			if newSub == nil {
 				// context was cancelled during reconnection
-				close(blockQueue)
 				workerGroup.Wait()
 				return nil
 			}
@@ -173,26 +171,32 @@ func (e *exporter) processBlocks(ctx context.Context, m *metrics.Metrics, worker
 	logger := e.logger.With().Int("worker_id", workerID).Logger()
 	logger.Debug().Msg("worker started")
 
-	for header := range blockQueue {
-		reEnqueue := e.verifyBlock(ctx, m, header)
-		if reEnqueue {
-			// re-queue with delay in background
-			go func(h *types.Header) {
-				select {
-				case <-time.After(5 * time.Minute):
+	// blockQueue is never closed: delayed re-queue goroutines may still send to it,
+	// so workers exit on ctx cancellation instead.
+	for {
+		select {
+		case <-ctx.Done():
+			logger.Debug().Msg("worker stopped")
+			return
+		case header := <-blockQueue:
+			reEnqueue := e.verifyBlock(ctx, m, header)
+			if reEnqueue {
+				// re-queue with delay in background
+				go func(h *types.Header) {
 					select {
-					case blockQueue <- h:
-						logger.Debug().Uint64("block", h.Number.Uint64()).Msg("re-queued block after cooldown")
+					case <-time.After(5 * time.Minute):
+						select {
+						case blockQueue <- h:
+							logger.Debug().Uint64("block", h.Number.Uint64()).Msg("re-queued block after cooldown")
+						case <-ctx.Done():
+							// graceful shutdown, don't send
+						}
 					case <-ctx.Done():
-						// graceful shutdown, don't send
 					}
-				case <-ctx.Done():
-				}
-			}(header)
+				}(header)
+			}
 		}
 	}
-
-	logger.Debug().Msg("worker stopped")
 }
 
 func (e *exporter) onVerified(m *metrics.Metrics, namespace string, blockHeight, daHeight uint64, verified bool, submissionDuration time.Duration) {

pkg/exporters/verifier/verifier_test.go

@@ -0,0 +1,30 @@
+package verifier
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/ethereum/go-ethereum/core/types"
+	"github.com/rs/zerolog"
+)
+
+func TestProcessBlocksStopsOnContextCancel(t *testing.T) {
+	e := &exporter{logger: zerolog.Nop()}
+	queue := make(chan *types.Header, 1)
+	ctx, cancel := context.WithCancel(context.Background())
+
+	done := make(chan struct{})
+	go func() {
+		e.processBlocks(ctx, nil, 0, queue)
+		close(done)
+	}()
+
+	cancel()
+
+	select {
+	case <-done:
+	case <-time.After(time.Second):
+		t.Fatal("processBlocks did not return after context cancellation")
+	}
+}

pkg/metrics/server.go

@@ -33,7 +33,8 @@ func StartServer(ctx context.Context, metricsPath string, port int, logger zerol
 	// initialize metrics (thread-safe, shared across all exporters)
 	m := New("ev_metrics")
 
-	var g errgroup.Group
+	// derived ctx is cancelled when any exporter or the server fails, stopping everything else
+	g, ctx := errgroup.WithContext(ctx)
 
 	// start all metric exporters
 	for _, exporter := range exporters {

pkg/metrics/server_test.go

@@ -0,0 +1,36 @@
+package metrics
+
+import (
+	"context"
+	"errors"
+	"net"
+	"testing"
+	"time"
+
+	"github.com/rs/zerolog"
+	"github.com/stretchr/testify/require"
+)
+
+type failingExporter struct{ err error }
+
+func (f failingExporter) ExportMetrics(context.Context, *Metrics) error { return f.err }
+
+func TestStartServer_ExporterErrorStopsServer(t *testing.T) {
+	l, err := net.Listen("tcp", ":0")
+	require.NoError(t, err)
+	port := l.Addr().(*net.TCPAddr).Port
+	require.NoError(t, l.Close())
+
+	wantErr := errors.New("subscribe failed")
+	done := make(chan error, 1)
+	go func() {
+		done <- StartServer(context.Background(), "/metrics", port, zerolog.Nop(), failingExporter{err: wantErr})
+	}()
+
+	select {
+	case err := <-done:
+		require.ErrorIs(t, err, wantErr)
+	case <-time.After(10 * time.Second):
+		t.Fatal("StartServer did not return after exporter failure")
+	}
+}