diff --git a/.devcontainer/Dockerfile b/.devcontainer/Dockerfile index b462177..db23d28 100644 --- a/.devcontainer/Dockerfile +++ b/.devcontainer/Dockerfile @@ -1,24 +1,39 @@ -FROM golang:1.25-trixie +FROM dhi.io/golang:1.26-dev # Run as root USER root +SHELL ["/bin/bash", "-c"] + # Install additional OS packages RUN apt update && apt upgrade -y && export DEBIAN_FRONTEND=noninteractive \ && apt -y install --no-install-recommends \ + bsdutils \ curl \ git \ - valkey-tools \ + gzip \ + libc-bin \ + locales \ + unzip \ + && ldconfig \ && apt clean -y \ && rm -rf /var/lib/apt/lists/* +# Generate and set the locale used by the shell +RUN sed -i '/en_US.UTF-8/s/^# //g' /etc/locale.gen \ + && locale-gen + +ENV LANG=en_US.UTF-8 +ENV LANGUAGE=en_US:en +ENV LC_ALL=en_US.UTF-8 + # Install Node.js (latest LTS) RUN cd /tmp; curl -fsSL https://deb.nodesource.com/setup_lts.x | bash - \ && apt install -y nodejs \ && npm install -g npm@latest # Install Claude Code -RUN curl -fsSL https://claude.ai/install.sh | bash \ +RUN curl -fsSL https://claude.ai/install.sh | bash -s stable \ && echo 'export PATH="$HOME/.local/bin:$PATH"' >> ~/.bashrc \ && echo 'export PATH="$HOME/.local/bin:$PATH"' >> ~/.bash_profile @@ -27,5 +42,11 @@ RUN go install golang.org/x/tools/gopls@latest \ && go install github.com/go-delve/delve/cmd/dlv@latest \ && go install github.com/air-verse/air@latest +# Install Valkey's CLI +RUN apt update \ + && (getent group adm >/dev/null || groupadd -r -g 4 adm) \ + && apt install -y --no-install-recommends valkey-tools \ + && rm -rf /var/lib/apt/lists/* + # Set up the workspace WORKDIR /workspace diff --git a/.devcontainer/devcontainer.json b/.devcontainer/devcontainer.json index 3cc4904..b70de18 100644 --- a/.devcontainer/devcontainer.json +++ b/.devcontainer/devcontainer.json @@ -14,13 +14,21 @@ "go.goroot": "/usr/local/go", "go.toolsEnvVars": { "GO111MODULE": "on" + }, + "terminal.integrated.defaultProfile.linux": "bash", + "terminal.integrated.profiles.linux": { + "bash": { + "path": "/bin/bash" + } } } } }, - "forwardPorts": [8080], + "forwardPorts": [ + 8080 + ], "remoteUser": "root", - "mounts": [ - "source=~/.claude,target=/root/.claude,type=bind,consistency=cached" - ] + "mounts": [ + "source=~/.claude,target=/root/.claude,type=bind,consistency=cached" + ] } diff --git a/.devcontainer/docker-compose.yml b/.devcontainer/docker-compose.yml index 3abb9d9..a860f1c 100644 --- a/.devcontainer/docker-compose.yml +++ b/.devcontainer/docker-compose.yml @@ -16,7 +16,7 @@ services: - "8081:8081" valkey: - image: valkey/valkey:8.1-alpine + image: dhi.io/valkey:8.1-dev ports: - "6379:6379" # If you want to enable TLS, uncomment the following line @@ -38,20 +38,20 @@ services: - valkey prometheus: - image: prom/prometheus:latest + image: dhi.io/prometheus:3.13 ports: - "9999:9090" volumes: - ./prometheus.yml:/etc/prometheus/prometheus.yml:ro command: - "--config.file=/etc/prometheus/prometheus.yml" - - "--storage.tsdb.path=/prometheus" + - "--storage.tsdb.path=/var/prometheus" - "--storage.tsdb.retention.time=24h" - "--web.enable-lifecycle" restart: unless-stopped grafana: - image: grafana/grafana:latest + image: dhi.io/grafana:13-dev ports: - "3000:3000" environment: diff --git a/.github/workflows/build-and-push-images.yaml b/.github/workflows/build-and-push-images.yaml index 1aaaf3e..eb24048 100644 --- a/.github/workflows/build-and-push-images.yaml +++ b/.github/workflows/build-and-push-images.yaml @@ -10,7 +10,7 @@ on: pull_request: types: [opened, synchronize] branches: - - '!misc/**' + - "!misc/**" paths: - "**.go" - "**/Dockerfile" @@ -47,7 +47,7 @@ jobs: id: setup_go uses: actions/setup-go@v6 with: - go-version: "1.25" + go-version: "1.26" - name: Run tests id: run_tests @@ -108,7 +108,7 @@ jobs: id: setup_go uses: actions/setup-go@v6 with: - go-version: "1.25" + go-version: "1.26" - name: Log in to GitHub Container Registry id: login_ghcr @@ -166,7 +166,7 @@ jobs: id: setup_go uses: actions/setup-go@v6 with: - go-version: "1.25" + go-version: "1.26" - name: Log in to GitHub Container Registry id: login_ghcr_lb diff --git a/.gitignore b/.gitignore index 81868ff..aadad04 100644 --- a/.gitignore +++ b/.gitignore @@ -48,3 +48,6 @@ configs/endpoints.json # Local benchmarks benchmarks/ + +# Misc +*.bak diff --git a/Makefile b/Makefile index 99c582f..35432e4 100644 --- a/Makefile +++ b/Makefile @@ -108,9 +108,23 @@ k8s-delete: kubectl delete -f k8s/health-checker.yaml kubectl delete -f k8s/namespace.yaml +# Build health checker for the local host (no GOOS/GOARCH override) +.PHONY: build-hc-native +build-hc-native: + @echo "Building Health Checker (native)..." + mkdir -p bin + env -u GOOS -u GOARCH go build -o bin/aetherlay-hc ./services/health-checker/main.go + +# Build load balancer for the local host (no GOOS/GOARCH override) +.PHONY: build-lb-native +build-lb-native: + @echo "Building RPC Load Balancer (native)..." + mkdir -p bin + env -u GOOS -u GOARCH go build -o bin/aetherlay-lb ./services/load-balancer/main.go + # Run both services in the background .PHONY: run -run: build +run: build-hc-native build-lb-native @echo "Starting both services..." ./bin/aetherlay-hc & ./bin/aetherlay-lb --metrics-port=9091 & @@ -118,13 +132,13 @@ run: build # Run health checker .PHONY: run-hc -run-hc: build-hc +run-hc: build-hc-native @echo "Running Health Checker..." ./bin/aetherlay-hc # Run load balancer .PHONY: run-lb -run-lb: build-lb +run-lb: build-lb-native @echo "Running RPC Load Balancer..." ./bin/aetherlay-lb --metrics-port=9091 @@ -167,7 +181,9 @@ help: @echo "Available targets:" @echo " build - Build both services" @echo " build-hc - Build health checker only" + @echo " build-hc-native - Build health checker for the local host (used by run/run-hc)" @echo " build-lb - Build load balancer only" + @echo " build-lb-native - Build load balancer for the local host (used by run/run-lb)" @echo " clean - Clean build artifacts" @echo " dev-setup - Set up development environment" @echo " docker-build - Build Docker images for both services" diff --git a/README.md b/README.md index c6187c0..67ec392 100644 --- a/README.md +++ b/README.md @@ -223,6 +223,15 @@ The service checks the health of an endpoint by sending these requests to it. Wh In both cases, the sync/health-status call is treated as optional: if an endpoint doesn't implement it (a JSON-RPC "method not found" error), it's assumed healthy rather than being marked down over a missing optional method. You can also skip it for a specific endpoint with `"skip_sync_check": true`. +### Failing-Method Detection + +The `eth_blockNumber`/`getSlot` and sync-status calls above only prove those specific methods work, they say nothing about a provider that's healthy overall but failing on a different method on which your traffic actually depends (e.g., `eth_getBlockByNumber` or `getBlock`). Aetherlay closes that gap automatically, with no configuration needed: + +1. **Capture**: When a proxied request to an endpoint fails with a real 5xx, Aetherlay checks whether the failed request's JSON-RPC method is on a small allowlist of methods it knows how to safely re-test on its own with a read-only, proven-to-be-valid request (currently `eth_getBlockByNumber` for `evm` chains and `getBlock` for `solana`). This is keyed off the method name only; the client's original request body is never replayed, so a captured failure can never cause Aetherlay to resubmit a state-mutating call. +2. **Targeted re-testing**: While a method is captured for an endpoint, every health check that would otherwise mark it healthy also re-tests that exact method. The endpoint isn't considered healthy again until both the regular probe and the captured method's request succeed. +3. **Stability window**: The captured method stays the target for `ephemeral-checks-healthy-threshold * ephemeral-checks-interval` seconds (plus a small fixed overhead), so an endpoint failing on every method doesn't have its target constantly overwritten before any single method can accumulate enough consecutive passes to prove recovery. +4. **Automatic reset**: Once the endpoint passes the configured consecutive-success threshold, the captured method is cleared and health checks revert to the default probe. + ### Chain Types Each endpoint can declare a `chain_type`, which selects the JSON-RPC dialect used for health checks (and the rate-limit recovery probe): diff --git a/go.mod b/go.mod index b28ba3c..4d131d2 100644 --- a/go.mod +++ b/go.mod @@ -1,6 +1,6 @@ module aetherlay -go 1.25 +go 1.26 require ( github.com/gorilla/mux v1.8.1 @@ -23,7 +23,7 @@ require ( github.com/pmezard/go-difflib v1.0.0 // indirect github.com/prometheus/common v0.65.0 // indirect github.com/prometheus/procfs v0.16.1 // indirect - golang.org/x/sys v0.34.0 // indirect + golang.org/x/sys v0.44.0 // indirect google.golang.org/protobuf v1.36.6 // indirect gopkg.in/yaml.v3 v3.0.1 // indirect ) diff --git a/go.sum b/go.sum index 898b6de..a68f0e6 100644 --- a/go.sum +++ b/go.sum @@ -60,8 +60,8 @@ golang.org/x/net v0.40.0/go.mod h1:y0hY0exeL2Pku80/zKK7tpntoX23cqL3Oa6njdgRtds= golang.org/x/sys v0.0.0-20220811171246-fbc7d0a398ab/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.12.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= -golang.org/x/sys v0.34.0 h1:H5Y5sJ2L2JRdyv7ROF1he/lPdvFsd0mJHFw2ThKHxLA= -golang.org/x/sys v0.34.0/go.mod h1:BJP2sWEmIv4KK5OTEluFJCKSidICx8ciO85XgH3Ak8k= +golang.org/x/sys v0.44.0 h1:ildZl3J4uzeKP07r2F++Op7E9B29JRUy+a27EibtBTQ= +golang.org/x/sys v0.44.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= golang.org/x/text v0.25.0 h1:qVyWApTSYLk/drJRO5mDlNYskwQznZmkpV2c8q9zls4= golang.org/x/text v0.25.0/go.mod h1:WEdwpYrmk1qmdHvhkSTNPm3app7v4rsT8F2UD6+VHIA= google.golang.org/protobuf v1.36.6 h1:z1NpPI8ku2WgiWnf+t9wTPsn6eP1L7ksHUlkfLvd9xY= diff --git a/internal/health/checker.go b/internal/health/checker.go index 64e3618..01bb3c6 100644 --- a/internal/health/checker.go +++ b/internal/health/checker.go @@ -26,6 +26,18 @@ import ( // ErrMethodNotFound indicates that the RPC method is not supported by the endpoint var ErrMethodNotFound = errors.New("method not found") +// ErrSlotSkipped indicates a Solana getBlock call failed because the requested slot has +// no block, not because the endpoint itself is unhealthy. Solana returns this same error +// code both when a slot was genuinely skipped (a routine, expected occurrence) and when +// the endpoint has pruned it from long-term storage, so it is deliberately not treated as +// equivalent to success; see its use in checkHTTPHealth's custom probe re-test. +var ErrSlotSkipped = errors.New("solana: slot skipped or missing from history") + +// solanaSlotSkippedCodes are the JSON-RPC error codes Solana returns for getBlock when +// the requested slot has no block: -32007 ("skipped, or missing due to ledger jump to +// recent snapshot") and -32009 ("skipped, or missing in long-term storage"). +var solanaSlotSkippedCodes = map[int]bool{-32007: true, -32009: true} + // JSON-RPC methods used for health checks, keyed by chain type. Block/slot methods are // always required; sync/health methods are only called when sync-status checking is // enabled and are tolerated as "not found" (see optionalHealthCheckMethods below). @@ -84,6 +96,17 @@ func checkRPCError(response *RpcResponse, method, protocol, chain, endpointID, u return nil } + if method == "getBlock" && solanaSlotSkippedCodes[response.Error.Code] { + log.Debug(). + Str("chain", chain). + Str("endpoint", helpers.RedactAPIKey(url)). + Str("endpoint_id", endpointID). + Int("error_code", response.Error.Code). + Str("error_message", response.Error.Message). + Msg("getBlock reported the requested slot as skipped or unavailable") + return ErrSlotSkipped + } + // Check for "method not found" errors methodNotFound := response.Error.Code == -32601 || containsMethodNotFound(response.Error.Message) @@ -134,6 +157,14 @@ type Checker struct { ephemeralChecksInterval time.Duration ephemeralChecksThreshold int + // statusMu serializes the read-modify-write status update sequence per endpoint (key: + // chain+":"+endpointID, value: *sync.Mutex), since checkEndpoint runs the HTTP and WS + // checks for the same endpoint concurrently and both can independently persist status + // via updateEndpointStatusInValkey. Without this, two concurrent get-then-put cycles on + // the same Valkey key can interleave, with the later write silently reverting the field + // the other one had just set. + statusMu sync.Map + // Rate limit handler function provided by server HandleRateLimitFunc func(chain, endpointID, protocol string, signal RateLimitSignal) @@ -318,16 +349,25 @@ func (c *Checker) runEphemeralCheckProtocol(ctx context.Context, chain, endpoint log.Debug().Str("chain", chain).Str("endpoint_id", endpointID).Str("protocol", protocol).Int("consecutive", consecutive).Msg("Ephemeral check: success") if consecutive >= threshold { log.Info().Str("chain", chain).Str("endpoint_id", endpointID).Str("protocol", protocol).Msg("Ephemeral check: protocol considered healthy again") - // Mark protocol healthy in Valkey - status, err := c.valkeyClient.GetEndpointStatus(ctx, chain, endpointID) - if err == nil { + // Mark protocol healthy in Valkey. Routed through updateEndpointStatusInValkey + // (the same locked read-modify-write path checkHTTPHealth/checkWSHealth/ + // checkEndpoint use) rather than a standalone get-then-put, since this can run + // concurrently with a periodic sweep checking the same endpoint. + c.updateEndpointStatusInValkey(ctx, chain, endpointID, func(status *store.EndpointStatus) { switch protocol { case "http": status.HealthyHTTP = true case "ws": status.HealthyWS = true } - c.updateStatus(ctx, chain, endpointID, *status) + }) + // Recovery confirmed via the same threshold used above, so any custom + // probe method targeted at this endpoint (see custom_probe.go) has now + // also passed that many times in a row, revert to the default probe. + if protocol == "http" { + if err := c.valkeyClient.ClearCustomProbeState(ctx, chain, endpointID); err != nil { + log.Error().Err(err).Str("chain", chain).Str("endpoint_id", endpointID).Msg("Failed to clear custom probe state") + } } // Remove from ephemeralChecks if state, ok := c.ephemeralChecks[key]; ok { @@ -402,9 +442,6 @@ func (c *Checker) checkEndpoint(ctx context.Context, chain, endpointID string, e return } - status := store.NewEndpointStatus() - status.LastHealthCheck = time.Now() - // Create channels to collect results from parallel health checks httpResult := make(chan bool, 1) wsResult := make(chan bool, 1) @@ -422,26 +459,59 @@ func (c *Checker) checkEndpoint(ctx context.Context, chain, endpointID string, e }() // Collect results - status.HasHTTP = endpoint.HTTPURL != "" - status.HasWS = endpoint.WSURL != "" - status.HealthyHTTP = <-httpResult - status.HealthyWS = <-wsResult - - // Get current request counts - r24h, r1m, rAll, err := c.valkeyClient.GetCombinedRequestCounts(ctx, chain, endpointID) - if err == nil { - status.Requests24h = r24h - status.Requests1Month = r1m - status.RequestsLifetime = rAll - } + httpHealthy := <-httpResult + wsHealthy := <-wsResult + + // Persist through updateEndpointStatusInValkey, the same locked read-modify-write path + // checkHTTPHealth/checkWSHealth just used above for their own per-protocol writes, + // instead of a raw overwrite. checkHTTPHealth/checkWSHealth may have independently + // persisted fields this function never learns about (e.g. BlockNumber); a raw overwrite + // here would silently erase those, and racing the read-modify-write cycles above would + // let this write revert whichever field the other finished last. + c.updateEndpointStatusInValkey(ctx, chain, endpointID, func(status *store.EndpointStatus) { + // wasHealthyHTTP/wasHealthyWS and hasPriorCheckHTTP/hasPriorCheckWS are read here, + // under the same lock this closure runs in, rather than before the probes above ran. + // The probes can take seconds; if a concurrent runEphemeralCheckProtocol confirmed + // recovery during that window, reading these values any earlier would resolve + // against a stale, already-superseded status and could revert that confirmed + // recovery back to unhealthy. + wasHealthyHTTP, wasHealthyWS := status.HealthyHTTP, status.HealthyWS + hasPriorCheckHTTP, hasPriorCheckWS := !status.LastHTTPHealthCheck.IsZero(), !status.LastWSHealthCheck.IsZero() - // Update status in Valkey - c.updateStatus(ctx, chain, endpointID, status) + status.HasHTTP = endpoint.HTTPURL != "" + status.HasWS = endpoint.WSURL != "" + // Only record a check timestamp for a protocol the endpoint actually has; otherwise + // an HTTP-only endpoint would end up with a LastWSHealthCheck timestamp despite + // checkWSHealth never having run a real probe for it (it returns early instead). + checkedAt := time.Now() + if status.HasHTTP { + status.LastHTTPHealthCheck = checkedAt + } + if status.HasWS { + status.LastWSHealthCheck = checkedAt + } + status.HealthyHTTP = c.resolveHealthTransition(hasPriorCheckHTTP, wasHealthyHTTP, httpHealthy) + status.HealthyWS = c.resolveHealthTransition(hasPriorCheckWS, wasHealthyWS, wsHealthy) + }) } -// makeRPCCall makes a single JSON-RPC call and returns the result +// makeRPCCall makes a single JSON-RPC call with empty params and returns the result func (c *Checker) makeRPCCall(ctx context.Context, url, method, chain, endpointID, provider string) (any, error) { - payload := []byte(`{"jsonrpc":"2.0","method":"` + method + `","params":[],"id":1}`) + return c.makeRPCCallWithParams(ctx, url, method, []any{}, chain, endpointID, provider) +} + +// makeRPCCallWithParams makes a single JSON-RPC call with the given params and returns +// the result. +func (c *Checker) makeRPCCallWithParams(ctx context.Context, url, method string, params []any, chain, endpointID, provider string) (any, error) { + payload, err := json.Marshal(map[string]any{ + "jsonrpc": "2.0", + "method": method, + "params": params, + "id": 1, + }) + if err != nil { + return nil, err + } req, err := http.NewRequestWithContext(ctx, "POST", url, bytes.NewBuffer(payload)) if err != nil { return nil, err @@ -757,8 +827,39 @@ func (c *Checker) incrementHealthRequestCount(ctx context.Context, chain, endpoi } } +// resolveHealthTransition decides whether a fresh probe result should overwrite the +// currently stored health status for a protocol. Failures are always applied +// immediately so a bad endpoint is ejected fast. The unhealthy to healthy transition is +// left to the ephemeral checker (see runEphemeralCheckProtocol), which requires several +// consecutive successful probes rather than accepting a single passing periodic check. +// Without this, a lucky, shallow probe on the main sweep can silently erase a failure +// surfaced by real production traffic (see the passive tracking in server.go) or by a +// prior periodic check. hasPriorCheck should be false only for an endpoint's very first +// ever check (no baseline to protect yet), so a fresh endpoint can still become healthy +// immediately at startup instead of waiting on the ephemeral checker's threshold. +func (c *Checker) resolveHealthTransition(hasPriorCheck, currentlyHealthy, probeHealthy bool) bool { + if !c.ephemeralChecksEnabled || !hasPriorCheck { + return probeHealthy // no other recovery path exists, preserve old behavior + } + if probeHealthy && !currentlyHealthy { + return false // stay unhealthy, the ephemeral checker owns recovery + } + return probeHealthy +} + +// statusLockFor returns the mutex guarding status read-modify-write cycles for a single +// endpoint, creating it on first use. +func (c *Checker) statusLockFor(chain, endpointID string) *sync.Mutex { + mu, _ := c.statusMu.LoadOrStore(chain+":"+endpointID, &sync.Mutex{}) + return mu.(*sync.Mutex) +} + // updateEndpointStatusInValkey fetches current status, updates it with new values, and stores it in Valkey func (c *Checker) updateEndpointStatusInValkey(ctx context.Context, chain, endpointID string, updateFn func(*store.EndpointStatus)) { + mu := c.statusLockFor(chain, endpointID) + mu.Lock() + defer mu.Unlock() + status, err := c.valkeyClient.GetEndpointStatus(ctx, chain, endpointID) if err != nil || status == nil { st := store.NewEndpointStatus() @@ -806,28 +907,56 @@ func (c *Checker) checkHTTPHealth(ctx context.Context, chain, endpointID string, c.incrementHealthRequestCount(ctx, chain, endpointID) } - // If the block/slot call failed, the endpoint is unhealthy - if blockErr != nil { - c.updateHealthMetrics(chain, endpointID, false) - return false - } - - // If sync status checking is enabled and the sync/health call failed (but not due to method not found), the endpoint is unhealthy - if c.healthCheckSyncStatus && syncErr != nil && !errors.Is(syncErr, ErrMethodNotFound) { - c.updateHealthMetrics(chain, endpointID, false) - return false + // A hard failure on either call means the endpoint is unhealthy. This used to return + // early here without ever persisting anything to Valkey, so an endpoint stuck + // failing this way could keep whatever stale HealthyHTTP value was already stored + // indefinitely. It now falls through to the same write path as every other outcome, + // same as a checkHealthParams failure would. + blockCallFailed := blockErr != nil + syncCallFailed := c.healthCheckSyncStatus && syncErr != nil && !errors.Is(syncErr, ErrMethodNotFound) + + var healthy bool + var blockNumber int64 + if !blockCallFailed && !syncCallFailed { + // Check all health parameters + healthy, blockNumber = c.checkHealthParams(chain, endpointID, endpoint.HTTPURL, "HTTP", endpoint.ChainType, syncResult, blockResult) + + // If a real proxied request recently failed on one of the allowlisted methods + // (see custom_probe.go and server.go's maybeSetCustomProbeMethod), additionally + // re-test that exact method with Aetherlay's own canned request. getSlot/getHealth + // passing says nothing about a failure isolated to a different method (e.g. + // getBlock); this check must also pass for the endpoint to be considered healthy. + if healthy { + if probeState, err := c.valkeyClient.GetCustomProbeState(ctx, chain, endpointID); err == nil && probeState != nil { + if build, ok := customProbeBuilderFor(probeState.Method, endpoint.ChainType); ok { + method, params := build(blockNumber) + if _, callErr := c.makeRPCCallWithParams(ctx, endpoint.HTTPURL, method, params, chain, endpointID, endpoint.Provider); callErr != nil { + if errors.Is(callErr, ErrSlotSkipped) { + // The target slot itself had no block; this says nothing about + // whether the endpoint can serve getBlock, so it's left out of the + // healthy determination rather than counted as a failure. + log.Debug().Str("chain", chain).Str("endpoint_id", endpointID).Str("method", method).Msg("Custom probe target slot skipped, treating as inconclusive rather than a failure") + } else { + healthy = false + log.Warn().Str("chain", chain).Str("endpoint_id", endpointID).Str("method", method).Err(callErr).Msg("Custom probe re-test failed, endpoint still considered unhealthy for this method") + } + } + c.incrementHealthRequestCount(ctx, chain, endpointID) + } + } + } } - // Check all health parameters - healthy, blockNumber := c.checkHealthParams(chain, endpointID, endpoint.HTTPURL, "HTTP", endpoint.ChainType, syncResult, blockResult) - // Update metrics and status in Valkey c.updateHealthMetrics(chain, endpointID, healthy) c.updateEndpointStatusInValkey(ctx, chain, endpointID, func(status *store.EndpointStatus) { - status.BlockNumber = blockNumber // Store the block number for future reference + hasPriorCheck := !status.LastHTTPHealthCheck.IsZero() + if !blockCallFailed { + status.BlockNumber = blockNumber // Store the block number for future reference; keep the last known value on a failed call + } status.HasHTTP = endpoint.HTTPURL != "" - status.HealthyHTTP = healthy - status.LastHealthCheck = time.Now() + status.HealthyHTTP = c.resolveHealthTransition(hasPriorCheck, status.HealthyHTTP, healthy) + status.LastHTTPHealthCheck = time.Now() }) return healthy } @@ -858,28 +987,30 @@ func (c *Checker) checkWSHealth(ctx context.Context, chain, endpointID string, e c.incrementHealthRequestCount(ctx, chain, endpointID) } - // If the block/slot call failed, the endpoint is unhealthy - if blockErr != nil { - c.updateHealthMetrics(chain, endpointID, false) - return false + // A hard failure on either call means the endpoint is unhealthy. This used to return + // early here without ever persisting anything to Valkey, so an endpoint stuck + // failing this way could keep whatever stale HealthyWS value was already stored + // indefinitely. It now falls through to the same write path as every other outcome, + // same as a checkHealthParams failure would. + blockCallFailed := blockErr != nil + syncCallFailed := c.healthCheckSyncStatus && syncErr != nil && !errors.Is(syncErr, ErrMethodNotFound) + + var healthy bool + var blockNumber int64 + if !blockCallFailed && !syncCallFailed { + healthy, blockNumber = c.checkHealthParams(chain, endpointID, endpoint.WSURL, "WS", endpoint.ChainType, syncResult, blockResult) } - // If sync status checking is enabled and the sync/health call failed (but not due to method not found), the endpoint is unhealthy - if c.healthCheckSyncStatus && syncErr != nil && !errors.Is(syncErr, ErrMethodNotFound) { - c.updateHealthMetrics(chain, endpointID, false) - return false - } - - // Check all health parameters - healthy, blockNumber := c.checkHealthParams(chain, endpointID, endpoint.WSURL, "WS", endpoint.ChainType, syncResult, blockResult) - // Update metrics and status in Valkey c.updateHealthMetrics(chain, endpointID, healthy) c.updateEndpointStatusInValkey(ctx, chain, endpointID, func(status *store.EndpointStatus) { - status.BlockNumber = blockNumber // Store the block number for future reference + hasPriorCheck := !status.LastWSHealthCheck.IsZero() + if !blockCallFailed { + status.BlockNumber = blockNumber // Store the block number for future reference; keep the last known value on a failed call + } status.HasWS = endpoint.WSURL != "" - status.HealthyWS = healthy - status.LastHealthCheck = time.Now() + status.HealthyWS = c.resolveHealthTransition(hasPriorCheck, status.HealthyWS, healthy) + status.LastWSHealthCheck = time.Now() }) return healthy } diff --git a/internal/health/checker_guard_test.go b/internal/health/checker_guard_test.go new file mode 100644 index 0000000..f82569a --- /dev/null +++ b/internal/health/checker_guard_test.go @@ -0,0 +1,482 @@ +package health + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "aetherlay/internal/config" + "aetherlay/internal/store" + + "github.com/gorilla/websocket" +) + +// TestResolveHealthTransition covers every (hasPriorCheck, currentlyHealthy, probeHealthy) +// combination, including the ephemeral-checks-disabled fallback. +func TestResolveHealthTransition(t *testing.T) { + tests := []struct { + name string + ephemeralChecksEnabled bool + hasPriorCheck bool + currentlyHealthy bool + probeHealthy bool + want bool + }{ + {"first ever check, probe healthy", true, false, false, true, true}, + {"first ever check, probe unhealthy", true, false, false, false, false}, + {"prior check, was healthy, probe healthy", true, true, true, true, true}, + {"prior check, was healthy, probe unhealthy applies immediately", true, true, true, false, false}, + {"prior check, was unhealthy, probe healthy stays unhealthy", true, true, false, true, false}, + {"prior check, was unhealthy, probe unhealthy", true, true, false, false, false}, + {"ephemeral disabled, was unhealthy, probe healthy flips immediately", false, true, false, true, true}, + {"ephemeral disabled, was healthy, probe unhealthy flips immediately", false, true, true, false, false}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + c := &Checker{ephemeralChecksEnabled: tt.ephemeralChecksEnabled} + got := c.resolveHealthTransition(tt.hasPriorCheck, tt.currentlyHealthy, tt.probeHealthy) + if got != tt.want { + t.Errorf("resolveHealthTransition(%v, %v, %v) = %v, want %v", tt.hasPriorCheck, tt.currentlyHealthy, tt.probeHealthy, got, tt.want) + } + }) + } +} + +// TestCheckHTTPHealthGuardKeepsUnhealthyOnPassingProbe verifies that a single passing +// periodic sweep does not flip a previously-unhealthy endpoint back to healthy. +func TestCheckHTTPHealthGuardKeepsUnhealthyOnPassingProbe(t *testing.T) { + server := solanaRPCTestServer(t, 123456, true) + defer server.Close() + + valkeyClient := store.NewMockValkeyClient() + valkeyClient.PopulateStatuses(map[string]*store.EndpointStatus{ + "solana-mainnet:test-1": {HasHTTP: true, HealthyHTTP: false, LastHTTPHealthCheck: time.Now().Add(-time.Minute)}, + }) + checker := &Checker{ + valkeyClient: valkeyClient, + healthCheckSyncStatus: true, + ephemeralChecksEnabled: true, + } + endpoint := config.Endpoint{Provider: "test", ChainType: config.ChainTypeSolana, HTTPURL: server.URL} + + if healthy := checker.checkHTTPHealth(context.Background(), "solana-mainnet", "test-1", endpoint); !healthy { + t.Error("expected the probe itself to report healthy") + } + + status, err := valkeyClient.GetEndpointStatus(context.Background(), "solana-mainnet", "test-1") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if status.HealthyHTTP { + t.Error("expected the periodic sweep NOT to flip a previously-unhealthy endpoint back to healthy on a single passing probe; that's the ephemeral checker's job") + } +} + +// TestCheckHTTPHealthGuardFlipsToUnhealthyImmediately verifies that a failing probe ejects +// a previously-healthy endpoint right away, with no debounce. +func TestCheckHTTPHealthGuardFlipsToUnhealthyImmediately(t *testing.T) { + // Slot 0 is an invalid/unhealthy result parsed by checkHealthParams itself, not a + // hard RPC-level error on the block/sync call (see + // TestCheckHTTPHealthPersistsUnhealthyOnHardSyncCallError below for that path). + server := solanaRPCTestServer(t, 0, true) + defer server.Close() + + valkeyClient := store.NewMockValkeyClient() + valkeyClient.PopulateStatuses(map[string]*store.EndpointStatus{ + "solana-mainnet:test-1": {HasHTTP: true, HealthyHTTP: true, LastHTTPHealthCheck: time.Now().Add(-time.Minute)}, + }) + checker := &Checker{ + valkeyClient: valkeyClient, + healthCheckSyncStatus: true, + ephemeralChecksEnabled: true, + } + endpoint := config.Endpoint{Provider: "test", ChainType: config.ChainTypeSolana, HTTPURL: server.URL} + + checker.checkHTTPHealth(context.Background(), "solana-mainnet", "test-1", endpoint) + + status, err := valkeyClient.GetEndpointStatus(context.Background(), "solana-mainnet", "test-1") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if status.HealthyHTTP { + t.Error("expected a failing probe to eject a previously-healthy endpoint immediately, with no debounce") + } +} + +// TestCheckHTTPHealthPersistsUnhealthyOnHardBlockCallError covers a hard RPC error on the +// block/slot call itself (a real 5xx from the endpoint, not just an invalid result), which +// used to return before ever calling updateEndpointStatusInValkey, leaving a stale +// HealthyHTTP value in Valkey indefinitely if the endpoint kept failing this way. +func TestCheckHTTPHealthPersistsUnhealthyOnHardBlockCallError(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusInternalServerError) + })) + defer server.Close() + + valkeyClient := store.NewMockValkeyClient() + valkeyClient.PopulateStatuses(map[string]*store.EndpointStatus{ + "solana-mainnet:test-1": {HasHTTP: true, HealthyHTTP: true, BlockNumber: 42, LastHTTPHealthCheck: time.Now().Add(-time.Minute)}, + }) + checker := &Checker{ + valkeyClient: valkeyClient, + healthCheckSyncStatus: true, + ephemeralChecksEnabled: true, + } + endpoint := config.Endpoint{Provider: "test", ChainType: config.ChainTypeSolana, HTTPURL: server.URL} + + if healthy := checker.checkHTTPHealth(context.Background(), "solana-mainnet", "test-1", endpoint); healthy { + t.Error("expected a hard error on the block call to report unhealthy") + } + + status, err := valkeyClient.GetEndpointStatus(context.Background(), "solana-mainnet", "test-1") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if status.HealthyHTTP { + t.Error("expected a hard error on the block call to persist HealthyHTTP=false, not leave the stale prior value in place") + } + if status.BlockNumber != 42 { + t.Errorf("expected the last known block number to be preserved when the block call itself fails, got %d", status.BlockNumber) + } +} + +// TestCheckHTTPHealthPersistsUnhealthyOnHardSyncCallError is the same as above but for a +// hard (non-method-not-found) JSON-RPC error on the sync/health-status call. +func TestCheckHTTPHealthPersistsUnhealthyOnHardSyncCallError(t *testing.T) { + server := solanaRPCTestServer(t, 123456, false) // getSlot ok, getHealth returns a real JSON-RPC error + defer server.Close() + + valkeyClient := store.NewMockValkeyClient() + valkeyClient.PopulateStatuses(map[string]*store.EndpointStatus{ + "solana-mainnet:test-1": {HasHTTP: true, HealthyHTTP: true, LastHTTPHealthCheck: time.Now().Add(-time.Minute)}, + }) + checker := &Checker{ + valkeyClient: valkeyClient, + healthCheckSyncStatus: true, + ephemeralChecksEnabled: true, + } + endpoint := config.Endpoint{Provider: "test", ChainType: config.ChainTypeSolana, HTTPURL: server.URL} + + if healthy := checker.checkHTTPHealth(context.Background(), "solana-mainnet", "test-1", endpoint); healthy { + t.Error("expected a hard error on the sync call to report unhealthy") + } + + status, err := valkeyClient.GetEndpointStatus(context.Background(), "solana-mainnet", "test-1") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if status.HealthyHTTP { + t.Error("expected a hard error on the sync call to persist HealthyHTTP=false, not leave the stale prior value in place") + } +} + +// TestCheckHTTPHealthGuardFallbackWhenEphemeralDisabled verifies that the old +// unconditional-overwrite behavior is preserved when ephemeral checks are disabled, since +// there's no other recovery path in that case. +func TestCheckHTTPHealthGuardFallbackWhenEphemeralDisabled(t *testing.T) { + server := solanaRPCTestServer(t, 123456, true) + defer server.Close() + + valkeyClient := store.NewMockValkeyClient() + valkeyClient.PopulateStatuses(map[string]*store.EndpointStatus{ + "solana-mainnet:test-1": {HasHTTP: true, HealthyHTTP: false, LastHTTPHealthCheck: time.Now().Add(-time.Minute)}, + }) + checker := &Checker{ + valkeyClient: valkeyClient, + healthCheckSyncStatus: true, + ephemeralChecksEnabled: false, + } + endpoint := config.Endpoint{Provider: "test", ChainType: config.ChainTypeSolana, HTTPURL: server.URL} + + checker.checkHTTPHealth(context.Background(), "solana-mainnet", "test-1", endpoint) + + status, err := valkeyClient.GetEndpointStatus(context.Background(), "solana-mainnet", "test-1") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if !status.HealthyHTTP { + t.Error("expected the old unconditional-overwrite behavior to be preserved when ephemeral checks are disabled, since there's no other recovery path") + } +} + +// TestCheckHTTPHealthFirstEverCheckBecomesHealthyImmediately verifies that a brand new +// endpoint's very first check can become healthy right away, without waiting on the +// ephemeral checker. +func TestCheckHTTPHealthFirstEverCheckBecomesHealthyImmediately(t *testing.T) { + server := solanaRPCTestServer(t, 123456, true) + defer server.Close() + + // No PopulateStatuses call: this endpoint has never been checked before. + valkeyClient := store.NewMockValkeyClient() + checker := &Checker{ + valkeyClient: valkeyClient, + healthCheckSyncStatus: true, + ephemeralChecksEnabled: true, + } + endpoint := config.Endpoint{Provider: "test", ChainType: config.ChainTypeSolana, HTTPURL: server.URL} + + checker.checkHTTPHealth(context.Background(), "solana-mainnet", "test-1", endpoint) + + status, err := valkeyClient.GetEndpointStatus(context.Background(), "solana-mainnet", "test-1") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if !status.HealthyHTTP { + t.Error("expected a brand new endpoint's very first check to be able to become healthy immediately, without waiting on the ephemeral checker") + } +} + +// TestCheckEndpointGuardKeepsUnhealthyOnPassingProbe verifies that checkEndpoint's own +// write site respects the same guard as checkHTTPHealth, instead of re-introducing the raw +// unguarded probe result. +func TestCheckEndpointGuardKeepsUnhealthyOnPassingProbe(t *testing.T) { + valkeyClient := store.NewMockValkeyClient() + valkeyClient.PopulateStatuses(map[string]*store.EndpointStatus{ + "solana-mainnet:test-1": {HasHTTP: true, HealthyHTTP: false, LastHTTPHealthCheck: time.Now().Add(-time.Minute)}, + }) + checker := &Checker{ + valkeyClient: valkeyClient, + ephemeralChecksEnabled: true, + } + checker.CheckHTTPHealthFunc = func(_ context.Context, _, _ string, _ config.Endpoint) bool { return true } + checker.CheckWSHealthFunc = func(_ context.Context, _, _ string, _ config.Endpoint) bool { return false } + + endpoint := config.Endpoint{Provider: "test", ChainType: config.ChainTypeSolana, HTTPURL: "http://example.invalid"} + checker.checkEndpoint(context.Background(), "solana-mainnet", "test-1", endpoint) + + status, err := valkeyClient.GetEndpointStatus(context.Background(), "solana-mainnet", "test-1") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if status.HealthyHTTP { + t.Error("expected checkEndpoint's own write to respect the same guard as checkHTTPHealth, not silently re-introduce the raw unguarded probe result") + } +} + +// solanaGetBlockTestServer extends the getSlot/getHealth fixture with a getBlock handler, +// so custom-probe consumption can be tested against a fully successful round. +func solanaGetBlockTestServer(t *testing.T, slot int64) *httptest.Server { + t.Helper() + return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + var req map[string]any + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + t.Errorf("failed to decode request: %v", err) + return + } + method, _ := req["method"].(string) + + w.Header().Set("Content-Type", "application/json") + switch method { + case "getSlot": + json.NewEncoder(w).Encode(map[string]any{"jsonrpc": "2.0", "id": 1, "result": slot}) + case "getHealth": + json.NewEncoder(w).Encode(map[string]any{"jsonrpc": "2.0", "id": 1, "result": "ok"}) + case "getBlock": + json.NewEncoder(w).Encode(map[string]any{"jsonrpc": "2.0", "id": 1, "result": map[string]any{"blockHeight": slot}}) + default: + json.NewEncoder(w).Encode(map[string]any{"jsonrpc": "2.0", "id": 1, "error": map[string]any{"code": -32601, "message": "Method not found"}}) + } + })) +} + +// TestCheckHTTPHealthCustomProbeFailureOverridesOtherwiseHealthy verifies that a failing +// custom probe re-test overrides an otherwise-healthy getSlot/getHealth result. +func TestCheckHTTPHealthCustomProbeFailureOverridesOtherwiseHealthy(t *testing.T) { + // solanaRPCTestServer only understands getSlot/getHealth; any custom probe method + // (like getBlock) hits its default "method not found" branch, which is a real + // failure since getBlock isn't in the optional-methods leniency list. + server := solanaRPCTestServer(t, 123456, true) + defer server.Close() + + valkeyClient := store.NewMockValkeyClient() + if err := valkeyClient.SetCustomProbeState(context.Background(), "solana-mainnet", "test-1", store.CustomProbeState{ + Method: "getBlock", + SetAt: time.Now(), + }); err != nil { + t.Fatalf("failed to seed custom probe state: %v", err) + } + + checker := &Checker{valkeyClient: valkeyClient, healthCheckSyncStatus: true} + endpoint := config.Endpoint{Provider: "test", ChainType: config.ChainTypeSolana, HTTPURL: server.URL} + + if healthy := checker.checkHTTPHealth(context.Background(), "solana-mainnet", "test-1", endpoint); healthy { + t.Error("expected a failing custom probe re-test to override an otherwise-healthy getSlot/getHealth result") + } +} + +// TestCheckHTTPHealthCustomProbeSuccessKeepsHealthy verifies that a passing custom probe +// re-test alongside a healthy default probe reports healthy overall. +func TestCheckHTTPHealthCustomProbeSuccessKeepsHealthy(t *testing.T) { + server := solanaGetBlockTestServer(t, 123456) + defer server.Close() + + valkeyClient := store.NewMockValkeyClient() + if err := valkeyClient.SetCustomProbeState(context.Background(), "solana-mainnet", "test-1", store.CustomProbeState{ + Method: "getBlock", + SetAt: time.Now(), + }); err != nil { + t.Fatalf("failed to seed custom probe state: %v", err) + } + + checker := &Checker{valkeyClient: valkeyClient, healthCheckSyncStatus: true} + endpoint := config.Endpoint{Provider: "test", ChainType: config.ChainTypeSolana, HTTPURL: server.URL} + + if healthy := checker.checkHTTPHealth(context.Background(), "solana-mainnet", "test-1", endpoint); !healthy { + t.Error("expected a passing custom probe re-test alongside a healthy default probe to report healthy") + } +} + +// TestCheckHTTPHealthCustomProbeSlotSkippedTreatedAsInconclusive verifies that a getBlock +// re-test failing with a skipped-slot error code (-32007/-32009) does not flip an +// otherwise-healthy endpoint to unhealthy, since that indicates the target slot has no +// block rather than that the endpoint can't serve getBlock. +func TestCheckHTTPHealthCustomProbeSlotSkippedTreatedAsInconclusive(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + var req map[string]any + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + t.Errorf("failed to decode request: %v", err) + return + } + method, _ := req["method"].(string) + + w.Header().Set("Content-Type", "application/json") + switch method { + case "getSlot": + json.NewEncoder(w).Encode(map[string]any{"jsonrpc": "2.0", "id": 1, "result": 123456}) + case "getHealth": + json.NewEncoder(w).Encode(map[string]any{"jsonrpc": "2.0", "id": 1, "result": "ok"}) + case "getBlock": + json.NewEncoder(w).Encode(map[string]any{"jsonrpc": "2.0", "id": 1, "error": map[string]any{"code": -32009, "message": "Slot 123424 was skipped, or missing in long-term storage"}}) + default: + json.NewEncoder(w).Encode(map[string]any{"jsonrpc": "2.0", "id": 1, "error": map[string]any{"code": -32601, "message": "Method not found"}}) + } + })) + defer server.Close() + + valkeyClient := store.NewMockValkeyClient() + if err := valkeyClient.SetCustomProbeState(context.Background(), "solana-mainnet", "test-1", store.CustomProbeState{ + Method: "getBlock", + SetAt: time.Now(), + }); err != nil { + t.Fatalf("failed to seed custom probe state: %v", err) + } + + checker := &Checker{valkeyClient: valkeyClient, healthCheckSyncStatus: true} + endpoint := config.Endpoint{Provider: "test", ChainType: config.ChainTypeSolana, HTTPURL: server.URL} + + if healthy := checker.checkHTTPHealth(context.Background(), "solana-mainnet", "test-1", endpoint); !healthy { + t.Error("expected a skipped-slot custom probe response to be treated as inconclusive, not a failure") + } +} + +// TestRunEphemeralCheckProtocolClearsCustomProbeStateOnRecovery verifies that reaching the +// ephemeral recovery threshold clears any active custom probe state. +func TestRunEphemeralCheckProtocolClearsCustomProbeStateOnRecovery(t *testing.T) { + valkeyClient := store.NewMockValkeyClient() + valkeyClient.PopulateStatuses(map[string]*store.EndpointStatus{ + "solana-devnet:ep1": {HasHTTP: true, HealthyHTTP: false}, + }) + if err := valkeyClient.SetCustomProbeState(context.Background(), "solana-devnet", "ep1", store.CustomProbeState{ + Method: "getBlock", + SetAt: time.Now(), + }); err != nil { + t.Fatalf("failed to seed custom probe state: %v", err) + } + + checker := &Checker{valkeyClient: valkeyClient} + checker.CheckHTTPHealthFunc = func(_ context.Context, _, _ string, _ config.Endpoint) bool { return true } + + checker.runEphemeralCheckProtocol(context.Background(), "solana-devnet", "ep1", config.Endpoint{}, time.Millisecond, 1, "solana-devnet|ep1|http", "http") + + status, err := valkeyClient.GetEndpointStatus(context.Background(), "solana-devnet", "ep1") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if !status.HealthyHTTP { + t.Error("expected the endpoint to be marked healthy after reaching the ephemeral threshold") + } + + state, err := valkeyClient.GetCustomProbeState(context.Background(), "solana-devnet", "ep1") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if state != nil { + t.Errorf("expected custom probe state to be cleared on confirmed recovery, got %+v", state) + } +} + +// solanaWSTestServer mirrors solanaRPCTestServer but over a WebSocket connection, so +// checkWSHealth can be exercised end-to-end. +func solanaWSTestServer(t *testing.T, slot int64, healthy bool) *httptest.Server { + t.Helper() + upgrader := websocket.Upgrader{} + return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + conn, err := upgrader.Upgrade(w, r, nil) + if err != nil { + t.Errorf("failed to upgrade connection: %v", err) + return + } + defer conn.Close() + + var req map[string]any + if err := conn.ReadJSON(&req); err != nil { + return + } + method, _ := req["method"].(string) + + var resp map[string]any + switch method { + case "getSlot": + resp = map[string]any{"jsonrpc": "2.0", "id": 1, "result": slot} + case "getHealth": + if healthy { + resp = map[string]any{"jsonrpc": "2.0", "id": 1, "result": "ok"} + } else { + resp = map[string]any{"jsonrpc": "2.0", "id": 1, "error": map[string]any{"code": -32005, "message": "Node is unhealthy"}} + } + default: + resp = map[string]any{"jsonrpc": "2.0", "id": 1, "error": map[string]any{"code": -32601, "message": "Method not found"}} + } + conn.WriteJSON(resp) + })) +} + +// TestCheckWSHealthTracksPriorCheckSeparatelyFromHTTP is a regression guard: HTTP and WS +// prior-check state must not share a single timestamp. StartEphemeralChecks' own startup +// sweep checks HTTP before WS for a given endpoint; if both protocols shared one marker, +// WS's own first-ever check would look like a prior observation (because HTTP had just +// set it) and get stuck unhealthy on a passing probe instead of being accepted right away. +func TestCheckWSHealthTracksPriorCheckSeparatelyFromHTTP(t *testing.T) { + server := solanaWSTestServer(t, 123456, true) + defer server.Close() + wsURL := "ws" + strings.TrimPrefix(server.URL, "http") + + valkeyClient := store.NewMockValkeyClient() + // HTTP has already been checked (LastHTTPHealthCheck set); WS never has. + valkeyClient.PopulateStatuses(map[string]*store.EndpointStatus{ + "solana-mainnet:test-1": {HasWS: true, HealthyWS: false, LastHTTPHealthCheck: time.Now()}, + }) + checker := &Checker{ + valkeyClient: valkeyClient, + healthCheckSyncStatus: true, + ephemeralChecksEnabled: true, + } + endpoint := config.Endpoint{Provider: "test", ChainType: config.ChainTypeSolana, WSURL: wsURL} + + if healthy := checker.checkWSHealth(context.Background(), "solana-mainnet", "test-1", endpoint); !healthy { + t.Error("expected the probe itself to report healthy") + } + + status, err := valkeyClient.GetEndpointStatus(context.Background(), "solana-mainnet", "test-1") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if !status.HealthyWS { + t.Error("expected WS's own first-ever check to be accepted immediately, not gated because HTTP had already been checked") + } +} diff --git a/internal/health/custom_probe.go b/internal/health/custom_probe.go new file mode 100644 index 0000000..1f7bec6 --- /dev/null +++ b/internal/health/custom_probe.go @@ -0,0 +1,70 @@ +package health + +import "aetherlay/internal/config" + +// solanaFinalizedSlotMargin keeps the canned getBlock probe well behind the reported tip +// slot so it targets a rooted, finalized block instead of one that may not exist yet. +const solanaFinalizedSlotMargin = 32 + +// customProbeBuilder pairs a canned request builder with the chain type it's valid for. +// getBlock and eth_getBlockByNumber are both real method names, but only for their own +// dialect; a Solana endpoint erroring on an EVM-shaped request name (or the reverse) +// must never be captured or replayed as that method, since the probe would just fail +// against the wrong chain type. +type customProbeBuilder struct { + chainType string + build func(currentBlockOrSlot int64) (method string, params []any) +} + +// customProbeBuilders maps a JSON-RPC method observed failing on real proxied traffic to +// a canned, read-only request Aetherlay can safely issue on its own to specifically +// re-test that method. This is deliberately not the client's original request body; +// replaying an arbitrary captured request could resubmit a state-mutating call such as +// sendTransaction. Only methods with a well-defined, side-effect-free, always-valid +// request shape belong here. currentBlockOrSlot is the value the calling check just +// obtained from the endpoint's regular getSlot/eth_blockNumber probe. +var customProbeBuilders = map[string]customProbeBuilder{ + "getBlock": { + chainType: config.ChainTypeSolana, + build: func(slot int64) (string, []any) { + target := max(slot-solanaFinalizedSlotMargin, 0) + return "getBlock", []any{target, map[string]any{ + "encoding": "json", + "maxSupportedTransactionVersion": 0, + }} + }, + }, + "eth_getBlockByNumber": { + chainType: config.ChainTypeEVM, + build: func(_ int64) (string, []any) { + return "eth_getBlockByNumber", []any{"latest", false} + }, + }, +} + +// normalizeChainType mirrors the default-to-EVM behavior blockNumberMethod/syncStatusMethod +// already apply elsewhere in this package: an endpoint with ChainType left unset is EVM. +func normalizeChainType(chainType string) string { + if chainType == config.ChainTypeSolana { + return config.ChainTypeSolana + } + return config.ChainTypeEVM +} + +// IsCustomProbeMethod reports whether method is on the allowlist of methods Aetherlay +// knows how to safely re-test on its own via customProbeBuilders, for the given endpoint +// chain type. A method valid for one chain type is never treated as valid for another. +func IsCustomProbeMethod(method, chainType string) bool { + entry, ok := customProbeBuilders[method] + return ok && entry.chainType == normalizeChainType(chainType) +} + +// customProbeBuilderFor returns the canned request builder for method, scoped to +// chainType, or false if method isn't allowlisted for that chain type. +func customProbeBuilderFor(method, chainType string) (func(currentBlockOrSlot int64) (string, []any), bool) { + entry, ok := customProbeBuilders[method] + if !ok || entry.chainType != normalizeChainType(chainType) { + return nil, false + } + return entry.build, true +} diff --git a/internal/health/custom_probe_test.go b/internal/health/custom_probe_test.go new file mode 100644 index 0000000..629cace --- /dev/null +++ b/internal/health/custom_probe_test.go @@ -0,0 +1,100 @@ +package health + +import ( + "testing" + + "aetherlay/internal/config" +) + +// TestCustomProbeBuilderGetBlockUsesMarginBehindTip verifies that the Solana getBlock +// builder targets a slot solanaFinalizedSlotMargin behind the reported tip. +func TestCustomProbeBuilderGetBlockUsesMarginBehindTip(t *testing.T) { + build, ok := customProbeBuilderFor("getBlock", config.ChainTypeSolana) + if !ok { + t.Fatal("expected getBlock to be a registered custom probe builder for Solana") + } + + method, params := build(1000) + if method != "getBlock" { + t.Errorf("expected method getBlock, got %q", method) + } + if len(params) != 2 { + t.Fatalf("expected 2 params, got %d: %v", len(params), params) + } + slot, ok := params[0].(int64) + if !ok || slot != 1000-solanaFinalizedSlotMargin { + t.Errorf("expected slot %d, got %v", 1000-solanaFinalizedSlotMargin, params[0]) + } +} + +// TestCustomProbeBuilderGetBlockClampsToZero verifies that the Solana getBlock builder +// never targets a negative slot when the current slot is below the finalized margin. +func TestCustomProbeBuilderGetBlockClampsToZero(t *testing.T) { + build, _ := customProbeBuilderFor("getBlock", config.ChainTypeSolana) + + _, params := build(5) // well under solanaFinalizedSlotMargin + slot, ok := params[0].(int64) + if !ok || slot != 0 { + t.Errorf("expected slot to clamp to 0 for a low current slot, got %v", params[0]) + } +} + +// TestCustomProbeBuilderEthGetBlockByNumberIsAlwaysLatest verifies that the EVM +// eth_getBlockByNumber builder always requests the latest block, ignoring its input. +func TestCustomProbeBuilderEthGetBlockByNumberIsAlwaysLatest(t *testing.T) { + build, ok := customProbeBuilderFor("eth_getBlockByNumber", config.ChainTypeEVM) + if !ok { + t.Fatal("expected eth_getBlockByNumber to be a registered custom probe builder for EVM") + } + + method, params := build(999999) + if method != "eth_getBlockByNumber" { + t.Errorf("expected method eth_getBlockByNumber, got %q", method) + } + if len(params) != 2 || params[0] != "latest" || params[1] != false { + t.Errorf("expected params [\"latest\", false], got %v", params) + } +} + +// TestCustomProbeBuilderForRejectsMismatchedChainType is a regression guard: getBlock is a +// real Solana method name, but must never be treated as valid for an EVM endpoint (or the +// reverse for eth_getBlockByNumber), even though both are registered method names. +func TestCustomProbeBuilderForRejectsMismatchedChainType(t *testing.T) { + if _, ok := customProbeBuilderFor("getBlock", config.ChainTypeEVM); ok { + t.Error("expected getBlock to be rejected for an EVM endpoint") + } + if _, ok := customProbeBuilderFor("eth_getBlockByNumber", config.ChainTypeSolana); ok { + t.Error("expected eth_getBlockByNumber to be rejected for a Solana endpoint") + } + // An empty ChainType defaults to EVM elsewhere in this package (blockNumberMethod, + // syncStatusMethod); the same default must apply here too. + if _, ok := customProbeBuilderFor("getBlock", ""); ok { + t.Error("expected getBlock to be rejected for an endpoint with no configured chain type (defaults to EVM)") + } + if _, ok := customProbeBuilderFor("eth_getBlockByNumber", ""); !ok { + t.Error("expected eth_getBlockByNumber to be accepted for an endpoint with no configured chain type (defaults to EVM)") + } +} + +// TestIsCustomProbeMethod covers allowlist membership across matching, mismatched, and +// unregistered method/chain-type combinations. +func TestIsCustomProbeMethod(t *testing.T) { + if !IsCustomProbeMethod("getBlock", config.ChainTypeSolana) { + t.Error("expected getBlock to be allowlisted for Solana") + } + if !IsCustomProbeMethod("eth_getBlockByNumber", config.ChainTypeEVM) { + t.Error("expected eth_getBlockByNumber to be allowlisted for EVM") + } + if IsCustomProbeMethod("getBlock", config.ChainTypeEVM) { + t.Error("expected getBlock to not be allowlisted for EVM") + } + if IsCustomProbeMethod("eth_getBlockByNumber", config.ChainTypeSolana) { + t.Error("expected eth_getBlockByNumber to not be allowlisted for Solana") + } + if IsCustomProbeMethod("sendTransaction", config.ChainTypeSolana) { + t.Error("expected sendTransaction to not be allowlisted") + } + if IsCustomProbeMethod("", config.ChainTypeSolana) { + t.Error("expected an empty method to not be allowlisted") + } +} diff --git a/internal/server/custom_probe_test.go b/internal/server/custom_probe_test.go new file mode 100644 index 0000000..d66bdde --- /dev/null +++ b/internal/server/custom_probe_test.go @@ -0,0 +1,301 @@ +package server + +import ( + "context" + "net/http" + "net/http/httptest" + "testing" + "time" + + "aetherlay/internal/config" + "aetherlay/internal/helpers" + "aetherlay/internal/store" +) + +// newCustomProbeTestServer builds a Server with a single healthy Solana endpoint, for +// tests exercising maybeSetCustomProbeMethod in isolation. +func newCustomProbeTestServer(chain, endpointID string) (*Server, *store.MockValkeyClient) { + cfg := &config.Config{ + Endpoints: map[string]config.ChainEndpoints{ + chain: { + endpointID: config.Endpoint{Provider: "test", ChainType: config.ChainTypeSolana, HTTPURL: "http://fail", Role: "primary", Type: "full"}, + }, + }, + } + valkeyClient := store.NewMockValkeyClient() + valkeyClient.PopulateStatuses(map[string]*store.EndpointStatus{ + chain + ":" + endpointID: {HasHTTP: true, HealthyHTTP: true}, + }) + server := NewServer(cfg, valkeyClient, createTestConfig()) + return server, valkeyClient +} + +// TestMaybeSetCustomProbeMethodSetsAllowlistedMethod verifies that a 5xx on an allowlisted +// method captures it as the endpoint's custom probe method. +func TestMaybeSetCustomProbeMethodSetsAllowlistedMethod(t *testing.T) { + server, valkeyClient := newCustomProbeTestServer("solana-devnet", "ep1") + + body := []byte(`{"jsonrpc":"2.0","method":"getBlock","params":[123],"id":1}`) + server.maybeSetCustomProbeMethod(context.Background(), "solana-devnet", "ep1", body) + + state, err := valkeyClient.GetCustomProbeState(context.Background(), "solana-devnet", "ep1") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if state == nil { + t.Fatal("expected a custom probe state to be set") + } + if state.Method != "getBlock" { + t.Errorf("expected method getBlock, got %q", state.Method) + } +} + +// TestMaybeSetCustomProbeMethodIgnoresNonAllowlistedMethod verifies that a state-mutating +// method like sendTransaction is never captured for replay, even on a real 5xx. +func TestMaybeSetCustomProbeMethodIgnoresNonAllowlistedMethod(t *testing.T) { + server, valkeyClient := newCustomProbeTestServer("solana-devnet", "ep1") + + // sendTransaction is state-mutating and must never be captured for replay. + body := []byte(`{"jsonrpc":"2.0","method":"sendTransaction","params":["deadbeef"],"id":1}`) + server.maybeSetCustomProbeMethod(context.Background(), "solana-devnet", "ep1", body) + + state, err := valkeyClient.GetCustomProbeState(context.Background(), "solana-devnet", "ep1") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if state != nil { + t.Errorf("expected no custom probe state for a non-allowlisted method, got %+v", state) + } +} + +// TestMaybeSetCustomProbeMethodIgnoresMethodForWrongChainType is a regression guard: +// eth_getBlockByNumber is a real, allowlisted method name, but not for a Solana endpoint. +// Capturing it there would later have the health checker replay an EVM-shaped request +// against a Solana node, which would just fail. +func TestMaybeSetCustomProbeMethodIgnoresMethodForWrongChainType(t *testing.T) { + server, valkeyClient := newCustomProbeTestServer("solana-devnet", "ep1") // Solana endpoint + + body := []byte(`{"jsonrpc":"2.0","method":"eth_getBlockByNumber","params":["latest",false],"id":1}`) + server.maybeSetCustomProbeMethod(context.Background(), "solana-devnet", "ep1", body) + + state, err := valkeyClient.GetCustomProbeState(context.Background(), "solana-devnet", "ep1") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if state != nil { + t.Errorf("expected no custom probe state for a method that belongs to a different chain type, got %+v", state) + } +} + +// TestMaybeSetCustomProbeMethodIgnoresGetBlockOnEVMEndpoint is the reverse case: getBlock +// is a real Solana method name, but must not be captured for an EVM endpoint. +func TestMaybeSetCustomProbeMethodIgnoresGetBlockOnEVMEndpoint(t *testing.T) { + cfg := &config.Config{ + Endpoints: map[string]config.ChainEndpoints{ + "ethereum": { + "ep1": config.Endpoint{Provider: "test", ChainType: config.ChainTypeEVM, HTTPURL: "http://fail", Role: "primary", Type: "full"}, + }, + }, + } + valkeyClient := store.NewMockValkeyClient() + valkeyClient.PopulateStatuses(map[string]*store.EndpointStatus{ + "ethereum:ep1": {HasHTTP: true, HealthyHTTP: true}, + }) + server := NewServer(cfg, valkeyClient, createTestConfig()) + + body := []byte(`{"jsonrpc":"2.0","method":"getBlock","params":[123],"id":1}`) + server.maybeSetCustomProbeMethod(context.Background(), "ethereum", "ep1", body) + + state, err := valkeyClient.GetCustomProbeState(context.Background(), "ethereum", "ep1") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if state != nil { + t.Errorf("expected no custom probe state for getBlock on an EVM endpoint, got %+v", state) + } +} + +// TestMaybeSetCustomProbeMethodIgnoresUnparseableBody verifies that a body that doesn't +// unmarshal into the expected single-object shape is skipped rather than erroring. +func TestMaybeSetCustomProbeMethodIgnoresUnparseableBody(t *testing.T) { + server, valkeyClient := newCustomProbeTestServer("solana-devnet", "ep1") + + // A batch (array) JSON-RPC request doesn't unmarshal into the single-object shape + // extractRPCMethod expects; the safe default is to skip capture. + body := []byte(`[{"jsonrpc":"2.0","method":"getBlock","params":[1],"id":1}]`) + server.maybeSetCustomProbeMethod(context.Background(), "solana-devnet", "ep1", body) + + state, err := valkeyClient.GetCustomProbeState(context.Background(), "solana-devnet", "ep1") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if state != nil { + t.Errorf("expected no custom probe state for an unparseable/batch body, got %+v", state) + } +} + +// TestMaybeSetCustomProbeMethodDoesNotOverwriteWithinRefreshPeriod verifies that a second +// failure on the same method within the refresh period is a no-op, since the gate acquired +// by the first call is still held. +func TestMaybeSetCustomProbeMethodDoesNotOverwriteWithinRefreshPeriod(t *testing.T) { + server, valkeyClient := newCustomProbeTestServer("solana-devnet", "ep1") + + body := []byte(`{"jsonrpc":"2.0","method":"getBlock","params":[123],"id":1}`) + server.maybeSetCustomProbeMethod(context.Background(), "solana-devnet", "ep1", body) + + firstState, err := valkeyClient.GetCustomProbeState(context.Background(), "solana-devnet", "ep1") + if err != nil || firstState == nil { + t.Fatalf("expected an initial custom probe state to be set, got %+v, err=%v", firstState, err) + } + + // Same method failing again shortly after: the gate acquired by the first call is + // still held, so this second call must be a no-op rather than bumping SetAt forward. + server.maybeSetCustomProbeMethod(context.Background(), "solana-devnet", "ep1", body) + + state, err := valkeyClient.GetCustomProbeState(context.Background(), "solana-devnet", "ep1") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if state == nil || !state.SetAt.Equal(firstState.SetAt) { + t.Errorf("expected SetAt to stay stable within the refresh period, got %+v (first was %+v)", state, firstState) + } +} + +// TestMaybeSetCustomProbeMethodOverwritesAfterRefreshPeriodElapses verifies that once the +// refresh period elapses, the gate expires and a fresh failure refreshes SetAt. +func TestMaybeSetCustomProbeMethodOverwritesAfterRefreshPeriodElapses(t *testing.T) { + server, valkeyClient := newCustomProbeTestServer("solana-devnet", "ep1") + + base := time.Now() + valkeyClient.NowFunc = func() time.Time { return base } + + body := []byte(`{"jsonrpc":"2.0","method":"getBlock","params":[123],"id":1}`) + server.maybeSetCustomProbeMethod(context.Background(), "solana-devnet", "ep1", body) + + firstState, err := valkeyClient.GetCustomProbeState(context.Background(), "solana-devnet", "ep1") + if err != nil || firstState == nil { + t.Fatalf("expected an initial custom probe state to be set, got %+v, err=%v", firstState, err) + } + + // Fast-forward the mock's clock past the refresh period so the gate looks expired, + // without a real sleep (see MockValkeyClient.NowFunc). + valkeyClient.NowFunc = func() time.Time { return base.Add(2 * server.customProbeRefreshPeriod) } + + server.maybeSetCustomProbeMethod(context.Background(), "solana-devnet", "ep1", body) + + state, err := valkeyClient.GetCustomProbeState(context.Background(), "solana-devnet", "ep1") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if state == nil || !state.SetAt.After(firstState.SetAt) { + t.Errorf("expected SetAt to refresh once the gate's refresh period elapsed, got %+v (first was %+v)", state, firstState) + } +} + +// TestMaybeSetCustomProbeMethodNoopWhenEphemeralChecksDisabled verifies that capture is +// skipped entirely when ephemeral checks are disabled, since runEphemeralCheckProtocol, +// the only path that ever clears a captured state, never runs in that case, and a +// captured target would otherwise stay pinned forever. +func TestMaybeSetCustomProbeMethodNoopWhenEphemeralChecksDisabled(t *testing.T) { + cfg := &config.Config{ + Endpoints: map[string]config.ChainEndpoints{ + "solana-devnet": { + "ep1": config.Endpoint{Provider: "test", ChainType: config.ChainTypeSolana, HTTPURL: "http://fail", Role: "primary", Type: "full"}, + }, + }, + } + valkeyClient := store.NewMockValkeyClient() + valkeyClient.PopulateStatuses(map[string]*store.EndpointStatus{ + "solana-devnet:ep1": {HasHTTP: true, HealthyHTTP: true}, + }) + appConfig := &helpers.LoadedConfig{ + EphemeralChecksEnabled: false, + EndpointFailureThreshold: 1, + EndpointSuccessThreshold: 1, + ProxyMaxRetries: 3, + ProxyTimeout: 15, + ProxyTimeoutPerTry: 5, + } + server := NewServer(cfg, valkeyClient, appConfig) + + body := []byte(`{"jsonrpc":"2.0","method":"getBlock","params":[123],"id":1}`) + server.maybeSetCustomProbeMethod(context.Background(), "solana-devnet", "ep1", body) + + state, err := valkeyClient.GetCustomProbeState(context.Background(), "solana-devnet", "ep1") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if state != nil { + t.Errorf("expected no custom probe state to be captured when ephemeral checks are disabled, got %+v", state) + } +} + +// TestForwardRequestCapturesCustomProbeMethodOn5xx is an integration-style check that the +// capture is actually wired into the live proxy path, not just directly callable. +func TestForwardRequestCapturesCustomProbeMethodOn5xx(t *testing.T) { + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusInternalServerError) + w.Write([]byte(`{"jsonrpc":"2.0","id":1,"error":{"code":19,"message":"Temporary internal error"}}`)) + })) + defer upstream.Close() + + cfg := &config.Config{ + Endpoints: map[string]config.ChainEndpoints{ + "solana-devnet": { + "ep1": config.Endpoint{Provider: "test", ChainType: config.ChainTypeSolana, HTTPURL: upstream.URL, Role: "primary", Type: "full"}, + }, + }, + } + valkeyClient := store.NewMockValkeyClient() + valkeyClient.PopulateStatuses(map[string]*store.EndpointStatus{ + "solana-devnet:ep1": {HasHTTP: true, HealthyHTTP: true}, + }) + server := NewServer(cfg, valkeyClient, createTestConfig()) + + body := []byte(`{"jsonrpc":"2.0","method":"getBlock","params":[123],"id":1}`) + err := server.defaultForwardRequestWithBodyFunc(httptest.NewRecorder(), context.Background(), "POST", upstream.URL, body, http.Header{}) + if err == nil { + t.Fatal("expected an error from the 500 response") + } + + state, err := valkeyClient.GetCustomProbeState(context.Background(), "solana-devnet", "ep1") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if state == nil || state.Method != "getBlock" { + t.Errorf("expected a real 500 on getBlock to capture it as the custom probe method, got %+v", state) + } +} + +// TestForwardRequestDoesNotCaptureOn400 ensures the capture only fires for real 5xx +// responses, matching the existing "defer to caller" handling for 400s. +func TestForwardRequestDoesNotCaptureOn400(t *testing.T) { + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusBadRequest) + })) + defer upstream.Close() + + cfg := &config.Config{ + Endpoints: map[string]config.ChainEndpoints{ + "solana-devnet": { + "ep1": config.Endpoint{Provider: "test", ChainType: config.ChainTypeSolana, HTTPURL: upstream.URL, Role: "primary", Type: "full"}, + }, + }, + } + valkeyClient := store.NewMockValkeyClient() + valkeyClient.PopulateStatuses(map[string]*store.EndpointStatus{ + "solana-devnet:ep1": {HasHTTP: true, HealthyHTTP: true}, + }) + server := NewServer(cfg, valkeyClient, createTestConfig()) + + body := []byte(`{"jsonrpc":"2.0","method":"getBlock","params":[123],"id":1}`) + _ = server.defaultForwardRequestWithBodyFunc(httptest.NewRecorder(), context.Background(), "POST", upstream.URL, body, http.Header{}) + + state, err := valkeyClient.GetCustomProbeState(context.Background(), "solana-devnet", "ep1") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if state != nil { + t.Errorf("expected no custom probe state to be captured for a 400 response, got %+v", state) + } +} diff --git a/internal/server/server.go b/internal/server/server.go index e7bb0fe..51f20b2 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -87,6 +87,12 @@ type Server struct { successThreshold int failureStatesMu sync.RWMutex + // customProbeRefreshPeriod gates how often the custom probe method captured for an + // endpoint (see maybeSetCustomProbeMethod) can change, so a 100%-down endpoint (every + // method failing) can't thrash the target before a full ephemeral recovery check + // cycle has a chance to complete. + customProbeRefreshPeriod time.Duration + // Health checker grace period state tracking initialCheckPassed bool hcFailureTimestamp time.Time @@ -99,21 +105,28 @@ type Server struct { proxyWebSocket func(w http.ResponseWriter, r *http.Request, backendURL string) error } +// customProbeRefreshOverheadSeconds is added on top of ephemeralChecksHealthyThreshold * +// ephemeralChecksInterval when deriving customProbeRefreshPeriod, so the derived window +// comfortably covers a full run of consecutive ephemeral recovery checks (e.g. the +// default 3 * 30s + 10s = 100s) instead of expiring right as the last one lands. +const customProbeRefreshOverheadSeconds = 10 + // NewServer creates a new server instance func NewServer(cfg *config.Config, valkeyClient store.ValkeyClientIface, appConfig *helpers.LoadedConfig) *Server { s := &Server{ - appConfig: appConfig, - config: cfg, - ephemeralChecksEnabled: appConfig.EphemeralChecksEnabled, - failureStates: make(map[string]*endpointFailureState), - failureThreshold: appConfig.EndpointFailureThreshold, - healthCache: cache.NewHealthCache(time.Duration(appConfig.HealthCacheTTL) * time.Second), - maxRetries: appConfig.ProxyMaxRetries, - requestTimeout: time.Duration(appConfig.ProxyTimeout) * time.Second, - requestTimeoutPerTry: time.Duration(appConfig.ProxyTimeoutPerTry) * time.Second, - router: mux.NewRouter(), - successThreshold: appConfig.EndpointSuccessThreshold, - valkeyClient: valkeyClient, + appConfig: appConfig, + config: cfg, + customProbeRefreshPeriod: time.Duration(appConfig.EphemeralChecksHealthyThreshold*appConfig.EphemeralChecksInterval+customProbeRefreshOverheadSeconds) * time.Second, + ephemeralChecksEnabled: appConfig.EphemeralChecksEnabled, + failureStates: make(map[string]*endpointFailureState), + failureThreshold: appConfig.EndpointFailureThreshold, + healthCache: cache.NewHealthCache(time.Duration(appConfig.HealthCacheTTL) * time.Second), + maxRetries: appConfig.ProxyMaxRetries, + requestTimeout: time.Duration(appConfig.ProxyTimeout) * time.Second, + requestTimeoutPerTry: time.Duration(appConfig.ProxyTimeoutPerTry) * time.Second, + router: mux.NewRouter(), + successThreshold: appConfig.EndpointSuccessThreshold, + valkeyClient: valkeyClient, } s.forwardRequestWithBody = s.defaultForwardRequestWithBodyFunc @@ -1182,6 +1195,79 @@ func (s *Server) markEndpointHealthyAttempt(chain, endpointID, protocol string) s.updateEndpointHealthState(chain, endpointID, protocol, true) } +// extractRPCMethod returns the JSON-RPC method name from a single (non-batch) JSON-RPC +// request body, or "" if the body isn't a single object with a string "method" field. +func extractRPCMethod(bodyBytes []byte) string { + var req struct { + Method string `json:"method"` + } + if err := json.Unmarshal(bodyBytes, &req); err != nil { + return "" + } + return req.Method +} + +// customProbeValkeyTimeout bounds every Valkey call maybeSetCustomProbeMethod makes. It's +// invoked from the request-forwarding path on a real 5xx; if Valkey stalls, this must not +// hold up the caller's retry-another-endpoint or return-a-response decision for longer +// than a short, fixed bound, regardless of how generous the request's own context is. +const customProbeValkeyTimeout = 2 * time.Second + +// maybeSetCustomProbeMethod records the JSON-RPC method of a request that just failed +// with a real 5xx, if that method is on the allowlist of methods the health checker +// knows how to safely re-test on its own (see health.IsCustomProbeMethod). This is +// deliberately keyed off the method name only, never the client's original request body, +// so a captured failure can never cause Aetherlay to replay a state-mutating call. +// +// Updates are gated by customProbeRefreshPeriod: once a method is recorded, it stays the +// target until that period elapses or the endpoint's ephemeral recovery threshold is +// reached (see runEphemeralCheckProtocol clearing it on recovery). Without this gate, an +// endpoint that's failing on every method would have its target method constantly +// overwritten by whichever request happened to fail last, before any single method could +// accumulate enough consecutive successful re-checks to prove it recovered. +// +// The gate itself is acquired via TryAcquireCustomProbeGate, a single atomic Valkey SET +// NX EX, rather than a get-then-compare-then-set from here: two concurrent failed +// requests could otherwise both observe a missing or expired gate and both write, with +// the second silently replacing the first's target inside what was supposed to be the +// debounce window. A process-local lock can't fix this either, since multiple server +// instances share the same Valkey. +// +// This is a no-op when ephemeral checks are disabled: the only path that ever clears a +// captured custom probe state is runEphemeralCheckProtocol's recovery handling, so +// without it running, a captured target would stay pinned forever, permanently +// re-testing a method that may no longer be relevant instead of falling back to the +// endpoint's default probe. +func (s *Server) maybeSetCustomProbeMethod(ctx context.Context, chain, endpointID string, bodyBytes []byte) { + if !s.ephemeralChecksEnabled { + return + } + + method := extractRPCMethod(bodyBytes) + if method == "" || !health.IsCustomProbeMethod(method, s.chainTypeForEndpoint(chain, endpointID)) { + return + } + + ctx, cancel := context.WithTimeout(ctx, customProbeValkeyTimeout) + defer cancel() + + acquired, err := s.valkeyClient.TryAcquireCustomProbeGate(ctx, chain, endpointID, s.customProbeRefreshPeriod) + if err != nil { + log.Error().Err(err).Str("chain", chain).Str("endpoint", endpointID).Msg("Failed to acquire custom probe gate") + return + } + if !acquired { + return // another request already captured/refreshed the target within this window + } + + if err := s.valkeyClient.SetCustomProbeState(ctx, chain, endpointID, store.CustomProbeState{ + Method: method, + SetAt: time.Now(), + }); err != nil { + log.Error().Err(err).Str("chain", chain).Str("endpoint", endpointID).Str("method", method).Msg("Failed to set custom probe method") + } +} + // findChainAndEndpointByURL searches the config for an endpoint matching the given URL (HTTPURL or WSURL) and returns the chain and endpoint ID. func (s *Server) findChainAndEndpointByURL(url string) (chain string, endpointID string, found bool) { for chainName, endpoints := range s.config.Endpoints { @@ -1261,6 +1347,9 @@ func (s *Server) defaultForwardRequestWithBodyFunc(w http.ResponseWriter, ctx co } else { s.markEndpointUnhealthyProtocol(chain, endpointID, "http") log.Debug().Str("url", helpers.RedactAPIKey(targetURL)).Int("status_code", resp.StatusCode).Msg("Endpoint returned non-2xx status, marked unhealthy") + if resp.StatusCode >= 500 { + s.maybeSetCustomProbeMethod(ctx, chain, endpointID, bodyBytes) + } } } @@ -1355,6 +1444,17 @@ func (s *Server) providerForEndpoint(chain, endpointID string) string { return chainEndpoints[endpointID].Provider } +// chainTypeForEndpoint looks up the configured chain type for a chain/endpoint, used to +// make sure a custom probe method is only ever captured or replayed against the chain +// type it's actually valid for (see health.IsCustomProbeMethod). +func (s *Server) chainTypeForEndpoint(chain, endpointID string) string { + chainEndpoints, ok := s.config.GetEndpointsForChain(chain) + if !ok { + return "" + } + return chainEndpoints[endpointID].ChainType +} + // capacityWindowSeconds resolves the window width to track usage against for the WRITE // path (recordCapacityUsage's usage counter). It must agree with whatever window // effectiveCapacityCeiling's READ path is watching, or gating silently stops working - diff --git a/internal/store/testutils.go b/internal/store/testutils.go index f4b392f..7c1e729 100644 --- a/internal/store/testutils.go +++ b/internal/store/testutils.go @@ -11,6 +11,8 @@ import ( // It supports in-memory endpoint status storage and is safe for concurrent use. type MockValkeyClient struct { rateLimitStates map[string]*RateLimitState + customProbeStates map[string]*CustomProbeState + customProbeGates map[string]time.Time // "chain:endpoint" -> when the gate expires requestCounts map[string]map[string]map[string][3]int64 // [0]=24h, [1]=1m, [2]=all capacityCounts map[string]map[int64]int64 // "chain:endpoint" -> bucket -> count capacityEstimates map[string]*CapacityEstimate // "chain:endpoint" -> learned estimate @@ -27,6 +29,8 @@ type MockValkeyClient struct { func NewMockValkeyClient() *MockValkeyClient { return &MockValkeyClient{ rateLimitStates: make(map[string]*RateLimitState), + customProbeStates: make(map[string]*CustomProbeState), + customProbeGates: make(map[string]time.Time), requestCounts: make(map[string]map[string]map[string][3]int64), capacityCounts: make(map[string]map[int64]int64), capacityEstimates: make(map[string]*CapacityEstimate), @@ -36,7 +40,13 @@ func NewMockValkeyClient() *MockValkeyClient { } } -// GetEndpointStatus returns the status for a given chain and endpoint. +// GetEndpointStatus returns the status for a given chain and endpoint. It returns a copy, +// not the stored pointer, matching the real ValkeyClient (which always hands back a +// freshly unmarshaled value). Callers that mutate the fields of an EndpointStatus they +// got from a prior Get (checker.go, server.go's updateEndpointHealthState) always +// explicitly write it back via UpdateEndpointStatus; if Get returned the live stored +// pointer instead, that in-place mutation would race with any concurrent reader of the +// same endpoint's status, such as a test polling health state from another goroutine. func (m *MockValkeyClient) GetEndpointStatus(_ context.Context, chain, endpointID string) (*EndpointStatus, error) { m.mu.RLock() defer m.mu.RUnlock() @@ -45,7 +55,8 @@ func (m *MockValkeyClient) GetEndpointStatus(_ context.Context, chain, endpointI if !ok { return &EndpointStatus{}, nil } - return status, nil + statusCopy := *status + return &statusCopy, nil } // UpdateEndpointStatus sets the status for a given chain and endpoint. @@ -129,6 +140,64 @@ func (m *MockValkeyClient) GetRateLimitState(_ context.Context, chain, endpoint return state, nil } +// GetCustomProbeState returns the custom probe state for a given chain and endpoint, if +// one has been set. A nil result (with a nil error) means no custom probe method is +// currently active, matching the real ValkeyClient's behavior on a cache miss. It returns +// a copy, not the stored pointer, for the same reason GetEndpointStatus does: exposing +// the map-owned pointer after releasing m.mu would let a caller mutate the stored state +// without synchronization, and would behave differently from the real ValkeyClient, which +// JSON-decodes a fresh value on every read. +func (m *MockValkeyClient) GetCustomProbeState(_ context.Context, chain, endpoint string) (*CustomProbeState, error) { + m.mu.RLock() + defer m.mu.RUnlock() + key := chain + ":" + endpoint + state := m.customProbeStates[key] + if state == nil { + return nil, nil + } + stateCopy := *state + return &stateCopy, nil +} + +// SetCustomProbeState sets the custom probe state for a given chain and endpoint. +func (m *MockValkeyClient) SetCustomProbeState(_ context.Context, chain, endpoint string, state CustomProbeState) error { + m.mu.Lock() + defer m.mu.Unlock() + key := chain + ":" + endpoint + m.customProbeStates[key] = &state + return nil +} + +// ClearCustomProbeState removes the custom probe state for a given chain and endpoint. +func (m *MockValkeyClient) ClearCustomProbeState(_ context.Context, chain, endpoint string) error { + m.mu.Lock() + defer m.mu.Unlock() + key := chain + ":" + endpoint + delete(m.customProbeStates, key) + return nil +} + +// TryAcquireCustomProbeGate mirrors the real ValkeyClient's SET NX EX gate: it atomically +// (under the mock's own lock) checks whether the gate for chain:endpoint is currently +// held and, if not, claims it for ttl and returns true. Concurrent callers under -race +// exercise the same lock, so this only "succeeds" for exactly one caller per window, same +// as SET NX would on a real Valkey server. +func (m *MockValkeyClient) TryAcquireCustomProbeGate(_ context.Context, chain, endpoint string, ttl time.Duration) (bool, error) { + if ttl < time.Second { + return false, fmt.Errorf("custom probe gate ttl must be at least 1 second, got %s", ttl) + } + + m.mu.Lock() + defer m.mu.Unlock() + key := chain + ":" + endpoint + now := m.NowFunc() + if expiresAt, held := m.customProbeGates[key]; held && now.Before(expiresAt) { + return false, nil + } + m.customProbeGates[key] = now.Add(ttl) + return true, nil +} + // CleanupStaleEndpoints is a no-op stub for tests; returns 0 deleted and no error. func (m *MockValkeyClient) CleanupStaleEndpoints(_ context.Context, _ map[string][]string) (int, error) { return 0, nil diff --git a/internal/store/valkey.go b/internal/store/valkey.go index 22b4244..9fe8645 100644 --- a/internal/store/valkey.go +++ b/internal/store/valkey.go @@ -19,6 +19,8 @@ const ( rateLimitPrefix = "rate_limit:" capacityPrefix = "capacity:" capacityEstimatePrefix = "capacity_estimate:" + customProbePrefix = "custom_probe:" + customProbeGatePrefix = "custom_probe_gate:" proxyRequests = "proxy_requests" healthRequests = "health_requests" requests24hKey = "requests_24h" @@ -29,10 +31,17 @@ const ( // EndpointStatus represents the health status and metrics of an endpoint. // It contains information about the endpoint's health, protocol support, and request counts. type EndpointStatus struct { - LastHealthCheck time.Time `json:"last_health_check"` // When the last health check was performed - Requests24h int64 `json:"requests_24h"` // Number of requests in the last 24 hours - Requests1Month int64 `json:"requests_1_month"` // Number of requests in the last month - RequestsLifetime int64 `json:"requests_lifetime"` // Total number of requests since start + // LastHTTPHealthCheck and LastWSHealthCheck are tracked separately, not as one shared + // timestamp: StartEphemeralChecks runs a protocol's first-ever check before the + // other's, so a single shared field would go non-zero after the first protocol + // checked and make the second protocol's own first-ever check look like a prior + // observation, keeping it stuck unhealthy on a passing probe instead of accepting it + // immediately (see resolveHealthTransition's hasPriorCheck parameter). + LastHTTPHealthCheck time.Time `json:"last_http_health_check"` // When the last HTTP health check was performed + LastWSHealthCheck time.Time `json:"last_ws_health_check"` // When the last WS health check was performed + Requests24h int64 `json:"requests_24h"` // Number of requests in the last 24 hours + Requests1Month int64 `json:"requests_1_month"` // Number of requests in the last month + RequestsLifetime int64 `json:"requests_lifetime"` // Total number of requests since start // Protocol support and health flags HasHTTP bool `json:"has_http"` // Whether the endpoint supports HTTP/HTTPS @@ -46,6 +55,9 @@ type EndpointStatus struct { // NewEndpointStatus creates a new endpoint status with default values. // All health flags are set to false and request counts are initialized to 0. +// LastHTTPHealthCheck and LastWSHealthCheck are left at their zero value; that's the +// signal callers use to tell a never-checked protocol apart from one that was actually +// observed unhealthy. func NewEndpointStatus() EndpointStatus { return EndpointStatus{ BlockNumber: 0, @@ -53,7 +65,6 @@ func NewEndpointStatus() EndpointStatus { HasWS: false, HealthyHTTP: false, HealthyWS: false, - LastHealthCheck: time.Now(), Requests24h: 0, Requests1Month: 0, RequestsLifetime: 0, @@ -70,6 +81,10 @@ type ValkeyClientIface interface { GetCombinedRequestCounts(ctx context.Context, chain, endpoint string) (int64, int64, int64, error) GetRateLimitState(ctx context.Context, chain, endpoint string) (*RateLimitState, error) SetRateLimitState(ctx context.Context, chain, endpoint string, state RateLimitState) error + GetCustomProbeState(ctx context.Context, chain, endpoint string) (*CustomProbeState, error) + SetCustomProbeState(ctx context.Context, chain, endpoint string, state CustomProbeState) error + ClearCustomProbeState(ctx context.Context, chain, endpoint string) error + TryAcquireCustomProbeGate(ctx context.Context, chain, endpoint string, ttl time.Duration) (bool, error) IncrementCapacityCount(ctx context.Context, chain, endpoint string, windowSeconds int) (int64, error) GetCapacityCount(ctx context.Context, chain, endpoint string, windowSeconds int) (int64, error) GetCapacityEstimate(ctx context.Context, chain, endpoint string) (*CapacityEstimate, error) @@ -303,7 +318,7 @@ func (r *ValkeyClient) CleanupStaleEndpoints(ctx context.Context, activeEndpoint } } - prefixes := []string{healthPrefix, metricsPrefix, rateLimitPrefix, capacityEstimatePrefix} + prefixes := []string{healthPrefix, metricsPrefix, rateLimitPrefix, capacityEstimatePrefix, customProbePrefix, customProbeGatePrefix} var staleKeys []string for _, prefix := range prefixes { @@ -416,6 +431,98 @@ func (r *ValkeyClient) SetRateLimitState(ctx context.Context, chain, endpoint st return r.client.Do(ctx, cmd).Error() } +// CustomProbeState records which allowlisted method the health checker should +// additionally re-test for an endpoint, captured from a real 5xx on that method, until +// either the refresh period elapses or the endpoint's ephemeral recovery threshold is +// reached (see health.IsCustomProbeMethod and Checker.runEphemeralCheckProtocol). +type CustomProbeState struct { + Method string `json:"method"` + SetAt time.Time `json:"set_at"` +} + +// GetCustomProbeState retrieves the custom probe state for an endpoint, if one is set. +// A nil result (with a nil error) means no custom probe method is currently active. +func (r *ValkeyClient) GetCustomProbeState(ctx context.Context, chain, endpoint string) (*CustomProbeState, error) { + key := customProbePrefix + chain + ":" + endpoint + cmd := r.client.B().Get().Key(key).Build() + result := r.client.Do(ctx, cmd) + + if valkey.IsValkeyNil(result.Error()) { + return nil, nil + } + + data, err := result.AsBytes() + if err != nil { + return nil, err + } + + var state CustomProbeState + if err := json.Unmarshal(data, &state); err != nil { + return nil, err + } + return &state, nil +} + +// SetCustomProbeState stores the custom probe state for an endpoint in Valkey. It has no +// expiration: an endpoint can legitimately stay unhealthy on the captured method for +// longer than any fixed TTL, and a time-based expiry would let the periodic/ephemeral +// checks silently fall back to the default probe (which the captured method may still +// fail) while nothing about the endpoint has actually changed. The state is removed only +// by ClearCustomProbeState on confirmed threshold-based recovery (see +// Checker.runEphemeralCheckProtocol), or by CleanupStaleEndpoints once the endpoint is no +// longer in the active config (customProbePrefix is included in its sweep). +func (r *ValkeyClient) SetCustomProbeState(ctx context.Context, chain, endpoint string, state CustomProbeState) error { + key := customProbePrefix + chain + ":" + endpoint + + jsonBytes, err := json.Marshal(state) + if err != nil { + return err + } + + cmd := r.client.B().Set().Key(key).Value(string(jsonBytes)).Build() + return r.client.Do(ctx, cmd).Error() +} + +// ClearCustomProbeState removes the custom probe state for an endpoint, reverting future +// health checks to the endpoint's default probe method. +func (r *ValkeyClient) ClearCustomProbeState(ctx context.Context, chain, endpoint string) error { + key := customProbePrefix + chain + ":" + endpoint + cmd := r.client.B().Del().Key(key).Build() + return r.client.Do(ctx, cmd).Error() +} + +// TryAcquireCustomProbeGate atomically decides whether the caller is allowed to +// (re)capture the custom probe method for an endpoint right now, using a separate, +// short-lived gate key (SET NX EX) rather than reading CustomProbeState and comparing a +// stored timestamp. A plain get-then-set from application code has a race: two concurrent +// requests can both observe a missing or expired gate before either writes, and both then +// write, with the later one winning even though it's supposed to be debounced. SET NX is +// atomic at the Valkey server itself, so exactly one caller ever acquires the gate in a +// given ttl window, even across multiple server instances sharing the same Valkey. Note +// the gate key's ttl only bounds how often the target method can change (see +// server.maybeSetCustomProbeMethod); it is not the lifetime of CustomProbeState itself, +// which has no expiration (see SetCustomProbeState). +func (r *ValkeyClient) TryAcquireCustomProbeGate(ctx context.Context, chain, endpoint string, ttl time.Duration) (bool, error) { + // Ex() below takes whole seconds; a sub-second ttl would round down to EX 0, which + // Valkey rejects as an invalid expire time, when the caller almost certainly meant + // "expire quickly" rather than "expire immediately." Reject it here with a clear error + // instead of letting that surface as an opaque Valkey command failure. + if ttl < time.Second { + return false, fmt.Errorf("custom probe gate ttl must be at least 1 second, got %s", ttl) + } + + key := customProbeGatePrefix + chain + ":" + endpoint + cmd := r.client.B().Set().Key(key).Value("1").Nx().Ex(ttl).Build() + result := r.client.Do(ctx, cmd) + if valkey.IsValkeyNil(result.Error()) { + return false, nil // gate already held by another caller within this window + } + if err := result.Error(); err != nil { + return false, err + } + return true, nil +} + // capacityBucketKey returns the Valkey key for the current fixed window of width // windowSeconds, e.g. window 10 buckets time into 10-second slices. The window // resets every windowSeconds because each slice gets its own key - unlike diff --git a/internal/store/valkey_test.go b/internal/store/valkey_test.go index a6da51b..5abaad6 100644 --- a/internal/store/valkey_test.go +++ b/internal/store/valkey_test.go @@ -5,6 +5,8 @@ import ( "fmt" "net" "os" + "sync" + "sync/atomic" "testing" "time" ) @@ -51,14 +53,15 @@ func TestUpdateAndGetEndpointStatus(t *testing.T) { // Create a test status status := EndpointStatus{ - LastHealthCheck: time.Now(), - Requests24h: 10, - Requests1Month: 100, - RequestsLifetime: 1000, - HasHTTP: true, - HasWS: true, - HealthyHTTP: true, - HealthyWS: false, + LastHTTPHealthCheck: time.Now(), + LastWSHealthCheck: time.Now(), + Requests24h: 10, + Requests1Month: 100, + RequestsLifetime: 1000, + HasHTTP: true, + HasWS: true, + HealthyHTTP: true, + HealthyWS: false, } // Update the status @@ -133,6 +136,121 @@ func TestGetEndpointStatusForNonExistentEndpoint(t *testing.T) { } } +// TestSetAndGetCustomProbeState verifies that a stored custom probe state round-trips +// with its method and timestamp intact. +func TestSetAndGetCustomProbeState(t *testing.T) { + client := NewMockValkeyClient() + ctx := context.Background() + chain := "solana-devnet" + endpoint := "ep1" + + setAt := time.Now() + err := client.SetCustomProbeState(ctx, chain, endpoint, CustomProbeState{Method: "getBlock", SetAt: setAt}) + if err != nil { + t.Fatalf("SetCustomProbeState failed: %v", err) + } + + state, err := client.GetCustomProbeState(ctx, chain, endpoint) + if err != nil { + t.Fatalf("GetCustomProbeState failed: %v", err) + } + if state == nil { + t.Fatal("expected a non-nil custom probe state") + } + if state.Method != "getBlock" { + t.Errorf("expected method getBlock, got %q", state.Method) + } + if !state.SetAt.Equal(setAt) { + t.Errorf("expected SetAt %v, got %v", setAt, state.SetAt) + } +} + +// TestTryAcquireCustomProbeGateIsExclusiveUnderConcurrency is a regression guard for the +// atomicity that's required here: many concurrent callers racing for the same endpoint's +// gate must see exactly one winner, never more. +func TestTryAcquireCustomProbeGateIsExclusiveUnderConcurrency(t *testing.T) { + client := NewMockValkeyClient() + ctx := context.Background() + chain := "solana-devnet" + endpoint := "ep1" + + const goroutines = 50 + var wg sync.WaitGroup + var acquiredCount int64 + wg.Add(goroutines) + for range goroutines { + go func() { + defer wg.Done() + acquired, err := client.TryAcquireCustomProbeGate(ctx, chain, endpoint, time.Minute) + if err != nil { + t.Errorf("TryAcquireCustomProbeGate failed: %v", err) + return + } + if acquired { + atomic.AddInt64(&acquiredCount, 1) + } + }() + } + wg.Wait() + + if acquiredCount != 1 { + t.Errorf("expected exactly 1 caller to acquire the gate, got %d", acquiredCount) + } +} + +// TestTryAcquireCustomProbeGateRejectsSubSecondTTL verifies that a ttl below one second is +// rejected up front, rather than silently rounding down to a zero-second Valkey EX, which +// the server would reject anyway with a far less clear error. +func TestTryAcquireCustomProbeGateRejectsSubSecondTTL(t *testing.T) { + client := NewMockValkeyClient() + acquired, err := client.TryAcquireCustomProbeGate(context.Background(), "solana-devnet", "ep1", 500*time.Millisecond) + if err == nil { + t.Error("expected an error for a sub-second ttl") + } + if acquired { + t.Error("expected acquired=false alongside the error") + } +} + +// TestGetCustomProbeStateForNonExistentEndpoint verifies a nil, error-free result for an +// endpoint that has never had a custom probe state set. +func TestGetCustomProbeStateForNonExistentEndpoint(t *testing.T) { + client := NewMockValkeyClient() + ctx := context.Background() + + state, err := client.GetCustomProbeState(ctx, "solana-devnet", "no-such-endpoint") + if err != nil { + t.Fatalf("GetCustomProbeState failed: %v", err) + } + if state != nil { + t.Errorf("expected a nil custom probe state for an endpoint that was never set, got %+v", state) + } +} + +// TestClearCustomProbeState verifies that clearing removes a previously set custom probe +// state. +func TestClearCustomProbeState(t *testing.T) { + client := NewMockValkeyClient() + ctx := context.Background() + chain := "solana-devnet" + endpoint := "ep1" + + if err := client.SetCustomProbeState(ctx, chain, endpoint, CustomProbeState{Method: "getBlock", SetAt: time.Now()}); err != nil { + t.Fatalf("SetCustomProbeState failed: %v", err) + } + if err := client.ClearCustomProbeState(ctx, chain, endpoint); err != nil { + t.Fatalf("ClearCustomProbeState failed: %v", err) + } + + state, err := client.GetCustomProbeState(ctx, chain, endpoint) + if err != nil { + t.Fatalf("GetCustomProbeState failed: %v", err) + } + if state != nil { + t.Errorf("expected custom probe state to be cleared, got %+v", state) + } +} + func uniqueTestKey(base string) string { return fmt.Sprintf("%s-%d", base, time.Now().UnixNano()) } diff --git a/services/health-checker/Dockerfile b/services/health-checker/Dockerfile index 4dfcece..0403768 100644 --- a/services/health-checker/Dockerfile +++ b/services/health-checker/Dockerfile @@ -1,5 +1,5 @@ # Stage 1: Build the Go binary -FROM golang:1.25-alpine AS builder +FROM dhi.io/golang:1.26-alpine AS builder # Set the working directory inside the container WORKDIR /app diff --git a/services/load-balancer/Dockerfile b/services/load-balancer/Dockerfile index 3ef7cd5..a75b485 100644 --- a/services/load-balancer/Dockerfile +++ b/services/load-balancer/Dockerfile @@ -1,5 +1,5 @@ # Stage 1: Build the Go binary -FROM golang:1.25-alpine AS builder +FROM dhi.io/golang:1.26-alpine AS builder # Set the working directory inside the container WORKDIR /app