Integrate reprocess behavior and break out services (#140)

* Integrate reprocess behavior and break out services

* Potential fix for pull request finding 'CodeQL / Uncontrolled data used in path expression'

Co-authored-by: Copilot Autofix powered by AI <62310815+github-advanced-security[bot]@users.noreply.github.com>

* perf improvements

* Fix linting errors

* Updates for SSE drops

* Potential fix for pull request finding 'CodeQL / Uncontrolled data used in path expression'

Co-authored-by: Copilot Autofix powered by AI <62310815+github-advanced-security[bot]@users.noreply.github.com>

* ci: remove unused //nolint:gosec on context.WithCancel

golangci-lint 2.12.2 in CI no longer flags context.WithCancel patterns
where the cancel is stored for later invocation (gosec G115 / G505
relaxation), so the suppression at services/sse/manager.go:232 became
unused under nolintlint. The local 2.11.4 still required it, masking
the CI failure.

---------

Co-authored-by: Copilot Autofix powered by AI <62310815+github-advanced-security[bot]@users.noreply.github.com>

Author: galt-tr

app/app.go

@@ -11,8 +11,12 @@ package app
 import (
 	"context"
 	"fmt"
+	"os"
+	"path"
 	"time"
 
+	chaintrackslib "github.com/bsv-blockchain/go-chaintracks/chaintracks"
+	"github.com/bsv-blockchain/go-sdk/chainhash"
 	"go.uber.org/zap"
 
 	"github.com/bsv-blockchain/arcade/config"
@@ -22,9 +26,12 @@ import (
 	"github.com/bsv-blockchain/arcade/services"
 	"github.com/bsv-blockchain/arcade/services/api_server"
 	"github.com/bsv-blockchain/arcade/services/bump_builder"
+	"github.com/bsv-blockchain/arcade/services/chaintracks_server"
 	"github.com/bsv-blockchain/arcade/services/p2p_client"
 	"github.com/bsv-blockchain/arcade/services/propagation"
+	"github.com/bsv-blockchain/arcade/services/sse"
 	"github.com/bsv-blockchain/arcade/services/tx_validator"
+	"github.com/bsv-blockchain/arcade/services/watchdog"
 	"github.com/bsv-blockchain/arcade/services/webhook"
 	"github.com/bsv-blockchain/arcade/store"
 	storefactory "github.com/bsv-blockchain/arcade/store/factory"
@@ -46,6 +53,11 @@ type Deps struct {
 	TeranodeClient *teranode.Client
 	MerkleClient   *merkleservice.Client // nil when MerkleService.URL is unset
 	Validator      *validator.Validator
+	// Chaintracks is the shared in-process header tracker. nil when
+	// chaintracks_server is disabled (regtest, or explicit opt-out) — services
+	// that consume it (chaintracks_server, bump-builder canonical-root
+	// validation) must nil-guard.
+	Chaintracks chaintrackslib.Chaintracks
 }
 
 // Bootstrap creates every shared dependency the services rely on, in the
@@ -78,22 +90,28 @@ func Bootstrap(ctx context.Context, cfg *config.Config, logger *zap.Logger) (*De
 		_ = producer.Close()
 		return nil, nil, fmt.Errorf("creating store: %w", err)
 	}
-	if err := st.EnsureIndexes(); err != nil {
+	if idxErr := st.EnsureIndexes(); idxErr != nil {
 		_ = st.Close()
 		_ = producer.Close()
-		return nil, nil, fmt.Errorf("ensuring store indexes: %w", err)
+		return nil, nil, fmt.Errorf("ensuring store indexes: %w", idxErr)
 	}
 
+	// Align store batch-helper concurrency with config. Zero falls back to
+	// runtime.NumCPU which matches validator parallelism — keeps DB write
+	// fanout from being the bottleneck.
+	store.SetBatchConcurrency(cfg.Store.BatchConcurrency)
+
 	txTracker := store.NewTxTracker()
 
 	teranodeClient := teranode.NewClient(cfg.DatahubURLs, cfg.Teranode.AuthToken, teranode.HealthConfig{
-		FailureThreshold:    cfg.Propagation.EndpointHealth.FailureThreshold,
-		ProbeInterval:       time.Duration(cfg.Propagation.EndpointHealth.ProbeIntervalMs) * time.Millisecond,
-		ProbeTimeout:        time.Duration(cfg.Propagation.EndpointHealth.ProbeTimeoutMs) * time.Millisecond,
-		MinHealthyEndpoints: cfg.Propagation.EndpointHealth.MinHealthyEndpoints,
-		RefreshInterval:     time.Duration(cfg.Propagation.EndpointHealth.RefreshIntervalMs) * time.Millisecond,
-		Source:              endpointSource{st: st, network: cfg.Network},
-		Logger:              logger,
+		FailureThreshold:          cfg.Propagation.EndpointHealth.FailureThreshold,
+		BroadcastFailureThreshold: cfg.Propagation.EndpointHealth.BroadcastFailureThreshold,
+		ProbeInterval:             time.Duration(cfg.Propagation.EndpointHealth.ProbeIntervalMs) * time.Millisecond,
+		ProbeTimeout:              time.Duration(cfg.Propagation.EndpointHealth.ProbeTimeoutMs) * time.Millisecond,
+		MinHealthyEndpoints:       cfg.Propagation.EndpointHealth.MinHealthyEndpoints,
+		RefreshInterval:           time.Duration(cfg.Propagation.EndpointHealth.RefreshIntervalMs) * time.Millisecond,
+		Source:                    endpointSource{st: st, network: cfg.Network},
+		Logger:                    logger,
 	})
 
 	// Seed the registry with statically configured URLs so a freshly started
@@ -103,16 +121,16 @@ func Bootstrap(ctx context.Context, cfg *config.Config, logger *zap.Logger) (*De
 	if len(cfg.DatahubURLs) > 0 {
 		seedCtx, seedCancel := context.WithTimeout(ctx, 5*time.Second)
 		for _, url := range cfg.DatahubURLs {
-			if err := st.UpsertDatahubEndpoint(seedCtx, store.DatahubEndpoint{
+			if seedErr := st.UpsertDatahubEndpoint(seedCtx, store.DatahubEndpoint{
 				URL:      url,
 				Network:  cfg.Network,
 				Source:   store.DatahubEndpointSourceConfigured,
 				LastSeen: time.Now(),
-			}); err != nil {
+			}); seedErr != nil {
 				logger.Warn(
 					"failed to seed configured datahub url",
 					zap.String("url", url),
-					zap.Error(err),
+					zap.Error(seedErr),
 				)
 			}
 		}
@@ -131,6 +149,22 @@ 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).
+	var chainTracks chaintrackslib.Chaintracks
+	if cfg.ChaintracksServer.Enabled {
+		ct, ctErr := initChaintracks(ctx, cfg, logger)
+		if ctErr != nil {
+			_ = st.Close()
+			_ = producer.Close()
+			return nil, nil, fmt.Errorf("chaintracks init: %w", ctErr)
+		}
+		chainTracks = ct
+	}
+
 	deps := &Deps{
 		Cfg:            cfg,
 		Logger:         logger,
@@ -143,6 +177,7 @@ func Bootstrap(ctx context.Context, cfg *config.Config, logger *zap.Logger) (*De
 		TeranodeClient: teranodeClient,
 		MerkleClient:   merkleClient,
 		Validator:      txVal,
+		Chaintracks:    chainTracks,
 	}
 
 	cleanup := func() {
@@ -154,6 +189,52 @@ func Bootstrap(ctx context.Context, cfg *config.Config, logger *zap.Logger) (*De
 	return deps, cleanup, nil
 }
 
+// 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.
+//
+// The construction logic mirrors what previously lived in
+// chaintracks_server.Service.initChaintracks. Moving it here lets bump-
+// builder use the same instance without depending on a service's
+// initialization timing or a duplicate P2P subscription.
+func initChaintracks(ctx context.Context, cfg *config.Config, logger *zap.Logger) (chaintrackslib.Chaintracks, error) {
+	// Default chaintracks storage to <storage_path>/chaintracks/ so
+	// operators only need to set a single storage root. Tilde expansion
+	// happens in config.Load.
+	if cfg.Chaintracks.StoragePath == "" {
+		root := cfg.StoragePath
+		if root == "" {
+			root = "."
+		}
+		if err := os.MkdirAll(root, 0o750); err != nil {
+			return nil, fmt.Errorf("creating storage directory %s: %w", root, err)
+		}
+		cfg.Chaintracks.StoragePath = path.Join(root, "chaintracks")
+	}
+
+	// Thread the top-level network into chaintracks' embedded p2p config.
+	// Without this go-chaintracks falls back to "main" silently. Chaintracks
+	// needs the upstream-strict spelling ("main"/"test"/"teratestnet").
+	_, defaultBootstrap := config.ResolveP2PNetwork(cfg.Network)
+	cfg.Chaintracks.P2P.Network = config.ResolveChaintracksNetwork(cfg.Network)
+	if len(cfg.Chaintracks.P2P.MsgBus.BootstrapPeers) == 0 {
+		cfg.Chaintracks.P2P.MsgBus.BootstrapPeers = defaultBootstrap
+	}
+
+	ct, err := cfg.Chaintracks.Initialize(ctx, "arcade", nil)
+	if err != nil {
+		return nil, fmt.Errorf("chaintracks initialize: %w", err)
+	}
+
+	network, _ := ct.GetNetwork(ctx)
+	logger.Info(
+		"chaintracks initialized",
+		zap.String("storage_path", cfg.Chaintracks.StoragePath),
+		zap.String("network", network),
+	)
+	return ct, nil
+}
+
 // BuildServices returns the services that should run for the configured mode.
 // Each service's lifetime is tied to the ctx passed to its Start method by
 // the caller — the supervisor in cmd/arcade or the test harness.
@@ -166,10 +247,37 @@ func BuildServices(d *Deps) []services.Service {
 	}
 
 	if shouldRun("api-server") {
-		svcs = append(svcs, api_server.New(cfg, d.Logger, d.Producer, d.Publisher, d.Store, d.TxTracker, d.TeranodeClient))
+		svcs = append(svcs, api_server.New(cfg, d.Logger, d.Producer, d.Publisher, d.Store, d.TxTracker, d.TeranodeClient, d.MerkleClient))
 	}
 	if shouldRun("bump-builder") {
-		svcs = append(svcs, bump_builder.New(cfg, d.Logger, d.Producer, d.Publisher, d.Store, d.TeranodeClient))
+		// chainHeader is nil when chaintracks is disabled — bump-builder
+		// nil-guards and falls back to subtree-count-only validation.
+		var chainHeader bump_builder.ChainHeaderReader
+		if d.Chaintracks != nil {
+			chainHeader = chaintracksHeaderReader{ct: d.Chaintracks}
+		}
+		svcs = append(svcs, bump_builder.New(cfg, d.Logger, d.Producer, d.Publisher, d.Store, d.TeranodeClient, d.TxTracker, chainHeader))
+	}
+	if shouldRun("watchdog") && cfg.Watchdog.Enabled {
+		if wd := watchdog.NewService(cfg, d.Logger, d.Store, d.Leaser, d.MerkleClient); wd != nil {
+			svcs = append(svcs, wd)
+		} else {
+			d.Logger.Info("watchdog skipped: merkle_service.url or leaser not configured")
+		}
+	}
+	if shouldRun("sse") {
+		if ssvc := sse.New(cfg, d.Logger, d.Publisher, d.Store); ssvc != nil {
+			svcs = append(svcs, ssvc)
+		} else {
+			d.Logger.Info("sse skipped: sse.enabled=false or publisher not configured")
+		}
+	}
+	if shouldRun("chaintracks") {
+		if ct := chaintracks_server.New(cfg, d.Logger, d.Store, d.Chaintracks); ct != nil {
+			svcs = append(svcs, ct)
+		} else {
+			d.Logger.Info("chaintracks skipped: chaintracks_server.enabled=false (regtest force-disables this)")
+		}
 	}
 	if shouldRun("tx-validator") {
 		svcs = append(svcs, tx_validator.New(cfg, d.Logger, d.Producer, d.Publisher, d.Store, d.TxTracker, d.Validator))
@@ -187,6 +295,19 @@ func BuildServices(d *Deps) []services.Service {
 	return svcs
 }
 
+// chaintracksHeaderReader adapts go-chaintracks's Chaintracks interface to
+// the narrower ChainHeaderReader contract bump-builder needs. The library
+// already exposes GetHeaderByHash with the same signature, so this is a
+// trivial passthrough; the wrapper exists only to keep arcade packages from
+// importing chaintracks types into their public APIs.
+type chaintracksHeaderReader struct {
+	ct chaintrackslib.Chaintracks
+}
+
+func (a chaintracksHeaderReader) GetHeaderByHash(ctx context.Context, hash *chainhash.Hash) (*chaintrackslib.BlockHeader, error) {
+	return a.ct.GetHeaderByHash(ctx, hash)
+}
+
 // endpointSource adapts store.Store to teranode.EndpointSource by extracting
 // just the URL list. network scopes the listing to the configured Bitcoin
 // network so a store shared across pods (or reused after a network change)

bump/datahub.go

@@ -32,6 +32,19 @@ const DefaultMaxBlockBytes int64 = 1 * 1024 * 1024 * 1024 // 1 GiB
 // stops a hostile server from inflating arcade's log lines.
 const maxErrorBodyBytes int64 = 512
 
+// BlockDataValidator is an optional response-acceptance predicate run after a
+// successful datahub fetch. Returning a non-nil error causes the fetch loop
+// to discard that peer's response (logging the reason into urlErrors) and try
+// the next URL — turning the otherwise-greedy "first 200 wins" into a
+// "first 200 that passes validation wins" policy.
+//
+// Validators are stateless and pure functions of the response; they receive
+// what was parsed from one peer and decide whether it's plausible. Callers
+// use this to reject obviously-truncated responses (a pruned peer returning
+// fewer subtrees than the STUMPs reference) and to cross-check the header
+// merkle root against an out-of-band canonical value (chaintracks).
+type BlockDataValidator func(subtreeHashes []chainhash.Hash, headerMerkleRoot *chainhash.Hash) error
+
 // FetchBlockDataForBUMP fetches subtree hashes, coinbase BUMP, and the block's
 // header merkle root from the binary block endpoint, trying all DataHub URLs.
 // Each attempt emits a Debug-level log line so operators can see which URLs
@@ -48,13 +61,27 @@ const maxErrorBodyBytes int64 = 512
 // subtreeCount (varint) | subtreeHashes (N×32) | coinbaseTx (variable) |
 // blockHeight (varint) | coinbaseBUMPLen (varint) | coinbaseBUMP (variable)
 func FetchBlockDataForBUMP(ctx context.Context, datahubURLs []string, blockHash string, logger *zap.Logger) (subtreeHashes []chainhash.Hash, coinbaseBUMP []byte, headerMerkleRoot *chainhash.Hash, err error) {
-	return FetchBlockDataForBUMPWithCap(ctx, datahubURLs, blockHash, DefaultMaxBlockBytes, logger)
+	return FetchBlockDataForBUMPWithOptions(ctx, datahubURLs, blockHash, DefaultMaxBlockBytes, nil, logger)
 }
 
 // FetchBlockDataForBUMPWithCap is the cap-aware variant of FetchBlockDataForBUMP.
 // maxBlockBytes <= 0 selects DefaultMaxBlockBytes — passing zero/negative does
 // not silently disable the protection.
+//
+// Retained for backwards compatibility with callers that don't need to plug
+// in a content validator. Equivalent to FetchBlockDataForBUMPWithOptions with
+// validator=nil.
 func FetchBlockDataForBUMPWithCap(ctx context.Context, datahubURLs []string, blockHash string, maxBlockBytes int64, logger *zap.Logger) (subtreeHashes []chainhash.Hash, coinbaseBUMP []byte, headerMerkleRoot *chainhash.Hash, err error) {
+	return FetchBlockDataForBUMPWithOptions(ctx, datahubURLs, blockHash, maxBlockBytes, nil, logger)
+}
+
+// FetchBlockDataForBUMPWithOptions adds an optional BlockDataValidator that
+// runs after each successful HTTP+parse. A non-nil error from the validator
+// is treated the same as a transport error: log it, append to urlErrors, and
+// try the next URL. This turns the loop into "first 200-that-passes-validation
+// wins" — defending against datahubs that return self-consistent-but-wrong
+// block representations (pruned peers, stale caches).
+func FetchBlockDataForBUMPWithOptions(ctx context.Context, datahubURLs []string, blockHash string, maxBlockBytes int64, validator BlockDataValidator, logger *zap.Logger) (subtreeHashes []chainhash.Hash, coinbaseBUMP []byte, headerMerkleRoot *chainhash.Hash, err error) {
 	if logger == nil {
 		logger = zap.NewNop()
 	}
@@ -68,22 +95,58 @@ func FetchBlockDataForBUMPWithCap(ctx context.Context, datahubURLs []string, blo
 	for i, dataHubURL := range datahubURLs {
 		start := time.Now()
 		hashes, cbBUMP, root, status, fetchErr := fetchBlockBinary(ctx, dataHubURL, blockHash, maxBlockBytes)
-		logger.Debug("datahub fetch attempt",
+		logger.Debug(
+			"datahub fetch attempt",
 			zap.Int("idx", i),
 			zap.String("url", dataHubURL),
 			zap.Int("status", status),
 			zap.Duration("elapsed", time.Since(start)),
 			zap.Int64("max_block_bytes", maxBlockBytes),
 			zap.Error(fetchErr),
 		)
-		if fetchErr == nil {
-			return hashes, cbBUMP, root, nil
+		if fetchErr != nil {
+			urlErrors = append(urlErrors, fmt.Sprintf("url[%d] %q: %v", i, dataHubURL, fetchErr))
+			continue
+		}
+		if validator != nil {
+			if vErr := validator(hashes, root); vErr != nil {
+				logger.Warn(
+					"datahub response rejected by validator",
+					zap.Int("idx", i),
+					zap.String("url", dataHubURL),
+					zap.Int("subtree_count", len(hashes)),
+					zap.Error(vErr),
+				)
+				urlErrors = append(urlErrors, fmt.Sprintf("url[%d] %q: validator rejected: %v", i, dataHubURL, vErr))
+				continue
+			}
 		}
-		urlErrors = append(urlErrors, fmt.Sprintf("url[%d] %q: %v", i, dataHubURL, fetchErr))
+		return hashes, cbBUMP, root, nil
 	}
 	return nil, nil, nil, fmt.Errorf("all DataHub URLs failed for block %s:\n  %s", blockHash, strings.Join(urlErrors, "\n  "))
 }
 
+// SubtreeCountValidator returns a BlockDataValidator that rejects responses
+// with fewer than minSubtrees subtree hashes. Pass max(stump.SubtreeIndex)+1
+// when STUMPs are already in hand — a peer whose response can't index every
+// STUMP we've collected is provably wrong, no expensive cryptographic check
+// required.
+//
+// minSubtrees <= 0 returns nil — no validation, which preserves the old
+// "first 200 wins" behavior for callers that haven't yet collected STUMPs.
+func SubtreeCountValidator(minSubtrees int) BlockDataValidator {
+	if minSubtrees <= 0 {
+		return nil
+	}
+	return func(subtreeHashes []chainhash.Hash, _ *chainhash.Hash) error {
+		if len(subtreeHashes) < minSubtrees {
+			return fmt.Errorf("subtree_count %d < required %d (STUMPs reference subtree indexes up to %d)",
+				len(subtreeHashes), minSubtrees, minSubtrees-1)
+		}
+		return nil
+	}
+}
+
 // fetchBlockBinary fetches a block from the binary endpoint and parses
 // subtree hashes, coinbase BUMP, and the header merkle root from the response.
 // Returns the HTTP status code (0 on transport error before a response was