Repository navigation
feat(merkle): register with multiple merkle-services and fan out reprocess #362
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -3,6 +3,7 @@ package config | |
| import ( | ||
| "errors" | ||
| "fmt" | ||
| "net/url" | ||
| "os" | ||
| "path/filepath" | ||
| "strings" | ||
|
|
@@ -310,9 +311,52 @@ type TeranodeConfig struct { | |
| AuthToken string `mapstructure:"auth_token"` | ||
| } | ||
|
|
||
| // MerkleServiceConfig selects the merkle-service endpoints arcade registers | ||
| // transactions with. URL is the original single-endpoint form; URLs is the | ||
| // multi-endpoint form. Both feed Endpoints(), so an existing config with only | ||
| // `url` keeps working and `urls` may be added alongside it. | ||
| // | ||
| // With several endpoints every tx is registered with all of them and every | ||
| // endpoint delivers callbacks; arcade treats a callback that carries nothing | ||
| // new as a no-op. A tx counts as registered once at least one endpoint | ||
| // accepted it, so a single unreachable merkle-service no longer blocks | ||
| // broadcast (F-024 still holds: zero acceptances means no broadcast). | ||
| type MerkleServiceConfig struct { | ||
| URL string `mapstructure:"url"` | ||
| URLs []string `mapstructure:"urls"` | ||
| AuthToken string `mapstructure:"auth_token"` | ||
| } | ||
|
|
||
| // Endpoints returns every configured merkle-service base URL, in config | ||
| // order (URL first, then URLs), trimmed, without a trailing slash, and with | ||
| // duplicates removed. Empty entries are dropped here; validate rejects them so | ||
| // a typo in `urls` fails loudly at startup instead of silently shrinking the | ||
| // redundancy set. | ||
| func (c MerkleServiceConfig) Endpoints() []string { | ||
| raw := make([]string, 0, 1+len(c.URLs)) | ||
| raw = append(raw, c.URL) | ||
| raw = append(raw, c.URLs...) | ||
| seen := make(map[string]struct{}, len(raw)) | ||
| out := make([]string, 0, len(raw)) | ||
| for _, u := range raw { | ||
| u = strings.TrimSuffix(strings.TrimSpace(u), "/") | ||
| if u == "" { | ||
| continue | ||
| } | ||
| if _, dup := seen[u]; dup { | ||
| continue | ||
| } | ||
| seen[u] = struct{}{} | ||
| out = append(out, u) | ||
| } | ||
| return out | ||
| } | ||
|
|
||
| // Enabled reports whether the Merkle integration is configured at all. It is | ||
| // the single toggle every consumer checks: no endpoints means no /watch | ||
| // registration, no /reprocess recovery, and no watchdog. | ||
| func (c MerkleServiceConfig) Enabled() bool { | ||
| return len(c.Endpoints()) > 0 | ||
| } | ||
|
|
||
| // P2PConfig controls the libp2p-based peer discovery service. Seeds is the | ||
|
|
@@ -548,10 +592,13 @@ type PropagationConfig struct { | |
| // MerkleRegisteredAt is within this window. /watch is INSERT ... ON | ||
| // CONFLICT DO NOTHING on the merkle-service side and does not refresh | ||
| // expires_at, so re-registering a tx merkle-service already knows about | ||
| // is wasted work. Default -1 (auto): 30 (matches merkle-service | ||
| // postMineTTLSec) with a single merkle-service, 0 with several, because | ||
| // the stamp is set once any one of them accepted the tx and the pool's | ||
| // queue of registrations owed to a lagging endpoint is in memory, so a | ||
| // restart must re-send them. Set to 0 to disable the skip and | ||
| // re-register every non-terminal tx — useful for forcing a full re-sync | ||
| // after a known merkle-service wipe. Issue #145. | ||
| MerkleReplaySkipRecentMinutes int `mapstructure:"merkle_replay_skip_recent_minutes"` | ||
| // MerkleReplayRPS caps the average requests-per-second the startup | ||
| // replay issues against merkle-service. Implemented as an inter-batch | ||
|
|
@@ -751,8 +798,8 @@ type ReconcilerConfig struct { | |
|
|
||
| // WatchdogConfig tunes the stale-block recovery watchdog. Defaults are | ||
| // applied in setDefaults; zero values for everything except Enabled fall | ||
| // back to those defaults. The watchdog requires at least one merkle_service | ||
| // endpoint (no /reprocess target otherwise) and runs as a standalone arcade | ||
| // service (mode=watchdog) or alongside other services when mode=all. At | ||
| // most one replica fires per tick — coordination via the | ||
| // `block-processing-watchdog` lease. | ||
|
|
@@ -1190,6 +1237,17 @@ func setDefaults() { | |
| viper.SetDefault("api.host", "0.0.0.0") | ||
| viper.SetDefault("api.port", 8080) | ||
| viper.SetDefault("kafka.backend", "sarama") | ||
| // callback_* and merkle_service.* defaults exist so viper.AutomaticEnv | ||
| // knows the keys: without a registered key an ARCADE_MERKLE_SERVICE_URLS | ||
| // or ARCADE_CALLBACK_TOKEN override is silently ignored (see | ||
| // TestPropagationBatchCapsBind for the same trap). Empty values keep the | ||
| // integration disabled. | ||
| viper.SetDefault("callback_url", "") | ||
| viper.SetDefault("callback_token", "") | ||
| viper.SetDefault("merkle_service.url", "") | ||
| viper.SetDefault("merkle_service.urls", []string{}) | ||
| viper.SetDefault("merkle_service.auth_token", "") | ||
|
|
||
| viper.SetDefault("kafka.brokers", []string{"localhost:9092"}) | ||
| viper.SetDefault("kafka.consumer_group", "arcade") | ||
| viper.SetDefault("kafka.max_retries", 5) | ||
|
|
@@ -1276,7 +1334,7 @@ func setDefaults() { | |
| // callbacks for everything in-flight. | ||
| viper.SetDefault("propagation.register_replay_on_start", true) | ||
| viper.SetDefault("propagation.register_replay_lookback_hours", 24) | ||
| viper.SetDefault("propagation.merkle_replay_skip_recent_minutes", -1) | ||
| viper.SetDefault("propagation.merkle_replay_rps", 50) | ||
|
|
||
| viper.SetDefault("network", NetworkMainnet) | ||
|
|
@@ -1345,7 +1403,7 @@ func setDefaults() { | |
| // Block-processing watchdog (standalone arcade service — mode=watchdog | ||
| // in production, in-process under mode=all): on by default. The runtime | ||
| // nil-guards the merkle-service client; an unconfigured deployment | ||
| // (no merkle_service.url/urls) skips the watchdog regardless of this flag. | ||
| viper.SetDefault("watchdog.enabled", true) | ||
| viper.SetDefault("watchdog.interval_ms", 30000) | ||
| viper.SetDefault("watchdog.stale_threshold_ms", 120000) | ||
|
|
@@ -1401,6 +1459,52 @@ func setDefaults() { | |
| viper.SetDefault("validator.observed_fee_refresh_ms", DefaultValidatorObservedFeeRefreshMs) | ||
| } | ||
|
|
||
| // validateMerkleService checks the merkle_service block. Every entry of | ||
| // `url`/`urls` must be an absolute http(s) URL; a blank entry in `urls` is a | ||
| // misconfiguration rather than "disabled" (only an entirely empty block | ||
| // disables the integration). When at least one endpoint is set the top-level | ||
| // callback_token is mandatory — see the comment in validate. | ||
| func validateMerkleService(cfg *Config) error { | ||
| for i, u := range cfg.MerkleService.URLs { | ||
| if strings.TrimSpace(u) == "" { | ||
| return fmt.Errorf("merkle_service.urls[%d] is empty; remove the entry or set a URL", i) | ||
| } | ||
| } | ||
| for _, u := range cfg.MerkleService.Endpoints() { | ||
| parsed, err := url.Parse(u) | ||
| if err != nil { | ||
| return fmt.Errorf("merkle_service url %q: %w", u, err) | ||
| } | ||
| // The URL is shown in /health, metric labels and startup logs, so | ||
| // credentials in it would leak there; auth_token is the supported | ||
| // way to authenticate. | ||
| if parsed.User != nil { | ||
| return fmt.Errorf("merkle_service url %q must not contain user info; use merkle_service.auth_token for authentication", redactUserinfo(parsed)) | ||
| } | ||
| if (parsed.Scheme != "http" && parsed.Scheme != "https") || parsed.Host == "" { | ||
| return fmt.Errorf("merkle_service url %q must be an absolute http(s) URL", u) | ||
| } | ||
|
Copilot marked this conversation as resolved.
Comment on lines
+1489
to
+1491
Contributor
Author
There was a problem hiding this comment. Fixed in 71dd3c3: URLs with user info are rejected at config validation (the error redacts the password); |
||
| // The client appends /watch, /reprocess and /health to the base URL | ||
| // as a string, so a query or fragment would swallow the path. | ||
| if parsed.RawQuery != "" || parsed.ForceQuery || parsed.Fragment != "" { | ||
| return fmt.Errorf("merkle_service url %q must not contain a query or fragment", u) | ||
| } | ||
| } | ||
| if cfg.MerkleService.Enabled() && cfg.CallbackToken == "" { | ||
| return fmt.Errorf("callback_token is required when merkle_service.url/urls is set " + | ||
| "(unauthenticated /api/v1/merkle-service/callback would accept forged callbacks; see issue #76)") | ||
| } | ||
| return nil | ||
| } | ||
|
|
||
| // redactUserinfo renders u with its user info replaced, so the validation | ||
| // error that rejects it does not print the password it is protecting. | ||
| func redactUserinfo(u *url.URL) string { | ||
| c := *u | ||
| c.User = url.User("REDACTED") | ||
| return c.String() | ||
| } | ||
|
|
||
| func validate(cfg *Config) error { | ||
| switch cfg.Kafka.Backend { | ||
| case "", "sarama": | ||
|
|
@@ -1456,15 +1560,16 @@ func validate(cfg *Config) error { | |
| default: | ||
| return fmt.Errorf("unknown store.backend %q (expected aerospike, pebble, postgres, or mongodb)", cfg.Store.Backend) | ||
| } | ||
| // merkle_service.url / urls are intentionally optional: no endpoints means | ||
| // the Merkle integration is disabled. The runtime treats | ||
| // MerkleServiceConfig.Enabled() as the toggle — app.Bootstrap only | ||
| // constructs a merkleservice.Pool when at least one endpoint is set, and | ||
| // propagation.Propagator nil-guards every dereference of the client. The | ||
| // documented standalone and zero-dependency profiles | ||
| // (config.example.standalone.yaml) ship with merkle_service.url: "" for | ||
| // exactly this reason. See issue #59 / finding F-001. | ||
| // | ||
| // When the Merkle integration IS enabled, callback_token is | ||
| // mandatory. The /api/v1/merkle-service/callback endpoint accepts forged | ||
| // status updates for any txid in the system if it runs without bearer-token | ||
| // auth, so we fail-closed here at config load rather than silently exposing | ||
|
|
@@ -1476,9 +1581,8 @@ func validate(cfg *Config) error { | |
| // callbacks. Without a configured token there's nothing to forward AND the | ||
| // inbound receiver would 401 anyway — the same fail-closed posture covers | ||
| // both ends, so a duplicate "outbound token required" check is unnecessary. | ||
| if err := validateMerkleService(cfg); err != nil { | ||
| return err | ||
| } | ||
| if cfg.Network == "" { | ||
| cfg.Network = NetworkMainnet | ||
|
|
||
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
0→Client's 30sdefaultTimeout. That number sets the cost of every hung-endpoint path: before 0366c51 it was the stall per batch, and it is still the stall for a single-endpoint deployment and for the all-down case until breakers trip (~31–38s here). Worth exposing asmerkle_service.timeout_ms(same default) so operators can tighten it; /watch answers in ~100–300ms on a healthy service. Follow-up, not blocking.