feat(api): Improve VPA handling

crates/etl-api/README.md

@@ -224,27 +224,23 @@ fall back to it. Deployments that start with VPA `Off` may enable updates after
 their chosen observation policy has collected representative usage. Deployments
 that want immediate actuation may configure `in_place_or_recreate` instead.
 
-Before restarting a running replicator, the API uses durable table state to
-predict whether the restart will repeat an initial table copy. If any table is
-before `SyncDone` and is not stopped in an error state, the API deletes the VPA
-before reconciling the pipeline. Reconciliation recreates the VPA from the
-current pipeline overrides, autoscaling configuration, and initial update mode.
-Deleting the VPA resource does not guarantee that a running upstream recommender
-forgets its in-memory usage aggregates. `SyncDone` and `Ready` tables keep their
-destination data across restart and therefore preserve the existing VPA and its
-live update mode. This applies to explicit restarts and configuration updates
-that restart a running replicator. Source connection, query, or state-decoding
-failures preserve the VPA rather than making the restart less reliable.
-
-Kubelet container restarts and Kubernetes- or controller-initiated Pod
-replacements do not pass through the API restart path, so they do not delete the
-VPA. A container restarted within the same Pod keeps that Pod's current
-resources. A replacement Pod is created from the StatefulSet template, but VPA
-admission may replace those resources with the existing recommendation. If a
-replacement happens while table copy will repeat, the API cannot reset the VPA
-to the initial configuration first. This is a known limitation of keeping the
-copy-aware decision at the API boundary. A future Kubernetes controller with
-access to durable table state could own this lifecycle.
+Before restarting a running replicator, the API checks current publication
+membership and durable table state in your database. If any published table
+needs initial sync, including a newly published table, it deletes the VPA.
+Reconciliation restores the configured bounds and initial update mode. This
+covers the table sync phase, including when copying existing rows is skipped.
+`SyncDone`, `Ready`, and errored tables do not trigger a reset. Inspection
+failures preserve the VPA and allow restart. This applies to explicit restarts
+and configuration updates that restart a running replicator.
+
+This is a best-effort reset, not a guarantee that memory stays at the startup
+allocation throughout initial sync. State or publication membership can change
+after inspection. Internal pipeline retries that start new table syncs,
+container restarts, and Kubernetes Pod replacements bypass this API check and
+do not reset the VPA, even during initial sync. The current Pod retains its
+resources; a replacement may receive an existing VPA recommendation. Kubernetes
+and the VPA's live policy govern those allocations. Deleting the VPA also does
+not guarantee that the recommender forgets usage history.
 
 Stopping and starting a pipeline still deletes the autoscaler resource: stop
 deletes the StatefulSet and VPA, and start always recreates both resources.

crates/etl-api/src/data/pipelines.rs

@@ -1,10 +1,14 @@
 use std::ops::DerefMut;
 
+use etl::store::TableState;
 use etl_postgres::{
+    publications::publication_table_ids_query,
     slots,
     store::{destination_table_metadata, health, schema, table_state},
 };
-use sqlx::{FromRow, PgConnection, PgExecutor, PgPool, PgTransaction};
+use sqlx::{
+    AssertSqlSafe, FromRow, PgConnection, PgExecutor, PgPool, PgTransaction, postgres::types::Oid,
+};
 use thiserror::Error;
 
 use crate::{
@@ -33,6 +37,59 @@ use crate::{
 /// only one pipeline will use it.
 pub const MAX_PIPELINES_PER_TENANT: i64 = 1;
 
+/// Returns published tables that would perform initial sync on restart.
+///
+/// Expands publication membership using the replicator's schema and partition
+/// rules, then joins current state for this pipeline. New tables start in
+/// [`TableState::Init`]. Copy selection does not affect this decision: skipping
+/// existing rows still requires table sync.
+///
+/// The publication must come from the configuration used for the replacement.
+/// This observation does not lock membership or replication progress; either
+/// can change before the replicator starts.
+pub(crate) async fn read_pipeline_tables_to_sync(
+    pool: &PgPool,
+    pipeline_id: i64,
+    publication_name: &str,
+) -> Result<Vec<u32>, PipelineError> {
+    let query = format!(
+        r#"
+        with publication_tables as ({})
+        select publication_table.oid, state.id, state.metadata
+        from publication_tables publication_table
+        left join etl.replication_state state
+            on state.table_id = publication_table.oid
+            and state.pipeline_id = $1
+            and state.is_current = true
+        "#,
+        publication_table_ids_query(publication_name),
+    );
+    let rows =
+        sqlx::query_as::<_, (Oid, Option<i64>, Option<serde_json::Value>)>(AssertSqlSafe(query))
+            .bind(pipeline_id)
+            .fetch_all(pool)
+            .await?;
+
+    let mut tables_to_sync = Vec::new();
+    for (table_id, state_id, metadata) in rows {
+        let state = if state_id.is_some() {
+            let metadata = metadata.ok_or(PipelineError::MissingTableState)?;
+            serde_json::from_value::<TableState>(metadata)
+                .map_err(PipelineError::InvalidTableState)?
+        } else {
+            // A newly published table has no stored state yet. Treat it as Init so
+            // its first initial sync triggers a VPA reset before the worker starts.
+            TableState::Init
+        };
+
+        if state.as_type().would_perform_table_sync() {
+            tables_to_sync.push(table_id.0);
+        }
+    }
+
+    Ok(tables_to_sync)
+}
+
 #[derive(Debug, Clone)]
 pub struct Pipeline {
     pub id: i64,