Bake channels: a bake reaches buyers only after its smoke clone passes

migrations/20261026100000_baked_snapshot_channels.down.sql

@@ -0,0 +1,3 @@
+DELETE FROM public.casbin_rule WHERE ptype = 'p' AND v1 = '/api/admin/bake/:id/promote';
+ALTER TABLE baked_snapshots DROP COLUMN IF EXISTS gated;
+DROP TABLE IF EXISTS baked_snapshot_channels;

migrations/20261026100000_baked_snapshot_channels.up.sql

@@ -0,0 +1,23 @@
+-- Release channels for baked snapshots (BAKE_PIPELINE_PLAN.md, Stage 4): the
+-- pointers helpers/bake_registry.rs moves. A bake started from the API lands on
+-- `unstable`; buyers get `latest` (promoted after a green smoke clone) or, when
+-- that is unhealthy, `stable` (the previous `latest`).
+CREATE TABLE IF NOT EXISTS baked_snapshot_channels (
+    stack       TEXT NOT NULL,
+    provider    TEXT NOT NULL,
+    channel     TEXT NOT NULL,
+    snapshot_id INTEGER NOT NULL REFERENCES baked_snapshots (id),
+    updated_at  TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+    PRIMARY KEY (stack, provider, channel)
+);
+
+CREATE INDEX IF NOT EXISTS idx_baked_snapshot_channels_snapshot
+    ON baked_snapshot_channels (snapshot_id);
+
+-- A gated snapshot reaches buyers only through `latest` or `stable`. Rows
+-- recorded before this migration, and by the `bake` binary, stay ungated and
+-- keep the "newest healthy" behaviour.
+ALTER TABLE baked_snapshots ADD COLUMN IF NOT EXISTS gated BOOLEAN NOT NULL DEFAULT FALSE;
+
+INSERT INTO public.casbin_rule (ptype, v0, v1, v2, v3, v4, v5) VALUES ('p', 'group_admin', '/api/admin/bake/:id/promote', 'POST', '', '', '') ON CONFLICT ON CONSTRAINT unique_key_sqlx_adapter DO NOTHING;
+INSERT INTO public.casbin_rule (ptype, v0, v1, v2, v3, v4, v5) VALUES ('p', 'admin_service', '/api/admin/bake/:id/promote', 'POST', '', '', '') ON CONFLICT ON CONSTRAINT unique_key_sqlx_adapter DO NOTHING;

src/db/bake_registry.rs

@@ -0,0 +1,155 @@
+//! Channel pointers for baked snapshots (`baked_snapshot_channels`). The
+//! promotion decision itself is `helpers::bake_registry::promote_ladder`.
+
+use crate::helpers::bake_registry::{
+    promote_ladder, ChannelState, PromoteMoves, CHANNEL_LATEST, CHANNEL_STABLE, CHANNEL_UNSTABLE,
+};
+use sqlx::PgPool;
+
+/// Point `channel` of a stack at a snapshot, replacing what it pointed at.
+pub async fn point<'e, E: sqlx::PgExecutor<'e>>(
+    executor: E,
+    stack: &str,
+    provider: &str,
+    channel: &str,
+    snapshot_id: i32,
+) -> Result<(), String> {
+    sqlx::query(
+        r#"
+        INSERT INTO baked_snapshot_channels (stack, provider, channel, snapshot_id)
+        VALUES ($1, $2, $3, $4)
+        ON CONFLICT (stack, provider, channel)
+        DO UPDATE SET snapshot_id = EXCLUDED.snapshot_id, updated_at = NOW()
+        "#,
+    )
+    .bind(stack)
+    .bind(provider)
+    .bind(channel)
+    .bind(snapshot_id)
+    .execute(executor)
+    .await
+    .map(|_| ())
+    .map_err(|e| format!("Failed to move the {channel} channel: {e}"))
+}
+
+/// Where each channel of a stack points now.
+pub async fn state(pool: &PgPool, stack: &str, provider: &str) -> Result<ChannelState, String> {
+    let rows: Vec<(String, i32)> = sqlx::query_as(
+        "SELECT channel, snapshot_id FROM baked_snapshot_channels WHERE stack = $1 AND provider = $2",
+    )
+    .bind(stack)
+    .bind(provider)
+    .fetch_all(pool)
+    .await
+    .map_err(|e| format!("Failed to read channels: {e}"))?;
+    Ok(to_state(rows))
+}
+
+/// The channels of the stack (and provider) a snapshot belongs to.
+pub async fn state_for_snapshot(pool: &PgPool, snapshot_id: i32) -> Result<ChannelState, String> {
+    let rows: Vec<(String, i32)> = sqlx::query_as(
+        r#"
+        SELECT c.channel, c.snapshot_id
+        FROM baked_snapshots s
+        JOIN baked_snapshot_channels c ON c.stack = s.stack AND c.provider = s.provider
+        WHERE s.id = $1
+        "#,
+    )
+    .bind(snapshot_id)
+    .fetch_all(pool)
+    .await
+    .map_err(|e| format!("Failed to read channels: {e}"))?;
+    Ok(to_state(rows))
+}
+
+fn to_state(rows: Vec<(String, i32)>) -> ChannelState {
+    let mut state = ChannelState {
+        unstable: None,
+        latest: None,
+        stable: None,
+    };
+    for (channel, id) in rows {
+        match channel.as_str() {
+            CHANNEL_UNSTABLE => state.unstable = Some(id),
+            CHANNEL_LATEST => state.latest = Some(id),
+            CHANNEL_STABLE => state.stable = Some(id),
+            _ => {}
+        }
+    }
+    state
+}
+
+#[derive(Debug)]
+pub enum PromoteError {
+    /// No such snapshot.
+    NotFound,
+    /// The snapshot is not what `unstable` points at: a newer bake replaced it,
+    /// or it was never baked through the API.
+    NotUnstable,
+    Db(String),
+}
+
+impl std::fmt::Display for PromoteError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            PromoteError::NotFound => write!(f, "no such baked snapshot"),
+            PromoteError::NotUnstable => write!(
+                f,
+                "the snapshot is not the stack's current unstable bake, so it cannot be promoted"
+            ),
+            PromoteError::Db(e) => write!(f, "{e}"),
+        }
+    }
+}
+
+/// Promote the stack's `unstable` snapshot to `latest` (old `latest` to
+/// `stable`) after a green smoke clone. `snapshot_id` must be the current
+/// `unstable`, so a smoke result is never applied to a different bake.
+/// `Ok(None)` when it is already `latest`: repeating a promote is harmless.
+pub async fn promote(
+    pool: &PgPool,
+    snapshot_id: i32,
+) -> Result<Option<PromoteMoves>, PromoteError> {
+    let db = |e: sqlx::Error| PromoteError::Db(format!("Failed to promote: {e}"));
+    let mut tx = pool.begin().await.map_err(db)?;
+    let (stack, provider): (String, String) =
+        sqlx::query_as("SELECT stack, provider FROM baked_snapshots WHERE id = $1")
+            .bind(snapshot_id)
+            .fetch_optional(&mut *tx)
+            .await
+            .map_err(db)?
+            .ok_or(PromoteError::NotFound)?;
+    // Lock the stack's pointers: two promotes, or a bake landing meanwhile,
+    // must not interleave.
+    let rows: Vec<(String, i32)> = sqlx::query_as(
+        "SELECT channel, snapshot_id FROM baked_snapshot_channels WHERE stack = $1 AND provider = $2 FOR UPDATE",
+    )
+    .bind(&stack)
+    .bind(&provider)
+    .fetch_all(&mut *tx)
+    .await
+    .map_err(db)?;
+    let state = to_state(rows);
+    if state.unstable != Some(snapshot_id) {
+        return Err(PromoteError::NotUnstable);
+    }
+    let Some(moves) = promote_ladder(&state) else {
+        return Ok(None);
+    };
+    point(
+        &mut *tx,
+        &stack,
+        &provider,
+        CHANNEL_LATEST,
+        moves.new_latest,
+    )
+    .await
+    .map_err(PromoteError::Db)?;
+    if let Some(stable) = moves.new_stable {
+        point(&mut *tx, &stack, &provider, CHANNEL_STABLE, stable)
+            .await
+            .map_err(PromoteError::Db)?;
+    }
+    tx.commit().await.map_err(db)?;
+    Ok(Some(moves))
+}

src/db/baked_snapshot.rs

@@ -1,26 +1,34 @@
 use crate::models::BakedSnapshot;
 use sqlx::PgPool;
 
-/// Resolve the most recent healthy baked snapshot image_id for a stack+version
-/// on a provider. Used by the clone deploy path to find what to clone from.
+/// Which snapshots buyers may get, and in what order: a gated snapshot (baked
+/// through the API) only through the `latest` or `stable` channel; `latest`
+/// first, then `stable`, then the newest ungated one, so stacks baked before
+/// channels existed keep working. `s` is `baked_snapshots`.
+const SERVABLE: &str = r#"
+    LEFT JOIN baked_snapshot_channels c
+      ON c.snapshot_id = s.id AND c.channel IN ('latest', 'stable')
+    WHERE s.healthy = TRUE
+      AND (s.gated = FALSE OR c.channel IS NOT NULL)
+"#;
+const SERVE_ORDER: &str = r#"
+    ORDER BY CASE c.channel WHEN 'latest' THEN 0 WHEN 'stable' THEN 1 ELSE 2 END,
+             s.created_at DESC
+    LIMIT 1
+"#;
+
+/// Resolve the snapshot buyers get for a stack+version on a provider. Used by
+/// the clone deploy path to find what to clone from.
 pub async fn resolve(
     pool: &PgPool,
     stack: &str,
     version: &str,
     provider: &str,
 ) -> Result<Option<BakedSnapshot>, String> {
-    sqlx::query_as::<_, BakedSnapshot>(
-        r#"
-        SELECT *
-        FROM baked_snapshots
-        WHERE stack = $1
-          AND version = $2
-          AND provider = $3
-          AND healthy = TRUE
-        ORDER BY created_at DESC
-        LIMIT 1
-        "#,
-    )
+    sqlx::query_as::<_, BakedSnapshot>(&format!(
+        "SELECT s.* FROM baked_snapshots s {SERVABLE} \
+         AND s.stack = $1 AND s.version = $2 AND s.provider = $3 {SERVE_ORDER}"
+    ))
     .bind(stack)
     .bind(version)
     .bind(provider)
@@ -29,32 +37,52 @@ pub async fn resolve(
     .map_err(|e| format!("Failed to resolve baked snapshot: {e}"))
 }
 
-/// Resolve any healthy snapshot for a stack (any version), newest first.
-/// Used when the exact version has not been baked but a generic one exists.
+/// Resolve the snapshot buyers get for a stack (any version): `latest`, else
+/// `stable`, else the newest ungated healthy one.
 pub async fn resolve_latest(
     pool: &PgPool,
     stack: &str,
     provider: &str,
+) -> Result<Option<BakedSnapshot>, String> {
+    sqlx::query_as::<_, BakedSnapshot>(&format!(
+        "SELECT s.* FROM baked_snapshots s {SERVABLE} \
+         AND s.stack = $1 AND s.provider = $2 {SERVE_ORDER}"
+    ))
+    .bind(stack)
+    .bind(provider)
+    .fetch_optional(pool)
+    .await
+    .map_err(|e| format!("Failed to resolve latest baked snapshot: {e}"))
+}
+
+/// The bake a stack's `unstable` channel points at: what the smoke clone checks
+/// before promotion. `version`, when given, must be that bake's version; there
+/// is no fallback to a released snapshot.
+pub async fn resolve_unstable(
+    pool: &PgPool,
+    stack: &str,
+    version: Option<&str>,
+    provider: &str,
 ) -> Result<Option<BakedSnapshot>, String> {
     sqlx::query_as::<_, BakedSnapshot>(
         r#"
-        SELECT *
-        FROM baked_snapshots
-        WHERE stack = $1
-          AND provider = $2
-          AND healthy = TRUE
-        ORDER BY created_at DESC
-        LIMIT 1
+        SELECT s.* FROM baked_snapshot_channels c
+        JOIN baked_snapshots s ON s.id = c.snapshot_id
+        WHERE c.stack = $1 AND c.provider = $2 AND c.channel = 'unstable'
+          AND s.healthy = TRUE AND ($3::TEXT IS NULL OR s.version = $3)
         "#,
     )
     .bind(stack)
     .bind(provider)
+    .bind(version)
     .fetch_optional(pool)
     .await
-    .map_err(|e| format!("Failed to resolve latest baked snapshot: {e}"))
+    .map_err(|e| format!("Failed to resolve the unstable baked snapshot: {e}"))
 }
 
-/// Record a published bake. Returns the new row.
+/// Record a published bake. Returns the new row. The snapshot is served at
+/// once (ungated): the `bake` binary's path.
+#[allow(clippy::too_many_arguments)]
 pub async fn record(
     pool: &PgPool,
     stack: &str,
@@ -66,11 +94,88 @@ pub async fn record(
     config_contract: Option<serde_json::Value>,
     required_env_keys: Option<serde_json::Value>,
     server_type: Option<&str>,
+) -> Result<BakedSnapshot, String> {
+    insert(
+        pool,
+        stack,
+        version,
+        provider,
+        image_id,
+        healthy,
+        digests,
+        config_contract,
+        required_env_keys,
+        server_type,
+        false,
+    )
+    .await
+}
+
+/// Record a bake that must pass a smoke clone before buyers get it: the row is
+/// gated and lands on the `unstable` channel, in one transaction.
+#[allow(clippy::too_many_arguments)]
+pub async fn record_unstable(
+    pool: &PgPool,
+    stack: &str,
+    version: &str,
+    provider: &str,
+    image_id: i64,
+    healthy: bool,
+    digests: Option<serde_json::Value>,
+    config_contract: Option<serde_json::Value>,
+    required_env_keys: Option<serde_json::Value>,
+    server_type: Option<&str>,
+) -> Result<BakedSnapshot, String> {
+    let mut tx = pool
+        .begin()
+        .await
+        .map_err(|e| format!("Failed to record baked snapshot: {e}"))?;
+    let row = insert(
+        &mut *tx,
+        stack,
+        version,
+        provider,
+        image_id,
+        healthy,
+        digests,
+        config_contract,
+        required_env_keys,
+        server_type,
+        true,
+    )
+    .await?;
+    crate::db::bake_registry::point(
+        &mut *tx,
+        &row.stack,
+        &row.provider,
+        crate::helpers::bake_registry::CHANNEL_UNSTABLE,
+        row.id,
+    )
+    .await?;
+    tx.commit()
+        .await
+        .map_err(|e| format!("Failed to record baked snapshot: {e}"))?;
+    Ok(row)
+}
+
+#[allow(clippy::too_many_arguments)]
+async fn insert<'e, E: sqlx::PgExecutor<'e>>(
+    executor: E,
+    stack: &str,
+    version: &str,
+    provider: &str,
+    image_id: i64,
+    healthy: bool,
+    digests: Option<serde_json::Value>,
+    config_contract: Option<serde_json::Value>,
+    required_env_keys: Option<serde_json::Value>,
+    server_type: Option<&str>,
+    gated: bool,
 ) -> Result<BakedSnapshot, String> {
     sqlx::query_as::<_, BakedSnapshot>(
         r#"
-        INSERT INTO baked_snapshots (stack, version, provider, image_id, healthy, digests, config_contract, required_env_keys, server_type)
-        VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
+        INSERT INTO baked_snapshots (stack, version, provider, image_id, healthy, digests, config_contract, required_env_keys, server_type, gated)
+        VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)
         RETURNING *
         "#,
     )
@@ -83,7 +188,8 @@ pub async fn record(
     .bind(config_contract)
     .bind(required_env_keys)
     .bind(server_type)
-    .fetch_one(pool)
+    .bind(gated)
+    .fetch_one(executor)
     .await
     .map_err(|e| format!("Failed to record baked snapshot: {e}"))
 }