Keep chaintracks isolated in microservice deployment (#156)

Author: galt-tr

app/app.go

@@ -162,13 +162,19 @@ func Bootstrap(ctx context.Context, cfg *config.Config, logger *zap.Logger) (*De
 
 	teranodeClient.Start(ctx)
 
-	// Construct chaintracks once at process startup so every consumer
-	// (chaintracks_server, bump-builder canonical-root validation, watchdog
-	// future use) shares one P2P subscription and header cache. Skipped when
-	// chaintracks_server is disabled — that flag is already the operator's
-	// process-wide "no chaintracks" switch (regtest force-disables it).
+	// Construct chaintracks once at process startup so every in-process
+	// consumer (chaintracks_server, bump-builder canonical-root validation)
+	// shares one P2P subscription and header cache. Skipped when:
+	//   - cfg.ChaintracksServer.Enabled is false (operator-wide off switch;
+	//     regtest force-disables it via config.validate), or
+	//   - the configured mode never dereferences deps.Chaintracks. Without
+	//     this gate every microservice pod (api-server, sse, propagation, …)
+	//     would spin up an embedded ChainManager and poll the upstream node
+	//     even though nothing in that pod reads from it. In microservice
+	//     deployments the chaintracks pod runs embedded and bump-builder
+	//     points chaintracks.mode=remote at it; everything else skips here.
 	var chainTracks chaintrackslib.Chaintracks
-	if cfg.ChaintracksServer.Enabled {
+	if cfg.ChaintracksServer.Enabled && modeNeedsChaintracks(cfg.Mode) {
 		ct, ctErr := initChaintracks(ctx, cfg, logger)
 		if ctErr != nil {
 			_ = st.Close()
@@ -233,6 +239,25 @@ func Bootstrap(ctx context.Context, cfg *config.Config, logger *zap.Logger) (*De
 	return deps, cleanup, nil
 }
 
+// modeNeedsChaintracks reports whether the configured service mode constructs
+// at least one service that reads from deps.Chaintracks. Other modes
+// (api-server, sse, propagation, p2p-client, watchdog) never dereference it,
+// so initializing it would spin up an unused embedded ChainManager + upstream
+// header poller in every pod — which is exactly the wrong thing for
+// microservice deployments where the dedicated chaintracks pod owns the
+// header store and bump-builder reads via chaintracks.mode=remote.
+//
+// Keep this list in sync with BuildServices: any new mode that wires
+// deps.Chaintracks into a service must be added here.
+func modeNeedsChaintracks(mode string) bool {
+	switch mode {
+	case "all", "chaintracks", "bump-builder":
+		return true
+	default:
+		return false
+	}
+}
+
 // initChaintracks brings up the embedded go-chaintracks instance shared
 // across the process. Caller gates the enabled-ness check; this function
 // always tries to construct and returns an error on failure.

app/app_test.go

@@ -0,0 +1,28 @@
+package app
+
+import "testing"
+
+func TestModeNeedsChaintracks(t *testing.T) {
+	cases := []struct {
+		mode string
+		want bool
+	}{
+		{"all", true},
+		{"chaintracks", true},
+		{"bump-builder", true},
+		{"api-server", false},
+		{"sse", false},
+		{"propagation", false},
+		{"p2p-client", false},
+		{"watchdog", false},
+		{"", false},
+		{"unknown-mode", false},
+	}
+	for _, tc := range cases {
+		t.Run(tc.mode, func(t *testing.T) {
+			if got := modeNeedsChaintracks(tc.mode); got != tc.want {
+				t.Errorf("modeNeedsChaintracks(%q) = %v, want %v", tc.mode, got, tc.want)
+			}
+		})
+	}
+}

config.example.yaml

@@ -181,8 +181,19 @@ chaintracks_server:
 
 # go-chaintracks library config. See that library's docs for the full set.
 # Defaults are inherited from the library's SetDefaults — uncomment to override.
+#
+# In single-binary (mode=all) and the dedicated chaintracks pod, leave mode at
+# the default "embedded" so this process owns the header store and the P2P
+# subscription. In a microservice deployment every OTHER consumer pod
+# (bump-builder) should run mode="remote" with url pointed at the chaintracks
+# pod's HTTP endpoint — otherwise each pod spins up its own embedded
+# ChainManager and hits the upstream node N times. arcade additionally skips
+# the chaintracks library entirely in modes that don't read from it
+# (api-server, sse, propagation, p2p-client, watchdog), so those pods need no
+# chaintracks config at all.
 # chaintracks:
-#   mode: embedded         # or "remote" with url set to a running chaintracks
-#   storage_path: ""       # defaults to <storage_path>/chaintracks
-#   bootstrap_url: ""
-#   bootstrap_mode: api
+#   mode: embedded         # or "remote" with url set
+#   url: ""                # required when mode=remote, e.g. http://chaintracks:8083
+#   storage_path: ""       # embedded mode only; defaults to <storage_path>/chaintracks
+#   bootstrap_url: ""      # embedded mode only
+#   bootstrap_mode: api    # embedded mode only

deploy/bump-builder.yaml

@@ -39,6 +39,13 @@ spec:
               value: "aerospike-0.aerospike.arcade.svc.cluster.local:3000"
             - name: ARCADE_AEROSPIKE_NAMESPACE
               value: "arcade"
+            # Talk to the chaintracks pod over HTTP instead of running an
+            # embedded ChainManager + upstream header poller in this pod.
+            # The chaintracks Service exposes /chaintracks/v1/* on 8083.
+            - name: ARCADE_CHAINTRACKS_MODE
+              value: "remote"
+            - name: ARCADE_CHAINTRACKS_URL
+              value: "http://chaintracks.arcade.svc.cluster.local:8083"
           readinessProbe:
             httpGet:
               path: /health

services/propagation/propagator.go

@@ -46,7 +46,11 @@ type Propagator struct {
 	leaser         store.Leaser
 	teranodeClient *teranode.Client
 	merkleClient   *merkleservice.Client
-	consumer       *kafka.ConsumerGroup
+	// consumer is written by Start (after kafka.NewConsumerGroup returns) and
+	// read by Stop. atomic.Pointer makes that handoff race-free without
+	// needing a mutex: in tests the harness can call Stop concurrently with
+	// an in-flight Start that's still blocked initializing the consumer.
+	consumer atomic.Pointer[kafka.ConsumerGroup]
 
 	maxPending int
 	// admitCh, requeueCh, terminalCh, and drainCh feed runDispatcher's
@@ -470,7 +474,7 @@ func (p *Propagator) Start(ctx context.Context) error {
 	if err != nil {
 		return fmt.Errorf("creating consumer group: %w", err)
 	}
-	p.consumer = consumer
+	p.consumer.Store(consumer)
 
 	// Spin up the persistent broadcast worker pool. Workers exit when
 	// Stop() closes the job channel; the WaitGroup lets Stop() block until
@@ -547,8 +551,8 @@ func (p *Propagator) WaitForBatches() {
 func (p *Propagator) Stop() error {
 	p.logger.Info("stopping propagation service")
 	var consumerErr error
-	if p.consumer != nil {
-		consumerErr = p.consumer.Close()
+	if c := p.consumer.Load(); c != nil {
+		consumerErr = c.Close()
 	}
 	// Wait for in-flight processBatch goroutines to finish before tearing
 	// down the broadcast worker pool. Otherwise an in-flight batch would

store/postgres/postgres.go

@@ -770,19 +770,31 @@ RETURNING t.txid, prev.status, prev.timestamp_at, prev.block_hash, prev.block_he
 			txid          string
 			prevStatus    string
 			prevTimestamp time.Time
-			prevBlockHash string
-			prevHeight    int64
+			// transactions.block_hash / block_height are nullable for any
+			// pre-MINED status (RECEIVED / SENT_TO_NETWORK / SEEN_*). The CTE
+			// captures the pre-update row so these will be NULL whenever a tx
+			// transitions to MINED for the first time. Scan via pointer so a
+			// NULL doesn't fail the scan; an empty BlockHash on the returned
+			// prev snapshot is the correct "no block yet" representation
+			// (matches models.TransactionStatus zero values used elsewhere).
+			prevBlockHash *string
+			prevHeight    *int64
 		)
 		if err := rows.Scan(&txid, &prevStatus, &prevTimestamp, &prevBlockHash, &prevHeight); err != nil {
 			return prevs, out, err
 		}
-		prevs = append(prevs, &models.TransactionStatus{
-			TxID:        txid,
-			Status:      models.Status(prevStatus),
-			Timestamp:   prevTimestamp,
-			BlockHash:   prevBlockHash,
-			BlockHeight: uint64(prevHeight), //nolint:gosec // value originated as uint64 in this column
-		})
+		prev := &models.TransactionStatus{
+			TxID:      txid,
+			Status:    models.Status(prevStatus),
+			Timestamp: prevTimestamp,
+		}
+		if prevBlockHash != nil {
+			prev.BlockHash = *prevBlockHash
+		}
+		if prevHeight != nil {
+			prev.BlockHeight = uint64(*prevHeight) //nolint:gosec // value originated as uint64 in this column
+		}
+		prevs = append(prevs, prev)
 		out = append(out, &models.TransactionStatus{
 			TxID:        txid,
 			Status:      models.StatusMined,

store/postgres/postgres_test.go

@@ -1062,3 +1062,36 @@ func hashesOf(rows []*models.BlockProcessingStatus) []string {
 	}
 	return out
 }
+
+// Regression: a tx at any pre-MINED status has NULL block_hash / block_height,
+// so the CTE in SetMinedByTxIDs returns NULLs for the previous-row snapshot.
+// Scanning those NULLs into bare string/int64 used to error with
+// "cannot scan NULL into *string", which then caused bump-builder to log
+// "failed to set mined status" and drop the MINED transition entirely.
+func TestSetMinedByTxIDs_HandlesNullPrevBlock(t *testing.T) {
+	s := newTestStore(t)
+	ctx := context.Background()
+
+	in := &models.TransactionStatus{TxID: "tx-prev-null", Status: models.StatusReceived}
+	if _, _, err := s.GetOrInsertStatus(ctx, in); err != nil {
+		t.Fatalf("seed: %v", err)
+	}
+
+	prevs, mined, err := s.SetMinedByTxIDs(ctx, "blockhashX", 42, []string{"tx-prev-null"})
+	if err != nil {
+		t.Fatalf("SetMinedByTxIDs: %v", err)
+	}
+	if len(prevs) != 1 || len(mined) != 1 {
+		t.Fatalf("prevs=%d mined=%d want 1/1", len(prevs), len(mined))
+	}
+	if prevs[0].Status != models.StatusReceived {
+		t.Errorf("prev status = %q, want RECEIVED", prevs[0].Status)
+	}
+	if prevs[0].BlockHash != "" || prevs[0].BlockHeight != 0 {
+		t.Errorf("prev block fields should be zero for pre-MINED row, got hash=%q height=%d",
+			prevs[0].BlockHash, prevs[0].BlockHeight)
+	}
+	if mined[0].BlockHash != "blockhashX" || mined[0].BlockHeight != 42 {
+		t.Errorf("mined snapshot mismatch: %+v", mined[0])
+	}
+}