Fixed clickhouse password config update propagation

src/backfill/backfill_staging.rs

@@ -17,14 +17,13 @@
 //! differs) from "already copied back" (staging name gone).
 
 use std::sync::Arc;
-use std::time::Duration;
 
 use anyhow::{Context, Result, bail};
 use clickhouse_c::{Block, Event};
 
 use crate::backfill::backfill_types::BackupRequest;
 use crate::ch::{ChConn, EmitterError, exec_drain, quote_ident, with_timeout};
-use crate::emit::ch_emitter::{EmitterConfig, RetryConfig};
+use crate::config::DestEmitter;
 use crate::mapping::{MappingHandle, TableMapping, TableTarget};
 use crate::schema::RelName;
 use ahash::{HashMap, HashMapExt, HashSet};
@@ -79,12 +78,12 @@ pub struct StagingPlan {
 /// `reuse` keeps the existing tables: a resumed pass is vouched for by rows
 /// already in them, so rebuilding would throw its own progress away
 pub async fn prepare(
-    emitter: Arc<EmitterConfig>,
+    dest: Arc<DestEmitter>,
     live: &MappingHandle,
     reqs: &[BackupRequest],
     reuse: bool,
 ) -> Result<StagingPlan> {
-    let mut sess = StagingSession::connect(emitter).await?;
+    let mut sess = StagingSession::connect(dest).await?;
     // Freeze routing for entire staging plan
     let live_map = live.snapshot().await;
     let mut staged: HashMap<RelName, TableMapping> = HashMap::with_capacity(reqs.len());
@@ -129,25 +128,18 @@ pub async fn prepare(
 /// inserter pool's bounded per-attempt timeout.
 pub struct StagingSession {
     client: ChConn,
-    /// Kept whole for reconnect; shared with the pass that opened the session
-    conn: Arc<EmitterConfig>,
     /// Per-relation destination rules, for the promote's `_lsn` predicate when
     /// a `[table.*]` block or `config_table` row renamed that column
     rules: Option<Arc<crate::table_rules::TableRules>>,
-    retry: RetryConfig,
-    timeout: Duration,
 }
 
 impl StagingSession {
-    pub async fn connect(emitter: Arc<EmitterConfig>) -> Result<Self> {
-        let client = ChConn::connect(&*emitter)
+    pub async fn connect(dest: Arc<DestEmitter>) -> Result<Self> {
+        let client = ChConn::connect(dest)
             .await
             .map_err(|e| anyhow::anyhow!("backfill_staging: connect: {e}"))?;
         Ok(Self {
             client,
-            retry: emitter.retry.clone(),
-            timeout: emitter.insert_timeout,
-            conn: emitter,
             rules: None,
         })
     }
@@ -162,27 +154,25 @@ impl StagingSession {
         match &self.rules {
             Some(rules) => rules
                 .settings(rel)
-                .system_columns(&self.conn.system_columns)
+                .system_columns(&self.client.config().system_columns)
                 .lsn
                 .clone(),
-            None => self.conn.system_columns.lsn.clone(),
+            None => self.client.config().system_columns.lsn.clone(),
         }
     }
 
     async fn attempt_write(&mut self, sql: &str) -> Result<(), EmitterError> {
-        let timeout = self.timeout;
-        let client = self.client.ready(&*self.conn).await?;
+        let timeout = self.client.config().insert_timeout;
+        let client = self.client.ready().await?;
         exec_drain(client, sql, timeout).await
     }
 
     /// Statement safe to re-apply (DROP/CREATE IF NOT EXISTS, dedup-absorbed
     /// INSERT..SELECT): reconnect + resend on retryable failure.
     pub(crate) async fn exec_retry(&mut self, sql: &str) -> Result<()> {
-        let timeout = self.timeout;
+        let timeout = self.client.config().insert_timeout;
         self.client
             .retry(
-                &*self.conn,
-                self.retry.backoff(),
                 |mut client| async move {
                     let result = exec_drain(&mut client, sql, timeout).await;
                     (client, result)
@@ -209,10 +199,10 @@ impl StagingSession {
 
     /// Single-column String SELECT, one attempt under the timeout.
     pub(crate) async fn query_strings(&mut self, sql: &str) -> Result<Vec<String>> {
-        let timeout = self.timeout;
+        let timeout = self.client.config().insert_timeout;
         let client = self
             .client
-            .ready(&*self.conn)
+            .ready()
             .await
             .map_err(|e| anyhow::anyhow!("backfill_staging: {sql}: {e}"))?;
         with_timeout(timeout, async {
@@ -383,7 +373,9 @@ mod tests {
         for retries in 0..=2 {
             let (config, server) =
                 crate::ch::test_support::retry_server(retries, sql, false, false).await;
-            let mut session = StagingSession::connect(Arc::new(config)).await.unwrap();
+            let mut session = StagingSession::connect(DestEmitter::new(Arc::new(config), None))
+                .await
+                .unwrap();
             assert_eq!(session.exec_retry(sql).await.is_ok(), retries == 2);
             server.await.unwrap();
         }
@@ -393,7 +385,9 @@ mod tests {
     async fn exchange_ambiguity_does_not_retry() {
         let sql = "EXCHANGE TABLES staging AND live";
         let (config, server) = crate::ch::test_support::retry_server(2, sql, false, false).await;
-        let mut session = StagingSession::connect(Arc::new(config)).await.unwrap();
+        let mut session = StagingSession::connect(DestEmitter::new(Arc::new(config), None))
+            .await
+            .unwrap();
         assert!(session.exec_once(sql).await.is_err());
         server.abort();
         assert!(server.await.unwrap_err().is_cancelled());

src/backfill/backfill_types.rs

@@ -9,7 +9,7 @@ use crate::budget::MemoryBudget;
 use crate::catalog::desc_log::DescriptorLog;
 use crate::catalog::shadow_catalog::ShadowCatalog;
 use crate::config::ResolvedConfig;
-use crate::emit::ch_emitter::{EmitterConfig, EmitterStats};
+use crate::emit::ch_emitter::EmitterStats;
 use crate::mapping::MappingHandle;
 use crate::ops::oracle::Oracle;
 use crate::schema::RelDescriptor;
@@ -23,7 +23,7 @@ pub struct BackupRequest {
 
 pub struct PassContext {
     pub pg: PgConfig,
-    pub emitter: Arc<EmitterConfig>,
+    pub dest: Arc<crate::config::DestEmitter>,
     /// Routing for this pass's rows; staging targets while a pass is
     /// unpublished
     pub mapping: MappingHandle,

src/backfill/backup_backfill.rs

@@ -158,8 +158,8 @@ async fn run_object_store_pass(ctx: &PassContext, reqs: &[BackupRequest]) -> Res
     // Archive from the `[backup]` config, never the source-PG overlay:
     // credentials in a source table is the wrong trust direction
     // (architecture/bootstrap.md)
-    let settings = ctx
-        .emitter
+    let emitter = ctx.dest.current();
+    let settings = emitter
         .backup
         .as_ref()
         .context("backup_backfill: object_store initial_load requires a [backup] section")?;
@@ -234,7 +234,8 @@ async fn run_object_store_pass(ctx: &PassContext, reqs: &[BackupRequest]) -> Res
             ctx.scratch_dir.clone(),
         )
         .with_parallelism(
-            ctx.emitter
+            ctx.dest
+                .current()
                 .bootstrap
                 .object_store_parallelism
                 .map_or(8, |n| n.get()),
@@ -306,7 +307,7 @@ async fn walk_and_ship(
 
     // Use live pipeline's store, `from_config` always selects ClickHouse mirror
     let mut resolver = ToastResolver::for_mode(
-        &ctx.emitter,
+        ctx.dest.clone(),
         ctx.stats.clone(),
         ctx.oracle
             .as_deref()
@@ -319,8 +320,8 @@ async fn walk_and_ship(
     // Dedicated tail: own connections, own seq space, own fatal — the
     // live pipeline never blocks on a backfill (Regime A)
     let tail = OwnedTail::spawn(
-        &ctx.emitter,
-        ctx.emitter.inserter_pool_size.clamp(1, 3),
+        ctx.dest.clone(),
+        ctx.dest.current().inserter_pool_size.clamp(1, 3),
         ctx.stats.clone(),
         Fatal::new(),
         ctx.config_rx.clone(),
@@ -435,7 +436,7 @@ async fn walk_and_ship(
                 &tail.ack,
                 &ctx.stats,
                 &resolver,
-                &ctx.emitter.row_policy(),
+                &ctx.dest.current().row_policy(),
                 config.as_deref(),
                 next_seq,
                 replay_checkpoint,
@@ -492,7 +493,7 @@ async fn walk_and_ship(
     outcome.pending_tables = visibility_pending::ship(
         pending,
         &ctx.published.snapshot().await,
-        ctx.emitter.clone(),
+        ctx.dest.clone(),
         ctx.stats.clone(),
         resolver,
         ctx.config_rx.as_ref().map(|rx| rx.borrow().clone()),
@@ -622,7 +623,7 @@ async fn run_walk(
         ctx.stats.clone(),
         resolver.clone(),
         bootstrap::Deferral::Handback(toast_spool),
-        ctx.emitter.row_policy(),
+        ctx.dest.current().row_policy(),
         ctx.config_rx.as_ref().map(|rx| rx.borrow().clone()),
         HashSet::new(),
         barrier.clone(),
@@ -893,10 +894,10 @@ async fn replay_gap(
         mapping: ctx.mapping.snapshot().await,
         stats: ctx.stats.clone(),
         budget: ctx.budget.clone(),
-        row_policy: ctx.emitter.row_policy(),
+        row_policy: ctx.dest.current().row_policy(),
         config: ctx.config_rx.as_ref().map(|rx| rx.borrow().clone()),
-        batch_rows: ctx.emitter.drain_batch_rows,
-        batch_bytes: ctx.emitter.drain_batch_bytes,
+        batch_rows: ctx.dest.current().drain_batch_rows,
+        batch_bytes: ctx.dest.current().drain_batch_bytes,
         msg_tx: tail.msg_tx.clone(),
         ack: tail.ack.clone(),
         next_seq,

src/backfill/bootstrap_window.rs

@@ -21,7 +21,7 @@ use crate::backfill::wal_replay::{
 use crate::catalog::desc_log::{BatchRecord, DescLogIdentity, DescriptorLog, LogEntry, LogValue};
 use crate::config::ResolvedConfig;
 use crate::decode::visibility::PgXactPatch;
-use crate::emit::ch_emitter::{EmitterConfig, EmitterStats};
+use crate::emit::ch_emitter::EmitterStats;
 use crate::emit::pipeline::Fatal;
 use crate::emit::pipeline::tail::OwnedTail;
 use crate::mapping::MappingHandle;
@@ -39,7 +39,7 @@ const STOP_POLL: Duration = Duration::from_millis(100);
 /// Inputs shared with concurrent bootstrap drain
 #[derive(Clone)]
 pub struct WindowLegConfig {
-    pub emitter: EmitterConfig,
+    pub dest: Arc<crate::config::DestEmitter>,
     pub mapping: MappingHandle,
     pub config: Arc<ResolvedConfig>,
     /// Shared emitter counters
@@ -192,8 +192,8 @@ impl Leg {
 
         let (filter_rfns, targets) = replay_scope(&cfg.catalog);
         let tail = OwnedTail::spawn(
-            &cfg.emitter,
-            cfg.emitter.inserter_pool_size,
+            cfg.dest.clone(),
+            cfg.dest.current().inserter_pool_size,
             cfg.stats.clone(),
             cfg.fatal.clone(),
             None,
@@ -215,10 +215,10 @@ impl Leg {
             mapping: cfg.mapping.snapshot().await,
             stats: cfg.stats.clone(),
             budget: cfg.resolver.budget().cloned(),
-            row_policy: cfg.emitter.row_policy(),
+            row_policy: cfg.dest.current().row_policy(),
             config: Some(cfg.config.clone()),
-            batch_rows: cfg.emitter.drain_batch_rows,
-            batch_bytes: cfg.emitter.drain_batch_bytes,
+            batch_rows: cfg.dest.current().drain_batch_rows,
+            batch_bytes: cfg.dest.current().drain_batch_bytes,
             msg_tx: tail.msg_tx.clone(),
             ack: tail.ack.clone(),
             next_seq: 0,