diff --git a/CLAUDE.md b/CLAUDE.md index 4e8e699..883f5ea 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -539,9 +539,20 @@ Key config sections: ### Settings Service & State Persistence (`--state-db`) -The **Settings Service** (`pkg/settings`) is the central authority for all *mutable* -runtime config. It owns the single effective `config.Config` that every module reads -and is the only writer. Setting values resolve from three layers: +The **Settings Service** (`config.Service` in `pkg/config/settings.go`) is the +central authority for all *mutable* runtime config and the dependency modules +hold instead of a raw `*config.Config`. It publishes **immutable config +snapshots**: `Current()` returns the latest generation, which never changes +underneath its reader; every applied change builds a fresh `Config` (shallow +copy — all fields are values) and swaps it in atomically. Consumers load +exactly ONE snapshot per operation (HTTP request, scheduler tick, build, +reconcile pass) and thread it down the call stack. The greppable rule: +`*config.Service` may live in struct fields; `*config.Config` (and section +pointers) appear only as function parameters and locals — storing a snapshot +freezes that consumer on a stale generation. One-shot commands and tests wrap +a fixed config via `config.NewStaticService(cfg)` (serves the same pointer +every call, so single-threaded tests may still mutate it between operations). +Setting values resolve from three layers: ``` hardcoded defaults < CLI-supplied (flag/env/config) < UI override @@ -555,12 +566,14 @@ CLI vs UI is resolved by **recency** (a monotonic seq), not fixed priority: form the CLI layer, so bumping a *hardcoded default* in a new release never clobbers a UI override. -The mutable-setting registry lives in `pkg/settings/fields.go` (keys in `keys.go`); -the per-module enable flags (`epbs_enabled`, `builder_api_enabled`, -`lifecycle_enabled`) are ordinary settings too. Write handlers call -`settingsSvc.SetMany`, which mutates the shared config in place, persists overrides, -and fires `OnChange` callbacks (registered in `cmd/run.go`) that trigger module -resets (`builder.UpdateConfig`, `epbs.UpdateConfig`) and `SetEnabled` syncs. +The mutable-setting registry lives in `pkg/config/settings_fields.go` (keys in +`settings_keys.go`); the per-module enable flags (`epbs_enabled`, +`builder_api_enabled`, `lifecycle_enabled`) are ordinary settings too. Write +handlers call `settingsSvc.SetMany`, which publishes the new snapshot +generation (a `SetMany` batch is atomic: readers see all of it or none of it), +persists overrides, and fires `OnChange` callbacks (registered in +`cmd/run.go`) that trigger module resets (plan-service schedule accounting) +and `SetEnabled` syncs from a fresh `Current()` snapshot. The optional **state-db** (`pkg/db`, mirrors spamoor: `glebarez/go-sqlite` + `sqlx` + goose migrations) persists across restarts when `--state-db ` is set: @@ -593,7 +606,7 @@ numbered step comments there match this list 1:1): 3. Initialize BLS signer 4. Initialize RPC client and wallet (if lifecycle available) 5. Fetch chain spec & genesis (wait for the beacon node), apply slot-time timing defaults -6. Open the state-db (`--state-db`) and initialize the central Settings Service (applies persisted overrides into `cfg` in place before any module reads it) +6. Open the state-db (`--state-db`) and initialize the central Settings Service (its first published snapshot already includes persisted overrides; the builder key registry is constructed right after it, since the fleet targets are mutable settings) 7. Start chain service 7b. Start the action plan service (the per-slot scheduling authority; persisted via the `kv_store` `slot_plans` namespace; a mandatory constructor dependency of every action module below) 8. Initialize lifecycle manager (if prerequisites available) diff --git a/cmd/deposit.go b/cmd/deposit.go index 1ed74fd..48d1885 100644 --- a/cmd/deposit.go +++ b/cmd/deposit.go @@ -8,6 +8,7 @@ import ( "github.com/spf13/cobra" "github.com/ethpandaops/buildoor/pkg/chain" + "github.com/ethpandaops/buildoor/pkg/config" "github.com/ethpandaops/buildoor/pkg/lifecycle" "github.com/ethpandaops/buildoor/pkg/rpc/beacon" "github.com/ethpandaops/buildoor/pkg/rpc/execution" @@ -52,8 +53,12 @@ var depositCmd = &cobra.Command{ } defer rpcClient.Close() + // One-shot command: wrap the resolved operator config in a static + // settings service (no UI overrides, no persistence). + cfgSvc := config.NewStaticService(cfg) + // Initialize the managed builder key set and select the key to deposit for - registry, err := newKeyRegistry(cfg, logger) + registry, err := newKeyRegistry(cfgSvc, logger) if err != nil { return err } @@ -91,7 +96,7 @@ var depositCmd = &cobra.Command{ } // Initialize chain service - chainSvc := chain.NewService(cfg, clClient, chainSpec, genesis, logger) + chainSvc := chain.NewService(cfgSvc, clClient, chainSpec, genesis, logger) if err := chainSvc.Start(ctx); err != nil { return fmt.Errorf("failed to start chain service: %w", err) } @@ -116,7 +121,7 @@ var depositCmd = &cobra.Command{ timeout, _ := cmd.Flags().GetDuration("timeout") // Initialize lifecycle manager - lifecycleMgr, err := lifecycle.NewManager(cfg, clClient, chainSvc, registry, w, logger) + lifecycleMgr, err := lifecycle.NewManager(cfgSvc, clClient, chainSvc, registry, w, logger) if err != nil { return fmt.Errorf("failed to initialize lifecycle manager: %w", err) } diff --git a/cmd/exit.go b/cmd/exit.go index 85add33..e25888c 100644 --- a/cmd/exit.go +++ b/cmd/exit.go @@ -7,6 +7,7 @@ import ( "github.com/spf13/cobra" "github.com/ethpandaops/buildoor/pkg/chain" + "github.com/ethpandaops/buildoor/pkg/config" "github.com/ethpandaops/buildoor/pkg/lifecycle" "github.com/ethpandaops/buildoor/pkg/rpc/beacon" "github.com/ethpandaops/buildoor/pkg/rpc/execution" @@ -52,7 +53,11 @@ var exitCmd = &cobra.Command{ defer rpcClient.Close() // Initialize the managed builder key set and select the key to exit - registry, err := newKeyRegistry(cfg, logger) + // One-shot command: wrap the resolved operator config in a static + // settings service (no UI overrides, no persistence). + cfgSvc := config.NewStaticService(cfg) + + registry, err := newKeyRegistry(cfgSvc, logger) if err != nil { return err } @@ -90,7 +95,7 @@ var exitCmd = &cobra.Command{ } // Initialize chain service - chainSvc := chain.NewService(cfg, clClient, chainSpec, genesis, logger) + chainSvc := chain.NewService(cfgSvc, clClient, chainSpec, genesis, logger) if err := chainSvc.Start(ctx); err != nil { return fmt.Errorf("failed to start chain service: %w", err) } diff --git a/cmd/keys.go b/cmd/keys.go index 95702ef..4f3a1e1 100644 --- a/cmd/keys.go +++ b/cmd/keys.go @@ -13,13 +13,15 @@ import ( // newKeyRegistry builds the managed builder key set from the configured entry // key (raw private key or mnemonic + account index). Internal key 0 is the entry // key itself, so a single-key deployment keeps its identity. -func newKeyRegistry(cfg *config.Config, log logrus.FieldLogger) (*builder_keys.Registry, error) { +func newKeyRegistry(cfgSvc *config.Service, log logrus.FieldLogger) (*builder_keys.Registry, error) { + cfg := cfgSvc.Current() + entryPrivkey, err := signer.ResolveEntryPrivkey(cfg.BuilderPrivkey, cfg.BuilderMnemonic, cfg.BuilderKeyIndex) if err != nil { return nil, fmt.Errorf("invalid builder key: %w", err) } - registry, err := builder_keys.NewRegistry(cfg, entryPrivkey, log) + registry, err := builder_keys.NewRegistry(cfgSvc, entryPrivkey, log) if err != nil { return nil, fmt.Errorf("failed to initialize builder key registry: %w", err) } diff --git a/cmd/run.go b/cmd/run.go index b08a6a0..26532e6 100644 --- a/cmd/run.go +++ b/cmd/run.go @@ -83,16 +83,6 @@ and begins building blocks according to configuration.`, return fmt.Errorf("failed to connect to EL engine API: %w", err) } - // 3. Initialize the managed builder key set (derived from the raw hex key - // or the mnemonic; internal key 0 is that entry key itself) - keyRegistry, err := newKeyRegistry(cfg, logger) - if err != nil { - return err - } - - pubkey := keyRegistry.Primary().Pubkey() - logger.WithField("pubkey", fmt.Sprintf("%x", pubkey[:8])).Info("Builder key loaded") - // 4. Initialize RPC client and wallet (if lifecycle enabled) var rpcClient *execution.Client @@ -178,6 +168,10 @@ and begins building blocks according to configuration.`, slotTimeMs := chainSpec.SecondsPerSlot.Milliseconds() cfg.ApplySlotDefaults(slotTimeMs) + if err := config.ValidateTimingBounds(cfg, chainSpec.SecondsPerSlot); err != nil { + return fmt.Errorf("invalid timing configuration: %w", err) + } + logger.WithFields(logrus.Fields{ "slot_time_ms": slotTimeMs, "build_start_time": cfg.EPBS.BuildStartTime, @@ -187,9 +181,12 @@ and begins building blocks according to configuration.`, }).Info("Timing defaults applied") // 6. Open the optional state-db and build the central settings service. - // The settings service applies persisted UI overrides (and detects CLI - // changes) into cfg in place BEFORE any service reads it, so every - // module starts from the effective configuration. + // The settings service publishes immutable config snapshots: persisted + // UI overrides (and detected CLI changes) are part of the first + // generation, so every module starts from the effective configuration. + // Modules receive the service itself and load one snapshot per + // operation; cfg below is rebound to the initial snapshot for the + // remaining startup reads. stateDB := db.NewDatabase(&db.Config{File: cfg.StateDBPath}, logger) if err := stateDB.Init(); err != nil { return fmt.Errorf("failed to init state-db: %w", err) @@ -205,13 +202,26 @@ and begins building blocks according to configuration.`, supplied[f.Key] = v.IsSet(f.FlagKey) } - settingsSvc, err := config.NewService(cfg, defaults, supplied, stateDB, logger) + settingsSvc, err := config.NewService(cfg, defaults, supplied, chainSpec.SecondsPerSlot, stateDB, logger) if err != nil { return fmt.Errorf("failed to init settings service: %w", err) } + cfg = settingsSvc.Current() + + // 6b. Initialize the managed builder key set (derived from the raw hex + // key or the mnemonic; internal key 0 is that entry key itself). Needs + // the settings service: the fleet targets are mutable settings. + keyRegistry, err := newKeyRegistry(settingsSvc, logger) + if err != nil { + return err + } + + pubkey := keyRegistry.Primary().Pubkey() + logger.WithField("pubkey", fmt.Sprintf("%x", pubkey[:8])).Info("Builder key loaded") + // 7. Start chain service (epoch-level state management) - chainSvc := chain.NewService(cfg, clClient, chainSpec, genesis, logger) + chainSvc := chain.NewService(settingsSvc, clClient, chainSpec, genesis, logger) if err := chainSvc.Start(ctx); err != nil { return fmt.Errorf("failed to start chain service: %w", err) } @@ -228,7 +238,7 @@ and begins building blocks according to configuration.`, // 7b. Initialize the per-slot action plan service. Decision points // (build/bid/serve/reveal) freeze the slot's plan on first use; plans // persist in the state-db's kv_store when --state-db is set. - planSvc := action_plan.NewPlanService(cfg, chainSvc, logger) + planSvc := action_plan.NewPlanService(settingsSvc, chainSvc, logger) planSvc.SetPersistence(ctx, stateDB) if err := planSvc.Start(ctx); err != nil { @@ -242,7 +252,7 @@ and begins building blocks according to configuration.`, var lifecycleMgr *lifecycle.Manager if lifecycleAvailable { - lifecycleMgr, err = lifecycle.NewManager(cfg, clClient, chainSvc, keyRegistry, w, logger) + lifecycleMgr, err = lifecycle.NewManager(settingsSvc, clClient, chainSvc, keyRegistry, w, logger) if err != nil { return fmt.Errorf("failed to initialize lifecycle: %w", err) } @@ -278,7 +288,7 @@ and begins building blocks according to configuration.`, defer validatorStore.Stop() } - builderSvc, err := payload_builder.NewService(cfg, clClient, chainSvc, planSvc, engineClient, feeRecipient, logger) + builderSvc, err := payload_builder.NewService(settingsSvc, clClient, chainSvc, planSvc, engineClient, feeRecipient, logger) if err != nil { return fmt.Errorf("failed to initialize builder: %w", err) } @@ -311,7 +321,7 @@ and begins building blocks according to configuration.`, // for payments settled since the last epoch snapshot. keyRegistry.SetBalanceAdjuster(paymentTracker) - revealSvc = payload_bidder.NewRevealService(cfg, keyRegistry, + revealSvc = payload_bidder.NewRevealService(settingsSvc, keyRegistry, clClient, chainSvc, builderSvc, paymentTracker, planSvc, chainSvc.GetHeadVoteTracker(), logger) if err := revealSvc.Start(ctx); err != nil { @@ -353,7 +363,7 @@ and begins building blocks according to configuration.`, gloasForkEpoch := chainSpec.GetForkEpoch(version.DataVersionGloas) logger.WithField("gloas_fork_epoch", gloasForkEpoch).Info("Initializing p2p bidder service...") - epbsSvc, err = p2p_bidder.NewService(clClient, chainSvc, keyRegistry, propPrefSvc.GetStore(), planSvc, logger) + epbsSvc, err = p2p_bidder.NewService(settingsSvc, clClient, chainSvc, keyRegistry, propPrefSvc.GetStore(), planSvc, logger) if err != nil { return fmt.Errorf("failed to initialize p2p bidder: %w", err) } @@ -381,7 +391,7 @@ and begins building blocks according to configuration.`, "genesis_validators_root": fmt.Sprintf("0x%x", genesisValidatorsRoot[:]), }).Info("Using genesis parameters from beacon node") - builderAPISrv = builderapi.NewServer(&cfg.BuilderAPI, logger, chainSvc, planSvc, builderSvc.GetPayloadCache(), keyRegistry, validatorStore) + builderAPISrv = builderapi.NewServer(settingsSvc, logger, chainSvc, planSvc, builderSvc.GetPayloadCache(), keyRegistry, validatorStore) builderAPISrv.SetCLClient(clClient) builderAPISrv.SetEnabled(cfg.BuilderAPIEnabled) @@ -405,7 +415,7 @@ and begins building blocks according to configuration.`, // artifacts. Started before the producer services so its blocking // subscriptions never miss an event; SetPersistence migrates any // legacy won_blocks namespace into slot results. - resultTracker := slot_results.NewTracker(cfg, chainSvc, stateDB, planSvc, + resultTracker := slot_results.NewTracker(settingsSvc, chainSvc, stateDB, planSvc, builderSvc, epbsSvc, revealSvc, inclusionTracker, keyRegistry, logger) resultTracker.SetPersistence(ctx, stateDB) @@ -425,28 +435,26 @@ and begins building blocks according to configuration.`, valRanges.Start(ctx) // 14. Register settings OnChange subscribers: route changes through the - // modules. The settings service has already mutated cfg in place; these - // callbacks trigger module-side resets (schedule counters, scheduler) and - // sync the enable flags. + // modules. The settings service has already published the new snapshot + // when these fire; they trigger module-side resets (schedule counters) + // and sync the enable flags from a fresh snapshot. settingsSvc.OnChange(func() { + current := settingsSvc.Current() + // The plan service is the scheduling authority: schedule-mode // changes reset its next_n accounting. planSvc.UpdateConfig() - if err := builderSvc.UpdateConfig(cfg); err != nil { - logger.WithError(err).Warn("failed to apply builder config update") - } - if epbsSvc != nil { - epbsSvc.SetEnabled(cfg.EPBSEnabled) + epbsSvc.SetEnabled(current.EPBSEnabled) } if builderAPISrv != nil { - builderAPISrv.SetEnabled(cfg.BuilderAPIEnabled) + builderAPISrv.SetEnabled(current.BuilderAPIEnabled) } if lifecycleMgr != nil { - lifecycleMgr.SetEnabled(cfg.LifecycleEnabled) + lifecycleMgr.SetEnabled(current.LifecycleEnabled) // A target key count change must act now, not at the next // reconcile tick. lifecycleMgr.Reconcile() diff --git a/pkg/action_plan/service.go b/pkg/action_plan/service.go index 52ca812..23d262c 100644 --- a/pkg/action_plan/service.go +++ b/pkg/action_plan/service.go @@ -47,7 +47,7 @@ type RuleChangeEvent struct { // that fill the slots it does not cover, their freeze state and their // persistence. It is the single writer; all reads return deep copies. type PlanService struct { - cfg *config.Config + cfgSvc *config.Service // settings source; one snapshot per decision chainSvc chain.Service store *memstore.Store[phase0.Slot, *SlotPlan] rules *memstore.Store[string, *SlotRule] @@ -75,9 +75,9 @@ type PlanService struct { // NewPlanService creates the plan service. The config pointer is the shared // live config; enable flags and default settings are read from it at freeze // time. -func NewPlanService(cfg *config.Config, chainSvc chain.Service, log logrus.FieldLogger) *PlanService { +func NewPlanService(cfgSvc *config.Service, chainSvc chain.Service, log logrus.FieldLogger) *PlanService { return &PlanService{ - cfg: cfg, + cfgSvc: cfgSvc, chainSvc: chainSvc, store: memstore.New[phase0.Slot, *SlotPlan](), rules: memstore.New[string, *SlotRule](), @@ -308,7 +308,7 @@ func (s *PlanService) Freeze(slot phase0.Slot) *FrozenPlan { plan := s.PlanForSlot(slot) fork := s.chainSvc.ActiveForkAtEpoch(s.chainSvc.GetEpochOfSlot(slot)) - frozen := resolveFrozenPlan(slot, plan, s.cfg, fork, time.Now(), s.slotsBuilt) + frozen := resolveFrozenPlan(slot, plan, s.cfgSvc.Current(), fork, time.Now(), s.slotsBuilt) s.frozen[slot] = frozen return frozen @@ -334,7 +334,7 @@ func (s *PlanService) UpdateConfig() { s.mu.Lock() defer s.mu.Unlock() - if s.cfg.Schedule.Mode == config.ScheduleModeNextN { + if s.cfgSvc.Current().Schedule.Mode == config.ScheduleModeNextN { s.slotsBuilt = 0 } } @@ -353,15 +353,17 @@ func (s *PlanService) GetSlotsRemaining() int { s.mu.Lock() defer s.mu.Unlock() - if s.cfg.Schedule.Mode != config.ScheduleModeNextN { + cfg := s.cfgSvc.Current() + + if cfg.Schedule.Mode != config.ScheduleModeNextN { return -1 } - if s.slotsBuilt >= s.cfg.Schedule.NextN { + if s.slotsBuilt >= cfg.Schedule.NextN { return 0 } - return int(s.cfg.Schedule.NextN - s.slotsBuilt) + return int(cfg.Schedule.NextN - s.slotsBuilt) } // IsFrozen reports whether the slot's plan has been frozen already. @@ -588,7 +590,7 @@ func matchRule(rules []*SlotRule, slot phase0.Slot, slotsPerEpoch uint64) *SlotR // pruneForEpoch drops past plans outside the retention window and stale // freeze markers. Future plans never match the cutoff and are never pruned. func (s *PlanService) pruneForEpoch(epoch phase0.Epoch) { - retention := s.cfg.SlotResultRetentionEpochs // live read; mutable setting + retention := s.cfgSvc.Current().SlotResultRetentionEpochs // mutable setting; fresh snapshot per prune if retention == 0 || uint64(epoch) <= retention { return } diff --git a/pkg/action_plan/service_test.go b/pkg/action_plan/service_test.go index 2fc9447..f637b59 100644 --- a/pkg/action_plan/service_test.go +++ b/pkg/action_plan/service_test.go @@ -59,7 +59,7 @@ func newTestService(chainSvc *stubChain, cfg *config.Config) *PlanService { log := logrus.New() log.SetLevel(logrus.ErrorLevel) - return NewPlanService(cfg, chainSvc, log) + return NewPlanService(config.NewStaticService(cfg), chainSvc, log) } func TestApplyUpdatesAndGet(t *testing.T) { @@ -701,7 +701,7 @@ func TestFrozenCandidateSettings(t *testing.T) { log := logrus.New() log.SetLevel(logrus.PanicLevel) - svc := NewPlanService(cfg, newStubChain(), log) + svc := NewPlanService(config.NewStaticService(cfg), newStubChain(), log) // Without a plan: the frozen snapshot carries the complete global // candidate policy and the global bid/serve selections. diff --git a/pkg/builder_keys/persistence_test.go b/pkg/builder_keys/persistence_test.go index 970c53c..07563b9 100644 --- a/pkg/builder_keys/persistence_test.go +++ b/pkg/builder_keys/persistence_test.go @@ -43,7 +43,7 @@ func newTestRegistry(t *testing.T, entryKey string) *Registry { MaxIndex: 50, }} - registry, err := NewRegistry(cfg, entryKey, log) + registry, err := NewRegistry(config.NewStaticService(cfg), entryKey, log) require.NoError(t, err) return registry diff --git a/pkg/builder_keys/registry.go b/pkg/builder_keys/registry.go index 0b7f9b2..f96bb30 100644 --- a/pkg/builder_keys/registry.go +++ b/pkg/builder_keys/registry.go @@ -69,7 +69,7 @@ type keyRuntime struct { // on-chain state, usage history and selection. It is the identity dependency of // every module that used to hold a single *signer.BLSSigner. type Registry struct { - cfg *config.Config + cfgSvc *config.Service // settings source; one snapshot per read entrySK string log logrus.FieldLogger adjuster BalanceAdjuster @@ -94,9 +94,9 @@ type Registry struct { // NewRegistry creates the key set rooted at the operator's entry key. It derives // key 0 eagerly, which validates the supplied key material; every other key is // derived on demand and cached for the process lifetime. -func NewRegistry(cfg *config.Config, entryPrivkeyHex string, log logrus.FieldLogger) (*Registry, error) { +func NewRegistry(cfgSvc *config.Service, entryPrivkeyHex string, log logrus.FieldLogger) (*Registry, error) { r := &Registry{ - cfg: cfg, + cfgSvc: cfgSvc, entrySK: entryPrivkeyHex, log: log.WithField("component", "builder-keys"), runtimes: make(map[uint64]*keyRuntime, 8), @@ -149,7 +149,7 @@ func (r *Registry) Start(ctx context.Context, chainSvc chain.Service, stateDB *d go r.run(runCtx) r.log.WithFields(logrus.Fields{ - "target": r.cfg.BuilderKeys.EffectiveTargetCount(), + "target": r.cfgSvc.Current().BuilderKeys.EffectiveTargetCount(), "known_keys": len(r.order), "primary_key": r.Primary().String(), }).Info("Builder key registry started") @@ -321,8 +321,8 @@ func (r *Registry) highestTracked() uint64 { // maxIndex returns the effective derivation cap. func (r *Registry) maxIndex() uint64 { - if r.cfg.BuilderKeys.MaxIndex > 0 { - return r.cfg.BuilderKeys.MaxIndex + if capIndex := r.cfgSvc.Current().BuilderKeys.MaxIndex; capIndex > 0 { + return capIndex } return defaultMaxIndex @@ -331,8 +331,8 @@ func (r *Registry) maxIndex() uint64 { // discoveryGap returns the effective number of consecutive unused indices that // ends the discovery scan. func (r *Registry) discoveryGap() uint64 { - if r.cfg.BuilderKeys.DiscoveryGap > 0 { - return r.cfg.BuilderKeys.DiscoveryGap + if gap := r.cfgSvc.Current().BuilderKeys.DiscoveryGap; gap > 0 { + return gap } return defaultDiscoveryGap @@ -349,7 +349,7 @@ func (r *Registry) discoveryGap() uint64 { func (r *Registry) Refresh() { snapshot := r.chainSnapshot() - target := r.cfg.BuilderKeys.EffectiveTargetCount() + target := r.cfgSvc.Current().BuilderKeys.EffectiveTargetCount() maxIndex := r.maxIndex() gapLimit := r.discoveryGap() @@ -698,7 +698,7 @@ func (r *Registry) States() []*State { // Aggregate summarises the key set for the dashboard. func (r *Registry) Aggregate() Aggregate { - aggregate := Aggregate{Target: r.cfg.BuilderKeys.EffectiveTargetCount()} + aggregate := Aggregate{Target: r.cfgSvc.Current().BuilderKeys.EffectiveTargetCount()} for _, state := range r.States() { switch state.Status { diff --git a/pkg/builder_keys/registry_test.go b/pkg/builder_keys/registry_test.go index 408c45e..5ce712b 100644 --- a/pkg/builder_keys/registry_test.go +++ b/pkg/builder_keys/registry_test.go @@ -21,7 +21,7 @@ func testRegistry(t *testing.T, keys config.BuilderKeysConfig) *Registry { log := logrus.New() log.SetLevel(logrus.PanicLevel) - registry, err := NewRegistry(&config.Config{BuilderKeys: keys}, testEntryKey, log) + registry, err := NewRegistry(config.NewStaticService(&config.Config{BuilderKeys: keys}), testEntryKey, log) require.NoError(t, err) return registry diff --git a/pkg/builderapi/builder_preferences_test.go b/pkg/builderapi/builder_preferences_test.go index fee9c55..a46e7f0 100644 --- a/pkg/builderapi/builder_preferences_test.go +++ b/pkg/builderapi/builder_preferences_test.go @@ -67,7 +67,7 @@ func TestSubmitBuilderPreferences_Success(t *testing.T) { require.NoError(t, err) cfg := &config.BuilderAPIConfig{BuilderURL: testBuilderURL} - srv := NewServer(cfg, logrus.New(), &mockChainService{}, newServingPlanService(), nil, nil, nil) + srv := NewServer(staticSvc(cfg), logrus.New(), &mockChainService{}, newServingPlanService(), nil, nil, nil) srv.SetEnabled(true) body := signBuilderPrefsRequest(t, blsSigner, testBuilderURL, 100, 5_000_000_000, gfv) @@ -91,7 +91,7 @@ func TestSubmitBuilderPreferences_SuccessSSZ(t *testing.T) { require.NoError(t, err) cfg := &config.BuilderAPIConfig{BuilderURL: testBuilderURL} - srv := NewServer(cfg, logrus.New(), &mockChainService{}, newServingPlanService(), nil, nil, nil) + srv := NewServer(staticSvc(cfg), logrus.New(), &mockChainService{}, newServingPlanService(), nil, nil, nil) srv.SetEnabled(true) // Build the same signed request as the JSON path, but submit it SSZ-encoded. @@ -116,7 +116,7 @@ func TestSubmitBuilderPreferences_SuccessSSZ(t *testing.T) { func TestSubmitBuilderPreferences_MalformedSSZ(t *testing.T) { cfg := &config.BuilderAPIConfig{BuilderURL: testBuilderURL} - srv := NewServer(cfg, logrus.New(), &mockChainService{}, newServingPlanService(), nil, nil, nil) + srv := NewServer(staticSvc(cfg), logrus.New(), &mockChainService{}, newServingPlanService(), nil, nil, nil) srv.SetEnabled(true) url := "/eth/v1/builder/builder_preferences/0x" + hex.EncodeToString(make([]byte, 48)) @@ -130,7 +130,7 @@ func TestSubmitBuilderPreferences_MalformedSSZ(t *testing.T) { func TestSubmitBuilderPreferences_UnknownContentType(t *testing.T) { cfg := &config.BuilderAPIConfig{BuilderURL: testBuilderURL} - srv := NewServer(cfg, logrus.New(), &mockChainService{}, newServingPlanService(), nil, nil, nil) + srv := NewServer(staticSvc(cfg), logrus.New(), &mockChainService{}, newServingPlanService(), nil, nil, nil) srv.SetEnabled(true) url := "/eth/v1/builder/builder_preferences/0x" + hex.EncodeToString(make([]byte, 48)) @@ -148,7 +148,7 @@ func TestSubmitBuilderPreferences_LatestOverwrites(t *testing.T) { require.NoError(t, err) cfg := &config.BuilderAPIConfig{BuilderURL: testBuilderURL} - srv := NewServer(cfg, logrus.New(), &mockChainService{}, newServingPlanService(), nil, nil, nil) + srv := NewServer(staticSvc(cfg), logrus.New(), &mockChainService{}, newServingPlanService(), nil, nil, nil) srv.SetEnabled(true) pk := blsSigner.PublicKey() url := "/eth/v1/builder/builder_preferences/0x" + hex.EncodeToString(pk[:]) @@ -172,7 +172,7 @@ func TestSubmitBuilderPreferences_WrongBuilderURL(t *testing.T) { require.NoError(t, err) cfg := &config.BuilderAPIConfig{BuilderURL: testBuilderURL} - srv := NewServer(cfg, logrus.New(), &mockChainService{}, newServingPlanService(), nil, nil, nil) + srv := NewServer(staticSvc(cfg), logrus.New(), &mockChainService{}, newServingPlanService(), nil, nil, nil) srv.SetEnabled(true) // Validly signed, but for a different builder URL than this builder's. @@ -198,7 +198,7 @@ func TestSubmitBuilderPreferences_BadSignature(t *testing.T) { require.NoError(t, err) cfg := &config.BuilderAPIConfig{BuilderURL: testBuilderURL} - srv := NewServer(cfg, logrus.New(), &mockChainService{}, newServingPlanService(), nil, nil, nil) + srv := NewServer(staticSvc(cfg), logrus.New(), &mockChainService{}, newServingPlanService(), nil, nil, nil) srv.SetEnabled(true) // Signed by `other` (correct builder URL), but submitted under `validator`'s pubkey. @@ -222,7 +222,7 @@ func TestSubmitBuilderPreferences_NoBuilderURLConfigured(t *testing.T) { require.NoError(t, err) cfg := &config.BuilderAPIConfig{} // BuilderURL empty - srv := NewServer(cfg, logrus.New(), &mockChainService{}, newServingPlanService(), nil, nil, nil) + srv := NewServer(staticSvc(cfg), logrus.New(), &mockChainService{}, newServingPlanService(), nil, nil, nil) srv.SetEnabled(true) body := signBuilderPrefsRequest(t, blsSigner, testBuilderURL, 100, 5_000_000_000, gfv) @@ -239,7 +239,7 @@ func TestSubmitBuilderPreferences_NoBuilderURLConfigured(t *testing.T) { func TestSubmitBuilderPreferences_InvalidJSON(t *testing.T) { cfg := &config.BuilderAPIConfig{BuilderURL: testBuilderURL} - srv := NewServer(cfg, logrus.New(), &mockChainService{}, newServingPlanService(), nil, nil, nil) + srv := NewServer(staticSvc(cfg), logrus.New(), &mockChainService{}, newServingPlanService(), nil, nil, nil) srv.SetEnabled(true) url := "/eth/v1/builder/builder_preferences/0x" + hex.EncodeToString(make([]byte, 48)) @@ -256,7 +256,7 @@ func TestSubmitBuilderPreferences_InvalidJSON(t *testing.T) { // preferences, so the handler rejects it with 400 (not 415). func TestSubmitBuilderPreferences_MissingContentType(t *testing.T) { cfg := &config.BuilderAPIConfig{BuilderURL: testBuilderURL} - srv := NewServer(cfg, logrus.New(), &mockChainService{}, newServingPlanService(), nil, nil, nil) + srv := NewServer(staticSvc(cfg), logrus.New(), &mockChainService{}, newServingPlanService(), nil, nil, nil) srv.SetEnabled(true) url := "/eth/v1/builder/builder_preferences/0x" + hex.EncodeToString(make([]byte, 48)) @@ -277,7 +277,7 @@ func TestSubmitBuilderPreferences_MissingContentType(t *testing.T) { func TestSubmitBuilderPreferences_Disabled(t *testing.T) { cfg := &config.BuilderAPIConfig{BuilderURL: testBuilderURL} - srv := NewServer(cfg, logrus.New(), &mockChainService{}, newServingPlanService(), nil, nil, nil) + srv := NewServer(staticSvc(cfg), logrus.New(), &mockChainService{}, newServingPlanService(), nil, nil, nil) // not enabled url := "/eth/v1/builder/builder_preferences/0x" + hex.EncodeToString(make([]byte, 48)) diff --git a/pkg/builderapi/epbs/handler.go b/pkg/builderapi/epbs/handler.go index 146445c..710429f 100644 --- a/pkg/builderapi/epbs/handler.go +++ b/pkg/builderapi/epbs/handler.go @@ -83,7 +83,7 @@ type BlockBroadcaster interface { // reveal to the shared payload_bidder.RevealService, which publishes the // envelope at the configured reveal time. type Handler struct { - cfg *config.BuilderAPIConfig // shared pointer, read live + cfgSvc *config.Service // settings source; one snapshot per request log logrus.FieldLogger chainSvc chain.Service payloadCache *payload_builder.PayloadCache @@ -118,15 +118,15 @@ type Handler struct { blocksAccepted atomic.Uint64 // count of accepted signed beacon blocks } -// NewHandler creates a new post-Gloas Builder API dialect handler. cfg is the -// shared mutable config pointer; values are read live, never copied out. +// NewHandler creates a new post-Gloas Builder API dialect handler. cfgSvc is +// the settings source; each request loads one immutable config snapshot. // planSvc is the mandatory per-slot scheduling/settings authority consulted // (via Freeze) on every getExecutionPayloadBid request. -func NewHandler(cfg *config.BuilderAPIConfig, log logrus.FieldLogger, chainSvc chain.Service, +func NewHandler(cfgSvc *config.Service, log logrus.FieldLogger, chainSvc chain.Service, planSvc *action_plan.PlanService, payloadCache *payload_builder.PayloadCache, registry *builder_keys.Registry) *Handler { return &Handler{ - cfg: cfg, + cfgSvc: cfgSvc, log: log.WithField("component", "builderapi-epbs"), chainSvc: chainSvc, planSvc: planSvc, diff --git a/pkg/builderapi/epbs/handler_test.go b/pkg/builderapi/epbs/handler_test.go index eebf617..1680fd8 100644 --- a/pkg/builderapi/epbs/handler_test.go +++ b/pkg/builderapi/epbs/handler_test.go @@ -206,18 +206,19 @@ func newBeaconBlockTestEnv(t *testing.T, slotDuration time.Duration, revealTimeM }, } - planSvc := action_plan.NewPlanService(cfg, chainSvc, log) + cfgSvc := config.NewStaticService(cfg) + planSvc := action_plan.NewPlanService(cfgSvc, chainSvc, log) - builderSvc, err := payload_builder.NewService(&config.Config{}, nil, chainSvc, planSvc, nil, common.Address{}, log) + builderSvc, err := payload_builder.NewService(cfgSvc, nil, chainSvc, planSvc, nil, common.Address{}, log) require.NoError(t, err) publisher := &stubEnvelopePublisher{} revealSvc := payload_bidder.NewRevealService( - cfg, registry, publisher, chainSvc, builderSvc, nil, planSvc, nil, log) + cfgSvc, registry, publisher, chainSvc, builderSvc, nil, planSvc, nil, log) broadcaster := &stubBlockBroadcaster{} - h := NewHandler(&cfg.BuilderAPI, log, chainSvc, planSvc, + h := NewHandler(cfgSvc, log, chainSvc, planSvc, payload_builder.NewPayloadCache(10), registry) h.SetBlockBroadcaster(broadcaster) h.SetRevealService(revealSvc) diff --git a/pkg/builderapi/epbs/keyset_test.go b/pkg/builderapi/epbs/keyset_test.go index 45fe775..96c7d09 100644 --- a/pkg/builderapi/epbs/keyset_test.go +++ b/pkg/builderapi/epbs/keyset_test.go @@ -27,7 +27,7 @@ func newTestKeyRegistry(t *testing.T, builderIndices ...uint64) *builder_keys.Re MaxIndex: 32, }} - registry, err := builder_keys.NewRegistry(cfg, testEntryPrivkey, log) + registry, err := builder_keys.NewRegistry(config.NewStaticService(cfg), testEntryPrivkey, log) require.NoError(t, err) for keyIndex, builderIndex := range builderIndices { diff --git a/pkg/builderapi/epbs/payload_bid.go b/pkg/builderapi/epbs/payload_bid.go index 0f19c52..ce40ef3 100644 --- a/pkg/builderapi/epbs/payload_bid.go +++ b/pkg/builderapi/epbs/payload_bid.go @@ -163,8 +163,10 @@ func (h *Handler) HandleGetExecutionPayloadBid(w http.ResponseWriter, r *http.Re } // Parse and validate SignedRequestAuth from the request body. - // Auth is always verified when present; h.cfg.RequireRequestAuth controls whether - // absence is an error. + // Auth is always verified when present; RequireRequestAuth controls whether + // absence is an error. One config snapshot covers the whole auth check. + bapiCfg := &h.cfgSvc.Current().BuilderAPI + authBody, readErr := io.ReadAll(r.Body) if readErr != nil { log.WithError(readErr).Warn("getExecutionPayloadBid: failed to read request body") @@ -195,10 +197,10 @@ func (h *Handler) HandleGetExecutionPayloadBid(w http.ResponseWriter, r *http.Re writeError(w, http.StatusBadRequest, "invalid SignedRequestAuthV1: auth.message.slot does not match the requested slot") return } - if h.cfg.BuilderURL != "" && string(signedAuth.Message.Data) != h.cfg.BuilderURL { + if bapiCfg.BuilderURL != "" && string(signedAuth.Message.Data) != bapiCfg.BuilderURL { log.WithFields(logrus.Fields{ "auth_url": string(signedAuth.Message.Data), - "builder_url": h.cfg.BuilderURL, + "builder_url": bapiCfg.BuilderURL, }).Warn("getExecutionPayloadBid: SignedRequestAuth data (builder_url) mismatch") writeError(w, http.StatusBadRequest, "invalid SignedRequestAuthV1: auth.message.data does not match this builder's URL") return @@ -210,7 +212,7 @@ func (h *Handler) HandleGetExecutionPayloadBid(w http.ResponseWriter, r *http.Re return } log.Info("getExecutionPayloadBid: SignedRequestAuth verified") - } else if h.cfg.RequireRequestAuth { + } else if bapiCfg.RequireRequestAuth { log.Warn("getExecutionPayloadBid: missing required SignedRequestAuth") writeError(w, http.StatusUnauthorized, "missing SignedRequestAuthV1: this builder requires authenticated requests") return @@ -456,8 +458,9 @@ func (h *Handler) matchPayloadForParent( ctx context.Context, slot phase0.Slot, parentRoot phase0.Root, parentHash phase0.Hash32, servePolicy string, ) (*payload_builder.Payload, *bidMatchError) { + bapiCfg := &h.cfgSvc.Current().BuilderAPI if servePolicy == "" { - servePolicy = h.cfg.ServeCandidates + servePolicy = bapiCfg.ServeCandidates } if payload := h.payloadCache.GetVariant(slot, beacon.AttrParentKey{Root: parentRoot, Hash: parentHash}); payload != nil { @@ -486,7 +489,7 @@ func (h *Handler) matchPayloadForParent( reason: "parent_hash is neither the parent block's committed payload nor its execution parent"} } - if h.cfg.OnDemandBuild && h.onDemandBuilder != nil { + if bapiCfg.OnDemandBuild && h.onDemandBuilder != nil { payload, err := h.onDemandBuilder.BuildCandidateOnDemand(ctx, slot, parentRoot, parentHash) if err != nil { return nil, &bidMatchError{status: http.StatusNoContent, diff --git a/pkg/builderapi/epbs/payload_bid_test.go b/pkg/builderapi/epbs/payload_bid_test.go index 17ceb87..0dc8089 100644 --- a/pkg/builderapi/epbs/payload_bid_test.go +++ b/pkg/builderapi/epbs/payload_bid_test.go @@ -153,9 +153,10 @@ func newPayloadBidTestEnv(t *testing.T, enabled bool) *payloadBidTestEnv { // Bid serving is decided exclusively by the plan service (frozen per-slot // plans over the cfg.BuilderAPIEnabled baseline) — the handler's enabled // flag is deliberately never set. - planSvc := action_plan.NewPlanService(cfg, chainSvc, log) + cfgSvc := config.NewStaticService(cfg) + planSvc := action_plan.NewPlanService(cfgSvc, chainSvc, log) - h := NewHandler(&cfg.BuilderAPI, log, chainSvc, planSvc, + h := NewHandler(cfgSvc, log, chainSvc, planSvc, payload_builder.NewPayloadCache(10), registry) recorder := &stubSlotResultRecorder{} diff --git a/pkg/builderapi/epbs/preferences.go b/pkg/builderapi/epbs/preferences.go index ab81599..8bbfc68 100644 --- a/pkg/builderapi/epbs/preferences.go +++ b/pkg/builderapi/epbs/preferences.go @@ -30,8 +30,10 @@ func (h *Handler) HandleSubmitBuilderPreferences(w http.ResponseWriter, r *http. // The builder MUST check auth.message.builder_url against its own URL. Without a // configured URL it cannot perform that mandatory check, so treat it as a server - // misconfiguration (500) rather than a client error. - if h.cfg.BuilderURL == "" { + // misconfiguration (500) rather than a client error. One config snapshot + // covers the whole request. + builderURL := h.cfgSvc.Current().BuilderAPI.BuilderURL + if builderURL == "" { log.Error("submitBuilderPreferences: 500 — builder URL not configured; cannot verify auth.message.builder_url") writeError(w, http.StatusInternalServerError, "builder URL not configured") return @@ -75,10 +77,10 @@ func (h *Handler) HandleSubmitBuilderPreferences(w http.ResponseWriter, r *http. } // Check auth.message.data (the builder URL) matches this builder's URL (400 on mismatch). - if string(req.Auth.Message.Data) != h.cfg.BuilderURL { + if string(req.Auth.Message.Data) != builderURL { log.WithFields(logrus.Fields{ "auth_url": string(req.Auth.Message.Data), - "builder_url": h.cfg.BuilderURL, + "builder_url": builderURL, }).Warn("submitBuilderPreferences: builder_url mismatch") writeError(w, http.StatusBadRequest, "auth.message.data does not match this builder's URL") return diff --git a/pkg/builderapi/keyset_test.go b/pkg/builderapi/keyset_test.go index f3929a8..bc802ae 100644 --- a/pkg/builderapi/keyset_test.go +++ b/pkg/builderapi/keyset_test.go @@ -27,7 +27,7 @@ func newTestKeyRegistry(t *testing.T, builderIndices ...uint64) *builder_keys.Re MaxIndex: 32, }} - registry, err := builder_keys.NewRegistry(cfg, testEntryPrivkey, log) + registry, err := builder_keys.NewRegistry(config.NewStaticService(cfg), testEntryPrivkey, log) require.NoError(t, err) for keyIndex, builderIndex := range builderIndices { diff --git a/pkg/builderapi/legacy/get_header_test.go b/pkg/builderapi/legacy/get_header_test.go index e37ae39..7e8e305 100644 --- a/pkg/builderapi/legacy/get_header_test.go +++ b/pkg/builderapi/legacy/get_header_test.go @@ -137,10 +137,10 @@ func newGetHeaderTestEnv(t *testing.T, enabled bool, blockValueWei *big.Int) *ge }, } - planSvc := action_plan.NewPlanService(cfg, chainSvc, log) + planSvc := action_plan.NewPlanService(config.NewStaticService(cfg), chainSvc, log) store := memstore.New[phase0.BLSPubKey, *apiv1.SignedValidatorRegistration]() - h := NewHandler(&cfg.BuilderAPI, log, chainSvc, planSvc, payload_builder.NewPayloadCache(10), + h := NewHandler(log, chainSvc, planSvc, payload_builder.NewPayloadCache(10), store, registry) recorder := &stubSlotResultRecorder{} diff --git a/pkg/builderapi/legacy/handler.go b/pkg/builderapi/legacy/handler.go index c9afaf5..6163c94 100644 --- a/pkg/builderapi/legacy/handler.go +++ b/pkg/builderapi/legacy/handler.go @@ -16,7 +16,6 @@ import ( "github.com/ethpandaops/buildoor/pkg/action_plan" "github.com/ethpandaops/buildoor/pkg/builder_keys" "github.com/ethpandaops/buildoor/pkg/chain" - "github.com/ethpandaops/buildoor/pkg/config" "github.com/ethpandaops/buildoor/pkg/memstore" "github.com/ethpandaops/buildoor/pkg/payload_builder" ) @@ -76,7 +75,6 @@ type ProposalSubmitter interface { // (registerValidators, getHeader, submitBlindedBlock). It is constructed and // mounted by the parent builderapi.Server. type Handler struct { - cfg *config.BuilderAPIConfig // shared pointer, read live log logrus.FieldLogger chainSvc chain.Service payloadCache *payload_builder.PayloadCache @@ -99,16 +97,14 @@ type Handler struct { blocksPublished atomic.Uint64 } -// NewHandler creates a new pre-Gloas Builder API dialect handler. cfg is the -// shared mutable config pointer; values are read live, never copied out. +// NewHandler creates a new pre-Gloas Builder API dialect handler. // planSvc is the mandatory per-slot scheduling/settings authority consulted // (via Freeze) on every getHeader request. -func NewHandler(cfg *config.BuilderAPIConfig, log logrus.FieldLogger, chainSvc chain.Service, +func NewHandler(log logrus.FieldLogger, chainSvc chain.Service, planSvc *action_plan.PlanService, payloadCache *payload_builder.PayloadCache, validatorsStore *memstore.Store[phase0.BLSPubKey, *apiv1.SignedValidatorRegistration], registry *builder_keys.Registry) *Handler { return &Handler{ - cfg: cfg, log: log.WithField("component", "builderapi-legacy"), chainSvc: chainSvc, planSvc: planSvc, diff --git a/pkg/builderapi/legacy/handler_test.go b/pkg/builderapi/legacy/handler_test.go index 62ac922..80d450e 100644 --- a/pkg/builderapi/legacy/handler_test.go +++ b/pkg/builderapi/legacy/handler_test.go @@ -104,11 +104,12 @@ func (p *stubProposalSubmitter) SubmitProposal(_ context.Context, opts *api.Subm // plans stored). func newServingPlanService(chainSvc chain.Service) *action_plan.PlanService { return action_plan.NewPlanService( - &config.Config{APIPort: 8080, BuilderAPIEnabled: true}, chainSvc, logrus.New()) + config.NewStaticService(&config.Config{APIPort: 8080, BuilderAPIEnabled: true}), + chainSvc, logrus.New()) } func newTestHandler(chainSvc chain.Service, registry *builder_keys.Registry) *Handler { - return NewHandler(&config.BuilderAPIConfig{}, logrus.New(), chainSvc, + return NewHandler(logrus.New(), chainSvc, newServingPlanService(chainSvc), payload_builder.NewPayloadCache(10), memstore.New[phase0.BLSPubKey, *apiv1.SignedValidatorRegistration](), registry) } @@ -147,7 +148,7 @@ func TestHandleGetHeader_Success(t *testing.T) { store := memstore.New[phase0.BLSPubKey, *apiv1.SignedValidatorRegistration]() chainSvc := &stubChainService{currentFork: version.DataVersionFulu} - h := NewHandler(&config.BuilderAPIConfig{}, logrus.New(), chainSvc, + h := NewHandler(logrus.New(), chainSvc, newServingPlanService(chainSvc), payload_builder.NewPayloadCache(10), store, registry) pk := blsSigner.PublicKey() diff --git a/pkg/builderapi/legacy/keyset_test.go b/pkg/builderapi/legacy/keyset_test.go index e1dac5b..88cf704 100644 --- a/pkg/builderapi/legacy/keyset_test.go +++ b/pkg/builderapi/legacy/keyset_test.go @@ -27,7 +27,7 @@ func newTestKeyRegistry(t *testing.T, builderIndices ...uint64) *builder_keys.Re MaxIndex: 32, }} - registry, err := builder_keys.NewRegistry(cfg, testEntryPrivkey, log) + registry, err := builder_keys.NewRegistry(config.NewStaticService(cfg), testEntryPrivkey, log) require.NoError(t, err) for keyIndex, builderIndex := range builderIndices { diff --git a/pkg/builderapi/legacy/registrations_test.go b/pkg/builderapi/legacy/registrations_test.go index 6557d9f..2e5b7ca 100644 --- a/pkg/builderapi/legacy/registrations_test.go +++ b/pkg/builderapi/legacy/registrations_test.go @@ -18,7 +18,6 @@ import ( "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" - "github.com/ethpandaops/buildoor/pkg/config" "github.com/ethpandaops/buildoor/pkg/memstore" "github.com/ethpandaops/buildoor/pkg/payload_builder" "github.com/ethpandaops/buildoor/pkg/signer" @@ -159,7 +158,7 @@ func TestHandleRegisterValidators_OverwriteAndFlush(t *testing.T) { defer store.Stop() - h := NewHandler(&config.BuilderAPIConfig{}, logrus.New(), &stubChainService{}, + h := NewHandler(logrus.New(), &stubChainService{}, newServingPlanService(&stubChainService{}), payload_builder.NewPayloadCache(10), store, registry) h.SetEnabled(true) diff --git a/pkg/builderapi/mockchain_test.go b/pkg/builderapi/mockchain_test.go index 015093b..9c47bd3 100644 --- a/pkg/builderapi/mockchain_test.go +++ b/pkg/builderapi/mockchain_test.go @@ -20,7 +20,7 @@ import ( // plans stored). func newServingPlanService() *action_plan.PlanService { return action_plan.NewPlanService( - &config.Config{APIPort: 8080, BuilderAPIEnabled: true}, + config.NewStaticService(&config.Config{APIPort: 8080, BuilderAPIEnabled: true}), &mockChainService{}, logrus.New()) } @@ -72,3 +72,9 @@ func (m *mockChainService) GetValidatorPubkeyByIndex(phase0.ValidatorIndex) *pha } func (m *mockChainService) RefreshBuilders(context.Context) error { return nil } + +// staticSvc wraps a Builder-API section config in a full-config static +// settings service, as the server takes the settings service now. +func staticSvc(cfg *config.BuilderAPIConfig) *config.Service { + return config.NewStaticService(&config.Config{BuilderAPI: *cfg}) +} diff --git a/pkg/builderapi/server.go b/pkg/builderapi/server.go index bca5aa2..102d268 100644 --- a/pkg/builderapi/server.go +++ b/pkg/builderapi/server.go @@ -84,7 +84,7 @@ type RequestStats struct { // is NOT done here — the shared payload_bidder.InclusionTracker is the single // owner of won-block records, recording actual inclusion. type Server struct { - cfg *config.BuilderAPIConfig + cfgSvc *config.Service // settings source shared with the dialect handlers log *logrus.Logger chainSvc chain.Service payloadCache *payload_builder.PayloadCache // debug endpoints + dialect construction @@ -103,7 +103,7 @@ type Server struct { // validatorStore is optional (an in-memory store is created when nil); when // provided it is the shared instance also read by the legacy registration // settings resolver. -func NewServer(cfg *config.BuilderAPIConfig, log *logrus.Logger, chainSvc chain.Service, +func NewServer(cfgSvc *config.Service, log *logrus.Logger, chainSvc chain.Service, planSvc *action_plan.PlanService, payloadCache *payload_builder.PayloadCache, registry *builder_keys.Registry, validatorStore *memstore.Store[phase0.BLSPubKey, *apiv1.SignedValidatorRegistration]) *Server { @@ -113,13 +113,13 @@ func NewServer(cfg *config.BuilderAPIConfig, log *logrus.Logger, chainSvc chain. } return &Server{ - cfg: cfg, + cfgSvc: cfgSvc, log: log, chainSvc: chainSvc, payloadCache: payloadCache, validatorsStore: store, - legacy: legacy.NewHandler(cfg, log, chainSvc, planSvc, payloadCache, store, registry), - epbs: epbsapi.NewHandler(cfg, log, chainSvc, planSvc, payloadCache, registry), + legacy: legacy.NewHandler(log, chainSvc, planSvc, payloadCache, store, registry), + epbs: epbsapi.NewHandler(cfgSvc, log, chainSvc, planSvc, payloadCache, registry), } } diff --git a/pkg/builderapi/server_test.go b/pkg/builderapi/server_test.go index 857b770..8cf9bd8 100644 --- a/pkg/builderapi/server_test.go +++ b/pkg/builderapi/server_test.go @@ -42,7 +42,7 @@ func TestRegisterValidators_BuilderSpecsExample(t *testing.T) { // Uses the official builder-specs example from validators/testdata/signed_validator_registrations.json cfg := &config.BuilderAPIConfig{} log := logrus.New() - srv := NewServer(cfg, log, &mockChainService{}, newServingPlanService(), nil, nil, nil) + srv := NewServer(staticSvc(cfg), log, &mockChainService{}, newServingPlanService(), nil, nil, nil) req := httptest.NewRequest(http.MethodPost, "/eth/v1/builder/validators", bytes.NewReader(builderSpecsExampleJSON)) req.Header.Set("Content-Type", "application/json") @@ -57,7 +57,7 @@ func TestRegisterValidators_BuilderSpecsExample(t *testing.T) { func TestRegisterValidators_EmptyArray(t *testing.T) { cfg := &config.BuilderAPIConfig{} log := logrus.New() - srv := NewServer(cfg, log, &mockChainService{}, newServingPlanService(), nil, nil, nil) + srv := NewServer(staticSvc(cfg), log, &mockChainService{}, newServingPlanService(), nil, nil, nil) req := httptest.NewRequest(http.MethodPost, "/eth/v1/builder/validators", bytes.NewReader([]byte("[]"))) req.Header.Set("Content-Type", "application/json") @@ -72,7 +72,7 @@ func TestRegisterValidators_EmptyArray(t *testing.T) { func TestRegisterValidators_InvalidJSON(t *testing.T) { cfg := &config.BuilderAPIConfig{} log := logrus.New() - srv := NewServer(cfg, log, &mockChainService{}, newServingPlanService(), nil, nil, nil) + srv := NewServer(staticSvc(cfg), log, &mockChainService{}, newServingPlanService(), nil, nil, nil) req := httptest.NewRequest(http.MethodPost, "/eth/v1/builder/validators", bytes.NewReader([]byte("not json"))) req.Header.Set("Content-Type", "application/json") @@ -124,7 +124,7 @@ func TestRegisterValidators_ValidSignature(t *testing.T) { cfg := &config.BuilderAPIConfig{} log := logrus.New() - srv := NewServer(cfg, log, &mockChainService{}, newServingPlanService(), nil, nil, nil) + srv := NewServer(staticSvc(cfg), log, &mockChainService{}, newServingPlanService(), nil, nil, nil) req := httptest.NewRequest(http.MethodPost, "/eth/v1/builder/validators", bytes.NewReader(body)) req.Header.Set("Content-Type", "application/json") @@ -142,7 +142,7 @@ func TestRegisterValidators_ValidSignature(t *testing.T) { func TestRegisterValidators_MissingContentType(t *testing.T) { cfg := &config.BuilderAPIConfig{} log := logrus.New() - srv := NewServer(cfg, log, &mockChainService{}, newServingPlanService(), nil, nil, nil) + srv := NewServer(staticSvc(cfg), log, &mockChainService{}, newServingPlanService(), nil, nil, nil) req := httptest.NewRequest(http.MethodPost, "/eth/v1/builder/validators", bytes.NewReader([]byte("[]"))) // no Content-Type @@ -159,7 +159,7 @@ func TestGetHeader_NoPayload(t *testing.T) { log := logrus.New() registry := newTestKeyRegistry(t, 1) blsSigner := registry.Primary().BLSSigner() - srv := NewServer(cfg, log, &mockChainService{}, newServingPlanService(), nil, registry, nil) + srv := NewServer(staticSvc(cfg), log, &mockChainService{}, newServingPlanService(), nil, registry, nil) pk := blsSigner.PublicKey() url := "/eth/v1/builder/header/1/0x0000000000000000000000000000000000000000000000000000000000000000/0x" + hex.EncodeToString(pk[:]) @@ -175,7 +175,7 @@ func TestGetHeader_InvalidSlot(t *testing.T) { cfg := &config.BuilderAPIConfig{} log := logrus.New() registry := newTestKeyRegistry(t, 1) - srv := NewServer(cfg, log, &mockChainService{}, newServingPlanService(), + srv := NewServer(staticSvc(cfg), log, &mockChainService{}, newServingPlanService(), payload_builder.NewPayloadCache(10), registry, nil) srv.SetEnabled(true) @@ -222,7 +222,6 @@ func TestGetHeader_SubsidyInBidValue(t *testing.T) { // so the server and the plan service must share one config. fullCfg := &config.Config{APIPort: 8080, BuilderAPIEnabled: true} fullCfg.BuilderAPI.BlockValueSubsidyGwei = 1_000_000 - cfg := &fullCfg.BuilderAPI log := logrus.New() cache := payload_builder.NewPayloadCache(10) parentHash := phase0.Hash32(common.HexToHash("0x0000000000000000000000000000000000000000000000000000000000000001")) @@ -241,8 +240,8 @@ func TestGetHeader_SubsidyInBidValue(t *testing.T) { BlockValue: new(big.Int).SetUint64(500_000_000_000_000), // 0.0005 ETH in wei } cache.Store(event) - planSvc := action_plan.NewPlanService(fullCfg, &mockChainService{}, log) - srv := NewServer(cfg, log, &mockChainService{currentFork: version.DataVersionDeneb}, planSvc, cache, registry, nil) + planSvc := action_plan.NewPlanService(config.NewStaticService(fullCfg), &mockChainService{}, log) + srv := NewServer(config.NewStaticService(fullCfg), log, &mockChainService{currentFork: version.DataVersionDeneb}, planSvc, cache, registry, nil) srv.SetEnabled(true) req := httptest.NewRequest(http.MethodPost, "/eth/v1/builder/validators", bytes.NewReader(regs)) @@ -273,7 +272,7 @@ func TestGetHeader_SubsidyInBidValue(t *testing.T) { func TestSubmitBlindedBlockV2_InvalidJSON(t *testing.T) { cfg := &config.BuilderAPIConfig{} log := logrus.New() - srv := NewServer(cfg, log, &mockChainService{}, newServingPlanService(), payload_builder.NewPayloadCache(10), nil, nil) + srv := NewServer(staticSvc(cfg), log, &mockChainService{}, newServingPlanService(), payload_builder.NewPayloadCache(10), nil, nil) srv.SetEnabled(true) req := httptest.NewRequest(http.MethodPost, "/eth/v2/builder/blinded_blocks", bytes.NewReader([]byte("not json"))) @@ -288,7 +287,7 @@ func TestSubmitBlindedBlockV2_InvalidJSON(t *testing.T) { func TestSubmitBlindedBlockV2_MissingContentType(t *testing.T) { cfg := &config.BuilderAPIConfig{} log := logrus.New() - srv := NewServer(cfg, log, &mockChainService{}, newServingPlanService(), nil, nil, nil) + srv := NewServer(staticSvc(cfg), log, &mockChainService{}, newServingPlanService(), nil, nil, nil) srv.SetEnabled(true) req := httptest.NewRequest(http.MethodPost, "/eth/v2/builder/blinded_blocks", bytes.NewReader([]byte("{}"))) @@ -312,7 +311,7 @@ func (m *mockProposalSubmitter) SubmitProposal(_ context.Context, opts *api.Subm func TestSubmitBlindedBlockV2_NoMatchingPayload(t *testing.T) { cfg := &config.BuilderAPIConfig{} log := logrus.New() - srv := NewServer(cfg, log, &mockChainService{}, newServingPlanService(), payload_builder.NewPayloadCache(10), nil, nil) + srv := NewServer(staticSvc(cfg), log, &mockChainService{}, newServingPlanService(), payload_builder.NewPayloadCache(10), nil, nil) srv.SetEnabled(true) // Minimal Fulu (Electra-shaped) blinded block body: message.body.execution_payload_header.block_hash that won't be in cache @@ -341,7 +340,7 @@ func TestSubmitBlindedBlockV2_Success_UnblindAndPublish(t *testing.T) { log := logrus.New() cache := payload_builder.NewPayloadCache(10) submitter := &mockProposalSubmitter{} - srv := NewServer(cfg, log, &mockChainService{currentFork: version.DataVersionFulu}, newServingPlanService(), cache, nil, nil) + srv := NewServer(staticSvc(cfg), log, &mockChainService{currentFork: version.DataVersionFulu}, newServingPlanService(), cache, nil, nil) srv.legacy.SetCLClient(submitter) srv.SetEnabled(true) @@ -388,7 +387,7 @@ func TestSubmitBlindedBlockV1_Success_ReturnsPayload(t *testing.T) { log := logrus.New() cache := payload_builder.NewPayloadCache(10) submitter := &mockProposalSubmitter{} - srv := NewServer(cfg, log, &mockChainService{currentFork: version.DataVersionFulu}, newServingPlanService(), cache, nil, nil) + srv := NewServer(staticSvc(cfg), log, &mockChainService{currentFork: version.DataVersionFulu}, newServingPlanService(), cache, nil, nil) srv.legacy.SetCLClient(submitter) srv.SetEnabled(true) @@ -479,7 +478,7 @@ func TestRegisterValidators_SSZ(t *testing.T) { cfg := &config.BuilderAPIConfig{} log := logrus.New() - srv := NewServer(cfg, log, &mockChainService{}, newServingPlanService(), nil, nil, nil) + srv := NewServer(staticSvc(cfg), log, &mockChainService{}, newServingPlanService(), nil, nil, nil) req := httptest.NewRequest(http.MethodPost, "/eth/v1/builder/validators", bytes.NewReader(body)) req.Header.Set("Content-Type", "application/octet-stream") @@ -510,7 +509,7 @@ func TestRegisterValidators_SSZ(t *testing.T) { func TestUnknownEthEndpoint_JSON404(t *testing.T) { cfg := &config.BuilderAPIConfig{} log := logrus.New() - srv := NewServer(cfg, log, &mockChainService{}, newServingPlanService(), nil, nil, nil) + srv := NewServer(staticSvc(cfg), log, &mockChainService{}, newServingPlanService(), nil, nil, nil) for _, tc := range []struct { method string diff --git a/pkg/chain/headvotes.go b/pkg/chain/headvotes.go index 86c53e5..e0efc8b 100644 --- a/pkg/chain/headvotes.go +++ b/pkg/chain/headvotes.go @@ -147,7 +147,7 @@ func (s *slotVoteState) primary() (phase0.Root, *rootVoteState) { // percent-steps on a flush interval; crossing the configured participation // threshold fires immediately. type HeadVoteTracker struct { - cfg *config.Config // shared live config; threshold read per check, never cached + cfgSvc *config.Service // settings source; threshold loads a fresh snapshot per check chainSvc Service clClient *beacon.Client log logrus.FieldLogger @@ -172,13 +172,13 @@ type HeadVoteTracker struct { // NewHeadVoteTracker creates a new head vote tracker. func NewHeadVoteTracker( - cfg *config.Config, + cfgSvc *config.Service, chainSvc Service, clClient *beacon.Client, log logrus.FieldLogger, ) *HeadVoteTracker { return &HeadVoteTracker{ - cfg: cfg, + cfgSvc: cfgSvc, chainSvc: chainSvc, clClient: clClient, log: log.WithField("component", "head-vote-tracker"), @@ -363,14 +363,15 @@ func (t *HeadVoteTracker) run() { } } -// thresholdPct returns the live-configured participation threshold in percent -// (0 = disabled). Read on every check so UI overrides apply immediately. +// thresholdPct returns the configured participation threshold in percent +// (0 = disabled). A fresh snapshot per check, so UI overrides apply on the +// next check. func (t *HeadVoteTracker) thresholdPct() float64 { - if t.cfg == nil { + if t.cfgSvc == nil { return 0 } - return float64(t.cfg.EPBS.HeadVoteThresholdPct) + return float64(t.cfgSvc.Current().EPBS.HeadVoteThresholdPct) } // handleHeadEvent marks the slot's head root as the primary tracked root and diff --git a/pkg/chain/headvotes_test.go b/pkg/chain/headvotes_test.go index d6ffaca..e7cebe9 100644 --- a/pkg/chain/headvotes_test.go +++ b/pkg/chain/headvotes_test.go @@ -124,7 +124,7 @@ func newVoteTestTracker( log := logrus.New() log.SetLevel(logrus.PanicLevel) - return NewHeadVoteTracker(cfg, chainSvc, nil, log), chainSvc + return NewHeadVoteTracker(config.NewStaticService(cfg), chainSvc, nil, log), chainSvc } // drainUpdates reads all buffered updates from the subscription. diff --git a/pkg/chain/service.go b/pkg/chain/service.go index 80d15bd..fb272cf 100644 --- a/pkg/chain/service.go +++ b/pkg/chain/service.go @@ -65,7 +65,7 @@ var _ Service = (*service)(nil) // service is the implementation of Service. type service struct { - cfg *config.Config + cfgSvc *config.Service // settings source; one snapshot per read clClient *beacon.Client chainSpec *ChainSpec genesis *beacon.Genesis @@ -99,14 +99,14 @@ type service struct { // NewService creates a new chain service. func NewService( - cfg *config.Config, + cfgSvc *config.Service, clClient *beacon.Client, chainSpec *ChainSpec, genesis *beacon.Genesis, log logrus.FieldLogger, ) Service { return &service{ - cfg: cfg, + cfgSvc: cfgSvc, clClient: clClient, chainSpec: chainSpec, genesis: genesis, @@ -127,7 +127,7 @@ func (s *service) Start(ctx context.Context) error { } // Start head vote tracker - s.headVoteTracker = NewHeadVoteTracker(s.cfg, s, s.clClient, s.log) + s.headVoteTracker = NewHeadVoteTracker(s.cfgSvc, s, s.clClient, s.log) s.headVoteTracker.Start(s.ctx) // Start canonical head tracker diff --git a/pkg/config/settings.go b/pkg/config/settings.go index cb593ef..6aed1b2 100644 --- a/pkg/config/settings.go +++ b/pkg/config/settings.go @@ -5,6 +5,7 @@ import ( "encoding/json" "fmt" "sync" + "sync/atomic" "time" "github.com/sirupsen/logrus" @@ -29,22 +30,28 @@ type keyState struct { uiSeq int64 } -// Service is the central authority for buildoor's mutable runtime configuration. -// It owns the effective Config every module reads and is the single writer, -// layering three sources: hardcoded defaults < CLI-supplied < UI override. CLI -// and UI are resolved by recency (a monotonic seq), not a fixed priority: a CLI -// value that changed since the last run wins over an older UI override, while an -// unchanged CLI flag lets a newer UI override win. UI overrides persist across -// restarts via the optional state-db. The effective Config is the same pointer -// handed to every module, so writes (applied in place under the service lock) -// are observed live by all readers. +// Service is the central authority for buildoor's mutable runtime configuration +// and the dependency modules hold instead of a raw *Config. It is the single +// writer, layering three sources: hardcoded defaults < CLI-supplied < UI +// override. CLI and UI are resolved by recency (a monotonic seq), not a fixed +// priority: a CLI value that changed since the last run wins over an older UI +// override, while an unchanged CLI flag lets a newer UI override win. UI +// overrides persist across restarts via the optional state-db. +// +// Published configs are immutable snapshots: every applied change builds a +// fresh Config generation and swaps it in atomically, so a snapshot obtained +// from Current() never changes underneath its reader. Consumers load exactly +// one snapshot per operation (request, tick, build, reconcile pass) and thread +// it down the call stack; only the Service itself may live in a struct field — +// storing a *Config freezes that consumer on a stale generation. type Service struct { - log logrus.FieldLogger - store *db.Database - fields []Field - byKey map[string]Field - defaults *Config // pristine, slot-adjusted defaults — the floor - effective *Config // shared config; mutated in place + log logrus.FieldLogger + store *db.Database + fields []Field + byKey map[string]Field + defaults *Config // pristine, slot-adjusted defaults — the floor + current atomic.Pointer[Config] // latest immutable snapshot; swapped by recompute + slotDuration time.Duration // 0 = unknown; skips the reveal-time upper bound check mu sync.Mutex seq int64 @@ -54,22 +61,30 @@ type Service struct { // New constructs the settings service. // -// - effective is the resolved operator config (defaults + flags/env/file, -// already slot-adjusted); it becomes the shared config modules read and is -// mutated in place to apply overrides. +// - operator is the resolved operator config (defaults + flags/env/file, +// already slot-adjusted); it seeds the first published snapshot and is the +// source of the CLI layer's values. The constructor's final recompute +// publishes a fresh generation with persisted overrides applied, so the +// operator config itself is never mutated. // - defaults is a pristine, slot-adjusted default Config used as the floor. // - supplied maps each field key to whether the operator explicitly provided // it (viper.IsSet); only supplied keys form the CLI layer. +// - slotDuration is the network's slot duration, used to bound +// reveal-relative timing overrides; pass 0 if not yet known. // - store is the optional state-db (may be disabled). -func NewService(effective, defaults *Config, supplied map[string]bool, store *db.Database, log logrus.FieldLogger) (*Service, error) { +func NewService( + operator, defaults *Config, supplied map[string]bool, slotDuration time.Duration, + store *db.Database, log logrus.FieldLogger, +) (*Service, error) { s := &Service{ - log: log.WithField("module", "settings"), - store: store, - fields: Fields(), - defaults: defaults, - effective: effective, - keyState: make(map[string]*keyState), + log: log.WithField("module", "settings"), + store: store, + fields: Fields(), + defaults: defaults, + slotDuration: slotDuration, + keyState: make(map[string]*keyState), } + s.current.Store(operator) s.byKey = make(map[string]Field, len(s.fields)) for _, f := range s.fields { @@ -117,6 +132,13 @@ func NewService(effective, defaults *Config, supplied map[string]bool, store *db } // Pass 2: reconcile the CLI layer against what the operator supplied now. + // Every changed row is batched into a single transaction below rather + // than persisted one key at a time, so a crash or a transient state-db + // failure mid-pass can never leave only a prefix of this reconciliation + // durable. + now := time.Now().UnixMilli() + changedRows := make([]db.SettingRow, 0, len(s.fields)) + for _, f := range s.fields { ks := s.keyState[f.Key] row := rowByKey[f.Key] @@ -137,7 +159,7 @@ func NewService(effective, defaults *Config, supplied map[string]bool, store *db switch { case isSupplied: - suppliedVal := f.Get(s.effective) + suppliedVal := f.Get(operator) if !storedHasCLI || !f.Equal(suppliedVal, storedCLI) { // New or changed operator value — counts as a fresh write. ks.hasCLI = true @@ -156,19 +178,52 @@ func NewService(effective, defaults *Config, supplied map[string]bool, store *db } if changed { - s.persist(f, ks, SourceCLI) + built, err := buildSettingRow(f, ks, SourceCLI, now) + if err != nil { + return nil, fmt.Errorf("encode %q: %w", f.Key, err) + } + + changedRows = append(changedRows, built) } } + if err := store.PutSettings(changedRows); err != nil { + return nil, fmt.Errorf("persist cli settings: %w", err) + } + s.recompute() + // Operator input was validated before construction; a persisted UI + // override from an earlier run may still violate the timing invariants + // (e.g. saved by an older release without these checks). It stays in + // effect — dropping it silently would surprise more — but is called out + // loudly. + if err := ValidateTimingBounds(s.current.Load(), s.slotDuration); err != nil { + s.log.WithError(err).Warn("persisted settings violate timing invariants; fix via the settings API") + } + return s, nil } -// Load returns the effective The returned pointer is the shared config -// modules read; callers must treat it as read-only. -func (s *Service) Load() *Config { - return s.effective +// NewStaticService wraps a fixed config in a read-only Service for consumers +// that need a config source without the settings machinery (tests, one-shot +// commands). Current() always returns the given pointer; Set/SetMany reject +// every key. Because the same pointer is served on every call, single-threaded +// callers may still mutate the config between operations. +func NewStaticService(cfg *Config) *Service { + s := &Service{} + s.current.Store(cfg) + + return s +} + +// Current returns the latest effective config snapshot. Snapshots are +// immutable: the returned Config never changes, so all reads from it are +// coherent (one settings generation). Load one snapshot per operation and pass +// it down; never store it in a struct field — later generations would not be +// observed. +func (s *Service) Current() *Config { + return s.current.Load() } // OnChange registers a callback invoked (outside the service lock) after every @@ -185,9 +240,15 @@ func (s *Service) Set(key string, raw json.RawMessage, actor string) error { return s.SetMany(map[string]json.RawMessage{key: raw}, actor) } -// SetMany applies a batch of UI overrides atomically: all values are validated -// and decoded first, then applied, persisted, and the effective config -// recomputed before subscribers are notified once. +// SetMany applies a batch of UI overrides atomically: all values are +// validated and decoded first, then checked together against the timing +// invariants (ValidateTimingBounds) the batch would produce, then persisted +// in one state-db transaction, and only once that durably commits (or +// persistence is disabled) applied and published as a fresh snapshot, +// before subscribers are notified once. Persistence failing anywhere in the +// batch leaves both the published snapshot and the state-db exactly as they +// were before the call — a caller that observes an error can rely on +// nothing having changed. func (s *Service) SetMany(updates map[string]json.RawMessage, actor string) error { s.mu.Lock() @@ -214,13 +275,54 @@ func (s *Service) SetMany(updates map[string]json.RawMessage, actor string) erro decoded[key] = v } + // Validate the batch's cross-field / slot-relative timing invariants + // against what the effective config would become if it commits: apply it + // to a scratch copy of the current snapshot first, leaving fields the + // batch doesn't touch at their current effective value. The scratch copy + // is never published — the new snapshot comes from recompute below. + scratch := *s.current.Load() for key, v := range decoded { - f := s.byKey[key] - ks := s.keyState[key] - ks.hasUI = true - ks.uiValue = v - ks.uiSeq = s.nextSeq() - s.persist(f, ks, actor) + if err := s.byKey[key].Set(&scratch, v); err != nil { + s.mu.Unlock() + return fmt.Errorf("apply %q: %w", key, err) + } + } + + if err := ValidateTimingBounds(&scratch, s.slotDuration); err != nil { + s.mu.Unlock() + return err + } + + // Stage the new key state and build every row up front: nothing is + // applied to s.keyState (and no snapshot is published) until the whole + // batch durably persists. + now := time.Now().UnixMilli() + rows := make([]db.SettingRow, 0, len(decoded)) + staged := make(map[string]*keyState, len(decoded)) + + for key, v := range decoded { + next := *s.keyState[key] + next.hasUI = true + next.uiValue = v + next.uiSeq = s.nextSeq() + + row, err := buildSettingRow(s.byKey[key], &next, actor, now) + if err != nil { + s.mu.Unlock() + return fmt.Errorf("encode %q: %w", key, err) + } + + rows = append(rows, row) + staged[key] = &next + } + + if err := s.store.PutSettings(rows); err != nil { + s.mu.Unlock() + return fmt.Errorf("persist settings: %w", err) + } + + for key, next := range staged { + s.keyState[key] = next } s.recompute() @@ -236,9 +338,15 @@ func (s *Service) SetMany(updates map[string]json.RawMessage, actor string) erro return nil } -// recompute rebuilds the effective config in place: each registered field is set -// to the highest-seq layer present (defaults are the seq-0 floor). Must hold mu. +// recompute publishes a fresh config generation: the current snapshot is +// shallow-copied (all Config fields are values, so the copy shares nothing +// mutable), each registered field is set to the highest-seq layer present +// (defaults are the seq-0 floor), and the result is swapped in atomically. +// Snapshots already handed out stay untouched. Must hold mu. func (s *Service) recompute() { + next := new(Config) + *next = *s.current.Load() + for _, f := range s.fields { ks := s.keyState[f.Key] val := f.Get(s.defaults) @@ -253,10 +361,12 @@ func (s *Service) recompute() { val = ks.uiValue } - if err := f.Set(s.effective, val); err != nil { + if err := f.Set(next, val); err != nil { s.log.WithError(err).WithField("key", f.Key).Error("failed to apply setting") } } + + s.current.Store(next) } // nextSeq allocates a monotonic sequence number. Must hold mu. @@ -265,35 +375,37 @@ func (s *Service) nextSeq() int64 { return s.seq } -// persist writes the full 3-way row for a key to the state-db. Must hold mu. -func (s *Service) persist(f Field, ks *keyState, actor string) { - if !s.store.Enabled() { - return - } - +// buildSettingRow builds the full 3-way persisted row for a key from its +// keyState, without writing anything: callers batch rows from several keys +// into a single db.Database.PutSettings transaction. +func buildSettingRow(f Field, ks *keyState, actor string, updatedAt int64) (db.SettingRow, error) { row := db.SettingRow{ Key: f.Key, - UpdatedAt: time.Now().UnixMilli(), + UpdatedAt: updatedAt, Actor: actor, } if ks.hasCLI { - if b, err := f.Encode(ks.cliValue); err == nil { - row.CLIValue = sql.NullString{String: string(b), Valid: true} - row.CLISeq = ks.cliSeq + b, err := f.Encode(ks.cliValue) + if err != nil { + return db.SettingRow{}, fmt.Errorf("encode %q cli value: %w", f.Key, err) } + + row.CLIValue = sql.NullString{String: string(b), Valid: true} + row.CLISeq = ks.cliSeq } if ks.hasUI { - if b, err := f.Encode(ks.uiValue); err == nil { - row.UIValue = sql.NullString{String: string(b), Valid: true} - row.UISeq = ks.uiSeq + b, err := f.Encode(ks.uiValue) + if err != nil { + return db.SettingRow{}, fmt.Errorf("encode %q ui value: %w", f.Key, err) } - } - if err := s.store.PutSetting(row); err != nil { - s.log.WithError(err).WithField("key", f.Key).Warn("failed to persist setting") + row.UIValue = sql.NullString{String: string(b), Valid: true} + row.UISeq = ks.uiSeq } + + return row, nil } // validateValue performs light per-field validation of incoming UI values. @@ -316,3 +428,32 @@ func validateValue(key string, v any) error { return nil } + +// ValidateTimingBounds checks the cross-field / slot-relative invariants a +// mutable timing setting must satisfy regardless of which layer (CLI, config +// file, or UI override) sets it. Neither violation crashes or corrupts +// anything by itself — an inverted bid window just suppresses bidding, and a +// too-late reveal time is cleanly skipped rather than published wrong — but +// both silently defeat the feature for the rest of the run, so they are +// rejected outright at the point a value is accepted. Per-slot action-plan +// overrides are validated separately (pkg/action_plan) and deliberately stay +// free of these bounds — chaos scenarios belong in plans, not the global +// baseline. slotDuration <= 0 skips the reveal-time upper bound (unknown +// yet, e.g. before the chain spec has been fetched at startup). +func ValidateTimingBounds(cfg *Config, slotDuration time.Duration) error { + if cfg.EPBS.BidStartTime > cfg.EPBS.BidEndTime { + return fmt.Errorf("%s (%dms) must not be after %s (%dms)", + KeyEPBSBidStartTime, cfg.EPBS.BidStartTime, KeyEPBSBidEndTime, cfg.EPBS.BidEndTime) + } + + if cfg.Reveal.TimeMs < 0 { + return fmt.Errorf("%s must not be negative, got %dms", KeyRevealTimeMs, cfg.Reveal.TimeMs) + } + + if slotDuration > 0 && time.Duration(cfg.Reveal.TimeMs)*time.Millisecond >= slotDuration { + return fmt.Errorf("%s (%dms) must be less than the slot duration (%s)", + KeyRevealTimeMs, cfg.Reveal.TimeMs, slotDuration) + } + + return nil +} diff --git a/pkg/config/settings_persist_test.go b/pkg/config/settings_persist_test.go new file mode 100644 index 0000000..77fcdbf --- /dev/null +++ b/pkg/config/settings_persist_test.go @@ -0,0 +1,195 @@ +package config + +// SetMany persistence and validation semantics: a batch persists in a single +// state-db transaction BEFORE anything is applied or published, so a failed +// write leaves both the published snapshot and the state-db exactly as they +// were (previously persistence was per-key, silently lossy, and always +// reported success). ValidateTimingBounds rejects writes that would invert +// the bid window or push the reveal time past the slot deadline — reachable +// by any client on --api-port when no auth provider is configured. + +import ( + "encoding/json" + "path/filepath" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/ethpandaops/buildoor/pkg/db" +) + +func setMany(t *testing.T, svc *Service, updates map[string]any, actor string) error { + t.Helper() + + raw := make(map[string]json.RawMessage, len(updates)) + + for k, v := range updates { + b, err := json.Marshal(v) + require.NoError(t, err) + raw[k] = b + } + + return svc.SetMany(raw, actor) +} + +// TestSetMany_PersistFailureLeavesEverythingUnchanged forces the state-db +// write to fail (by closing the underlying connection out from under the +// service) and confirms SetMany reports the failure and changes nothing — +// neither the published snapshot nor what was actually durable. +func TestSetMany_PersistFailureLeavesEverythingUnchanged(t *testing.T) { + dir := t.TempDir() + store := db.NewDatabase(&db.Config{File: filepath.Join(dir, "state.db")}, testLogger()) + require.NoError(t, store.Init()) + + defaults := defaultsConfig() + svc := boot(t, store, defaults, nil) + + require.NoError(t, setMany(t, svc, map[string]any{subsidyKey: uint64(600)}, "tester")) + require.Equal(t, uint64(600), svc.Current().EPBS.BidSubsidy) + + // Force every subsequent write to fail. + require.NoError(t, store.Close()) + + err := setMany(t, svc, map[string]any{subsidyKey: uint64(700)}, "tester") + require.Error(t, err) + + // The published snapshot is untouched by the failed write. + assert.Equal(t, uint64(600), svc.Current().EPBS.BidSubsidy) +} + +// TestSetMany_BatchIsAtomic confirms a batch touching several keys either +// commits entirely or not at all: forcing the persist step to fail must +// leave every key in the batch unchanged, not just the one that happened to +// trigger the failure. +func TestSetMany_BatchIsAtomic(t *testing.T) { + dir := t.TempDir() + store := db.NewDatabase(&db.Config{File: filepath.Join(dir, "state.db")}, testLogger()) + require.NoError(t, store.Init()) + + defaults := defaultsConfig() + svc := boot(t, store, defaults, nil) + + require.NoError(t, setMany(t, svc, map[string]any{ + subsidyKey: uint64(500), + KeyEPBSBidMinAmount: uint64(1000), + KeyBuilderAPIOnDemandBuild: true, + }, "tester")) + + require.Equal(t, uint64(500), svc.Current().EPBS.BidSubsidy) + require.Equal(t, uint64(1000), svc.Current().EPBS.BidMinAmount) + require.True(t, svc.Current().BuilderAPI.OnDemandBuild) + + require.NoError(t, store.Close()) + + err := setMany(t, svc, map[string]any{ + subsidyKey: uint64(999), + KeyEPBSBidMinAmount: uint64(999), + KeyBuilderAPIOnDemandBuild: false, + }, "tester") + require.Error(t, err) + + // None of the three keys moved, not just the one whose write happened + // to fail first. + assert.Equal(t, uint64(500), svc.Current().EPBS.BidSubsidy) + assert.Equal(t, uint64(1000), svc.Current().EPBS.BidMinAmount) + assert.True(t, svc.Current().BuilderAPI.OnDemandBuild) +} + +func TestValidateTimingBounds(t *testing.T) { + tests := []struct { + name string + mutate func(*Config) + slotDuration time.Duration + wantErr bool + }{ + { + name: "defaults are valid", + mutate: func(*Config) {}, + slotDuration: 12 * time.Second, + }, + { + name: "inverted bid window rejected", + mutate: func(c *Config) { + c.EPBS.BidStartTime = -100 + c.EPBS.BidEndTime = -400 + }, + slotDuration: 12 * time.Second, + wantErr: true, + }, + { + name: "equal bid start/end allowed", + mutate: func(c *Config) { + c.EPBS.BidStartTime = -400 + c.EPBS.BidEndTime = -400 + }, + slotDuration: 12 * time.Second, + }, + { + name: "negative reveal time rejected", + mutate: func(c *Config) { + c.Reveal.TimeMs = -1 + }, + slotDuration: 12 * time.Second, + wantErr: true, + }, + { + name: "reveal time past the slot deadline rejected", + mutate: func(c *Config) { + c.Reveal.TimeMs = 12000 + }, + slotDuration: 12 * time.Second, + wantErr: true, + }, + { + name: "reveal time bound skipped when slot duration unknown", + mutate: func(c *Config) { + c.Reveal.TimeMs = 999999 + }, + slotDuration: 0, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + cfg := defaultsConfig() + tt.mutate(cfg) + + err := ValidateTimingBounds(cfg, tt.slotDuration) + if tt.wantErr { + assert.Error(t, err) + } else { + assert.NoError(t, err) + } + }) + } +} + +// TestSetMany_RejectsInvertedBidWindow guards the actual write path: SetMany +// itself, not just the standalone validator, must reject a write that would +// invert the bid window — including when only one of the two fields is +// touched by this particular call (the other must be compared against its +// current effective value, not silently skipped). +func TestSetMany_RejectsInvertedBidWindow(t *testing.T) { + dir := t.TempDir() + store := db.NewDatabase(&db.Config{File: filepath.Join(dir, "state.db")}, testLogger()) + require.NoError(t, store.Init()) + + defaults := defaultsConfig() + svc := boot(t, store, defaults, nil) + + originalStart := svc.Current().EPBS.BidStartTime + originalEnd := svc.Current().EPBS.BidEndTime + require.Less(t, originalStart, originalEnd) + + // Only bid_end_time is touched, but pushed before the CURRENT + // (untouched) bid_start_time — must still be rejected. + err := setMany(t, svc, map[string]any{KeyEPBSBidEndTime: originalStart - 1}, "tester") + require.Error(t, err) + assert.Equal(t, originalEnd, svc.Current().EPBS.BidEndTime, "rejected write must not partially apply") + + // A reveal time at or past the slot deadline is rejected too. + err = setMany(t, svc, map[string]any{KeyRevealTimeMs: int64(12000)}, "tester") + require.Error(t, err) +} diff --git a/pkg/config/settings_race_test.go b/pkg/config/settings_race_test.go new file mode 100644 index 0000000..efd74d0 --- /dev/null +++ b/pkg/config/settings_race_test.go @@ -0,0 +1,70 @@ +package config + +import ( + "encoding/json" + "strconv" + "strings" + "sync" + "testing" + + "github.com/stretchr/testify/require" + + "github.com/ethpandaops/buildoor/pkg/db" +) + +// TestSnapshots_RaceFree drives SetMany (the real write path) against +// concurrent snapshot readers under the race detector. Published snapshots are +// immutable, so a reader must always observe a coherent settings generation: +// no torn string reads, and never a mix of two SetMany batches within one +// snapshot. +func TestSnapshots_RaceFree(t *testing.T) { + store := db.NewDatabase(&db.Config{}, testLogger()) + require.NoError(t, store.Init()) + + svc := boot(t, store, defaultsConfig(), nil) + + var wg sync.WaitGroup + + stop := make(chan struct{}) + + // Writer: every batch sets the subsidy and the extra-data prefix to the + // same sequence number, so a coherent snapshot always agrees on both. + wg.Add(1) + + go func() { + defer wg.Done() + + for i := 0; ; i++ { + select { + case <-stop: + return + default: + } + + n := strconv.Itoa(i) + err := svc.SetMany(map[string]json.RawMessage{ + KeyEPBSBidSubsidy: json.RawMessage(n), + KeyExtraData: json.RawMessage(`"gen-` + n + `"`), + }, "race-test") + if err != nil { + t.Errorf("SetMany: %v", err) + return + } + } + }() + + for range 10_000 { + cfg := svc.Current() + + // Snapshots taken before the first batch keep the defaults; once a + // batch is visible, both fields must agree on the generation. + if gen, ok := strings.CutPrefix(cfg.ExtraData, "gen-"); ok { + if want := strconv.FormatUint(cfg.EPBS.BidSubsidy, 10); gen != want { + t.Fatalf("torn snapshot: subsidy %d but extra data %q", cfg.EPBS.BidSubsidy, cfg.ExtraData) + } + } + } + + close(stop) + wg.Wait() +} diff --git a/pkg/config/settings_test.go b/pkg/config/settings_test.go index 373773e..c353b1f 100644 --- a/pkg/config/settings_test.go +++ b/pkg/config/settings_test.go @@ -5,6 +5,7 @@ import ( "io" "path/filepath" "testing" + "time" "github.com/sirupsen/logrus" "github.com/stretchr/testify/require" @@ -42,7 +43,7 @@ func boot(t *testing.T, store *db.Database, defaults *Config, suppliedVal *uint6 supplied[subsidyKey] = true } - svc, err := NewService(&eff, defaults, supplied, store, testLogger()) + svc, err := NewService(&eff, defaults, supplied, 12*time.Second, store, testLogger()) require.NoError(t, err) return svc @@ -71,34 +72,34 @@ func TestThreeWayResolution(t *testing.T) { // 1. Boot with --epbs-bid-subsidy 500 -> CLI wins over default. svc := boot(t, store, defaults, u64(500)) - require.Equal(t, uint64(500), svc.Load().EPBS.BidSubsidy) + require.Equal(t, uint64(500), svc.Current().EPBS.BidSubsidy) // 2. UI sets 600 -> UI wins. setSubsidy(t, svc, 600) - require.Equal(t, uint64(600), svc.Load().EPBS.BidSubsidy) + require.Equal(t, uint64(600), svc.Current().EPBS.BidSubsidy) // 3. Restart, flag unchanged at 500 -> UI override (600) still wins. svc = boot(t, store, defaults, u64(500)) - require.Equal(t, uint64(600), svc.Load().EPBS.BidSubsidy) + require.Equal(t, uint64(600), svc.Current().EPBS.BidSubsidy) // 4. Operator changes the flag to 700 -> CLI change wins over old UI value. svc = boot(t, store, defaults, u64(700)) - require.Equal(t, uint64(700), svc.Load().EPBS.BidSubsidy) + require.Equal(t, uint64(700), svc.Current().EPBS.BidSubsidy) // 5. Restart, flag unchanged at 700 -> CLI stays (UI 600 does not resurrect). svc = boot(t, store, defaults, u64(700)) - require.Equal(t, uint64(700), svc.Load().EPBS.BidSubsidy) + require.Equal(t, uint64(700), svc.Current().EPBS.BidSubsidy) // 6. UI sets 800 -> UI wins again. setSubsidy(t, svc, 800) - require.Equal(t, uint64(800), svc.Load().EPBS.BidSubsidy) + require.Equal(t, uint64(800), svc.Current().EPBS.BidSubsidy) // 7. Upgrade safety: flag removed, hardcoded default bumped to 550M. // The default bump must NOT clobber the UI override. bumped := defaultsConfig() bumped.EPBS.BidSubsidy = 550000000 svc = boot(t, store, bumped, nil) - require.Equal(t, uint64(800), svc.Load().EPBS.BidSubsidy) + require.Equal(t, uint64(800), svc.Current().EPBS.BidSubsidy) require.NoError(t, store.Close()) } @@ -113,14 +114,14 @@ func TestDisabledDBNoPersistence(t *testing.T) { defaults := defaultsConfig() svc := boot(t, store, defaults, nil) - require.Equal(t, defaults.EPBS.BidSubsidy, svc.Load().EPBS.BidSubsidy) + require.Equal(t, defaults.EPBS.BidSubsidy, svc.Current().EPBS.BidSubsidy) setSubsidy(t, svc, 123) - require.Equal(t, uint64(123), svc.Load().EPBS.BidSubsidy) + require.Equal(t, uint64(123), svc.Current().EPBS.BidSubsidy) // New "boot" — no persistence, falls back to the default. svc = boot(t, store, defaults, nil) - require.Equal(t, defaults.EPBS.BidSubsidy, svc.Load().EPBS.BidSubsidy) + require.Equal(t, defaults.EPBS.BidSubsidy, svc.Current().EPBS.BidSubsidy) } // TestUnsuppliedUsesDefault verifies an unsupplied key resolves to the default @@ -134,7 +135,7 @@ func TestUnsuppliedUsesDefault(t *testing.T) { eff := *defaults eff.EPBS.BidSubsidy = 999 // present in effective but NOT operator-supplied - svc, err := NewService(&eff, defaults, map[string]bool{}, store, testLogger()) + svc, err := NewService(&eff, defaults, map[string]bool{}, 12*time.Second, store, testLogger()) require.NoError(t, err) - require.Equal(t, defaults.EPBS.BidSubsidy, svc.Load().EPBS.BidSubsidy) + require.Equal(t, defaults.EPBS.BidSubsidy, svc.Current().EPBS.BidSubsidy) } diff --git a/pkg/db/settings.go b/pkg/db/settings.go index d82a331..32886ce 100644 --- a/pkg/db/settings.go +++ b/pkg/db/settings.go @@ -2,6 +2,7 @@ package db import ( "database/sql" + "fmt" "github.com/jmoiron/sqlx" ) @@ -43,21 +44,27 @@ func (d *Database) GetSettings() ([]SettingRow, error) { return rows, nil } -// PutSetting upserts the full 3-way state for a settings key. No-op when the -// database is disabled. The settings service owns the in-memory authority and -// always writes the complete row, so a plain INSERT OR REPLACE is correct. -func (d *Database) PutSetting(row SettingRow) error { - if !d.enabled { +// PutSettings upserts the full 3-way state for a batch of settings keys in a +// single transaction: either every row commits or none do. No-op when the +// database is disabled or the batch is empty. The settings service owns the +// in-memory authority and always writes the complete row, so a plain INSERT +// OR REPLACE per row is correct. +func (d *Database) PutSettings(rows []SettingRow) error { + if !d.enabled || len(rows) == 0 { return nil } return d.RunDBTransaction(func(tx *sqlx.Tx) error { - _, err := tx.Exec(` - INSERT OR REPLACE INTO settings - (key, cli_value, cli_seq, ui_value, ui_seq, updated_at, actor) - VALUES ($1, $2, $3, $4, $5, $6, $7)`, - row.Key, row.CLIValue, row.CLISeq, row.UIValue, row.UISeq, row.UpdatedAt, row.Actor) + for _, row := range rows { + if _, err := tx.Exec(` + INSERT OR REPLACE INTO settings + (key, cli_value, cli_seq, ui_value, ui_seq, updated_at, actor) + VALUES ($1, $2, $3, $4, $5, $6, $7)`, + row.Key, row.CLIValue, row.CLISeq, row.UIValue, row.UISeq, row.UpdatedAt, row.Actor); err != nil { + return fmt.Errorf("upsert setting %q: %w", row.Key, err) + } + } - return err + return nil }) } diff --git a/pkg/db/settings_test.go b/pkg/db/settings_test.go new file mode 100644 index 0000000..85409b9 --- /dev/null +++ b/pkg/db/settings_test.go @@ -0,0 +1,60 @@ +package db + +import ( + "database/sql" + "io" + "testing" + + "github.com/sirupsen/logrus" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// TestPutSettingsBatchRoundTrip confirms a multi-row batch upserts every row +// in the single transaction PutSettings runs. +func TestPutSettingsBatchRoundTrip(t *testing.T) { + d := testDB(t) + + require.NoError(t, d.PutSettings([]SettingRow{ + {Key: "a", UIValue: sql.NullString{String: `1`, Valid: true}, UISeq: 1, UpdatedAt: 100}, + {Key: "b", UIValue: sql.NullString{String: `2`, Valid: true}, UISeq: 1, UpdatedAt: 100}, + {Key: "c", UIValue: sql.NullString{String: `3`, Valid: true}, UISeq: 1, UpdatedAt: 100}, + })) + + rows, err := d.GetSettings() + require.NoError(t, err) + + byKey := make(map[string]SettingRow, len(rows)) + for _, r := range rows { + byKey[r.Key] = r + } + + require.Len(t, byKey, 3) + assert.Equal(t, `1`, byKey["a"].UIValue.String) + assert.Equal(t, `2`, byKey["b"].UIValue.String) + assert.Equal(t, `3`, byKey["c"].UIValue.String) +} + +// TestPutSettingsNoopWhenDisabled confirms a batch write against a disabled +// (no state-db configured) database is a silent no-op, not an error — +// callers rely on this to keep working in-memory-only. +func TestPutSettingsNoopWhenDisabled(t *testing.T) { + log := logrus.New() + log.SetOutput(io.Discard) + + d := NewDatabase(&Config{}, log) + require.NoError(t, d.Init()) + require.False(t, d.Enabled()) + + require.NoError(t, d.PutSettings([]SettingRow{{Key: "a"}})) +} + +// TestPutSettingsFailsAfterClose confirms a batch write against a closed +// connection reports an error rather than silently succeeding. +func TestPutSettingsFailsAfterClose(t *testing.T) { + d := testDB(t) + require.NoError(t, d.Close()) + + err := d.PutSettings([]SettingRow{{Key: "a"}}) + assert.Error(t, err) +} diff --git a/pkg/lifecycle/balance.go b/pkg/lifecycle/balance.go index 87dbfc7..715ae8c 100644 --- a/pkg/lifecycle/balance.go +++ b/pkg/lifecycle/balance.go @@ -23,7 +23,7 @@ const topupCooldownEpochs phase0.Epoch = 8 // snapshot for the whole key set, so monitoring hundreds of keys costs one pass // rather than one beacon query per key. type BalanceService struct { - cfg *config.Config + cfgSvc *config.Service // settings source; one snapshot per check chainSvc chain.Service registry *builder_keys.Registry depositSvc *DepositService @@ -32,14 +32,14 @@ type BalanceService struct { // NewBalanceService creates a new balance service. func NewBalanceService( - cfg *config.Config, + cfgSvc *config.Service, chainSvc chain.Service, registry *builder_keys.Registry, depositSvc *DepositService, log logrus.FieldLogger, ) *BalanceService { return &BalanceService{ - cfg: cfg, + cfgSvc: cfgSvc, chainSvc: chainSvc, registry: registry, depositSvc: depositSvc, @@ -75,7 +75,9 @@ func (s *BalanceService) NeedsTopup(key *builder_keys.Key) (bool, uint64, error) return false, 0, nil } - threshold := s.cfg.TopupThreshold + cfg := s.cfgSvc.Current() + + threshold := cfg.TopupThreshold if s.registry.EffectiveBalance(key.KeyIndex()) >= threshold { return false, 0, nil } @@ -88,7 +90,7 @@ func (s *BalanceService) NeedsTopup(key *builder_keys.Key) (bool, uint64, error) } } - topupAmount := s.cfg.TopupAmount + topupAmount := cfg.TopupAmount if topupAmount == 0 { topupAmount = threshold } diff --git a/pkg/lifecycle/deposit.go b/pkg/lifecycle/deposit.go index a2e0705..5775d1c 100644 --- a/pkg/lifecycle/deposit.go +++ b/pkg/lifecycle/deposit.go @@ -56,7 +56,7 @@ const depositConfirmTimeout = 5 * time.Minute // deposit system contract. It is key-agnostic: every operation names the builder // key it acts on, so one service serves the whole managed key set. type DepositService struct { - cfg *config.Config + cfgSvc *config.Service // settings source; one snapshot per read chainSvc chain.Service wallet *wallet.Wallet log logrus.FieldLogger @@ -64,7 +64,7 @@ type DepositService struct { // NewDepositService creates a new deposit service. func NewDepositService( - cfg *config.Config, + cfgSvc *config.Service, chainSvc chain.Service, w *wallet.Wallet, log logrus.FieldLogger, @@ -80,7 +80,7 @@ func NewDepositService( Info("Using builder deposit contract") return &DepositService{ - cfg: cfg, + cfgSvc: cfgSvc, chainSvc: chainSvc, wallet: w, log: depositLog, @@ -272,7 +272,7 @@ func (s *DepositService) resolveDepositFee(ctx context.Context) (*big.Int, error return nil, ErrContractNotActive } - if maxFeeGwei := s.cfg.DepositMaxFeeGwei; maxFeeGwei > 0 { + if maxFeeGwei := s.cfgSvc.Current().DepositMaxFeeGwei; maxFeeGwei > 0 { maxFeeWei := GweiToWei(maxFeeGwei) if fee.Cmp(maxFeeWei) > 0 { s.log.WithFields(logrus.Fields{ diff --git a/pkg/lifecycle/early_deposit.go b/pkg/lifecycle/early_deposit.go index a3faddb..2965115 100644 --- a/pkg/lifecycle/early_deposit.go +++ b/pkg/lifecycle/early_deposit.go @@ -37,7 +37,7 @@ const depositContractABI = `[{"name":"deposit","type":"function","stateMutabilit // validator deposit that sits in the beacon state's pending_deposits queue and is // converted into a builder at the Gloas fork boundary. type EarlyDepositService struct { - cfg *config.Config + cfgSvc *config.Service // settings source; one snapshot per read chainSvc chain.Service wallet *wallet.Wallet depositABI abi.ABI @@ -46,7 +46,7 @@ type EarlyDepositService struct { // NewEarlyDepositService creates a new early deposit service. func NewEarlyDepositService( - cfg *config.Config, + cfgSvc *config.Service, chainSvc chain.Service, w *wallet.Wallet, log logrus.FieldLogger, @@ -57,7 +57,7 @@ func NewEarlyDepositService( } return &EarlyDepositService{ - cfg: cfg, + cfgSvc: cfgSvc, chainSvc: chainSvc, wallet: w, depositABI: depositABI, diff --git a/pkg/lifecycle/early_onboard.go b/pkg/lifecycle/early_onboard.go index 8fbae50..4477371 100644 --- a/pkg/lifecycle/early_onboard.go +++ b/pkg/lifecycle/early_onboard.go @@ -49,11 +49,11 @@ func (m *Manager) earlyOnboard(ctx context.Context) { m.log.WithFields(logrus.Fields{ "gloas_fork_epoch": forkEpoch, - "target_keys": m.cfg.BuilderKeys.EffectiveTargetCount(), + "target_keys": m.cfgSvc.Current().BuilderKeys.EffectiveTargetCount(), }).Info("Gloas scheduled, evaluating early builder onboarding") m.fireEvent("early_onboard", fmt.Sprintf( "Gloas fork at epoch %d, preparing early onboarding of %d builder keys", - forkEpoch, m.cfg.BuilderKeys.EffectiveTargetCount()), "info") + forkEpoch, m.cfgSvc.Current().BuilderKeys.EffectiveTargetCount()), "info") // Subscribe before the first evaluation so an epoch transition can't slip through // between a "wait" decision and the subscription. @@ -82,7 +82,7 @@ func (m *Manager) earlyOnboard(ctx context.Context) { // not yet in the builder registry and not already waiting in the pending-deposit // queue (restart safety — a prior run's deposits must not be submitted twice). func (m *Manager) earlyOnboardTargets() []*builder_keys.Key { - target := m.cfg.BuilderKeys.EffectiveTargetCount() + target := m.cfgSvc.Current().BuilderKeys.EffectiveTargetCount() pending := make([]*builder_keys.Key, 0, target) @@ -159,7 +159,7 @@ func (m *Manager) tryEarlyOnboardOnce(ctx context.Context, forkEpoch phase0.Epoc return true } - amount := m.cfg.DepositAmount + amount := m.cfgSvc.Current().DepositAmount // Two deposit windows (per design): at least earlyOnboardFinalizationMargin epochs // before the fork if the pending-deposit queue is long enough to shield the whole diff --git a/pkg/lifecycle/manager.go b/pkg/lifecycle/manager.go index 8ec3142..203c7f5 100644 --- a/pkg/lifecycle/manager.go +++ b/pkg/lifecycle/manager.go @@ -39,7 +39,7 @@ type LifecycleEvent struct { // Manager orchestrates builder lifecycle operations. type Manager struct { - cfg *config.Config + cfgSvc *config.Service // settings source; one snapshot per pass clClient *beacon.Client chainSvc chain.Service registry *builder_keys.Registry @@ -73,7 +73,7 @@ type Manager struct { // NewManager creates a new lifecycle manager. func NewManager( - cfg *config.Config, + cfgSvc *config.Service, clClient *beacon.Client, chainSvc chain.Service, registry *builder_keys.Registry, @@ -83,7 +83,7 @@ func NewManager( managerLog := log.WithField("component", "lifecycle-manager") m := &Manager{ - cfg: cfg, + cfgSvc: cfgSvc, clClient: clClient, chainSvc: chainSvc, registry: registry, @@ -95,7 +95,7 @@ func NewManager( } // Initialize services - depositSvc, err := NewDepositService(cfg, chainSvc, w, managerLog) + depositSvc, err := NewDepositService(cfgSvc, chainSvc, w, managerLog) if err != nil { return nil, fmt.Errorf("failed to create deposit service: %w", err) } @@ -104,7 +104,7 @@ func NewManager( // Early deposit service (regular validator deposit contract, used to onboard the // builder before the Gloas fork so there is no Builder-API-to-Gloas coverage gap). - earlyDepositSvc, err := NewEarlyDepositService(cfg, chainSvc, w, managerLog) + earlyDepositSvc, err := NewEarlyDepositService(cfgSvc, chainSvc, w, managerLog) if err != nil { return nil, fmt.Errorf("failed to create early deposit service: %w", err) } @@ -222,13 +222,13 @@ func (m *Manager) EnsureBuilderRegistered(ctx context.Context, key *builder_keys } m.log.Info("Builder not registered, creating deposit") - m.fireEvent("deposit", fmt.Sprintf("Builder not registered, submitting deposit (%d gwei)", m.cfg.DepositAmount), "info") + m.fireEvent("deposit", fmt.Sprintf("Builder not registered, submitting deposit (%d gwei)", m.cfgSvc.Current().DepositAmount), "info") if m.depositPendingCallback != nil { m.depositPendingCallback() } - if err := m.depositSvc.CreateDeposit(ctx, key, m.cfg.DepositAmount); err != nil { + if err := m.depositSvc.CreateDeposit(ctx, key, m.cfgSvc.Current().DepositAmount); err != nil { if isDepositDeferred(err) { // Fee too high or contract not active yet — delay, don't treat as failure. m.fireEvent("deposit", fmt.Sprintf("Deposit deferred: %v", err), "info") @@ -270,7 +270,7 @@ func (m *Manager) CheckAndTopup(ctx context.Context, key *builder_keys.Key) erro // configured top-up amount. func (m *Manager) TopupKey(ctx context.Context, key *builder_keys.Key, amountGwei uint64) error { if amountGwei == 0 { - amountGwei = m.cfg.TopupAmount + amountGwei = m.cfgSvc.Current().TopupAmount } if err := m.depositSvc.CreateTopup(ctx, key, amountGwei); err != nil { @@ -382,7 +382,7 @@ func (m *Manager) WaitForRegistration( // stores it for direct access. func (m *Manager) SetPaymentTracker(payments *payload_bidder.PaymentTracker) { m.payments = payments - m.balanceSvc = NewBalanceService(m.cfg, m.chainSvc, m.registry, m.depositSvc, m.log) + m.balanceSvc = NewBalanceService(m.cfgSvc, m.chainSvc, m.registry, m.depositSvc, m.log) } // GetPaymentTracker returns the shared payment tracker. diff --git a/pkg/lifecycle/reconcile.go b/pkg/lifecycle/reconcile.go index a86ca28..7cf75ff 100644 --- a/pkg/lifecycle/reconcile.go +++ b/pkg/lifecycle/reconcile.go @@ -111,14 +111,15 @@ func (m *Manager) reconcileOnce(ctx context.Context) bool { return false } - target := m.cfg.BuilderKeys.EffectiveTargetCount() + cfg := m.cfgSvc.Current() + target := cfg.BuilderKeys.EffectiveTargetCount() managed := m.registry.Aggregate().Managed switch { - case managed < target && m.cfg.BuilderKeys.AutoDeposit: + case managed < target && cfg.BuilderKeys.AutoDeposit: return m.depositKeysToTarget(ctx, target, managed) - case managed > target && m.cfg.BuilderKeys.AutoExit: + case managed > target && cfg.BuilderKeys.AutoExit: return m.exitSurplusKey(ctx, target, managed) } @@ -152,7 +153,7 @@ func (m *Manager) depositKeysToTarget(ctx context.Context, target, managed uint6 return false } - amount := m.cfg.DepositAmount + amount := m.cfgSvc.Current().DepositAmount //nolint:gosec // the batch size is bounded by wallet.MaxBatchSize if !m.walletCanFund(ctx, amount*uint64(len(keys))) { diff --git a/pkg/p2p_bidder/keyset_test.go b/pkg/p2p_bidder/keyset_test.go index 5d6f399..18dab5e 100644 --- a/pkg/p2p_bidder/keyset_test.go +++ b/pkg/p2p_bidder/keyset_test.go @@ -24,7 +24,7 @@ func newTestKeyRegistry(t *testing.T, builderIndices ...uint64) *builder_keys.Re MaxIndex: 32, }} - registry, err := builder_keys.NewRegistry(cfg, testBuilderPrivkey, log) + registry, err := builder_keys.NewRegistry(config.NewStaticService(cfg), testBuilderPrivkey, log) require.NoError(t, err) for keyIndex, builderIndex := range builderIndices { diff --git a/pkg/p2p_bidder/registration_test.go b/pkg/p2p_bidder/registration_test.go index 111f047..a151966 100644 --- a/pkg/p2p_bidder/registration_test.go +++ b/pkg/p2p_bidder/registration_test.go @@ -41,9 +41,9 @@ func TestSetRegistrationPendingKeepsFleetBidding(t *testing.T) { log.SetLevel(logrus.PanicLevel) registry, err := builder_keys.NewRegistry( - &config.Config{BuilderKeys: config.BuilderKeysConfig{ + config.NewStaticService(&config.Config{BuilderKeys: config.BuilderKeysConfig{ TargetCount: 1, DiscoveryGap: 1, MaxIndex: 32, - }}, testBuilderPrivkey, log) + }}), testBuilderPrivkey, log) require.NoError(t, err) require.False(t, registry.AnyActive()) diff --git a/pkg/p2p_bidder/scheduler.go b/pkg/p2p_bidder/scheduler.go index 4369980..f0c387b 100644 --- a/pkg/p2p_bidder/scheduler.go +++ b/pkg/p2p_bidder/scheduler.go @@ -78,7 +78,7 @@ type Scheduler struct { registry *builder_keys.Registry propPrefsStore *memstore.Store[phase0.Slot, *gloasspec.SignedProposerPreferences] planSvc *action_plan.PlanService // per-slot scheduling/settings authority - cfg *config.Config // shared config; mutable settings read live + cfgSvc *config.Service // settings source; one snapshot per selection log logrus.FieldLogger // Simple state tracking per slot @@ -110,7 +110,7 @@ func NewScheduler( registry *builder_keys.Registry, propPrefsStore *memstore.Store[phase0.Slot, *gloasspec.SignedProposerPreferences], planSvc *action_plan.PlanService, - cfg *config.Config, + cfgSvc *config.Service, log logrus.FieldLogger, ) *Scheduler { return &Scheduler{ @@ -122,7 +122,7 @@ func NewScheduler( registry: registry, propPrefsStore: propPrefsStore, planSvc: planSvc, - cfg: cfg, + cfgSvc: cfgSvc, slotStates: make(map[phase0.Slot]*SlotState), log: log.WithField("component", "scheduler"), } @@ -328,11 +328,13 @@ func (s *Scheduler) checkSlotForBidding(ctx context.Context, slot phase0.Slot, n func (s *Scheduler) selectBidPayloads( slot phase0.Slot, bidSettings *action_plan.ResolvedBidSettings, ) []*payload_builder.Payload { - // The frozen per-slot selection wins; the live config covers snapshots - // frozen before the setting existed. + // One config snapshot per selection. The frozen per-slot selection wins; + // the snapshot covers plans frozen before the setting existed. + cfg := s.cfgSvc.Current() + mode := bidSettings.BidCandidate if mode == "" { - mode = s.cfg.EPBS.BidCandidate + mode = cfg.EPBS.BidCandidate } switch { @@ -358,7 +360,7 @@ func (s *Scheduler) selectBidPayloads( chosen, chosenSet := state.BidCandidate, state.BidCandidateSet s.mu.Unlock() - if chosenSet && !s.cfg.EPBS.BidCandidateSwitch { + if chosenSet && !cfg.EPBS.BidCandidateSwitch { // Sticky: keep bidding the committed candidate (the gossip first-seen // rule makes a switched bid unlikely to propagate anyway); fall back // to the primary payload when that candidate produced none. diff --git a/pkg/p2p_bidder/scheduler_test.go b/pkg/p2p_bidder/scheduler_test.go index b62df19..0bef3eb 100644 --- a/pkg/p2p_bidder/scheduler_test.go +++ b/pkg/p2p_bidder/scheduler_test.go @@ -207,7 +207,8 @@ func newSchedulerHarness(t *testing.T, opts harnessOptions) *schedulerHarness { cfg.EPBS.BidSubsidy = 0 cfg.EPBS.BidValueOverride = 0 - planSvc := action_plan.NewPlanService(cfg, chainSvc, log) + cfgSvc := config.NewStaticService(cfg) + planSvc := action_plan.NewPlanService(cfgSvc, chainSvc, log) // Several keys by default: a key is spent once it has bid a slot, so tests // that re-bid or bid several candidates need more than one. @@ -216,7 +217,7 @@ func newSchedulerHarness(t *testing.T, opts harnessOptions) *schedulerHarness { prefs := memstore.New[phase0.Slot, *gloasspec.SignedProposerPreferences]() - svc, err := NewService(nil, chainSvc, registry, prefs, planSvc, log) + svc, err := NewService(cfgSvc, nil, chainSvc, registry, prefs, planSvc, log) require.NoError(t, err) svc.SetEnabled(opts.serviceEnabled) @@ -226,7 +227,7 @@ func newSchedulerHarness(t *testing.T, opts harnessOptions) *schedulerHarness { cache := payload_builder.NewPayloadCache(8) scheduler := NewScheduler(chainSvc, bidCreator, bidTracker, - cache, svc, registry, prefs, planSvc, cfg, log) + cache, svc, registry, prefs, planSvc, cfgSvc, log) events := svc.SubscribeBidSubmissions(16, false) t.Cleanup(events.Unsubscribe) diff --git a/pkg/p2p_bidder/service.go b/pkg/p2p_bidder/service.go index 7c6c94b..e763ed6 100644 --- a/pkg/p2p_bidder/service.go +++ b/pkg/p2p_bidder/service.go @@ -18,6 +18,7 @@ import ( "github.com/ethpandaops/buildoor/pkg/action_plan" "github.com/ethpandaops/buildoor/pkg/builder_keys" "github.com/ethpandaops/buildoor/pkg/chain" + "github.com/ethpandaops/buildoor/pkg/config" "github.com/ethpandaops/buildoor/pkg/memstore" "github.com/ethpandaops/buildoor/pkg/payload_builder" "github.com/ethpandaops/buildoor/pkg/rpc/beacon" @@ -99,6 +100,7 @@ type BidSubmissionEvent struct { // reveals, inclusion tracking, and payment accounting live in the shared // payload_bidder services. type Service struct { + cfgSvc *config.Service // settings source, handed to the scheduler registry *builder_keys.Registry scheduler *Scheduler bidCreator *BidCreator @@ -128,6 +130,7 @@ type Service struct { // how the slot is bid on (a plan may activate bidding for a slot even when // ePBS is globally disabled). func NewService( + cfgSvc *config.Service, clClient *beacon.Client, chainSvc chain.Service, registry *builder_keys.Registry, @@ -138,6 +141,7 @@ func NewService( serviceLog := log.WithField("component", "p2p-bidder") s := &Service{ + cfgSvc: cfgSvc, registry: registry, clClient: clClient, chainSvc: chainSvc, @@ -220,7 +224,7 @@ func (s *Service) Start(ctx context.Context, builderSvc *payload_builder.Service s.registry, s.propPrefsStore, s.planSvc, - builderSvc.GetConfig(), + s.cfgSvc, s.log, ) diff --git a/pkg/payload_bidder/inclusion_tracker_test.go b/pkg/payload_bidder/inclusion_tracker_test.go index af50c6e..7a5f70c 100644 --- a/pkg/payload_bidder/inclusion_tracker_test.go +++ b/pkg/payload_bidder/inclusion_tracker_test.go @@ -253,8 +253,8 @@ func TestInclusionTracker_GloasGatingAndInclusion(t *testing.T) { // Not started: RequestReveal just queues on the buffered channel. cfg := &config.Config{} - revealSvc := NewRevealService(cfg, registry, &mockEnvelopePublisher{}, - chainSvc, builderSvc, payments, action_plan.NewPlanService(cfg, chainSvc, logger), nil, logger) + revealSvc := NewRevealService(config.NewStaticService(cfg), registry, &mockEnvelopePublisher{}, + chainSvc, builderSvc, payments, action_plan.NewPlanService(config.NewStaticService(cfg), chainSvc, logger), nil, logger) tracker := NewInclusionTracker(nil, chainSvc, builderSvc, registry, revealSvc, payments, logger) includedSub := tracker.SubscribeIncluded(4, false) @@ -455,8 +455,8 @@ func TestInclusionTracker_BindsWinToTheBiddingKey(t *testing.T) { registry := newTestKeyRegistry(t, 4, 9) cfg := &config.Config{} - revealSvc := NewRevealService(cfg, registry, &mockEnvelopePublisher{}, - chainSvc, builderSvc, payments, action_plan.NewPlanService(cfg, chainSvc, logger), nil, logger) + revealSvc := NewRevealService(config.NewStaticService(cfg), registry, &mockEnvelopePublisher{}, + chainSvc, builderSvc, payments, action_plan.NewPlanService(config.NewStaticService(cfg), chainSvc, logger), nil, logger) tracker := NewInclusionTracker(nil, chainSvc, builderSvc, registry, revealSvc, payments, logger) blockHash := phase0.Hash32{0xab} @@ -492,8 +492,8 @@ func TestInclusionTracker_RejectsForeignBuilderIndex(t *testing.T) { registry := newTestKeyRegistry(t, 4, 9) cfg := &config.Config{} - revealSvc := NewRevealService(cfg, registry, &mockEnvelopePublisher{}, - chainSvc, builderSvc, payments, action_plan.NewPlanService(cfg, chainSvc, logger), nil, logger) + revealSvc := NewRevealService(config.NewStaticService(cfg), registry, &mockEnvelopePublisher{}, + chainSvc, builderSvc, payments, action_plan.NewPlanService(config.NewStaticService(cfg), chainSvc, logger), nil, logger) tracker := NewInclusionTracker(nil, chainSvc, builderSvc, registry, revealSvc, payments, logger) blockHash := phase0.Hash32{0xcd} diff --git a/pkg/payload_bidder/keyset_test.go b/pkg/payload_bidder/keyset_test.go index 34420bc..f7f3db5 100644 --- a/pkg/payload_bidder/keyset_test.go +++ b/pkg/payload_bidder/keyset_test.go @@ -27,7 +27,7 @@ func newTestKeyRegistry(t *testing.T, builderIndices ...uint64) *builder_keys.Re MaxIndex: 32, }} - registry, err := builder_keys.NewRegistry(cfg, testEntryPrivkey, log) + registry, err := builder_keys.NewRegistry(config.NewStaticService(cfg), testEntryPrivkey, log) require.NoError(t, err) for keyIndex, builderIndex := range builderIndices { diff --git a/pkg/payload_bidder/mockchain_test.go b/pkg/payload_bidder/mockchain_test.go index 99ded49..472bf38 100644 --- a/pkg/payload_bidder/mockchain_test.go +++ b/pkg/payload_bidder/mockchain_test.go @@ -107,7 +107,7 @@ func newTestBuilderSvc(chainSvc chain.Service) *payload_builder.Service { log := logrus.New() log.SetLevel(logrus.PanicLevel) - svc, err := payload_builder.NewService(&config.Config{}, nil, chainSvc, nil, nil, common.Address{}, log) + svc, err := payload_builder.NewService(config.NewStaticService(&config.Config{}), nil, chainSvc, nil, nil, common.Address{}, log) if err != nil { panic(err) } diff --git a/pkg/payload_bidder/payment_tracker.go b/pkg/payload_bidder/payment_tracker.go index 256828b..6a872a4 100644 --- a/pkg/payload_bidder/payment_tracker.go +++ b/pkg/payload_bidder/payment_tracker.go @@ -34,6 +34,18 @@ type keyPayments struct { // pending holds unrevealed won bids, kept for 2 epochs. Only these count as // "pending" in the UI and for topup checks. pending map[phase0.Slot]*PendingPayment + + // earlyReveals records slots whose MarkRevealed call arrived before + // RecordWonBid created the pending entry: the two are fed by independent + // goroutines (RevealService's gate/timer vs. InclusionTracker's + // head-event loop, and the Builder API requests reveals already at block + // submission), so either order is possible. Without this, the inverted + // order silently drops the balance deduction (MarkRevealed finds no + // pending entry and no-ops) and the later RecordWonBid creates a pending + // entry that is never marked revealed, orphaned until it expires two + // epochs later. Values are the slot's epoch, pruned on the same schedule + // as pending payments. + earlyReveals map[phase0.Slot]phase0.Epoch } // PaymentTracker tracks payment obligations and live balance adjustments per @@ -67,7 +79,10 @@ func NewPaymentTracker(chainSvc chain.Service, log logrus.FieldLogger) *PaymentT func (t *PaymentTracker) forKey(keyIndex uint64) *keyPayments { entry, ok := t.keys[keyIndex] if !ok { - entry = &keyPayments{pending: make(map[phase0.Slot]*PendingPayment, 8)} + entry = &keyPayments{ + pending: make(map[phase0.Slot]*PendingPayment, 8), + earlyReveals: make(map[phase0.Slot]phase0.Epoch, 2), + } t.keys[keyIndex] = entry } @@ -75,15 +90,38 @@ func (t *PaymentTracker) forKey(keyIndex uint64) *keyPayments { } // RecordWonBid records a won bid as a pending payment (unrevealed) against the -// key whose bid was included. If we later reveal, MarkRevealed moves it from -// pending to a balance deduction; otherwise it stays pending for 2 epochs and -// then expires. +// key whose bid was included — unless MarkRevealed already ran for this slot +// (see keyPayments.earlyReveals), in which case the deduction that was +// deferred for lack of a known value applies immediately and no pending +// entry is created: a bid that was already revealed is never "pending". +// If we later reveal (the common order), MarkRevealed moves it from pending +// to a balance deduction; otherwise it stays pending for 2 epochs and then +// expires. func (t *PaymentTracker) RecordWonBid(keyIndex uint64, slot phase0.Slot, value uint64) { epoch := t.chainSvc.GetEpochOfSlot(slot) t.mu.Lock() - t.forKey(keyIndex).pending[slot] = &PendingPayment{ + entry := t.forKey(keyIndex) + + if _, revealedEarly := entry.earlyReveals[slot]; revealedEarly { + delete(entry.earlyReveals, slot) + + entry.balanceAdjustment -= int64(value) + anchorEpoch(entry, epoch) + + t.mu.Unlock() + + t.log.WithFields(logrus.Fields{ + "key_index": keyIndex, + "slot": slot, + "value": value, + }).Info("Won bid was already revealed before it was recorded: deducted from live balance") + + return + } + + entry.pending[slot] = &PendingPayment{ KeyIndex: keyIndex, Slot: slot, Epoch: epoch, @@ -101,7 +139,10 @@ func (t *PaymentTracker) RecordWonBid(keyIndex uint64, slot phase0.Slot, value u } // MarkRevealed moves a won bid from pending to an immediate balance deduction on -// the key that owes it. +// the key that owes it. If RecordWonBid hasn't run yet for this slot (no +// pending entry exists), the value isn't known here, so the deduction can't +// apply yet: the slot is recorded in earlyReveals instead and RecordWonBid +// applies the deduction as soon as it runs. func (t *PaymentTracker) MarkRevealed(keyIndex uint64, slot phase0.Slot) { slotEpoch := t.chainSvc.GetEpochOfSlot(slot) @@ -111,7 +152,14 @@ func (t *PaymentTracker) MarkRevealed(keyIndex uint64, slot phase0.Slot) { payment, ok := entry.pending[slot] if !ok { + entry.earlyReveals[slot] = slotEpoch t.mu.Unlock() + + t.log.WithFields(logrus.Fields{ + "key_index": keyIndex, + "slot": slot, + }).Warn("Reveal completed before the won bid was recorded — deferring the balance deduction") + return } @@ -293,5 +341,23 @@ func (t *PaymentTracker) PruneExpiredPayments(currentEpoch phase0.Epoch) { delete(entry.pending, slot) } + + // Early-reveal markers of the same age whose matching RecordWonBid + // never arrived (the block was never seen included, or the report + // was lost) would otherwise sit here forever. + for slot, epoch := range entry.earlyReveals { + if currentEpoch <= epoch+1 { + continue + } + + t.log.WithFields(logrus.Fields{ + "key_index": keyIndex, + "slot": slot, + "reveal_epoch": epoch, + "current_epoch": currentEpoch, + }).Debug("Pruning stale early-reveal marker (matching won bid never recorded)") + + delete(entry.earlyReveals, slot) + } } } diff --git a/pkg/payload_bidder/payment_tracker_test.go b/pkg/payload_bidder/payment_tracker_test.go index 86286ca..7f1d057 100644 --- a/pkg/payload_bidder/payment_tracker_test.go +++ b/pkg/payload_bidder/payment_tracker_test.go @@ -138,3 +138,70 @@ func TestPaymentTracker_PerKeyAccounting(t *testing.T) { assert.Equal(t, uint64(0), tracker.GetPendingPayments(5)) assert.Equal(t, uint64(800), tracker.GetPendingPayments(6)) } + +// TestPaymentTracker_MarkRevealedBeforeRecordWonBid: MarkRevealed and +// RecordWonBid are fed by independent goroutines with no happens-before edge +// (the Builder API even requests reveals at block submission, before the +// head event), so either order is possible. A MarkRevealed arriving first +// must not drop the deduction: it defers it, and the later RecordWonBid +// applies it immediately without ever treating the bid as pending. +func TestPaymentTracker_MarkRevealedBeforeRecordWonBid(t *testing.T) { + tracker := newTestPaymentTracker() + + const keyIndex = uint64(3) + + const slot = phase0.Slot(200) + + const value = uint64(5000) + + // The reveal completes before the win is even recorded. + tracker.MarkRevealed(keyIndex, slot) + assert.Equal(t, int64(0), tracker.GetBalanceAdjustment(keyIndex), + "the value isn't known yet, so no deduction can apply until RecordWonBid runs") + assert.Equal(t, uint64(0), tracker.GetTotalPendingPayments()) + + // The delayed head event now lands and the win is recorded. + tracker.RecordWonBid(keyIndex, slot, value) + + assert.Equal(t, -int64(value), tracker.GetBalanceAdjustment(keyIndex), + "the deduction must still apply once the won bid is recorded") + assert.Equal(t, uint64(0), tracker.GetTotalPendingPayments(), + "a bid revealed before it was recorded must never sit in pending") +} + +// TestPaymentTracker_EarlyRevealIsPerKey confirms an early reveal recorded +// against one key never satisfies another key's won bid for the same slot. +func TestPaymentTracker_EarlyRevealIsPerKey(t *testing.T) { + tracker := newTestPaymentTracker() + + tracker.MarkRevealed(1, 300) + tracker.RecordWonBid(2, 300, 700) + + assert.Equal(t, int64(0), tracker.GetBalanceAdjustment(2), + "key 2's bid was not revealed; it must stay pending") + assert.Equal(t, uint64(700), tracker.GetTotalPendingPayments()) + assert.Equal(t, int64(0), tracker.GetBalanceAdjustment(1)) +} + +// TestPaymentTracker_EarlyRevealNeverRecordedIsPruned confirms an early +// reveal whose matching won bid never arrives is cleared by the same +// two-epoch prune pass as expired pending payments, instead of deducting +// from a balance forever later. +func TestPaymentTracker_EarlyRevealNeverRecordedIsPruned(t *testing.T) { + tracker := newTestPaymentTracker() + + const keyIndex = uint64(0) + + const slot = phase0.Slot(64) // epoch 2 on the stub chain + + tracker.MarkRevealed(keyIndex, slot) + + epoch := tracker.chainSvc.GetEpochOfSlot(slot) + tracker.PruneExpiredPayments(epoch + 2) + + // A won-bid report arriving after the prune window must be treated as a + // fresh pending payment, not matched to the long-gone reveal. + tracker.RecordWonBid(keyIndex, slot, 900) + assert.Equal(t, int64(0), tracker.GetBalanceAdjustment(keyIndex)) + assert.Equal(t, uint64(900), tracker.GetTotalPendingPayments()) +} diff --git a/pkg/payload_bidder/reveal_service.go b/pkg/payload_bidder/reveal_service.go index 6fdc98b..07eaf57 100644 --- a/pkg/payload_bidder/reveal_service.go +++ b/pkg/payload_bidder/reveal_service.go @@ -107,7 +107,7 @@ const ( // broadcast validation, deadline bypass) — the plan service is the single // per-slot settings authority. type RevealService struct { - cfg *config.Config // shared live config (reveal settings resolve via planSvc.Freeze) + cfgSvc *config.Service // settings source (reveal settings resolve via planSvc.Freeze) registry *builder_keys.Registry publisher envelopePublisher chainSvc chain.Service @@ -174,7 +174,7 @@ func (st *revealState) gateSatisfied(now time.Time) bool { // programming error. votes is the head-vote tracker backing the vote gates; // it may be nil (vote gates then never open and expire at the slot end). func NewRevealService( - cfg *config.Config, + cfgSvc *config.Service, registry *builder_keys.Registry, publisher envelopePublisher, chainSvc chain.Service, @@ -185,7 +185,7 @@ func NewRevealService( log logrus.FieldLogger, ) *RevealService { return &RevealService{ - cfg: cfg, + cfgSvc: cfgSvc, registry: registry, publisher: publisher, chainSvc: chainSvc, @@ -477,7 +477,7 @@ func (s *RevealService) schedule(req *RevealRequest) { // a reorg). The rebuilt envelope is re-signed for the new root — the envelope // signature covers the beacon block root, so the old one cannot be reused. func (s *RevealService) shouldRebind(slot phase0.Slot, existing *revealState, req *RevealRequest) bool { - if !s.cfg.Reveal.RebindOnReorg { + if !s.cfgSvc.Current().Reveal.RebindOnReorg { return false } diff --git a/pkg/payload_bidder/reveal_service_test.go b/pkg/payload_bidder/reveal_service_test.go index 4b5c167..03f3209 100644 --- a/pkg/payload_bidder/reveal_service_test.go +++ b/pkg/payload_bidder/reveal_service_test.go @@ -154,10 +154,11 @@ func newRevealTestEnv(t *testing.T, slotDuration time.Duration, revealTimeMs int builderSvc := newTestBuilderSvc(chainSvc) payments := NewPaymentTracker(chainSvc, log) publisher := &mockEnvelopePublisher{} - planSvc := action_plan.NewPlanService(cfg, chainSvc, log) + cfgSvc := config.NewStaticService(cfg) + planSvc := action_plan.NewPlanService(cfgSvc, chainSvc, log) votes := newStubVoteSource() - svc := NewRevealService(cfg, registry, publisher, chainSvc, builderSvc, + svc := NewRevealService(cfgSvc, registry, publisher, chainSvc, builderSvc, payments, planSvc, votes, log) return &revealTestEnv{ diff --git a/pkg/payload_builder/attributes_fallback_test.go b/pkg/payload_builder/attributes_fallback_test.go index 5682292..a90dce0 100644 --- a/pkg/payload_builder/attributes_fallback_test.go +++ b/pkg/payload_builder/attributes_fallback_test.go @@ -35,9 +35,10 @@ func TestApplyAttributesFallback(t *testing.T) { clClient, err := beacon.NewClient(context.Background(), "http://127.0.0.1:1", log) require.NoError(t, err) - planSvc := action_plan.NewPlanService(cfg, chainSvc, log) + cfgSvc := config.NewStaticService(cfg) + planSvc := action_plan.NewPlanService(cfgSvc, chainSvc, log) - svc, err := NewService(cfg, clClient, chainSvc, planSvc, nil, common.Address{}, log) + svc, err := NewService(cfgSvc, clClient, chainSvc, planSvc, nil, common.Address{}, log) require.NoError(t, err) svc.ctx = context.Background() @@ -107,9 +108,10 @@ func TestApplyAttributesFallbackMultiSlotGap(t *testing.T) { clClient, err := beacon.NewClient(context.Background(), "http://127.0.0.1:1", log) require.NoError(t, err) - planSvc := action_plan.NewPlanService(cfg, chainSvc, log) + cfgSvc := config.NewStaticService(cfg) + planSvc := action_plan.NewPlanService(cfgSvc, chainSvc, log) - svc, err := NewService(cfg, clClient, chainSvc, planSvc, nil, common.Address{}, log) + svc, err := NewService(cfgSvc, clClient, chainSvc, planSvc, nil, common.Address{}, log) require.NoError(t, err) svc.ctx = context.Background() diff --git a/pkg/payload_builder/attributes_test.go b/pkg/payload_builder/attributes_test.go index 7ac0e04..907db0e 100644 --- a/pkg/payload_builder/attributes_test.go +++ b/pkg/payload_builder/attributes_test.go @@ -57,9 +57,10 @@ func sanitizeTestSetup(t *testing.T, blocks ...*beacon.BlockInfo) *Service { chainSvc.headTracker = newPrimedHeadTracker(spec, blocks...) cfg := config.DefaultConfig() - planSvc := action_plan.NewPlanService(cfg, chainSvc, log) + cfgSvc := config.NewStaticService(cfg) + planSvc := action_plan.NewPlanService(cfgSvc, chainSvc, log) - svc, err := NewService(cfg, nil, chainSvc, planSvc, nil, common.Address{}, log) + svc, err := NewService(cfgSvc, nil, chainSvc, planSvc, nil, common.Address{}, log) require.NoError(t, err) svc.ctx = context.Background() @@ -161,12 +162,13 @@ func TestHandlePayloadAttributes_RescheduleOnParentChange(t *testing.T) { cfg := config.DefaultConfig() cfg.EPBSEnabled = true - planSvc := action_plan.NewPlanService(cfg, chainSvc, log) + cfgSvc := config.NewStaticService(cfg) + planSvc := action_plan.NewPlanService(cfgSvc, chainSvc, log) clClient, err := beacon.NewClient(context.Background(), "http://127.0.0.1:1", log) require.NoError(t, err) - svc, err := NewService(cfg, clClient, chainSvc, planSvc, nil, common.Address{}, log) + svc, err := NewService(cfgSvc, clClient, chainSvc, planSvc, nil, common.Address{}, log) require.NoError(t, err) svc.ctx = context.Background() diff --git a/pkg/payload_builder/build_skip_test.go b/pkg/payload_builder/build_skip_test.go index c02b53d..b3b10ec 100644 --- a/pkg/payload_builder/build_skip_test.go +++ b/pkg/payload_builder/build_skip_test.go @@ -55,9 +55,10 @@ func newSkipTestService(t *testing.T, cfg *config.Config) (*Service, *action_pla log := logrus.New() log.SetLevel(logrus.ErrorLevel) - planSvc := action_plan.NewPlanService(cfg, chainSvc, log) + cfgSvc := config.NewStaticService(cfg) + planSvc := action_plan.NewPlanService(cfgSvc, chainSvc, log) - svc, err := NewService(cfg, nil, chainSvc, planSvc, nil, common.Address{}, log) + svc, err := NewService(cfgSvc, nil, chainSvc, planSvc, nil, common.Address{}, log) require.NoError(t, err) return svc, planSvc diff --git a/pkg/payload_builder/candidate_retry_test.go b/pkg/payload_builder/candidate_retry_test.go new file mode 100644 index 0000000..a9ef61d --- /dev/null +++ b/pkg/payload_builder/candidate_retry_test.go @@ -0,0 +1,94 @@ +package payload_builder + +// slotBuildState.started (the per-candidate build dedup map) is set before a +// build attempt; a failed attempt must clear it again, or a single transient +// engine error permanently blocks any retry of that (slot, parent-tuple) +// candidate for the rest of the slot — including a legitimate CL-client +// attributes redelivery for the exact same parent. + +import ( + "context" + "testing" + "time" + + "github.com/ethereum/go-ethereum/common" + "github.com/ethpandaops/go-eth2-client/spec/phase0" + "github.com/ethpandaops/go-eth2-client/spec/version" + "github.com/sirupsen/logrus" + "github.com/stretchr/testify/require" + + "github.com/ethpandaops/buildoor/pkg/action_plan" + "github.com/ethpandaops/buildoor/pkg/chain" + "github.com/ethpandaops/buildoor/pkg/config" + "github.com/ethpandaops/buildoor/pkg/rpc/beacon" +) + +// unknownForkChainService forces every build attempt to fail deterministically +// and immediately: chain.EngineVersion(DataVersionUnknown) errors before any +// engine or beacon-API call is made, so engineClient/clClient can stay nil. +type unknownForkChainService struct { + stubChainService +} + +func (s *unknownForkChainService) ActiveForkAtEpoch(phase0.Epoch) version.DataVersion { + return version.DataVersionUnknown +} + +func TestExecuteCandidateBuild_RetriesAfterAFailedAttempt(t *testing.T) { + chainSvc := &unknownForkChainService{stubChainService{spec: &chain.ChainSpec{ + SecondsPerSlot: 12 * time.Second, + SlotsPerEpoch: 32, + }}} + + log := logrus.New() + log.SetLevel(logrus.PanicLevel) + + cfg := config.DefaultConfig() + cfgSvc := config.NewStaticService(cfg) + planSvc := action_plan.NewPlanService(cfgSvc, chainSvc, log) + + svc, err := NewService(cfgSvc, nil, chainSvc, planSvc, nil, common.Address{}, log) + require.NoError(t, err) + + // Set up the state Start() would normally set up, without calling Start() + // itself (which needs a live beacon client for its event stream). + svc.ctx = context.Background() + svc.payloadBuilder = NewPayloadBuilder(nil, nil, chainSvc, common.Address{}, cfgSvc, log, nil) + + sub := svc.SubscribePayloadBuildFailed(4, false) + defer sub.Unsubscribe() + + slot := phase0.Slot(500) + attrs := &beacon.PayloadAttributesEvent{ + ProposalSlot: slot, + ParentBlockRoot: phase0.Root{0x11}, + ParentBlockHash: phase0.Hash32{0x22}, + } + target := &buildTarget{candidate: chain.CandidateParentFull, attrs: attrs} + + // First attempt fails (the unknown-fork stand-in for a transient engine + // error) and marks the tuple started. + svc.executeCandidateBuild(slot, target) + + select { + case event := <-sub.Channel(): + require.Equal(t, slot, event.Slot) + case <-time.After(time.Second): + t.Fatal("expected the first build attempt to fail and fire buildFailedDispatcher") + } + + // A fresh payload_attributes redelivery for the EXACT SAME parent tuple + // arrives — real CL client behavior (reorgs, retries, some clients + // simply re-emit payload_attributes). It must actually retry, not be + // silently dropped because the tuple is still marked "started" from the + // failed attempt. + svc.executeCandidateBuild(slot, target) + + select { + case event := <-sub.Channel(): + require.Equal(t, slot, event.Slot) + case <-time.After(time.Second): + t.Fatal("second attempt for the same parent tuple never ran " + + "— the started marker was not cleared after the first failure") + } +} diff --git a/pkg/payload_builder/candidates.go b/pkg/payload_builder/candidates.go index f3142ae..017a641 100644 --- a/pkg/payload_builder/candidates.go +++ b/pkg/payload_builder/candidates.go @@ -147,7 +147,7 @@ func (s *Service) candidateMode(slot phase0.Slot, key chain.CandidateKey) string } } - return s.cfg.Build.CandidateMode(string(key)) + return s.cfgSvc.Current().Build.CandidateMode(string(key)) } // resolveChainCandidates fetches the chain view's build-parent candidates for @@ -212,7 +212,7 @@ func (s *Service) candidateAutoSignal(candidate *chain.CandidateParent) bool { // participation is below the configured weak-head threshold — the signal that // the next proposer may reorg it out. func (s *Service) headContested() bool { - threshold := s.cfg.Build.AutoWeakHeadPct + threshold := s.cfgSvc.Current().Build.AutoWeakHeadPct if threshold == 0 { return false } @@ -429,16 +429,18 @@ func (s *Service) resolveOnDemandTarget(slot phase0.Slot, tuple beacon.AttrParen // the full build time; only serialized builds shorten the speculative ones to // fit them alongside the canonical build. func (s *Service) candidateBuildTime(target *buildTarget) uint64 { - if s.cfg.Build.Parallel { - return s.cfg.PayloadBuildTime + cfg := s.cfgSvc.Current() + + if cfg.Build.Parallel { + return cfg.PayloadBuildTime } if target.candidate != chain.CandidateParentFull && target.candidate != "" && - s.cfg.Build.SpeculativeBuildTimeMs != 0 { - return s.cfg.Build.SpeculativeBuildTimeMs + cfg.Build.SpeculativeBuildTimeMs != 0 { + return cfg.Build.SpeculativeBuildTimeMs } - return s.cfg.PayloadBuildTime + return cfg.PayloadBuildTime } // slotEndTime returns the wall-clock end of a slot (the bound for late diff --git a/pkg/payload_builder/candidates_test.go b/pkg/payload_builder/candidates_test.go index e00a109..c8a1729 100644 --- a/pkg/payload_builder/candidates_test.go +++ b/pkg/payload_builder/candidates_test.go @@ -40,12 +40,13 @@ func candidateTestSetup(t *testing.T, gp, parent *beacon.BlockInfo) *Service { cfg := config.DefaultConfig() cfg.EPBSEnabled = true - planSvc := action_plan.NewPlanService(cfg, chainSvc, log) + cfgSvc := config.NewStaticService(cfg) + planSvc := action_plan.NewPlanService(cfgSvc, chainSvc, log) clClient, err := beacon.NewClient(context.Background(), "http://127.0.0.1:1", log) require.NoError(t, err) - svc, err := NewService(cfg, clClient, chainSvc, planSvc, nil, common.Address{}, log) + svc, err := NewService(cfgSvc, clClient, chainSvc, planSvc, nil, common.Address{}, log) require.NoError(t, err) svc.ctx = context.Background() @@ -97,7 +98,7 @@ func TestResolveBuildTargets_PayloadMissAddsEmptyCandidate(t *testing.T) { func TestResolveBuildTargets_NeverModeSuppresses(t *testing.T) { gp, parent := testCandidateChain() svc := candidateTestSetup(t, gp, parent) - svc.cfg.Build.CandidateParentEmpty = config.CandidateModeNever + svc.cfgSvc.Current().Build.CandidateParentEmpty = config.CandidateModeNever events := svc.clClient.Events() require.True(t, events.InjectPayloadAttributes(&beacon.PayloadAttributesEvent{ diff --git a/pkg/payload_builder/payload_builder.go b/pkg/payload_builder/payload_builder.go index c28ac85..32f0bbd 100644 --- a/pkg/payload_builder/payload_builder.go +++ b/pkg/payload_builder/payload_builder.go @@ -27,7 +27,7 @@ type PayloadBuilder struct { feeRecipient common.Address settingsResolvers []ProposerSettingsResolver // asked in order for proposer settings; first match wins - cfg *config.Config // shared config; mutable settings are read live, never cached + cfgSvc *config.Service // settings source; one snapshot per build log logrus.FieldLogger // Active build tracking: multiple candidate builds may run for the same @@ -55,7 +55,8 @@ type activeBuild struct { } // NewPayloadBuilder creates a new payload builder. -// cfg is the shared config pointer; mutable settings (e.g. PayloadBuildTime) are read live from it. +// cfgSvc is the settings source; each build loads one immutable config +// snapshot at its start, so a build sees a single settings generation. // settingsResolvers are asked in order for the proposer's announced fee recipient and gas // limit; the first match wins. func NewPayloadBuilder( @@ -63,7 +64,7 @@ func NewPayloadBuilder( engineClient EngineClient, chainSvc chain.Service, feeRecipient common.Address, - cfg *config.Config, + cfgSvc *config.Service, log logrus.FieldLogger, settingsResolvers []ProposerSettingsResolver, ) *PayloadBuilder { @@ -73,7 +74,7 @@ func NewPayloadBuilder( engineClient: engineClient, feeRecipient: feeRecipient, settingsResolvers: settingsResolvers, - cfg: cfg, + cfgSvc: cfgSvc, activeBuilds: make(map[activeBuildKey]*activeBuild, 4), log: log.WithField("component", "payload-builder"), } @@ -96,6 +97,10 @@ func (b *PayloadBuilder) BuildPayloadFromAttributes( attrs *beacon.PayloadAttributesEvent, buildTimeMs uint64, ) (*Payload, error) { + // One config snapshot for the whole build: every setting read below sees + // the same settings generation. + cfg := b.cfgSvc.Current() + buildKey := activeBuildKey{ slot: attrs.ProposalSlot, parentRoot: attrs.ParentBlockRoot, @@ -270,9 +275,9 @@ func (b *PayloadBuilder) BuildPayloadFromAttributes( "payload_id": fmt.Sprintf("%x", payloadID[:]), }).Debug("Payload build requested from attributes") - // Read the build time live from config so UI overrides take effect - // immediately; an explicit per-build time (speculative candidates) wins. - payloadBuildTime := b.cfg.PayloadBuildTime + // An explicit per-build time (speculative candidates) wins over the + // configured build time. + payloadBuildTime := cfg.PayloadBuildTime if buildTimeMs != 0 { payloadBuildTime = buildTimeMs } @@ -302,7 +307,7 @@ func (b *PayloadBuilder) BuildPayloadFromAttributes( return nil, fmt.Errorf("getPayload returned no execution payload") } - gasLimitOverride := b.resolveGasLimitOverride(buildCtx, attrs, beaconFork, + gasLimitOverride := b.resolveGasLimitOverride(buildCtx, cfg, attrs, beaconFork, targetGasLimit, enginePayload.GasLimit, enginePayload.GasUsed) // Inject our extra-data marker (and the gas limit override, if any) and @@ -310,7 +315,7 @@ func (b *PayloadBuilder) BuildPayloadFromAttributes( newHash, err := ModifyPayloadExtraData( enginePayload, resp.ExecutionRequests, - []byte(b.cfg.ExtraData), + []byte(cfg.ExtraData), common.Hash(attrs.ParentBeaconBlockRoot), gasLimitOverride, ) @@ -366,11 +371,12 @@ func (b *PayloadBuilder) BuildPayloadFromAttributes( // or the payload's gas usage exceeds the required limit. func (b *PayloadBuilder) resolveGasLimitOverride( ctx context.Context, + cfg *config.Config, attrs *beacon.PayloadAttributesEvent, beaconFork version.DataVersion, targetGasLimit, payloadGasLimit, payloadGasUsed uint64, ) uint64 { - if !b.cfg.Build.EnforceBidGasLimit || beaconFork < version.DataVersionGloas || targetGasLimit == 0 { + if !cfg.Build.EnforceBidGasLimit || beaconFork < version.DataVersionGloas || targetGasLimit == 0 { return 0 } diff --git a/pkg/payload_builder/payload_builder_test.go b/pkg/payload_builder/payload_builder_test.go index 320a097..b8316c5 100644 --- a/pkg/payload_builder/payload_builder_test.go +++ b/pkg/payload_builder/payload_builder_test.go @@ -16,7 +16,7 @@ func TestNewPayloadBuilder(t *testing.T) { nil, nil, common.HexToAddress("0x1111"), - &config.Config{PayloadBuildTime: 100}, + config.NewStaticService(&config.Config{PayloadBuildTime: 100}), logrus.New(), nil, ) @@ -26,5 +26,5 @@ func TestNewPayloadBuilder(t *testing.T) { func TestNewPayloadBuilder_AcceptsNilClients(t *testing.T) { // Constructor allows nil clients (used in tests); actual build will fail if they're nil. - _ = NewPayloadBuilder(nil, nil, nil, common.Address{}, &config.Config{}, logrus.New(), nil) + _ = NewPayloadBuilder(nil, nil, nil, common.Address{}, config.NewStaticService(&config.Config{}), logrus.New(), nil) } diff --git a/pkg/payload_builder/service.go b/pkg/payload_builder/service.go index 51b8348..409432b 100644 --- a/pkg/payload_builder/service.go +++ b/pkg/payload_builder/service.go @@ -38,7 +38,7 @@ const transformTimeout = 2 * time.Second // Building is triggered by payload_attributes events from the beacon node, // which contain all the information needed to build a payload. type Service struct { - cfg *config.Config + cfgSvc *config.Service // settings source; load one snapshot per operation clClient *beacon.Client chainSvc chain.Service planSvc *action_plan.PlanService // per-slot scheduling authority @@ -102,7 +102,7 @@ type ELClientVersion struct { // planSvc is the per-slot scheduling authority: every build decision polls it // for the slot's frozen plan (schedule, force/suppress, build timing). func NewService( - cfg *config.Config, + cfgSvc *config.Service, clClient *beacon.Client, chainSvc chain.Service, planSvc *action_plan.PlanService, @@ -113,7 +113,7 @@ func NewService( serviceLog := log.WithField("component", "builder-service") s := &Service{ - cfg: cfg, + cfgSvc: cfgSvc, clClient: clClient, chainSvc: chainSvc, planSvc: planSvc, @@ -145,7 +145,7 @@ func (s *Service) Start(ctx context.Context) error { s.engineClient, s.chainSvc, s.feeRecipient, - s.cfg, + s.cfgSvc, s.log, s.settingsResolvers, ) @@ -261,19 +261,10 @@ func (s *Service) GetStats() BuilderStats { return *s.stats } -// GetConfig returns the current configuration. +// GetConfig returns the latest effective config snapshot (immutable; one +// settings generation per call). func (s *Service) GetConfig() *config.Config { - return s.cfg -} - -// UpdateConfig updates the service configuration at runtime. Schedule -// changes are handled by the plan service (the scheduling authority). -func (s *Service) UpdateConfig(cfg *config.Config) error { - s.cfg = cfg - - s.log.Info("Configuration updated") - - return nil + return s.cfgSvc.Current() } // GetCurrentSlot returns the most recently built slot. @@ -480,6 +471,20 @@ func newSlotBuildState() *slotBuildState { return &slotBuildState{started: make(map[beacon.AttrParentKey]bool, 4)} } +// clearBuildStarted un-marks a (slot, parent-tuple) candidate as started +// after a failed build attempt, so a later trigger for the exact same tuple +// (a fresh payload_attributes redelivery, a late-build check, ...) is free +// to retry it instead of being silently dropped by the started check in +// executeCandidateBuild for the rest of the slot. +func (s *Service) clearBuildStarted(slot phase0.Slot, tuple beacon.AttrParentKey) { + s.scheduledBuildMu.Lock() + defer s.scheduledBuildMu.Unlock() + + if state := s.slotBuilds[slot]; state != nil { + delete(state.started, tuple) + } +} + // maybeLateBuild activates a candidate build for an attributes variant that // arrived after the slot's build pass already ran: the chain moved (reorg, // payload-miss flip, late reveal) and the new parent still deserves a payload @@ -586,7 +591,7 @@ func (s *Service) scheduleAttributesFallback(targetSlot phase0.Slot) { // attributes would have arrived long before it, and synthesizing then // leaves the normal scheduling path an immediate build. fireAt := s.chainSvc.SlotToTime(targetSlot). - Add(time.Duration(s.cfg.EPBS.BuildStartTime) * time.Millisecond) + Add(time.Duration(s.cfgSvc.Current().EPBS.BuildStartTime) * time.Millisecond) time.AfterFunc(max(time.Until(fireAt), 0), func() { s.applyAttributesFallback(targetSlot) @@ -736,7 +741,7 @@ func (s *Service) executeBuildForSlot(slot phase0.Slot) { return } - if s.cfg.Build.Parallel { + if s.cfgSvc.Current().Build.Parallel { var wg sync.WaitGroup for _, target := range targets { @@ -826,6 +831,11 @@ func (s *Service) executeCandidateBuild(slot phase0.Slot, target *buildTarget) { FailedAt: time.Now(), }) + // A transient failure must not permanently forfeit this parent tuple + // for the rest of the slot: clear the started marker so a later + // trigger for the same tuple can retry. + s.clearBuildStarted(slot, tuple) + return } @@ -849,6 +859,11 @@ func (s *Service) executeCandidateBuild(slot phase0.Slot, target *buildTarget) { FailedAt: time.Now(), }) + // A transient failure must not permanently forfeit this parent tuple + // for the rest of the slot: clear the started marker so a later + // trigger for the same tuple can retry. + s.clearBuildStarted(slot, tuple) + return } diff --git a/pkg/slot_results/tracker.go b/pkg/slot_results/tracker.go index 7181f32..2d99797 100644 --- a/pkg/slot_results/tracker.go +++ b/pkg/slot_results/tracker.go @@ -42,7 +42,7 @@ const ( // request-scoped Builder API handlers — and prunes both stores to their // retention windows on epoch transitions. type Tracker struct { - cfg *config.Config + cfgSvc *config.Service // settings source; one snapshot per event/prune chainSvc chain.Service stateDB *db.Database planSvc *action_plan.PlanService @@ -76,7 +76,7 @@ type Tracker struct { // NewTracker creates the slot results tracker. stateDB must be non-nil (a // disabled database is fine); epbsSvc and revealSvc may be nil when the Gloas // fork is not scheduled. -func NewTracker(cfg *config.Config, chainSvc chain.Service, stateDB *db.Database, +func NewTracker(cfgSvc *config.Service, chainSvc chain.Service, stateDB *db.Database, planSvc *action_plan.PlanService, builderSvc *payload_builder.Service, epbsSvc *p2p_bidder.Service, revealSvc *payload_bidder.RevealService, inclusionTracker *payload_bidder.InclusionTracker, registry *builder_keys.Registry, @@ -84,7 +84,7 @@ func NewTracker(cfg *config.Config, chainSvc chain.Service, stateDB *db.Database trackerLog := log.WithField("component", "slot-results") return &Tracker{ - cfg: cfg, + cfgSvc: cfgSvc, chainSvc: chainSvc, stateDB: stateDB, planSvc: planSvc, @@ -414,7 +414,7 @@ func (t *Tracker) handlePayloadReady(payload *payload_builder.Payload) { outcome.Candidate = string(payload.Candidate) - if t.cfg.SlotArtifactCaptureEnabled && payload.ExecutionPayload != nil { + if t.cfgSvc.Current().SlotArtifactCaptureEnabled && payload.ExecutionPayload != nil { idx, err := t.artifacts.StorePayload(slot, forkVersion, payload.ExecutionPayload, PayloadArtifactMeta{ Candidate: string(payload.Candidate), @@ -641,7 +641,7 @@ func (t *Tracker) handleBidSubmission(event *p2p_bidder.BidSubmissionEvent) { } } - if t.cfg.SlotArtifactCaptureEnabled && event.SignedBid != nil { + if t.cfgSvc.Current().SlotArtifactCaptureEnabled && event.SignedBid != nil { idx, err := t.artifacts.StoreBid(event.Slot, event.SignedBid.Version, event.SignedBid, BidArtifactMeta{ Transport: string(payload_builder.BidTransportP2P), @@ -691,7 +691,7 @@ func (t *Tracker) handleRevealResult(result *payload_bidder.RevealResult) { attempt.Status = RevealStatusFailed } - if t.cfg.SlotArtifactCaptureEnabled && result.Envelope != nil { + if t.cfgSvc.Current().SlotArtifactCaptureEnabled && result.Envelope != nil { if err := t.artifacts.StoreEnvelope(result.Slot, result.Envelope.Version, result.Envelope); err != nil { t.log.WithError(err).WithField("slot", result.Slot). @@ -771,7 +771,7 @@ func (t *Tracker) RecordBuilderAPIBid(slot phase0.Slot, forkName string, signedB } marshaler, isMarshaler := signedBid.(sszMarshaler) - if t.cfg.SlotArtifactCaptureEnabled && isMarshaler { + if t.cfgSvc.Current().SlotArtifactCaptureEnabled && isMarshaler { fork, err := version.DataVersionFromString(forkName) if err != nil { fork = version.DataVersionUnknown @@ -1016,8 +1016,9 @@ func (t *Tracker) SubscribeUpdates(capacity int) *utils.Subscription[*SlotResult // retention windows. func (t *Tracker) pruneForEpoch(epoch phase0.Epoch) { slotsPerEpoch := t.chainSvc.GetChainSpec().SlotsPerEpoch + cfg := t.cfgSvc.Current() - if retention := t.cfg.SlotResultRetentionEpochs; retention > 0 && uint64(epoch) > retention { + if retention := cfg.SlotResultRetentionEpochs; retention > 0 && uint64(epoch) > retention { cutoff := phase0.Slot((uint64(epoch) - retention) * slotsPerEpoch) pruned := t.store.Prune(func(slot phase0.Slot) bool { return slot < cutoff }) @@ -1038,7 +1039,7 @@ func (t *Tracker) pruneForEpoch(epoch phase0.Epoch) { t.mu.Unlock() } - if retention := t.cfg.SlotArtifactRetentionEpochs; retention > 0 && uint64(epoch) > retention { + if retention := cfg.SlotArtifactRetentionEpochs; retention > 0 && uint64(epoch) > retention { cutoff := phase0.Slot((uint64(epoch) - retention) * slotsPerEpoch) t.artifacts.PruneBefore(cutoff) } diff --git a/pkg/slot_results/tracker_test.go b/pkg/slot_results/tracker_test.go index e7e7c02..1c1db55 100644 --- a/pkg/slot_results/tracker_test.go +++ b/pkg/slot_results/tracker_test.go @@ -81,7 +81,8 @@ func newTrackerTestEnv(t *testing.T, withDB bool) *trackerTestEnv { cfg.APIPort = 8080 chainSvc := newStubChain() - planSvc := action_plan.NewPlanService(cfg, chainSvc, log) + cfgSvc := config.NewStaticService(cfg) + planSvc := action_plan.NewPlanService(cfgSvc, chainSvc, log) dbFile := "" if withDB { @@ -93,7 +94,7 @@ func newTrackerTestEnv(t *testing.T, withDB bool) *trackerTestEnv { t.Cleanup(func() { _ = stateDB.Close() }) - tracker := NewTracker(cfg, chainSvc, stateDB, planSvc, nil, nil, nil, nil, nil, log) + tracker := NewTracker(cfgSvc, chainSvc, stateDB, planSvc, nil, nil, nil, nil, nil, log) return &trackerTestEnv{ cfg: cfg, @@ -375,7 +376,7 @@ func TestPersistenceRoundTrip(t *testing.T) { log := logrus.New() log.SetOutput(io.Discard) - fresh := NewTracker(env.cfg, env.chainSvc, env.stateDB, env.planSvc, nil, nil, nil, nil, nil, log) + fresh := NewTracker(config.NewStaticService(env.cfg), env.chainSvc, env.stateDB, env.planSvc, nil, nil, nil, nil, nil, log) fresh.SetPersistence(t.Context(), env.stateDB) defer fresh.store.Stop() diff --git a/pkg/validatorranges/resolver.go b/pkg/validatorranges/resolver.go index 631590f..e4e5446 100644 --- a/pkg/validatorranges/resolver.go +++ b/pkg/validatorranges/resolver.go @@ -28,7 +28,7 @@ type rangeEntry struct { // Resolver maps validator indices to client names using configured ranges. type Resolver struct { - cfg *config.ValidatorRangesConfig + cfg *config.ValidatorRangesConfig // static section (not in the mutable-settings registry) log logrus.FieldLogger ranges []rangeEntry mu sync.RWMutex diff --git a/pkg/webui/handlers/api/action_plan_test.go b/pkg/webui/handlers/api/action_plan_test.go index ccd0558..b4ae0ae 100644 --- a/pkg/webui/handlers/api/action_plan_test.go +++ b/pkg/webui/handlers/api/action_plan_test.go @@ -84,14 +84,15 @@ func newPlanAPITestEnv(t *testing.T) *planAPITestEnv { cfg.APIPort = 8080 chainSvc := newPlanTestChain() - planSvc := action_plan.NewPlanService(cfg, chainSvc, log) + cfgSvc := config.NewStaticService(cfg) + planSvc := action_plan.NewPlanService(cfgSvc, chainSvc, log) stateDB := db.NewDatabase(&db.Config{File: filepath.Join(t.TempDir(), "state.db")}, log) require.NoError(t, stateDB.Init()) t.Cleanup(func() { _ = stateDB.Close() }) - tracker := slot_results.NewTracker(cfg, chainSvc, stateDB, planSvc, nil, nil, nil, nil, nil, log) + tracker := slot_results.NewTracker(cfgSvc, chainSvc, stateDB, planSvc, nil, nil, nil, nil, nil, log) authHandler, err := auth.NewAuthHandler(context.Background(), "") require.NoError(t, err) @@ -477,7 +478,7 @@ func TestUpdateSettingsPathBased(t *testing.T) { cfg := config.DefaultConfig() defaults := config.DefaultConfig() - settingsSvc, err := config.NewService(cfg, defaults, map[string]bool{}, stateDB, log) + settingsSvc, err := config.NewService(cfg, defaults, map[string]bool{}, 12*time.Second, stateDB, log) require.NoError(t, err) authHandler, err := auth.NewAuthHandler(context.Background(), "") @@ -497,16 +498,18 @@ func TestUpdateSettingsPathBased(t *testing.T) { return rec } - // Partial update of two unrelated settings in one call. + // Partial update of two unrelated settings in one call. Applied changes + // publish a fresh snapshot; the operator config passed to NewService is + // never mutated. rec := post(`{"epbs.bid_subsidy": 12345, "schedule.mode": "every_nth"}`) require.Equal(t, http.StatusOK, rec.Code, rec.Body.String()) - assert.Equal(t, uint64(12345), cfg.EPBS.BidSubsidy) - assert.Equal(t, config.ScheduleModeEveryN, cfg.Schedule.Mode) + assert.Equal(t, uint64(12345), settingsSvc.Current().EPBS.BidSubsidy) + assert.Equal(t, config.ScheduleModeEveryN, settingsSvc.Current().Schedule.Mode) // Unknown key → 400, nothing applied. rec = post(`{"epbs.bid_subsidy": 1, "no.such.key": 2}`) assert.Equal(t, http.StatusBadRequest, rec.Code) - assert.Equal(t, uint64(12345), cfg.EPBS.BidSubsidy, "atomic: nothing applied on unknown key") + assert.Equal(t, uint64(12345), settingsSvc.Current().EPBS.BidSubsidy, "atomic: nothing applied on unknown key") // Invalid value → 400. rec = post(`{"slot_result_retention_epochs": 0}`) diff --git a/pkg/webui/handlers/api/builder_keys.go b/pkg/webui/handlers/api/builder_keys.go index 7547631..b5d2043 100644 --- a/pkg/webui/handlers/api/builder_keys.go +++ b/pkg/webui/handlers/api/builder_keys.go @@ -94,7 +94,7 @@ func (h *APIHandler) GetBuilderKeys(w http.ResponseWriter, _ *http.Request) { // builderKeysSettings snapshots the mutable key-set configuration. func (h *APIHandler) builderKeysSettings() BuilderKeysSettings { - cfg := h.settingsSvc.Load() + cfg := h.settingsSvc.Current() return BuilderKeysSettings{ TargetCount: cfg.BuilderKeys.EffectiveTargetCount(), @@ -341,7 +341,7 @@ func (h *APIHandler) ExitBuilderKey(w http.ResponseWriter, r *http.Request) { // Lower the target after the exit landed, so a failed exit never shrinks // the fleet the operator asked for. if req.LowerTarget { - target := h.settingsSvc.Load().BuilderKeys.EffectiveTargetCount() + target := h.settingsSvc.Current().BuilderKeys.EffectiveTargetCount() if target > 1 { h.applyKeyTarget(w, r, token, "builder_keys.target", detail, target-1) } diff --git a/pkg/webui/handlers/api/builder_keys_test.go b/pkg/webui/handlers/api/builder_keys_test.go index 95378a9..34aa72f 100644 --- a/pkg/webui/handlers/api/builder_keys_test.go +++ b/pkg/webui/handlers/api/builder_keys_test.go @@ -34,11 +34,11 @@ func keysTestHandler(t *testing.T, target uint64) *APIHandler { // The settings service owns cfg, so it must be built before the registry // reads the resolved values. - settingsSvc, err := config.NewService(cfg, defaults, map[string]bool{}, + settingsSvc, err := config.NewService(cfg, defaults, map[string]bool{}, 0, db.NewDatabase(&db.Config{}, log), log) require.NoError(t, err) - registry, err := builder_keys.NewRegistry(cfg, testEntryPrivkey, log) + registry, err := builder_keys.NewRegistry(config.NewStaticService(cfg), testEntryPrivkey, log) require.NoError(t, err) registry.Refresh() diff --git a/pkg/webui/handlers/api/builder_preferences_test.go b/pkg/webui/handlers/api/builder_preferences_test.go index 2212df4..044b487 100644 --- a/pkg/webui/handlers/api/builder_preferences_test.go +++ b/pkg/webui/handlers/api/builder_preferences_test.go @@ -29,7 +29,7 @@ func TestGetBuilderPreferences_NotEnabled(t *testing.T) { func TestGetBuilderPreferences_ReturnsEntries(t *testing.T) { cfg := &config.BuilderAPIConfig{} - srv := builderapi.NewServer(cfg, logrus.New(), nil, nil, nil, nil, nil) + srv := builderapi.NewServer(config.NewStaticService(&config.Config{BuilderAPI: *cfg}), logrus.New(), nil, nil, nil, nil, nil) var pk phase0.BLSPubKey pk[0] = 0xab