polyc-controller 2026.8.3

//! Shared infrastructure for the reconcilers that fan out into the apps
//! namespace (`ServiceDefinition`, `Workflow`).
//!
//! The pre-flight that waits until a fanned-out watch is serveable, and the
//! label every fanned-out object carries. Both reconcilers run the same
//! watch-availability dance and stamp the same `managed-by` label, so it
//! lives here rather than in two near-identical copies.
//!
//! A watch preflight can fail for two distinct reasons, and
//! `classify_watch_error` tells them apart so each gets its own log line
//! and remedy instead of one message that points every reader at the wrong
//! fix:
//!
//! - The namespace, its RBAC, or the apiserver isn't ready yet — self-healing,
//!   worth a quiet retry.
//! - The list response's body fails to decode into the expected type because
//!   the cluster's installed CRD schema is older than what this binary
//!   expects — a schema skew. This never self-heals (the CRD doesn't change
//!   underneath a running process), so it gets a distinct, loud log line
//!   naming the real fix.
//!
//! [`await_watchable`] and its cancellable sibling
//! [`await_watchable_cancellable`] are `pub` (the rest of this module stays
//! `pub(crate)`) so `polyc-control-plane`'s in-process routine scheduler
//! (#1370) can reuse the exact same "stay dormant until the watch is
//! serveable" preflight its own Routine reflector watch needs, instead of a
//! second near-identical retry loop — the cancellable variant, since the
//! scheduler's own shutdown must be able to unwind this wait (#1370 review
//! finding 3).

use std::time::Duration;

use kube::Api;
use serde::de::DeserializeOwned;
use tokio_util::sync::CancellationToken;

/// Why a fan-out watch's list preflight failed.
///
/// `classify_watch_error` is the sole place that tells the two apart, so
/// the classification can't drift out of sync between the two call sites
/// ([`await_watchable`] and [`await_watchable_cancellable`]).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum WatchErrorClass {
    /// The list HTTP call failed, or the apiserver returned a well-formed
    /// error response — the namespace, its RBAC, or the apiserver itself
    /// isn't ready yet. Self-healing: retrying the same list later is the
    /// correct remedy.
    NamespaceOrRbacUnready,
    /// The list response's body failed to deserialize into the expected
    /// type — the cluster's installed CRD schema is older than the schema
    /// this binary expects. Not self-healing: the CRD won't change
    /// underneath a running process, so retrying the same list forever
    /// cannot fix it; only applying the newer CRD does.
    SchemaSkew,
}

/// Classifies a failed `Api::list` call as a schema skew or a not-yet-ready
/// namespace/RBAC/apiserver, so callers can log — and alert on — the two
/// differently.
///
/// `kube`'s HTTP client deserializes a list response's body with
/// `serde_json::from_str` and reports a failure to parse it into the
/// requested type as [`kube::Error::SerdeError`] — the same variant a
/// genuinely malformed JSON payload would produce, but in practice this is
/// what a `list::<K>` call sees when the installed CRD's schema no longer
/// matches `K` (for example, a tagged-enum field whose accepted variants
/// changed out from under a running binary). [`kube::Error::Api`] (an
/// HTTP-level error the apiserver itself returned, already pattern-matched
/// elsewhere in this crate — see `routine_reconcile.rs` and `reconcile.rs`)
/// and every other `kube::Error` variant fall through to
/// [`WatchErrorClass::NamespaceOrRbacUnready`], the pre-existing behavior.
pub(crate) const fn classify_watch_error(err: &kube::Error) -> WatchErrorClass {
    match err {
        kube::Error::SerdeError(_) => WatchErrorClass::SchemaSkew,
        _ => WatchErrorClass::NamespaceOrRbacUnready,
    }
}

/// Logs one line for a failed watch-preflight list, in the class-appropriate
/// severity and wording. Shared by [`await_watchable`] and
/// [`await_watchable_cancellable`] so the two loops can't drift apart on
/// what they tell the reader.
///
/// A schema skew logs at `ERROR`, immediately, every retry — it can't
/// self-heal, so folding it into the namespace/RBAC `WARN` would let a
/// permanent misconfiguration hide inside a message that reads as transient
/// start-up noise. It stays on the same retry cadence rather than aborting
/// the process: the caller may be serving unrelated duties (`AgentService`,
/// other reconcilers) that a decoding problem in one fan-out watch shouldn't
/// take down.
fn log_watch_error(namespace: &str, kind: &str, err: &kube::Error) {
    match classify_watch_error(err) {
        WatchErrorClass::SchemaSkew => {
            tracing::error!(
                %namespace,
                %kind,
                error = %err,
                "fan-out watch response failed to decode: the installed CRD schema is \
                 older than what this binary expects (schema skew, not a namespace/RBAC \
                 problem); apply the current CRDs with `kubectl apply -k manifests/crds`; \
                 retrying in 5m"
            );
        }
        WatchErrorClass::NamespaceOrRbacUnready => {
            tracing::warn!(
                %namespace,
                %kind,
                error = %err,
                "fan-out watch not available (apps namespace/RBAC unprovisioned? \
                 apply manifests/apps-namespace); retrying in 5m"
            );
        }
    }
}

/// How often [`await_watchable`] retries while the watch is unavailable —
/// either the apps namespace/RBAC is unprovisioned (the namespace and its
/// Role ship separately from the base manifests, so a base-only install has
/// neither) or the installed CRD schema is skewed against this binary.
pub(crate) const CHECK_RETRY_INTERVAL: Duration = Duration::from_mins(5);

/// Block until a one-item list over `api` succeeds.
///
/// Logs one actionable line every 5 minutes (`CHECK_RETRY_INTERVAL`)
/// meanwhile (`kind` names the watched resource); see the module docs for
/// how a namespace/RBAC failure and a CRD schema skew are told apart and
/// logged differently.
///
/// Lets a fan-out reconciler stay dormant — rather than error-loop its watch
/// — until the target namespace/RBAC is provisioned, then start serving.
///
/// Not cancellable: every existing reconciler call site runs this with no
/// shutdown token in scope, and is unaffected by this doc update. A caller
/// whose own shutdown must be able to unwind this wait — `polyc-control-
/// plane`'s in-process routine scheduler (#1370) — wants
/// [`await_watchable_cancellable`] instead.
pub async fn await_watchable<K>(api: &Api<K>, namespace: &str, kind: &str)
where
    K: Clone + DeserializeOwned + std::fmt::Debug,
{
    loop {
        match api.list(&kube::api::ListParams::default().limit(1)).await {
            Ok(_) => break,
            Err(err) => {
                log_watch_error(namespace, kind, &err);
                tokio::time::sleep(CHECK_RETRY_INTERVAL).await;
            }
        }
    }
}

/// Cancellable variant of [`await_watchable`].
///
/// Identical retry/backoff behavior — including how a namespace/RBAC failure
/// and a CRD schema skew are classified and logged, see the module docs —
/// but `shutdown` firing aborts the wait immediately instead of blocking it
/// for up to a full `CHECK_RETRY_INTERVAL` sleep — or, while the watch is
/// never serveable, forever. Needed by a caller whose preflight runs on the
/// same task a graceful shutdown must be able to unwind: the in-process
/// routine scheduler (#1370 review finding 3) — a missing `Routine` CRD, a
/// schema skew, or an unreachable apiserver at `SIGTERM` must never wedge
/// shutdown until `SIGKILL`.
///
/// Returns `true` once the watch is serveable, `false` if `shutdown` fired
/// first. The existing reconciler call sites keep calling
/// [`await_watchable`] unchanged — they have no shutdown token in scope at
/// that call site and always want the unconditional blocking wait.
pub async fn await_watchable_cancellable<K>(
    api: &Api<K>,
    namespace: &str,
    kind: &str,
    shutdown: &CancellationToken,
) -> bool
where
    K: Clone + DeserializeOwned + std::fmt::Debug,
{
    loop {
        let params = kube::api::ListParams::default().limit(1);
        tokio::select! {
            biased;
            () = shutdown.cancelled() => return false,
            res = api.list(&params) => match res {
                Ok(_) => return true,
                Err(err) => {
                    log_watch_error(namespace, kind, &err);
                    tokio::select! {
                        biased;
                        () = shutdown.cancelled() => return false,
                        () = tokio::time::sleep(CHECK_RETRY_INTERVAL) => {}
                    }
                }
            },
        }
    }
}

/// Labels stamped on every fanned-out object: the standard selector name
/// label, the [`MANAGED_BY_KEY`] marker, and a lineage label (`lineage_key`)
/// pointing back at the owning custom resource by name.
///
/// One definition instead of a near-identical copy per reconciler, so a new
/// fan-out kind can't drift the label set.
pub(crate) fn labels(name: &str, lineage_key: &str) -> std::collections::BTreeMap<String, String> {
    std::collections::BTreeMap::from([
        ("app.kubernetes.io/name".to_owned(), name.to_owned()),
        (MANAGED_BY_KEY.to_owned(), MANAGED_BY_VALUE.to_owned()),
        (lineage_key.to_owned(), name.to_owned()),
    ])
}

/// `app.kubernetes.io/managed-by` label key stamped on every object the control
/// plane fans out, so operators and selectors can find platform-owned objects.
pub(crate) const MANAGED_BY_KEY: &str = "app.kubernetes.io/managed-by";

/// Value for [`MANAGED_BY_KEY`].
pub(crate) const MANAGED_BY_VALUE: &str = "polychrome";

#[cfg(test)]
mod tests {
    #![allow(clippy::pedantic, clippy::nursery, missing_docs)]

    use std::time::Duration;

    use http::StatusCode;
    use k8s_openapi::api::core::v1::ConfigMap;
    use kube::Api;
    use tokio_util::sync::CancellationToken;

    use super::{WatchErrorClass, await_watchable_cancellable, classify_watch_error};

    /// A hand-rolled fake `kube::Client` whose every request errors (a
    /// permanent 500) — models an apiserver `await_watchable_cancellable`
    /// can never reach, so its retry loop keeps sleeping
    /// `CHECK_RETRY_INTERVAL` between attempts forever unless cancelled.
    /// Mirrors `polyc_control_plane`'s `lease.rs::fakes::fake_client`
    /// `tower::service_fn` pattern (this crate's own precedent: see
    /// `reconcile.rs`'s `fake_conversations`, `routine_reconcile/tests.rs`'s
    /// `fake_routines`) — no reusable kube-API mock exists in this crate.
    fn always_erroring_client() -> kube::Client {
        let svc = tower::service_fn(|_req: http::Request<kube::client::Body>| async move {
            let payload = serde_json::to_vec(&serde_json::json!({
                "apiVersion": "v1",
                "kind": "Status",
                "status": "Failure",
                "reason": "InternalError",
                "code": 500,
            }))
            .unwrap();
            let resp = http::Response::builder()
                .status(StatusCode::INTERNAL_SERVER_ERROR)
                .header("content-type", "application/json")
                .body(kube::client::Body::from(payload))
                .expect("build fake response");
            Ok::<_, std::convert::Infallible>(resp)
        });
        kube::Client::new(svc, "test-ns")
    }

    /// A fake `kube::Client` whose list call always succeeds with zero items
    /// — models a provisioned, reachable apiserver.
    fn always_listable_client() -> kube::Client {
        let svc = tower::service_fn(|_req: http::Request<kube::client::Body>| async move {
            let payload = serde_json::to_vec(&serde_json::json!({
                "apiVersion": "v1",
                "kind": "ConfigMapList",
                "items": [],
            }))
            .unwrap();
            let resp = http::Response::builder()
                .status(StatusCode::OK)
                .header("content-type", "application/json")
                .body(kube::client::Body::from(payload))
                .expect("build fake response");
            Ok::<_, std::convert::Infallible>(resp)
        });
        kube::Client::new(svc, "test-ns")
    }

    /// A fake `kube::Client` whose list call succeeds at the HTTP level but
    /// returns a body that cannot deserialize into the requested type —
    /// models the #1672 failure: the installed CRD's schema (here, a
    /// `ConfigMapList` item shape that doesn't match what `k8s_openapi`'s
    /// `ConfigMap` expects) has skewed against what this binary expects.
    fn schema_skewed_client() -> kube::Client {
        let svc = tower::service_fn(|_req: http::Request<kube::client::Body>| async move {
            // `items` holds a value `ConfigMap` cannot deserialize into
            // (a bare string where an object is required) — the same shape
            // of failure #1672 hit when a `RoutineSchedule` string variant
            // the binary no longer recognized came back from an
            // out-of-date CRD.
            let payload = serde_json::to_vec(&serde_json::json!({
                "apiVersion": "v1",
                "kind": "ConfigMapList",
                "items": ["not an object"],
            }))
            .unwrap();
            let resp = http::Response::builder()
                .status(StatusCode::OK)
                .header("content-type", "application/json")
                .body(kube::client::Body::from(payload))
                .expect("build fake response");
            Ok::<_, std::convert::Infallible>(resp)
        });
        kube::Client::new(svc, "test-ns")
    }

    // Review finding 3: the in-process routine scheduler's preflight must
    // never wedge shutdown on a missing CRD/unreachable apiserver.
    #[tokio::test(start_paused = true)]
    async fn cancellation_during_the_retry_sleep_returns_false_promptly() {
        let api: Api<ConfigMap> = Api::namespaced(always_erroring_client(), "test-ns");
        let shutdown = CancellationToken::new();

        let shutdown_task = shutdown.clone();
        let task = tokio::spawn(async move {
            await_watchable_cancellable(&api, "test-ns", "ConfigMap", &shutdown_task).await
        });

        // Let the first (erroring) list attempt settle into its 5-minute
        // retry sleep before cancelling.
        tokio::time::sleep(Duration::from_millis(10)).await;
        shutdown.cancel();

        let watchable = tokio::time::timeout(Duration::from_secs(5), task)
            .await
            .expect("cancellation must return promptly, never wait out the 5-minute retry sleep")
            .expect("task did not panic");
        assert!(
            !watchable,
            "shutdown fired before the watch ever became serveable"
        );
    }

    #[tokio::test]
    async fn a_serveable_watch_returns_true_without_ever_sleeping() {
        let api: Api<ConfigMap> = Api::namespaced(always_listable_client(), "test-ns");
        let shutdown = CancellationToken::new();

        let watchable = await_watchable_cancellable(&api, "test-ns", "ConfigMap", &shutdown).await;
        assert!(watchable, "a successful list means the watch is serveable");
    }

    // #1672: cancellation must also unwind promptly while stuck on a
    // permanent decode skew, not just a permanent HTTP failure — a skewed
    // CRD never self-heals, so this loop retries forever without shutdown.
    #[tokio::test(start_paused = true)]
    async fn cancellation_during_a_schema_skew_retry_sleep_returns_false_promptly() {
        let api: Api<ConfigMap> = Api::namespaced(schema_skewed_client(), "test-ns");
        let shutdown = CancellationToken::new();

        let shutdown_task = shutdown.clone();
        let task = tokio::spawn(async move {
            await_watchable_cancellable(&api, "test-ns", "ConfigMap", &shutdown_task).await
        });

        tokio::time::sleep(Duration::from_millis(10)).await;
        shutdown.cancel();

        let watchable = tokio::time::timeout(Duration::from_secs(5), task)
            .await
            .expect("cancellation must return promptly even mid schema-skew retry")
            .expect("task did not panic");
        assert!(
            !watchable,
            "shutdown fired before a skewed watch ever became serveable"
        );
    }

    #[test]
    fn a_body_decode_failure_classifies_as_schema_skew() {
        let json_err = serde_json::from_str::<serde_json::Value>("not json").unwrap_err();
        let err = kube::Error::SerdeError(json_err);

        assert_eq!(classify_watch_error(&err), WatchErrorClass::SchemaSkew);
    }

    #[test]
    fn an_api_error_response_classifies_as_namespace_or_rbac_unready() {
        let status = kube::core::Status {
            code: 403,
            reason: "Forbidden".to_owned(),
            ..Default::default()
        };
        let err = kube::Error::Api(Box::new(status));

        assert_eq!(
            classify_watch_error(&err),
            WatchErrorClass::NamespaceOrRbacUnready
        );
    }
}