Automated Baseline Promotion and Rollback

{
  "$id": "https://queryplan.org/schemas/baseline-version-transition.json",
  "type": "object",
  "additionalProperties": false,
  "required": ["query_fingerprint", "from_version", "to_version", "trigger", "probation_executions", "probation_seconds", "cost_delta_ratio", "p95_ratio"],
  "properties": {
    "query_fingerprint":     { "type": "string", "pattern": "^[a-f0-9]{64}$" },
    "from_version":          { "type": "integer", "minimum": 0 },
    "to_version":            { "type": "integer", "exclusiveMinimum": 0 },
    "trigger":               { "type": "string", "enum": ["PROMOTE", "ROLLBACK"] },
    "probation_executions":  { "type": "integer", "minimum": 0 },
    "probation_seconds":     { "type": "integer", "minimum": 0 },
    "cost_delta_ratio":      { "type": "number", "exclusiveMinimum": 0 },
    "p95_ratio":             { "type": "number", "exclusiveMinimum": 0 }
  }
}

from __future__ import annotations

import os
from dataclasses import dataclass
from typing import Literal

import asyncpg
import structlog
from opentelemetry import metrics, trace

log = structlog.get_logger(__name__)
tracer = trace.get_tracer(__name__)
meter = metrics.get_meter(__name__)

PROMOTION_TOTAL = meter.create_counter("baseline_promotion_total", unit="1")
ROLLBACK_TOTAL = meter.create_counter("baseline_rollback_total", unit="1")
SWAP_DURATION = meter.create_histogram("baseline_swap_duration_ms", unit="ms")

# --- Exact promotion / rollback thresholds (mirror promotion.yaml) --------
MIN_PROBATION_EXECS = 200
MIN_PROBATION_SECONDS = 86_400          # 24 h canary window
MAX_COST_DELTA_RATIO = 1.1              # candidate cost ≤ 1.1× incumbent
ROLLBACK_P95_RATIO = 2.0               # revert if p95 > 2× within window

Trigger = Literal["PROMOTE", "ROLLBACK", "HOLD"]


@dataclass(frozen=True)
class CandidateEvidence:
    query_fingerprint: str
    candidate_plan_id: int
    probation_executions: int
    probation_seconds: int
    cost_delta_ratio: float
    p95_ratio: float
    idempotency_key: str


def decide(ev: CandidateEvidence, has_incumbent: bool) -> Trigger:
    """Pure routing over the evidence bundle — no I/O, fully replayable."""
    if has_incumbent and ev.p95_ratio > ROLLBACK_P95_RATIO:
        return "ROLLBACK"
    if (
        ev.probation_executions >= MIN_PROBATION_EXECS
        and ev.probation_seconds >= MIN_PROBATION_SECONDS
        and ev.cost_delta_ratio <= MAX_COST_DELTA_RATIO
    ):
        return "PROMOTE"
    return "HOLD"


class PromotionController:
    """Atomic, idempotent baseline anchor transitions over asyncpg."""

    def __init__(self, pool: asyncpg.Pool) -> None:
        self._pool = pool

    async def run(self, ev: CandidateEvidence) -> Trigger:
        with tracer.start_as_current_span("baseline.promotion") as span:
            span.set_attribute("query_fingerprint", ev.query_fingerprint)
            async with self._pool.acquire() as conn:
                async with conn.transaction():
                    incumbent = await conn.fetchrow(
                        """
                        SELECT current_version, plan_id
                        FROM baseline_anchor
                        WHERE query_fingerprint = $1
                        FOR UPDATE
                        """,
                        ev.query_fingerprint,
                    )
                    trigger = decide(ev, incumbent is not None)
                    span.set_attribute("trigger", trigger)

                    if trigger == "HOLD":
                        log.info("promotion_hold",
                                 fingerprint=ev.query_fingerprint,
                                 execs=ev.probation_executions,
                                 seconds=ev.probation_seconds)
                        return "HOLD"

                    from_version = incumbent["current_version"] if incumbent else 0
                    to_version = from_version + 1
                    target_plan = (
                        incumbent["plan_id"] if trigger == "ROLLBACK"
                        else ev.candidate_plan_id
                    )
                    # For ROLLBACK the target is the *prior* pinned plan, resolved
                    # from the ledger rather than the live incumbent.
                    if trigger == "ROLLBACK":
                        prior = await conn.fetchval(
                            """
                            SELECT plan_id FROM baseline_version_transition
                            WHERE query_fingerprint = $1 AND trigger = 'PROMOTE'
                            ORDER BY to_version DESC OFFSET 1 LIMIT 1
                            """,
                            ev.query_fingerprint,
                        )
                        if prior is None:
                            raise RuntimeError("no prior baseline to roll back to")
                        target_plan = prior

                    # Idempotent transition insert; a replay collides and no-ops.
                    inserted = await conn.fetchval(
                        """
                        INSERT INTO baseline_version_transition
                            (query_fingerprint, from_version, to_version, trigger,
                             plan_id, cost_delta_ratio, p95_ratio,
                             probation_executions, probation_seconds,
                             idempotency_key, created_at)
                        VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10, now())
                        ON CONFLICT (idempotency_key) DO NOTHING
                        RETURNING to_version
                        """,
                        ev.query_fingerprint, from_version, to_version, trigger,
                        target_plan, ev.cost_delta_ratio, ev.p95_ratio,
                        ev.probation_executions, ev.probation_seconds,
                        ev.idempotency_key,
                    )
                    if inserted is None:
                        log.info("promotion_replay_noop",
                                 fingerprint=ev.query_fingerprint,
                                 idempotency_key=ev.idempotency_key)
                        return trigger

                    # Move the anchor and write the audit event in the same txn.
                    await conn.execute(
                        """
                        INSERT INTO baseline_anchor
                            (query_fingerprint, current_version, plan_id, updated_at)
                        VALUES ($1, $2, $3, now())
                        ON CONFLICT (query_fingerprint)
                        DO UPDATE SET current_version = EXCLUDED.current_version,
                                      plan_id = EXCLUDED.plan_id,
                                      updated_at = now()
                        """,
                        ev.query_fingerprint, to_version, target_plan,
                    )
                    await conn.execute(
                        """
                        INSERT INTO baseline_audit_event
                            (query_fingerprint, from_version, to_version,
                             trigger, actor, created_at)
                        VALUES ($1,$2,$3,$4,'promotion-controller', now())
                        """,
                        ev.query_fingerprint, from_version, to_version, trigger,
                    )

            if trigger == "PROMOTE":
                PROMOTION_TOTAL.add(1, {"result": "committed"})
            else:
                ROLLBACK_TOTAL.add(1, {"result": "committed"})
            log.info("baseline_transition",
                     fingerprint=ev.query_fingerprint,
                     from_version=from_version, to_version=to_version,
                     trigger=trigger)
            return trigger


async def build_pool() -> asyncpg.Pool:
    return await asyncpg.create_pool(
        dsn=os.environ["PROMOTION_DSN"],
        min_size=2,
        max_size=int(os.environ.get("PROMOTION_POOL_MAX", "8")),
        command_timeout=5.0,
    )

groups:
  - name: baseline-promotion-controller
    rules:
      - alert: BaselineRollbackStorm
        expr: |
          (
            sum(rate(baseline_rollback_total[30m]))
            /
            clamp_min(sum(rate(baseline_promotion_total[30m])), 1)
          ) > 0.06
        for: 15m
        labels:
          severity: page
        annotations:
          summary: "Rollback rate above 6% for 15m — probation window likely too short"
          runbook: "Raise min_probation_seconds toward 172800 (48 h) before re-promoting."

      - alert: BaselineSwapLatencyHigh
        expr: |
          histogram_quantile(0.95, sum(rate(baseline_swap_duration_ms_bucket[10m])) by (le)) > 120
        for: 10m
        labels:
          severity: ticket
        annotations:
          summary: "Version-swap p95 > 120ms — check anchor-row lock contention"

SELECT query_fingerprint, probation_executions, probation_seconds,
       (probation_executions >= 200) AS exec_ok,
       (probation_seconds >= 86400)  AS window_ok
FROM candidate_probation
WHERE query_fingerprint = $1;

jq -r 'select(.query_fingerprint=="'$FP'") | .trace_id' probation.jsonl | sort -u | wc -l

SELECT to_version, trigger, cost_delta_ratio, created_at
FROM baseline_version_transition
WHERE query_fingerprint = $1
ORDER BY to_version DESC LIMIT 8;

SELECT a.query_fingerprint, a.current_version
FROM baseline_anchor a
LEFT JOIN baseline_audit_event e
  ON e.query_fingerprint = a.query_fingerprint
 AND e.to_version = a.current_version
WHERE e.query_fingerprint IS NULL;