diff --git a/.env.example b/.env.example index c56c182..d20bf0e 100644 --- a/.env.example +++ b/.env.example @@ -51,6 +51,25 @@ GRPC_API_DB_POOL_SIZE=10 # against PgBouncer. GO_API_DB_POOL_SIZE=5 +# OPTIONAL go-api default: 0 +# Minimum pool connections kept open (issue #238). 0 matches pgxpool's own +# default — raise it to pre-warm connections and cut cold-start latency. +GO_API_DB_POOL_MIN_CONNS=0 + +# OPTIONAL go-api default: 1800000 (30 min) +# Maximum age of a pooled connection before it's recycled (issue #238). A +# random jitter of 10% of this value is applied automatically so connections +# don't all recycle at once. +GO_API_DB_POOL_MAX_CONN_LIFETIME_MS=1800000 + +# OPTIONAL go-api default: 300000 (5 min) +# How long an idle pooled connection is kept before it's closed (issue #238). +GO_API_DB_POOL_MAX_CONN_IDLE_TIME_MS=300000 + +# OPTIONAL go-api default: 60000 (1 min) +# How often idle pooled connections are health-checked (issue #238). +GO_API_DB_POOL_HEALTH_CHECK_PERIOD_MS=60000 + # OPTIONAL go-api # PgBouncer admin console connection, used by GET /v1/admin/db to read # SHOW POOLS / SHOW STATS. Connect to the virtual "pgbouncer" database. @@ -151,8 +170,11 @@ RUST_LOG=info # Port serving /healthz and /readyz. HEALTH_PORT=8080 -# OPTIONAL indexer default: 30000 / 10000 -# Postgres per-statement and idle-in-transaction timeout bounds (ms). +# OPTIONAL indexer + go-api default: 30000 / 10000 +# Postgres per-statement and idle-in-transaction timeout bounds (ms), applied +# to every connection at connect time. Shared across both services (issue +# #238) so they agree on how long a query or idle transaction may hold a +# connection; bounded to [100, 3600000]ms, out-of-range values are clamped. DB_STATEMENT_TIMEOUT_MS=30000 DB_IDLE_IN_TRANSACTION_TIMEOUT_MS=10000 @@ -287,6 +309,13 @@ RETENTION_SOROBAN_EVENTS_DAYS=0 PPROF_ENABLED=false PPROF_ADDR=127.0.0.1:6060 +# OPTIONAL go-api default: 9091 +# Port the Go REST API serves its Prometheus /metrics endpoint on (issue #58), +# separate from the public API port. This is a distinct process/port from the +# Rust indexer's own METRICS_PORT (default 9090) documented above — same env +# var name, different service. +METRICS_PORT=9091 + # --------------------------------------------------------------------------- # Internal /internal/status auth (issue #316) # --------------------------------------------------------------------------- diff --git a/api/openapi.yaml b/api/openapi.yaml index 301c6f0..8dac5a6 100644 --- a/api/openapi.yaml +++ b/api/openapi.yaml @@ -46,22 +46,67 @@ security: paths: /v1/health: get: - summary: Health check - description: Returns indexer health status and last indexed ledger + summary: Liveness check + description: >- + Cheap process-liveness check (issue #243) — no dependency calls (no + Postgres/Redis/gRPC). Always 200 while the process is up and serving + requests. Intended for Kubernetes' liveness probe. For dependency + health (Postgres/Redis/gRPC), see GET /v1/ready instead. operationId: getHealth tags: - System security: [] responses: "200": - description: Indexer is healthy or degraded + description: Process is alive content: application/json: schema: - $ref: "#/components/schemas/HealthResponse" + $ref: "#/components/schemas/LivenessResponse" + "429": + $ref: "#/components/responses/TooManyRequestsIPOnly" "503": $ref: "#/components/responses/ServiceUnavailable" + /v1/ready: + get: + summary: Readiness check + description: >- + Verifies Postgres, Redis, and the gRPC backend concurrently, each + with a 3-second timeout (issue #243). Returns 503 if any dependency + check fails. Intended for Kubernetes' readiness probe / Fly's HTTP + service check, so a pod with a broken dependency is pulled out of + rotation instead of continuing to receive traffic it can't serve. + operationId: getReady + tags: + - System + security: [] + responses: + "200": + description: All dependencies reachable + content: + application/json: + schema: + $ref: "#/components/schemas/ReadyResponse" + "429": + $ref: "#/components/responses/TooManyRequestsIPOnly" + "503": + description: >- + Either one or more dependencies is unreachable (ReadyResponse + body, status "degraded", the failing entry in checks set to + "error: ..."), or the server is shedding load under the global + concurrency cap (ErrorResponse body, Retry-After header set) — + the same outermost load-shedding behavior every endpoint shares. + headers: + Retry-After: + $ref: "#/components/headers/Retry-After" + content: + application/json: + schema: + oneOf: + - $ref: "#/components/schemas/ReadyResponse" + - $ref: "#/components/schemas/ErrorResponse" + /v1/events: get: summary: List Soroban contract events @@ -114,14 +159,21 @@ paths: maximum: 200 default: 50 description: Maximum number of events to return - - name: after + - name: cursor in: query schema: type: string - description: Opaque pagination cursor from previous response (for next page) + description: Opaque pagination cursor from previous response's next_cursor (for next page) responses: "200": description: List of events with pagination metadata + headers: + X-RateLimit-Limit: + $ref: "#/components/headers/X-RateLimit-Limit" + X-RateLimit-Remaining: + $ref: "#/components/headers/X-RateLimit-Remaining" + X-RateLimit-Reset: + $ref: "#/components/headers/X-RateLimit-Reset" content: application/json: schema: @@ -130,6 +182,8 @@ paths: $ref: "#/components/responses/BadRequest" "401": $ref: "#/components/responses/Unauthorized" + "429": + $ref: "#/components/responses/RateLimitExceeded" "503": $ref: "#/components/responses/ServiceUnavailable" @@ -152,6 +206,13 @@ paths: responses: "200": description: Event details + headers: + X-RateLimit-Limit: + $ref: "#/components/headers/X-RateLimit-Limit" + X-RateLimit-Remaining: + $ref: "#/components/headers/X-RateLimit-Remaining" + X-RateLimit-Reset: + $ref: "#/components/headers/X-RateLimit-Reset" content: application/json: schema: @@ -171,6 +232,8 @@ paths: application/json: schema: $ref: "#/components/schemas/ErrorResponse" + "429": + $ref: "#/components/responses/RateLimitExceeded" "503": $ref: "#/components/responses/ServiceUnavailable" @@ -197,6 +260,13 @@ paths: responses: "200": description: Server-Sent Events stream + headers: + X-RateLimit-Limit: + $ref: "#/components/headers/X-RateLimit-Limit" + X-RateLimit-Remaining: + $ref: "#/components/headers/X-RateLimit-Remaining" + X-RateLimit-Reset: + $ref: "#/components/headers/X-RateLimit-Reset" content: text/event-stream: schema: @@ -205,6 +275,8 @@ paths: $ref: "#/components/responses/BadRequest" "401": $ref: "#/components/responses/Unauthorized" + "429": + $ref: "#/components/responses/RateLimitExceeded" "503": $ref: "#/components/responses/ServiceUnavailable" @@ -253,6 +325,13 @@ paths: description: | Batch result. Always returns 200 even when some IDs are not found; check `missing` for unresolved IDs. + headers: + X-RateLimit-Limit: + $ref: "#/components/headers/X-RateLimit-Limit" + X-RateLimit-Remaining: + $ref: "#/components/headers/X-RateLimit-Remaining" + X-RateLimit-Reset: + $ref: "#/components/headers/X-RateLimit-Reset" content: application/json: schema: @@ -278,6 +357,8 @@ paths: $ref: "#/components/responses/BadRequest" "401": $ref: "#/components/responses/Unauthorized" + "429": + $ref: "#/components/responses/RateLimitExceeded" "503": $ref: "#/components/responses/ServiceUnavailable" @@ -299,6 +380,13 @@ paths: responses: "200": description: Contract event schema registry entry + headers: + X-RateLimit-Limit: + $ref: "#/components/headers/X-RateLimit-Limit" + X-RateLimit-Remaining: + $ref: "#/components/headers/X-RateLimit-Remaining" + X-RateLimit-Reset: + $ref: "#/components/headers/X-RateLimit-Reset" content: application/json: schema: @@ -307,6 +395,8 @@ paths: $ref: "#/components/responses/BadRequest" "401": $ref: "#/components/responses/Unauthorized" + "429": + $ref: "#/components/responses/RateLimitExceeded" "503": $ref: "#/components/responses/ServiceUnavailable" @@ -328,6 +418,13 @@ paths: responses: "200": description: Contract spec and detected interfaces + headers: + X-RateLimit-Limit: + $ref: "#/components/headers/X-RateLimit-Limit" + X-RateLimit-Remaining: + $ref: "#/components/headers/X-RateLimit-Remaining" + X-RateLimit-Reset: + $ref: "#/components/headers/X-RateLimit-Reset" content: application/json: schema: @@ -338,6 +435,8 @@ paths: $ref: "#/components/responses/Unauthorized" "404": $ref: "#/components/responses/NotFound" + "429": + $ref: "#/components/responses/RateLimitExceeded" "503": $ref: "#/components/responses/ServiceUnavailable" @@ -359,6 +458,13 @@ paths: responses: "200": description: Latest known value per storage key + headers: + X-RateLimit-Limit: + $ref: "#/components/headers/X-RateLimit-Limit" + X-RateLimit-Remaining: + $ref: "#/components/headers/X-RateLimit-Remaining" + X-RateLimit-Reset: + $ref: "#/components/headers/X-RateLimit-Reset" content: application/json: schema: @@ -367,6 +473,8 @@ paths: $ref: "#/components/responses/BadRequest" "401": $ref: "#/components/responses/Unauthorized" + "429": + $ref: "#/components/responses/RateLimitExceeded" "503": $ref: "#/components/responses/ServiceUnavailable" @@ -394,6 +502,13 @@ paths: responses: "200": description: Recorded changes for the requested storage key + headers: + X-RateLimit-Limit: + $ref: "#/components/headers/X-RateLimit-Limit" + X-RateLimit-Remaining: + $ref: "#/components/headers/X-RateLimit-Remaining" + X-RateLimit-Reset: + $ref: "#/components/headers/X-RateLimit-Reset" content: application/json: schema: @@ -402,6 +517,8 @@ paths: $ref: "#/components/responses/BadRequest" "401": $ref: "#/components/responses/Unauthorized" + "429": + $ref: "#/components/responses/RateLimitExceeded" "503": $ref: "#/components/responses/ServiceUnavailable" @@ -423,6 +540,8 @@ paths: application/json: schema: $ref: "#/components/schemas/IndexerStatsResponse" + "429": + $ref: "#/components/responses/TooManyRequestsIPOnly" "503": $ref: "#/components/responses/ServiceUnavailable" @@ -464,7 +583,19 @@ paths: description: Number of top contracts to return responses: "200": - description: Contract activity statistics + description: >- + Contract activity statistics. X-Cache indicates whether this + response was served from the 60s Redis response cache (HIT) or + freshly computed (MISS). + headers: + X-RateLimit-Limit: + $ref: "#/components/headers/X-RateLimit-Limit" + X-RateLimit-Remaining: + $ref: "#/components/headers/X-RateLimit-Remaining" + X-RateLimit-Reset: + $ref: "#/components/headers/X-RateLimit-Reset" + X-Cache: + $ref: "#/components/headers/X-Cache" content: application/json: schema: @@ -473,6 +604,8 @@ paths: $ref: "#/components/responses/BadRequest" "401": $ref: "#/components/responses/Unauthorized" + "429": + $ref: "#/components/responses/RateLimitExceeded" "503": $ref: "#/components/responses/ServiceUnavailable" @@ -526,6 +659,8 @@ paths: $ref: "#/components/responses/BadRequest" "401": $ref: "#/components/responses/Unauthorized" + "429": + $ref: "#/components/responses/TooManyRequestsIPOnly" get: summary: List API keys @@ -569,6 +704,8 @@ paths: format: date-time "401": $ref: "#/components/responses/Unauthorized" + "429": + $ref: "#/components/responses/TooManyRequestsIPOnly" /v1/api-keys/{id}: delete: @@ -598,6 +735,8 @@ paths: application/json: schema: $ref: "#/components/schemas/ErrorResponse" + "429": + $ref: "#/components/responses/TooManyRequestsIPOnly" /v1/admin/db: get: @@ -617,6 +756,8 @@ paths: type: object "401": $ref: "#/components/responses/Unauthorized" + "429": + $ref: "#/components/responses/TooManyRequestsIPOnly" /metrics: get: @@ -709,31 +850,53 @@ components: nullable: true description: Opaque cursor for next page (null if has_more is false) - HealthResponse: + LivenessResponse: + type: object + required: + - status + properties: + status: + type: string + enum: [ok] + description: Always "ok" while the process is up — no dependency checks. + + ReadyChecks: + type: object + required: + - postgres + - redis + - grpc_api + properties: + postgres: + type: string + description: '"ok" or "error: "' + redis: + type: string + description: '"ok" or "error: "' + grpc_api: + type: string + description: '"ok" or "error: "' + + ReadyResponse: type: object required: - status - - indexer + - indexer_lag + - checks properties: status: type: string enum: [ok, degraded] - description: Overall system status - indexer: - type: object - required: - - last_ledger_indexed - properties: - last_ledger_indexed: - type: integer - format: int64 - nullable: true - description: Latest indexed ledger sequence - last_poll_at: - type: string - format: date-time - nullable: true - description: Timestamp of last successful indexer poll + description: '"degraded" when any dependency check in `checks` failed.' + indexer_lag: + type: integer + format: int64 + nullable: true + description: >- + Ledgers behind chain tip, from system_state. Null when Postgres + is unreachable or the chain-tip cache hasn't been populated yet. + checks: + $ref: "#/components/schemas/ReadyChecks" IndexerStatsResponse: type: object @@ -1057,6 +1220,60 @@ components: type: string description: Request ID for debugging + headers: + X-RateLimit-Limit: + description: >- + Requests allowed per window for this API key's rate-limit tier. + Present on every response from an endpoint secured by ApiKeyAuth + (2xx and 429 alike) once a valid X-API-Key was presented. + required: true + schema: + type: integer + minimum: 0 + example: 50 + + X-RateLimit-Remaining: + description: >- + Requests remaining in the current window for this API key. 0 on the + response that triggers a 429. + required: true + schema: + type: integer + minimum: 0 + example: 12 + + X-RateLimit-Reset: + description: >- + Unix timestamp (seconds) when the current rate-limit window resets. + required: true + schema: + type: integer + format: int64 + example: 1732900000 + + Retry-After: + description: >- + Seconds to wait before retrying. Present on 429 (rate limit + exceeded, per-API-key or per-IP) and on 503 responses caused by the + global concurrency cap shedding load. Not present on a 503 caused by + an unavailable dependency (database/Redis/gRPC backend) — check the + error envelope's `error.code` to distinguish the two. + schema: + type: integer + minimum: 1 + example: 1 + + X-Cache: + description: >- + Whether this response was served from the Redis response cache + (HIT) or freshly computed (MISS). Only emitted by endpoints that + cache their response. + required: true + schema: + type: string + enum: [HIT, MISS] + example: HIT + responses: BadRequest: description: Invalid request parameters @@ -1079,8 +1296,48 @@ components: schema: $ref: "#/components/schemas/ErrorResponse" + RateLimitExceeded: + description: >- + Rate limit exceeded for this API key's tier (error.code + RATE_LIMITED). Carries the same X-RateLimit-* headers as a + successful response (X-RateLimit-Remaining is 0) plus Retry-After. + headers: + X-RateLimit-Limit: + $ref: "#/components/headers/X-RateLimit-Limit" + X-RateLimit-Remaining: + $ref: "#/components/headers/X-RateLimit-Remaining" + X-RateLimit-Reset: + $ref: "#/components/headers/X-RateLimit-Reset" + Retry-After: + $ref: "#/components/headers/Retry-After" + content: + application/json: + schema: + $ref: "#/components/schemas/ErrorResponse" + + TooManyRequestsIPOnly: + description: >- + Per-IP rate limit exceeded (error.code RATE_LIMITED). Applies to + endpoints not covered by per-API-key limiting (public endpoints, or + admin endpoints authenticated via ADMIN_API_KEY rather than + X-API-Key) — only Retry-After is set, no X-RateLimit-* headers. + headers: + Retry-After: + $ref: "#/components/headers/Retry-After" + content: + application/json: + schema: + $ref: "#/components/schemas/ErrorResponse" + ServiceUnavailable: - description: Service temporarily unavailable + description: >- + Service temporarily unavailable — either a dependency (database, + Redis, gRPC backend) is down, or the server is shedding load under + the global concurrency cap. When load-shedding is the cause, + Retry-After is set; otherwise it is absent. + headers: + Retry-After: + $ref: "#/components/headers/Retry-After" content: application/json: schema: diff --git a/docker/docker-compose.yml b/docker/docker-compose.yml index dddc41c..0ee0a1c 100644 --- a/docker/docker-compose.yml +++ b/docker/docker-compose.yml @@ -129,13 +129,16 @@ services: LOG_LEVEL: ${LOG_LEVEL} ports: - "${PORT}:${PORT}" - # Probe the health endpoint we just implemented. wget is used instead of - # curl because it is available in the scratch/alpine base images without - # adding an extra layer. The 4-second timeout matches the handler's - # 3-second per-check timeout plus one second of slack. + # Probe /v1/ready (issue #243), not /v1/health — compose's healthcheck + # gates other services' `depends_on: condition: service_healthy`, so it + # needs the readiness semantics (DB/Redis/gRPC reachable), not just + # liveness. wget is used instead of curl because it is available in the + # scratch/alpine base images without adding an extra layer. The + # 4-second timeout matches the handler's 3-second per-check timeout plus + # one second of slack. healthcheck: test: ["CMD", "wget", "--quiet", "--tries=1", "--spider", - "http://localhost:${PORT}/v1/health"] + "http://localhost:${PORT}/v1/ready"] interval: 15s timeout: 5s retries: 3 diff --git a/docs/deployment.md b/docs/deployment.md index b26ba3c..8c06338 100644 --- a/docs/deployment.md +++ b/docs/deployment.md @@ -147,13 +147,13 @@ docker compose -f docker/docker-compose.yml -f docker/docker-compose.prod.yml up ### 7. Verify health ```bash -curl https://your-domain.com/v1/health +curl https://your-domain.com/v1/ready ``` -Expected response: +Expected response (Postgres/Redis/gRPC all reachable): ```json -{"status":"ok"} +{"status":"ok","indexer_lag":0,"checks":{"postgres":"ok","redis":"ok","grpc_api":"ok"}} ``` --- @@ -198,7 +198,7 @@ docker compose -f docker/docker-compose.yml -f docker/docker-compose.prod.yml \ ### 4. Verify deployment ```bash -curl https://your-domain.com/v1/health +curl https://your-domain.com/v1/ready docker compose -f docker/docker-compose.yml -f docker/docker-compose.prod.yml \ logs --tail=50 api ``` @@ -227,7 +227,7 @@ Migrations in `database/migrations/` are plain SQL and have no automated down pa ### 4. Verify health after rollback ```bash -curl https://your-domain.com/v1/health +curl https://your-domain.com/v1/ready ``` --- @@ -285,20 +285,26 @@ curl https://your-domain.com/v1/health | Endpoint | Description | |---|---| -| `GET /v1/health` | Public liveness check. Returns indexer poll status. | +| `GET /v1/health` | Public liveness check. No dependency calls — just confirms the process is up (issue #243). | +| `GET /v1/ready` | Public readiness check. Verifies Postgres, Redis, and the gRPC backend (issue #243); 503 if any is unreachable. | | `GET /internal/status` | Internal metrics endpoint (planned for a future release). | -`/v1/health` response shapes: +`/v1/health` response shape (always 200 while the process is alive): ```json {"status":"ok"} ``` -Indexer is polling within the last 60 seconds. + +`/v1/ready` response shapes: + +```json +{"status":"ok","indexer_lag":3,"checks":{"postgres":"ok","redis":"ok","grpc_api":"ok"}} +``` ```json -{"status":"degraded"} +{"status":"degraded","indexer_lag":null,"checks":{"postgres":"error: dial tcp: connection refused","redis":"ok","grpc_api":"ok"}} ``` -Indexer has stalled or the database is unreachable. +`indexer_lag` is null whenever Postgres is unreachable or the chain-tip cache hasn't been populated yet. Any non-"ok" entry in `checks` returns HTTP 503. ### PostgreSQL Disk Usage @@ -321,10 +327,10 @@ A growing `trident:events` stream length indicates consumer lag. Investigate the ### Indexer Lag -Check `last_poll_at` in the health response. If `status` is `degraded` or `last_poll_at` is more than 5 minutes ago, the indexer has stalled. +Check `indexer_lag` in the readiness response, or query `GET /v1/stats/indexer` directly for `last_poll_at` and `status`. If `indexer_lag` is large or `checks.postgres` is not `ok`, investigate the indexer/database. ```bash -curl https://your-domain.com/v1/health | jq . +curl https://your-domain.com/v1/ready | jq . ``` ### nginx / WebSocket Connections @@ -588,8 +594,10 @@ lasting change. under the top-level `[checks]` section so Fly restarts the machine if readiness fails, not just if the TCP port stops accepting connections. - **Go API metrics**: `GET /metrics` on the public `trident-api` endpoint -- **Go API health check**: `GET /v1/health` (used by Fly's HTTP service check - in `fly/api.toml`) +- **Go API health check**: `GET /v1/ready` (used by Fly's HTTP service check + in `fly/api.toml` to gate traffic routing). `/v1/ready` verifies + Postgres/Redis/gRPC reachability, unlike `/v1/health`, which is a cheap + liveness check that says only that the process is up (issue #243). - **gRPC API**: no HTTP health endpoint exists in `crates/api` today; Fly's check in `fly/grpc-api.toml` is a plain TCP check against port 50051. This proves the socket is listening, not that gRPC calls actually succeed — if a @@ -617,6 +625,7 @@ fly config validate -c fly/api.toml fly config validate -c fly/grpc-api.toml fly config validate -c fly/indexer.toml ``` +>>>>>>> origin/dev ### Updating secrets diff --git a/docs/kubernetes.md b/docs/kubernetes.md index 09fb30b..ee84015 100644 --- a/docs/kubernetes.md +++ b/docs/kubernetes.md @@ -508,10 +508,10 @@ any of these, redact the credential portion — don't log the raw env var value. ## Health checks -The Go API exposes `GET /v1/health`. Kubernetes liveness and readiness probes are pre-configured in the chart: +The Go API exposes two endpoints for Kubernetes' liveness and readiness probes, split per issue #243 so a dependency outage never causes a restart loop that can't fix it: -- **Liveness** (`failureThreshold: 3`): restarts the container after 3 consecutive failures. -- **Readiness** (`failureThreshold: 1`): removes the pod from the Service load balancer on the first failure for faster traffic isolation. +- **Liveness** — `GET /v1/health` (`failureThreshold: 3`): cheap, no dependency calls, just confirms the process is up. Restarts the container after 3 consecutive failures. Never fails because Postgres/Redis/the gRPC backend is down — restarting the pod doesn't fix an external dependency, so liveness must not conflate "the process is stuck" with "a dependency is unreachable." +- **Readiness** — `GET /v1/ready` (`failureThreshold: 1`): checks Postgres, Redis, and the gRPC backend concurrently (3s timeout per dependency) and returns 503 if any fail. Removes the pod from the Service load balancer on the first failure for faster traffic isolation. ## Upgrading diff --git a/fly/api.toml b/fly/api.toml index 4c2285c..3b29da2 100644 --- a/fly/api.toml +++ b/fly/api.toml @@ -27,12 +27,16 @@ primary_region = "iad" auto_start_machines = true min_machines_running = 1 + # Fly has no separate liveness/readiness concept — this single check gates + # whether the proxy routes traffic to the machine, so it needs readiness + # semantics (issue #243): /v1/ready checks Postgres/Redis/gRPC reachability, + # not just /v1/health's cheap liveness check. [[http_service.checks]] interval = "10s" timeout = "5s" grace_period = "15s" method = "GET" - path = "/v1/health" + path = "/v1/ready" [[vm]] size = "shared-cpu-1x" diff --git a/helm/trident/values.yaml b/helm/trident/values.yaml index 171f74c..3048149 100644 --- a/helm/trident/values.yaml +++ b/helm/trident/values.yaml @@ -190,6 +190,9 @@ goApi: minReplicas: 2 maxReplicas: 10 targetCPUUtilizationPercentage: 70 + # Liveness hits the cheap, dependency-free /v1/health (issue #243) — it + # must never fail just because Postgres/Redis/the gRPC backend is + # unreachable, since restarting this pod can't fix an external dependency. livenessProbe: httpGet: path: /v1/health @@ -198,15 +201,20 @@ goApi: periodSeconds: 10 failureThreshold: 3 timeoutSeconds: 2 + # Readiness hits /v1/ready (issue #243), which checks Postgres, Redis, and + # the gRPC backend concurrently with a 3s-per-dependency timeout in the + # handler (services/api/handlers/health.go) — timeoutSeconds here must + # exceed that, or the probe itself times out before a legitimately slow + # (but healthy) dependency check can complete. readinessProbe: httpGet: - path: /v1/health + path: /v1/ready port: 3000 initialDelaySeconds: 5 periodSeconds: 5 # Stricter than liveness — pod is removed from rotation faster failureThreshold: 1 - timeoutSeconds: 2 + timeoutSeconds: 4 # Grace period must be >= the bounded shutdown window in main.go so SIGTERM # drains in-flight requests before K8s force-kills the pod (issue #233). terminationGracePeriodSeconds: 30 diff --git a/services/api/go.mod b/services/api/go.mod index 1304d3d..8ce1b49 100644 --- a/services/api/go.mod +++ b/services/api/go.mod @@ -3,8 +3,11 @@ module github.com/Depo-dev/trident/services/api go 1.25.0 require ( + github.com/alicebob/miniredis/v2 v2.38.0 + github.com/getkin/kin-openapi v0.145.0 github.com/google/uuid v1.6.0 github.com/jackc/pgx/v5 v5.10.0 + github.com/prometheus/client_golang v1.24.1 github.com/redis/go-redis/v9 v9.21.0 github.com/stellar/go v0.0.0-20251210100531-aab2ea4aca88 github.com/stretchr/testify v1.11.1 @@ -12,30 +15,37 @@ require ( go.opentelemetry.io/otel v1.44.0 go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.44.0 go.opentelemetry.io/otel/sdk v1.44.0 - google.golang.org/grpc v1.82.1 + google.golang.org/grpc v1.81.1 google.golang.org/protobuf v1.36.11 ) require ( + github.com/beorn7/perks v1.0.1 // indirect github.com/cenkalti/backoff/v5 v5.0.3 // indirect github.com/cespare/xxhash/v2 v2.3.0 // indirect github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect - github.com/getkin/kin-openapi v0.145.0 // indirect github.com/go-logr/logr v1.4.3 // indirect github.com/go-logr/stdr v1.2.2 // indirect github.com/go-openapi/jsonpointer v0.22.5 // indirect github.com/go-openapi/swag/jsonname v0.25.5 // indirect + github.com/gorilla/mux v1.8.1 // indirect github.com/grpc-ecosystem/grpc-gateway/v2 v2.29.0 // indirect github.com/jackc/pgpassfile v1.0.0 // indirect github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect github.com/jackc/puddle/v2 v2.2.2 // indirect - github.com/klauspost/compress v1.17.6 // indirect + github.com/klauspost/compress v1.19.1 // indirect + github.com/kylelemons/godebug v1.1.0 // indirect + github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect github.com/oasdiff/yaml v0.1.1 // indirect github.com/oasdiff/yaml3 v0.0.14 // indirect github.com/pkg/errors v0.9.1 // indirect github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect + github.com/prometheus/client_model v0.6.2 // indirect + github.com/prometheus/common v0.70.1 // indirect + github.com/prometheus/procfs v0.21.1 // indirect github.com/santhosh-tekuri/jsonschema/v6 v6.0.2 // indirect github.com/stellar/go-xdr v0.0.0-20231122183749-b53fb00bcac2 // indirect + github.com/yuin/gopher-lua v1.1.1 // indirect go.opentelemetry.io/auto/sdk v1.2.1 // indirect go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.44.0 // indirect go.opentelemetry.io/otel/metric v1.44.0 // indirect @@ -43,10 +53,10 @@ require ( go.opentelemetry.io/proto/otlp v1.10.0 // indirect go.uber.org/atomic v1.11.0 // indirect golang.org/x/exp v0.0.0-20231006140011-7918f672742d // indirect - golang.org/x/net v0.55.0 // indirect - golang.org/x/sync v0.21.0 // indirect - golang.org/x/sys v0.45.0 // indirect - golang.org/x/text v0.39.0 // indirect + golang.org/x/net v0.57.0 // indirect + golang.org/x/sync v0.22.0 // indirect + golang.org/x/sys v0.47.0 // indirect + golang.org/x/text v0.40.0 // indirect google.golang.org/genproto/googleapis/api v0.0.0-20260526163538-3dc84a4a5aaa // indirect google.golang.org/genproto/googleapis/rpc v0.0.0-20260526163538-3dc84a4a5aaa // indirect gopkg.in/yaml.v3 v3.0.1 // indirect diff --git a/services/api/go.sum b/services/api/go.sum index b62bd20..b1d9b59 100644 --- a/services/api/go.sum +++ b/services/api/go.sum @@ -1,3 +1,7 @@ +github.com/alicebob/miniredis/v2 v2.38.0 h1:nZAzCR+Lj+Vxk4ZXzm2NuKq2O33RXj1XxJ2e2uP9jiw= +github.com/alicebob/miniredis/v2 v2.38.0/go.mod h1:TcL7YfarKPGDAthEtl5NBeHZfeUQj6OXMm/+iu5cLMM= +github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM= +github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw= github.com/bsm/ginkgo/v2 v2.12.0 h1:Ny8MWAHyOepLGlLKYmXG4IEkioBysk6GpaRTLC8zwWs= github.com/bsm/ginkgo/v2 v2.12.0/go.mod h1:SwYbGRRDovPVboqFv0tPTcG1sN61LM1Z4ARdbAV9g4c= github.com/bsm/gomega v1.27.10 h1:yeMWxP2pV2fG3FgAODIY8EiRE3dy0aeFYt4l7wh6yKA= @@ -9,6 +13,8 @@ github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XL github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc h1:U9qPSI2PIWSS1VwoXQT9A3Wy9MM3WgvqSxFWenqJduM= github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/dlclark/regexp2 v1.11.0 h1:G/nrcoOa7ZXlpoa/91N3X7mM3r8eIlMBBJZvsz/mxKI= +github.com/dlclark/regexp2 v1.11.0/go.mod h1:DHkYz0B9wPfa6wondMfaivmHpzrQ3v9q8cnmRbL6yW8= github.com/fsnotify/fsnotify v1.6.0 h1:n+5WquG0fcWoWp6xPWfHdbskMCQaFnG6PfBrh1Ky4HY= github.com/fsnotify/fsnotify v1.6.0/go.mod h1:sl3t1tCWJFWoRz9R8WJCbQihKKwmorjAbSClcnxKAGw= github.com/getkin/kin-openapi v0.145.0 h1:htBX+Q7SevVaCUqymFegUKzH2WCbewl9tsmyn2FMGWY= @@ -26,12 +32,16 @@ github.com/go-openapi/jsonpointer v0.22.5 h1:8on/0Yp4uTb9f4XvTrM2+1CPrV05QPZXu+r github.com/go-openapi/jsonpointer v0.22.5/go.mod h1:gyUR3sCvGSWchA2sUBJGluYMbe1zazrYWIkWPjjMUY0= github.com/go-openapi/swag/jsonname v0.25.5 h1:8p150i44rv/Drip4vWI3kGi9+4W9TdI3US3uUYSFhSo= github.com/go-openapi/swag/jsonname v0.25.5/go.mod h1:jNqqikyiAK56uS7n8sLkdaNY/uq6+D2m2LANat09pKU= +github.com/go-openapi/testify/v2 v2.4.0 h1:8nsPrHVCWkQ4p8h1EsRVymA2XABB4OT40gcvAu+voFM= +github.com/go-openapi/testify/v2 v2.4.0/go.mod h1:HCPmvFFnheKK2BuwSA0TbbdxJ3I16pjwMkYkP4Ywn54= github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek= github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6rSs7xps= github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= +github.com/gorilla/mux v1.8.1 h1:TuBL49tXwgrFYWhqrNgrUNEY92u81SPhu7sTdzQEiWY= +github.com/gorilla/mux v1.8.1/go.mod h1:AKf9I4AEqPTmMytcMc0KkNouC66V3BtZ4qD5fmWSiMQ= github.com/gorilla/schema v1.4.1 h1:jUg5hUjCSDZpNGLuXQOgIWGdlgrIdYvgQ0wZtdK1M3E= github.com/gorilla/schema v1.4.1/go.mod h1:Dg5SSm5PV60mhF2NFaTV1xuYYj8tV8NOPRo4FggUMnM= github.com/grpc-ecosystem/grpc-gateway/v2 v2.29.0 h1:5VipnvEpbqr2gA2VbM+nYVbkIF28c5ZQfqCBQ5g2xfk= @@ -44,16 +54,20 @@ github.com/jackc/pgx/v5 v5.10.0 h1:VhSvgU2jSli8o3AqIEOTJr7rZwAEUVo4E4XhR94Zfr0= github.com/jackc/pgx/v5 v5.10.0/go.mod h1:mal1tBGAFfLHvZzaYh77YS/eC6IX9OWbRV1QIIM0Jn4= github.com/jackc/puddle/v2 v2.2.2 h1:PR8nw+E/1w0GLuRFSmiioY6UooMp6KJv0/61nB7icHo= github.com/jackc/puddle/v2 v2.2.2/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4= -github.com/klauspost/compress v1.17.6 h1:60eq2E/jlfwQXtvZEeBUYADs+BwKBWURIY+Gj2eRGjI= -github.com/klauspost/compress v1.17.6/go.mod h1:/dCuZOvVtNoHsyb+cuJD3itjs3NbnF6KH9zAO4BDxPM= +github.com/klauspost/compress v1.19.1 h1:VsB4HPswih7mmZ8WleSFQ75c/Ui1M4trX5oAsJnhSlk= +github.com/klauspost/compress v1.19.1/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ= github.com/klauspost/cpuid/v2 v2.2.10 h1:tBs3QSyvjDyFTq3uoc/9xFpCuOsJQFNPiAhYdw2skhE= github.com/klauspost/cpuid/v2 v2.2.10/go.mod h1:hqwkgyIinND0mEev00jJYCxPNVRVXFQeu1XKlok6oO0= github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk= github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= +github.com/kylelemons/godebug v1.1.0 h1:RPNrshWIDI6G2gRW9EHilWtl7Z6Sb1BR0xunSBf0SNc= +github.com/kylelemons/godebug v1.1.0/go.mod h1:9/0rRGxNHcop5bhtWyNeEfOS8JIWk580+fNqagV/RAw= github.com/manucorporat/sse v0.0.0-20160126180136-ee05b128a739 h1:ykXz+pRRTibcSjG1yRhpdSHInF8yZY/mfn+Rz2Nd1rE= github.com/manucorporat/sse v0.0.0-20160126180136-ee05b128a739/go.mod h1:zUx1mhth20V3VKgL5jbd1BSQcW4Fy6Qs4PZvQwRFwzM= +github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA= +github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ= github.com/nxadm/tail v1.4.8 h1:nPr65rt6Y5JFSKQO7qToXr7pePgD6Gwiw05lkbyAQTE= github.com/nxadm/tail v1.4.8/go.mod h1:+ncqLTQzXmGhMZNUePPaPqPvBxHAIsmXswZKocGu+AU= github.com/oasdiff/yaml v0.1.1 h1:6nHx+pn9gBRM6YpBlFZFQGCCd1nuvqOBtTD3KKTgGxY= @@ -69,6 +83,14 @@ github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINE github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 h1:Jamvg5psRIccs7FGNTlIRMkT8wgtp5eCXdBlqhYGL6U= github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/prometheus/client_golang v1.24.1 h1:JnJkREXzWxUdCuPFpIWZiPispT9xVV59uiuyR2bPlnU= +github.com/prometheus/client_golang v1.24.1/go.mod h1:F+oSRECHg4sse5ucfYpYDeIv/hu68Zo0uoHKetWnzcE= +github.com/prometheus/client_model v0.6.2 h1:oBsgwpGs7iVziMvrGhE53c/GrLUsZdHnqNwqPLxwZyk= +github.com/prometheus/client_model v0.6.2/go.mod h1:y3m2F6Gdpfy6Ut/GBsUqTWZqCUvMVzSfMLjcu6wAwpE= +github.com/prometheus/common v0.70.1 h1:1HvjP4D5oL3t8RsPlwxA9onvvStjtIHYE5XuuwOi/PY= +github.com/prometheus/common v0.70.1/go.mod h1:VdFUQDMZK3VLkurFUVhia6uys/0suUp86TJz5qbJRhc= +github.com/prometheus/procfs v0.21.1 h1:GljZCt+zSTS+NZq88cyQ1LjZ+RCHp3uVuabBWA5+OJI= +github.com/prometheus/procfs v0.21.1/go.mod h1:aB55Cww9pdSJVHk0hUf0inxWyyjPogFIjmHKYgMKmtY= github.com/redis/go-redis/v9 v9.21.0 h1:FPBE4hhbAke+TLmcY3WkpbDffJEomdqPn3HYiqAtL9E= github.com/redis/go-redis/v9 v9.21.0/go.mod h1:v/M13XI1PVCDcm01VtPFOADfZtHf8YW3baQf57KlIkA= github.com/rogpeppe/go-internal v1.14.1 h1:UQB4HGPB6osV0SQTLymcB4TgvyWu6ZyliaW0tI/otEQ= @@ -92,6 +114,8 @@ github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= github.com/xdrpp/goxdr v0.1.1 h1:E1B2c6E8eYhOVyd7yEpOyopzTPirUeF6mVOfXfGyJyc= github.com/xdrpp/goxdr v0.1.1/go.mod h1:dXo1scL/l6s7iME1gxHWo2XCppbHEKZS7m/KyYWkNzA= +github.com/yuin/gopher-lua v1.1.1 h1:kYKnWBjvbNP4XLT3+bPEwAXJx262OhaHDWDVOPjL46M= +github.com/yuin/gopher-lua v1.1.1/go.mod h1:GBR0iDaNXjAgGg9zfCvksxSRnQx76gclCIb7kdAd1Pw= github.com/zeebo/xxh3 v1.1.0 h1:s7DLGDK45Dyfg7++yxI0khrfwq9661w9EN78eP/UZVs= github.com/zeebo/xxh3 v1.1.0/go.mod h1:IisAie1LELR4xhVinxWS5+zf1lA4p0MW4T+w+W07F5s= go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ64= @@ -118,24 +142,26 @@ go.uber.org/atomic v1.11.0 h1:ZvwS0R+56ePWxUNi+Atn9dWONBPp/AUETXlHW0DxSjE= go.uber.org/atomic v1.11.0/go.mod h1:LUxbIzbOniOlMKjJjyPfpl4v+PKK2cNJn91OQbhoJI0= go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto= go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE= +go.yaml.in/yaml/v2 v2.4.4 h1:tuyd0P+2Ont/d6e2rl3be67goVK4R6deVxCUX5vyPaQ= +go.yaml.in/yaml/v2 v2.4.4/go.mod h1:gMZqIpDtDqOfM0uNfy0SkpRhvUryYH0Z6wdMYcacYXQ= golang.org/x/exp v0.0.0-20231006140011-7918f672742d h1:jtJma62tbqLibJ5sFQz8bKtEM8rJBtfilJ2qTU199MI= golang.org/x/exp v0.0.0-20231006140011-7918f672742d/go.mod h1:ldy0pHrwJyGW56pPQzzkH36rKxoZW1tw7ZJpeKx+hdo= -golang.org/x/net v0.55.0 h1:bcvxaJn3e1U6InsFWt1JUq1aSjnRxLzT2rtD2KfkDF8= -golang.org/x/net v0.55.0/go.mod h1:L5U2KuzuOe1lY7Z+aWVIKK6qEeJXnXV9yzGA+WCHJww= -golang.org/x/sync v0.21.0 h1:HLII4xRRTtCRkxYp4HNFF0Js/Og6q2i++KXbg0gHCwM= -golang.org/x/sync v0.21.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0= -golang.org/x/sys v0.45.0 h1:dO4czNzziLiiXplLQgBCEpCvXQ3dnkn0SdaZSYdQ+FY= -golang.org/x/sys v0.45.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= -golang.org/x/text v0.39.0 h1:UbZz4pLOvn600D6Oh6GGEI6VAmndrEBLv8/6BEXzyus= -golang.org/x/text v0.39.0/go.mod h1:3UwRclnC2g0TU9x8PZiyfOajCd1zaUNHF9cvqcQZ+ZM= +golang.org/x/net v0.57.0 h1:K5+3DljvIuDG9/Jv9rvyMywYNFCQ9RSUY6OOTTkT+tE= +golang.org/x/net v0.57.0/go.mod h1:KpXc8iv+r3XplLAG/f7Jsf9RPszJzdR0f58q9vGOuEU= +golang.org/x/sync v0.22.0 h1:SZjpbeLmrCk4xhRSZFNZW5gFUeCeFgjekvI/+gfScek= +golang.org/x/sync v0.22.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0= +golang.org/x/sys v0.47.0 h1:o7XGOvZQCADBQQ4Y7VNq2dRWQR7JmOUW8Kxx4ZsNgWs= +golang.org/x/sys v0.47.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= +golang.org/x/text v0.40.0 h1:Ub2Z6/xjgF1WrYQz2nuITOEegKFtiIy+rieRJ5lHZKs= +golang.org/x/text v0.40.0/go.mod h1:hpnzDAfGV753zIKo+wk3u1bVKCGPbrnF7+7LBF/UHVY= gonum.org/v1/gonum v0.17.0 h1:VbpOemQlsSMrYmn7T2OUvQ4dqxQXU+ouZFQsZOx50z4= gonum.org/v1/gonum v0.17.0/go.mod h1:El3tOrEuMpv2UdMrbNlKEh9vd86bmQ6vqIcDwxEOc1E= google.golang.org/genproto/googleapis/api v0.0.0-20260526163538-3dc84a4a5aaa h1:Kjn0N0tCrDgiAFW+lGO4JZ3ck44CehvJQMAwj9QF0G8= google.golang.org/genproto/googleapis/api v0.0.0-20260526163538-3dc84a4a5aaa/go.mod h1:q4lMZS6kskjT5HvCPrnnypcDPVJqT/f4nfxmkE7gryY= google.golang.org/genproto/googleapis/rpc v0.0.0-20260526163538-3dc84a4a5aaa h1:mZHHdPZl0dbGHCflZgAq/Q468DWVFcU2whhB2KAo8fk= google.golang.org/genproto/googleapis/rpc v0.0.0-20260526163538-3dc84a4a5aaa/go.mod h1:4Hqkh8ycfw05ld/3BWL7rJOSfebL2Q+DVDeRgYgxUU8= -google.golang.org/grpc v1.82.1 h1:NnAxzGRA0677vCa4BUkOAnO5+FfQqVl9iUXeD0IqcGE= -google.golang.org/grpc v1.82.1/go.mod h1:yzTZ1TB1Z3SG+LIYaI+WiE8D5+PZ3ArnrSp8zF3+/ZA= +google.golang.org/grpc v1.81.1 h1:VnnIIZ88UzOOKLukQi+ImGz8O1Wdp8nAGGnvOfEIWQQ= +google.golang.org/grpc v1.81.1/go.mod h1:xGH9GfzOyMTGIOXBJmXt+BX/V0kcdQbdcuwQ/zNw42I= google.golang.org/protobuf v1.36.11 h1:fV6ZwhNocDyBLK0dj+fg8ektcVegBBuEolpbTQyBNVE= google.golang.org/protobuf v1.36.11/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= diff --git a/services/api/grpc/metrics.go b/services/api/grpc/metrics.go index b19c85c..31edef9 100644 --- a/services/api/grpc/metrics.go +++ b/services/api/grpc/metrics.go @@ -9,6 +9,7 @@ import ( "sync" "time" + "github.com/Depo-dev/trident/services/api/internal/metrics" "google.golang.org/grpc" "google.golang.org/grpc/status" ) @@ -48,7 +49,11 @@ func metricsUnaryInterceptor( ) error { start := time.Now() err := invoker(ctx, method, req, reply, cc, opts...) - clientMetrics.record(method, status.Code(err).String(), time.Since(start)) + elapsed := time.Since(start) + code := status.Code(err).String() + clientMetrics.record(method, code, elapsed) + metrics.GRPCClientRequestsTotal.WithLabelValues(method, code).Inc() + metrics.GRPCClientRequestDuration.WithLabelValues(method, code).Observe(elapsed.Seconds()) return err } diff --git a/services/api/grpc/metrics_prom_test.go b/services/api/grpc/metrics_prom_test.go new file mode 100644 index 0000000..bcd6dcb --- /dev/null +++ b/services/api/grpc/metrics_prom_test.go @@ -0,0 +1,44 @@ +package grpc + +import ( + "context" + "testing" + + "github.com/Depo-dev/trident/services/api/internal/metrics" + "github.com/prometheus/client_golang/prometheus/testutil" + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" +) + +// TestMetricsUnaryInterceptor_RecordsPrometheusMetrics verifies each call +// attempt updates trident_grpc_client_requests_total and +// trident_grpc_client_request_duration_seconds by method and status code +// (issue #58), alongside the pre-existing dependency-free counters. +func TestMetricsUnaryInterceptor_RecordsPrometheusMetrics(t *testing.T) { + const method = "/trident.Events/Stream" + + okBefore := testutil.ToFloat64(metrics.GRPCClientRequestsTotal.WithLabelValues(method, codes.OK.String())) + invoker := func(ctx context.Context, _ string, _, _ any, _ *grpc.ClientConn, _ ...grpc.CallOption) error { + return nil + } + if err := metricsUnaryInterceptor(context.Background(), method, nil, nil, nil, invoker); err != nil { + t.Fatalf("interceptor returned error: %v", err) + } + if got := testutil.ToFloat64(metrics.GRPCClientRequestsTotal.WithLabelValues(method, codes.OK.String())); got != okBefore+1 { + t.Errorf("OK requests total: want %v, got %v", okBefore+1, got) + } + + failCode := codes.Unavailable.String() + failBefore := testutil.ToFloat64(metrics.GRPCClientRequestsTotal.WithLabelValues(method, failCode)) + failInvoker := func(ctx context.Context, _ string, _, _ any, _ *grpc.ClientConn, _ ...grpc.CallOption) error { + return status.Error(codes.Unavailable, "backend down") + } + err := metricsUnaryInterceptor(context.Background(), method, nil, nil, nil, failInvoker) + if status.Code(err) != codes.Unavailable { + t.Fatalf("expected Unavailable error, got %v", err) + } + if got := testutil.ToFloat64(metrics.GRPCClientRequestsTotal.WithLabelValues(method, failCode)); got != failBefore+1 { + t.Errorf("Unavailable requests total: want %v, got %v", failBefore+1, got) + } +} diff --git a/services/api/handlers/apikeys.go b/services/api/handlers/apikeys.go index 9aca79c..224c3e3 100644 --- a/services/api/handlers/apikeys.go +++ b/services/api/handlers/apikeys.go @@ -18,6 +18,11 @@ import ( "github.com/redis/go-redis/v9" ) +// apiKeyQueryTimeout bounds the DB calls in the api-key admin handlers so a +// runaway query can't hold a pool connection for the request's full budget +// (issue #238). +const apiKeyQueryTimeout = 5 * time.Second + // APIKeyConfig wires the api-key handlers. type APIKeyConfig struct { AdminKey string @@ -116,9 +121,12 @@ func CreateAPIKey(cfg APIKeyConfig) http.HandlerFunc { createdBy = &req.CreatedBy } + ctx, cancel := context.WithTimeout(r.Context(), apiKeyQueryTimeout) + defer cancel() + var id string var createdAt time.Time - err := cfg.DB.QueryRow(r.Context(), + err := cfg.DB.QueryRow(ctx, `INSERT INTO api_keys (key_hash, key_prefix, label, network, rate_limit_tier, created_by) VALUES ($1, $2, $3, $4, $5, $6) RETURNING id, created_at`, @@ -153,7 +161,10 @@ func ListAPIKeys(cfg APIKeyConfig) http.HandlerFunc { return } - rows, err := cfg.DB.Query(r.Context(), + ctx, cancel := context.WithTimeout(r.Context(), apiKeyQueryTimeout) + defer cancel() + + rows, err := cfg.DB.Query(ctx, `SELECT id, key_prefix, label, network, rate_limit_tier, created_by, last_used_at, request_count, revoked_at, created_at FROM api_keys @@ -227,11 +238,14 @@ func UpdateAPIKey(cfg APIKeyConfig) http.HandlerFunc { return } + ctx, cancel := context.WithTimeout(r.Context(), apiKeyQueryTimeout) + defer cancel() + var k APIKeyResponse var lastUsedAt *time.Time var createdAt time.Time var keyHash string - err := cfg.DB.QueryRow(r.Context(), + err := cfg.DB.QueryRow(ctx, `UPDATE api_keys SET label = COALESCE($2, label), rate_limit_tier = COALESCE($3, rate_limit_tier) @@ -284,8 +298,11 @@ func DeleteAPIKey(cfg APIKeyConfig) http.HandlerFunc { return } + ctx, cancel := context.WithTimeout(r.Context(), apiKeyQueryTimeout) + defer cancel() + var keyHash string - err := cfg.DB.QueryRow(r.Context(), + err := cfg.DB.QueryRow(ctx, `UPDATE api_keys SET revoked_at = NOW() WHERE id = $1 AND revoked_at IS NULL diff --git a/services/api/handlers/contract_schemas.go b/services/api/handlers/contract_schemas.go index 882ff45..bf77286 100644 --- a/services/api/handlers/contract_schemas.go +++ b/services/api/handlers/contract_schemas.go @@ -7,6 +7,7 @@ import ( "net/http" "sort" "strings" + "time" "github.com/Depo-dev/trident/services/api/internal/httputil" "github.com/Depo-dev/trident/services/api/middleware" @@ -15,6 +16,11 @@ import ( "github.com/jackc/pgx/v5/pgconn" ) +// schemaQueryTimeout bounds the DB calls in ContractEventSchemas so a +// runaway query can't hold a pool connection for the request's full budget +// (issue #238). +const schemaQueryTimeout = 5 * time.Second + const unknownSchemaCodeHash = "unknown" type SchemaRegistryDB interface { @@ -94,19 +100,22 @@ func ContractEventSchemas(db SchemaRegistryDB) http.HandlerFunc { return } + ctx, cancel := context.WithTimeout(r.Context(), schemaQueryTimeout) + defer cancel() + network := middleware.NetworkFromContext(r.Context()) - codeHash, err := resolveContractCodeHash(r.Context(), db, contractID, network) + codeHash, err := resolveContractCodeHash(ctx, db, contractID, network) if err != nil { httputil.WriteErrorCtx(r.Context(), w, http.StatusServiceUnavailable, httputil.INTERNAL, "failed to load contract schema") return } - schemas, err := observeContractSchemas(r.Context(), db, contractID, network) + schemas, err := observeContractSchemas(ctx, db, contractID, network) if err != nil { httputil.WriteErrorCtx(r.Context(), w, http.StatusServiceUnavailable, httputil.INTERNAL, "failed to load contract schema") return } - if err := persistContractSchemas(r.Context(), db, contractID, network, codeHash, schemas); err != nil { + if err := persistContractSchemas(ctx, db, contractID, network, codeHash, schemas); err != nil { httputil.WriteErrorCtx(r.Context(), w, http.StatusServiceUnavailable, httputil.INTERNAL, "failed to persist contract schema") return } diff --git a/services/api/handlers/contract_spec.go b/services/api/handlers/contract_spec.go index d87ec7d..05fa2a4 100644 --- a/services/api/handlers/contract_spec.go +++ b/services/api/handlers/contract_spec.go @@ -5,6 +5,7 @@ import ( "encoding/json" "errors" "net/http" + "time" "github.com/Depo-dev/trident/services/api/internal/httputil" "github.com/Depo-dev/trident/services/api/middleware" @@ -12,6 +13,11 @@ import ( "github.com/jackc/pgx/v5" ) +// contractSpecQueryTimeout bounds the DB call in ContractSpec so a runaway +// query can't hold a pool connection for the request's full budget (issue +// #238). +const contractSpecQueryTimeout = 5 * time.Second + // ContractSpecFunction is one function captured from a contract's parsed // spec (issue #260). type ContractSpecFunction struct { @@ -46,8 +52,11 @@ func ContractSpec(db SchemaRegistryDB) http.HandlerFunc { return } + ctx, cancel := context.WithTimeout(r.Context(), contractSpecQueryTimeout) + defer cancel() + network := middleware.NetworkFromContext(r.Context()) - resp, err := loadContractSpec(r.Context(), db, contractID, network) + resp, err := loadContractSpec(ctx, db, contractID, network) if errors.Is(err, pgx.ErrNoRows) { httputil.WriteErrorCtx(r.Context(), w, http.StatusNotFound, httputil.NOT_FOUND, "no spec recorded for this contract") return diff --git a/services/api/handlers/contract_storage.go b/services/api/handlers/contract_storage.go index 538e2ac..db631ba 100644 --- a/services/api/handlers/contract_storage.go +++ b/services/api/handlers/contract_storage.go @@ -1,6 +1,7 @@ package handlers import ( + "context" "encoding/json" "net/http" "time" @@ -10,6 +11,11 @@ import ( "github.com/Depo-dev/trident/services/api/validation" ) +// contractStorageQueryTimeout bounds the DB calls in ContractStorageLatest/ +// ContractStorageHistory so a runaway query can't hold a pool connection for +// the request's full budget (issue #238). +const contractStorageQueryTimeout = 5 * time.Second + // ContractStorageValue is one contract-storage key's value at a given ledger // (issue #270). type ContractStorageValue struct { @@ -42,8 +48,11 @@ func ContractStorageLatest(db SchemaRegistryDB) http.HandlerFunc { return } + ctx, cancel := context.WithTimeout(r.Context(), contractStorageQueryTimeout) + defer cancel() + network := middleware.NetworkFromContext(r.Context()) - rows, err := db.Query(r.Context(), ` + rows, err := db.Query(ctx, ` SELECT DISTINCT ON (storage_key) storage_key, key_json, value_json, ledger_sequence, created_at FROM contract_storage_snapshots @@ -102,8 +111,11 @@ func ContractStorageHistory(db SchemaRegistryDB) http.HandlerFunc { return } + ctx, cancel := context.WithTimeout(r.Context(), contractStorageQueryTimeout) + defer cancel() + network := middleware.NetworkFromContext(r.Context()) - rows, err := db.Query(r.Context(), ` + rows, err := db.Query(ctx, ` SELECT storage_key, key_json, value_json, ledger_sequence, created_at FROM contract_storage_snapshots WHERE contract_id = $1 AND network = $2 AND storage_key = $3 diff --git a/services/api/handlers/contract_test.go b/services/api/handlers/contract_test.go index fa46cce..61edc14 100644 --- a/services/api/handlers/contract_test.go +++ b/services/api/handlers/contract_test.go @@ -125,7 +125,7 @@ func TestContract_OpenAPIResponseValidation(t *testing.T) { req := httptest.NewRequest(http.MethodGet, "/v1/health", nil) rr := httptest.NewRecorder() - handlers.Health(nil, nil, nil)(rr, req) + handlers.Health()(rr, req) if rr.Code == http.StatusOK { validateResponseAgainstSchema(t, doc, http.MethodGet, "/v1/health", rr.Code, rr.Body.Bytes()) @@ -239,6 +239,7 @@ func TestContract_RouteParity(t *testing.T) { // Note: Admin routes and webhook routes are also currently not documented in OpenAPI expectedRoutes := map[string]bool{ "GET /v1/health": true, + "GET /v1/ready": true, "GET /v1/events": true, "POST /v1/events/batch": true, "GET /v1/events/{id}": true, diff --git a/services/api/handlers/contract_xcache_test.go b/services/api/handlers/contract_xcache_test.go new file mode 100644 index 0000000..c5fec5e --- /dev/null +++ b/services/api/handlers/contract_xcache_test.go @@ -0,0 +1,145 @@ +package handlers + +import ( + "context" + "net/http" + "net/http/httptest" + "testing" + "time" + + "github.com/Depo-dev/trident/services/api/internal/contracttest" + "github.com/Depo-dev/trident/services/api/middleware" + "github.com/alicebob/miniredis/v2" + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgconn" + "github.com/redis/go-redis/v9" +) + +// wrapRateLimited mirrors main.go's real middleware chain closely enough +// for contract testing: GET /v1/stats/contracts's documented 200 response +// requires the X-RateLimit-* headers TieredRateLimit adds, which the bare +// handler under test doesn't set on its own (issue #242). +func wrapRateLimited(h http.Handler) http.Handler { + cfg := middleware.RateLimitConfig{ + SliderFn: func(_ context.Context, _ string, limit, _ int64) (bool, int64, error) { + return true, 1, nil + }, + Tiers: map[string]middleware.TierConfig{"free": {RPS: 1000, Window: time.Second}}, + } + return middleware.TieredRateLimit(cfg)(h) +} + +type xcacheMissDB struct{} + +func (xcacheMissDB) Ping(_ context.Context) error { return nil } +func (xcacheMissDB) QueryRow(_ context.Context, _ string, _ ...any) pgx.Row { + return nil +} +func (xcacheMissDB) Query(_ context.Context, _ string, _ ...any) (pgx.Rows, error) { + return &noRowsResult{}, nil +} + +// noRowsResult is a zero-row pgx.Rows stand-in for the ContractsStats +// live-aggregation query (issue #242's X-Cache MISS test) — no contracts are +// returned, only that the query executed and X-Cache: MISS was set. +type noRowsResult struct{ closed bool } + +func (r *noRowsResult) Close() { r.closed = true } +func (r *noRowsResult) Err() error { return nil } +func (r *noRowsResult) CommandTag() pgconn.CommandTag { return pgconn.CommandTag{} } +func (r *noRowsResult) FieldDescriptions() []pgconn.FieldDescription { return nil } +func (r *noRowsResult) Next() bool { return false } +func (r *noRowsResult) Scan(_ ...any) error { return nil } +func (r *noRowsResult) Values() ([]any, error) { return nil, nil } +func (r *noRowsResult) RawValues() [][]byte { return nil } +func (r *noRowsResult) Conn() *pgx.Conn { return nil } + +func newMiniredisClient(t *testing.T) *redis.Client { + t.Helper() + mr, err := miniredis.Run() + if err != nil { + t.Fatalf("start miniredis: %v", err) + } + t.Cleanup(mr.Close) + return redis.NewClient(&redis.Options{Addr: mr.Addr()}) +} + +// contractsStatsExplicitRangeReq builds a request with an explicit ledger +// range so ContractsStats takes the single-query live-aggregation path +// (queryContractStats) rather than the rollup fallback — keeps the DB mock +// trivial (issue #242). +func contractsStatsExplicitRangeReq(t *testing.T) *http.Request { + t.Helper() + req := httptest.NewRequest(http.MethodGet, "/v1/stats/contracts?from_ledger=0&to_ledger=1000000&network=testnet&limit=10", nil) + req.URL.Scheme = "http" + req.URL.Host = "localhost:3000" + req.Host = "localhost:3000" + req.Header.Set("X-API-Key", "contract-test-key") + return req +} + +// TestContractsStats_XCache_Miss verifies a cache-miss response sets +// X-Cache: MISS and conforms to GET /v1/stats/contracts's documented +// contract (issue #242). +func TestContractsStats_XCache_Miss(t *testing.T) { + rdb := newMiniredisClient(t) + req := contractsStatsExplicitRangeReq(t) + + rr := httptest.NewRecorder() + wrapRateLimited(ContractsStats(xcacheMissDB{}, rdb)).ServeHTTP(rr, req) + + if rr.Code != http.StatusOK { + t.Fatalf("want 200, got %d: %s", rr.Code, rr.Body.String()) + } + if got := rr.Header().Get("X-Cache"); got != "MISS" { + t.Errorf("X-Cache: want MISS, got %q", got) + } + + doc := contracttest.LoadSpec(t) + router := contracttest.NewRouter(t, doc) + contracttest.ValidateResponse(t, router, req, rr.Code, rr.Header(), rr.Body.Bytes()) +} + +// TestContractsStats_XCache_Hit verifies a cache-hit response sets +// X-Cache: HIT, is served without touching the DB, and conforms to the same +// documented contract as a MISS (issue #242). +func TestContractsStats_XCache_Hit(t *testing.T) { + rdb := newMiniredisClient(t) + req := contractsStatsExplicitRangeReq(t) + + cacheKey := "stats:contracts:testnet:0:1000000:10" + cachedBody := `{"contracts":[],"from_ledger":0,"to_ledger":1000000,"network":"testnet","generated_at":"` + + time.Now().UTC().Format(time.RFC3339) + `"}` + if err := rdb.Set(context.Background(), cacheKey, cachedBody, time.Minute).Err(); err != nil { + t.Fatalf("seed cache: %v", err) + } + + // A DB that panics if queried — a HIT must never reach it. + var panicsIfQueried DBPool = panicDB{t} + + rr := httptest.NewRecorder() + wrapRateLimited(ContractsStats(panicsIfQueried, rdb)).ServeHTTP(rr, req) + + if rr.Code != http.StatusOK { + t.Fatalf("want 200, got %d: %s", rr.Code, rr.Body.String()) + } + if got := rr.Header().Get("X-Cache"); got != "HIT" { + t.Errorf("X-Cache: want HIT, got %q", got) + } + + doc := contracttest.LoadSpec(t) + router := contracttest.NewRouter(t, doc) + contracttest.ValidateResponse(t, router, req, rr.Code, rr.Header(), rr.Body.Bytes()) +} + +type panicDB struct{ t *testing.T } + +func (p panicDB) Ping(_ context.Context) error { p.t.Fatal("unexpected Ping on cache HIT"); return nil } +func (p panicDB) QueryRow(_ context.Context, _ string, _ ...any) pgx.Row { + p.t.Fatal("unexpected QueryRow on cache HIT") + return nil +} +func (p panicDB) Query(_ context.Context, _ string, _ ...any) (pgx.Rows, error) { + p.t.Fatal("unexpected Query on cache HIT") + return nil, nil +} diff --git a/services/api/handlers/health.go b/services/api/handlers/health.go index 99e2ec2..7196992 100644 --- a/services/api/handlers/health.go +++ b/services/api/handlers/health.go @@ -49,27 +49,48 @@ type EventsLister interface { ListEvents(ctx context.Context, in *gen.ListEventsRequest, opts ...grpc.CallOption) (*gen.ListEventsResponse, error) } -// HealthChecks holds the per-dependency check results. -type HealthChecks struct { +// LivenessResponse is the JSON body for GET /v1/health. +type LivenessResponse struct { + Status string `json:"status"` +} + +// Health handles GET /v1/health — a liveness check (issue #243). +// +// Deliberately cheap: no dependency calls (no DB/Redis/gRPC), just confirms +// the process is up and serving requests. This is what Kubernetes' liveness +// probe should hit — restarting the pod never fixes an unreachable external +// dependency, so liveness must not fail because Postgres/Redis/the gRPC +// backend is down. For that, see Ready (GET /v1/ready). +func Health() http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + writeJSON(w, http.StatusOK, LivenessResponse{Status: "ok"}) + } +} + +// ReadyChecks holds the per-dependency check results. +type ReadyChecks struct { Postgres string `json:"postgres"` Redis string `json:"redis"` GRPCAPI string `json:"grpc_api"` } -// HealthResponse is the JSON body for GET /v1/health. -type HealthResponse struct { - Status string `json:"status"` - IndexerLag *int64 `json:"indexer_lag"` - Checks HealthChecks `json:"checks"` +// ReadyResponse is the JSON body for GET /v1/ready. +type ReadyResponse struct { + Status string `json:"status"` + IndexerLag *int64 `json:"indexer_lag"` + Checks ReadyChecks `json:"checks"` } -// Health handles GET /v1/health. +// Ready handles GET /v1/ready — a readiness check (issue #243). // // Runs Postgres, Redis, and gRPC checks concurrently with a shared -// 3-second timeout. Returns 200 when all checks pass, 503 when any fail. -// The indexer_lag field is populated from system_state when Postgres is -// healthy and the chain tip is available in the cache; null otherwise. -func Health(db DBPool, redisClient RedisPinger, grpcClient EventsLister) http.HandlerFunc { +// 3-second timeout. Returns 200 when all checks pass, 503 when any fail — +// this is what Kubernetes' readiness probe should hit, so a pod with a +// broken dependency is pulled out of the Service's endpoint rotation +// instead of continuing to receive traffic it can't serve. The indexer_lag +// field is populated from system_state when Postgres is healthy and the +// chain tip is available in the cache; null otherwise. +func Ready(db DBPool, redisClient RedisPinger, grpcClient EventsLister) http.HandlerFunc { return func(w http.ResponseWriter, r *http.Request) { parentCtx := r.Context() @@ -106,8 +127,8 @@ func Health(db DBPool, redisClient RedisPinger, grpcClient EventsLister) http.Ha wg.Wait() - resp := HealthResponse{ - Checks: HealthChecks{ + resp := ReadyResponse{ + Checks: ReadyChecks{ Postgres: resultString(pgErr), Redis: resultString(redisErr), GRPCAPI: resultString(grpcErr), diff --git a/services/api/handlers/health_test.go b/services/api/handlers/health_test.go index ffef4a1..301a74f 100644 --- a/services/api/handlers/health_test.go +++ b/services/api/handlers/health_test.go @@ -1,4 +1,4 @@ -package handlers_test +package handlers import ( "context" @@ -6,184 +6,237 @@ import ( "errors" "net/http" "net/http/httptest" - "strings" "testing" "github.com/Depo-dev/trident/services/api/gen" - "github.com/Depo-dev/trident/services/api/handlers" + "github.com/Depo-dev/trident/services/api/internal/contracttest" "github.com/jackc/pgx/v5" "github.com/redis/go-redis/v9" "google.golang.org/grpc" - "google.golang.org/grpc/codes" - "google.golang.org/grpc/status" ) -type fakeHealthDB struct { +// fakeEventsClient implements EventsLister for Ready() tests (package +// handlers, not handlers_test — kept local rather than reusing +// events_test.go's MockEventsClient, which lives in the separate +// handlers_test package and isn't visible here). +type fakeEventsClient struct { + listEvents func(context.Context, *gen.ListEventsRequest) (*gen.ListEventsResponse, error) +} + +func (f *fakeEventsClient) ListEvents(ctx context.Context, in *gen.ListEventsRequest, _ ...grpc.CallOption) (*gen.ListEventsResponse, error) { + return f.listEvents(ctx, in) +} + +// healthMockDB implements DBPool for Ready() tests. When pingErr is set, +// Ping fails and QueryRow is never expected to matter (checkPostgres +// returns before calling it in production code paths that check the ping +// error first — this double doesn't need a real Row in that case). +type healthMockDB struct { pingErr error lastLedger *int64 - rowErr error + scanErr error } -func (db fakeHealthDB) Ping(context.Context) error { - return db.pingErr +func (m *healthMockDB) Ping(_ context.Context) error { return m.pingErr } +func (m *healthMockDB) QueryRow(_ context.Context, _ string, _ ...any) pgx.Row { + return &healthMockRow{m: m} } +func (m *healthMockDB) Query(_ context.Context, _ string, _ ...any) (pgx.Rows, error) { + return nil, nil +} + +type healthMockRow struct{ m *healthMockDB } -func (db fakeHealthDB) QueryRow(context.Context, string, ...any) pgx.Row { - return fakeHealthRow{lastLedger: db.lastLedger, err: db.rowErr} +func (r *healthMockRow) Scan(dest ...any) error { + if r.m.scanErr != nil { + return r.m.scanErr + } + *dest[0].(**int64) = r.m.lastLedger + return nil } -func (db fakeHealthDB) Query(context.Context, string, ...any) (pgx.Rows, error) { - return nil, nil +// healthyRedis and unhealthyRedis satisfy RedisPinger. +type fakeRedisPinger struct{ err error } + +func (f fakeRedisPinger) Ping(ctx context.Context) *redis.StatusCmd { + return redis.NewStatusResult("PONG", f.err) } -type fakeHealthRow struct { - lastLedger *int64 - err error +func healthReadyReq(path string) *http.Request { + req := httptest.NewRequest(http.MethodGet, path, nil) + req.URL.Scheme = "http" + req.URL.Host = "localhost:3000" + req.Host = "localhost:3000" + return req } -func (r fakeHealthRow) Scan(dest ...any) error { - if r.err != nil { - return r.err +// TestHealth_AlwaysReturns200 verifies GET /v1/health is a cheap liveness +// check: no dependencies wired at all, always 200 (issue #243). +func TestHealth_AlwaysReturns200(t *testing.T) { + req := healthReadyReq("/v1/health") + rr := httptest.NewRecorder() + Health().ServeHTTP(rr, req) + + if rr.Code != http.StatusOK { + t.Fatalf("want 200, got %d", rr.Code) } - if len(dest) == 0 || r.lastLedger == nil { - return nil + var body LivenessResponse + if err := json.Unmarshal(rr.Body.Bytes(), &body); err != nil { + t.Fatalf("decode body: %v", err) } - target, ok := dest[0].(**int64) - if !ok { - return nil + if body.Status != "ok" { + t.Errorf("status: want ok, got %q", body.Status) } - value := *r.lastLedger - *target = &value - return nil -} -type fakeRedisPinger struct { - err error + doc := contracttest.LoadSpec(t) + router := contracttest.NewRouter(t, doc) + contracttest.ValidateResponse(t, router, req, rr.Code, rr.Header(), rr.Body.Bytes()) } -func (p fakeRedisPinger) Ping(ctx context.Context) *redis.StatusCmd { - cmd := redis.NewStatusCmd(ctx) - if p.err != nil { - cmd.SetErr(p.err) - return cmd +// TestReady_AllHealthy_Returns200 verifies GET /v1/ready reports 200 with +// status ok when Postgres, Redis, and gRPC all succeed (issue #243). +func TestReady_AllHealthy_Returns200(t *testing.T) { + ledger := int64(42) + db := &healthMockDB{lastLedger: &ledger} + rdb := fakeRedisPinger{} + grpcClient := &fakeEventsClient{ + listEvents: func(_ context.Context, _ *gen.ListEventsRequest) (*gen.ListEventsResponse, error) { + return &gen.ListEventsResponse{}, nil + }, } - cmd.SetVal("PONG") - return cmd -} -type fakeHealthEventsClient struct { - err error -} + req := healthReadyReq("/v1/ready") + rr := httptest.NewRecorder() + Ready(db, rdb, grpcClient).ServeHTTP(rr, req) -func (c fakeHealthEventsClient) ListEvents(ctx context.Context, in *gen.ListEventsRequest, opts ...grpc.CallOption) (*gen.ListEventsResponse, error) { - if c.err != nil { - return nil, c.err + if rr.Code != http.StatusOK { + t.Fatalf("want 200, got %d: %s", rr.Code, rr.Body.String()) + } + var body ReadyResponse + if err := json.Unmarshal(rr.Body.Bytes(), &body); err != nil { + t.Fatalf("decode body: %v", err) } - return &gen.ListEventsResponse{}, nil + if body.Status != "ok" { + t.Errorf("status: want ok, got %q", body.Status) + } + if body.Checks.Postgres != "ok" || body.Checks.Redis != "ok" || body.Checks.GRPCAPI != "ok" { + t.Errorf("want all checks ok, got %+v", body.Checks) + } + + doc := contracttest.LoadSpec(t) + router := contracttest.NewRouter(t, doc) + contracttest.ValidateResponse(t, router, req, rr.Code, rr.Header(), rr.Body.Bytes()) } -func TestHealthHandler_TableDriven(t *testing.T) { - dbErr := errors.New("database unavailable") - redisErr := errors.New("redis unavailable") - grpcErr := status.Error(codes.Unavailable, "grpc unavailable") - - tests := []struct { - name string - db handlers.DBPool - redis handlers.RedisPinger - grpc handlers.EventsLister - wantStatus int - wantBody handlers.HealthResponse - check func(t *testing.T, body handlers.HealthResponse) - }{ - { - name: "all dependencies reachable", - db: fakeHealthDB{}, - redis: fakeRedisPinger{}, - grpc: fakeHealthEventsClient{}, - wantStatus: http.StatusOK, - wantBody: handlers.HealthResponse{ - Status: "ok", - Checks: handlers.HealthChecks{ - Postgres: "ok", - Redis: "ok", - GRPCAPI: "ok", - }, - }, +// TestReady_PostgresDown_Returns503 verifies a Postgres ping failure alone +// degrades the whole readiness check to 503 (issue #243). +func TestReady_PostgresDown_Returns503(t *testing.T) { + db := &healthMockDB{pingErr: errors.New("connection refused")} + rdb := fakeRedisPinger{} + grpcClient := &fakeEventsClient{ + listEvents: func(_ context.Context, _ *gen.ListEventsRequest) (*gen.ListEventsResponse, error) { + return &gen.ListEventsResponse{}, nil }, - { - name: "db unreachable returns degraded 503", - db: fakeHealthDB{pingErr: dbErr}, - redis: fakeRedisPinger{}, - grpc: fakeHealthEventsClient{}, - wantStatus: http.StatusServiceUnavailable, - check: func(t *testing.T, body handlers.HealthResponse) { - if body.Status != "degraded" { - t.Fatalf("status: got %q, want degraded", body.Status) - } - if !strings.Contains(body.Checks.Postgres, dbErr.Error()) { - t.Fatalf("postgres check: got %q, want db error", body.Checks.Postgres) - } - }, - }, - { - name: "grpc unreachable reflected in checks", - db: fakeHealthDB{}, - redis: fakeRedisPinger{}, - grpc: fakeHealthEventsClient{err: grpcErr}, - wantStatus: http.StatusServiceUnavailable, - check: func(t *testing.T, body handlers.HealthResponse) { - if body.Status != "degraded" { - t.Fatalf("status: got %q, want degraded", body.Status) - } - if !strings.Contains(body.Checks.GRPCAPI, "Unavailable") { - t.Fatalf("grpc_api check: got %q, want Unavailable", body.Checks.GRPCAPI) - } - }, + } + + req := healthReadyReq("/v1/ready") + rr := httptest.NewRecorder() + Ready(db, rdb, grpcClient).ServeHTTP(rr, req) + + assertDegraded(t, rr, "postgres") + + doc := contracttest.LoadSpec(t) + router := contracttest.NewRouter(t, doc) + contracttest.ValidateResponse(t, router, req, rr.Code, rr.Header(), rr.Body.Bytes()) +} + +// TestReady_RedisDown_Returns503 verifies a Redis ping failure alone +// degrades the whole readiness check to 503 (issue #243). +func TestReady_RedisDown_Returns503(t *testing.T) { + ledger := int64(1) + db := &healthMockDB{lastLedger: &ledger} + rdb := fakeRedisPinger{err: errors.New("dial tcp: connection refused")} + grpcClient := &fakeEventsClient{ + listEvents: func(_ context.Context, _ *gen.ListEventsRequest) (*gen.ListEventsResponse, error) { + return &gen.ListEventsResponse{}, nil }, - { - name: "redis unreachable reflected in checks", - db: fakeHealthDB{}, - redis: fakeRedisPinger{err: redisErr}, - grpc: fakeHealthEventsClient{}, - wantStatus: http.StatusServiceUnavailable, - check: func(t *testing.T, body handlers.HealthResponse) { - if body.Status != "degraded" { - t.Fatalf("status: got %q, want degraded", body.Status) - } - if !strings.Contains(body.Checks.Redis, redisErr.Error()) { - t.Fatalf("redis check: got %q, want redis error", body.Checks.Redis) - } - }, + } + + req := healthReadyReq("/v1/ready") + rr := httptest.NewRecorder() + Ready(db, rdb, grpcClient).ServeHTTP(rr, req) + + assertDegraded(t, rr, "redis") + + doc := contracttest.LoadSpec(t) + router := contracttest.NewRouter(t, doc) + contracttest.ValidateResponse(t, router, req, rr.Code, rr.Header(), rr.Body.Bytes()) +} + +// TestReady_GRPCDown_Returns503 verifies a gRPC backend failure alone +// degrades the whole readiness check to 503 (issue #243). +func TestReady_GRPCDown_Returns503(t *testing.T) { + ledger := int64(1) + db := &healthMockDB{lastLedger: &ledger} + rdb := fakeRedisPinger{} + grpcClient := &fakeEventsClient{ + listEvents: func(_ context.Context, _ *gen.ListEventsRequest) (*gen.ListEventsResponse, error) { + return nil, errors.New("backend unreachable") }, } - for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - req := httptest.NewRequest(http.MethodGet, "/v1/health", nil) - rr := httptest.NewRecorder() - - handlers.Health(tt.db, tt.redis, tt.grpc)(rr, req) - - if rr.Code != tt.wantStatus { - t.Fatalf("status: got %d, want %d; body: %s", rr.Code, tt.wantStatus, rr.Body.String()) - } - - var body handlers.HealthResponse - if err := json.NewDecoder(rr.Body).Decode(&body); err != nil { - t.Fatalf("decode response: %v", err) - } - if tt.wantBody.Status != "" { - if body.Status != tt.wantBody.Status { - t.Fatalf("body.status: got %q, want %q", body.Status, tt.wantBody.Status) - } - if body.Checks != tt.wantBody.Checks { - t.Fatalf("checks: got %+v, want %+v", body.Checks, tt.wantBody.Checks) - } - } - if tt.check != nil { - tt.check(t, body) - } - }) + req := healthReadyReq("/v1/ready") + rr := httptest.NewRecorder() + Ready(db, rdb, grpcClient).ServeHTTP(rr, req) + + assertDegraded(t, rr, "grpc_api") + + doc := contracttest.LoadSpec(t) + router := contracttest.NewRouter(t, doc) + contracttest.ValidateResponse(t, router, req, rr.Code, rr.Header(), rr.Body.Bytes()) +} + +// TestReady_NilDependencies_Returns503 verifies unconfigured dependencies +// (nil db/redis/grpc, e.g. at cold start before DATABASE_URL connects) are +// treated as failures, not silently skipped (issue #243). +func TestReady_NilDependencies_Returns503(t *testing.T) { + req := healthReadyReq("/v1/ready") + rr := httptest.NewRecorder() + Ready(nil, nil, nil).ServeHTTP(rr, req) + + assertDegraded(t, rr, "postgres") + + var body ReadyResponse + if err := json.Unmarshal(rr.Body.Bytes(), &body); err != nil { + t.Fatalf("decode body: %v", err) + } + if body.Checks.Redis == "ok" || body.Checks.GRPCAPI == "ok" { + t.Errorf("want redis and grpc_api also reported as failing, got %+v", body.Checks) + } +} + +func assertDegraded(t *testing.T, rr *httptest.ResponseRecorder, failingCheck string) { + t.Helper() + if rr.Code != http.StatusServiceUnavailable { + t.Fatalf("want 503, got %d: %s", rr.Code, rr.Body.String()) + } + var body ReadyResponse + if err := json.Unmarshal(rr.Body.Bytes(), &body); err != nil { + t.Fatalf("decode body: %v", err) + } + if body.Status != "degraded" { + t.Errorf("status: want degraded, got %q", body.Status) + } + var got string + switch failingCheck { + case "postgres": + got = body.Checks.Postgres + case "redis": + got = body.Checks.Redis + case "grpc_api": + got = body.Checks.GRPCAPI + } + if got == "ok" || got == "" { + t.Errorf("checks.%s: want a failure reason, got %q", failingCheck, got) } } diff --git a/services/api/handlers/stats.go b/services/api/handlers/stats.go index 5f9da80..142008e 100644 --- a/services/api/handlers/stats.go +++ b/services/api/handlers/stats.go @@ -631,7 +631,10 @@ func queryContractStats(ctx context.Context, db DBPool, params *validation.Query } defer rows.Close() - var stats []*ContractStats + // Non-nil so a zero-row result serializes as JSON [] rather than null + // (issue #242) — the OpenAPI spec documents ContractStatsResponse.contracts + // as a non-nullable array. + stats := []*ContractStats{} for rows.Next() { var cs ContractStats var lastSeenAt time.Time @@ -711,7 +714,10 @@ func queryContractStatsFromRollup(ctx context.Context, db DBPool, params *valida } defer rows.Close() - var stats []*ContractStats + // Non-nil so a zero-row result serializes as JSON [] rather than null + // (issue #242) — the OpenAPI spec documents ContractStatsResponse.contracts + // as a non-nullable array. + stats := []*ContractStats{} for rows.Next() { var cs ContractStats var lastSeenAt time.Time diff --git a/services/api/handlers/token_metadata.go b/services/api/handlers/token_metadata.go index 4011aa3..95c5267 100644 --- a/services/api/handlers/token_metadata.go +++ b/services/api/handlers/token_metadata.go @@ -1,8 +1,10 @@ package handlers import ( + "context" "errors" "net/http" + "time" "github.com/Depo-dev/trident/services/api/internal/httputil" "github.com/Depo-dev/trident/services/api/middleware" @@ -10,6 +12,11 @@ import ( "github.com/jackc/pgx/v5" ) +// tokenMetadataQueryTimeout bounds the DB call in TokenMetadata so a +// runaway query can't hold a pool connection for the request's full budget +// (issue #238). +const tokenMetadataQueryTimeout = 5 * time.Second + // TokenMetadataResponse is the JSON body for GET /v1/contracts/{id}/metadata. // // Name/Symbol/Decimals/ResolvedAt are null whenever IsToken is false — either @@ -55,7 +62,10 @@ func TokenMetadata(db DBPool) http.HandlerFunc { decimals *int32 resolvedAt *string ) - err := db.QueryRow(r.Context(), ` + ctx, cancel := context.WithTimeout(r.Context(), tokenMetadataQueryTimeout) + defer cancel() + + err := db.QueryRow(ctx, ` SELECT is_token, name, symbol, decimals, resolved_at::text FROM token_metadata WHERE contract_id = $1 AND network = $2 diff --git a/services/api/handlers/token_metadata_slowquery_test.go b/services/api/handlers/token_metadata_slowquery_test.go new file mode 100644 index 0000000..83bfe30 --- /dev/null +++ b/services/api/handlers/token_metadata_slowquery_test.go @@ -0,0 +1,102 @@ +package handlers + +import ( + "context" + "net/http" + "net/http/httptest" + "testing" + "time" + + "github.com/jackc/pgx/v5" +) + +// slowQueryDB is a DBPool whose QueryRow blocks until the caller's context +// is done, simulating a runaway/slow query for issue #238's "test +// demonstrating slow query cancellation and pool recovery" acceptance +// criterion. +type slowQueryDB struct{} + +func (slowQueryDB) Ping(_ context.Context) error { return nil } + +func (slowQueryDB) QueryRow(ctx context.Context, _ string, _ ...any) pgx.Row { + return slowQueryRow{ctx: ctx} +} + +func (slowQueryDB) Query(_ context.Context, _ string, _ ...any) (pgx.Rows, error) { + return nil, nil +} + +type slowQueryRow struct{ ctx context.Context } + +func (r slowQueryRow) Scan(_ ...any) error { + <-r.ctx.Done() + return r.ctx.Err() +} + +// fastQueryDB is a DBPool that resolves immediately with a not-found result, +// standing in for a healthy pool connection. +type fastQueryDB struct{} + +func (fastQueryDB) Ping(_ context.Context) error { return nil } + +func (fastQueryDB) QueryRow(_ context.Context, _ string, _ ...any) pgx.Row { + return fastQueryRow{} +} + +func (fastQueryDB) Query(_ context.Context, _ string, _ ...any) (pgx.Rows, error) { + return nil, nil +} + +type fastQueryRow struct{} + +func (fastQueryRow) Scan(_ ...any) error { return pgx.ErrNoRows } + +// TestTokenMetadata_SlowQueryIsCancelledByTimeout demonstrates issue #238's +// per-call deadline: even though the incoming request context has no +// deadline of its own (httptest.NewRequest's default), the handler's own +// tokenMetadataQueryTimeout bounds the DB call — a runaway query can't hold +// the connection indefinitely. +func TestTokenMetadata_SlowQueryIsCancelledByTimeout(t *testing.T) { + req := httptest.NewRequest(http.MethodGet, "/v1/contracts/"+validSchemaContractID+"/metadata", nil) + req.SetPathValue("id", validSchemaContractID) + rr := httptest.NewRecorder() + + start := time.Now() + TokenMetadata(slowQueryDB{}).ServeHTTP(rr, req) + elapsed := time.Since(start) + + if elapsed > tokenMetadataQueryTimeout+2*time.Second { + t.Fatalf("handler took %v, want bounded near the %v query timeout — deadline was not applied", elapsed, tokenMetadataQueryTimeout) + } + if elapsed < tokenMetadataQueryTimeout-500*time.Millisecond { + t.Fatalf("handler returned after only %v, want it to have waited out the %v query timeout", elapsed, tokenMetadataQueryTimeout) + } + if rr.Code != http.StatusServiceUnavailable { + t.Fatalf("want 503 on a cancelled query, got %d", rr.Code) + } +} + +// TestTokenMetadata_PoolRecoversAfterSlowQuery demonstrates issue #238's +// "pool recovery" criterion: a request against a healthy connection +// immediately after a cancelled slow query succeeds normally and quickly — +// the earlier timeout does not leave the handler path wedged. +func TestTokenMetadata_PoolRecoversAfterSlowQuery(t *testing.T) { + slowReq := httptest.NewRequest(http.MethodGet, "/v1/contracts/"+validSchemaContractID+"/metadata", nil) + slowReq.SetPathValue("id", validSchemaContractID) + TokenMetadata(slowQueryDB{}).ServeHTTP(httptest.NewRecorder(), slowReq) + + fastReq := httptest.NewRequest(http.MethodGet, "/v1/contracts/"+validSchemaContractID+"/metadata", nil) + fastReq.SetPathValue("id", validSchemaContractID) + rr := httptest.NewRecorder() + + start := time.Now() + TokenMetadata(fastQueryDB{}).ServeHTTP(rr, fastReq) + elapsed := time.Since(start) + + if elapsed > time.Second { + t.Fatalf("recovery request took %v, want a fast response — pool/handler path looks wedged after the prior timeout", elapsed) + } + if rr.Code != http.StatusOK { + t.Fatalf("want 200 on recovery request, got %d", rr.Code) + } +} diff --git a/services/api/internal/contracttest/contracttest.go b/services/api/internal/contracttest/contracttest.go new file mode 100644 index 0000000..dbc153b --- /dev/null +++ b/services/api/internal/contracttest/contracttest.go @@ -0,0 +1,87 @@ +// Package contracttest validates live HTTP responses against +// api/openapi.yaml (issue #242), so a header or body shape drifting from +// the documented contract fails a test instead of surfacing only in +// production against real clients. +package contracttest + +import ( + "bytes" + "context" + "io" + "net/http" + "path/filepath" + "runtime" + "testing" + + "github.com/getkin/kin-openapi/openapi3" + "github.com/getkin/kin-openapi/openapi3filter" + "github.com/getkin/kin-openapi/routers" + "github.com/getkin/kin-openapi/routers/gorillamux" +) + +// specPath resolves api/openapi.yaml relative to this source file (not the +// test's working directory), so callers in any package under services/api +// find the same spec regardless of `go test`'s per-package cwd. +func specPath() string { + _, thisFile, _, _ := runtime.Caller(0) + return filepath.Join(filepath.Dir(thisFile), "..", "..", "..", "..", "api", "openapi.yaml") +} + +// LoadSpec loads and validates api/openapi.yaml. Fails the test immediately +// on a malformed spec, since every contract test depends on it. +func LoadSpec(t *testing.T) *openapi3.T { + t.Helper() + loader := &openapi3.Loader{IsExternalRefsAllowed: false} + doc, err := loader.LoadFromFile(specPath()) + if err != nil { + t.Fatalf("contracttest: load api/openapi.yaml: %v", err) + } + if err := doc.Validate(context.Background()); err != nil { + t.Fatalf("contracttest: api/openapi.yaml failed its own validation: %v", err) + } + return doc +} + +// NewRouter builds a router used to resolve an *http.Request to the +// operation (and its documented responses) it matches in doc. +func NewRouter(t *testing.T, doc *openapi3.T) routers.Router { + t.Helper() + router, err := gorillamux.NewRouter(doc) + if err != nil { + t.Fatalf("contracttest: build router: %v", err) + } + return router +} + +// ValidateResponse asserts that status/header/body for req's matched +// operation conform to what api/openapi.yaml documents — the response code +// is a documented one, every documented header for that response is +// present and matches its schema, and the body matches the documented +// content schema. Fails the test (via t.Error, not Fatal, so multiple +// contract violations in a suite are all reported) on any mismatch. +func ValidateResponse(t *testing.T, router routers.Router, req *http.Request, status int, header http.Header, body []byte) { + t.Helper() + + route, pathParams, err := router.FindRoute(req) + if err != nil { + t.Errorf("contracttest: %s %s does not match any documented route: %v", req.Method, req.URL.Path, err) + return + } + + reqInput := &openapi3filter.RequestValidationInput{ + Request: req, + PathParams: pathParams, + Route: route, + } + + respInput := &openapi3filter.ResponseValidationInput{ + RequestValidationInput: reqInput, + Status: status, + Header: header, + Body: io.NopCloser(bytes.NewReader(body)), + } + + if err := openapi3filter.ValidateResponse(context.Background(), respInput); err != nil { + t.Errorf("contracttest: %s %s -> %d response does not conform to api/openapi.yaml: %v", req.Method, req.URL.Path, status, err) + } +} diff --git a/services/api/internal/contracttest/contracttest_test.go b/services/api/internal/contracttest/contracttest_test.go new file mode 100644 index 0000000..3e40a65 --- /dev/null +++ b/services/api/internal/contracttest/contracttest_test.go @@ -0,0 +1,38 @@ +package contracttest + +import ( + "net/http" + "testing" +) + +// TestLoadSpec_Valid is the baseline check that api/openapi.yaml itself is +// well-formed and internally consistent (issue #242) — every other contract +// test depends on this succeeding. +func TestLoadSpec_Valid(t *testing.T) { + doc := LoadSpec(t) + if doc.Info == nil || doc.Info.Title == "" { + t.Fatal("loaded spec has no info.title") + } + if _, ok := doc.Paths.Map()["/v1/events"]; !ok { + t.Fatal("loaded spec is missing /v1/events") + } +} + +// TestNewRouter_ResolvesKnownRoute verifies the router can match a +// documented path/method pair. +func TestNewRouter_ResolvesKnownRoute(t *testing.T) { + doc := LoadSpec(t) + router := NewRouter(t, doc) + + req, err := http.NewRequest(http.MethodGet, "http://localhost:3000/v1/events", nil) + if err != nil { + t.Fatalf("build request: %v", err) + } + route, _, err := router.FindRoute(req) + if err != nil { + t.Fatalf("FindRoute(GET /v1/events): %v", err) + } + if route.Operation.OperationID != "listEvents" { + t.Errorf("want operationId listEvents, got %s", route.Operation.OperationID) + } +} diff --git a/services/api/internal/metrics/dbpool_test.go b/services/api/internal/metrics/dbpool_test.go new file mode 100644 index 0000000..8bacbf7 --- /dev/null +++ b/services/api/internal/metrics/dbpool_test.go @@ -0,0 +1,81 @@ +package metrics + +import ( + "context" + "testing" + "time" + + "github.com/jackc/pgx/v5/pgxpool" + "github.com/prometheus/client_golang/prometheus/testutil" +) + +// newUnconnectedPool builds a pool that never successfully connects (bogus +// port) but is otherwise fully constructed, so Stat() is safe to call +// without a live database — pgxpool dials lazily/in the background and +// Stat() reflects whatever state exists at call time. +func newUnconnectedPool(t *testing.T) *pgxpool.Pool { + t.Helper() + cfg, err := pgxpool.ParseConfig("postgres://user:pass@127.0.0.1:1/testdb") + if err != nil { + t.Fatalf("ParseConfig: %v", err) + } + cfg.MaxConns = 4 + pool, err := pgxpool.NewWithConfig(context.Background(), cfg) + if err != nil { + t.Fatalf("NewWithConfig: %v", err) + } + t.Cleanup(pool.Close) + return pool +} + +// TestPollDBPool_ReportsStatImmediately verifies PollDBPool populates the +// gauges from an initial report before the first tick (issue #238). +func TestPollDBPool_ReportsStatImmediately(t *testing.T) { + pool := newUnconnectedPool(t) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + done := make(chan struct{}) + go func() { + PollDBPool(ctx, pool, time.Hour) // long interval — only the immediate report matters here + close(done) + }() + + // Give the immediate report a moment to run. + time.Sleep(50 * time.Millisecond) + + if got := testutil.ToFloat64(DBPoolMaxConns); got != 4 { + t.Errorf("trident_db_pool_max_conns: want 4, got %v", got) + } + + cancel() + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("PollDBPool did not return after context cancellation") + } +} + +// TestPollDBPool_StopsOnContextCancel verifies the polling loop exits +// promptly when ctx is done, rather than leaking a goroutine. +func TestPollDBPool_StopsOnContextCancel(t *testing.T) { + pool := newUnconnectedPool(t) + + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + go func() { + PollDBPool(ctx, pool, 10*time.Millisecond) + close(done) + }() + + // Let a few ticks happen, then cancel. + time.Sleep(30 * time.Millisecond) + cancel() + + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("PollDBPool did not stop after context cancellation") + } +} diff --git a/services/api/internal/metrics/metrics.go b/services/api/internal/metrics/metrics.go new file mode 100644 index 0000000..9f3bf19 --- /dev/null +++ b/services/api/internal/metrics/metrics.go @@ -0,0 +1,221 @@ +// Package metrics provides a client_golang-backed Prometheus registry served +// on its own internal port (issue #58), separate from the public API port. +// +// It is additive to the pre-existing hand-rolled, dependency-free +// Prometheus-text metrics mounted at GET /metrics on the public mux +// (handlers.MetricsHandler and friends) — that endpoint is untouched. This +// package covers the specific gaps called out by #58: per-endpoint HTTP +// request counts/latency, active WebSocket connections and message totals, +// outbound gRPC call metrics, and rate-limiting rejections. +package metrics + +import ( + "context" + "log/slog" + "net/http" + "os" + "time" + + "github.com/jackc/pgx/v5/pgxpool" + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/promauto" + "github.com/prometheus/client_golang/prometheus/promhttp" +) + +// DefaultPort is used when METRICS_PORT is unset. +const DefaultPort = "9091" + +// Registry is a dedicated registry (not the global default) so this endpoint +// exposes exactly the collectors defined here — no Go-runtime default +// collectors mixed in. +var Registry = prometheus.NewRegistry() + +var ( + HTTPRequestsTotal = promauto.With(Registry).NewCounterVec(prometheus.CounterOpts{ + Name: "trident_http_requests_total", + Help: "Total HTTP requests handled by the Go API, by method, route pattern, and status code.", + }, []string{"method", "path", "status"}) + + HTTPRequestDuration = promauto.With(Registry).NewHistogramVec(prometheus.HistogramOpts{ + Name: "trident_http_request_duration_seconds", + Help: "HTTP request duration in seconds, by method, route pattern, and status code.", + Buckets: prometheus.DefBuckets, + }, []string{"method", "path", "status"}) + + WSActiveConnections = promauto.With(Registry).NewGauge(prometheus.GaugeOpts{ + Name: "trident_ws_active_connections", + Help: "Currently active WebSocket subscribers (REST WS + GraphQL subscriptions).", + }) + + WSConnectsTotal = promauto.With(Registry).NewCounter(prometheus.CounterOpts{ + Name: "trident_ws_connects_total", + Help: "Total WebSocket subscriber registrations since startup.", + }) + + WSDisconnectsTotal = promauto.With(Registry).NewCounter(prometheus.CounterOpts{ + Name: "trident_ws_disconnects_total", + Help: "Total WebSocket subscriber unregistrations since startup.", + }) + + WSMessagesTotal = promauto.With(Registry).NewCounterVec(prometheus.CounterOpts{ + Name: "trident_ws_messages_total", + Help: "Total WebSocket broadcast messages, by outcome.", + }, []string{"result"}) // result: sent|dropped + + GRPCClientRequestsTotal = promauto.With(Registry).NewCounterVec(prometheus.CounterOpts{ + Name: "trident_grpc_client_requests_total", + Help: "Total outbound gRPC client call attempts, by method and status code.", + }, []string{"method", "code"}) + + GRPCClientRequestDuration = promauto.With(Registry).NewHistogramVec(prometheus.HistogramOpts{ + Name: "trident_grpc_client_request_duration_seconds", + Help: "Outbound gRPC client call duration in seconds, by method and status code.", + Buckets: prometheus.DefBuckets, + }, []string{"method", "code"}) + + RateLimitRejectionsTotal = promauto.With(Registry).NewCounterVec(prometheus.CounterOpts{ + Name: "trident_ratelimit_rejections_total", + Help: "Total requests rejected by a rate limiter, by limiter.", + }, []string{"limiter"}) // limiter: per_key|per_ip|global_concurrency + + // DB pool saturation metrics (issue #238), sourced from pgxpool.Pool.Stat() + // by PollDBPool. All exposed as Gauges — Stat() itself only returns + // point-in-time cumulative totals (not deltas), which Set() reflects + // directly; Prometheus rate()/increase() work the same over a + // monotonically-increasing Gauge as over a Counter. + DBPoolMaxConns = promauto.With(Registry).NewGauge(prometheus.GaugeOpts{ + Name: "trident_db_pool_max_conns", + Help: "Configured maximum size of the Postgres connection pool.", + }) + DBPoolTotalConns = promauto.With(Registry).NewGauge(prometheus.GaugeOpts{ + Name: "trident_db_pool_total_conns", + Help: "Current total connections in the Postgres pool (idle + in-use + being established).", + }) + DBPoolAcquiredConns = promauto.With(Registry).NewGauge(prometheus.GaugeOpts{ + Name: "trident_db_pool_acquired_conns", + Help: "Connections currently acquired (in use) from the Postgres pool.", + }) + DBPoolIdleConns = promauto.With(Registry).NewGauge(prometheus.GaugeOpts{ + Name: "trident_db_pool_idle_conns", + Help: "Idle connections currently available in the Postgres pool.", + }) + DBPoolConstructingConns = promauto.With(Registry).NewGauge(prometheus.GaugeOpts{ + Name: "trident_db_pool_constructing_conns", + Help: "Connections currently being established for the Postgres pool.", + }) + DBPoolAcquireCount = promauto.With(Registry).NewGauge(prometheus.GaugeOpts{ + Name: "trident_db_pool_acquire_count", + Help: "Cumulative number of successful connection acquisitions from the Postgres pool.", + }) + DBPoolEmptyAcquireCount = promauto.With(Registry).NewGauge(prometheus.GaugeOpts{ + Name: "trident_db_pool_empty_acquire_count", + Help: "Cumulative number of acquisitions that had to wait because the Postgres pool had no idle connection — a direct saturation signal.", + }) + DBPoolCanceledAcquireCount = promauto.With(Registry).NewGauge(prometheus.GaugeOpts{ + Name: "trident_db_pool_canceled_acquire_count", + Help: "Cumulative number of connection acquisitions canceled before completion (e.g. caller's context expired while waiting).", + }) + DBPoolAcquireDurationSeconds = promauto.With(Registry).NewGauge(prometheus.GaugeOpts{ + Name: "trident_db_pool_acquire_duration_seconds", + Help: "Cumulative time spent acquiring connections from the Postgres pool, in seconds.", + }) + DBPoolEmptyAcquireWaitSeconds = promauto.With(Registry).NewGauge(prometheus.GaugeOpts{ + Name: "trident_db_pool_empty_acquire_wait_seconds", + Help: "Cumulative time acquisitions spent waiting for a connection because the Postgres pool was empty, in seconds — a direct saturation signal.", + }) + DBPoolNewConnsCount = promauto.With(Registry).NewGauge(prometheus.GaugeOpts{ + Name: "trident_db_pool_new_conns_count", + Help: "Cumulative number of new connections established for the Postgres pool.", + }) + DBPoolMaxIdleDestroyCount = promauto.With(Registry).NewGauge(prometheus.GaugeOpts{ + Name: "trident_db_pool_max_idle_destroy_count", + Help: "Cumulative number of connections destroyed for exceeding MaxConnIdleTime.", + }) + DBPoolMaxLifetimeDestroyCount = promauto.With(Registry).NewGauge(prometheus.GaugeOpts{ + Name: "trident_db_pool_max_lifetime_destroy_count", + Help: "Cumulative number of connections destroyed for exceeding MaxConnLifetime.", + }) +) + +// PollDBPool periodically snapshots pool.Stat() into the DB pool gauges +// above (issue #238) until ctx is done. Runs once immediately so the gauges +// are populated before the first tick. +func PollDBPool(ctx context.Context, pool *pgxpool.Pool, interval time.Duration) { + report := func() { + stat := pool.Stat() + DBPoolMaxConns.Set(float64(stat.MaxConns())) + DBPoolTotalConns.Set(float64(stat.TotalConns())) + DBPoolAcquiredConns.Set(float64(stat.AcquiredConns())) + DBPoolIdleConns.Set(float64(stat.IdleConns())) + DBPoolConstructingConns.Set(float64(stat.ConstructingConns())) + DBPoolAcquireCount.Set(float64(stat.AcquireCount())) + DBPoolEmptyAcquireCount.Set(float64(stat.EmptyAcquireCount())) + DBPoolCanceledAcquireCount.Set(float64(stat.CanceledAcquireCount())) + DBPoolAcquireDurationSeconds.Set(stat.AcquireDuration().Seconds()) + DBPoolEmptyAcquireWaitSeconds.Set(stat.EmptyAcquireWaitTime().Seconds()) + DBPoolNewConnsCount.Set(float64(stat.NewConnsCount())) + DBPoolMaxIdleDestroyCount.Set(float64(stat.MaxIdleDestroyCount())) + DBPoolMaxLifetimeDestroyCount.Set(float64(stat.MaxLifetimeDestroyCount())) + } + + report() + + ticker := time.NewTicker(interval) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + report() + } + } +} + +// Port returns the port the metrics server listens on (METRICS_PORT, or +// DefaultPort). +func Port() string { + if p := os.Getenv("METRICS_PORT"); p != "" { + return p + } + return DefaultPort +} + +// Handler builds a mux exposing only GET /metrics, backed by Registry. +// Exposed for testing. +func Handler() *http.ServeMux { + mux := http.NewServeMux() + mux.Handle("GET /metrics", promhttp.HandlerFor(Registry, promhttp.HandlerOpts{})) + return mux +} + +// Start launches the internal metrics server on METRICS_PORT (default 9091) +// and returns its *http.Server so the caller can shut it down. +func Start() *http.Server { + addr := ":" + Port() + srv := &http.Server{ + Addr: addr, + Handler: Handler(), + ReadHeaderTimeout: 5 * time.Second, + } + + slog.Info("metrics server listening", "addr", addr) + + go func() { + if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed { + slog.Error("metrics server error", "err", err) + } + }() + + return srv +} + +// Shutdown gracefully stops the metrics server (nil-safe). +func Shutdown(srv *http.Server) { + if srv == nil { + return + } + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + _ = srv.Shutdown(ctx) +} diff --git a/services/api/main.go b/services/api/main.go index 6a7d342..530633d 100644 --- a/services/api/main.go +++ b/services/api/main.go @@ -14,6 +14,7 @@ import ( "github.com/Depo-dev/trident/services/api/grpc" "github.com/Depo-dev/trident/services/api/handlers" + "github.com/Depo-dev/trident/services/api/internal/metrics" "github.com/Depo-dev/trident/services/api/internal/profiling" "github.com/Depo-dev/trident/services/api/internal/sorobanrpc" "github.com/Depo-dev/trident/services/api/middleware" @@ -35,6 +36,10 @@ import ( // allow it to be. const contractStatsRollupRefreshInterval = 60 * time.Second +// How often pgxpool saturation stats are polled into Prometheus gauges +// (issue #238). +const dbPoolMetricsPollInterval = 15 * time.Second + // Usage rollup: re-aggregate audit_log into usage_rollup every 5 minutes, // covering the last 48h so late-arriving audit rows (the writer batches // asynchronously) and the UTC day boundary are always caught by the next run. @@ -47,6 +52,26 @@ const ( const defaultDBPoolSize = 5 +// Pool lifecycle defaults (issue #238). Applied in buildPoolConfig unless +// overridden by env vars. +const ( + defaultDBPoolMinConns = 0 + defaultDBPoolMaxConnLifetimeMS = 1_800_000 // 30 min + defaultDBPoolMaxConnIdleTimeMS = 300_000 // 5 min + defaultDBPoolHealthCheckPeriodMS = 60_000 // 1 min, matches pgxpool's own default + dbPoolMaxConnLifetimeJitterPercent = 10 // spreads reconnects so the pool doesn't empty all at once +) + +// Statement-timeout defaults (issue #238), shared with the Rust indexer +// (crates/indexer/src/config.rs) via the same env vars so both services agree +// on how long a query or idle transaction may hold a connection. +const ( + defaultStatementTimeoutMS = 30_000 + defaultIdleInTransactionTimeoutMS = 10_000 + statementTimeoutMinMS = 100 + statementTimeoutMaxMS = 3_600_000 +) + // connErrRegexp matches a userinfo-bearing connection URI (scheme://user:pass@host) // so DB/Redis connection errors — which some drivers embed the DSN in — never // leak the credential portion to logs (issue #305). @@ -198,6 +223,13 @@ func main() { go runContractStatsRollupRefresh(ctx, pool) } + // Periodically export pgxpool saturation stats (issue #238) — total/idle/ + // acquired conns and acquire-wait time are the direct signal that a burst + // of slow queries is starving the pool. + if pool != nil { + go metrics.PollDBPool(ctx, pool, dbPoolMetricsPollInterval) + } + // Re-aggregate audit_log into usage_rollup so GET /v1/usage reads a // pre-aggregated table, and bound that table's growth. Both loops and // their four interval constants already existed but were never started — @@ -247,7 +279,8 @@ func main() { handlers.SetInternalStatusDeps(pool, redisClient, hub) mux := http.NewServeMux() - mux.HandleFunc("GET /v1/health", handlers.Health(healthDB, redisClient, grpcClient)) + mux.HandleFunc("GET /v1/health", handlers.Health()) + mux.HandleFunc("GET /v1/ready", handlers.Ready(healthDB, redisClient, grpcClient)) mux.HandleFunc("GET /v1/events", handlers.ListEvents) mux.HandleFunc("POST /v1/events/batch", handlers.BatchGetEvents) mux.HandleFunc("GET /v1/events/{id}", handlers.GetEvent) @@ -333,12 +366,21 @@ func main() { // Redis calls, logging — is spent on a request that's going to be // rejected anyway. handler = middleware.NewGlobalConcurrencyLimitFromEnv()(handler) - + // Metrics middleware is the absolute outermost wrap (issue #58): it must + // see every response, including ones shed by GlobalConcurrencyLimit, to + // report accurate per-endpoint counts/latency. + handler = middleware.NewMetrics(mux)(handler) // Opt-in, internal-only pprof server (off unless PPROF_ENABLED=true). It is // never mounted on the public mux above (#299). pprofSrv := profiling.Start() defer profiling.Shutdown(pprofSrv) + // Internal Prometheus metrics server on METRICS_PORT (default 9091, + // issue #58) — separate port from the public API and from the legacy + // /metrics route mounted above. + metricsSrv := metrics.Start() + defer metrics.Shutdown(metricsSrv) + // Grace period mirrors Helm terminationGracePeriodSeconds (default 30s). const shutdownGrace = 30 * time.Second @@ -380,13 +422,55 @@ func main() { slog.Info("shutdown complete") } -func newDBPool(ctx context.Context, dsn string, poolSize int32) (*pgxpool.Pool, error) { +// buildPoolConfig parses dsn and applies pool sizing, lifecycle, and +// statement-timeout settings (issue #238). It does not connect — safe to +// call from tests without a live database. +func buildPoolConfig(dsn string, poolSize int32) (*pgxpool.Config, error) { cfg, err := pgxpool.ParseConfig(dsn) if err != nil { return nil, fmt.Errorf("parse DATABASE_URL: %w", err) } - cfg.MaxConns = poolSize cfg.ConnConfig.DefaultQueryExecMode = pgx.QueryExecModeSimpleProtocol + + cfg.MaxConns = poolSize + cfg.MinConns = envInt32("GO_API_DB_POOL_MIN_CONNS", defaultDBPoolMinConns) + cfg.MaxConnLifetime = envDurationMS("GO_API_DB_POOL_MAX_CONN_LIFETIME_MS", defaultDBPoolMaxConnLifetimeMS) + cfg.MaxConnLifetimeJitter = cfg.MaxConnLifetime * dbPoolMaxConnLifetimeJitterPercent / 100 + cfg.MaxConnIdleTime = envDurationMS("GO_API_DB_POOL_MAX_CONN_IDLE_TIME_MS", defaultDBPoolMaxConnIdleTimeMS) + cfg.HealthCheckPeriod = envDurationMS("GO_API_DB_POOL_HEALTH_CHECK_PERIOD_MS", defaultDBPoolHealthCheckPeriodMS) + + // Bound how long a single statement or an idle-in-transaction connection + // may hold a pool slot (issue #238) — a runaway query or a leaked + // transaction must not be able to stall the whole pool. Shared env vars + // with the Rust indexer (crates/indexer/src/config.rs) so both services + // agree; see #249 for database-level (role/cluster) coordination. + // + // These SETs run once per physical connection at AfterConnect time, + // before any transaction begins — safe for a direct DATABASE_URL + // connection. If DATABASE_URL is ever pointed at PgBouncer in + // transaction-pooling mode (today only PGBOUNCER_ADMIN_URL, the admin + // console, is used — see pgbouncer.go), this needs revisiting against + // PgBouncer's parameter-tracking behavior (#249, #256). + stmtTimeoutMS := envIntBounded("DB_STATEMENT_TIMEOUT_MS", defaultStatementTimeoutMS, statementTimeoutMinMS, statementTimeoutMaxMS) + idleTimeoutMS := envIntBounded("DB_IDLE_IN_TRANSACTION_TIMEOUT_MS", defaultIdleInTransactionTimeoutMS, statementTimeoutMinMS, statementTimeoutMaxMS) + cfg.AfterConnect = func(ctx context.Context, conn *pgx.Conn) error { + if _, err := conn.Exec(ctx, fmt.Sprintf("SET statement_timeout = '%dms'", stmtTimeoutMS)); err != nil { + return fmt.Errorf("set statement_timeout: %w", err) + } + if _, err := conn.Exec(ctx, fmt.Sprintf("SET idle_in_transaction_session_timeout = '%dms'", idleTimeoutMS)); err != nil { + return fmt.Errorf("set idle_in_transaction_session_timeout: %w", err) + } + return nil + } + + return cfg, nil +} + +func newDBPool(ctx context.Context, dsn string, poolSize int32) (*pgxpool.Pool, error) { + cfg, err := buildPoolConfig(dsn, poolSize) + if err != nil { + return nil, err + } pool, err := pgxpool.NewWithConfig(ctx, cfg) if err != nil { return nil, err @@ -408,6 +492,60 @@ func dbPoolSizeFromEnv() int32 { return defaultDBPoolSize } +// envInt32 reads a non-negative int32 env var, falling back to def on +// missing/invalid input (issue #238). +func envInt32(key string, def int32) int32 { + raw := os.Getenv(key) + if raw == "" { + return def + } + n, err := strconv.Atoi(raw) + if err != nil || n < 0 { + slog.Warn("invalid env value; using default", "key", key, "value", raw, "default", def) + return def + } + return int32(n) +} + +// envDurationMS reads a millisecond duration env var, falling back to defMS +// on missing/invalid input (issue #238). +func envDurationMS(key string, defMS int) time.Duration { + raw := os.Getenv(key) + if raw == "" { + return time.Duration(defMS) * time.Millisecond + } + n, err := strconv.Atoi(raw) + if err != nil || n < 0 { + slog.Warn("invalid env value; using default", "key", key, "value", raw, "default_ms", defMS) + return time.Duration(defMS) * time.Millisecond + } + return time.Duration(n) * time.Millisecond +} + +// envIntBounded reads an int env var clamped to [min, max], falling back to +// def on missing/invalid input (issue #238). Mirrors the Rust indexer's +// parse_bounded_u64 (crates/indexer/src/config.rs) so both services validate +// DB_STATEMENT_TIMEOUT_MS/DB_IDLE_IN_TRANSACTION_TIMEOUT_MS the same way. +func envIntBounded(key string, def, min, max int) int { + raw := os.Getenv(key) + if raw == "" { + return def + } + n, err := strconv.Atoi(raw) + if err != nil { + slog.Warn("invalid env value; using default", "key", key, "value", raw, "default", def) + return def + } + if n < min || n > max { + slog.Warn("env value out of range; clamping", "key", key, "value", n, "min", min, "max", max) + if n < min { + return min + } + return max + } + return n +} + // runContractStatsRollupRefresh recomputes contract_stats_rollup on a fixed // interval until ctx is cancelled (issue #257). Runs once immediately so the // rollup is populated shortly after startup rather than only after the first diff --git a/services/api/main_test.go b/services/api/main_test.go new file mode 100644 index 0000000..fb5aa99 --- /dev/null +++ b/services/api/main_test.go @@ -0,0 +1,120 @@ +package main + +import ( + "testing" + "time" +) + +const testDSN = "postgres://user:pass@localhost:5432/testdb" + +// TestBuildPoolConfig_Defaults verifies buildPoolConfig applies the +// documented defaults when no pool env vars are set (issue #238). +// pgxpool.ParseConfig does not connect, so this needs no live database. +func TestBuildPoolConfig_Defaults(t *testing.T) { + cfg, err := buildPoolConfig(testDSN, 7) + if err != nil { + t.Fatalf("buildPoolConfig: %v", err) + } + + if cfg.MaxConns != 7 { + t.Errorf("MaxConns: want 7, got %d", cfg.MaxConns) + } + if cfg.MinConns != defaultDBPoolMinConns { + t.Errorf("MinConns: want %d, got %d", defaultDBPoolMinConns, cfg.MinConns) + } + if want := time.Duration(defaultDBPoolMaxConnLifetimeMS) * time.Millisecond; cfg.MaxConnLifetime != want { + t.Errorf("MaxConnLifetime: want %v, got %v", want, cfg.MaxConnLifetime) + } + if want := cfg.MaxConnLifetime * dbPoolMaxConnLifetimeJitterPercent / 100; cfg.MaxConnLifetimeJitter != want { + t.Errorf("MaxConnLifetimeJitter: want %v, got %v", want, cfg.MaxConnLifetimeJitter) + } + if want := time.Duration(defaultDBPoolMaxConnIdleTimeMS) * time.Millisecond; cfg.MaxConnIdleTime != want { + t.Errorf("MaxConnIdleTime: want %v, got %v", want, cfg.MaxConnIdleTime) + } + if want := time.Duration(defaultDBPoolHealthCheckPeriodMS) * time.Millisecond; cfg.HealthCheckPeriod != want { + t.Errorf("HealthCheckPeriod: want %v, got %v", want, cfg.HealthCheckPeriod) + } + if cfg.AfterConnect == nil { + t.Error("AfterConnect: want non-nil (statement_timeout / idle_in_transaction_session_timeout hook)") + } +} + +// TestBuildPoolConfig_EnvOverrides verifies pool lifecycle env vars are +// honored (issue #238). +func TestBuildPoolConfig_EnvOverrides(t *testing.T) { + t.Setenv("GO_API_DB_POOL_MIN_CONNS", "3") + t.Setenv("GO_API_DB_POOL_MAX_CONN_LIFETIME_MS", "60000") + t.Setenv("GO_API_DB_POOL_MAX_CONN_IDLE_TIME_MS", "20000") + t.Setenv("GO_API_DB_POOL_HEALTH_CHECK_PERIOD_MS", "5000") + + cfg, err := buildPoolConfig(testDSN, 10) + if err != nil { + t.Fatalf("buildPoolConfig: %v", err) + } + + if cfg.MinConns != 3 { + t.Errorf("MinConns: want 3, got %d", cfg.MinConns) + } + if cfg.MaxConnLifetime != 60*time.Second { + t.Errorf("MaxConnLifetime: want 60s, got %v", cfg.MaxConnLifetime) + } + if cfg.MaxConnLifetimeJitter != 6*time.Second { + t.Errorf("MaxConnLifetimeJitter: want 6s (10%% of lifetime), got %v", cfg.MaxConnLifetimeJitter) + } + if cfg.MaxConnIdleTime != 20*time.Second { + t.Errorf("MaxConnIdleTime: want 20s, got %v", cfg.MaxConnIdleTime) + } + if cfg.HealthCheckPeriod != 5*time.Second { + t.Errorf("HealthCheckPeriod: want 5s, got %v", cfg.HealthCheckPeriod) + } +} + +// TestBuildPoolConfig_InvalidEnvFallsBackToDefault verifies unparsable pool +// env vars fall back to defaults rather than erroring (issue #238), matching +// dbPoolSizeFromEnv's existing warn-and-fallback convention. +func TestBuildPoolConfig_InvalidEnvFallsBackToDefault(t *testing.T) { + t.Setenv("GO_API_DB_POOL_MIN_CONNS", "not-a-number") + + cfg, err := buildPoolConfig(testDSN, 5) + if err != nil { + t.Fatalf("buildPoolConfig: %v", err) + } + if cfg.MinConns != defaultDBPoolMinConns { + t.Errorf("MinConns: want default %d on invalid input, got %d", defaultDBPoolMinConns, cfg.MinConns) + } +} + +// TestEnvIntBounded_ClampsOutOfRange verifies DB_STATEMENT_TIMEOUT_MS / +// DB_IDLE_IN_TRANSACTION_TIMEOUT_MS style vars are clamped into range rather +// than accepted as-is (issue #238), mirroring the Rust indexer's +// parse_bounded_u64. +func TestEnvIntBounded_ClampsOutOfRange(t *testing.T) { + cases := []struct { + name string + value string + want int + }{ + {"below min clamps to min", "50", statementTimeoutMinMS}, + {"above max clamps to max", "10000000", statementTimeoutMaxMS}, + {"within range passes through", "15000", 15000}, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + t.Setenv("TEST_BOUNDED_TIMEOUT_MS", tc.value) + got := envIntBounded("TEST_BOUNDED_TIMEOUT_MS", defaultStatementTimeoutMS, statementTimeoutMinMS, statementTimeoutMaxMS) + if got != tc.want { + t.Errorf("envIntBounded(%q): want %d, got %d", tc.value, tc.want, got) + } + }) + } +} + +// TestEnvIntBounded_DefaultsWhenUnset verifies an unset env var returns def +// rather than 0 or an error. +func TestEnvIntBounded_DefaultsWhenUnset(t *testing.T) { + got := envIntBounded("TEST_BOUNDED_UNSET_VAR", defaultStatementTimeoutMS, statementTimeoutMinMS, statementTimeoutMaxMS) + if got != defaultStatementTimeoutMS { + t.Errorf("want default %d, got %d", defaultStatementTimeoutMS, got) + } +} diff --git a/services/api/middleware/abuse.go b/services/api/middleware/abuse.go index fb7c088..c4a9ef4 100644 --- a/services/api/middleware/abuse.go +++ b/services/api/middleware/abuse.go @@ -13,6 +13,7 @@ import ( "time" "github.com/Depo-dev/trident/services/api/internal/httputil" + "github.com/Depo-dev/trident/services/api/internal/metrics" "github.com/redis/go-redis/v9" ) @@ -156,6 +157,7 @@ func PerIPRateLimit(cfg PerIPRateLimitConfig) func(http.Handler) http.Handler { if !allowed { perIPRejected.Add(1) + metrics.RateLimitRejectionsTotal.WithLabelValues("per_ip").Inc() retryAfter := int64(window.Seconds()) if retryAfter < 1 { retryAfter = 1 @@ -221,6 +223,7 @@ func GlobalConcurrencyLimit(maxInFlight int) func(http.Handler) http.Handler { if n > limit { globalRejected.Add(1) + metrics.RateLimitRejectionsTotal.WithLabelValues("global_concurrency").Inc() w.Header().Set("Retry-After", "1") httputil.WriteErrorCtx(r.Context(), w, http.StatusServiceUnavailable, httputil.UNAVAILABLE, "server is shedding load; try again shortly") return diff --git a/services/api/middleware/abuse_test.go b/services/api/middleware/abuse_test.go index cfa4dca..49e04c1 100644 --- a/services/api/middleware/abuse_test.go +++ b/services/api/middleware/abuse_test.go @@ -7,6 +7,9 @@ import ( "sync" "testing" "time" + + "github.com/Depo-dev/trident/services/api/internal/metrics" + "github.com/prometheus/client_golang/prometheus/testutil" ) // fakeSlider is a deterministic in-memory stand-in for the Redis sliding @@ -60,6 +63,76 @@ func TestPerIPRateLimit_ExceedingIPBlocked_OtherIPUnaffected(t *testing.T) { } } +// TestPerIPRateLimit_RejectionRecordsPrometheusMetric verifies a 429 from the +// per-IP limiter increments trident_ratelimit_rejections_total{limiter="per_ip"} +// (issue #58). +func TestPerIPRateLimit_RejectionRecordsPrometheusMetric(t *testing.T) { + handler := PerIPRateLimit(PerIPRateLimitConfig{ + RPS: 1, + Window: time.Second, + SliderFn: fakeSlider(t), + })(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusOK) + })) + + before := testutil.ToFloat64(metrics.RateLimitRejectionsTotal.WithLabelValues("per_ip")) + + do := func() int { + req := httptest.NewRequest(http.MethodGet, "/v1/events", nil) + req.RemoteAddr = "8.8.8.8:1" + rec := httptest.NewRecorder() + handler.ServeHTTP(rec, req) + return rec.Code + } + do() // allowed + code := do() // rejected + + if code != http.StatusTooManyRequests { + t.Fatalf("expected second request to be rejected, got %d", code) + } + if got := testutil.ToFloat64(metrics.RateLimitRejectionsTotal.WithLabelValues("per_ip")); got != before+1 { + t.Errorf("per_ip rejections total: want %v, got %v", before+1, got) + } +} + +// TestGlobalConcurrencyLimit_RejectionRecordsPrometheusMetric verifies a shed +// request increments trident_ratelimit_rejections_total{limiter="global_concurrency"}. +func TestGlobalConcurrencyLimit_RejectionRecordsPrometheusMetric(t *testing.T) { + release := make(chan struct{}) + started := make(chan struct{}, 1) + + slow := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + started <- struct{}{} + <-release + w.WriteHeader(http.StatusOK) + }) + handler := GlobalConcurrencyLimit(1)(slow) + + before := testutil.ToFloat64(metrics.RateLimitRejectionsTotal.WithLabelValues("global_concurrency")) + + var wg sync.WaitGroup + codes := make([]int, 2) + for i := 0; i < 2; i++ { + wg.Add(1) + go func(i int) { + defer wg.Done() + req := httptest.NewRequest(http.MethodGet, "/v1/events", nil) + rec := httptest.NewRecorder() + handler.ServeHTTP(rec, req) + codes[i] = rec.Code + }(i) + } + + <-started + time.Sleep(50 * time.Millisecond) + close(release) + wg.Wait() + + if got := testutil.ToFloat64(metrics.RateLimitRejectionsTotal.WithLabelValues("global_concurrency")); got != before+1 { + t.Errorf("global_concurrency rejections total: want %v, got %v", before+1, got) + } +} + func TestPerIPRateLimit_NonPublicPath_Skipped(t *testing.T) { handler := PerIPRateLimit(PerIPRateLimitConfig{ RPS: 0, // would reject request 1 if applied diff --git a/services/api/middleware/auth.go b/services/api/middleware/auth.go index 80f9192..8c1caee 100644 --- a/services/api/middleware/auth.go +++ b/services/api/middleware/auth.go @@ -29,6 +29,12 @@ type DBAuthConfig struct { const authCacheTTL = 5 * time.Minute +// authDBQueryTimeout bounds the DB fallback lookup in NewDBAuth (issue #238) +// — this runs on nearly every request, so it gets a tight deadline rather +// than the full request budget, matching handlers/status.go's convention for +// other hot/lightweight DB reads. +const authDBQueryTimeout = 2 * time.Second + // ParseKeyHashes parses a comma-separated list of HMAC-SHA256 hex digests // (as stored in API_KEY_HASHES) into a set for O(1) lookup. func ParseKeyHashes(raw string) map[string]struct{} { @@ -96,7 +102,7 @@ func NewDBAuth(cfg DBAuthConfig) func(http.Handler) http.Handler { return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { // Public paths — skip auth entirely. path := r.URL.Path - if path == "/v1/health" || path == "/metrics" { + if path == "/v1/health" || path == "/v1/ready" || path == "/metrics" { next.ServeHTTP(w, r) return } @@ -129,8 +135,11 @@ func NewDBAuth(cfg DBAuthConfig) func(http.Handler) http.Handler { // ── 2. Database lookup ────────────────────────────────────────── if cfg.DB != nil { + dbCtx, cancel := context.WithTimeout(r.Context(), authDBQueryTimeout) + defer cancel() + var id, network string - err := cfg.DB.QueryRow(r.Context(), + err := cfg.DB.QueryRow(dbCtx, `SELECT id, network FROM api_keys WHERE key_hash = $1 AND revoked_at IS NULL`, dbHash, ).Scan(&id, &network) diff --git a/services/api/middleware/contract_test.go b/services/api/middleware/contract_test.go new file mode 100644 index 0000000..a4b0566 --- /dev/null +++ b/services/api/middleware/contract_test.go @@ -0,0 +1,144 @@ +package middleware + +import ( + "net/http" + "net/http/httptest" + "sync" + "testing" + "time" + + "github.com/Depo-dev/trident/services/api/internal/contracttest" +) + +// validEventListBody is a minimal EventListResponse-shaped body (issue +// #242), used by contract tests that exercise rate-limit middleware in +// isolation — the middleware under test doesn't care what the wrapped +// handler returns, but the contract test validates the full response +// against api/openapi.yaml's GET /v1/events schema, so the body must be +// shaped correctly too. +const validEventListBody = `{"events":[],"has_more":false,"next_cursor":null}` + +// withDevServer rewrites req's URL to match api/openapi.yaml's declared +// "http://localhost:3000" dev server — the gorillamux contract-test router +// matches routes against declared servers, but httptest.NewRequest builds a +// relative-only URL that doesn't match any of them. +func withDevServer(req *http.Request) *http.Request { + req.URL.Scheme = "http" + req.URL.Host = "localhost:3000" + req.Host = "localhost:3000" + return req +} + +func eventListStub() http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte(validEventListBody)) + }) +} + +// TestContract_TieredRateLimit_Success200 verifies a rate-limit-allowed +// response through TieredRateLimit conforms to GET /v1/events's documented +// 200 response — the X-RateLimit-* headers declared in api/openapi.yaml are +// marked required, so a middleware regression that stops setting one of +// them fails this test (issue #242). +func TestContract_TieredRateLimit_Success200(t *testing.T) { + resetCounters() + doc := contracttest.LoadSpec(t) + router := contracttest.NewRouter(t, doc) + + cfg := RateLimitConfig{SliderFn: alwaysAllow, Tiers: testTiers()} + mw := TieredRateLimit(cfg)(eventListStub()) + + req := withDevServer(apiKeyReq("contract-test-key")) + rr := httptest.NewRecorder() + mw.ServeHTTP(rr, req) + + if rr.Code != http.StatusOK { + t.Fatalf("want 200, got %d", rr.Code) + } + contracttest.ValidateResponse(t, router, req, rr.Code, rr.Header(), rr.Body.Bytes()) +} + +// TestContract_TieredRateLimit_429 verifies a rejected request through +// TieredRateLimit conforms to GET /v1/events's documented 429 +// (RateLimitExceeded) response — X-RateLimit-* and Retry-After headers, +// plus the ErrorResponse body shape (issue #242). +func TestContract_TieredRateLimit_429(t *testing.T) { + resetCounters() + doc := contracttest.LoadSpec(t) + router := contracttest.NewRouter(t, doc) + + cfg := RateLimitConfig{SliderFn: alwaysReject, Tiers: testTiers()} + mw := TieredRateLimit(cfg)(eventListStub()) + + req := withDevServer(apiKeyReq("contract-test-key")) + rr := httptest.NewRecorder() + mw.ServeHTTP(rr, req) + + if rr.Code != http.StatusTooManyRequests { + t.Fatalf("want 429, got %d", rr.Code) + } + contracttest.ValidateResponse(t, router, req, rr.Code, rr.Header(), rr.Body.Bytes()) +} + +// TestContract_GlobalConcurrencyLimit_503 verifies a load-shed request +// conforms to GET /v1/events's documented 503 (ServiceUnavailable) +// response, including the Retry-After header that's only present in the +// load-shedding case (issue #242). Chains GlobalConcurrencyLimit outside +// TieredRateLimit, mirroring main.go's real middleware order, so the +// successful path also carries the X-RateLimit-* headers the 200 response +// requires. +func TestContract_GlobalConcurrencyLimit_503(t *testing.T) { + resetCounters() + doc := contracttest.LoadSpec(t) + router := contracttest.NewRouter(t, doc) + + rlCfg := RateLimitConfig{SliderFn: alwaysAllow, Tiers: testTiers()} + release := make(chan struct{}) + started := make(chan struct{}, 1) + slow := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + started <- struct{}{} + <-release + eventListStub().ServeHTTP(w, r) + }) + mw := GlobalConcurrencyLimit(1)(TieredRateLimit(rlCfg)(slow)) + + var wg sync.WaitGroup + results := make([]*httptest.ResponseRecorder, 2) + reqs := make([]*http.Request, 2) + for i := 0; i < 2; i++ { + reqs[i] = withDevServer(apiKeyReq("contract-test-key")) + results[i] = httptest.NewRecorder() + wg.Add(1) + go func(i int) { + defer wg.Done() + mw.ServeHTTP(results[i], reqs[i]) + }(i) + } + + <-started + time.Sleep(50 * time.Millisecond) + close(release) + wg.Wait() + + var okIdx, shedIdx = -1, -1 + for i, rr := range results { + switch rr.Code { + case http.StatusOK: + okIdx = i + case http.StatusServiceUnavailable: + shedIdx = i + } + } + if okIdx == -1 || shedIdx == -1 { + t.Fatalf("want one 200 and one 503, got %d and %d", results[0].Code, results[1].Code) + } + + contracttest.ValidateResponse(t, router, reqs[okIdx], results[okIdx].Code, results[okIdx].Header(), results[okIdx].Body.Bytes()) + contracttest.ValidateResponse(t, router, reqs[shedIdx], results[shedIdx].Code, results[shedIdx].Header(), results[shedIdx].Body.Bytes()) + + if got := results[shedIdx].Header().Get("Retry-After"); got == "" { + t.Error("shed response missing Retry-After header") + } +} diff --git a/services/api/middleware/metrics_registry.go b/services/api/middleware/metrics_registry.go new file mode 100644 index 0000000..3d1cbe9 --- /dev/null +++ b/services/api/middleware/metrics_registry.go @@ -0,0 +1,58 @@ +package middleware + +// Prometheus-registry HTTP metrics (issue #58), served on the dedicated +// METRICS_PORT listener in internal/metrics. +// +// This lives alongside metrics.go rather than inside it: that file holds the +// hand-rolled counters rendered into the public GET /metrics route by +// handlers.MetricsHandler, and the two are independent — different storage, +// different exposition path, no shared state. Keeping them in separate files +// keeps that boundary obvious. + +import ( + "net/http" + "strconv" + "time" + + "github.com/Depo-dev/trident/services/api/internal/metrics" +) + +// legacyMetricsPattern is the route pattern of the pre-existing hand-rolled +// /metrics endpoint on the public mux (handlers.MetricsHandler, main.go). +// Excluded from duration tracking per issue #58 — it isn't a "real" endpoint +// whose latency is meaningful, and self-scraping would otherwise skew the +// distribution. +const legacyMetricsPattern = "GET /metrics" + +// NewMetrics returns middleware that records per-endpoint HTTP request +// counts and latency to the internal Prometheus registry (issue #58). +// +// mux is the same *http.ServeMux the request will ultimately be routed +// through; mux.Handler(r) is a side-effect-free lookup that resolves the +// registered route pattern (e.g. "GET /v1/events/{id}") for use as a +// bounded-cardinality label, and works even for requests rejected by an +// earlier middleware (auth, rate limiting) before ever reaching mux. +func NewMetrics(mux *http.ServeMux) func(http.Handler) http.Handler { + return func(next http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _, pattern := mux.Handler(r) + if pattern == "" { + pattern = "unmatched" + } + + if pattern == legacyMetricsPattern { + next.ServeHTTP(w, r) + return + } + + start := time.Now() + wrapped := &LoggingResponseWriter{ResponseWriter: w, statusCode: http.StatusOK} + next.ServeHTTP(wrapped, r) + duration := time.Since(start) + + status := strconv.Itoa(wrapped.statusCode) + metrics.HTTPRequestsTotal.WithLabelValues(r.Method, pattern, status).Inc() + metrics.HTTPRequestDuration.WithLabelValues(r.Method, pattern, status).Observe(duration.Seconds()) + }) + } +} diff --git a/services/api/middleware/ratelimit.go b/services/api/middleware/ratelimit.go index a90080d..b6ce335 100644 --- a/services/api/middleware/ratelimit.go +++ b/services/api/middleware/ratelimit.go @@ -13,6 +13,7 @@ import ( "time" "github.com/Depo-dev/trident/services/api/internal/httputil" + "github.com/Depo-dev/trident/services/api/internal/metrics" "github.com/jackc/pgx/v5" "github.com/redis/go-redis/v9" ) @@ -235,6 +236,7 @@ func TieredRateLimit(cfg RateLimitConfig) func(http.Handler) http.Handler { if !allowed { rlRejected.Add(1) + metrics.RateLimitRejectionsTotal.WithLabelValues("per_key").Inc() retryAfter := int64(math.Ceil(tcfg.Window.Seconds())) w.Header().Set("Retry-After", strconv.FormatInt(retryAfter, 10)) httputil.WriteErrorCtx(r.Context(), w, http.StatusTooManyRequests, httputil.RATE_LIMITED, "rate limit exceeded") diff --git a/services/api/middleware/ratelimit_test.go b/services/api/middleware/ratelimit_test.go index 64c0ad6..703f9f2 100644 --- a/services/api/middleware/ratelimit_test.go +++ b/services/api/middleware/ratelimit_test.go @@ -9,7 +9,9 @@ import ( "testing" "time" + "github.com/Depo-dev/trident/services/api/internal/metrics" "github.com/jackc/pgx/v5" + "github.com/prometheus/client_golang/prometheus/testutil" ) // --------------------------------------------------------------------------- @@ -153,6 +155,26 @@ func TestTieredRateLimit_Rejects_Returns429WithHeaders(t *testing.T) { } } +// TestTieredRateLimit_Rejects_RecordsPrometheusMetric verifies a 429 from the +// per-key tiered limiter increments trident_ratelimit_rejections_total{limiter="per_key"} +// (issue #58). +func TestTieredRateLimit_Rejects_RecordsPrometheusMetric(t *testing.T) { + resetCounters() + before := testutil.ToFloat64(metrics.RateLimitRejectionsTotal.WithLabelValues("per_key")) + + cfg := RateLimitConfig{SliderFn: alwaysReject, Tiers: testTiers()} + mw := TieredRateLimit(cfg)(noop()) + rec := httptest.NewRecorder() + mw.ServeHTTP(rec, apiKeyReq("key")) + + if rec.Code != http.StatusTooManyRequests { + t.Fatalf("want 429, got %d", rec.Code) + } + if got := testutil.ToFloat64(metrics.RateLimitRejectionsTotal.WithLabelValues("per_key")); got != before+1 { + t.Errorf("per_key rejections total: want %v, got %v", before+1, got) + } +} + func TestTieredRateLimit_FailOpen_OnSliderError(t *testing.T) { resetCounters() errSlider := func(_ context.Context, _ string, _, _ int64) (bool, int64, error) { diff --git a/services/api/ws/hub.go b/services/api/ws/hub.go index 51f518e..36d0465 100644 --- a/services/api/ws/hub.go +++ b/services/api/ws/hub.go @@ -5,6 +5,8 @@ package ws import ( "log/slog" "sync" + + "github.com/Depo-dev/trident/services/api/internal/metrics" ) // maxConsecutiveDrops is the fill policy threshold (issue #224): a subscriber @@ -76,6 +78,8 @@ func (h *Hub) register(s subscriber) { h.mu.Lock() h.clients[s] = struct{}{} h.mu.Unlock() + metrics.WSActiveConnections.Inc() + metrics.WSConnectsTotal.Inc() slog.Debug("ws: client registered", "contractId", s.getContractID()) } @@ -83,12 +87,17 @@ func (h *Hub) register(s subscriber) { // can exit cleanly. func (h *Hub) unregister(s subscriber) { h.mu.Lock() - if _, ok := h.clients[s]; ok { + _, ok := h.clients[s] + if ok { delete(h.clients, s) delete(h.dropStreak, s) s.shutdown() } h.mu.Unlock() + if ok { + metrics.WSActiveConnections.Dec() + metrics.WSDisconnectsTotal.Inc() + } slog.Debug("ws: client unregistered", "contractId", s.getContractID()) } @@ -111,10 +120,12 @@ func (h *Hub) Broadcast(contractID string, msg []byte) { } if s.trySend(msg) { h.dropStreak[s] = 0 + metrics.WSMessagesTotal.WithLabelValues("sent").Inc() continue } metricMessagesDropped.Add(1) + metrics.WSMessagesTotal.WithLabelValues("dropped").Inc() h.dropStreak[s]++ slog.Warn("ws: dropping message for slow client", "contractId", contractID, "streak", h.dropStreak[s]) diff --git a/services/api/ws/metrics_prom_test.go b/services/api/ws/metrics_prom_test.go new file mode 100644 index 0000000..2821090 --- /dev/null +++ b/services/api/ws/metrics_prom_test.go @@ -0,0 +1,66 @@ +package ws + +import ( + "testing" + + "github.com/Depo-dev/trident/services/api/internal/metrics" + "github.com/prometheus/client_golang/prometheus/testutil" +) + +// TestHub_RegisterUnregisterUpdatesPrometheusMetrics verifies register/ +// unregister move the active-connections gauge and connect/disconnect +// counters exposed on the internal metrics port (issue #58). +func TestHub_RegisterUnregisterUpdatesPrometheusMetrics(t *testing.T) { + h := NewHub() + c := &client{contractID: "contract-abc", send: make(chan []byte, 8)} + + activeBefore := testutil.ToFloat64(metrics.WSActiveConnections) + connectsBefore := testutil.ToFloat64(metrics.WSConnectsTotal) + + h.register(c) + + if got := testutil.ToFloat64(metrics.WSActiveConnections); got != activeBefore+1 { + t.Errorf("active connections after register: want %v, got %v", activeBefore+1, got) + } + if got := testutil.ToFloat64(metrics.WSConnectsTotal); got != connectsBefore+1 { + t.Errorf("connects total after register: want %v, got %v", connectsBefore+1, got) + } + + disconnectsBefore := testutil.ToFloat64(metrics.WSDisconnectsTotal) + h.unregister(c) + + if got := testutil.ToFloat64(metrics.WSActiveConnections); got != activeBefore { + t.Errorf("active connections after unregister: want %v, got %v", activeBefore, got) + } + if got := testutil.ToFloat64(metrics.WSDisconnectsTotal); got != disconnectsBefore+1 { + t.Errorf("disconnects total after unregister: want %v, got %v", disconnectsBefore+1, got) + } + + // A second unregister of the same (already-removed) client must not double-count. + h.unregister(c) + if got := testutil.ToFloat64(metrics.WSDisconnectsTotal); got != disconnectsBefore+1 { + t.Errorf("disconnects total after redundant unregister: want %v, got %v", disconnectsBefore+1, got) + } +} + +// TestHub_BroadcastUpdatesMessageCounters verifies sent/dropped outcomes are +// recorded on trident_ws_messages_total. +func TestHub_BroadcastUpdatesMessageCounters(t *testing.T) { + h := NewHub() + c := &client{contractID: "contract-msg", send: make(chan []byte, 1)} + h.register(c) + defer h.unregister(c) + + sentBefore := testutil.ToFloat64(metrics.WSMessagesTotal.WithLabelValues("sent")) + droppedBefore := testutil.ToFloat64(metrics.WSMessagesTotal.WithLabelValues("dropped")) + + h.Broadcast("contract-msg", []byte("first")) // fills the buffer, delivered + h.Broadcast("contract-msg", []byte("second")) // buffer full, dropped + + if got := testutil.ToFloat64(metrics.WSMessagesTotal.WithLabelValues("sent")); got != sentBefore+1 { + t.Errorf("sent counter: want %v, got %v", sentBefore+1, got) + } + if got := testutil.ToFloat64(metrics.WSMessagesTotal.WithLabelValues("dropped")); got != droppedBefore+1 { + t.Errorf("dropped counter: want %v, got %v", droppedBefore+1, got) + } +}