optimise use of inv queries; tighten hpa watch predicate

release-notes/unreleased/185-reduce-introspection-query-load.md

@@ -0,0 +1,45 @@
+# Reduce RestateDeployment introspection query load
+
+## Bug Fix
+
+### What Changed
+
+The RestateDeployment controller no longer issues a Restate introspection query
+(`POST /query` over `sys_invocation_status`) on every reconcile, and no longer reconciles on
+every HorizontalPodAutoscaler status update. Two changes:
+
+- **Owned HPAs are watched for spec changes only.** The Kubernetes HPA controller re-writes an
+  HPA's status on a ~15s timer; with per-version autoscaling that is one such write per version
+  (including every draining one), and each previously woke the owning RestateDeployment. The
+  watch now filters to generation (spec) changes, so those heartbeats no longer trigger
+  reconciles. The operator's HPA cache still sees every update.
+- **The deployment-usage query runs only when it is needed.** The query answers two questions:
+  is this version the one Restate routes new invocations to (registration), and have older
+  versions drained (cleanup). In the common case — this version already latest, with no older
+  version to drain — the operator determines "already latest" from the cheaper `GET /services`
+  and skips the invocation-status query entirely. Registration, promotion, rollback,
+  foreign-takeover, and drain paths run the full query exactly as before.
+
+### Why This Matters
+
+With per-version autoscaling enabled, a deployment carrying several draining versions produced
+a steady stream of reconciles — roughly one HPA heartbeat every 15 seconds per version — each
+issuing an invocation-status query. Across many deployments sharing one Restate admin endpoint
+this multiplied into sustained query load that could overwhelm Restate's query engine,
+surfacing as `500 Internal Server Error` ("No such scanner") responses from `/query`.
+
+### Impact on Users
+
+- No configuration change, and no change to registration, rollback, or drain behaviour.
+- Restate admin/query load from the operator drops sharply in the steady state: introspection
+  queries are issued during rollouts and drains, not continuously.
+- HorizontalPodAutoscaler status changes no longer trigger RestateDeployment reconciles; spec
+  changes and (re)creation still do.
+
+### Migration Guidance
+
+None. No CRD, Helm value, or CLI flag changes.
+
+### Related Issues
+
+- #185: reducing expensive RestateDeployment usage-query load (also approached by #186 and #188).

src/controllers/restatedeployment/controller.rs

@@ -465,13 +465,29 @@ impl RestateDeployment {
         };
 
         // this path only runs for a live RestateDeployment; deletion goes to `cleanup`.
-        let mut deployments = self.list_deployments(&ctx, CleanupMode::Rollout).await?;
-
         let existing_deployment_id = replicaset
             .annotations()
             .get(RESTATE_DEPLOYMENT_ID_ANNOTATION)
             .cloned();
 
+        // Optimisation: only run the expensive invocation-status query when we really need it,
+        // not in the common case where this version is already latest with nothing to drain.
+        if !registration::has_other_owned(
+            &ctx.replicasets_store,
+            namespace,
+            &my_uid,
+            &versioned_name,
+        ) && let Some(recorded_id) = existing_deployment_id.as_deref()
+        {
+            let latest_ids =
+                registration::latest_deployment_ids(&ctx, &self.spec.restate.register).await?;
+            if latest_ids.contains(recorded_id) {
+                return Ok((replicaset, None));
+            }
+        }
+
+        let mut deployments = self.list_deployments(&ctx, CleanupMode::Rollout).await?;
+
         let action = registration::plan_registration(
             existing_deployment_id.as_deref(),
             &deployments,
@@ -1386,7 +1402,10 @@ pub async fn run(client: Client, metrics: Metrics, state: State) {
     let hpa_reflector =
         kube::runtime::reflector(hpa_writer, kube::runtime::watcher(hpas, cfg.clone()))
             .touched_objects()
-            .default_backoff();
+            .default_backoff()
+            // Wake the owner on a managed HPA's spec change, not the controller's frequent
+            // status heartbeats. After the reflector write, so `hpa_store` still sees every update.
+            .predicate_filter(predicates::generation);
 
     let (rce_store, rce_writer) = kube::runtime::reflector::store();
     let rce_reflector = kube::runtime::reflector(

src/controllers/restatedeployment/registration.rs

@@ -158,6 +158,35 @@ where
         .collect()
 }
 
+/// Whether this RestateDeployment owns any versioned object other than `except_name` in the
+/// namespace — i.e. whether a previous version exists that cleanup might still need to drain.
+/// Read from the already-synced reflector cache, so it costs no admin call and mirrors the
+/// filter [`cleanup_old_replicasets`](super::reconcilers::replicaset::cleanup_old_replicasets)
+/// applies when deciding what to drain.
+pub(super) fn has_other_owned<K>(
+    store: &Store<K>,
+    namespace: &str,
+    rsd_uid: &str,
+    except_name: &str,
+) -> bool
+where
+    K: Resource + Clone + 'static,
+    K::DynamicType: std::hash::Hash + Eq + Clone + Default,
+{
+    store.state().into_iter().any(|obj| {
+        let obj_namespace = match obj.meta().namespace.as_deref() {
+            Some("") | None => "default",
+            Some(ns) => ns,
+        };
+        obj_namespace == namespace
+            && obj.name_any() != except_name
+            && obj.owner_references().iter().any(|reference| {
+                reference.uid == rsd_uid
+                    && reference.kind == <RestateDeployment as Resource>::kind(&())
+            })
+    })
+}
+
 /// A deployment as Restate returned it from registration.
 #[derive(Debug, Clone)]
 pub(super) struct RegisteredDeployment {
@@ -318,6 +347,22 @@ async fn latest_deployment_by_service(
         .collect())
 }
 
+/// The set of deployment ids Restate currently routes at least one service to.
+///
+/// This is the same `latest_for_service` fact the usage query carries, but read from
+/// `GET /services` — a metadata call that never scans `sys_invocation_status`. It answers "is
+/// my recorded version still the one taking new invocations?" cheaply enough to run on the hot
+/// reconcile path, so the steady state can decide it is already latest without the usage query.
+pub(super) async fn latest_deployment_ids(
+    ctx: &Context,
+    endpoint: &RestateAdminEndpoint,
+) -> Result<BTreeSet<String>> {
+    Ok(latest_deployment_by_service(ctx, endpoint)
+        .await?
+        .into_values()
+        .collect())
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -478,4 +523,77 @@ mod tests {
         assert!(!Overwrite::No.force());
         assert!(Overwrite::Yes.force());
     }
+
+    mod has_other_owned {
+        use super::super::has_other_owned;
+        use k8s_openapi::api::apps::v1::ReplicaSet;
+        use k8s_openapi::apimachinery::pkg::apis::meta::v1::{ObjectMeta, OwnerReference};
+        use kube::runtime::{reflector, watcher};
+
+        const UID: &str = "rd-uid";
+        const NS: &str = "app";
+
+        fn replica_set(name: &str, namespace: &str, owner_uid: Option<&str>) -> ReplicaSet {
+            ReplicaSet {
+                metadata: ObjectMeta {
+                    name: Some(name.into()),
+                    namespace: Some(namespace.into()),
+                    owner_references: owner_uid.map(|uid| {
+                        vec![OwnerReference {
+                            kind: "RestateDeployment".into(),
+                            name: "rd".into(),
+                            uid: uid.into(),
+                            api_version: "restate.dev/v1beta1".into(),
+                            ..Default::default()
+                        }]
+                    }),
+                    ..Default::default()
+                },
+                ..Default::default()
+            }
+        }
+
+        // The writer is returned boxed and must be kept alive: the Store reads through the
+        // shared handle the writer owns.
+        fn store_of(
+            objects: Vec<ReplicaSet>,
+        ) -> (reflector::Store<ReplicaSet>, Box<dyn std::any::Any>) {
+            let (reader, mut writer) = reflector::store::<ReplicaSet>();
+            writer.apply_watcher_event(&watcher::Event::Init);
+            for object in objects {
+                writer.apply_watcher_event(&watcher::Event::InitApply(object));
+            }
+            writer.apply_watcher_event(&watcher::Event::InitDone);
+            (reader, Box::new(writer))
+        }
+
+        #[test]
+        fn a_lone_version_has_no_others() {
+            let (store, _writer) = store_of(vec![replica_set("rd-current", NS, Some(UID))]);
+            assert!(!has_other_owned(&store, NS, UID, "rd-current"));
+        }
+
+        #[test]
+        fn an_older_owned_version_counts() {
+            let (store, _writer) = store_of(vec![
+                replica_set("rd-current", NS, Some(UID)),
+                replica_set("rd-old", NS, Some(UID)),
+            ]);
+            assert!(has_other_owned(&store, NS, UID, "rd-current"));
+        }
+
+        #[test]
+        fn foreign_other_namespace_and_orphaned_replicasets_do_not_count() {
+            let (store, _writer) = store_of(vec![
+                replica_set("rd-current", NS, Some(UID)),
+                // ours, but in another namespace
+                replica_set("rd-old", "other-ns", Some(UID)),
+                // same namespace, owned by a different RestateDeployment
+                replica_set("foreign", NS, Some("other-uid")),
+                // same namespace, no controller at all
+                replica_set("orphan", NS, None),
+            ]);
+            assert!(!has_other_owned(&store, NS, UID, "rd-current"));
+        }
+    }
 }