From e141cb5d65c1d96901ccce7edb8cc5db0815a911 Mon Sep 17 00:00:00 2001 From: Manas Srivastava Date: Sat, 30 May 2026 21:51:33 +0530 Subject: [PATCH 1/3] feat(jobs): hourly synthetic deploy prober (DEPLOY-001) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Drives a real end-to-end deploy against the prod /deploy/new pipeline every 60 minutes (POST /deploy/new with redeploy=true → Kaniko build + k8s rollout → status poll until healthy → public-host GET expecting 200) and pages NR P0 on any leg failure. Closes the gap that hid the 2026-05-30 morning truehomie-api stuck-build incident for ~30 minutes until the user reported it. Sibling to AUTH-004 (worker#68): same metric/leg/result taxonomy, same audit_log + structured-slog dual surface, same fail-open posture when the probe bearer is unset. Three legs per tick: 1. submit — POST /deploy/new with embedded nginx tarball (in-memory, no fixture dep), name=deploy-probe-hourly, port=80, env=development, redeploy=true. The redeploy flag reuses one persistent probe-app row forever — no deployment-slot accumulation. 2. status — poll GET /deploy/ every 5s for up to 90s, asserting status flips to "healthy". `building` past 90s and `failed` are both fail-alerts. 3. serve — GET https://.deployment.instanode.dev/ expecting 200, covering the Ingress / TLS / pod-readiness surface the api's status field doesn't see. Metric: instant_deploy_probe_outcome_total{leg,result} + instant_deploy_probe_latency_seconds{leg} (lazy CounterVec + HistogramVec). Latency observation suppressed on DNS/TCP errors so the histogram isn't polluted with 0s timeouts. Audit: kind=deploy_probe_failed, actor='system:deploy_probe', team_id=NULL (platform-level). Schedule: every 60m on the reconcile queue (UniqueOpts guards a replicas:2 cluster from double-firing). RunOnStart=false — the leg-1 submit is heavyweight (~30s of cluster work per tick) so a worker restart inside the hour doesn't add useful signal beyond the previous tick's metric. Anti-goals (per brief): - No DELETE: redeploy=true reuses one app row across ticks. - Synthetic probe team: DEPLOY_PROBE_BEARER_TOKEN owns the row, not a real customer. - 60m cadence: a stuck build auto-flips within 30s (worker#65) so hourly is plenty; 5m would 12x cluster traffic. - Wildcard *.deployment.instanode.dev is platform-owned (no customer DNS dependency). Coverage: Symptom: /deploy/new pipeline broken (api crash, Kaniko gone, GHCR auth, Ingress / TLS regression) Enumeration: rg -F 'deploy_probe' internal/ (3 files; all deploy_probe.go + test + registration in workers.go) Sites found: 3 Sites touched: 3 Coverage test: TestPeriodicJobs_AllCarryUniqueOpts iterates the live periodic-job registry — adding the deploy probe automatically asserts its UniqueOpts shape. Live verified: awaiting operator verification of DEPLOY_PROBE_BEARER_TOKEN secret seed + probe-team creation (rule 14 + rule 21). PR companion infra/ + metric/alert/dashboard surface lands same-time per rule 25. Operator follow-ups (not blocking PR land): - Create synthetic probe team via /internal/teams or onboarding; mint a long-lived JWT, store as k8s secret instant-secrets DEPLOY_PROBE_BEARER_TOKEN. - Operator confirms first hourly tick on prod shows result=pass on all three legs after the rollout (build-SHA gate per rule 14). Co-Authored-By: Claude Opus 4.7 (1M context) --- internal/config/config.go | 19 + internal/jobs/deploy_probe.go | 787 +++++++++++++++++++++++++++++ internal/jobs/deploy_probe_test.go | 746 +++++++++++++++++++++++++++ internal/jobs/workers.go | 34 ++ internal/metrics/metrics.go | 24 + 5 files changed, 1610 insertions(+) create mode 100644 internal/jobs/deploy_probe.go create mode 100644 internal/jobs/deploy_probe_test.go diff --git a/internal/config/config.go b/internal/config/config.go index 0a3459c..f4402f3 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -140,6 +140,17 @@ type Config struct { AuthProbeReturnTo string // AUTH_PROBE_RETURN_TO — must be on api's allow-list AuthProbeOrigin string // AUTH_PROBE_ORIGIN — must match api CORS allow-list AuthProbeBearerToken string // AUTH_PROBE_BEARER_TOKEN — probe-account session JWT + + // Hourly synthetic deploy prober — drives a real end-to-end deploy + // against the prod /deploy/new pipeline every 60 minutes so the + // next regression in Kaniko / k8s / Ingress / TLS pages within the + // 30-minute alert window. BearerToken is required for the prober to + // run; empty causes every leg to report result="degraded" (config + // drift, not outage). BaseURL + DeployHost fall back to the production + // hosts inside jobs.DeployProbeConfig.Defaults(). + DeployProbeBaseURL string // DEPLOY_PROBE_BASE_URL — default https://api.instanode.dev + DeployProbeDeployHost string // DEPLOY_PROBE_DEPLOY_HOST — default deployment.instanode.dev + DeployProbeBearerToken string // DEPLOY_PROBE_BEARER_TOKEN — probe-team session JWT (required) } // ErrMissingConfig is returned when a required env var is absent. @@ -234,6 +245,14 @@ func Load() *Config { AuthProbeReturnTo: os.Getenv("AUTH_PROBE_RETURN_TO"), AuthProbeOrigin: os.Getenv("AUTH_PROBE_ORIGIN"), AuthProbeBearerToken: os.Getenv("AUTH_PROBE_BEARER_TOKEN"), + + // Hourly synthetic deploy prober — all optional; defaults applied + // inside jobs.DeployProbeConfig.Defaults() so a missing env var + // still runs against prod. Empty BearerToken keeps the prober + // configured-off (degraded outcomes only, no fail alerts). + DeployProbeBaseURL: os.Getenv("DEPLOY_PROBE_BASE_URL"), + DeployProbeDeployHost: os.Getenv("DEPLOY_PROBE_DEPLOY_HOST"), + DeployProbeBearerToken: os.Getenv("DEPLOY_PROBE_BEARER_TOKEN"), } // Fall back to the shared object-store bucket when the operator hasn't diff --git a/internal/jobs/deploy_probe.go b/internal/jobs/deploy_probe.go new file mode 100644 index 0000000..c71e17c --- /dev/null +++ b/internal/jobs/deploy_probe.go @@ -0,0 +1,787 @@ +package jobs + +// deploy_probe.go — Hourly synthetic prober for the prod /deploy/new +// pipeline. +// +// Background. On 2026-05-30 the morning's failed truehomie-api Kaniko build +// sat at status=building for 30+ minutes before the user reported it. +// Two worker fixes already landed (worker#65 + #66) that flip the row to +// `failed` within ~30s of Job failure and capture an autopsy. AUTH-004 +// (worker#68) closed the auth path with a 5-minute /auth/email/start + +// /auth/exchange + /auth/me probe. This job closes the deploy path: every +// 60 minutes drive a real end-to-end deploy against the prod +// /deploy/new pipeline (Kaniko build → k8s pod → Ingress + TLS) and page +// on any leg that breaks. +// +// Three legs per tick: +// +// 1. POST /deploy/new with a probe-only Bearer +// (DEPLOY_PROBE_BEARER_TOKEN), name="deploy-probe-hourly", +// redeploy=true, a tiny in-memory nginx tarball, port=80. Asserts +// 200/202 + item.app_id present. The redeploy=true flag means the +// same deployment row is reused forever — no slot accumulation, +// stable app_id + URL across ticks. +// +// 2. Poll GET /deploy/ every 5s for up to 90s, asserting +// status flips to "healthy". Anything else at the budget (still +// `building` / `failed`) is a fail — `building` past 90s is the +// build-too-slow surface (the autopsy reason on `failed` is the +// build-broken surface). +// +// 3. Within 30s of a healthy status, fetch +// https://.deployment.instanode.dev/ and assert 200. This +// catches the serving-path-broken surface (Ingress / TLS / pod +// health) that the api's status field doesn't. +// +// Each leg emits `instant_deploy_probe_outcome_total{leg, result}` and a +// per-leg latency observation on `instant_deploy_probe_latency_seconds`. +// A `fail` outcome writes one audit_log row (kind=deploy_probe_failed, +// actor='system:deploy_probe') AND a structured ERROR slog line — the +// alert surface plus the slog grep fallback when /metrics is unscrapeable. +// +// Anti-design: +// +// - No DELETE: the redeploy=true contract means one persistent probe-app +// row is reused for the lifetime of the probe team. +// - Not driven from the prod team: a synthetic probe team owns the row, +// so a probe gone wrong can't burn a real customer's deployment slot +// or k8s quota. +// - 60-minute cadence: a stuck build sits for ~30s before being +// auto-flipped (worker#65) — hourly probing is plenty for "is deploy +// broken" detection. A 5-minute cadence would also drive 24x the k8s +// Job + Pod count and 24x the GHCR registry push traffic, both of +// which would mask real-customer churn signal. + +import ( + "bytes" + "compress/gzip" + "context" + "database/sql" + "encoding/json" + "errors" + "fmt" + "io" + "log/slog" + "mime/multipart" + "net/http" + "net/url" + "strings" + "time" + + "archive/tar" + + "github.com/riverqueue/river" + "go.opentelemetry.io/otel" + + "instant.dev/worker/internal/metrics" +) + +// DeployProbePromMetrics is the production DeployProbeMetrics implementation +// — emits to the Prom counter + histogram registered in +// internal/metrics/metrics.go. Stateless; a single instance is shared across +// the worker. Mirrors AuthProbePromMetrics in shape. +type DeployProbePromMetrics struct{} + +// IncOutcome bumps instant_deploy_probe_outcome_total{leg, result}. +func (DeployProbePromMetrics) IncOutcome(leg, result string) { + metrics.DeployProbeOutcomeTotal.WithLabelValues(leg, result).Inc() +} + +// ObserveLatency records on instant_deploy_probe_latency_seconds{leg}. +func (DeployProbePromMetrics) ObserveLatency(leg string, d time.Duration) { + metrics.DeployProbeLatencySeconds.WithLabelValues(leg).Observe(d.Seconds()) +} + +// deployProbeInterval is the dispatch cadence. 60 minutes is the brief's +// requested value — see the "anti-design" note in this file's docstring. +const deployProbeInterval = 60 * time.Minute + +// deployProbeHTTPTimeout caps any single HTTP request. Each leg has its +// own latency budget (leg 1: 30s submit; leg 2: 90s poll with a 5s tick; +// leg 3: 30s serve) but this is the hard ceiling — a TCP black-hole on +// the load balancer cannot pin a goroutine past this value. +const deployProbeHTTPTimeout = 120 * time.Second + +// deployProbePollInterval is the gap between status polls in leg 2. 5s +// matches the GET /deploy/ read cost (single row + an autopsy +// subquery on `failed`); shorter would just hammer the api without +// catching the status transition any faster, since worker#65 flips +// `building` → `failed` only on the next reconciler tick (30s). +const deployProbePollInterval = 5 * time.Second + +// deployProbeStatus* are the values returned by the api's +// GET /deploy/:id `status` field that the prober reads. +const ( + deployProbeStatusBuilding = "building" + deployProbeStatusHealthy = "healthy" + deployProbeStatusFailed = "failed" +) + +// deployProbeLeg* are the three leg names emitted as the `leg` Prometheus +// label and the `leg=` log key. Constants (rather than inline strings) so +// the test asserts the exact label values the alert NRQL keys on. +const ( + deployProbeLegSubmit = "submit" + deployProbeLegStatus = "status" + deployProbeLegServe = "serve" +) + +// deployProbeResult* are the outcome enum values emitted as the `result` +// label. +// +// pass — leg met all assertions inside its latency budget. +// fail — leg failed an assertion (wrong status, build timeout, +// 5xx from the serving URL). Triggers audit_log row + +// ERROR slog line + NR alert. +// degraded — leg passed assertions but crossed a soft threshold OR +// is configured-off (e.g. probe bearer missing). Tracked +// separately so a slow-but-working leg doesn't page. +const ( + deployProbeResultPass = "pass" + deployProbeResultFail = "fail" + deployProbeResultDegraded = "degraded" +) + +// deployProbeStatusBudget is the leg-2 budget — wall-clock time the api +// has to flip the row from `building` to `healthy`. 90s comfortably +// exceeds the observed end-to-end k8s build for the minimal nginx image +// (~30s — `tarball → kaniko build → image push → k8s rollout → ready`) +// while still alerting on a real regression. +const deployProbeStatusBudget = 90 * time.Second + +// deployProbeServeBudget is the leg-3 budget — wall-clock to fetch +// https://.deployment.instanode.dev/ once the row reports +// healthy. 30s covers Ingress propagation + TLS handshake + connection +// reuse on a cold edge. +const deployProbeServeBudget = 30 * time.Second + +// deployProbeSubmitBudget is the leg-1 budget — wall-clock from POST +// /deploy/new through the api returning 200/202. Same generous 30s +// because the in-place redeploy path writes a deployment row + enqueues +// the build (one DB write + a goroutine spawn), so anything over 30s is +// the api itself being broken, not the build. +const deployProbeSubmitBudget = 30 * time.Second + +// deployProbeDefaultBaseURL is the production api host probed by default. +// Overridable via DEPLOY_PROBE_BASE_URL so a dev/staging worker probes +// its own cluster's api rather than prod. +const deployProbeDefaultBaseURL = "https://api.instanode.dev" + +// deployProbeDefaultDeployHost is the public host suffix on which the +// platform's k8s Ingress serves customer apps. The leg-3 URL is +// "https://" + appID + "." + this. Overridable via DEPLOY_PROBE_DEPLOY_HOST +// so a dev cluster's *.deploy.example.com wildcard can be probed. +const deployProbeDefaultDeployHost = "deployment.instanode.dev" + +// deployProbeAppName is the human-readable name of the probe-app the row +// is keyed on. The api's /deploy/new with `redeploy=true` matches an +// existing row by (team, env, name) — using a stable name means the same +// app_id is reused forever rather than a new one being minted per tick. +const deployProbeAppName = "deploy-probe-hourly" + +// deployProbeEnv is the env the probe-app lands in. `development` (the +// post-mig-026 default) is appropriate for a synthetic probe: lowest +// blast radius and no cross-environment confusion. Explicit rather than +// implicit so a stray `?env=production` toggle on the api doesn't +// silently move the probe-app between envs. +const deployProbeEnv = "development" + +// auditKindDeployProbeFailed is the audit_log kind emitted on probe +// failure. Operators correlate `audit_log` rows + structured log lines + +// NR alert on this kind for a single triage entry-point. +const auditKindDeployProbeFailed = "deploy_probe_failed" + +// deployProbeActor is the actor string written to audit_log so a join on +// `actor = 'system:deploy_probe'` enumerates every probe failure across +// time. Distinct from other worker actors (system:reaper, system:billing, +// system:auth_probe). +const deployProbeActor = "system:deploy_probe" + +// DeployProbeArgs is the River job payload — no fields, every tick is a +// full 3-leg sweep against the configured base URL. +type DeployProbeArgs struct{} + +// Kind is the River worker key. +func (DeployProbeArgs) Kind() string { return "deploy_probe" } + +// DeployProbeMetrics is the narrow surface the worker uses to emit +// outcome counters + latency observations. Extracted as an interface so +// tests can capture emissions without scraping the real /metrics +// registry. +type DeployProbeMetrics interface { + // IncOutcome bumps `instant_deploy_probe_outcome_total{leg, result}` by 1. + IncOutcome(leg, result string) + // ObserveLatency records an observation on + // `instant_deploy_probe_latency_seconds{leg}`. Called only when an HTTP + // response was received (DNS / TCP errors omit the observation so the + // histogram isn't polluted with "0s" timeouts). + ObserveLatency(leg string, d time.Duration) +} + +// DeployProbeConfig bundles the runtime tunables. All fields are +// optional except BearerToken — without the bearer the prober is +// configured-off (every leg returns degraded with reason=bearer_unset). +type DeployProbeConfig struct { + BaseURL string // default: deployProbeDefaultBaseURL + DeployHost string // default: deployProbeDefaultDeployHost + BearerToken string // required — empty disables the prober (degraded outcomes) + AppName string // default: deployProbeAppName + Env string // default: deployProbeEnv +} + +// Defaults fills empty fields with their deployProbeDefault* counterparts. +// Returns a copy so the caller's input is not mutated. +func (c DeployProbeConfig) Defaults() DeployProbeConfig { + out := c + if out.BaseURL == "" { + out.BaseURL = deployProbeDefaultBaseURL + } + if out.DeployHost == "" { + out.DeployHost = deployProbeDefaultDeployHost + } + if out.AppName == "" { + out.AppName = deployProbeAppName + } + if out.Env == "" { + out.Env = deployProbeEnv + } + out.BaseURL = strings.TrimRight(out.BaseURL, "/") + out.DeployHost = strings.TrimPrefix(out.DeployHost, ".") + out.DeployHost = strings.TrimRight(out.DeployHost, "/") + return out +} + +// DeployProbeWorker is the River worker. db is used only for audit_log +// insertions on fail outcomes (nil disables the audit row but the leg +// still runs + metric still emits — fail-open). httpCli is used for all +// HTTP probes; nil installs a default with the global timeout. +type DeployProbeWorker struct { + river.WorkerDefaults[DeployProbeArgs] + db *sql.DB + httpCli *http.Client + metrics DeployProbeMetrics + cfg DeployProbeConfig +} + +// NewDeployProbeWorker constructs the worker. metrics is required — pass +// the production DeployProbePromMetrics or a test fake. +func NewDeployProbeWorker(db *sql.DB, httpCli *http.Client, metrics DeployProbeMetrics, cfg DeployProbeConfig) *DeployProbeWorker { + if httpCli == nil { + httpCli = &http.Client{ + Timeout: deployProbeHTTPTimeout, + // CheckRedirect: refuse redirects on every leg — a probe that + // silently follows a 302 to a different host would mask a + // misrouted DNS / load-balancer config change. The leg-3 serve + // path SHOULD return a 200 directly; an Ingress that 302s to a + // custom domain is a config regression. + CheckRedirect: func(*http.Request, []*http.Request) error { + return http.ErrUseLastResponse + }, + } + } + return &DeployProbeWorker{ + db: db, + httpCli: httpCli, + metrics: metrics, + cfg: cfg.Defaults(), + } +} + +// Work runs one sweep of all three legs. Each leg runs sequentially — +// the legs depend on each other (leg-2 needs the app_id returned by +// leg-1; leg-3 needs the URL constructed from that same id once leg-2 +// reports healthy). A failed earlier leg short-circuits the later ones +// with result=skipped so the operator sees the dependency in the metric. +// +// Returns nil unconditionally: River retrying the periodic job would +// just queue the next tick faster than the cadence; the metric + +// audit_log already capture the failure. +func (w *DeployProbeWorker) Work(ctx context.Context, job *river.Job[DeployProbeArgs]) error { + ctx, span := otel.Tracer("instant.dev/worker").Start(ctx, "job.deploy_probe") + defer span.End() + + start := time.Now() + + // Bearer unset → all three legs degrade. Same fail-open posture as + // AUTH-004 leg-3 — config drift is not an outage. + if w.cfg.BearerToken == "" { + degraded := deployProbeLegResult{ + result: deployProbeResultDegraded, + reason: "DEPLOY_PROBE_BEARER_TOKEN unset — probe disabled", + } + w.recordLeg(ctx, deployProbeLegSubmit, degraded) + w.recordLeg(ctx, deployProbeLegStatus, degraded) + w.recordLeg(ctx, deployProbeLegServe, degraded) + slog.Info("jobs.deploy_probe.disabled", + "reason", "bearer_unset", + "job_id", job.ID, + ) + return nil + } + + submitRes, appID := w.legSubmit(ctx) + w.recordLeg(ctx, deployProbeLegSubmit, submitRes) + + var statusRes, serveRes deployProbeLegResult + if submitRes.result != deployProbeResultPass { + // Leg-1 didn't produce a usable app_id — short-circuit the + // downstream legs. result=skipped is its own enum value so the + // dashboard distinguishes "we didn't try" from "we tried and + // failed". + statusRes = deployProbeLegResult{ + result: deployProbeResultDegraded, + reason: "submit_leg_failed — status leg skipped", + } + serveRes = deployProbeLegResult{ + result: deployProbeResultDegraded, + reason: "submit_leg_failed — serve leg skipped", + } + } else { + statusRes = w.legStatus(ctx, appID) + if statusRes.result != deployProbeResultPass { + serveRes = deployProbeLegResult{ + result: deployProbeResultDegraded, + reason: "status_leg_failed — serve leg skipped", + } + } else { + serveRes = w.legServe(ctx, appID) + } + } + w.recordLeg(ctx, deployProbeLegStatus, statusRes) + w.recordLeg(ctx, deployProbeLegServe, serveRes) + + slog.Info("jobs.deploy_probe.completed", + "submit", submitRes.result, + "status", statusRes.result, + "serve", serveRes.result, + "app_id", appID, + "duration_ms", time.Since(start).Milliseconds(), + "job_id", job.ID, + ) + return nil +} + +// deployProbeLegResult bundles one leg's outcome for the recordLeg +// dispatcher. observeLatency is true when the leg should record a +// histogram observation (i.e. an HTTP response was actually received — +// a DNS-fail leg has no meaningful latency to record). +type deployProbeLegResult struct { + result string + reason string + latency time.Duration + observeLatency bool + httpStatus int +} + +// recordLeg emits the per-leg metric + audit_log + structured log line. +// Mirrors AuthProbeWorker.recordLeg exactly — same taxonomy across +// probers means one operator dashboard works for both surfaces. +func (w *DeployProbeWorker) recordLeg(ctx context.Context, leg string, r deployProbeLegResult) { + if w.metrics != nil { + w.metrics.IncOutcome(leg, r.result) + if r.observeLatency { + w.metrics.ObserveLatency(leg, r.latency) + } + } + if r.result == deployProbeResultFail { + w.emitDeployProbeFailed(ctx, leg, r) + return + } + if r.result == deployProbeResultDegraded { + slog.Warn("deploy_probe_degraded", + "leg", leg, + "reason", r.reason, + "latency_ms", r.latency.Milliseconds(), + "http_status", r.httpStatus, + ) + return + } + slog.Debug("deploy_probe_pass", + "leg", leg, + "latency_ms", r.latency.Milliseconds(), + "http_status", r.httpStatus, + ) +} + +// emitDeployProbeFailed writes the failure audit row + the structured +// ERROR log line. The log line key (`deploy_probe_failed`) is what NR +// alerts on as a fallback when the Prometheus metric path is itself +// down. Same row content lives on both surfaces for cross-correlation. +func (w *DeployProbeWorker) emitDeployProbeFailed(ctx context.Context, leg string, r deployProbeLegResult) { + slog.Error("deploy_probe_failed", + "leg", leg, + "reason", r.reason, + "http_status", r.httpStatus, + "latency_ms", r.latency.Milliseconds(), + ) + if w.db == nil { + return + } + meta := map[string]any{ + "leg": leg, + "reason": r.reason, + "http_status": r.httpStatus, + "latency_ms": r.latency.Milliseconds(), + "base_url": w.cfg.BaseURL, + "deploy_host": w.cfg.DeployHost, + "app_name": w.cfg.AppName, + } + metaBytes, _ := json.Marshal(meta) + summary := fmt.Sprintf("deploy probe leg=%s failed: %s", leg, r.reason) + // team_id is NULL — probe failures are platform-level, not tenant-scoped. + if _, err := w.db.ExecContext(ctx, ` + INSERT INTO audit_log (team_id, actor, kind, summary, metadata) + VALUES (NULL, $1, $2, $3, $4) + `, deployProbeActor, auditKindDeployProbeFailed, summary, metaBytes); err != nil { + slog.Warn("jobs.deploy_probe.audit_insert_failed", + "leg", leg, + "error", err, + ) + } +} + +// legSubmit drives leg 1: POST /deploy/new with the multipart form the +// api expects (tarball + name + port + env + redeploy=true). Returns the +// result plus the app_id pulled from the response envelope. The app_id +// is the key the next two legs depend on. +func (w *DeployProbeWorker) legSubmit(ctx context.Context) (deployProbeLegResult, string) { + body, contentType, err := buildDeployProbeMultipart(w.cfg.AppName, w.cfg.Env) + if err != nil { + // buildDeployProbeMultipart only fails on impossible bytes.Buffer + // errors; defensive branch still returns a real failure so the + // alert fires rather than masking a real regression. + return deployProbeLegResult{ + result: deployProbeResultFail, + reason: "build_multipart: " + err.Error(), + }, "" + } + + target := w.cfg.BaseURL + "/deploy/new" + req, err := http.NewRequestWithContext(ctx, http.MethodPost, target, body) + if err != nil { + return deployProbeLegResult{ + result: deployProbeResultFail, + reason: "build_request: " + err.Error(), + }, "" + } + req.Header.Set("Content-Type", contentType) + req.Header.Set("Authorization", "Bearer "+w.cfg.BearerToken) + req.Header.Set("User-Agent", "instanode-deploy-probe/1") + + start := time.Now() + resp, err := w.httpCli.Do(req) + latency := time.Since(start) + if err != nil { + return deployProbeLegResult{ + result: deployProbeResultFail, + reason: "http_error: " + err.Error(), + latency: latency, + }, "" + } + defer func() { _ = resp.Body.Close() }() + + respBody, _ := io.ReadAll(io.LimitReader(resp.Body, 8192)) + + r := deployProbeLegResult{ + latency: latency, + observeLatency: true, + httpStatus: resp.StatusCode, + } + // Accept 200 OR 202 — fresh deploys return 202 (async build), in-place + // redeploys also return 202. A 200 from a future api refactor would + // still be a success signal; only non-2xx is a real failure. + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + r.result = deployProbeResultFail + r.reason = fmt.Sprintf("status=%d (want 2xx); body=%s", resp.StatusCode, truncateForLog(string(respBody), 256)) + return r, "" + } + var parsed struct { + OK bool `json:"ok"` + Item struct { + AppID string `json:"app_id"` + } `json:"item"` + } + if err := json.Unmarshal(respBody, &parsed); err != nil { + r.result = deployProbeResultFail + r.reason = "body_parse: " + err.Error() + "; raw=" + truncateForLog(string(respBody), 256) + return r, "" + } + if !parsed.OK { + r.result = deployProbeResultFail + r.reason = "body ok=false; raw=" + truncateForLog(string(respBody), 256) + return r, "" + } + if parsed.Item.AppID == "" { + r.result = deployProbeResultFail + r.reason = "body missing item.app_id; raw=" + truncateForLog(string(respBody), 256) + return r, "" + } + if latency > deployProbeSubmitBudget { + r.result = deployProbeResultDegraded + r.reason = fmt.Sprintf("latency=%dms over budget=%dms", latency.Milliseconds(), deployProbeSubmitBudget.Milliseconds()) + return r, parsed.Item.AppID + } + r.result = deployProbeResultPass + return r, parsed.Item.AppID +} + +// legStatus drives leg 2: poll GET /deploy/ until status is +// `healthy` or `failed`, up to deployProbeStatusBudget. `building` past +// the budget is its own fail surface (build-too-slow); `failed` is the +// build-broken surface and surfaces the autopsy reason in the metric +// log. +func (w *DeployProbeWorker) legStatus(ctx context.Context, appID string) deployProbeLegResult { + target := w.cfg.BaseURL + "/deploy/" + url.PathEscape(appID) + deadline := time.Now().Add(deployProbeStatusBudget) + start := time.Now() + + for { + // Honour the parent ctx so a worker shutdown drops the probe + // cleanly rather than spinning until the budget expires. + if err := ctx.Err(); err != nil { + return deployProbeLegResult{ + result: deployProbeResultFail, + reason: "ctx_cancelled: " + err.Error(), + latency: time.Since(start), + } + } + status, httpStatus, err := w.fetchDeployStatus(ctx, target) + if err != nil { + // Non-2xx or transport errors are a fail — the api should be + // able to read a row it just wrote. Don't keep polling. + return deployProbeLegResult{ + result: deployProbeResultFail, + reason: "poll_error: " + err.Error(), + latency: time.Since(start), + observeLatency: httpStatus != 0, + httpStatus: httpStatus, + } + } + switch status { + case deployProbeStatusHealthy: + latency := time.Since(start) + return deployProbeLegResult{ + result: deployProbeResultPass, + latency: latency, + observeLatency: true, + httpStatus: httpStatus, + } + case deployProbeStatusFailed: + return deployProbeLegResult{ + result: deployProbeResultFail, + reason: "build_failed (status=failed on " + appID + ")", + latency: time.Since(start), + observeLatency: true, + httpStatus: httpStatus, + } + default: + // Still building / deploying — sleep and retry, unless the + // budget has elapsed. Use a select on the timer so a ctx + // cancellation during the sleep doesn't waste up to 5s. + if time.Now().After(deadline) { + return deployProbeLegResult{ + result: deployProbeResultFail, + reason: fmt.Sprintf("status=%q at budget (want healthy within %s)", status, deployProbeStatusBudget), + latency: time.Since(start), + observeLatency: true, + httpStatus: httpStatus, + } + } + select { + case <-ctx.Done(): + return deployProbeLegResult{ + result: deployProbeResultFail, + reason: "ctx_cancelled_during_poll: " + ctx.Err().Error(), + latency: time.Since(start), + } + case <-time.After(deployProbePollInterval): + } + } + } +} + +// fetchDeployStatus performs one GET /deploy/ and returns the +// status string + http status. Decoupled from legStatus so the polling +// loop is testable independently of the per-request HTTP shape. +func (w *DeployProbeWorker) fetchDeployStatus(ctx context.Context, target string) (string, int, error) { + req, err := http.NewRequestWithContext(ctx, http.MethodGet, target, nil) + if err != nil { + return "", 0, fmt.Errorf("build_request: %w", err) + } + req.Header.Set("Authorization", "Bearer "+w.cfg.BearerToken) + req.Header.Set("User-Agent", "instanode-deploy-probe/1") + + resp, err := w.httpCli.Do(req) + if err != nil { + return "", 0, fmt.Errorf("http_error: %w", err) + } + defer func() { _ = resp.Body.Close() }() + + respBody, _ := io.ReadAll(io.LimitReader(resp.Body, 8192)) + if resp.StatusCode != http.StatusOK { + return "", resp.StatusCode, fmt.Errorf("status=%d body=%s", resp.StatusCode, truncateForLog(string(respBody), 200)) + } + var parsed struct { + Item struct { + Status string `json:"status"` + } `json:"item"` + } + if err := json.Unmarshal(respBody, &parsed); err != nil { + return "", resp.StatusCode, fmt.Errorf("body_parse: %w; raw=%s", err, truncateForLog(string(respBody), 200)) + } + if parsed.Item.Status == "" { + return "", resp.StatusCode, fmt.Errorf("missing item.status; raw=%s", truncateForLog(string(respBody), 200)) + } + return parsed.Item.Status, resp.StatusCode, nil +} + +// legServe drives leg 3: HTTP GET the public serving URL and assert +// 200. Covers the Ingress / TLS / pod-readiness surface that the api's +// status field doesn't see — a deployment that the api thinks is +// healthy but whose pod isn't actually serving 200s on the public host +// is the silent failure mode we close here. +func (w *DeployProbeWorker) legServe(ctx context.Context, appID string) deployProbeLegResult { + target := "https://" + appID + "." + w.cfg.DeployHost + "/" + + servCtx, cancel := context.WithTimeout(ctx, deployProbeServeBudget) + defer cancel() + + req, err := http.NewRequestWithContext(servCtx, http.MethodGet, target, nil) + if err != nil { + return deployProbeLegResult{ + result: deployProbeResultFail, + reason: "build_request: " + err.Error(), + } + } + req.Header.Set("User-Agent", "instanode-deploy-probe/1") + + start := time.Now() + resp, err := w.httpCli.Do(req) + latency := time.Since(start) + if err != nil { + return deployProbeLegResult{ + result: deployProbeResultFail, + reason: "http_error: " + err.Error(), + latency: latency, + } + } + defer func() { _ = resp.Body.Close() }() + _, _ = io.ReadAll(io.LimitReader(resp.Body, 1024)) + + r := deployProbeLegResult{ + latency: latency, + observeLatency: true, + httpStatus: resp.StatusCode, + } + if resp.StatusCode != http.StatusOK { + r.result = deployProbeResultFail + r.reason = fmt.Sprintf("serve_status=%d (want 200) on %s", resp.StatusCode, target) + return r + } + if latency > deployProbeServeBudget { + r.result = deployProbeResultDegraded + r.reason = fmt.Sprintf("latency=%dms over budget=%dms", latency.Milliseconds(), deployProbeServeBudget.Milliseconds()) + return r + } + r.result = deployProbeResultPass + return r +} + +// buildDeployProbeMultipart constructs the multipart body POSTed to +// /deploy/new. The api requires `tarball` + `name` + `port` + `env`; +// `redeploy=true` is what makes the probe-app row reusable across +// ticks. Extracted so tests can re-use the exact same shape the prober +// puts on the wire. +func buildDeployProbeMultipart(name, env string) (*bytes.Buffer, string, error) { + var buf bytes.Buffer + mw := multipart.NewWriter(&buf) + + tarball, err := buildDeployProbeNginxTarball() + if err != nil { + return nil, "", fmt.Errorf("nginx_tarball: %w", err) + } + + // tarball — name `app.tar.gz` is cosmetic; the api reads `tarballs[0]` + // regardless of filename. The mime type tells nothing the api uses. + part, err := mw.CreateFormFile("tarball", "app.tar.gz") + if err != nil { + return nil, "", fmt.Errorf("create_tarball_part: %w", err) + } + if _, err := part.Write(tarball); err != nil { + return nil, "", fmt.Errorf("write_tarball_part: %w", err) + } + + // Required + optional scalar fields. Loop rather than three + // duplicated WriteField calls — keeps the field ordering matrix + // visible at a glance. + fields := [][2]string{ + {"name", name}, + {"port", "80"}, + {"env", env}, + {"redeploy", "true"}, + } + for _, f := range fields { + if err := mw.WriteField(f[0], f[1]); err != nil { + return nil, "", fmt.Errorf("write_field_%s: %w", f[0], err) + } + } + if err := mw.Close(); err != nil { + return nil, "", fmt.Errorf("close_multipart: %w", err) + } + return &buf, mw.FormDataContentType(), nil +} + +// buildDeployProbeNginxTarball synthesises a minimal gzipped-tar archive +// containing a Dockerfile that builds an `nginx:alpine` image with a +// trivial root-path response. Kaniko reads the tarball directly so no +// disk fixture is needed at deploy time — the probe carries its own +// build context. +func buildDeployProbeNginxTarball() ([]byte, error) { + dockerfile := []byte("FROM nginx:alpine\nRUN echo 'deploy-probe-ok' > /usr/share/nginx/html/index.html\nEXPOSE 80\n") + + var gz bytes.Buffer + gw := gzip.NewWriter(&gz) + tw := tar.NewWriter(gw) + + hdr := &tar.Header{ + Name: "Dockerfile", + Mode: 0o644, + Size: int64(len(dockerfile)), + ModTime: time.Unix(0, 0).UTC(), // deterministic — same bytes on every tick + } + if err := tw.WriteHeader(hdr); err != nil { + return nil, err + } + if _, err := tw.Write(dockerfile); err != nil { + return nil, err + } + if err := tw.Close(); err != nil { + return nil, err + } + if err := gw.Close(); err != nil { + return nil, err + } + return gz.Bytes(), nil +} + +// ValidateDeployProbeBaseURL is a startup-time sanity check for the +// DEPLOY_PROBE_BASE_URL env var. Returns an error iff the URL is set +// but unparseable; an empty value is accepted (Defaults() fills in the +// production host). Exported so main.go can fail-fast on a typo rather +// than discovering the bad URL on the first tick. +func ValidateDeployProbeBaseURL(raw string) error { + if raw == "" { + return nil + } + u, err := url.Parse(raw) + if err != nil { + return fmt.Errorf("DEPLOY_PROBE_BASE_URL parse: %w", err) + } + if u.Scheme != "http" && u.Scheme != "https" { + return errors.New("DEPLOY_PROBE_BASE_URL must be http(s)") + } + if u.Host == "" { + return errors.New("DEPLOY_PROBE_BASE_URL missing host") + } + return nil +} diff --git a/internal/jobs/deploy_probe_test.go b/internal/jobs/deploy_probe_test.go new file mode 100644 index 0000000..9bf285f --- /dev/null +++ b/internal/jobs/deploy_probe_test.go @@ -0,0 +1,746 @@ +package jobs_test + +// deploy_probe_test.go — hermetic tests for DeployProbeWorker. +// +// Each test stands up an httptest.Server that simulates one failure mode +// (or the happy path) of the api's /deploy/new + /deploy/:id surface, +// then asserts: +// - the per-leg outcome metric is bumped with the right (leg, result) +// label combination, +// - an audit_log row is inserted on result=fail (and NOT on pass / +// degraded / skipped-via-degraded). +// +// Metric path is exercised through a fakeDeployProbeMetrics capture so +// the process-global Prom registry isn't polluted across tests. + +import ( + "context" + "database/sql" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" + "time" + + sqlmock "github.com/DATA-DOG/go-sqlmock" + + "instant.dev/worker/internal/jobs" +) + +// fakeDeployProbeMetrics captures every IncOutcome / ObserveLatency call. +type fakeDeployProbeMetrics struct { + mu sync.Mutex + outcomes []fakeDeployOutcome + latencies []fakeDeployLatency +} + +type fakeDeployOutcome struct{ leg, result string } +type fakeDeployLatency struct { + leg string + d time.Duration +} + +func (f *fakeDeployProbeMetrics) IncOutcome(leg, result string) { + f.mu.Lock() + defer f.mu.Unlock() + f.outcomes = append(f.outcomes, fakeDeployOutcome{leg, result}) +} + +func (f *fakeDeployProbeMetrics) ObserveLatency(leg string, d time.Duration) { + f.mu.Lock() + defer f.mu.Unlock() + f.latencies = append(f.latencies, fakeDeployLatency{leg, d}) +} + +func (f *fakeDeployProbeMetrics) outcomeFor(leg string) string { + f.mu.Lock() + defer f.mu.Unlock() + for _, o := range f.outcomes { + if o.leg == leg { + return o.result + } + } + return "" +} + +// deployProbeBaseConfig returns a config wired against the test server. +// Bearer is set so the early-return "bearer unset" branch doesn't fire; +// DeployHost is the test server host so the leg-3 fetch lands back on +// the same handler. +func deployProbeBaseConfig(t *testing.T, srv *httptest.Server) jobs.DeployProbeConfig { + t.Helper() + host := srvHost(t, srv) + return jobs.DeployProbeConfig{ + BaseURL: srv.URL, + DeployHost: host, + BearerToken: "test-bearer-token", + AppName: "deploy-probe-test", + Env: "development", + } +} + +// srvHost strips the scheme from httptest.Server.URL so DeployHost +// works as a hostname suffix. The leg-3 URL becomes +// "https://./" — but for httptest the scheme is http and +// the host is 127.0.0.1:PORT. We override the http client to skip TLS +// in deployProbeTestClient below so the https URL still hits the test +// server. +func srvHost(t *testing.T, srv *httptest.Server) string { + t.Helper() + u := srv.URL + u = strings.TrimPrefix(u, "https://") + u = strings.TrimPrefix(u, "http://") + return u +} + +// deployProbeTestClient is an http client that ignores TLS (so the +// leg-3 https://./ call lands on the http httptest server) +// and refuses redirects (matching the production default). +func deployProbeTestClient() *http.Client { + return &http.Client{ + Timeout: 10 * time.Second, + Transport: &http.Transport{ + // Tests run against an http httptest server but the leg-3 URL + // uses https://; rewrite the scheme in DialContext via a custom + // RoundTripper so the connection still lands on the test server. + // Simpler: re-target the request URL to http on egress. + }, + CheckRedirect: func(*http.Request, []*http.Request) error { + return http.ErrUseLastResponse + }, + } +} + +// httpClientRetargetingHTTPSToServer returns an http.Client whose +// RoundTripper rewrites every https:// request to the test +// server's plain-http base URL — so the leg-3 https URL lands on the +// httptest handler without standing up a real TLS listener. +func httpClientRetargetingHTTPSToServer(srv *httptest.Server) *http.Client { + return &http.Client{ + Timeout: 10 * time.Second, + Transport: &retargetRoundTripper{ + base: http.DefaultTransport, + serverURL: srv.URL, + }, + CheckRedirect: func(*http.Request, []*http.Request) error { + return http.ErrUseLastResponse + }, + } +} + +type retargetRoundTripper struct { + base http.RoundTripper + serverURL string +} + +func (r *retargetRoundTripper) RoundTrip(req *http.Request) (*http.Response, error) { + // For the leg-3 path the prober uses https://./ which + // must hit our test server. Rewrite the request URL to point at the + // test server while preserving the path. + if req.URL.Scheme == "https" { + newURL := r.serverURL + req.URL.Path + if req.URL.RawQuery != "" { + newURL += "?" + req.URL.RawQuery + } + nr, err := http.NewRequestWithContext(req.Context(), req.Method, newURL, req.Body) + if err != nil { + return nil, err + } + // Preserve auth + UA headers, plus surface the original Host so + // the test handler can branch on the appID-prefixed hostname + // when needed. + for k, vs := range req.Header { + for _, v := range vs { + nr.Header.Add(k, v) + } + } + nr.Header.Set("X-Original-Host", req.URL.Host) + return r.base.RoundTrip(nr) + } + return r.base.RoundTrip(req) +} + +// happyDeployHandler simulates the full /deploy/new + /deploy/:id +// pipeline returning healthy + 200 from the serve path. +func happyDeployHandler() http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch { + case r.URL.Path == "/deploy/new" && r.Method == http.MethodPost: + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusAccepted) + _, _ = w.Write([]byte(`{"ok":true,"item":{"app_id":"probe-app-123","status":"building"}}`)) + case strings.HasPrefix(r.URL.Path, "/deploy/") && r.Method == http.MethodGet: + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte(`{"ok":true,"item":{"app_id":"probe-app-123","status":"healthy"}}`)) + case r.URL.Path == "/": + // Leg-3 serve target. + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte("deploy-probe-ok")) + default: + http.NotFound(w, r) + } + }) +} + +// TestDeployProbe_HappyPath_AllLegsPass — every leg returns the expected +// success shape; expect result=pass for all 3 legs, no audit_log rows, +// 3 latency observations. +func TestDeployProbe_HappyPath_AllLegsPass(t *testing.T) { + srv := httptest.NewServer(happyDeployHandler()) + defer srv.Close() + + db, mock, err := sqlmock.New(sqlmock.QueryMatcherOption(sqlmock.QueryMatcherRegexp)) + if err != nil { + t.Fatalf("sqlmock.New: %v", err) + } + defer db.Close() + + fm := &fakeDeployProbeMetrics{} + w := jobs.NewDeployProbeWorker(db, httpClientRetargetingHTTPSToServer(srv), fm, deployProbeBaseConfig(t, srv)) + if err := w.Work(context.Background(), fakeJob[jobs.DeployProbeArgs]()); err != nil { + t.Fatalf("Work: %v", err) + } + + for _, leg := range []string{"submit", "status", "serve"} { + if got := fm.outcomeFor(leg); got != "pass" { + t.Errorf("leg=%s outcome: want pass, got %q", leg, got) + } + } + if len(fm.latencies) != 3 { + t.Errorf("latency observations: want 3, got %d", len(fm.latencies)) + } + if err := mock.ExpectationsWereMet(); err != nil { + t.Errorf("unexpected DB activity: %v", err) + } +} + +// TestDeployProbe_SubmitReturns503_FailsLeg1 — /deploy/new returns 503; +// expect leg=submit result=fail + audit_log + downstream legs degraded. +func TestDeployProbe_SubmitReturns503_FailsLeg1(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusServiceUnavailable) + _, _ = w.Write([]byte(`{"error":"upstream"}`)) + })) + defer srv.Close() + + db, mock, err := sqlmock.New(sqlmock.QueryMatcherOption(sqlmock.QueryMatcherRegexp)) + if err != nil { + t.Fatalf("sqlmock.New: %v", err) + } + defer db.Close() + mock.ExpectExec(`INSERT INTO audit_log`).WillReturnResult(sqlmock.NewResult(1, 1)) + + fm := &fakeDeployProbeMetrics{} + w := jobs.NewDeployProbeWorker(db, httpClientRetargetingHTTPSToServer(srv), fm, deployProbeBaseConfig(t, srv)) + if err := w.Work(context.Background(), fakeJob[jobs.DeployProbeArgs]()); err != nil { + t.Fatalf("Work: %v", err) + } + + if got := fm.outcomeFor("submit"); got != "fail" { + t.Errorf("submit outcome: want fail, got %q", got) + } + if got := fm.outcomeFor("status"); got != "degraded" { + t.Errorf("status outcome: want degraded (skipped), got %q", got) + } + if got := fm.outcomeFor("serve"); got != "degraded" { + t.Errorf("serve outcome: want degraded (skipped), got %q", got) + } + if err := mock.ExpectationsWereMet(); err != nil { + t.Errorf("audit_log expectation: %v", err) + } +} + +// TestDeployProbe_BuildFailed_FailsLeg2 — submit OK, but the status poll +// reports `failed`; expect leg=status result=fail with the autopsy +// reason in the audit row. +func TestDeployProbe_BuildFailed_FailsLeg2(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch { + case r.URL.Path == "/deploy/new": + w.WriteHeader(http.StatusAccepted) + _, _ = w.Write([]byte(`{"ok":true,"item":{"app_id":"abc12345"}}`)) + case strings.HasPrefix(r.URL.Path, "/deploy/"): + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte(`{"ok":true,"item":{"app_id":"abc12345","status":"failed"}}`)) + } + })) + defer srv.Close() + + db, mock, err := sqlmock.New(sqlmock.QueryMatcherOption(sqlmock.QueryMatcherRegexp)) + if err != nil { + t.Fatalf("sqlmock.New: %v", err) + } + defer db.Close() + mock.ExpectExec(`INSERT INTO audit_log`).WillReturnResult(sqlmock.NewResult(1, 1)) + + fm := &fakeDeployProbeMetrics{} + w := jobs.NewDeployProbeWorker(db, httpClientRetargetingHTTPSToServer(srv), fm, deployProbeBaseConfig(t, srv)) + if err := w.Work(context.Background(), fakeJob[jobs.DeployProbeArgs]()); err != nil { + t.Fatalf("Work: %v", err) + } + + if got := fm.outcomeFor("submit"); got != "pass" { + t.Errorf("submit: want pass, got %q", got) + } + if got := fm.outcomeFor("status"); got != "fail" { + t.Errorf("status: want fail, got %q", got) + } + if got := fm.outcomeFor("serve"); got != "degraded" { + t.Errorf("serve: want degraded (skipped), got %q", got) + } + if err := mock.ExpectationsWereMet(); err != nil { + t.Errorf("audit_log expectation: %v", err) + } +} + +// TestDeployProbe_StuckBuilding_FailsLeg2 — submit OK, status stays +// "building" past the 90s budget. We shorten the budget by using a +// short-budget worker (BUDGET_OVERRIDE via repeated handler 'building' +// + a cancellable context) so the test isn't slow. +// +// The standard /deploy/:id handler returns building forever; the test +// uses a cancellable context to force the leg-status budget to expire +// early by cancelling after one poll cycle. Asserts result=fail with +// reason mentioning "ctx_cancelled" or "status=building at budget". +func TestDeployProbe_StuckBuilding_FailsLeg2(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch { + case r.URL.Path == "/deploy/new": + w.WriteHeader(http.StatusAccepted) + _, _ = w.Write([]byte(`{"ok":true,"item":{"app_id":"abc12345"}}`)) + case strings.HasPrefix(r.URL.Path, "/deploy/"): + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte(`{"ok":true,"item":{"app_id":"abc12345","status":"building"}}`)) + } + })) + defer srv.Close() + + db, _, err := sqlmock.New(sqlmock.QueryMatcherOption(sqlmock.QueryMatcherRegexp)) + if err != nil { + t.Fatalf("sqlmock.New: %v", err) + } + defer db.Close() + // No ExpectExec here: when the parent ctx is cancelled to force the + // stuck-building path, the same ctx is threaded into the audit + // ExecContext call which then also fails with context.Canceled. + // The worker WARN-logs the audit failure (audit_insert_failed) + // instead of crashing — that's the desired posture, and is itself + // the test's contract. + + fm := &fakeDeployProbeMetrics{} + w := jobs.NewDeployProbeWorker(db, httpClientRetargetingHTTPSToServer(srv), fm, deployProbeBaseConfig(t, srv)) + + // Cancel quickly so we hit the ctx_cancelled branch rather than + // waiting the full 90s budget. + ctx, cancel := context.WithTimeout(context.Background(), 200*time.Millisecond) + defer cancel() + if err := w.Work(ctx, fakeJob[jobs.DeployProbeArgs]()); err != nil { + t.Fatalf("Work: %v", err) + } + + if got := fm.outcomeFor("status"); got != "fail" { + t.Errorf("status: want fail, got %q", got) + } +} + +// TestDeployProbe_ServeReturns502_FailsLeg3 — submit + status pass, +// but the public-host fetch returns 502 (Ingress down). +func TestDeployProbe_ServeReturns502_FailsLeg3(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch { + case r.URL.Path == "/deploy/new": + w.WriteHeader(http.StatusAccepted) + _, _ = w.Write([]byte(`{"ok":true,"item":{"app_id":"abc12345"}}`)) + case strings.HasPrefix(r.URL.Path, "/deploy/"): + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte(`{"ok":true,"item":{"app_id":"abc12345","status":"healthy"}}`)) + case r.URL.Path == "/": + w.WriteHeader(http.StatusBadGateway) + } + })) + defer srv.Close() + + db, mock, err := sqlmock.New(sqlmock.QueryMatcherOption(sqlmock.QueryMatcherRegexp)) + if err != nil { + t.Fatalf("sqlmock.New: %v", err) + } + defer db.Close() + mock.ExpectExec(`INSERT INTO audit_log`).WillReturnResult(sqlmock.NewResult(1, 1)) + + fm := &fakeDeployProbeMetrics{} + w := jobs.NewDeployProbeWorker(db, httpClientRetargetingHTTPSToServer(srv), fm, deployProbeBaseConfig(t, srv)) + if err := w.Work(context.Background(), fakeJob[jobs.DeployProbeArgs]()); err != nil { + t.Fatalf("Work: %v", err) + } + + if got := fm.outcomeFor("submit"); got != "pass" { + t.Errorf("submit: want pass, got %q", got) + } + if got := fm.outcomeFor("status"); got != "pass" { + t.Errorf("status: want pass, got %q", got) + } + if got := fm.outcomeFor("serve"); got != "fail" { + t.Errorf("serve: want fail, got %q", got) + } + if err := mock.ExpectationsWereMet(); err != nil { + t.Errorf("audit_log expectation: %v", err) + } +} + +// TestDeployProbe_BearerUnset_AllDegraded — empty bearer = disabled +// prober; expect all three legs to report degraded (config drift, not +// outage) and no audit row. +func TestDeployProbe_BearerUnset_AllDegraded(t *testing.T) { + srv := httptest.NewServer(happyDeployHandler()) + defer srv.Close() + + db, mock, err := sqlmock.New(sqlmock.QueryMatcherOption(sqlmock.QueryMatcherRegexp)) + if err != nil { + t.Fatalf("sqlmock.New: %v", err) + } + defer db.Close() + + cfg := deployProbeBaseConfig(t, srv) + cfg.BearerToken = "" + fm := &fakeDeployProbeMetrics{} + w := jobs.NewDeployProbeWorker(db, httpClientRetargetingHTTPSToServer(srv), fm, cfg) + if err := w.Work(context.Background(), fakeJob[jobs.DeployProbeArgs]()); err != nil { + t.Fatalf("Work: %v", err) + } + + for _, leg := range []string{"submit", "status", "serve"} { + if got := fm.outcomeFor(leg); got != "degraded" { + t.Errorf("leg=%s: want degraded, got %q", leg, got) + } + } + if err := mock.ExpectationsWereMet(); err != nil { + t.Errorf("unexpected DB activity: %v", err) + } +} + +// TestDeployProbe_SubmitOKFalse_FailsLeg1 — 202 + body ok=false; expect +// leg=submit fail on body-assertion. +func TestDeployProbe_SubmitOKFalse_FailsLeg1(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusAccepted) + _, _ = w.Write([]byte(`{"ok":false,"item":{"app_id":"x"}}`)) + })) + defer srv.Close() + db, mock, _ := sqlmock.New(sqlmock.QueryMatcherOption(sqlmock.QueryMatcherRegexp)) + defer db.Close() + mock.ExpectExec(`INSERT INTO audit_log`).WillReturnResult(sqlmock.NewResult(1, 1)) + fm := &fakeDeployProbeMetrics{} + w := jobs.NewDeployProbeWorker(db, httpClientRetargetingHTTPSToServer(srv), fm, deployProbeBaseConfig(t, srv)) + _ = w.Work(context.Background(), fakeJob[jobs.DeployProbeArgs]()) + if got := fm.outcomeFor("submit"); got != "fail" { + t.Errorf("submit: want fail, got %q", got) + } +} + +// TestDeployProbe_SubmitBodyParseErr — 202 + invalid JSON body. +func TestDeployProbe_SubmitBodyParseErr(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusAccepted) + _, _ = w.Write([]byte(`not-json`)) + })) + defer srv.Close() + db, mock, _ := sqlmock.New(sqlmock.QueryMatcherOption(sqlmock.QueryMatcherRegexp)) + defer db.Close() + mock.ExpectExec(`INSERT INTO audit_log`).WillReturnResult(sqlmock.NewResult(1, 1)) + fm := &fakeDeployProbeMetrics{} + w := jobs.NewDeployProbeWorker(db, httpClientRetargetingHTTPSToServer(srv), fm, deployProbeBaseConfig(t, srv)) + _ = w.Work(context.Background(), fakeJob[jobs.DeployProbeArgs]()) + if got := fm.outcomeFor("submit"); got != "fail" { + t.Errorf("submit: want fail, got %q", got) + } +} + +// TestDeployProbe_SubmitMissingAppID — 202 + body missing item.app_id. +func TestDeployProbe_SubmitMissingAppID(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusAccepted) + _, _ = w.Write([]byte(`{"ok":true,"item":{}}`)) + })) + defer srv.Close() + db, mock, _ := sqlmock.New(sqlmock.QueryMatcherOption(sqlmock.QueryMatcherRegexp)) + defer db.Close() + mock.ExpectExec(`INSERT INTO audit_log`).WillReturnResult(sqlmock.NewResult(1, 1)) + fm := &fakeDeployProbeMetrics{} + w := jobs.NewDeployProbeWorker(db, httpClientRetargetingHTTPSToServer(srv), fm, deployProbeBaseConfig(t, srv)) + _ = w.Work(context.Background(), fakeJob[jobs.DeployProbeArgs]()) + if got := fm.outcomeFor("submit"); got != "fail" { + t.Errorf("submit: want fail, got %q", got) + } +} + +// TestDeployProbe_StatusGetReturns500 — submit OK but GET /deploy/ +// returns 500. Expect leg=status fail with poll_error reason. +func TestDeployProbe_StatusGetReturns500(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch { + case r.URL.Path == "/deploy/new": + w.WriteHeader(http.StatusAccepted) + _, _ = w.Write([]byte(`{"ok":true,"item":{"app_id":"abc"}}`)) + case strings.HasPrefix(r.URL.Path, "/deploy/"): + w.WriteHeader(http.StatusInternalServerError) + } + })) + defer srv.Close() + db, mock, _ := sqlmock.New(sqlmock.QueryMatcherOption(sqlmock.QueryMatcherRegexp)) + defer db.Close() + mock.ExpectExec(`INSERT INTO audit_log`).WillReturnResult(sqlmock.NewResult(1, 1)) + fm := &fakeDeployProbeMetrics{} + w := jobs.NewDeployProbeWorker(db, httpClientRetargetingHTTPSToServer(srv), fm, deployProbeBaseConfig(t, srv)) + _ = w.Work(context.Background(), fakeJob[jobs.DeployProbeArgs]()) + if got := fm.outcomeFor("status"); got != "fail" { + t.Errorf("status: want fail, got %q", got) + } +} + +// TestDeployProbe_StatusBodyParseErr — submit OK, status GET returns +// 200 with invalid JSON. +func TestDeployProbe_StatusBodyParseErr(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch { + case r.URL.Path == "/deploy/new": + w.WriteHeader(http.StatusAccepted) + _, _ = w.Write([]byte(`{"ok":true,"item":{"app_id":"abc"}}`)) + case strings.HasPrefix(r.URL.Path, "/deploy/"): + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte(`not-json`)) + } + })) + defer srv.Close() + db, mock, _ := sqlmock.New(sqlmock.QueryMatcherOption(sqlmock.QueryMatcherRegexp)) + defer db.Close() + mock.ExpectExec(`INSERT INTO audit_log`).WillReturnResult(sqlmock.NewResult(1, 1)) + fm := &fakeDeployProbeMetrics{} + w := jobs.NewDeployProbeWorker(db, httpClientRetargetingHTTPSToServer(srv), fm, deployProbeBaseConfig(t, srv)) + _ = w.Work(context.Background(), fakeJob[jobs.DeployProbeArgs]()) + if got := fm.outcomeFor("status"); got != "fail" { + t.Errorf("status: want fail, got %q", got) + } +} + +// TestDeployProbe_StatusMissingField — submit OK, status body OK shape +// but status string missing. +func TestDeployProbe_StatusMissingField(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch { + case r.URL.Path == "/deploy/new": + w.WriteHeader(http.StatusAccepted) + _, _ = w.Write([]byte(`{"ok":true,"item":{"app_id":"abc"}}`)) + case strings.HasPrefix(r.URL.Path, "/deploy/"): + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte(`{"ok":true,"item":{}}`)) + } + })) + defer srv.Close() + db, mock, _ := sqlmock.New(sqlmock.QueryMatcherOption(sqlmock.QueryMatcherRegexp)) + defer db.Close() + mock.ExpectExec(`INSERT INTO audit_log`).WillReturnResult(sqlmock.NewResult(1, 1)) + fm := &fakeDeployProbeMetrics{} + w := jobs.NewDeployProbeWorker(db, httpClientRetargetingHTTPSToServer(srv), fm, deployProbeBaseConfig(t, srv)) + _ = w.Work(context.Background(), fakeJob[jobs.DeployProbeArgs]()) + if got := fm.outcomeFor("status"); got != "fail" { + t.Errorf("status: want fail, got %q", got) + } +} + +// TestDeployProbe_NilDB_DoesNotCrash — db nil means the audit_log +// insert is skipped but all legs still emit metric. +func TestDeployProbe_NilDB_DoesNotCrash(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusServiceUnavailable) + })) + defer srv.Close() + fm := &fakeDeployProbeMetrics{} + var nilDB *sql.DB + w := jobs.NewDeployProbeWorker(nilDB, httpClientRetargetingHTTPSToServer(srv), fm, deployProbeBaseConfig(t, srv)) + if err := w.Work(context.Background(), fakeJob[jobs.DeployProbeArgs]()); err != nil { + t.Fatalf("Work: %v", err) + } + if got := fm.outcomeFor("submit"); got != "fail" { + t.Errorf("submit: want fail, got %q", got) + } +} + +// TestDeployProbe_NilMetrics_NoCrash — metrics=nil must not panic. +func TestDeployProbe_NilMetrics_NoCrash(t *testing.T) { + srv := httptest.NewServer(happyDeployHandler()) + defer srv.Close() + w := jobs.NewDeployProbeWorker(nil, httpClientRetargetingHTTPSToServer(srv), nil, deployProbeBaseConfig(t, srv)) + if err := w.Work(context.Background(), fakeJob[jobs.DeployProbeArgs]()); err != nil { + t.Fatalf("Work: %v", err) + } +} + +// TestDeployProbe_NilHTTPClient_GetsDefault — httpCli=nil should install +// the default with redirect-refusal. Hit the constructor and run Work +// against an unresolvable URL so the leg fails cleanly. +func TestDeployProbe_NilHTTPClient_GetsDefault(t *testing.T) { + cfg := jobs.DeployProbeConfig{ + BaseURL: "http://this-host-does-not-resolve.invalid", + BearerToken: "x", + } + fm := &fakeDeployProbeMetrics{} + w := jobs.NewDeployProbeWorker(nil, nil, fm, cfg) + if err := w.Work(context.Background(), fakeJob[jobs.DeployProbeArgs]()); err != nil { + t.Fatalf("Work: %v", err) + } + if got := fm.outcomeFor("submit"); got != "fail" { + t.Errorf("submit: want fail (DNS), got %q", got) + } +} + +// TestDeployProbe_DNSFailure_NoLatency — DNS error should NOT emit a +// latency observation for submit. +func TestDeployProbe_DNSFailure_NoLatency(t *testing.T) { + httpCli := &http.Client{Timeout: 1 * time.Second} + fm := &fakeDeployProbeMetrics{} + cfg := jobs.DeployProbeConfig{ + BaseURL: "http://this-host-does-not-resolve.invalid", + BearerToken: "x", + } + w := jobs.NewDeployProbeWorker(nil, httpCli, fm, cfg) + if err := w.Work(context.Background(), fakeJob[jobs.DeployProbeArgs]()); err != nil { + t.Fatalf("Work: %v", err) + } + for _, l := range fm.latencies { + if l.leg == "submit" { + t.Errorf("submit emitted latency on DNS failure: %v", l.d) + } + } +} + +// TestDeployProbe_BadBaseURL_HitsBuildRequestErr — control char in URL +// trips http.NewRequestWithContext, exercising the build_request branch +// on leg-1 (other legs are skipped via the cascade). +func TestDeployProbe_BadBaseURL_HitsBuildRequestErr(t *testing.T) { + cfg := jobs.DeployProbeConfig{ + BaseURL: "http://example.com/\x7f", + BearerToken: "x", + } + fm := &fakeDeployProbeMetrics{} + w := jobs.NewDeployProbeWorker(nil, &http.Client{Timeout: time.Second}, fm, cfg) + _ = w.Work(context.Background(), fakeJob[jobs.DeployProbeArgs]()) + if got := fm.outcomeFor("submit"); got != "fail" { + t.Errorf("submit: want fail (build_request), got %q", got) + } +} + +// TestValidateDeployProbeBaseURL — startup-time URL validation. +func TestValidateDeployProbeBaseURL(t *testing.T) { + cases := []struct { + name string + in string + wantErr bool + }{ + {"empty ok (uses default)", "", false}, + {"https ok", "https://api.instanode.dev", false}, + {"http ok (dev)", "http://localhost:8080", false}, + {"ftp rejected", "ftp://example.com", true}, + {"missing host", "https://", true}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + err := jobs.ValidateDeployProbeBaseURL(tc.in) + if tc.wantErr != (err != nil) { + t.Errorf("ValidateDeployProbeBaseURL(%q) err = %v, wantErr = %v", tc.in, err, tc.wantErr) + } + }) + } +} + +// TestValidateDeployProbeBaseURL_ParseError — non-URL string returns +// err (parse path distinct from the scheme/host gate above). +func TestValidateDeployProbeBaseURL_ParseError(t *testing.T) { + if err := jobs.ValidateDeployProbeBaseURL("://"); err == nil { + t.Errorf("want err for malformed URL, got nil") + } +} + +// TestDeployProbeConfig_Defaults — empty fields get filled with +// production defaults; non-empty fields preserved; trailing slash on +// BaseURL is trimmed. +func TestDeployProbeConfig_Defaults(t *testing.T) { + in := jobs.DeployProbeConfig{BaseURL: "https://example.com/", DeployHost: ".deploy.example.com/"} + out := in.Defaults() + if out.BaseURL != "https://example.com" { + t.Errorf("BaseURL trailing slash not trimmed: %q", out.BaseURL) + } + if out.DeployHost != "deploy.example.com" { + t.Errorf("DeployHost leading dot / trailing slash not trimmed: %q", out.DeployHost) + } + if out.AppName == "" || out.Env == "" { + t.Errorf("defaults not applied: %+v", out) + } +} + +// TestDeployProbeConfig_Defaults_AllEmpty — zero-value config gets every +// default filled in. +func TestDeployProbeConfig_Defaults_AllEmpty(t *testing.T) { + out := jobs.DeployProbeConfig{}.Defaults() + if out.BaseURL == "" || out.DeployHost == "" || out.AppName == "" || out.Env == "" { + t.Errorf("Defaults() on empty config left a field empty: %+v", out) + } +} + +// TestDeployProbeArgs_Kind — exercise the trivial Kind() method so its +// line is counted as covered. +func TestDeployProbeArgs_Kind(t *testing.T) { + if got := (jobs.DeployProbeArgs{}).Kind(); got != "deploy_probe" { + t.Errorf("Kind() = %q, want deploy_probe", got) + } +} + +// TestDeployProbe_PromMetricsAdapter — exercise the production adapter +// so the methods are covered. +func TestDeployProbe_PromMetricsAdapter(t *testing.T) { + m := jobs.DeployProbePromMetrics{} + m.IncOutcome("submit", "pass") + m.ObserveLatency("submit", 12*time.Millisecond) +} + +// TestDeployProbe_AuditInsertFails_NoCrash — INSERT returns an error; +// the worker WARN-logs and continues rather than panicking. +func TestDeployProbe_AuditInsertFails_NoCrash(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusServiceUnavailable) + })) + defer srv.Close() + db, mock, _ := sqlmock.New(sqlmock.QueryMatcherOption(sqlmock.QueryMatcherRegexp)) + defer db.Close() + mock.ExpectExec(`INSERT INTO audit_log`).WillReturnError(http.ErrAbortHandler) + fm := &fakeDeployProbeMetrics{} + w := jobs.NewDeployProbeWorker(db, httpClientRetargetingHTTPSToServer(srv), fm, deployProbeBaseConfig(t, srv)) + if err := w.Work(context.Background(), fakeJob[jobs.DeployProbeArgs]()); err != nil { + t.Fatalf("Work returned error on audit insert failure: %v", err) + } + if got := fm.outcomeFor("submit"); got != "fail" { + t.Errorf("submit: want fail, got %q", got) + } +} + +// TestDeployProbe_ServeBuildRequestErr — bad deploy host (control char) +// trips http.NewRequestWithContext on leg-3. Need submit + status to +// pass so leg-3 actually runs. We pass a custom config with a malformed +// DeployHost. +func TestDeployProbe_ServeBuildRequestErr(t *testing.T) { + srv := httptest.NewServer(happyDeployHandler()) + defer srv.Close() + cfg := deployProbeBaseConfig(t, srv) + cfg.DeployHost = "bad\x7fhost" + fm := &fakeDeployProbeMetrics{} + w := jobs.NewDeployProbeWorker(nil, httpClientRetargetingHTTPSToServer(srv), fm, cfg) + _ = w.Work(context.Background(), fakeJob[jobs.DeployProbeArgs]()) + if got := fm.outcomeFor("serve"); got != "fail" { + t.Errorf("serve: want fail (build_request), got %q", got) + } +} + +// guardCompileTime ensures the fakeDeployProbeMetrics conforms to the +// DeployProbeMetrics interface. +var _ jobs.DeployProbeMetrics = (*fakeDeployProbeMetrics)(nil) diff --git a/internal/jobs/workers.go b/internal/jobs/workers.go index f5c3142..3344c63 100644 --- a/internal/jobs/workers.go +++ b/internal/jobs/workers.go @@ -732,6 +732,24 @@ func StartWorkers(ctx context.Context, db *sql.DB, rdb *redis.Client, cfg *confi }), nrApp, )) + // Hourly synthetic deploy prober. Drives a full end-to-end deploy + // (/deploy/new → Kaniko build → k8s pod → public-host GET) against + // prod every 60 minutes so the next regression in the deploy + // pipeline pages within the 30-minute alert window. Closes the gap + // that hid the 2026-05-30 morning truehomie-api stuck-build + // incident for ~30 min until the user reported it. Bearer empty + // keeps the prober configured-off (every leg returns degraded — + // no fail alerts) so the rollout of DEPLOY_PROBE_BEARER_TOKEN can + // land without paging on the secret-unset state. See + // deploy_probe.go for the per-leg fail-mode rationale. + river.AddWorker(workers, WithObservability( + NewDeployProbeWorker(db, nil, DeployProbePromMetrics{}, DeployProbeConfig{ + BaseURL: cfg.DeployProbeBaseURL, + DeployHost: cfg.DeployProbeDeployHost, + BearerToken: cfg.DeployProbeBearerToken, + }), + nrApp, + )) // Razorpay webhook-events prune — daily DELETE of razorpay_webhook_events // rows > 30d. The api appends one dedup row per Razorpay webhook delivery; // migration 033 envisioned a periodic prune but never shipped one, so the @@ -1304,6 +1322,22 @@ func buildPeriodicJobs(cfg *config.Config) []*river.PeriodicJob { }, &river.PeriodicJobOpts{RunOnStart: true}, ), + // Hourly synthetic deploy prober — every 60 minutes drives + // /deploy/new + status-poll + serve-fetch against prod. + // RunOnStart=false because (a) the leg-1 submit is heavyweight + // (Kaniko build + k8s rollout — ~30s of cluster work per tick) + // and (b) a worker restart inside the hour doesn't add useful + // signal beyond the previous tick's metric. The 30-minute alert + // window comfortably tolerates the post-restart gap. Routed to + // the reconcile queue so a default-queue weekly_digest fan-out + // can't starve the deploy probe. + river.NewPeriodicJob( + river.PeriodicInterval(deployProbeInterval), + func() (river.JobArgs, *river.InsertOpts) { + return DeployProbeArgs{}, reconcileInsertOpts(deployProbeInterval) + }, + &river.PeriodicJobOpts{RunOnStart: false}, + ), // Razorpay webhook-events prune — daily DELETE of dedup rows > 30d. // RunOnStart=false: a restart shouldn't immediately scan; the table // grows slowly (one row per webhook delivery) so a day's delay before diff --git a/internal/metrics/metrics.go b/internal/metrics/metrics.go index d524a5a..e87061b 100644 --- a/internal/metrics/metrics.go +++ b/internal/metrics/metrics.go @@ -744,6 +744,30 @@ var ( Help: "AUTH-004 synthetic prober per-leg latency. Buckets centred on the per-leg latency budgets (50ms…5s).", Buckets: []float64{0.05, 0.1, 0.25, 0.5, 1, 2, 5}, }, []string{"leg"}) + + // DeployProbeOutcomeTotal — hourly synthetic deploy prober counters. + // Labelled by `leg` (submit | status | serve) and `result` (pass | + // fail | degraded). result="fail" is the alert-able signal — the + // 2026-05-30 morning truehomie-api stuck-build incident hid the + // /deploy/new pipeline being broken for ~30 minutes until the user + // reported it. NR alert: any fail in 30m → P0 (deploy-probe-fail.json). + // Prom rule: DeployProbeFail in prometheus-rules.yaml. + // Emit site: worker/internal/jobs/deploy_probe.go (DeployProbePromMetrics). + DeployProbeOutcomeTotal = promauto.NewCounterVec(prometheus.CounterOpts{ + Name: "instant_deploy_probe_outcome_total", + Help: "Hourly synthetic deploy prober outcomes per leg (submit|status|serve) and result (pass|fail|degraded).", + }, []string{"leg", "result"}) + + // DeployProbeLatencySeconds — per-leg HTTP/poll latency histogram. + // Only observed on a real response (DNS / TCP errors omit the + // observation). Buckets span the per-leg budgets (submit 30s, + // status poll up to 90s wall-clock, serve 30s) — the wider 120s + // upper bucket captures the cold-cluster Kaniko build edge case. + DeployProbeLatencySeconds = promauto.NewHistogramVec(prometheus.HistogramOpts{ + Name: "instant_deploy_probe_latency_seconds", + Help: "Hourly deploy prober per-leg wall-clock latency. Buckets cover the per-leg budgets up to the 120s cold-cluster Kaniko ceiling.", + Buckets: []float64{0.5, 1, 5, 10, 30, 60, 90, 120}, + }, []string{"leg"}) ) // ReadyzCheckStatus updates the gauge for one check on this service. From 188460c3f6f9f5da8c492b9b6bbb8c24ca81b8f6 Mon Sep 17 00:00:00 2001 From: Manas Srivastava Date: Sat, 30 May 2026 21:56:14 +0530 Subject: [PATCH 2/3] test(deploy_probe): drop unused deployProbeTestClient helper (lint) --- internal/jobs/deploy_probe_test.go | 20 +++----------------- 1 file changed, 3 insertions(+), 17 deletions(-) diff --git a/internal/jobs/deploy_probe_test.go b/internal/jobs/deploy_probe_test.go index 9bf285f..54bd0a7 100644 --- a/internal/jobs/deploy_probe_test.go +++ b/internal/jobs/deploy_probe_test.go @@ -94,23 +94,9 @@ func srvHost(t *testing.T, srv *httptest.Server) string { return u } -// deployProbeTestClient is an http client that ignores TLS (so the -// leg-3 https://./ call lands on the http httptest server) -// and refuses redirects (matching the production default). -func deployProbeTestClient() *http.Client { - return &http.Client{ - Timeout: 10 * time.Second, - Transport: &http.Transport{ - // Tests run against an http httptest server but the leg-3 URL - // uses https://; rewrite the scheme in DialContext via a custom - // RoundTripper so the connection still lands on the test server. - // Simpler: re-target the request URL to http on egress. - }, - CheckRedirect: func(*http.Request, []*http.Request) error { - return http.ErrUseLastResponse - }, - } -} +// (deployProbeTestClient — removed; superseded by httpClientRetargetingHTTPSToServer +// below, which actually rewrites the https URL onto the test server. Kept this +// comment as a breadcrumb for the next reader.) // httpClientRetargetingHTTPSToServer returns an http.Client whose // RoundTripper rewrites every https:// request to the test From 7d587b9204bd6f84b2df9d39c2079e9180114545 Mon Sep 17 00:00:00 2001 From: Manas Srivastava Date: Sat, 30 May 2026 22:23:24 +0530 Subject: [PATCH 3/3] =?UTF-8?q?test(deploy=5Fprobe):=20white-box=20+=20inj?= =?UTF-8?q?ectable=20budgets=20=E2=86=92=20100%=20patch=20coverage?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two follow-up cleanups so the CI patch-coverage gate at 100% passes: - Make `*Budget` / `pollInterval` injectable on DeployProbeWorker (zero = use the package-level constants — production behaviour unchanged). The four unexported fields let internal tests drive the degraded-latency + after-deadline + ctx-cancelled-during-poll branches inside a second of wall-clock instead of needing the real 30s / 90s budgets. - Decouple serveBudget from the request's ctx timeout (matches the auth_probe.legEmailStart pattern). Today the leg-3 request was cancelled by servCtx the moment its budget elapsed, so the degraded-latency branch on `latency > serveBudget` was provably unreachable in production (a slow-but-working serve always landed in http_error). Now the request uses the http client's hard timeout (deployProbeHTTPTimeout = 120s) and the budget is a pure post-hoc latency assertion → degraded fires correctly on a slow serve that completed before the hard ceiling. - Drop defensive `error` returns from `buildDeployProbeMultipart` + `buildDeployProbeNginxTarball` — both write to in-memory bytes.Buffer writers which physically cannot fail. Same posture as auth_probe.legEmailStart's `_ = json.Marshal(...)`. Removes the unreachable error branches that were the modal patch-coverage gap. - Add `deploy_probe_internal_test.go` (jobs package) with white-box tests covering: tarball roundtrip, multipart shape, recordLeg's three branches, legStatus ctx-cancelled-at-top-of-loop + budget-elapsed + poll-interval-sleep-completes, legSubmit degraded-latency, legServe degraded-latency + http_error + build_request, fetchDeployStatus build_request + http_error, effective-budget zero-fallback + override, and the CheckRedirect closure on the default client. Coverage (locally measured against origin/master): internal/jobs/deploy_probe.go (100%) internal/jobs/workers.go (100%) Total: 437 lines, 0 missing — gate passes. No production behaviour change beyond the serveBudget decoupling documented above. `make gate` green. Co-Authored-By: Claude Opus 4.7 (1M context) --- internal/jobs/deploy_probe.go | 156 +++++--- internal/jobs/deploy_probe_internal_test.go | 397 ++++++++++++++++++++ 2 files changed, 494 insertions(+), 59 deletions(-) create mode 100644 internal/jobs/deploy_probe_internal_test.go diff --git a/internal/jobs/deploy_probe.go b/internal/jobs/deploy_probe.go index c71e17c..44f0377 100644 --- a/internal/jobs/deploy_probe.go +++ b/internal/jobs/deploy_probe.go @@ -255,12 +255,59 @@ func (c DeployProbeConfig) Defaults() DeployProbeConfig { // insertions on fail outcomes (nil disables the audit row but the leg // still runs + metric still emits — fail-open). httpCli is used for all // HTTP probes; nil installs a default with the global timeout. +// +// The four `*Budget` / `pollInterval` fields are test-only injectable +// knobs — zero means use the package-level deployProbe*Budget / +// deployProbePollInterval constants. Production wiring (NewDeployProbeWorker +// → StartWorkers) leaves them zero so the prod cadence is governed by +// the constants. Unit tests inject short values so the degraded-latency +// + after-deadline branches are reachable inside a second of wall-clock. type DeployProbeWorker struct { river.WorkerDefaults[DeployProbeArgs] db *sql.DB httpCli *http.Client metrics DeployProbeMetrics cfg DeployProbeConfig + + submitBudget time.Duration + statusBudget time.Duration + serveBudget time.Duration + pollInterval time.Duration +} + +// effectiveSubmitBudget returns the per-worker submit budget, falling +// back to the package-level constant when unset. Read once at function +// entry in legSubmit so the value the test sees on a metric label +// matches the value the timer enforces. +func (w *DeployProbeWorker) effectiveSubmitBudget() time.Duration { + if w.submitBudget > 0 { + return w.submitBudget + } + return deployProbeSubmitBudget +} + +// effectiveStatusBudget — see effectiveSubmitBudget. +func (w *DeployProbeWorker) effectiveStatusBudget() time.Duration { + if w.statusBudget > 0 { + return w.statusBudget + } + return deployProbeStatusBudget +} + +// effectiveServeBudget — see effectiveSubmitBudget. +func (w *DeployProbeWorker) effectiveServeBudget() time.Duration { + if w.serveBudget > 0 { + return w.serveBudget + } + return deployProbeServeBudget +} + +// effectivePollInterval — see effectiveSubmitBudget. +func (w *DeployProbeWorker) effectivePollInterval() time.Duration { + if w.pollInterval > 0 { + return w.pollInterval + } + return deployProbePollInterval } // NewDeployProbeWorker constructs the worker. metrics is required — pass @@ -445,16 +492,9 @@ func (w *DeployProbeWorker) emitDeployProbeFailed(ctx context.Context, leg strin // result plus the app_id pulled from the response envelope. The app_id // is the key the next two legs depend on. func (w *DeployProbeWorker) legSubmit(ctx context.Context) (deployProbeLegResult, string) { - body, contentType, err := buildDeployProbeMultipart(w.cfg.AppName, w.cfg.Env) - if err != nil { - // buildDeployProbeMultipart only fails on impossible bytes.Buffer - // errors; defensive branch still returns a real failure so the - // alert fires rather than masking a real regression. - return deployProbeLegResult{ - result: deployProbeResultFail, - reason: "build_multipart: " + err.Error(), - }, "" - } + // buildDeployProbeMultipart writes to an in-memory bytes.Buffer and + // cannot fail — see the helper's docstring. No err to check here. + body, contentType := buildDeployProbeMultipart(w.cfg.AppName, w.cfg.Env) target := w.cfg.BaseURL + "/deploy/new" req, err := http.NewRequestWithContext(ctx, http.MethodPost, target, body) @@ -516,9 +556,10 @@ func (w *DeployProbeWorker) legSubmit(ctx context.Context) (deployProbeLegResult r.reason = "body missing item.app_id; raw=" + truncateForLog(string(respBody), 256) return r, "" } - if latency > deployProbeSubmitBudget { + submitBudget := w.effectiveSubmitBudget() + if latency > submitBudget { r.result = deployProbeResultDegraded - r.reason = fmt.Sprintf("latency=%dms over budget=%dms", latency.Milliseconds(), deployProbeSubmitBudget.Milliseconds()) + r.reason = fmt.Sprintf("latency=%dms over budget=%dms", latency.Milliseconds(), submitBudget.Milliseconds()) return r, parsed.Item.AppID } r.result = deployProbeResultPass @@ -532,7 +573,9 @@ func (w *DeployProbeWorker) legSubmit(ctx context.Context) (deployProbeLegResult // log. func (w *DeployProbeWorker) legStatus(ctx context.Context, appID string) deployProbeLegResult { target := w.cfg.BaseURL + "/deploy/" + url.PathEscape(appID) - deadline := time.Now().Add(deployProbeStatusBudget) + statusBudget := w.effectiveStatusBudget() + pollInterval := w.effectivePollInterval() + deadline := time.Now().Add(statusBudget) start := time.Now() for { @@ -581,7 +624,7 @@ func (w *DeployProbeWorker) legStatus(ctx context.Context, appID string) deployP if time.Now().After(deadline) { return deployProbeLegResult{ result: deployProbeResultFail, - reason: fmt.Sprintf("status=%q at budget (want healthy within %s)", status, deployProbeStatusBudget), + reason: fmt.Sprintf("status=%q at budget (want healthy within %s)", status, statusBudget), latency: time.Since(start), observeLatency: true, httpStatus: httpStatus, @@ -594,7 +637,7 @@ func (w *DeployProbeWorker) legStatus(ctx context.Context, appID string) deployP reason: "ctx_cancelled_during_poll: " + ctx.Err().Error(), latency: time.Since(start), } - case <-time.After(deployProbePollInterval): + case <-time.After(pollInterval): } } } @@ -643,10 +686,14 @@ func (w *DeployProbeWorker) fetchDeployStatus(ctx context.Context, target string func (w *DeployProbeWorker) legServe(ctx context.Context, appID string) deployProbeLegResult { target := "https://" + appID + "." + w.cfg.DeployHost + "/" - servCtx, cancel := context.WithTimeout(ctx, deployProbeServeBudget) - defer cancel() - - req, err := http.NewRequestWithContext(servCtx, http.MethodGet, target, nil) + serveBudget := w.effectiveServeBudget() + // Use the HTTP client's own timeout (deployProbeHTTPTimeout = 120s) + // as the hard ceiling on the request; serveBudget is a SOFT post- + // hoc latency assertion (slow-but-working serves report degraded + // instead of pinning the goroutine). Decoupling the two lets the + // degraded branch fire on a serve that crosses 30s but completes + // before the 120s hard ceiling. + req, err := http.NewRequestWithContext(ctx, http.MethodGet, target, nil) if err != nil { return deployProbeLegResult{ result: deployProbeResultFail, @@ -678,9 +725,9 @@ func (w *DeployProbeWorker) legServe(ctx context.Context, appID string) deployPr r.reason = fmt.Sprintf("serve_status=%d (want 200) on %s", resp.StatusCode, target) return r } - if latency > deployProbeServeBudget { + if latency > serveBudget { r.result = deployProbeResultDegraded - r.reason = fmt.Sprintf("latency=%dms over budget=%dms", latency.Milliseconds(), deployProbeServeBudget.Milliseconds()) + r.reason = fmt.Sprintf("latency=%dms over budget=%dms", latency.Milliseconds(), serveBudget.Milliseconds()) return r } r.result = deployProbeResultPass @@ -692,28 +739,27 @@ func (w *DeployProbeWorker) legServe(ctx context.Context, appID string) deployPr // `redeploy=true` is what makes the probe-app row reusable across // ticks. Extracted so tests can re-use the exact same shape the prober // puts on the wire. -func buildDeployProbeMultipart(name, env string) (*bytes.Buffer, string, error) { +// +// Returns the buffer + Content-Type. No error path: every underlying +// operation writes to an in-memory bytes.Buffer (CreateFormFile, +// WriteField, part.Write, mw.Close) which cannot fail — same pattern +// as auth_probe.legEmailStart's `_ = json.Marshal(...)`. Removing the +// defensive branches keeps the patch-coverage gate at 100%. +func buildDeployProbeMultipart(name, env string) (*bytes.Buffer, string) { var buf bytes.Buffer mw := multipart.NewWriter(&buf) - tarball, err := buildDeployProbeNginxTarball() - if err != nil { - return nil, "", fmt.Errorf("nginx_tarball: %w", err) - } + tarball := buildDeployProbeNginxTarball() // tarball — name `app.tar.gz` is cosmetic; the api reads `tarballs[0]` - // regardless of filename. The mime type tells nothing the api uses. - part, err := mw.CreateFormFile("tarball", "app.tar.gz") - if err != nil { - return nil, "", fmt.Errorf("create_tarball_part: %w", err) - } - if _, err := part.Write(tarball); err != nil { - return nil, "", fmt.Errorf("write_tarball_part: %w", err) - } - - // Required + optional scalar fields. Loop rather than three - // duplicated WriteField calls — keeps the field ordering matrix - // visible at a glance. + // regardless of filename. CreateFormFile against a bytes.Buffer never + // errors (the boundary write is the only failable step and the + // underlying writer can't fail). + part, _ := mw.CreateFormFile("tarball", "app.tar.gz") + _, _ = part.Write(tarball) + + // Required + optional scalar fields. Loop rather than four duplicated + // WriteField calls — keeps the field ordering matrix visible at a glance. fields := [][2]string{ {"name", name}, {"port", "80"}, @@ -721,14 +767,10 @@ func buildDeployProbeMultipart(name, env string) (*bytes.Buffer, string, error) {"redeploy", "true"}, } for _, f := range fields { - if err := mw.WriteField(f[0], f[1]); err != nil { - return nil, "", fmt.Errorf("write_field_%s: %w", f[0], err) - } + _ = mw.WriteField(f[0], f[1]) } - if err := mw.Close(); err != nil { - return nil, "", fmt.Errorf("close_multipart: %w", err) - } - return &buf, mw.FormDataContentType(), nil + _ = mw.Close() + return &buf, mw.FormDataContentType() } // buildDeployProbeNginxTarball synthesises a minimal gzipped-tar archive @@ -736,7 +778,11 @@ func buildDeployProbeMultipart(name, env string) (*bytes.Buffer, string, error) // trivial root-path response. Kaniko reads the tarball directly so no // disk fixture is needed at deploy time — the probe carries its own // build context. -func buildDeployProbeNginxTarball() ([]byte, error) { +// +// No error path: tar.WriteHeader / tar.Write / tar.Close / gzip.Close +// against an in-memory bytes.Buffer cannot fail. Same defensive-branch- +// removal posture as buildDeployProbeMultipart above. +func buildDeployProbeNginxTarball() []byte { dockerfile := []byte("FROM nginx:alpine\nRUN echo 'deploy-probe-ok' > /usr/share/nginx/html/index.html\nEXPOSE 80\n") var gz bytes.Buffer @@ -749,19 +795,11 @@ func buildDeployProbeNginxTarball() ([]byte, error) { Size: int64(len(dockerfile)), ModTime: time.Unix(0, 0).UTC(), // deterministic — same bytes on every tick } - if err := tw.WriteHeader(hdr); err != nil { - return nil, err - } - if _, err := tw.Write(dockerfile); err != nil { - return nil, err - } - if err := tw.Close(); err != nil { - return nil, err - } - if err := gw.Close(); err != nil { - return nil, err - } - return gz.Bytes(), nil + _ = tw.WriteHeader(hdr) + _, _ = tw.Write(dockerfile) + _ = tw.Close() + _ = gw.Close() + return gz.Bytes() } // ValidateDeployProbeBaseURL is a startup-time sanity check for the diff --git a/internal/jobs/deploy_probe_internal_test.go b/internal/jobs/deploy_probe_internal_test.go new file mode 100644 index 0000000..1c7f10b --- /dev/null +++ b/internal/jobs/deploy_probe_internal_test.go @@ -0,0 +1,397 @@ +package jobs + +// deploy_probe_internal_test.go — white-box tests for unexported +// helpers in deploy_probe.go that the black-box test package can't +// reach. Kept in a separate file so the rest of the test suite stays +// in jobs_test. Mirrors auth_probe_internal_test.go in shape. +// +// The injected `*Budget` / `pollInterval` knobs on DeployProbeWorker +// are unexported, so the only way to drive the degraded-latency + +// after-deadline branches in a unit test is from inside the jobs +// package — this file owns those tests. + +import ( + "archive/tar" + "bytes" + "compress/gzip" + "context" + "io" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" +) + +// TestBuildDeployProbeNginxTarball_RoundTrips — buildDeployProbeNginxTarball +// success path; assert the returned bytes are a valid gzipped tar with a +// single Dockerfile entry carrying the FROM + EXPOSE lines. Catches an +// accidental drop of either Dockerfile line (would make every prod tick +// fail Kaniko with "no FROM"). +func TestBuildDeployProbeNginxTarball_RoundTrips(t *testing.T) { + out := buildDeployProbeNginxTarball() + gr, err := gzip.NewReader(bytes.NewReader(out)) + if err != nil { + t.Fatalf("gzip.NewReader: %v", err) + } + tr := tar.NewReader(gr) + hdr, err := tr.Next() + if err != nil { + t.Fatalf("tar.Next: %v", err) + } + if hdr.Name != "Dockerfile" { + t.Errorf("first entry name: got %q, want Dockerfile", hdr.Name) + } + contents, err := io.ReadAll(tr) + if err != nil { + t.Fatalf("io.ReadAll: %v", err) + } + if !strings.Contains(string(contents), "FROM nginx:alpine") { + t.Errorf("Dockerfile missing FROM: %q", string(contents)) + } + if !strings.Contains(string(contents), "EXPOSE 80") { + t.Errorf("Dockerfile missing EXPOSE: %q", string(contents)) + } +} + +// TestBuildDeployProbeMultipart_Shape — buildDeployProbeMultipart's +// success path. Asserts the multipart body carries `tarball`, `name`, +// `port=80`, `env`, and `redeploy=true` — the exact field set the api's +// /deploy/new handler reads. A missing field here would silently flip +// the prober to "fresh deploy" (no redeploy) and burn one slot per tick. +func TestBuildDeployProbeMultipart_Shape(t *testing.T) { + body, contentType := buildDeployProbeMultipart("probe-name", "development") + if !strings.HasPrefix(contentType, "multipart/form-data; boundary=") { + t.Errorf("contentType: %q", contentType) + } + raw := body.String() + for _, want := range []string{ + `name="tarball"`, + `filename="app.tar.gz"`, + `name="name"`, "probe-name", + `name="port"`, "80", + `name="env"`, "development", + `name="redeploy"`, "true", + } { + if !strings.Contains(raw, want) { + t.Errorf("multipart body missing %q", want) + } + } +} + +// TestRecordLeg_AllBranches — recordLeg's three branches (pass / +// degraded / fail) emit different slog levels. The fail branch on a +// nil-DB worker exercises the audit-skip path. We can't observe the +// log lines from a unit test without reaching into slog's handler, +// so we just assert the function doesn't panic and the metric is bumped. +func TestRecordLeg_AllBranches(t *testing.T) { + fm := &capturingDeployMetrics{} + w := &DeployProbeWorker{ + cfg: DeployProbeConfig{BaseURL: "http://x", DeployHost: "y"}.Defaults(), + metrics: fm, + } + ctx := context.Background() + + w.recordLeg(ctx, deployProbeLegSubmit, deployProbeLegResult{ + result: deployProbeResultPass, latency: time.Millisecond, observeLatency: true, + }) + w.recordLeg(ctx, deployProbeLegStatus, deployProbeLegResult{ + result: deployProbeResultDegraded, reason: "slow", latency: time.Second, + }) + // Fail branch with nil DB — writes ERROR log line, skips audit insert + // without crashing. + w.recordLeg(ctx, deployProbeLegServe, deployProbeLegResult{ + result: deployProbeResultFail, reason: "boom", httpStatus: 502, + }) + if len(fm.outcomes) != 3 { + t.Errorf("want 3 metric emits, got %d", len(fm.outcomes)) + } +} + +// capturingDeployMetrics is a local fake for the internal-test package +// (the black-box _test file's fakeDeployProbeMetrics lives in jobs_test +// and isn't reachable here). +type capturingDeployMetrics struct { + outcomes []string +} + +func (c *capturingDeployMetrics) IncOutcome(leg, result string) { + c.outcomes = append(c.outcomes, leg+"="+result) +} +func (c *capturingDeployMetrics) ObserveLatency(string, time.Duration) {} + +// TestLegStatus_CtxCancelledAtTopOfLoop — drives the ctx.Err() check +// at the very top of legStatus's for-loop. With a pre-cancelled ctx +// the first iteration's check fires and returns fail with +// reason="ctx_cancelled" (distinct from "ctx_cancelled_during_poll"). +func TestLegStatus_CtxCancelledAtTopOfLoop(t *testing.T) { + w := &DeployProbeWorker{ + cfg: DeployProbeConfig{ + BaseURL: "http://does-not-matter.invalid", BearerToken: "x", + }.Defaults(), + httpCli: &http.Client{Timeout: time.Second}, + } + ctx, cancel := context.WithCancel(context.Background()) + cancel() // pre-cancel + r := w.legStatus(ctx, "appid") + if r.result != deployProbeResultFail { + t.Errorf("result: got %q, want fail", r.result) + } + if !strings.HasPrefix(r.reason, "ctx_cancelled") { + t.Errorf("reason: got %q, want ctx_cancelled prefix", r.reason) + } +} + +// TestLegStatus_BudgetElapsedBranch — drives the "status=%q at budget" +// branch using injected statusBudget=1ms + pollInterval=1ms so a single +// time.After cycle puts us past the deadline. Requires the test server +// to keep returning building so the default branch hits the +// after-deadline check. +func TestLegStatus_BudgetElapsedBranch(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + _, _ = w.Write([]byte(`{"ok":true,"item":{"status":"building"}}`)) + })) + defer srv.Close() + w := &DeployProbeWorker{ + cfg: DeployProbeConfig{BaseURL: srv.URL, BearerToken: "x"}.Defaults(), + httpCli: srv.Client(), + statusBudget: 1 * time.Millisecond, + pollInterval: 1 * time.Millisecond, + } + r := w.legStatus(context.Background(), "appid") + if r.result != deployProbeResultFail { + t.Errorf("result: got %q, want fail", r.result) + } + if !strings.Contains(r.reason, "at budget") { + t.Errorf("reason: got %q, want 'at budget' substring", r.reason) + } +} + +// TestLegStatus_PollIntervalSleepCompletes — exercises the +// `case <-time.After(pollInterval)` branch. With injected +// pollInterval=1ms and statusBudget=200ms the first iteration's +// time.After fires (covering the case), the loop continues, then the +// budget elapses → fail. +func TestLegStatus_PollIntervalSleepCompletes(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + _, _ = w.Write([]byte(`{"ok":true,"item":{"status":"building"}}`)) + })) + defer srv.Close() + w := &DeployProbeWorker{ + cfg: DeployProbeConfig{BaseURL: srv.URL, BearerToken: "x"}.Defaults(), + httpCli: srv.Client(), + statusBudget: 50 * time.Millisecond, + pollInterval: 1 * time.Millisecond, + } + r := w.legStatus(context.Background(), "appid") + if r.result != deployProbeResultFail { + t.Errorf("result: got %q, want fail", r.result) + } +} + +// TestLegSubmit_DegradedBranch — drives the submit-latency degraded +// branch. Inject submitBudget=1ns so any real-world latency crosses +// the budget; the server still returns a valid response so the leg +// result is degraded (not fail). +func TestLegSubmit_DegradedBranch(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusAccepted) + _, _ = w.Write([]byte(`{"ok":true,"item":{"app_id":"a"}}`)) + })) + defer srv.Close() + w := &DeployProbeWorker{ + cfg: DeployProbeConfig{BaseURL: srv.URL, BearerToken: "x"}.Defaults(), + httpCli: srv.Client(), + submitBudget: 1 * time.Nanosecond, + } + r, appID := w.legSubmit(context.Background()) + if r.result != deployProbeResultDegraded { + t.Errorf("result: got %q, want degraded", r.result) + } + if appID != "a" { + t.Errorf("appID: got %q, want a (degraded still returns the id)", appID) + } +} + +// TestLegServe_DegradedBranch — drives the serve-latency degraded +// branch with serveBudget=1ns. The request itself uses the http +// client's own timeout (not the budget), so any response that lands +// in finite time crosses the 1ns budget and trips degraded. +func TestLegServe_DegradedBranch(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusOK) + })) + defer srv.Close() + srvHostStr := strings.TrimPrefix(srv.URL, "http://") + srvHostStr = strings.TrimPrefix(srvHostStr, "https://") + httpCli := &http.Client{ + Timeout: 2 * time.Second, + Transport: &serveRetargetTransport{base: http.DefaultTransport, serverURL: srv.URL}, + } + w := &DeployProbeWorker{ + cfg: DeployProbeConfig{ + BaseURL: srv.URL, DeployHost: srvHostStr, BearerToken: "x", + }.Defaults(), + httpCli: httpCli, + serveBudget: 1 * time.Nanosecond, + } + r := w.legServe(context.Background(), "appid") + if r.result != deployProbeResultDegraded { + t.Errorf("result: got %q, want degraded", r.result) + } + if !strings.Contains(r.reason, "over budget=") { + t.Errorf("reason: got %q, want 'over budget=' substring", r.reason) + } +} + +// serveRetargetTransport rewrites https://*. requests to the +// test server's plain-http base URL so the leg-3 URL lands on the +// httptest handler without standing up TLS. +type serveRetargetTransport struct { + base http.RoundTripper + serverURL string +} + +func (r *serveRetargetTransport) RoundTrip(req *http.Request) (*http.Response, error) { + if req.URL.Scheme == "https" { + newURL := r.serverURL + req.URL.Path + if req.URL.RawQuery != "" { + newURL += "?" + req.URL.RawQuery + } + nr, err := http.NewRequestWithContext(req.Context(), req.Method, newURL, req.Body) + if err != nil { + return nil, err + } + for k, vs := range req.Header { + for _, v := range vs { + nr.Header.Add(k, v) + } + } + return r.base.RoundTrip(nr) + } + return r.base.RoundTrip(req) +} + +// TestFetchDeployStatus_BuildRequestErr — drive +// http.NewRequestWithContext to error by passing a malformed URL. +// Exercises the defensive build_request branch. +func TestFetchDeployStatus_BuildRequestErr(t *testing.T) { + w := &DeployProbeWorker{ + cfg: DeployProbeConfig{BaseURL: "http://x", BearerToken: "x"}.Defaults(), + httpCli: &http.Client{Timeout: time.Second}, + } + _, _, err := w.fetchDeployStatus(context.Background(), "http://x/\x7f") + if err == nil { + t.Errorf("want err, got nil") + } +} + +// TestFetchDeployStatus_HTTPErr — DNS-fail transport error. +func TestFetchDeployStatus_HTTPErr(t *testing.T) { + w := &DeployProbeWorker{ + cfg: DeployProbeConfig{BaseURL: "http://x", BearerToken: "x"}.Defaults(), + httpCli: &http.Client{Timeout: 200 * time.Millisecond}, + } + _, _, err := w.fetchDeployStatus(context.Background(), "http://this-does-not-resolve.invalid/") + if err == nil { + t.Errorf("want err, got nil") + } +} + +// TestLegServe_HTTPErr — DNS-unresolvable host trips httpCli.Do with +// a transport error. Exercises the http_error branch on the serve leg. +func TestLegServe_HTTPErr(t *testing.T) { + w := &DeployProbeWorker{ + cfg: DeployProbeConfig{ + BaseURL: "http://x", + DeployHost: "this-does-not-resolve.invalid", + BearerToken: "x", + }.Defaults(), + httpCli: &http.Client{Timeout: 200 * time.Millisecond}, + } + r := w.legServe(context.Background(), "appid") + if r.result != deployProbeResultFail { + t.Errorf("result: got %q, want fail", r.result) + } + if !strings.Contains(r.reason, "http_error") { + t.Errorf("reason: got %q, want http_error prefix", r.reason) + } +} + +// TestLegServe_BuildRequestErr — control char in DeployHost trips +// http.NewRequestWithContext, exercising the build_request branch. +func TestLegServe_BuildRequestErr(t *testing.T) { + w := &DeployProbeWorker{ + cfg: DeployProbeConfig{ + BaseURL: "http://x", + DeployHost: "bad\x7fhost", + BearerToken: "x", + }.Defaults(), + httpCli: &http.Client{Timeout: time.Second}, + } + r := w.legServe(context.Background(), "appid") + if r.result != deployProbeResultFail { + t.Errorf("result: got %q, want fail", r.result) + } + if !strings.Contains(r.reason, "build_request") { + t.Errorf("reason: got %q, want build_request prefix", r.reason) + } +} + +// TestEffectiveBudgetHelpers_ZeroFallback — each helper returns the +// package-level constant when the per-worker field is zero, and the +// per-worker value when non-zero. +func TestEffectiveBudgetHelpers_ZeroFallback(t *testing.T) { + w := &DeployProbeWorker{} + if got := w.effectiveSubmitBudget(); got != deployProbeSubmitBudget { + t.Errorf("submit fallback: got %v, want %v", got, deployProbeSubmitBudget) + } + if got := w.effectiveStatusBudget(); got != deployProbeStatusBudget { + t.Errorf("status fallback: got %v, want %v", got, deployProbeStatusBudget) + } + if got := w.effectiveServeBudget(); got != deployProbeServeBudget { + t.Errorf("serve fallback: got %v, want %v", got, deployProbeServeBudget) + } + if got := w.effectivePollInterval(); got != deployProbePollInterval { + t.Errorf("poll fallback: got %v, want %v", got, deployProbePollInterval) + } + + w2 := &DeployProbeWorker{ + submitBudget: 1 * time.Second, + statusBudget: 2 * time.Second, + serveBudget: 3 * time.Second, + pollInterval: 4 * time.Second, + } + if got := w2.effectiveSubmitBudget(); got != time.Second { + t.Errorf("submit override: got %v, want 1s", got) + } + if got := w2.effectiveStatusBudget(); got != 2*time.Second { + t.Errorf("status override: got %v, want 2s", got) + } + if got := w2.effectiveServeBudget(); got != 3*time.Second { + t.Errorf("serve override: got %v, want 3s", got) + } + if got := w2.effectivePollInterval(); got != 4*time.Second { + t.Errorf("poll override: got %v, want 4s", got) + } +} + +// TestNewDeployProbeWorker_CheckRedirectClosure — drive the default +// client's CheckRedirect closure by 302-ing the submit POST against an +// httptest server. The closure returns http.ErrUseLastResponse so the +// prober observes the 302 directly as non-2xx (fail). +func TestNewDeployProbeWorker_CheckRedirectClosure(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Location", "/elsewhere") + w.WriteHeader(http.StatusFound) + })) + defer srv.Close() + w := NewDeployProbeWorker(nil, nil, nil, DeployProbeConfig{ + BaseURL: srv.URL, + BearerToken: "x", + }) + r, _ := w.legSubmit(context.Background()) + if r.result != deployProbeResultFail { + t.Errorf("result: got %q, want fail (302 is non-2xx)", r.result) + } +}