diff --git a/CLAUDE.md b/CLAUDE.md index a1621c6a..d88e4bf6 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -227,7 +227,7 @@ obol sell delete ollama-gated -n llm ## RPC Gateway -`obol network add|remove|status` manages remote RPCs via eRPC ConfigMap. Default: read-only (blocks `eth_sendRawTransaction`, `eth_sendTransaction`). `--allow-writes` flips readOnly → route `eth_sendRawTransaction` to the user-chosen upstream via eRPC per-method selection policy. `--endpoint` adds a custom RPC directly (skips ChainList). Key functions in `internal/network/rpc.go`: `AddPublicRPCs()` (ChainList), `AddCustomRPC()`, `ListRPCNetworks()`. Record-on-write: add/remove update `$CONFIG_DIR/rpc/recorded-upstreams.yaml` (`internal/network/record.go`); `obol stack up` replays via `ReconcileRecordedRPCs()` so remote RPCs survive cluster recreation. Local-node upstreams are NOT recorded (`network sync` re-registers them). +`obol network add|remove|status` manages remote RPCs via eRPC ConfigMap. Default: read-only (blocks `eth_sendRawTransaction`, `eth_sendTransaction`). `--allow-writes` flips readOnly → route `eth_sendRawTransaction` to the user-chosen upstream via eRPC per-method selection policy. `--endpoint` adds a custom RPC directly (skips ChainList). Key functions in `internal/network/rpc.go`: `AddPublicRPCs()` (ChainList), `AddCustomRPC()`, `ListRPCNetworks()`. Record-on-write: add/remove update `$CONFIG_DIR/rpc/recorded-upstreams.yaml` (`internal/network/record.go`); `obol stack up` replays via `ReconcileRecordedRPCs()` so remote RPCs survive cluster recreation. Local-node upstreams are NOT recorded (`network sync` re-registers them). **Durable eRPC operator config** (so local baskets are not lost on stack up): `obol network erpc set|status|reset` persists `$CONFIG_DIR/rpc/erpc-overlay.yaml` (merge-by-id into the live ConfigMap; not a full replace) and re-applies after recorded remotes on stack up via `ReconcileERPCOverlay()` (`internal/network/overlay.go`, #763). Example: `examples/erpc-overlay-hyperevm.yaml`. ## Network Management @@ -242,7 +242,7 @@ Two-stage templating: `values.yaml.gotmpl` annotated with `@enum`/`@default`/`@d | Command | Action | |---------|--------| | `obol stack init` | Generate cluster ID (petname), resolve absolute paths, write k3d.yaml, copy infrastructure | -| `obol stack up` | `k3d cluster create`, export kubeconfig, k3s auto-applies manifests, auto-configures LiteLLM with Ollama models (preserves Ollama modified-time order; `:cloud` aliases demoted behind local chat models; embedding-only models last; warns + suggests `ollama pull qwen3.5:4b` when empty or all-`:cloud`), re-imposes recorded model config (`model.ReconcileRecorded` — operator's `obol model` choices win over auto-detect), deploys default Hermes agent, applies agent capabilities, starts persistent Cloudflare tunnel, then replays recorded state: RPC upstreams (`network.ReconcileRecordedRPCs`) → Agent CRs (`agentcrd.ResumeAll`) → sell offers (`resumeSellOffers`; agent-type offers ride the sell-http store). Guard: `cmd/obol/stackup_resume_guard_test.go` | +| `obol stack up` | `k3d cluster create`, export kubeconfig, k3s auto-applies manifests, auto-configures LiteLLM with Ollama models (preserves Ollama modified-time order; `:cloud` aliases demoted behind local chat models; embedding-only models last; warns + suggests `ollama pull qwen3.5:4b` when empty or all-`:cloud`), re-imposes recorded model config (`model.ReconcileRecorded` — operator's `obol model` choices win over auto-detect), deploys default Hermes agent, applies agent capabilities, starts persistent Cloudflare tunnel, then replays recorded state: RPC upstreams (`network.ReconcileRecordedRPCs`) → eRPC operator overlay (`network.ReconcileERPCOverlay`) → Agent CRs (`agentcrd.ResumeAll`) → sell offers (`resumeSellOffers`; agent-type offers ride the sell-http store). Guard: `cmd/obol/stackup_resume_guard_test.go` | | `obol stack down` | `k3d cluster stop` (delete fallback; preserves config + data) | | `obol stack purge [-f]` | Delete config; `-f` also deletes root-owned PVCs; `-f` offers a full `stack export` first (fallback: OpenClaw wallet prompt) | | `obol stack export` | Full backup archive: config dir (minus kubeconfig/defaults), agent data dirs (brains + keystores, deployments quiesced for consistency), encrypted wallet backups, etcd-drift resources (Agent CRs, ServiceOffers, LiteLLM/eRPC CMs). `internal/stackbackup/` | diff --git a/cmd/obol/main.go b/cmd/obol/main.go index d82d2f69..3d27d386 100644 --- a/cmd/obol/main.go +++ b/cmd/obol/main.go @@ -212,6 +212,11 @@ GLOBAL OPTIONS:{{template "visibleFlagTemplate" .}}{{end}} // Replay recorded remote RPC upstreams into the // (possibly fresh) eRPC ConfigMap. Best-effort. network.ReconcileRecordedRPCs(cfg, u) + // Re-apply durable multi-upstream eRPC operator + // overlays (baskets/scoring/rate-limits) AFTER + // simple recorded remotes. Best-effort. + // See ObolNetwork/obol-stack#763. + network.ReconcileERPCOverlay(cfg, u) // Re-apply recorded Agent CRs BEFORE sell offers: // agent-backed ServiceOffers resolve agent.ref and // would dangle without their Agent. Best-effort. diff --git a/cmd/obol/network.go b/cmd/obol/network.go index 701b2226..6a30deb0 100644 --- a/cmd/obol/network.go +++ b/cmd/obol/network.go @@ -76,6 +76,7 @@ func networkCommand(cfg *config.Config) *cli.Command { networkAddCommand(cfg), networkRemoveCommand(cfg), networkStatusCommand(cfg), + networkERPCCommand(cfg), }, } } @@ -534,6 +535,81 @@ func networkStatusCommand(cfg *config.Config) *cli.Command { } } +// network erpc — durable operator eRPC config that survives stack up (#763). +// Verbs match sell-info set/reset (host-side intent) + status. +// --------------------------------------------------------------------------- + +func networkERPCCommand(cfg *config.Config) *cli.Command { + return &cli.Command{ + Name: "erpc", + Usage: "Manage durable operator eRPC config (host-side; re-applied on stack up so local baskets are not lost)", + Commands: []*cli.Command{ + { + Name: "set", + Usage: "Set durable eRPC config from a YAML file and merge it into the live ConfigMap", + Flags: []cli.Flag{ + &cli.StringFlag{ + Name: "file", + Aliases: []string{"f"}, + Usage: "YAML fragment: networks, upstreams, rateLimiters, cachePoliciesAdd (saved to $CONFIG_DIR/rpc/erpc-overlay.yaml)", + Required: true, + }, + }, + Action: func(_ context.Context, cmd *cli.Command) error { + return network.SetERPC(cfg, getUI(cmd), cmd.String("file")) + }, + }, + { + Name: "status", + Usage: "Show the durable eRPC config on disk", + Action: func(_ context.Context, cmd *cli.Command) error { + u := getUI(cmd) + st, err := network.StatusERPC(cfg) + if err != nil { + return err + } + u.Printf("eRPC operator config\n") + u.Printf("====================\n\n") + u.Printf("Path: %s\n", st.Path) + if !st.Present { + u.Info("Status: not set") + u.Info("Set with: obol network erpc set -f ") + return nil + } + u.Printf("Status: set (v%d, hash %s)\n", st.Version, st.ContentHash) + u.Printf("Networks (%d):\n", st.NetworkCount) + for _, k := range st.NetworkKeys { + u.Printf(" - %s\n", k) + } + if st.NetworkCount == 0 { + u.Info(" (none)") + } + u.Printf("Upstreams (%d):\n", st.UpstreamCount) + for _, id := range st.UpstreamIDs { + u.Printf(" - %s\n", id) + } + if st.UpstreamCount == 0 { + u.Info(" (none)") + } + u.Printf("Rate-limit budgets: %d\n", st.BudgetCount) + u.Printf("Cache policies to add: %d\n", st.CachePolicyAdd) + return nil + }, + }, + { + Name: "reset", + Usage: "Reset durable eRPC config (strip from live ConfigMap and delete host file)", + Action: func(_ context.Context, cmd *cli.Command) error { + return network.ResetERPC(cfg, getUI(cmd)) + }, + }, + }, + Action: func(ctx context.Context, cmd *cli.Command) error { + return cli.ShowSubcommandHelp(cmd) + }, + } +} + // chainIDToName returns a human-readable name for a chain ID. func chainIDToName(chainID int) string { names := map[int]string{ @@ -550,6 +626,7 @@ func chainIDToName(chainID int) string { 43114: "Avalanche", 59144: "Linea", 84532: "Base Sepolia", + 999: "HyperEVM", 534352: "Scroll", 560048: "Hoodi", 11155111: "Sepolia", diff --git a/cmd/obol/stackup_resume_guard_test.go b/cmd/obol/stackup_resume_guard_test.go index e37d3f9a..c089f947 100644 --- a/cmd/obol/stackup_resume_guard_test.go +++ b/cmd/obol/stackup_resume_guard_test.go @@ -21,6 +21,7 @@ func TestStackUpAction_ReplaysRecordedState(t *testing.T) { upIdx := strings.Index(body, "stack.Up(cfg") rpcIdx := strings.Index(body, "network.ReconcileRecordedRPCs(") + overlayIdx := strings.Index(body, "network.ReconcileERPCOverlay(") agentsIdx := strings.Index(body, "agentcrd.ResumeAll(") appsIdx := strings.Index(body, "app.ResumeAll(") offersIdx := strings.Index(body, "resumeSellOffers(") @@ -28,6 +29,9 @@ func TestStackUpAction_ReplaysRecordedState(t *testing.T) { if rpcIdx < 0 { t.Fatal("cmd/obol/main.go must call network.ReconcileRecordedRPCs — without it recorded remote RPCs never reach a freshly-recreated cluster") } + if overlayIdx < 0 { + t.Fatal("cmd/obol/main.go must call network.ReconcileERPCOverlay — without it durable eRPC baskets (e.g. HyperEVM) never re-apply after stack up (#763)") + } if agentsIdx < 0 { t.Fatal("cmd/obol/main.go must call agentcrd.ResumeAll — without it recorded Agent CRs never reach a freshly-recreated cluster") } @@ -37,9 +41,12 @@ func TestStackUpAction_ReplaysRecordedState(t *testing.T) { if upIdx < 0 || offersIdx < 0 { t.Fatalf("expected stack.Up and resumeSellOffers in main.go; upIdx=%d offersIdx=%d", upIdx, offersIdx) } - if rpcIdx < upIdx || agentsIdx < upIdx || appsIdx < upIdx { + if rpcIdx < upIdx || overlayIdx < upIdx || agentsIdx < upIdx || appsIdx < upIdx { t.Error("recorded-state replay must run AFTER stack.Up — before it there is no kubeconfig/cluster") } + if overlayIdx < rpcIdx { + t.Error("ReconcileERPCOverlay must run AFTER ReconcileRecordedRPCs — overlay merges onto base+recorded remotes") + } if agentsIdx > offersIdx { t.Error("agentcrd.ResumeAll must run BEFORE resumeSellOffers — agent-backed ServiceOffers need their Agent CR first") } diff --git a/examples/erpc-overlay-hyperevm.yaml b/examples/erpc-overlay-hyperevm.yaml new file mode 100644 index 00000000..9e7c9789 --- /dev/null +++ b/examples/erpc-overlay-hyperevm.yaml @@ -0,0 +1,103 @@ +# Example eRPC operator overlay — HyperEVM (chain 999) basket. +# +# Durable operator eRPC config so local baskets survive `obol stack up`. Set with: +# +# obol network erpc set -f examples/erpc-overlay-hyperevm.yaml +# obol network erpc status +# +# Host copy lives at $OBOL_CONFIG_DIR/rpc/erpc-overlay.yaml (0600) and is +# re-merged after recorded remotes on every stack up (see #763). +# +# Edit the local endpoint for your co-located hl-node before applying. + +version: 1 + +networks: + - alias: hyperevm + architecture: evm + evm: + chainId: 999 + failsafe: + timeout: + duration: 10s + retry: + maxAttempts: 3 + delay: 100ms + hedge: + delay: 150ms + maxCount: 1 + +upstreams: + # Prefer a co-located non-validator HyperEVM RPC when present. + - id: local-hl-node + endpoint: http://192.168.50.21:3001/evm + evm: + chainId: 999 + routing: + scoreMultipliers: + - network: "*" + method: "*" + overall: 5.0 + + - id: hyperevm-official + endpoint: https://rpc.hyperliquid.xyz/evm + evm: + chainId: 999 + tags: ["tier:fallback"] + rateLimitBudget: hyperevm-official + routing: + scoreMultipliers: + - network: "*" + method: "*" + overall: 0.3 + + - id: hyperevm-drpc + endpoint: https://hyperliquid.drpc.org + evm: + chainId: 999 + tags: ["tier:fallback"] + rateLimitBudget: hyperevm-drpc + routing: + scoreMultipliers: + - network: "*" + method: "*" + overall: 0.3 + + - id: hyperevm-onfinality + endpoint: https://hyperliquid.api.onfinality.io/evm/public + evm: + chainId: 999 + tags: ["tier:fallback"] + rateLimitBudget: hyperevm-onfinality + routing: + scoreMultipliers: + - network: "*" + method: "*" + overall: 0.3 + +rateLimiters: + budgets: + - id: hyperevm-official + rules: + - method: "*" + maxCount: 100 + period: second + - id: hyperevm-drpc + rules: + - method: "*" + maxCount: 30 + period: second + - id: hyperevm-onfinality + rules: + - method: "*" + maxCount: 25 + period: second + +# eth_call @ latest (HyperCore precompiles) needs realtime caching; the chart +# default finality:unfinalized can serve stale precompile reads under burst. +cachePoliciesAdd: + - network: "*" + method: eth_call + finality: realtime + connector: memory-cache + ttl: 2s diff --git a/internal/network/overlay.go b/internal/network/overlay.go new file mode 100644 index 00000000..9fbdbb0f --- /dev/null +++ b/internal/network/overlay.go @@ -0,0 +1,618 @@ +package network + +import ( + "crypto/sha256" + "encoding/hex" + "fmt" + "os" + "path/filepath" + "strings" + + "github.com/ObolNetwork/obol-stack/internal/config" + "github.com/ObolNetwork/obol-stack/internal/kubectl" + "github.com/ObolNetwork/obol-stack/internal/ui" + "gopkg.in/yaml.v3" +) + +// Host-side eRPC operator overlay. Complements recorded-upstreams.yaml +// (simple remote remotes) with durable multi-upstream baskets, scoring, +// rate-limit budgets, and cache policy fragments that survive +// `obol stack up` / helmfile re-renders of the eRPC chart. +// +// Lifecycle (mirrors Hermes mergePreservedHermesConfigKeys + recorded RPCs): +// 1. Operator writes overlay via `obol network overlay apply -f` +// 2. Overlay is stored at $CONFIG_DIR/rpc/erpc-overlay.yaml (0600) +// 3. applyOverlayToCluster deep-merges into the live eRPC ConfigMap +// 4. ReconcileERPCOverlay re-applies after stack up (after ReconcileRecordedRPCs) +// +// See ObolNetwork/obol-stack#763. + +const ( + erpcOverlayVersion = 1 + erpcOverlayAnnotKey = "obol.stack/erpc-overlay-hash" + erpcOverlayAnnotSource = "obol.stack/erpc-overlay-source" +) + +// ERPCOverlay is the durable operator overlay document. +// Fragments are free-form maps so operators can pass full eRPC YAML objects +// (scoreMultipliers, failsafe, rateLimitBudget, …) without a rigid schema. +type ERPCOverlay struct { + Version int `yaml:"version"` + + // Networks are merged into projects[0].networks by chainId (preferred) + // or alias. Matching entries are replaced; others are appended. + Networks []map[string]any `yaml:"networks,omitempty"` + + // Upstreams are merged into projects[0].upstreams by id. Matching + // entries are replaced; others are appended (after existing). + Upstreams []map[string]any `yaml:"upstreams,omitempty"` + + // RateLimiters is deep-merged at the top level. Budget entries under + // rateLimiters.budgets are merged by id. + RateLimiters map[string]any `yaml:"rateLimiters,omitempty"` + + // CachePoliciesAdd are appended to database.evmJsonRpcCache.policies + // when no existing policy has the same network+method+finality triple. + CachePoliciesAdd []map[string]any `yaml:"cachePoliciesAdd,omitempty"` +} + +// ERPCOverlayStatus is a read-only summary for CLI status. +type ERPCOverlayStatus struct { + Path string + Present bool + Version int + NetworkCount int + UpstreamCount int + BudgetCount int + CachePolicyAdd int + NetworkKeys []string + UpstreamIDs []string + ContentHash string +} + +func erpcOverlayPath(cfg *config.Config) string { + return filepath.Join(cfg.ConfigDir, "rpc", "erpc-overlay.yaml") +} + +// ReconcileERPCOverlay re-applies the host-side overlay into the live eRPC +// ConfigMap. Called after ReconcileRecordedRPCs on stack up. Best-effort: +// missing overlay is a silent no-op; apply errors are warned, not fatal. +func ReconcileERPCOverlay(cfg *config.Config, u *ui.UI) { + ov, err := readERPCOverlay(cfg) + if err != nil { + u.Warnf("Could not read durable eRPC config: %v", err) + return + } + if ov == nil { + return + } + if err := applyOverlayToCluster(cfg, ov, "reconcile"); err != nil { + u.Warnf("Could not restore durable eRPC operator config: %v", err) + return + } + u.Successf("Restored durable eRPC operator config (%d network(s), %d upstream(s))", + len(ov.Networks), len(ov.Upstreams)) +} + +// ApplyERPCOverlayFile loads an overlay YAML from path, persists it under +// ConfigDir, and merges it into the live eRPC ConfigMap. +func SetERPC(cfg *config.Config, u *ui.UI, path string) error { + data, err := os.ReadFile(path) + if err != nil { + return fmt.Errorf("read overlay file: %w", err) + } + ov, err := parseERPCOverlay(data) + if err != nil { + return err + } + if err := writeERPCOverlay(cfg, ov); err != nil { + return err + } + if err := applyOverlayToCluster(cfg, ov, filepath.Base(path)); err != nil { + return err + } + u.Successf("eRPC config set and saved to %s", erpcOverlayPath(cfg)) + u.Infof(" networks=%d upstreams=%d cachePoliciesAdd=%d", + len(ov.Networks), len(ov.Upstreams), len(ov.CachePoliciesAdd)) + return nil +} + +// ClearERPCOverlay removes overlay-owned fragments from the live ConfigMap +// (best-effort), then deletes the host-side overlay file. +func ResetERPC(cfg *config.Config, u *ui.UI) error { + ov, err := readERPCOverlay(cfg) + if err != nil { + return err + } + if ov == nil { + u.Info("No durable eRPC config on disk") + return nil + } + + if err := removeOverlayFromCluster(cfg, ov); err != nil { + u.Warnf("Could not strip eRPC config from live ConfigMap (will still reset host file): %v", err) + } else { + u.Success("Removed operator eRPC networks/upstreams from live ConfigMap") + } + + path := erpcOverlayPath(cfg) + if err := os.Remove(path); err != nil && !os.IsNotExist(err) { + return fmt.Errorf("remove overlay file: %w", err) + } + // Best-effort annotation clear + _ = annotateERPCOverlay(cfg, "", "") + u.Successf("Reset eRPC config at %s", path) + u.Info("Chart base + recorded remotes remain; re-run `obol stack up` if you need a full eRPC re-render") + return nil +} + +// StatusERPCOverlay returns a summary of the on-disk overlay (no cluster I/O). +func StatusERPC(cfg *config.Config) (*ERPCOverlayStatus, error) { + path := erpcOverlayPath(cfg) + st := &ERPCOverlayStatus{Path: path} + ov, err := readERPCOverlay(cfg) + if err != nil { + return nil, err + } + if ov == nil { + return st, nil + } + st.Present = true + st.Version = ov.Version + st.NetworkCount = len(ov.Networks) + st.UpstreamCount = len(ov.Upstreams) + st.CachePolicyAdd = len(ov.CachePoliciesAdd) + st.BudgetCount = countRateLimitBudgets(ov.RateLimiters) + for _, n := range ov.Networks { + st.NetworkKeys = append(st.NetworkKeys, describeNetwork(n)) + } + for _, u := range ov.Upstreams { + if id, _ := u["id"].(string); id != "" { + st.UpstreamIDs = append(st.UpstreamIDs, id) + } + } + if data, err := os.ReadFile(path); err == nil { + sum := sha256.Sum256(data) + st.ContentHash = hex.EncodeToString(sum[:8]) + } + return st, nil +} + +// --- persistence --- + +func readERPCOverlay(cfg *config.Config) (*ERPCOverlay, error) { + data, err := os.ReadFile(erpcOverlayPath(cfg)) + if os.IsNotExist(err) { + return nil, nil + } + if err != nil { + return nil, err + } + return parseERPCOverlay(data) +} + +func parseERPCOverlay(data []byte) (*ERPCOverlay, error) { + var ov ERPCOverlay + if err := yaml.Unmarshal(data, &ov); err != nil { + return nil, fmt.Errorf("parse eRPC overlay: %w", err) + } + if ov.Version == 0 { + ov.Version = erpcOverlayVersion + } + if ov.Version != erpcOverlayVersion { + return nil, fmt.Errorf("unsupported erpc-overlay version %d (want %d)", ov.Version, erpcOverlayVersion) + } + if len(ov.Networks) == 0 && len(ov.Upstreams) == 0 && + len(ov.RateLimiters) == 0 && len(ov.CachePoliciesAdd) == 0 { + return nil, fmt.Errorf("eRPC overlay is empty (need networks, upstreams, rateLimiters, and/or cachePoliciesAdd)") + } + // Upstreams must have ids for merge keys + for i, u := range ov.Upstreams { + id, _ := u["id"].(string) + if strings.TrimSpace(id) == "" { + return nil, fmt.Errorf("upstreams[%d]: missing id", i) + } + } + return &ov, nil +} + +func writeERPCOverlay(cfg *config.Config, ov *ERPCOverlay) error { + if ov.Version == 0 { + ov.Version = erpcOverlayVersion + } + path := erpcOverlayPath(cfg) + if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil { + return err + } + data, err := yaml.Marshal(ov) + if err != nil { + return err + } + tmp := path + ".tmp" + if err := os.WriteFile(tmp, data, 0o600); err != nil { + return err + } + return os.Rename(tmp, path) +} + +// --- cluster apply / merge --- + +func applyOverlayToCluster(cfg *config.Config, ov *ERPCOverlay, source string) error { + erpcConfig, err := readERPCConfig(cfg) + if err != nil { + return err + } + if err := mergeERPCOverlay(erpcConfig, ov); err != nil { + return err + } + if err := writeERPCConfig(cfg, erpcConfig); err != nil { + return err + } + hash := "" + if data, err := os.ReadFile(erpcOverlayPath(cfg)); err == nil { + sum := sha256.Sum256(data) + hash = hex.EncodeToString(sum[:8]) + } + _ = annotateERPCOverlay(cfg, hash, source) + return nil +} + +func removeOverlayFromCluster(cfg *config.Config, ov *ERPCOverlay) error { + erpcConfig, err := readERPCConfig(cfg) + if err != nil { + return err + } + if err := stripERPCOverlay(erpcConfig, ov); err != nil { + return err + } + return writeERPCConfig(cfg, erpcConfig) +} + +// mergeERPCOverlay mutates erpcConfig in place. Exported for tests via +// the unexported name used from overlay_test.go (same package). +func mergeERPCOverlay(erpcConfig map[string]any, ov *ERPCOverlay) error { + if ov == nil { + return nil + } + projects, ok := erpcConfig["projects"].([]any) + if !ok || len(projects) == 0 { + return fmt.Errorf("eRPC config has no projects") + } + project, ok := projects[0].(map[string]any) + if !ok { + return fmt.Errorf("eRPC config project[0] is not a map") + } + + if len(ov.Networks) > 0 { + networks, _ := project["networks"].([]any) + project["networks"] = mergeNetworksByKey(networks, ov.Networks) + } + if len(ov.Upstreams) > 0 { + upstreams, _ := project["upstreams"].([]any) + project["upstreams"] = mergeUpstreamsByID(upstreams, ov.Upstreams) + } + if len(ov.RateLimiters) > 0 { + existing, _ := erpcConfig["rateLimiters"].(map[string]any) + erpcConfig["rateLimiters"] = mergeRateLimiters(existing, ov.RateLimiters) + } + if len(ov.CachePoliciesAdd) > 0 { + if err := mergeCachePolicies(erpcConfig, ov.CachePoliciesAdd); err != nil { + return err + } + } + return nil +} + +func stripERPCOverlay(erpcConfig map[string]any, ov *ERPCOverlay) error { + projects, ok := erpcConfig["projects"].([]any) + if !ok || len(projects) == 0 { + return fmt.Errorf("eRPC config has no projects") + } + project, ok := projects[0].(map[string]any) + if !ok { + return fmt.Errorf("eRPC config project[0] is not a map") + } + + if len(ov.Upstreams) > 0 { + drop := map[string]struct{}{} + for _, u := range ov.Upstreams { + if id, _ := u["id"].(string); id != "" { + drop[id] = struct{}{} + } + } + upstreams, _ := project["upstreams"].([]any) + kept := make([]any, 0, len(upstreams)) + for _, u := range upstreams { + um, ok := u.(map[string]any) + if !ok { + kept = append(kept, u) + continue + } + id, _ := um["id"].(string) + if _, hit := drop[id]; hit { + continue + } + kept = append(kept, u) + } + project["upstreams"] = kept + } + + if len(ov.Networks) > 0 { + drop := map[string]struct{}{} + for _, n := range ov.Networks { + if k := networkMergeKey(n); k != "" { + drop[k] = struct{}{} + } + } + networks, _ := project["networks"].([]any) + kept := make([]any, 0, len(networks)) + for _, n := range networks { + nm, ok := n.(map[string]any) + if !ok { + kept = append(kept, n) + continue + } + if _, hit := drop[networkMergeKey(nm)]; hit { + continue + } + kept = append(kept, n) + } + project["networks"] = kept + } + // Leave rateLimiters / cache policies in place — removing budgets that + // other upstreams still reference is riskier than leaving unused ones. + return nil +} + +func mergeNetworksByKey(existing []any, overlay []map[string]any) []any { + out := make([]any, 0) // no capacity hint: len(a)+len(b) trips CodeQL allocation-size-overflow + index := map[string]int{} // mergeKey → index in out + for _, n := range existing { + nm, ok := n.(map[string]any) + if !ok { + out = append(out, n) + continue + } + k := networkMergeKey(nm) + if k != "" { + index[k] = len(out) + } + out = append(out, cloneMap(nm)) + } + for _, n := range overlay { + // Deep-ish copy so later mutations don't touch the overlay struct + entry := cloneMap(n) + k := networkMergeKey(entry) + if k != "" { + if i, ok := index[k]; ok { + out[i] = entry + continue + } + index[k] = len(out) + } + out = append(out, entry) + } + return out +} + +func mergeUpstreamsByID(existing []any, overlay []map[string]any) []any { + out := make([]any, 0) // no capacity hint: len(a)+len(b) trips CodeQL allocation-size-overflow + index := map[string]int{} + for _, u := range existing { + um, ok := u.(map[string]any) + if !ok { + out = append(out, u) + continue + } + id, _ := um["id"].(string) + if id != "" { + index[id] = len(out) + } + out = append(out, cloneMap(um)) + } + for _, u := range overlay { + entry := cloneMap(u) + id, _ := entry["id"].(string) + if id != "" { + if i, ok := index[id]; ok { + out[i] = entry + continue + } + index[id] = len(out) + } + out = append(out, entry) + } + return out +} + +func mergeRateLimiters(existing, overlay map[string]any) map[string]any { + if existing == nil { + existing = map[string]any{} + } + out := cloneMap(existing) + // budgets: merge by id + if ob, ok := overlay["budgets"]; ok { + out["budgets"] = mergeBudgetList(out["budgets"], ob) + } + for k, v := range overlay { + if k == "budgets" { + continue + } + out[k] = v + } + return out +} + +func mergeBudgetList(existing any, overlay any) []any { + var out []any + index := map[string]int{} + appendBudget := func(b any) { + bm, ok := b.(map[string]any) + if !ok { + out = append(out, b) + return + } + id, _ := bm["id"].(string) + entry := cloneMap(bm) + if id != "" { + if i, ok := index[id]; ok { + out[i] = entry + return + } + index[id] = len(out) + } + out = append(out, entry) + } + switch e := existing.(type) { + case []any: + for _, b := range e { + appendBudget(b) + } + } + switch o := overlay.(type) { + case []any: + for _, b := range o { + appendBudget(b) + } + case []map[string]any: + for _, b := range o { + appendBudget(b) + } + } + return out +} + +func mergeCachePolicies(erpcConfig map[string]any, add []map[string]any) error { + db, _ := erpcConfig["database"].(map[string]any) + if db == nil { + db = map[string]any{} + erpcConfig["database"] = db + } + cache, _ := db["evmJsonRpcCache"].(map[string]any) + if cache == nil { + cache = map[string]any{} + db["evmJsonRpcCache"] = cache + } + policies, _ := cache["policies"].([]any) + seen := map[string]struct{}{} + for _, p := range policies { + pm, ok := p.(map[string]any) + if !ok { + continue + } + seen[cachePolicyKey(pm)] = struct{}{} + } + for _, p := range add { + entry := cloneMap(p) + k := cachePolicyKey(entry) + if _, ok := seen[k]; ok { + // Replace existing policy with same key + for i, cur := range policies { + cm, ok := cur.(map[string]any) + if ok && cachePolicyKey(cm) == k { + policies[i] = entry + break + } + } + continue + } + policies = append(policies, entry) + seen[k] = struct{}{} + } + cache["policies"] = policies + return nil +} + +func cachePolicyKey(p map[string]any) string { + return fmt.Sprintf("%v|%v|%v", p["network"], p["method"], p["finality"]) +} + +func networkMergeKey(n map[string]any) string { + if evm, ok := n["evm"].(map[string]any); ok { + if cid := yamlInt(evm["chainId"]); cid != 0 { + return fmt.Sprintf("chain:%d", cid) + } + } + if alias, ok := n["alias"].(string); ok && alias != "" { + return "alias:" + alias + } + return "" +} + +func describeNetwork(n map[string]any) string { + alias, _ := n["alias"].(string) + cid := 0 + if evm, ok := n["evm"].(map[string]any); ok { + cid = yamlInt(evm["chainId"]) + } + switch { + case alias != "" && cid != 0: + return fmt.Sprintf("%s (chainId %d)", alias, cid) + case alias != "": + return alias + case cid != 0: + return fmt.Sprintf("chainId %d", cid) + default: + return "(unnamed network)" + } +} + +func countRateLimitBudgets(rl map[string]any) int { + if rl == nil { + return 0 + } + switch b := rl["budgets"].(type) { + case []any: + return len(b) + case []map[string]any: + return len(b) + default: + return 0 + } +} + +func cloneMap(m map[string]any) map[string]any { + // YAML round-trip keeps nested maps as map[string]any for our purposes + // without needing a full deep-copy library. + b, err := yaml.Marshal(m) + if err != nil { + out := make(map[string]any, len(m)) + for k, v := range m { + out[k] = v + } + return out + } + var out map[string]any + if err := yaml.Unmarshal(b, &out); err != nil { + out = make(map[string]any, len(m)) + for k, v := range m { + out[k] = v + } + } + return out +} + +func annotateERPCOverlay(cfg *config.Config, hash, source string) error { + if err := kubectl.EnsureCluster(cfg); err != nil { + return err + } + kubectlBin, kubeconfigPath := kubectl.Paths(cfg) + // Clear when both empty + if hash == "" && source == "" { + _ = kubectl.RunSilent(kubectlBin, kubeconfigPath, + "annotate", "configmap", erpcConfigMapName, "-n", erpcNamespace, + erpcOverlayAnnotKey+"-", erpcOverlayAnnotSource+"-", "--overwrite") + return nil + } + args := []string{ + "annotate", "configmap", erpcConfigMapName, "-n", erpcNamespace, + "--overwrite", + } + if hash != "" { + args = append(args, erpcOverlayAnnotKey+"="+hash) + } + if source != "" { + args = append(args, erpcOverlayAnnotSource+"="+source) + } + return kubectl.RunSilent(kubectlBin, kubeconfigPath, args...) +} diff --git a/internal/network/overlay_test.go b/internal/network/overlay_test.go new file mode 100644 index 00000000..ef7e45cc --- /dev/null +++ b/internal/network/overlay_test.go @@ -0,0 +1,352 @@ +package network + +import ( + "os" + "path/filepath" + "testing" + + "github.com/ObolNetwork/obol-stack/internal/config" + "gopkg.in/yaml.v3" +) + +func TestParseERPCOverlay_RequiresContent(t *testing.T) { + _, err := parseERPCOverlay([]byte("version: 1\n")) + if err == nil { + t.Fatal("expected empty overlay to fail") + } +} + +func TestParseERPCOverlay_RequiresUpstreamID(t *testing.T) { + _, err := parseERPCOverlay([]byte(` +version: 1 +upstreams: + - endpoint: https://example.com +`)) + if err == nil { + t.Fatal("expected missing id to fail") + } +} + +func TestERPCOverlayRoundTrip(t *testing.T) { + cfg := &config.Config{ConfigDir: t.TempDir()} + ov := &ERPCOverlay{ + Version: 1, + Networks: []map[string]any{ + { + "alias": "hyperevm", + "architecture": "evm", + "evm": map[string]any{"chainId": 999}, + }, + }, + Upstreams: []map[string]any{ + { + "id": "local-hl-node", + "endpoint": "http://192.168.50.21:3001/evm", + "evm": map[string]any{"chainId": 999}, + }, + { + "id": "hyperevm-official", + "endpoint": "https://rpc.hyperliquid.xyz/evm", + "evm": map[string]any{"chainId": 999}, + }, + }, + RateLimiters: map[string]any{ + "budgets": []any{ + map[string]any{ + "id": "hyperevm-official", + "rules": []any{ + map[string]any{"method": "*", "maxCount": 100, "period": "second"}, + }, + }, + }, + }, + CachePoliciesAdd: []map[string]any{ + {"network": "*", "method": "eth_call", "finality": "realtime", "connector": "memory-cache", "ttl": "2s"}, + }, + } + if err := writeERPCOverlay(cfg, ov); err != nil { + t.Fatal(err) + } + info, err := os.Stat(erpcOverlayPath(cfg)) + if err != nil { + t.Fatal(err) + } + if info.Mode().Perm() != 0o600 { + t.Errorf("mode %v, want 0600", info.Mode().Perm()) + } + + got, err := readERPCOverlay(cfg) + if err != nil || got == nil { + t.Fatalf("read: %v %v", got, err) + } + if len(got.Networks) != 1 || len(got.Upstreams) != 2 { + t.Fatalf("round-trip sizes: nets=%d ups=%d", len(got.Networks), len(got.Upstreams)) + } + if got.Upstreams[0]["id"] != "local-hl-node" { + t.Errorf("upstream id = %v", got.Upstreams[0]["id"]) + } + + st, err := StatusERPC(cfg) + if err != nil || !st.Present { + t.Fatalf("status: %+v err=%v", st, err) + } + if st.UpstreamCount != 2 || st.NetworkCount != 1 || st.ContentHash == "" { + t.Errorf("status incomplete: %+v", st) + } +} + +func TestMergeERPCOverlay_Idempotent(t *testing.T) { + baseYAML := ` +logLevel: debug +database: + evmJsonRpcCache: + connectors: + - id: memory-cache + driver: memory + policies: + - network: "*" + method: "*" + finality: unfinalized + connector: memory-cache + ttl: 10s +projects: + - id: rpc + networks: + - alias: base + architecture: evm + evm: + chainId: 8453 + upstreams: + - id: obol-rpc-base + endpoint: https://example.com/base + evm: + chainId: 8453 +` + var erpcConfig map[string]any + if err := yaml.Unmarshal([]byte(baseYAML), &erpcConfig); err != nil { + t.Fatal(err) + } + + ov := &ERPCOverlay{ + Version: 1, + Networks: []map[string]any{ + { + "alias": "hyperevm", + "architecture": "evm", + "evm": map[string]any{"chainId": 999}, + "failsafe": map[string]any{"timeout": map[string]any{"duration": "10s"}}, + }, + }, + Upstreams: []map[string]any{ + { + "id": "local-hl-node", + "endpoint": "http://192.168.50.21:3001/evm", + "evm": map[string]any{"chainId": 999}, + }, + { + "id": "hyperevm-official", + "endpoint": "https://rpc.hyperliquid.xyz/evm", + "evm": map[string]any{"chainId": 999}, + "rateLimitBudget": "hyperevm-official", + }, + }, + RateLimiters: map[string]any{ + "budgets": []any{ + map[string]any{"id": "hyperevm-official", "rules": []any{}}, + }, + }, + CachePoliciesAdd: []map[string]any{ + {"network": "*", "method": "eth_call", "finality": "realtime", "ttl": "2s"}, + }, + } + + if err := mergeERPCOverlay(erpcConfig, ov); err != nil { + t.Fatal(err) + } + // Second apply must be idempotent (no duplicate ids) + if err := mergeERPCOverlay(erpcConfig, ov); err != nil { + t.Fatal(err) + } + + project := erpcConfig["projects"].([]any)[0].(map[string]any) + networks := project["networks"].([]any) + upstreams := project["upstreams"].([]any) + + if len(networks) != 2 { + t.Fatalf("networks = %d, want 2 (base + hyperevm)", len(networks)) + } + if len(upstreams) != 3 { + t.Fatalf("upstreams = %d, want 3 (base + 2 overlay)", len(upstreams)) + } + + // base preserved + ids := map[string]bool{} + for _, u := range upstreams { + ids[u.(map[string]any)["id"].(string)] = true + } + for _, want := range []string{"obol-rpc-base", "local-hl-node", "hyperevm-official"} { + if !ids[want] { + t.Errorf("missing upstream %s", want) + } + } + + // hyperevm network present + foundHyper := false + for _, n := range networks { + nm := n.(map[string]any) + if nm["alias"] == "hyperevm" && yamlInt(nm["evm"].(map[string]any)["chainId"]) == 999 { + foundHyper = true + } + } + if !foundHyper { + t.Error("hyperevm network missing") + } + + // rate limiters + rl := erpcConfig["rateLimiters"].(map[string]any) + budgets := rl["budgets"].([]any) + if len(budgets) != 1 { + t.Fatalf("budgets = %d, want 1", len(budgets)) + } + + // cache policy added once + policies := erpcConfig["database"].(map[string]any)["evmJsonRpcCache"].(map[string]any)["policies"].([]any) + if len(policies) != 2 { + t.Fatalf("cache policies = %d, want 2 (base unfinalized + realtime eth_call)", len(policies)) + } +} + +func TestMergeERPCOverlay_ReplacesByChainIDAndUpstreamID(t *testing.T) { + erpcConfig := map[string]any{ + "projects": []any{ + map[string]any{ + "id": "rpc", + "networks": []any{ + map[string]any{ + "alias": "hyperevm", + "evm": map[string]any{"chainId": 999}, + "failsafe": map[string]any{ + "timeout": map[string]any{"duration": "30s"}, + }, + }, + }, + "upstreams": []any{ + map[string]any{ + "id": "local-hl-node", + "endpoint": "http://old:3001/evm", + "evm": map[string]any{"chainId": 999}, + }, + }, + }, + }, + } + ov := &ERPCOverlay{ + Version: 1, + Networks: []map[string]any{ + { + "alias": "hyperevm", + "evm": map[string]any{"chainId": 999}, + "failsafe": map[string]any{ + "timeout": map[string]any{"duration": "10s"}, + }, + }, + }, + Upstreams: []map[string]any{ + { + "id": "local-hl-node", + "endpoint": "http://192.168.50.21:3001/evm", + "evm": map[string]any{"chainId": 999}, + }, + }, + } + if err := mergeERPCOverlay(erpcConfig, ov); err != nil { + t.Fatal(err) + } + project := erpcConfig["projects"].([]any)[0].(map[string]any) + networks := project["networks"].([]any) + upstreams := project["upstreams"].([]any) + if len(networks) != 1 || len(upstreams) != 1 { + t.Fatalf("replace must not duplicate: nets=%d ups=%d", len(networks), len(upstreams)) + } + ep := upstreams[0].(map[string]any)["endpoint"] + if ep != "http://192.168.50.21:3001/evm" { + t.Errorf("endpoint not replaced: %v", ep) + } + to := networks[0].(map[string]any)["failsafe"].(map[string]any)["timeout"].(map[string]any)["duration"] + if to != "10s" { + t.Errorf("network failsafe not replaced: %v", to) + } +} + +func TestStripERPCOverlay(t *testing.T) { + erpcConfig := map[string]any{ + "projects": []any{ + map[string]any{ + "id": "rpc", + "networks": []any{ + map[string]any{"alias": "base", "evm": map[string]any{"chainId": 8453}}, + map[string]any{"alias": "hyperevm", "evm": map[string]any{"chainId": 999}}, + }, + "upstreams": []any{ + map[string]any{"id": "obol-rpc-base", "evm": map[string]any{"chainId": 8453}}, + map[string]any{"id": "local-hl-node", "evm": map[string]any{"chainId": 999}}, + }, + }, + }, + } + ov := &ERPCOverlay{ + Version: 1, + Networks: []map[string]any{ + {"alias": "hyperevm", "evm": map[string]any{"chainId": 999}}, + }, + Upstreams: []map[string]any{ + {"id": "local-hl-node", "endpoint": "http://x", "evm": map[string]any{"chainId": 999}}, + }, + } + if err := stripERPCOverlay(erpcConfig, ov); err != nil { + t.Fatal(err) + } + project := erpcConfig["projects"].([]any)[0].(map[string]any) + if len(project["networks"].([]any)) != 1 { + t.Fatalf("networks after strip: %v", project["networks"]) + } + if len(project["upstreams"].([]any)) != 1 { + t.Fatalf("upstreams after strip: %v", project["upstreams"]) + } + if project["upstreams"].([]any)[0].(map[string]any)["id"] != "obol-rpc-base" { + t.Error("base upstream must remain") + } +} + +func TestApplyERPCOverlayFile_PersistsThenReadable(t *testing.T) { + cfg := &config.Config{ConfigDir: t.TempDir()} + src := filepath.Join(t.TempDir(), "basket.yaml") + body := []byte(` +version: 1 +networks: + - alias: hyperevm + architecture: evm + evm: + chainId: 999 +upstreams: + - id: hyperevm-official + endpoint: https://rpc.hyperliquid.xyz/evm + evm: + chainId: 999 +`) + if err := os.WriteFile(src, body, 0o600); err != nil { + t.Fatal(err) + } + // applyOverlayToCluster needs a cluster — only test parse+persist via write path + ov, err := parseERPCOverlay(body) + if err != nil { + t.Fatal(err) + } + if err := writeERPCOverlay(cfg, ov); err != nil { + t.Fatal(err) + } + got, err := readERPCOverlay(cfg) + if err != nil || got == nil || len(got.Upstreams) != 1 { + t.Fatalf("persist failed: %+v %v", got, err) + } +}