From 9be5ab4eeb0590a250b176db89304001f8922a8c Mon Sep 17 00:00:00 2001 From: akselbor Date: Sun, 28 Dec 2025 19:07:17 +0100 Subject: [PATCH 01/14] Add support for TEMPORARY replication slots --- etl-api/src/configs/pipeline.rs | 18 ++++- etl-api/src/k8s/http.rs | 5 +- etl-api/tests/support/mocks.rs | 2 + etl-config/src/shared/pipeline.rs | 12 +++- etl-examples/src/main.rs | 5 +- etl/src/lib.rs | 3 +- etl/src/replication/client.rs | 29 ++++++-- etl/src/replication/table_sync.rs | 4 +- etl/src/test_utils/pipeline.rs | 4 +- etl/src/workers/apply.rs | 16 +++-- etl/tests/replication.rs | 111 ++++++++++++++++++++++++++---- 11 files changed, 176 insertions(+), 33 deletions(-) diff --git a/etl-api/src/configs/pipeline.rs b/etl-api/src/configs/pipeline.rs index 905f2b3e2..612665894 100644 --- a/etl-api/src/configs/pipeline.rs +++ b/etl-api/src/configs/pipeline.rs @@ -1,4 +1,4 @@ -use etl_config::shared::{BatchConfig, PgConnectionConfig, PipelineConfig}; +use etl_config::shared::{BatchConfig, PgConnectionConfig, PipelineConfig, ReplicationSlotConfig}; use serde::{Deserialize, Serialize}; use utoipa::ToSchema; @@ -31,6 +31,7 @@ pub struct FullApiPipelineConfig { #[schema(example = "my_publication")] #[serde(deserialize_with = "crate::utils::trim_string")] pub publication_name: String, + pub temporary_replication_slot: bool, #[serde(skip_serializing_if = "Option::is_none")] pub batch: Option, #[schema(example = 1000)] @@ -49,6 +50,10 @@ impl From for FullApiPipelineConfig { fn from(value: StoredPipelineConfig) -> Self { Self { publication_name: value.publication_name, + temporary_replication_slot: match value.replication_slot { + ReplicationSlotConfig::Temporary => true, + ReplicationSlotConfig::Permanent => false, + }, batch: Some(ApiBatchConfig { max_size: Some(value.batch.max_size), max_fill_ms: Some(value.batch.max_fill_ms), @@ -89,6 +94,7 @@ pub struct PartialApiPipelineConfig { #[derive(Debug, Clone, Serialize, Deserialize)] pub struct StoredPipelineConfig { pub publication_name: String, + pub replication_slot: ReplicationSlotConfig, pub batch: BatchConfig, pub table_error_retry_delay_ms: u64, #[serde(default = "default_table_error_retry_max_attempts")] @@ -106,6 +112,7 @@ impl StoredPipelineConfig { PipelineConfig { id: pipeline_id, publication_name: self.publication_name, + replication_slot: self.replication_slot, pg_connection: pg_connection_config, batch: self.batch, table_error_retry_delay_ms: self.table_error_retry_delay_ms, @@ -161,6 +168,11 @@ impl From for StoredPipelineConfig { Self { publication_name: value.publication_name, + replication_slot: if value.temporary_replication_slot { + ReplicationSlotConfig::Temporary + } else { + ReplicationSlotConfig::Permanent + }, batch, table_error_retry_delay_ms: value .table_error_retry_delay_ms @@ -185,6 +197,7 @@ mod tests { fn test_stored_pipeline_config_serialization() { let config = StoredPipelineConfig { publication_name: "test_publication".to_string(), + replication_slot: ReplicationSlotConfig::Permanent, batch: BatchConfig { max_size: 1000, max_fill_ms: 5000, @@ -223,6 +236,7 @@ mod tests { table_error_retry_max_attempts: None, max_table_sync_workers: None, log_level: Some(LogLevel::Debug), + temporary_replication_slot: false, }; let stored: StoredPipelineConfig = full_config.clone().into(); @@ -235,6 +249,7 @@ mod tests { fn test_full_api_pipeline_config_defaults() { let full_config = FullApiPipelineConfig { publication_name: "test_publication".to_string(), + temporary_replication_slot: false, batch: None, table_error_retry_delay_ms: None, table_error_retry_max_attempts: None, @@ -264,6 +279,7 @@ mod tests { fn test_partial_api_pipeline_config_merge() { let mut stored = StoredPipelineConfig { publication_name: "old_publication".to_string(), + replication_slot: ReplicationSlotConfig::Permanent, batch: BatchConfig { max_size: 500, max_fill_ms: 2000, diff --git a/etl-api/src/k8s/http.rs b/etl-api/src/k8s/http.rs index dd8361756..98462c989 100644 --- a/etl-api/src/k8s/http.rs +++ b/etl-api/src/k8s/http.rs @@ -1052,8 +1052,8 @@ mod tests { use super::*; use etl_config::shared::{ - BatchConfig, DestinationConfig, PgConnectionConfig, PipelineConfig, ReplicatorConfig, - ReplicatorConfigWithoutSecrets, TlsConfig, + BatchConfig, DestinationConfig, PgConnectionConfig, PipelineConfig, ReplicationSlotConfig, + ReplicatorConfig, ReplicatorConfigWithoutSecrets, TlsConfig, }; use insta::assert_json_snapshot; @@ -1139,6 +1139,7 @@ mod tests { pipeline: PipelineConfig { id: 42, publication_name: "all-pub".to_string(), + replication_slot: ReplicationSlotConfig::Permanent, pg_connection: PgConnectionConfig { host: "localhost".to_string(), port: 5432, diff --git a/etl-api/tests/support/mocks.rs b/etl-api/tests/support/mocks.rs index 623933cc3..2ea653627 100644 --- a/etl-api/tests/support/mocks.rs +++ b/etl-api/tests/support/mocks.rs @@ -238,6 +238,7 @@ pub mod pipelines { table_error_retry_max_attempts: Some(5), max_table_sync_workers: Some(2), log_level: Some(LogLevel::Info), + temporary_replication_slot: false, } } @@ -253,6 +254,7 @@ pub mod pipelines { table_error_retry_max_attempts: Some(10), max_table_sync_workers: Some(4), log_level: Some(LogLevel::Info), + temporary_replication_slot: false, } } diff --git a/etl-config/src/shared/pipeline.rs b/etl-config/src/shared/pipeline.rs index 1d23f9ee3..7c7cef4a7 100644 --- a/etl-config/src/shared/pipeline.rs +++ b/etl-config/src/shared/pipeline.rs @@ -3,7 +3,6 @@ use serde::{Deserialize, Serialize}; use crate::shared::{ PgConnectionConfig, PgConnectionConfigWithoutSecrets, ValidationError, batch::BatchConfig, }; - /// Configuration for an ETL pipeline. /// /// Contains all settings required to run a replication pipeline including @@ -20,6 +19,8 @@ pub struct PipelineConfig { pub id: u64, /// Name of the Postgres publication to use for logical replication. pub publication_name: String, + /// Whether to use a temporary replication slot + pub replication_slot: ReplicationSlotConfig, /// The connection configuration for the Postgres instance to which the pipeline connects for /// replication. pub pg_connection: PgConnectionConfig, @@ -64,6 +65,8 @@ pub struct PipelineConfigWithoutSecrets { pub id: u64, /// Name of the Postgres publication to use for logical replication. pub publication_name: String, + /// Whether to use a temporary replication slot + pub replication_slot: ReplicationSlotConfig, /// The connection configuration for the Postgres instance to which the pipeline connects for /// replication. pub pg_connection: PgConnectionConfigWithoutSecrets, @@ -82,6 +85,7 @@ impl From for PipelineConfigWithoutSecrets { PipelineConfigWithoutSecrets { id: value.id, publication_name: value.publication_name, + replication_slot: value.replication_slot, pg_connection: value.pg_connection.into(), batch: value.batch, table_error_retry_delay_ms: value.table_error_retry_delay_ms, @@ -90,3 +94,9 @@ impl From for PipelineConfigWithoutSecrets { } } } + +#[derive(Clone, Copy, Debug, Deserialize, Serialize)] +pub enum ReplicationSlotConfig { + Temporary, + Permanent, +} diff --git a/etl-examples/src/main.rs b/etl-examples/src/main.rs index 5eb0345a9..7f1418e4d 100644 --- a/etl-examples/src/main.rs +++ b/etl-examples/src/main.rs @@ -32,7 +32,9 @@ The pipeline will automatically: */ use clap::{Args, Parser}; -use etl::config::{BatchConfig, PgConnectionConfig, PipelineConfig, TlsConfig}; +use etl::config::{ + BatchConfig, PgConnectionConfig, PipelineConfig, ReplicationSlotConfig, TlsConfig, +}; use etl::pipeline::Pipeline; use etl::store::both::memory::MemoryStore; use etl_destinations::bigquery::BigQueryDestination; @@ -174,6 +176,7 @@ async fn main_impl() -> Result<(), Box> { table_error_retry_delay_ms: 10000, table_error_retry_max_attempts: 5, max_table_sync_workers: args.bq_args.max_table_sync_workers, + replication_slot: ReplicationSlotConfig::Permanent, }; // Initialize BigQuery destination with service account authentication diff --git a/etl/src/lib.rs b/etl/src/lib.rs index 5001eff06..1f166483b 100644 --- a/etl/src/lib.rs +++ b/etl/src/lib.rs @@ -49,7 +49,7 @@ //! //! ```rust,no_run //! use etl::{ -//! config::{BatchConfig, PgConnectionConfig, PipelineConfig, TlsConfig}, +//! config::{BatchConfig, PgConnectionConfig, PipelineConfig, TlsConfig, ReplicationSlotConfig}, //! destination::memory::MemoryDestination, //! pipeline::Pipeline, //! store::both::memory::MemoryStore, @@ -76,6 +76,7 @@ //! let config = PipelineConfig { //! id: 1, //! publication_name: "my_publication".to_string(), +//! replication_slot: ReplicationSlotConfig::Permanent, //! pg_connection: pg_config, //! batch: BatchConfig { max_size: 1000, max_fill_ms: 5000 }, //! table_error_retry_delay_ms: 10000, diff --git a/etl/src/replication/client.rs b/etl/src/replication/client.rs index 7db618685..835b54c14 100644 --- a/etl/src/replication/client.rs +++ b/etl/src/replication/client.rs @@ -1,7 +1,7 @@ use crate::error::{ErrorKind, EtlResult}; use crate::utils::tokio::MakeRustlsConnect; use crate::{bail, etl_error}; -use etl_config::shared::{IntoConnectOptions, PgConnectionConfig}; +use etl_config::shared::{IntoConnectOptions, PgConnectionConfig, ReplicationSlotConfig}; use etl_postgres::replication::extract_server_version; use etl_postgres::types::convert_type_oid_to_type; use etl_postgres::types::{ColumnSchema, TableId, TableName, TableSchema}; @@ -261,18 +261,23 @@ impl PgReplicationClient { pub async fn create_slot_with_transaction( &self, slot_name: &str, + config: ReplicationSlotConfig, ) -> EtlResult<(PgReplicationSlotTransaction, CreateSlotResult)> { // TODO: check if we want to consume the client and return it on commit to avoid any other // operations on a connection that has started a transaction. let transaction = PgReplicationSlotTransaction::new(self.clone()).await?; - let slot = self.create_slot_internal(slot_name, true).await?; + let slot = self.create_slot_internal(slot_name, true, config).await?; Ok((transaction, slot)) } /// Creates a new logical replication slot with the specified name and no snapshot. - pub async fn create_slot(&self, slot_name: &str) -> EtlResult { - self.create_slot_internal(slot_name, false).await + pub async fn create_slot( + &self, + slot_name: &str, + config: ReplicationSlotConfig, + ) -> EtlResult { + self.create_slot_internal(slot_name, false, config).await } /// Gets the slot by `slot_name`. @@ -317,7 +322,11 @@ impl PgReplicationClient { /// - A boolean indicating whether the slot was created (true) or already existed (false) /// - The slot result containing either the confirmed_flush_lsn (for existing slots) /// or the consistent_point (for newly created slots) - pub async fn get_or_create_slot(&self, slot_name: &str) -> EtlResult { + pub async fn get_or_create_slot( + &self, + slot_name: &str, + config: ReplicationSlotConfig, + ) -> EtlResult { match self.get_slot(slot_name).await { Ok(slot) => { info!("using existing replication slot '{}'", slot_name); @@ -327,7 +336,7 @@ impl PgReplicationClient { Err(err) if err.kind() == ErrorKind::ReplicationSlotNotFound => { info!("creating new replication slot '{}'", slot_name); - let create_result = self.create_slot_internal(slot_name, false).await?; + let create_result = self.create_slot_internal(slot_name, false, config).await?; Ok(GetOrCreateSlotResult::CreateSlot(create_result)) } @@ -591,6 +600,7 @@ impl PgReplicationClient { &self, slot_name: &str, use_snapshot: bool, + config: ReplicationSlotConfig, ) -> EtlResult { // Do not convert the query or the options to lowercase, since the lexer for // replication commands (repl_scanner.l) in Postgres code expects the commands @@ -601,9 +611,14 @@ impl PgReplicationClient { } else { "NOEXPORT_SNAPSHOT" }; + let temporary = match config { + ReplicationSlotConfig::Temporary => " TEMPORARY ", + ReplicationSlotConfig::Permanent => " ", + }; let query = format!( - r#"CREATE_REPLICATION_SLOT {} LOGICAL pgoutput {}"#, + r#"CREATE_REPLICATION_SLOT {}{}LOGICAL pgoutput {}"#, quote_identifier(slot_name), + temporary, snapshot_option ); match self.client.simple_query(&query).await { diff --git a/etl/src/replication/table_sync.rs b/etl/src/replication/table_sync.rs index 27d28e635..a72628b44 100644 --- a/etl/src/replication/table_sync.rs +++ b/etl/src/replication/table_sync.rs @@ -1,4 +1,4 @@ -use etl_config::shared::PipelineConfig; +use etl_config::shared::{PipelineConfig, ReplicationSlotConfig}; use etl_postgres::replication::slots::EtlReplicationSlot; use etl_postgres::types::TableId; use futures::StreamExt; @@ -183,7 +183,7 @@ where // If a slot already exists at this point, we could delete it and try to recover, but it means // that the state was somehow reset without the slot being deleted, and we want to surface this. let (transaction, slot) = replication_client - .create_slot_with_transaction(&slot_name) + .create_slot_with_transaction(&slot_name, ReplicationSlotConfig::Permanent) .await?; // We copy the table schema and write it both to the state store and destination. diff --git a/etl/src/test_utils/pipeline.rs b/etl/src/test_utils/pipeline.rs index 045d21a87..f71318b53 100644 --- a/etl/src/test_utils/pipeline.rs +++ b/etl/src/test_utils/pipeline.rs @@ -1,4 +1,4 @@ -use etl_config::shared::{BatchConfig, PgConnectionConfig, PipelineConfig}; +use etl_config::shared::{BatchConfig, PgConnectionConfig, PipelineConfig, ReplicationSlotConfig}; use uuid::Uuid; use crate::destination::Destination; @@ -39,6 +39,7 @@ where table_error_retry_delay_ms: 1000, table_error_retry_max_attempts: 2, max_table_sync_workers: 1, + replication_slot: ReplicationSlotConfig::Permanent, }; Pipeline::new(config, store, destination) @@ -64,6 +65,7 @@ where let config = PipelineConfig { id: pipeline_id, publication_name, + replication_slot: ReplicationSlotConfig::Permanent, pg_connection: pg_connection_config.clone(), batch, table_error_retry_delay_ms: 1000, diff --git a/etl/src/workers/apply.rs b/etl/src/workers/apply.rs index 593d9b810..4e2780552 100644 --- a/etl/src/workers/apply.rs +++ b/etl/src/workers/apply.rs @@ -1,4 +1,4 @@ -use etl_config::shared::PipelineConfig; +use etl_config::shared::{PipelineConfig, ReplicationSlotConfig}; use etl_postgres::replication::slots::EtlReplicationSlot; use etl_postgres::replication::worker::WorkerType; use etl_postgres::types::TableId; @@ -136,8 +136,13 @@ where publication_name = self.config.publication_name ); let apply_worker = async move { - let start_lsn = - get_start_lsn(self.pipeline_id, &self.replication_client, &self.store).await?; + let start_lsn = get_start_lsn( + self.pipeline_id, + &self.replication_client, + &self.store, + self.config.replication_slot, + ) + .await?; // We create the signal used to notify the apply worker that it should force syncing tables. let (force_syncing_tables_tx, force_syncing_tables_rx) = create_signal(); @@ -192,12 +197,15 @@ async fn get_start_lsn( pipeline_id: PipelineId, replication_client: &PgReplicationClient, store: &S, + replication_slot: ReplicationSlotConfig, ) -> EtlResult { let slot_name: String = EtlReplicationSlot::for_apply_worker(pipeline_id).try_into()?; // We try to get or create the slot. Both operations will return an LSN that we can use to start // streaming events. - let slot = replication_client.get_or_create_slot(&slot_name).await?; + let slot = replication_client + .get_or_create_slot(&slot_name, replication_slot) + .await?; // When creating a new apply worker slot, all tables must be in the `Init` state. If any table // is not in Init state, it means the table was synchronized based on another apply worker diff --git a/etl/tests/replication.rs b/etl/tests/replication.rs index 806ca818f..68d413ef9 100644 --- a/etl/tests/replication.rs +++ b/etl/tests/replication.rs @@ -2,8 +2,9 @@ use std::collections::HashSet; +use etl::config::ReplicationSlotConfig; use etl::error::ErrorKind; -use etl::replication::client::PgReplicationClient; +use etl::replication::client::{CreateSlotResult, PgReplicationClient}; use etl::test_utils::database::{spawn_source_database, test_table_name}; use etl::test_utils::pipeline::test_slot_name; use etl::test_utils::table::assert_table_schema; @@ -93,7 +94,10 @@ async fn test_replication_client_creates_slot() { .unwrap(); let slot_name = test_slot_name("my_slot"); - let create_slot = client.create_slot(&slot_name).await.unwrap(); + let create_slot = client + .create_slot(&slot_name, ReplicationSlotConfig::Permanent) + .await + .unwrap(); assert!(!create_slot.consistent_point.to_string().is_empty()); let get_slot = client.get_slot(&slot_name).await.unwrap(); @@ -116,7 +120,10 @@ async fn test_create_and_delete_slot() { let slot_name = test_slot_name("my_slot"); // Create the slot and verify it exists - let create_slot = client.create_slot(&slot_name).await.unwrap(); + let create_slot = client + .create_slot(&slot_name, ReplicationSlotConfig::Permanent) + .await + .unwrap(); assert!(!create_slot.consistent_point.to_string().is_empty()); let get_slot = client.get_slot(&slot_name).await.unwrap(); @@ -156,13 +163,88 @@ async fn test_replication_client_doesnt_recreate_slot() { .unwrap(); let slot_name = test_slot_name("my_slot"); - assert!(client.create_slot(&slot_name).await.is_ok()); + assert!( + client + .create_slot(&slot_name, ReplicationSlotConfig::Permanent) + .await + .is_ok() + ); assert!(matches!( - client.create_slot(&slot_name).await, + client.create_slot(&slot_name, ReplicationSlotConfig::Permanent).await, Err(ref err) if err.kind() == ErrorKind::ReplicationSlotAlreadyExists )); } +#[tokio::test(flavor = "multi_thread")] +async fn test_replication_client_temporary_slot_dropped_on_disconnect() { + let database = spawn_source_database().await; + + let client = PgReplicationClient::connect(database.config.clone()) + .await + .unwrap(); + + let slot_name = test_slot_name("my_slot"); + assert!(matches!( + client + .create_slot(&slot_name, ReplicationSlotConfig::Temporary) + .await, + Ok(CreateSlotResult { + consistent_point: _ + }) + )); + + // Dropping the client should close the connection, causing our replication + // slot to be cleaned up + drop(client); + + // This means that we should be able to execute the exact same sequence again, and create the same replication slot again + let client = PgReplicationClient::connect(database.config.clone()) + .await + .unwrap(); + + assert!(matches!( + client + .create_slot(&slot_name, ReplicationSlotConfig::Temporary) + .await, + Ok(CreateSlotResult { + consistent_point: _ + }) + )); +} + +#[tokio::test(flavor = "multi_thread")] +async fn test_replication_client_permanent_slot_persisted_on_disconnect() { + let database = spawn_source_database().await; + + let client = PgReplicationClient::connect(database.config.clone()) + .await + .unwrap(); + + let slot_name = test_slot_name("my_slot"); + assert!(matches!( + client + .create_slot(&slot_name, ReplicationSlotConfig::Permanent) + .await, + Ok(CreateSlotResult { + consistent_point: _ + }) + )); + + // Dropping the client should close the connection, causing our replication + // slot to be cleaned up + drop(client); + + // This means that we should be able to execute the exact same sequence again, and the slot should still exist + let client = PgReplicationClient::connect(database.config.clone()) + .await + .unwrap(); + + assert!(matches!( + client.create_slot(&slot_name, ReplicationSlotConfig::Permanent).await, + Err(ref err) if err.kind() == ErrorKind::ReplicationSlotAlreadyExists, + )); +} + #[tokio::test(flavor = "multi_thread")] async fn test_table_schema_copy_is_consistent() { init_test_tracing(); @@ -187,7 +269,7 @@ async fn test_table_schema_copy_is_consistent() { // We create the slot when the database schema contains only 'table_1'. let (transaction, _) = client - .create_slot_with_transaction(&test_slot_name("my_slot")) + .create_slot_with_transaction(&test_slot_name("my_slot"), ReplicationSlotConfig::Permanent) .await .unwrap(); @@ -239,7 +321,7 @@ async fn test_table_schema_copy_across_multiple_connections() { // We create the slot when the database schema contains only 'table_1'. let (transaction, _) = first_client - .create_slot_with_transaction(&test_slot_name("my_slot")) + .create_slot_with_transaction(&test_slot_name("my_slot"), ReplicationSlotConfig::Permanent) .await .unwrap(); @@ -274,7 +356,7 @@ async fn test_table_schema_copy_across_multiple_connections() { // We create the slot when the database schema contains both 'table_1' and 'table_2'. let (transaction, _) = second_client - .create_slot_with_transaction(&test_slot_name("my_slot")) + .create_slot_with_transaction(&test_slot_name("my_slot"), ReplicationSlotConfig::Permanent) .await .unwrap(); @@ -338,7 +420,7 @@ async fn test_table_copy_stream_is_consistent() { // We create the slot when the database schema contains only 'table_1' data. let (transaction, _) = parent_client - .create_slot_with_transaction(&test_slot_name("my_slot")) + .create_slot_with_transaction(&test_slot_name("my_slot"), ReplicationSlotConfig::Permanent) .await .unwrap(); @@ -414,7 +496,7 @@ async fn test_table_copy_stream_respects_row_filter() { // We create the slot when the database schema contains only 'table_1' data. let (transaction, _) = parent_client - .create_slot_with_transaction(&test_slot_name("my_slot")) + .create_slot_with_transaction(&test_slot_name("my_slot"), ReplicationSlotConfig::Permanent) .await .unwrap(); @@ -501,7 +583,7 @@ async fn test_table_copy_stream_respects_column_filter() { // Create the slot when the database schema contains the test data. let (transaction, _) = parent_client - .create_slot_with_transaction(&test_slot_name("my_slot")) + .create_slot_with_transaction(&test_slot_name("my_slot"), ReplicationSlotConfig::Permanent) .await .unwrap(); @@ -589,7 +671,7 @@ async fn test_table_copy_stream_no_row_filter() { // We create the slot when the database schema contains only 'table_1' data. let (transaction, _) = parent_client - .create_slot_with_transaction(&test_slot_name("my_slot")) + .create_slot_with_transaction(&test_slot_name("my_slot"), ReplicationSlotConfig::Permanent) .await .unwrap(); @@ -716,7 +798,10 @@ async fn test_start_logical_replication() { // We create a slot which is going to replicate data before we insert the data. let slot_name = test_slot_name("my_slot"); - let slot = parent_client.create_slot(&slot_name).await.unwrap(); + let slot = parent_client + .create_slot(&slot_name, ReplicationSlotConfig::Permanent) + .await + .unwrap(); // We create a table with a publication and 10 entries. database From fc583aa11a7895491b55a3d48fb28e5673e29920 Mon Sep 17 00:00:00 2001 From: akselbor Date: Sun, 28 Dec 2025 19:12:12 +0100 Subject: [PATCH 02/14] Minor comment fixes --- etl/tests/replication.rs | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/etl/tests/replication.rs b/etl/tests/replication.rs index 68d413ef9..5477febf0 100644 --- a/etl/tests/replication.rs +++ b/etl/tests/replication.rs @@ -230,11 +230,11 @@ async fn test_replication_client_permanent_slot_persisted_on_disconnect() { }) )); - // Dropping the client should close the connection, causing our replication - // slot to be cleaned up + // Dropping the client should close the connection. Since we use a persisted replication slot + // it should still be present when we start a fresh connection drop(client); - // This means that we should be able to execute the exact same sequence again, and the slot should still exist + // This means that we should be able to execute the exact same sequence again, and the slot should still exist (i.e., we fail to create a new one) let client = PgReplicationClient::connect(database.config.clone()) .await .unwrap(); From 4d04b0a432d7fdd4a89a3ba171c69fb3d71e933a Mon Sep 17 00:00:00 2001 From: akselbor Date: Sun, 28 Dec 2025 22:33:37 +0100 Subject: [PATCH 03/14] Fix PipelineConfig for benches/table_copies.rs --- etl-benchmarks/benches/table_copies.rs | 2 ++ 1 file changed, 2 insertions(+) diff --git a/etl-benchmarks/benches/table_copies.rs b/etl-benchmarks/benches/table_copies.rs index 01503d18d..2a7a2e46f 100644 --- a/etl-benchmarks/benches/table_copies.rs +++ b/etl-benchmarks/benches/table_copies.rs @@ -1,4 +1,5 @@ use clap::{Parser, Subcommand, ValueEnum}; +use etl::config::ReplicationSlotConfig; use etl::destination::Destination; use etl::error::EtlResult; use etl::pipeline::Pipeline; @@ -327,6 +328,7 @@ async fn start_pipeline(args: RunArgs) -> Result<(), Box> { let pipeline_config = PipelineConfig { id: 1, publication_name: args.publication_name, + replication_slot: ReplicationSlotConfig::Permanent, pg_connection: pg_connection_config, batch: BatchConfig { max_size: args.batch_max_size, From 8ffbf215ba65e170103feda5f09ef10843074fe0 Mon Sep 17 00:00:00 2001 From: akselbor Date: Sun, 28 Dec 2025 22:52:00 +0100 Subject: [PATCH 04/14] Update .snaps --- ...pi__k8s__http__tests__create_replicator_config_map_json.snap | 2 +- ...ting_bigquery_destination_and_pipeline_can_be_updated-2.snap | 1 + ...berg_supabase_destination_and_pipeline_can_be_updated-2.snap | 1 + ...nes__bigquery_destination_and_pipeline_can_be_created-2.snap | 1 + ...berg_supabase_destination_and_pipeline_can_be_created-2.snap | 1 + .../tests/snapshots/pipelines__all_pipelines_can_be_read.snap | 1 + .../snapshots/pipelines__an_existing_pipeline_can_be_read.snap | 1 + .../pipelines__an_existing_pipeline_can_be_updated.snap | 1 + .../snapshots/pipelines__pipeline_config_can_be_updated-2.snap | 1 + .../snapshots/pipelines__pipeline_config_can_be_updated-3.snap | 1 + .../snapshots/pipelines__pipeline_config_can_be_updated-4.snap | 1 + .../snapshots/pipelines__pipeline_config_can_be_updated-5.snap | 1 + .../snapshots/pipelines__pipeline_config_can_be_updated.snap | 1 + 13 files changed, 13 insertions(+), 1 deletion(-) diff --git a/etl-api/src/k8s/snapshots/etl_api__k8s__http__tests__create_replicator_config_map_json.snap b/etl-api/src/k8s/snapshots/etl_api__k8s__http__tests__create_replicator_config_map_json.snap index 10a8d4417..c183894e9 100644 --- a/etl-api/src/k8s/snapshots/etl_api__k8s__http__tests__create_replicator_config_map_json.snap +++ b/etl-api/src/k8s/snapshots/etl_api__k8s__http__tests__create_replicator_config_map_json.snap @@ -6,7 +6,7 @@ expression: config_map_json "apiVersion": "v1", "data": { "base.json": "", - "prod.json": "{\"destination\":{\"big_query\":{\"project_id\":\"project-id\",\"dataset_id\":\"dataset-id\",\"max_concurrent_streams\":4}},\"pipeline\":{\"id\":42,\"publication_name\":\"all-pub\",\"pg_connection\":{\"host\":\"localhost\",\"port\":5432,\"name\":\"postgres\",\"username\":\"postgres\",\"tls\":{\"trusted_root_certs\":\"\",\"enabled\":false}},\"batch\":{\"max_size\":10000,\"max_fill_ms\":1000},\"table_error_retry_delay_ms\":500,\"table_error_retry_max_attempts\":3,\"max_table_sync_workers\":4}}" + "prod.json": "{\"destination\":{\"big_query\":{\"project_id\":\"project-id\",\"dataset_id\":\"dataset-id\",\"max_concurrent_streams\":4}},\"pipeline\":{\"id\":42,\"publication_name\":\"all-pub\",\"replication_slot\":\"Permanent\",\"pg_connection\":{\"host\":\"localhost\",\"port\":5432,\"name\":\"postgres\",\"username\":\"postgres\",\"tls\":{\"trusted_root_certs\":\"\",\"enabled\":false}},\"batch\":{\"max_size\":10000,\"max_fill_ms\":1000},\"table_error_retry_delay_ms\":500,\"table_error_retry_max_attempts\":3,\"max_table_sync_workers\":4}}" }, "kind": "ConfigMap", "metadata": { diff --git a/etl-api/tests/snapshots/destinations_pipelines__an_existing_bigquery_destination_and_pipeline_can_be_updated-2.snap b/etl-api/tests/snapshots/destinations_pipelines__an_existing_bigquery_destination_and_pipeline_can_be_updated-2.snap index efb242e54..24f2771f8 100644 --- a/etl-api/tests/snapshots/destinations_pipelines__an_existing_bigquery_destination_and_pipeline_can_be_updated-2.snap +++ b/etl-api/tests/snapshots/destinations_pipelines__an_existing_bigquery_destination_and_pipeline_can_be_updated-2.snap @@ -4,6 +4,7 @@ expression: response.config --- FullApiPipelineConfig { publication_name: "updated_publication", + temporary_replication_slot: false, batch: Some( ApiBatchConfig { max_size: Some( diff --git a/etl-api/tests/snapshots/destinations_pipelines__an_existing_iceberg_supabase_destination_and_pipeline_can_be_updated-2.snap b/etl-api/tests/snapshots/destinations_pipelines__an_existing_iceberg_supabase_destination_and_pipeline_can_be_updated-2.snap index efb242e54..24f2771f8 100644 --- a/etl-api/tests/snapshots/destinations_pipelines__an_existing_iceberg_supabase_destination_and_pipeline_can_be_updated-2.snap +++ b/etl-api/tests/snapshots/destinations_pipelines__an_existing_iceberg_supabase_destination_and_pipeline_can_be_updated-2.snap @@ -4,6 +4,7 @@ expression: response.config --- FullApiPipelineConfig { publication_name: "updated_publication", + temporary_replication_slot: false, batch: Some( ApiBatchConfig { max_size: Some( diff --git a/etl-api/tests/snapshots/destinations_pipelines__bigquery_destination_and_pipeline_can_be_created-2.snap b/etl-api/tests/snapshots/destinations_pipelines__bigquery_destination_and_pipeline_can_be_created-2.snap index 5dc86306c..42dbb80d1 100644 --- a/etl-api/tests/snapshots/destinations_pipelines__bigquery_destination_and_pipeline_can_be_created-2.snap +++ b/etl-api/tests/snapshots/destinations_pipelines__bigquery_destination_and_pipeline_can_be_created-2.snap @@ -4,6 +4,7 @@ expression: response.config --- FullApiPipelineConfig { publication_name: "publication", + temporary_replication_slot: false, batch: Some( ApiBatchConfig { max_size: Some( diff --git a/etl-api/tests/snapshots/destinations_pipelines__iceberg_supabase_destination_and_pipeline_can_be_created-2.snap b/etl-api/tests/snapshots/destinations_pipelines__iceberg_supabase_destination_and_pipeline_can_be_created-2.snap index 5dc86306c..42dbb80d1 100644 --- a/etl-api/tests/snapshots/destinations_pipelines__iceberg_supabase_destination_and_pipeline_can_be_created-2.snap +++ b/etl-api/tests/snapshots/destinations_pipelines__iceberg_supabase_destination_and_pipeline_can_be_created-2.snap @@ -4,6 +4,7 @@ expression: response.config --- FullApiPipelineConfig { publication_name: "publication", + temporary_replication_slot: false, batch: Some( ApiBatchConfig { max_size: Some( diff --git a/etl-api/tests/snapshots/pipelines__all_pipelines_can_be_read.snap b/etl-api/tests/snapshots/pipelines__all_pipelines_can_be_read.snap index aed37eea7..2307c2170 100644 --- a/etl-api/tests/snapshots/pipelines__all_pipelines_can_be_read.snap +++ b/etl-api/tests/snapshots/pipelines__all_pipelines_can_be_read.snap @@ -4,6 +4,7 @@ expression: pipeline.config --- FullApiPipelineConfig { publication_name: "publication", + temporary_replication_slot: false, batch: Some( ApiBatchConfig { max_size: Some( diff --git a/etl-api/tests/snapshots/pipelines__an_existing_pipeline_can_be_read.snap b/etl-api/tests/snapshots/pipelines__an_existing_pipeline_can_be_read.snap index 979216c6e..74c78729d 100644 --- a/etl-api/tests/snapshots/pipelines__an_existing_pipeline_can_be_read.snap +++ b/etl-api/tests/snapshots/pipelines__an_existing_pipeline_can_be_read.snap @@ -4,6 +4,7 @@ expression: response.config --- FullApiPipelineConfig { publication_name: "publication", + temporary_replication_slot: false, batch: Some( ApiBatchConfig { max_size: Some( diff --git a/etl-api/tests/snapshots/pipelines__an_existing_pipeline_can_be_updated.snap b/etl-api/tests/snapshots/pipelines__an_existing_pipeline_can_be_updated.snap index 2db56e867..b41ea8fcb 100644 --- a/etl-api/tests/snapshots/pipelines__an_existing_pipeline_can_be_updated.snap +++ b/etl-api/tests/snapshots/pipelines__an_existing_pipeline_can_be_updated.snap @@ -4,6 +4,7 @@ expression: response.config --- FullApiPipelineConfig { publication_name: "updated_publication", + temporary_replication_slot: false, batch: Some( ApiBatchConfig { max_size: Some( diff --git a/etl-api/tests/snapshots/pipelines__pipeline_config_can_be_updated-2.snap b/etl-api/tests/snapshots/pipelines__pipeline_config_can_be_updated-2.snap index 667e3ea3a..f779e9b37 100644 --- a/etl-api/tests/snapshots/pipelines__pipeline_config_can_be_updated-2.snap +++ b/etl-api/tests/snapshots/pipelines__pipeline_config_can_be_updated-2.snap @@ -4,6 +4,7 @@ expression: response.config --- FullApiPipelineConfig { publication_name: "publication", + temporary_replication_slot: false, batch: Some( ApiBatchConfig { max_size: Some( diff --git a/etl-api/tests/snapshots/pipelines__pipeline_config_can_be_updated-3.snap b/etl-api/tests/snapshots/pipelines__pipeline_config_can_be_updated-3.snap index d565d89fb..2d4073a24 100644 --- a/etl-api/tests/snapshots/pipelines__pipeline_config_can_be_updated-3.snap +++ b/etl-api/tests/snapshots/pipelines__pipeline_config_can_be_updated-3.snap @@ -4,6 +4,7 @@ expression: response.config --- FullApiPipelineConfig { publication_name: "publication", + temporary_replication_slot: false, batch: Some( ApiBatchConfig { max_size: Some( diff --git a/etl-api/tests/snapshots/pipelines__pipeline_config_can_be_updated-4.snap b/etl-api/tests/snapshots/pipelines__pipeline_config_can_be_updated-4.snap index 1a5c71ed0..e7de59663 100644 --- a/etl-api/tests/snapshots/pipelines__pipeline_config_can_be_updated-4.snap +++ b/etl-api/tests/snapshots/pipelines__pipeline_config_can_be_updated-4.snap @@ -4,6 +4,7 @@ expression: response.config --- FullApiPipelineConfig { publication_name: "publication", + temporary_replication_slot: false, batch: Some( ApiBatchConfig { max_size: Some( diff --git a/etl-api/tests/snapshots/pipelines__pipeline_config_can_be_updated-5.snap b/etl-api/tests/snapshots/pipelines__pipeline_config_can_be_updated-5.snap index 7c715f879..f708c2706 100644 --- a/etl-api/tests/snapshots/pipelines__pipeline_config_can_be_updated-5.snap +++ b/etl-api/tests/snapshots/pipelines__pipeline_config_can_be_updated-5.snap @@ -4,6 +4,7 @@ expression: response.config --- FullApiPipelineConfig { publication_name: "publication", + temporary_replication_slot: false, batch: Some( ApiBatchConfig { max_size: Some( diff --git a/etl-api/tests/snapshots/pipelines__pipeline_config_can_be_updated.snap b/etl-api/tests/snapshots/pipelines__pipeline_config_can_be_updated.snap index a20177232..49419c0a1 100644 --- a/etl-api/tests/snapshots/pipelines__pipeline_config_can_be_updated.snap +++ b/etl-api/tests/snapshots/pipelines__pipeline_config_can_be_updated.snap @@ -4,6 +4,7 @@ expression: response.config --- FullApiPipelineConfig { publication_name: "publication", + temporary_replication_slot: false, batch: Some( ApiBatchConfig { max_size: Some( From ee3533bbf68741387bb4f93e2c6362ab98076147 Mon Sep 17 00:00:00 2001 From: Aksel Borgen Date: Sun, 4 Jan 2026 22:47:06 +0100 Subject: [PATCH 05/14] Change ReplicationSlotConfig from enum to struct --- etl-api/src/configs/pipeline.rs | 48 +++++++++++++------ etl-api/src/k8s/http.rs | 2 +- ...create_replicator_config_map_json.snap.new | 21 ++++++++ etl-api/tests/support/mocks.rs | 6 +-- etl-benchmarks/benches/table_copies.rs | 2 +- etl-config/src/shared/pipeline.rs | 12 +++-- etl-examples/src/main.rs | 2 +- etl/src/lib.rs | 2 +- etl/src/replication/client.rs | 6 +-- etl/src/replication/table_sync.rs | 4 +- etl/src/test_utils/pipeline.rs | 4 +- etl/tests/replication.rs | 32 ++++++------- 12 files changed, 94 insertions(+), 47 deletions(-) create mode 100644 etl-api/src/k8s/snapshots/etl_api__k8s__http__tests__create_replicator_config_map_json.snap.new diff --git a/etl-api/src/configs/pipeline.rs b/etl-api/src/configs/pipeline.rs index 612665894..d9edd3790 100644 --- a/etl-api/src/configs/pipeline.rs +++ b/etl-api/src/configs/pipeline.rs @@ -26,12 +26,37 @@ pub struct ApiBatchConfig { pub max_fill_ms: Option, } + +#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)] +pub struct ApiReplicationSlotConfig { + #[schema(example = false)] + pub temporary: bool, +} + +impl Default for ApiReplicationSlotConfig { + fn default() -> Self { + Self { temporary: false } + } +} + +impl From for ApiReplicationSlotConfig { + fn from(value: ReplicationSlotConfig) -> Self { + ApiReplicationSlotConfig { temporary: value.temporary } + } +} + +impl From for ReplicationSlotConfig { + fn from(value: ApiReplicationSlotConfig) -> Self { + ReplicationSlotConfig { temporary: value.temporary } + } +} + #[derive(Debug, Clone, Serialize, Deserialize, ToSchema)] pub struct FullApiPipelineConfig { #[schema(example = "my_publication")] #[serde(deserialize_with = "crate::utils::trim_string")] pub publication_name: String, - pub temporary_replication_slot: bool, + pub replication_slot: ApiReplicationSlotConfig, #[serde(skip_serializing_if = "Option::is_none")] pub batch: Option, #[schema(example = 1000)] @@ -46,14 +71,12 @@ pub struct FullApiPipelineConfig { pub log_level: Option, } + impl From for FullApiPipelineConfig { fn from(value: StoredPipelineConfig) -> Self { Self { publication_name: value.publication_name, - temporary_replication_slot: match value.replication_slot { - ReplicationSlotConfig::Temporary => true, - ReplicationSlotConfig::Permanent => false, - }, + replication_slot: value.replication_slot.into(), batch: Some(ApiBatchConfig { max_size: Some(value.batch.max_size), max_fill_ms: Some(value.batch.max_fill_ms), @@ -94,6 +117,7 @@ pub struct PartialApiPipelineConfig { #[derive(Debug, Clone, Serialize, Deserialize)] pub struct StoredPipelineConfig { pub publication_name: String, + #[serde(default)] pub replication_slot: ReplicationSlotConfig, pub batch: BatchConfig, pub table_error_retry_delay_ms: u64, @@ -168,11 +192,7 @@ impl From for StoredPipelineConfig { Self { publication_name: value.publication_name, - replication_slot: if value.temporary_replication_slot { - ReplicationSlotConfig::Temporary - } else { - ReplicationSlotConfig::Permanent - }, + replication_slot: value.replication_slot.into(), batch, table_error_retry_delay_ms: value .table_error_retry_delay_ms @@ -197,7 +217,7 @@ mod tests { fn test_stored_pipeline_config_serialization() { let config = StoredPipelineConfig { publication_name: "test_publication".to_string(), - replication_slot: ReplicationSlotConfig::Permanent, + replication_slot: ReplicationSlotConfig::default(), batch: BatchConfig { max_size: 1000, max_fill_ms: 5000, @@ -236,7 +256,7 @@ mod tests { table_error_retry_max_attempts: None, max_table_sync_workers: None, log_level: Some(LogLevel::Debug), - temporary_replication_slot: false, + replication_slot: ApiReplicationSlotConfig::default() }; let stored: StoredPipelineConfig = full_config.clone().into(); @@ -249,7 +269,7 @@ mod tests { fn test_full_api_pipeline_config_defaults() { let full_config = FullApiPipelineConfig { publication_name: "test_publication".to_string(), - temporary_replication_slot: false, + replication_slot: ApiReplicationSlotConfig::default(), batch: None, table_error_retry_delay_ms: None, table_error_retry_max_attempts: None, @@ -279,7 +299,7 @@ mod tests { fn test_partial_api_pipeline_config_merge() { let mut stored = StoredPipelineConfig { publication_name: "old_publication".to_string(), - replication_slot: ReplicationSlotConfig::Permanent, + replication_slot: ReplicationSlotConfig::default(), batch: BatchConfig { max_size: 500, max_fill_ms: 2000, diff --git a/etl-api/src/k8s/http.rs b/etl-api/src/k8s/http.rs index 98462c989..309570394 100644 --- a/etl-api/src/k8s/http.rs +++ b/etl-api/src/k8s/http.rs @@ -1139,7 +1139,7 @@ mod tests { pipeline: PipelineConfig { id: 42, publication_name: "all-pub".to_string(), - replication_slot: ReplicationSlotConfig::Permanent, + replication_slot: ReplicationSlotConfig::default(), pg_connection: PgConnectionConfig { host: "localhost".to_string(), port: 5432, diff --git a/etl-api/src/k8s/snapshots/etl_api__k8s__http__tests__create_replicator_config_map_json.snap.new b/etl-api/src/k8s/snapshots/etl_api__k8s__http__tests__create_replicator_config_map_json.snap.new new file mode 100644 index 000000000..db4063159 --- /dev/null +++ b/etl-api/src/k8s/snapshots/etl_api__k8s__http__tests__create_replicator_config_map_json.snap.new @@ -0,0 +1,21 @@ +--- +source: etl-api/src/k8s/http.rs +assertion_line: 1184 +expression: config_map_json +--- +{ + "apiVersion": "v1", + "data": { + "base.json": "", + "prod.json": "{\"destination\":{\"big_query\":{\"project_id\":\"project-id\",\"dataset_id\":\"dataset-id\",\"max_concurrent_streams\":4}},\"pipeline\":{\"id\":42,\"publication_name\":\"all-pub\",\"replication_slot\":{\"temporary\":false},\"pg_connection\":{\"host\":\"localhost\",\"port\":5432,\"name\":\"postgres\",\"username\":\"postgres\",\"tls\":{\"trusted_root_certs\":\"\",\"enabled\":false}},\"batch\":{\"max_size\":10000,\"max_fill_ms\":1000},\"table_error_retry_delay_ms\":500,\"table_error_retry_max_attempts\":3,\"max_table_sync_workers\":4}}" + }, + "kind": "ConfigMap", + "metadata": { + "labels": { + "etl.supabase.com/app-name": "abcdefghijklmnopqrst-42-replicator-app", + "etl.supabase.com/app-type": "etl-replicator-app" + }, + "name": "abcdefghijklmnopqrst-42-replicator-config", + "namespace": "etl-data-plane" + } +} diff --git a/etl-api/tests/support/mocks.rs b/etl-api/tests/support/mocks.rs index 2ea653627..5b51c887b 100644 --- a/etl-api/tests/support/mocks.rs +++ b/etl-api/tests/support/mocks.rs @@ -224,7 +224,7 @@ pub mod tenants { /// Pipeline config helpers. pub mod pipelines { use super::*; - use etl_api::configs::{log::LogLevel, pipeline::ApiBatchConfig}; + use etl_api::configs::{log::LogLevel, pipeline::{ApiBatchConfig, ApiReplicationSlotConfig}}; /// Returns a default pipeline config. pub fn new_pipeline_config() -> FullApiPipelineConfig { @@ -238,7 +238,7 @@ pub mod pipelines { table_error_retry_max_attempts: Some(5), max_table_sync_workers: Some(2), log_level: Some(LogLevel::Info), - temporary_replication_slot: false, + replication_slot: ApiReplicationSlotConfig::default(), } } @@ -254,7 +254,7 @@ pub mod pipelines { table_error_retry_max_attempts: Some(10), max_table_sync_workers: Some(4), log_level: Some(LogLevel::Info), - temporary_replication_slot: false, + replication_slot: ApiReplicationSlotConfig::default(), } } diff --git a/etl-benchmarks/benches/table_copies.rs b/etl-benchmarks/benches/table_copies.rs index 2a7a2e46f..250078db6 100644 --- a/etl-benchmarks/benches/table_copies.rs +++ b/etl-benchmarks/benches/table_copies.rs @@ -328,7 +328,7 @@ async fn start_pipeline(args: RunArgs) -> Result<(), Box> { let pipeline_config = PipelineConfig { id: 1, publication_name: args.publication_name, - replication_slot: ReplicationSlotConfig::Permanent, + replication_slot: ReplicationSlotConfig::default(), pg_connection: pg_connection_config, batch: BatchConfig { max_size: args.batch_max_size, diff --git a/etl-config/src/shared/pipeline.rs b/etl-config/src/shared/pipeline.rs index 7c7cef4a7..20c0d21fe 100644 --- a/etl-config/src/shared/pipeline.rs +++ b/etl-config/src/shared/pipeline.rs @@ -96,7 +96,13 @@ impl From for PipelineConfigWithoutSecrets { } #[derive(Clone, Copy, Debug, Deserialize, Serialize)] -pub enum ReplicationSlotConfig { - Temporary, - Permanent, +#[serde(rename_all = "snake_case")] +pub struct ReplicationSlotConfig { + pub temporary: bool, } + +impl Default for ReplicationSlotConfig { + fn default() -> Self { + ReplicationSlotConfig { temporary: false } + } +} \ No newline at end of file diff --git a/etl-examples/src/main.rs b/etl-examples/src/main.rs index 7f1418e4d..56b4cd0b3 100644 --- a/etl-examples/src/main.rs +++ b/etl-examples/src/main.rs @@ -176,7 +176,7 @@ async fn main_impl() -> Result<(), Box> { table_error_retry_delay_ms: 10000, table_error_retry_max_attempts: 5, max_table_sync_workers: args.bq_args.max_table_sync_workers, - replication_slot: ReplicationSlotConfig::Permanent, + replication_slot: ReplicationSlotConfig::default(), }; // Initialize BigQuery destination with service account authentication diff --git a/etl/src/lib.rs b/etl/src/lib.rs index 1f166483b..ee6ba3024 100644 --- a/etl/src/lib.rs +++ b/etl/src/lib.rs @@ -76,7 +76,7 @@ //! let config = PipelineConfig { //! id: 1, //! publication_name: "my_publication".to_string(), -//! replication_slot: ReplicationSlotConfig::Permanent, +//! replication_slot: ReplicationSlotConfig::default(), //! pg_connection: pg_config, //! batch: BatchConfig { max_size: 1000, max_fill_ms: 5000 }, //! table_error_retry_delay_ms: 10000, diff --git a/etl/src/replication/client.rs b/etl/src/replication/client.rs index 835b54c14..a06ecd8c0 100644 --- a/etl/src/replication/client.rs +++ b/etl/src/replication/client.rs @@ -611,9 +611,9 @@ impl PgReplicationClient { } else { "NOEXPORT_SNAPSHOT" }; - let temporary = match config { - ReplicationSlotConfig::Temporary => " TEMPORARY ", - ReplicationSlotConfig::Permanent => " ", + let temporary = match config.temporary { + true => " TEMPORARY ", + false => " ", }; let query = format!( r#"CREATE_REPLICATION_SLOT {}{}LOGICAL pgoutput {}"#, diff --git a/etl/src/replication/table_sync.rs b/etl/src/replication/table_sync.rs index a72628b44..faeeecf7f 100644 --- a/etl/src/replication/table_sync.rs +++ b/etl/src/replication/table_sync.rs @@ -1,4 +1,4 @@ -use etl_config::shared::{PipelineConfig, ReplicationSlotConfig}; +use etl_config::shared::PipelineConfig; use etl_postgres::replication::slots::EtlReplicationSlot; use etl_postgres::types::TableId; use futures::StreamExt; @@ -183,7 +183,7 @@ where // If a slot already exists at this point, we could delete it and try to recover, but it means // that the state was somehow reset without the slot being deleted, and we want to surface this. let (transaction, slot) = replication_client - .create_slot_with_transaction(&slot_name, ReplicationSlotConfig::Permanent) + .create_slot_with_transaction(&slot_name, config.replication_slot) .await?; // We copy the table schema and write it both to the state store and destination. diff --git a/etl/src/test_utils/pipeline.rs b/etl/src/test_utils/pipeline.rs index f71318b53..6347ff921 100644 --- a/etl/src/test_utils/pipeline.rs +++ b/etl/src/test_utils/pipeline.rs @@ -39,7 +39,7 @@ where table_error_retry_delay_ms: 1000, table_error_retry_max_attempts: 2, max_table_sync_workers: 1, - replication_slot: ReplicationSlotConfig::Permanent, + replication_slot: ReplicationSlotConfig::default(), }; Pipeline::new(config, store, destination) @@ -65,7 +65,7 @@ where let config = PipelineConfig { id: pipeline_id, publication_name, - replication_slot: ReplicationSlotConfig::Permanent, + replication_slot: ReplicationSlotConfig::default(), pg_connection: pg_connection_config.clone(), batch, table_error_retry_delay_ms: 1000, diff --git a/etl/tests/replication.rs b/etl/tests/replication.rs index 5477febf0..107b3e1b4 100644 --- a/etl/tests/replication.rs +++ b/etl/tests/replication.rs @@ -95,7 +95,7 @@ async fn test_replication_client_creates_slot() { let slot_name = test_slot_name("my_slot"); let create_slot = client - .create_slot(&slot_name, ReplicationSlotConfig::Permanent) + .create_slot(&slot_name, ReplicationSlotConfig::default()) .await .unwrap(); assert!(!create_slot.consistent_point.to_string().is_empty()); @@ -121,7 +121,7 @@ async fn test_create_and_delete_slot() { // Create the slot and verify it exists let create_slot = client - .create_slot(&slot_name, ReplicationSlotConfig::Permanent) + .create_slot(&slot_name, ReplicationSlotConfig::default()) .await .unwrap(); assert!(!create_slot.consistent_point.to_string().is_empty()); @@ -165,12 +165,12 @@ async fn test_replication_client_doesnt_recreate_slot() { let slot_name = test_slot_name("my_slot"); assert!( client - .create_slot(&slot_name, ReplicationSlotConfig::Permanent) + .create_slot(&slot_name, ReplicationSlotConfig::default()) .await .is_ok() ); assert!(matches!( - client.create_slot(&slot_name, ReplicationSlotConfig::Permanent).await, + client.create_slot(&slot_name, ReplicationSlotConfig::default()).await, Err(ref err) if err.kind() == ErrorKind::ReplicationSlotAlreadyExists )); } @@ -186,7 +186,7 @@ async fn test_replication_client_temporary_slot_dropped_on_disconnect() { let slot_name = test_slot_name("my_slot"); assert!(matches!( client - .create_slot(&slot_name, ReplicationSlotConfig::Temporary) + .create_slot(&slot_name, ReplicationSlotConfig { temporary: true}) .await, Ok(CreateSlotResult { consistent_point: _ @@ -204,7 +204,7 @@ async fn test_replication_client_temporary_slot_dropped_on_disconnect() { assert!(matches!( client - .create_slot(&slot_name, ReplicationSlotConfig::Temporary) + .create_slot(&slot_name, ReplicationSlotConfig { temporary: true}) .await, Ok(CreateSlotResult { consistent_point: _ @@ -223,7 +223,7 @@ async fn test_replication_client_permanent_slot_persisted_on_disconnect() { let slot_name = test_slot_name("my_slot"); assert!(matches!( client - .create_slot(&slot_name, ReplicationSlotConfig::Permanent) + .create_slot(&slot_name, ReplicationSlotConfig { temporary: false}) .await, Ok(CreateSlotResult { consistent_point: _ @@ -240,7 +240,7 @@ async fn test_replication_client_permanent_slot_persisted_on_disconnect() { .unwrap(); assert!(matches!( - client.create_slot(&slot_name, ReplicationSlotConfig::Permanent).await, + client.create_slot(&slot_name, ReplicationSlotConfig { temporary: false }).await, Err(ref err) if err.kind() == ErrorKind::ReplicationSlotAlreadyExists, )); } @@ -269,7 +269,7 @@ async fn test_table_schema_copy_is_consistent() { // We create the slot when the database schema contains only 'table_1'. let (transaction, _) = client - .create_slot_with_transaction(&test_slot_name("my_slot"), ReplicationSlotConfig::Permanent) + .create_slot_with_transaction(&test_slot_name("my_slot"), ReplicationSlotConfig::default()) .await .unwrap(); @@ -321,7 +321,7 @@ async fn test_table_schema_copy_across_multiple_connections() { // We create the slot when the database schema contains only 'table_1'. let (transaction, _) = first_client - .create_slot_with_transaction(&test_slot_name("my_slot"), ReplicationSlotConfig::Permanent) + .create_slot_with_transaction(&test_slot_name("my_slot"), ReplicationSlotConfig::default()) .await .unwrap(); @@ -356,7 +356,7 @@ async fn test_table_schema_copy_across_multiple_connections() { // We create the slot when the database schema contains both 'table_1' and 'table_2'. let (transaction, _) = second_client - .create_slot_with_transaction(&test_slot_name("my_slot"), ReplicationSlotConfig::Permanent) + .create_slot_with_transaction(&test_slot_name("my_slot"), ReplicationSlotConfig::default()) .await .unwrap(); @@ -420,7 +420,7 @@ async fn test_table_copy_stream_is_consistent() { // We create the slot when the database schema contains only 'table_1' data. let (transaction, _) = parent_client - .create_slot_with_transaction(&test_slot_name("my_slot"), ReplicationSlotConfig::Permanent) + .create_slot_with_transaction(&test_slot_name("my_slot"), ReplicationSlotConfig::default()) .await .unwrap(); @@ -496,7 +496,7 @@ async fn test_table_copy_stream_respects_row_filter() { // We create the slot when the database schema contains only 'table_1' data. let (transaction, _) = parent_client - .create_slot_with_transaction(&test_slot_name("my_slot"), ReplicationSlotConfig::Permanent) + .create_slot_with_transaction(&test_slot_name("my_slot"), ReplicationSlotConfig::default()) .await .unwrap(); @@ -583,7 +583,7 @@ async fn test_table_copy_stream_respects_column_filter() { // Create the slot when the database schema contains the test data. let (transaction, _) = parent_client - .create_slot_with_transaction(&test_slot_name("my_slot"), ReplicationSlotConfig::Permanent) + .create_slot_with_transaction(&test_slot_name("my_slot"), ReplicationSlotConfig::default()) .await .unwrap(); @@ -671,7 +671,7 @@ async fn test_table_copy_stream_no_row_filter() { // We create the slot when the database schema contains only 'table_1' data. let (transaction, _) = parent_client - .create_slot_with_transaction(&test_slot_name("my_slot"), ReplicationSlotConfig::Permanent) + .create_slot_with_transaction(&test_slot_name("my_slot"), ReplicationSlotConfig::default()) .await .unwrap(); @@ -799,7 +799,7 @@ async fn test_start_logical_replication() { // We create a slot which is going to replicate data before we insert the data. let slot_name = test_slot_name("my_slot"); let slot = parent_client - .create_slot(&slot_name, ReplicationSlotConfig::Permanent) + .create_slot(&slot_name, ReplicationSlotConfig::default()) .await .unwrap(); From 22990a4f7b3fa74de3ab03da3f8e5d66de1b127a Mon Sep 17 00:00:00 2001 From: Aksel Borgen Date: Sun, 4 Jan 2026 23:02:33 +0100 Subject: [PATCH 06/14] Update .snaps --- ...ts__create_replicator_config_map_json.snap | 2 +- ...create_replicator_config_map_json.snap.new | 21 ------------------- ...ination_and_pipeline_can_be_updated-2.snap | 4 +++- ...ination_and_pipeline_can_be_updated-2.snap | 4 +++- ...ination_and_pipeline_can_be_created-2.snap | 4 +++- ...ination_and_pipeline_can_be_created-2.snap | 4 +++- .../pipelines__all_pipelines_can_be_read.snap | 4 +++- ...nes__an_existing_pipeline_can_be_read.snap | 4 +++- ...__an_existing_pipeline_can_be_updated.snap | 4 +++- ...nes__pipeline_config_can_be_updated-2.snap | 4 +++- ...nes__pipeline_config_can_be_updated-3.snap | 4 +++- ...nes__pipeline_config_can_be_updated-4.snap | 4 +++- ...nes__pipeline_config_can_be_updated-5.snap | 4 +++- ...lines__pipeline_config_can_be_updated.snap | 4 +++- 14 files changed, 37 insertions(+), 34 deletions(-) delete mode 100644 etl-api/src/k8s/snapshots/etl_api__k8s__http__tests__create_replicator_config_map_json.snap.new diff --git a/etl-api/src/k8s/snapshots/etl_api__k8s__http__tests__create_replicator_config_map_json.snap b/etl-api/src/k8s/snapshots/etl_api__k8s__http__tests__create_replicator_config_map_json.snap index c183894e9..1599ef403 100644 --- a/etl-api/src/k8s/snapshots/etl_api__k8s__http__tests__create_replicator_config_map_json.snap +++ b/etl-api/src/k8s/snapshots/etl_api__k8s__http__tests__create_replicator_config_map_json.snap @@ -6,7 +6,7 @@ expression: config_map_json "apiVersion": "v1", "data": { "base.json": "", - "prod.json": "{\"destination\":{\"big_query\":{\"project_id\":\"project-id\",\"dataset_id\":\"dataset-id\",\"max_concurrent_streams\":4}},\"pipeline\":{\"id\":42,\"publication_name\":\"all-pub\",\"replication_slot\":\"Permanent\",\"pg_connection\":{\"host\":\"localhost\",\"port\":5432,\"name\":\"postgres\",\"username\":\"postgres\",\"tls\":{\"trusted_root_certs\":\"\",\"enabled\":false}},\"batch\":{\"max_size\":10000,\"max_fill_ms\":1000},\"table_error_retry_delay_ms\":500,\"table_error_retry_max_attempts\":3,\"max_table_sync_workers\":4}}" + "prod.json": "{\"destination\":{\"big_query\":{\"project_id\":\"project-id\",\"dataset_id\":\"dataset-id\",\"max_concurrent_streams\":4}},\"pipeline\":{\"id\":42,\"publication_name\":\"all-pub\",\"replication_slot\":{\"temporary\":false},\"pg_connection\":{\"host\":\"localhost\",\"port\":5432,\"name\":\"postgres\",\"username\":\"postgres\",\"tls\":{\"trusted_root_certs\":\"\",\"enabled\":false}},\"batch\":{\"max_size\":10000,\"max_fill_ms\":1000},\"table_error_retry_delay_ms\":500,\"table_error_retry_max_attempts\":3,\"max_table_sync_workers\":4}}" }, "kind": "ConfigMap", "metadata": { diff --git a/etl-api/src/k8s/snapshots/etl_api__k8s__http__tests__create_replicator_config_map_json.snap.new b/etl-api/src/k8s/snapshots/etl_api__k8s__http__tests__create_replicator_config_map_json.snap.new deleted file mode 100644 index db4063159..000000000 --- a/etl-api/src/k8s/snapshots/etl_api__k8s__http__tests__create_replicator_config_map_json.snap.new +++ /dev/null @@ -1,21 +0,0 @@ ---- -source: etl-api/src/k8s/http.rs -assertion_line: 1184 -expression: config_map_json ---- -{ - "apiVersion": "v1", - "data": { - "base.json": "", - "prod.json": "{\"destination\":{\"big_query\":{\"project_id\":\"project-id\",\"dataset_id\":\"dataset-id\",\"max_concurrent_streams\":4}},\"pipeline\":{\"id\":42,\"publication_name\":\"all-pub\",\"replication_slot\":{\"temporary\":false},\"pg_connection\":{\"host\":\"localhost\",\"port\":5432,\"name\":\"postgres\",\"username\":\"postgres\",\"tls\":{\"trusted_root_certs\":\"\",\"enabled\":false}},\"batch\":{\"max_size\":10000,\"max_fill_ms\":1000},\"table_error_retry_delay_ms\":500,\"table_error_retry_max_attempts\":3,\"max_table_sync_workers\":4}}" - }, - "kind": "ConfigMap", - "metadata": { - "labels": { - "etl.supabase.com/app-name": "abcdefghijklmnopqrst-42-replicator-app", - "etl.supabase.com/app-type": "etl-replicator-app" - }, - "name": "abcdefghijklmnopqrst-42-replicator-config", - "namespace": "etl-data-plane" - } -} diff --git a/etl-api/tests/snapshots/destinations_pipelines__an_existing_bigquery_destination_and_pipeline_can_be_updated-2.snap b/etl-api/tests/snapshots/destinations_pipelines__an_existing_bigquery_destination_and_pipeline_can_be_updated-2.snap index 24f2771f8..349cc2622 100644 --- a/etl-api/tests/snapshots/destinations_pipelines__an_existing_bigquery_destination_and_pipeline_can_be_updated-2.snap +++ b/etl-api/tests/snapshots/destinations_pipelines__an_existing_bigquery_destination_and_pipeline_can_be_updated-2.snap @@ -4,7 +4,9 @@ expression: response.config --- FullApiPipelineConfig { publication_name: "updated_publication", - temporary_replication_slot: false, + replication_slot: ApiReplicationSlotConfig { + temporary: false, + }, batch: Some( ApiBatchConfig { max_size: Some( diff --git a/etl-api/tests/snapshots/destinations_pipelines__an_existing_iceberg_supabase_destination_and_pipeline_can_be_updated-2.snap b/etl-api/tests/snapshots/destinations_pipelines__an_existing_iceberg_supabase_destination_and_pipeline_can_be_updated-2.snap index 24f2771f8..349cc2622 100644 --- a/etl-api/tests/snapshots/destinations_pipelines__an_existing_iceberg_supabase_destination_and_pipeline_can_be_updated-2.snap +++ b/etl-api/tests/snapshots/destinations_pipelines__an_existing_iceberg_supabase_destination_and_pipeline_can_be_updated-2.snap @@ -4,7 +4,9 @@ expression: response.config --- FullApiPipelineConfig { publication_name: "updated_publication", - temporary_replication_slot: false, + replication_slot: ApiReplicationSlotConfig { + temporary: false, + }, batch: Some( ApiBatchConfig { max_size: Some( diff --git a/etl-api/tests/snapshots/destinations_pipelines__bigquery_destination_and_pipeline_can_be_created-2.snap b/etl-api/tests/snapshots/destinations_pipelines__bigquery_destination_and_pipeline_can_be_created-2.snap index 42dbb80d1..9a801d6b0 100644 --- a/etl-api/tests/snapshots/destinations_pipelines__bigquery_destination_and_pipeline_can_be_created-2.snap +++ b/etl-api/tests/snapshots/destinations_pipelines__bigquery_destination_and_pipeline_can_be_created-2.snap @@ -4,7 +4,9 @@ expression: response.config --- FullApiPipelineConfig { publication_name: "publication", - temporary_replication_slot: false, + replication_slot: ApiReplicationSlotConfig { + temporary: false, + }, batch: Some( ApiBatchConfig { max_size: Some( diff --git a/etl-api/tests/snapshots/destinations_pipelines__iceberg_supabase_destination_and_pipeline_can_be_created-2.snap b/etl-api/tests/snapshots/destinations_pipelines__iceberg_supabase_destination_and_pipeline_can_be_created-2.snap index 42dbb80d1..9a801d6b0 100644 --- a/etl-api/tests/snapshots/destinations_pipelines__iceberg_supabase_destination_and_pipeline_can_be_created-2.snap +++ b/etl-api/tests/snapshots/destinations_pipelines__iceberg_supabase_destination_and_pipeline_can_be_created-2.snap @@ -4,7 +4,9 @@ expression: response.config --- FullApiPipelineConfig { publication_name: "publication", - temporary_replication_slot: false, + replication_slot: ApiReplicationSlotConfig { + temporary: false, + }, batch: Some( ApiBatchConfig { max_size: Some( diff --git a/etl-api/tests/snapshots/pipelines__all_pipelines_can_be_read.snap b/etl-api/tests/snapshots/pipelines__all_pipelines_can_be_read.snap index 2307c2170..0f42d34eb 100644 --- a/etl-api/tests/snapshots/pipelines__all_pipelines_can_be_read.snap +++ b/etl-api/tests/snapshots/pipelines__all_pipelines_can_be_read.snap @@ -4,7 +4,9 @@ expression: pipeline.config --- FullApiPipelineConfig { publication_name: "publication", - temporary_replication_slot: false, + replication_slot: ApiReplicationSlotConfig { + temporary: false, + }, batch: Some( ApiBatchConfig { max_size: Some( diff --git a/etl-api/tests/snapshots/pipelines__an_existing_pipeline_can_be_read.snap b/etl-api/tests/snapshots/pipelines__an_existing_pipeline_can_be_read.snap index 74c78729d..d4fef874f 100644 --- a/etl-api/tests/snapshots/pipelines__an_existing_pipeline_can_be_read.snap +++ b/etl-api/tests/snapshots/pipelines__an_existing_pipeline_can_be_read.snap @@ -4,7 +4,9 @@ expression: response.config --- FullApiPipelineConfig { publication_name: "publication", - temporary_replication_slot: false, + replication_slot: ApiReplicationSlotConfig { + temporary: false, + }, batch: Some( ApiBatchConfig { max_size: Some( diff --git a/etl-api/tests/snapshots/pipelines__an_existing_pipeline_can_be_updated.snap b/etl-api/tests/snapshots/pipelines__an_existing_pipeline_can_be_updated.snap index b41ea8fcb..b7b63c3d4 100644 --- a/etl-api/tests/snapshots/pipelines__an_existing_pipeline_can_be_updated.snap +++ b/etl-api/tests/snapshots/pipelines__an_existing_pipeline_can_be_updated.snap @@ -4,7 +4,9 @@ expression: response.config --- FullApiPipelineConfig { publication_name: "updated_publication", - temporary_replication_slot: false, + replication_slot: ApiReplicationSlotConfig { + temporary: false, + }, batch: Some( ApiBatchConfig { max_size: Some( diff --git a/etl-api/tests/snapshots/pipelines__pipeline_config_can_be_updated-2.snap b/etl-api/tests/snapshots/pipelines__pipeline_config_can_be_updated-2.snap index f779e9b37..c6d78ec5a 100644 --- a/etl-api/tests/snapshots/pipelines__pipeline_config_can_be_updated-2.snap +++ b/etl-api/tests/snapshots/pipelines__pipeline_config_can_be_updated-2.snap @@ -4,7 +4,9 @@ expression: response.config --- FullApiPipelineConfig { publication_name: "publication", - temporary_replication_slot: false, + replication_slot: ApiReplicationSlotConfig { + temporary: false, + }, batch: Some( ApiBatchConfig { max_size: Some( diff --git a/etl-api/tests/snapshots/pipelines__pipeline_config_can_be_updated-3.snap b/etl-api/tests/snapshots/pipelines__pipeline_config_can_be_updated-3.snap index 2d4073a24..a3f04ffc9 100644 --- a/etl-api/tests/snapshots/pipelines__pipeline_config_can_be_updated-3.snap +++ b/etl-api/tests/snapshots/pipelines__pipeline_config_can_be_updated-3.snap @@ -4,7 +4,9 @@ expression: response.config --- FullApiPipelineConfig { publication_name: "publication", - temporary_replication_slot: false, + replication_slot: ApiReplicationSlotConfig { + temporary: false, + }, batch: Some( ApiBatchConfig { max_size: Some( diff --git a/etl-api/tests/snapshots/pipelines__pipeline_config_can_be_updated-4.snap b/etl-api/tests/snapshots/pipelines__pipeline_config_can_be_updated-4.snap index e7de59663..fc6bbe70c 100644 --- a/etl-api/tests/snapshots/pipelines__pipeline_config_can_be_updated-4.snap +++ b/etl-api/tests/snapshots/pipelines__pipeline_config_can_be_updated-4.snap @@ -4,7 +4,9 @@ expression: response.config --- FullApiPipelineConfig { publication_name: "publication", - temporary_replication_slot: false, + replication_slot: ApiReplicationSlotConfig { + temporary: false, + }, batch: Some( ApiBatchConfig { max_size: Some( diff --git a/etl-api/tests/snapshots/pipelines__pipeline_config_can_be_updated-5.snap b/etl-api/tests/snapshots/pipelines__pipeline_config_can_be_updated-5.snap index f708c2706..7e9247f85 100644 --- a/etl-api/tests/snapshots/pipelines__pipeline_config_can_be_updated-5.snap +++ b/etl-api/tests/snapshots/pipelines__pipeline_config_can_be_updated-5.snap @@ -4,7 +4,9 @@ expression: response.config --- FullApiPipelineConfig { publication_name: "publication", - temporary_replication_slot: false, + replication_slot: ApiReplicationSlotConfig { + temporary: false, + }, batch: Some( ApiBatchConfig { max_size: Some( diff --git a/etl-api/tests/snapshots/pipelines__pipeline_config_can_be_updated.snap b/etl-api/tests/snapshots/pipelines__pipeline_config_can_be_updated.snap index 49419c0a1..aa54fff75 100644 --- a/etl-api/tests/snapshots/pipelines__pipeline_config_can_be_updated.snap +++ b/etl-api/tests/snapshots/pipelines__pipeline_config_can_be_updated.snap @@ -4,7 +4,9 @@ expression: response.config --- FullApiPipelineConfig { publication_name: "publication", - temporary_replication_slot: false, + replication_slot: ApiReplicationSlotConfig { + temporary: false, + }, batch: Some( ApiBatchConfig { max_size: Some( From 313810842c6769bbb1c8482824a785b34da08fca Mon Sep 17 00:00:00 2001 From: Aksel Borgen Date: Sun, 4 Jan 2026 23:02:49 +0100 Subject: [PATCH 07/14] Run cargo fmt --- etl-api/src/configs/pipeline.rs | 12 +++++++----- etl-api/tests/support/mocks.rs | 5 ++++- etl-config/src/shared/pipeline.rs | 2 +- etl/tests/replication.rs | 6 +++--- 4 files changed, 15 insertions(+), 10 deletions(-) diff --git a/etl-api/src/configs/pipeline.rs b/etl-api/src/configs/pipeline.rs index d9edd3790..1841a7541 100644 --- a/etl-api/src/configs/pipeline.rs +++ b/etl-api/src/configs/pipeline.rs @@ -26,7 +26,6 @@ pub struct ApiBatchConfig { pub max_fill_ms: Option, } - #[derive(Debug, Clone, Serialize, Deserialize, ToSchema)] pub struct ApiReplicationSlotConfig { #[schema(example = false)] @@ -41,13 +40,17 @@ impl Default for ApiReplicationSlotConfig { impl From for ApiReplicationSlotConfig { fn from(value: ReplicationSlotConfig) -> Self { - ApiReplicationSlotConfig { temporary: value.temporary } + ApiReplicationSlotConfig { + temporary: value.temporary, + } } } impl From for ReplicationSlotConfig { fn from(value: ApiReplicationSlotConfig) -> Self { - ReplicationSlotConfig { temporary: value.temporary } + ReplicationSlotConfig { + temporary: value.temporary, + } } } @@ -71,7 +74,6 @@ pub struct FullApiPipelineConfig { pub log_level: Option, } - impl From for FullApiPipelineConfig { fn from(value: StoredPipelineConfig) -> Self { Self { @@ -256,7 +258,7 @@ mod tests { table_error_retry_max_attempts: None, max_table_sync_workers: None, log_level: Some(LogLevel::Debug), - replication_slot: ApiReplicationSlotConfig::default() + replication_slot: ApiReplicationSlotConfig::default(), }; let stored: StoredPipelineConfig = full_config.clone().into(); diff --git a/etl-api/tests/support/mocks.rs b/etl-api/tests/support/mocks.rs index 5b51c887b..60b790acf 100644 --- a/etl-api/tests/support/mocks.rs +++ b/etl-api/tests/support/mocks.rs @@ -224,7 +224,10 @@ pub mod tenants { /// Pipeline config helpers. pub mod pipelines { use super::*; - use etl_api::configs::{log::LogLevel, pipeline::{ApiBatchConfig, ApiReplicationSlotConfig}}; + use etl_api::configs::{ + log::LogLevel, + pipeline::{ApiBatchConfig, ApiReplicationSlotConfig}, + }; /// Returns a default pipeline config. pub fn new_pipeline_config() -> FullApiPipelineConfig { diff --git a/etl-config/src/shared/pipeline.rs b/etl-config/src/shared/pipeline.rs index 20c0d21fe..0741b4824 100644 --- a/etl-config/src/shared/pipeline.rs +++ b/etl-config/src/shared/pipeline.rs @@ -105,4 +105,4 @@ impl Default for ReplicationSlotConfig { fn default() -> Self { ReplicationSlotConfig { temporary: false } } -} \ No newline at end of file +} diff --git a/etl/tests/replication.rs b/etl/tests/replication.rs index 107b3e1b4..66bbe9b67 100644 --- a/etl/tests/replication.rs +++ b/etl/tests/replication.rs @@ -186,7 +186,7 @@ async fn test_replication_client_temporary_slot_dropped_on_disconnect() { let slot_name = test_slot_name("my_slot"); assert!(matches!( client - .create_slot(&slot_name, ReplicationSlotConfig { temporary: true}) + .create_slot(&slot_name, ReplicationSlotConfig { temporary: true }) .await, Ok(CreateSlotResult { consistent_point: _ @@ -204,7 +204,7 @@ async fn test_replication_client_temporary_slot_dropped_on_disconnect() { assert!(matches!( client - .create_slot(&slot_name, ReplicationSlotConfig { temporary: true}) + .create_slot(&slot_name, ReplicationSlotConfig { temporary: true }) .await, Ok(CreateSlotResult { consistent_point: _ @@ -223,7 +223,7 @@ async fn test_replication_client_permanent_slot_persisted_on_disconnect() { let slot_name = test_slot_name("my_slot"); assert!(matches!( client - .create_slot(&slot_name, ReplicationSlotConfig { temporary: false}) + .create_slot(&slot_name, ReplicationSlotConfig { temporary: false }) .await, Ok(CreateSlotResult { consistent_point: _ From 548b6a1fdb77b25c250a736ca6aa34ee59ec402d Mon Sep 17 00:00:00 2001 From: Aksel Borgen Date: Sun, 4 Jan 2026 23:08:59 +0100 Subject: [PATCH 08/14] Change ReplicationSlotConfig to derive Default --- etl-config/src/shared/pipeline.rs | 7 +------ 1 file changed, 1 insertion(+), 6 deletions(-) diff --git a/etl-config/src/shared/pipeline.rs b/etl-config/src/shared/pipeline.rs index 0741b4824..f837e8b18 100644 --- a/etl-config/src/shared/pipeline.rs +++ b/etl-config/src/shared/pipeline.rs @@ -95,14 +95,9 @@ impl From for PipelineConfigWithoutSecrets { } } -#[derive(Clone, Copy, Debug, Deserialize, Serialize)] +#[derive(Clone, Copy, Debug, Deserialize, Serialize, Default)] #[serde(rename_all = "snake_case")] pub struct ReplicationSlotConfig { pub temporary: bool, } -impl Default for ReplicationSlotConfig { - fn default() -> Self { - ReplicationSlotConfig { temporary: false } - } -} From dc6f1f288510ce2ea3f0b94b34c1e4a0359191e6 Mon Sep 17 00:00:00 2001 From: Aksel Borgen Date: Sun, 4 Jan 2026 23:10:44 +0100 Subject: [PATCH 09/14] cargo fmt --- etl-config/src/shared/pipeline.rs | 1 - 1 file changed, 1 deletion(-) diff --git a/etl-config/src/shared/pipeline.rs b/etl-config/src/shared/pipeline.rs index f837e8b18..f45220e54 100644 --- a/etl-config/src/shared/pipeline.rs +++ b/etl-config/src/shared/pipeline.rs @@ -100,4 +100,3 @@ impl From for PipelineConfigWithoutSecrets { pub struct ReplicationSlotConfig { pub temporary: bool, } - From 8fabadb2a41f431aa2431879563a692fd8ab33a8 Mon Sep 17 00:00:00 2001 From: Aksel Borgen Date: Sun, 4 Jan 2026 23:16:35 +0100 Subject: [PATCH 10/14] Run clippy --- etl-api/src/configs/pipeline.rs | 8 +------- 1 file changed, 1 insertion(+), 7 deletions(-) diff --git a/etl-api/src/configs/pipeline.rs b/etl-api/src/configs/pipeline.rs index 1841a7541..ff1c4c41f 100644 --- a/etl-api/src/configs/pipeline.rs +++ b/etl-api/src/configs/pipeline.rs @@ -26,18 +26,12 @@ pub struct ApiBatchConfig { pub max_fill_ms: Option, } -#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)] +#[derive(Debug, Clone, Serialize, Deserialize, ToSchema, Default)] pub struct ApiReplicationSlotConfig { #[schema(example = false)] pub temporary: bool, } -impl Default for ApiReplicationSlotConfig { - fn default() -> Self { - Self { temporary: false } - } -} - impl From for ApiReplicationSlotConfig { fn from(value: ReplicationSlotConfig) -> Self { ApiReplicationSlotConfig { From 1b1f8aa9d70f693e98e00022d7f8d656d67c97bf Mon Sep 17 00:00:00 2001 From: Aksel Borgen Date: Wed, 7 Jan 2026 23:10:30 +0100 Subject: [PATCH 11/14] Cargo fmt --- etl/src/replication/client.rs | 3 ++- etl/tests/replication.rs | 21 ++++++++++++++++++--- 2 files changed, 20 insertions(+), 4 deletions(-) diff --git a/etl/src/replication/client.rs b/etl/src/replication/client.rs index d9f5429fe..adeeca10d 100644 --- a/etl/src/replication/client.rs +++ b/etl/src/replication/client.rs @@ -2,7 +2,8 @@ use crate::error::{ErrorKind, EtlResult}; use crate::utils::tokio::MakeRustlsConnect; use crate::{bail, etl_error}; use etl_config::shared::{ - ETL_REPLICATION_OPTIONS, IntoConnectOptions, PgConnectionConfig, ReplicationSlotConfig, ReplicationSlotPersistence, + ETL_REPLICATION_OPTIONS, IntoConnectOptions, PgConnectionConfig, ReplicationSlotConfig, + ReplicationSlotPersistence, }; use etl_postgres::replication::extract_server_version; use etl_postgres::types::convert_type_oid_to_type; diff --git a/etl/tests/replication.rs b/etl/tests/replication.rs index 7f2502d78..face6251e 100644 --- a/etl/tests/replication.rs +++ b/etl/tests/replication.rs @@ -186,7 +186,12 @@ async fn test_replication_client_temporary_slot_dropped_on_disconnect() { let slot_name = test_slot_name("my_slot"); assert!(matches!( client - .create_slot(&slot_name, ReplicationSlotConfig { persistence: ReplicationSlotPersistence::Temporary }) + .create_slot( + &slot_name, + ReplicationSlotConfig { + persistence: ReplicationSlotPersistence::Temporary + } + ) .await, Ok(CreateSlotResult { consistent_point: _ @@ -204,7 +209,12 @@ async fn test_replication_client_temporary_slot_dropped_on_disconnect() { assert!(matches!( client - .create_slot(&slot_name, ReplicationSlotConfig { persistence: ReplicationSlotPersistence::Temporary }) + .create_slot( + &slot_name, + ReplicationSlotConfig { + persistence: ReplicationSlotPersistence::Temporary + } + ) .await, Ok(CreateSlotResult { consistent_point: _ @@ -223,7 +233,12 @@ async fn test_replication_client_permanent_slot_persisted_on_disconnect() { let slot_name = test_slot_name("my_slot"); assert!(matches!( client - .create_slot(&slot_name, ReplicationSlotConfig { persistence: ReplicationSlotPersistence::Permanent }) + .create_slot( + &slot_name, + ReplicationSlotConfig { + persistence: ReplicationSlotPersistence::Permanent + } + ) .await, Ok(CreateSlotResult { consistent_point: _ From 69cbbd5a98ab38f6c4451ddd39f6f58668f8b9ee Mon Sep 17 00:00:00 2001 From: Aksel Borgen Date: Wed, 7 Jan 2026 23:11:44 +0100 Subject: [PATCH 12/14] Remove left-over schema example for replication slot config --- etl-api/src/configs/pipeline.rs | 1 - 1 file changed, 1 deletion(-) diff --git a/etl-api/src/configs/pipeline.rs b/etl-api/src/configs/pipeline.rs index bf8f56bc9..e005a9053 100644 --- a/etl-api/src/configs/pipeline.rs +++ b/etl-api/src/configs/pipeline.rs @@ -58,7 +58,6 @@ impl From for ReplicationSlotPersistence { #[derive(Debug, Clone, Serialize, Deserialize, ToSchema, Default)] #[serde(rename_all = "snake_case")] pub struct ApiReplicationSlotConfig { - #[schema(example = false)] pub persistence: ApiReplicationSlotPersistence, } From c9d032682ddc9a3761206df6d769ee872f60b79b Mon Sep 17 00:00:00 2001 From: Aksel Borgen Date: Wed, 7 Jan 2026 23:20:09 +0100 Subject: [PATCH 13/14] Docs --- etl-config/src/shared/pipeline.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/etl-config/src/shared/pipeline.rs b/etl-config/src/shared/pipeline.rs index 66238abfb..e5fc99e01 100644 --- a/etl-config/src/shared/pipeline.rs +++ b/etl-config/src/shared/pipeline.rs @@ -64,7 +64,7 @@ pub struct PipelineConfig { pub id: u64, /// Name of the Postgres publication to use for logical replication. pub publication_name: String, - /// Whether to use a temporary replication slot + /// Configuration for the replication slot used pub replication_slot: ReplicationSlotConfig, /// The connection configuration for the Postgres instance to which the pipeline connects for /// replication. @@ -112,7 +112,7 @@ pub struct PipelineConfigWithoutSecrets { pub id: u64, /// Name of the Postgres publication to use for logical replication. pub publication_name: String, - /// Whether to use a temporary replication slot + /// Configuration for the replication slot used pub replication_slot: ReplicationSlotConfig, /// The connection configuration for the Postgres instance to which the pipeline connects for /// replication. From 9b7d2a3998ba75f4a9ebac533dcea151761834e1 Mon Sep 17 00:00:00 2001 From: Aksel Borgen Date: Wed, 7 Jan 2026 23:34:28 +0100 Subject: [PATCH 14/14] Ignore PipelineBuilder doctest --- etl/src/test_utils/pipeline.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/etl/src/test_utils/pipeline.rs b/etl/src/test_utils/pipeline.rs index be1cc21b8..3a7348c05 100644 --- a/etl/src/test_utils/pipeline.rs +++ b/etl/src/test_utils/pipeline.rs @@ -27,7 +27,7 @@ pub fn test_slot_name(slot_name: &str) -> String { /// /// # Examples /// -/// ``` +/// ```ignore /// // Create a pipeline with default settings /// let pipeline = PipelineBuilder::new(pg_config, id, pub_name, store, dest) /// .build();