From 60f7183f43b77cf99e3d4d0284d65d3f98e33613 Mon Sep 17 00:00:00 2001 From: psteinroe Date: Tue, 21 Jul 2026 15:46:22 +0000 Subject: [PATCH] fix: preserve transaction commit LSN on live events --- src/sink/memory.rs | 1 + src/store.rs | 2 + src/types/event.rs | 54 ++++++--- tests/elasticsearch_sink_tests.rs | 3 + tests/gcp_pubsub_sink_tests.rs | 2 + tests/kafka_sink_tests.rs | 2 + tests/kinesis_sink_tests.rs | 2 + tests/meilisearch_sink_tests.rs | 3 + tests/nats_sink_tests.rs | 2 + tests/rabbitmq_sink_tests.rs | 2 + tests/redis_streams_sink_tests.rs | 2 + tests/redis_strings_sink_tests.rs | 2 + tests/replay_client_tests.rs | 27 ++++- tests/sns_sink_tests.rs | 2 + tests/sqs_sink_tests.rs | 2 + tests/stream_tests.rs | 30 ++++- tests/transaction_lsn_tests.rs | 190 ++++++++++++++++++++++++++++++ tests/webhook_sink_tests.rs | 4 + 18 files changed, 304 insertions(+), 28 deletions(-) create mode 100644 tests/transaction_lsn_tests.rs diff --git a/src/sink/memory.rs b/src/sink/memory.rs index 4ad4761..1d19d8b 100644 --- a/src/sink/memory.rs +++ b/src/sink/memory.rs @@ -92,6 +92,7 @@ mod tests { payload: serde_json::json!({ "test": id }), metadata: None, stream_id: StreamId::from(1u64), + commit_lsn: None, lsn: Some("0/16B3748".parse().unwrap()), } } diff --git a/src/store.rs b/src/store.rs index 668c8ec..e9b7355 100644 --- a/src/store.rs +++ b/src/store.rs @@ -78,6 +78,7 @@ impl StreamStore { payload, metadata: row.event_metadata, stream_id: crate::types::StreamId::from(event_stream_id as u64), + commit_lsn: None, lsn: row.event_lsn.and_then(|s| s.parse().ok()), })); } @@ -176,6 +177,7 @@ impl StreamStore { payload: row.payload, metadata: row.metadata, stream_id: crate::types::StreamId::from(row.stream_id as u64), + commit_lsn: None, lsn: row.lsn.and_then(|s| s.parse().ok()), }); diff --git a/src/types/event.rs b/src/types/event.rs index 88d0112..f248795 100644 --- a/src/types/event.rs +++ b/src/types/event.rs @@ -49,8 +49,12 @@ pub struct TriggeredEvent { pub payload: serde_json::Value, pub metadata: Option, pub stream_id: StreamId, + /// The logical replication transaction commit LSN. + /// Available for live events and safe to use as a replica consistency watermark. + pub commit_lsn: Option, /// The WAL LSN at the time the event was inserted. - /// Used for precise replay point lookup during slot recovery. + /// Used for precise replay point lookup during slot recovery, but not safe as a + /// replica consistency watermark because it precedes the transaction commit. pub lsn: Option, } @@ -78,6 +82,7 @@ macro_rules! missing { pub fn convert_event_from_table( table_row: &mut TableRow, column_schemas: &[ColumnSchema], + commit_lsn: Option, ) -> EtlResult { let mut id = None; let mut created_at = None; @@ -110,6 +115,7 @@ pub fn convert_event_from_table( payload: payload.ok_or_else(|| missing!("payload"))?, stream_id: stream_id.ok_or_else(|| missing!("stream_id"))?, metadata, + commit_lsn, lsn, }) } @@ -120,7 +126,7 @@ pub fn convert_events_from_table_rows( ) -> EtlResult> { table_rows .into_iter() - .map(|mut table_row| convert_event_from_table(&mut table_row, column_schemas)) + .map(|mut table_row| convert_event_from_table(&mut table_row, column_schemas, None)) .collect() } @@ -131,10 +137,15 @@ pub fn convert_stream_events_from_events( events .into_iter() .filter_map(|event| match event { - Event::Insert(mut insert_event) => Some(convert_event_from_table( - &mut insert_event.table_row, - column_schemas, - )), + Event::Insert(mut insert_event) => { + let commit_lsn = Some(insert_event.commit_lsn); + + Some(convert_event_from_table( + &mut insert_event.table_row, + column_schemas, + commit_lsn, + )) + } Event::Begin(_) | Event::Commit(_) | Event::Update(_) @@ -175,9 +186,9 @@ mod tests { Cell::Uuid(id), Cell::TimestampTz(created_at), Cell::Json(payload), - Cell::Null, // metadata - Cell::I64(1), // stream_id - Cell::String("0/16B3748".to_string()), // lsn (parsed to PgLsn) + Cell::Null, // metadata + Cell::I64(1), // stream_id + Cell::String("0/100".to_string()), // lsn (parsed to PgLsn) ], } } @@ -200,7 +211,7 @@ mod tests { } #[test] - fn test_convert_event_from_table_valid() { + fn test_convert_replayed_event_preserves_row_lsn_without_commit_lsn() { let column_schemas = make_column_schemas(); let id = Uuid::new_v4(); let created_at = Utc::now(); @@ -208,12 +219,13 @@ mod tests { let mut table_row = make_table_row(id, created_at, payload.clone()); - let result = convert_event_from_table(&mut table_row, &column_schemas).unwrap(); + let result = convert_event_from_table(&mut table_row, &column_schemas, None).unwrap(); assert_eq!(result.id.id, id.to_string()); assert_eq!(result.id.created_at, created_at); assert_eq!(result.payload, payload); - assert_eq!(result.lsn, Some("0/16B3748".parse().unwrap())); + assert_eq!(result.lsn, Some("0/100".parse().unwrap())); + assert_eq!(result.commit_lsn, None); } #[test] @@ -225,12 +237,13 @@ mod tests { let mut table_row = make_table_row_without_lsn(id, created_at, payload.clone()); - let result = convert_event_from_table(&mut table_row, &column_schemas).unwrap(); + let result = convert_event_from_table(&mut table_row, &column_schemas, None).unwrap(); assert_eq!(result.id.id, id.to_string()); assert_eq!(result.id.created_at, created_at); assert_eq!(result.payload, payload); assert_eq!(result.lsn, None); + assert_eq!(result.commit_lsn, None); } #[test] @@ -247,7 +260,7 @@ mod tests { ], }; - let result = convert_event_from_table(&mut table_row, &column_schemas); + let result = convert_event_from_table(&mut table_row, &column_schemas, None); assert!(result.is_err()); assert!(result.unwrap_err().to_string().contains("Missing id")); @@ -267,7 +280,7 @@ mod tests { ], }; - let result = convert_event_from_table(&mut table_row, &column_schemas); + let result = convert_event_from_table(&mut table_row, &column_schemas, None); assert!(result.is_err()); assert!( @@ -292,7 +305,7 @@ mod tests { ], }; - let result = convert_event_from_table(&mut table_row, &column_schemas); + let result = convert_event_from_table(&mut table_row, &column_schemas, None); assert!(result.is_err()); assert!(result.unwrap_err().to_string().contains("Missing payload")); @@ -330,7 +343,7 @@ mod tests { } #[test] - fn test_convert_stream_events_from_events_filters_inserts_only() { + fn test_convert_live_insert_preserves_commit_lsn() { let column_schemas = make_column_schemas(); let id = Uuid::new_v4(); let ts = Utc::now(); @@ -338,7 +351,7 @@ mod tests { let events = vec![ Event::Insert(InsertEvent { start_lsn: PgLsn::from(0), - commit_lsn: PgLsn::from(0), + commit_lsn: "0/200".parse().unwrap(), table_id: TableId::new(1), table_row: make_table_row(id, ts, serde_json::json!({"test": 1})), }), @@ -353,6 +366,8 @@ mod tests { let first = result.first().expect("first element exists"); assert_eq!(first.id.id, id.to_string()); assert_eq!(first.payload.get("test").and_then(|v| v.as_i64()), Some(1)); + assert_eq!(first.lsn, Some("0/100".parse().unwrap())); + assert_eq!(first.commit_lsn, Some("0/200".parse().unwrap())); } #[test] @@ -455,6 +470,7 @@ mod tests { payload: payload.clone(), metadata: None, stream_id: StreamId::from(1u64), + commit_lsn: Some("0/200".parse().unwrap()), lsn: Some("0/16B3748".parse().unwrap()), }; let event2 = TriggeredEvent { @@ -462,6 +478,7 @@ mod tests { payload, metadata: None, stream_id: StreamId::from(1u64), + commit_lsn: Some("0/200".parse().unwrap()), lsn: Some("0/16B3748".parse().unwrap()), }; @@ -479,6 +496,7 @@ mod tests { payload, metadata: None, stream_id: StreamId::from(1u64), + commit_lsn: None, lsn: None, }; let (id_returned, created_at) = event.primary_keys(); diff --git a/tests/elasticsearch_sink_tests.rs b/tests/elasticsearch_sink_tests.rs index 89139c6..a1e5578 100644 --- a/tests/elasticsearch_sink_tests.rs +++ b/tests/elasticsearch_sink_tests.rs @@ -22,6 +22,7 @@ fn make_test_event(key: &str) -> TriggeredEvent { stream_id: StreamId::default(), payload: serde_json::json!({ "key": key, "value": "test_data" }), metadata: Some(serde_json::json!({ "source": "test" })), + commit_lsn: None, lsn: Some(PgLsn::from(12345u64)), } } @@ -131,6 +132,7 @@ async fn test_elasticsearch_sink_indexes_only_payload() { stream_id: StreamId::default(), payload: serde_json::json!({ "action": "created", "user_id": 456 }), metadata: Some(serde_json::json!({ "source": "api" })), + commit_lsn: None, lsn: Some(PgLsn::from(99999u64)), }; let event_id = event.id.id.clone(); @@ -247,6 +249,7 @@ async fn test_elasticsearch_sink_uses_index_from_metadata() { stream_id: StreamId::default(), payload: serde_json::json!({ "routed": true }), metadata: Some(serde_json::json!({ "index": metadata_index })), + commit_lsn: None, lsn: None, }; let event_id = event.id.id.clone(); diff --git a/tests/gcp_pubsub_sink_tests.rs b/tests/gcp_pubsub_sink_tests.rs index 2b6554a..1bac3c7 100644 --- a/tests/gcp_pubsub_sink_tests.rs +++ b/tests/gcp_pubsub_sink_tests.rs @@ -21,6 +21,7 @@ fn make_test_event(key: &str) -> TriggeredEvent { stream_id: StreamId::default(), payload: serde_json::json!({ "key": key, "value": "test_data" }), metadata: Some(serde_json::json!({ "source": "test" })), + commit_lsn: None, lsn: Some(PgLsn::from(12345u64)), } } @@ -201,6 +202,7 @@ async fn test_gcp_pubsub_sink_sends_only_payload() { stream_id: StreamId::default(), payload: serde_json::json!({ "action": "created", "user_id": 123 }), metadata: Some(serde_json::json!({ "routing_key": "orders" })), + commit_lsn: None, lsn: Some(PgLsn::from(99999u64)), }; diff --git a/tests/kafka_sink_tests.rs b/tests/kafka_sink_tests.rs index 8c540f0..ed1af41 100644 --- a/tests/kafka_sink_tests.rs +++ b/tests/kafka_sink_tests.rs @@ -25,6 +25,7 @@ fn make_test_event(id: &str) -> TriggeredEvent { }), metadata: Some(serde_json::json!({ "source": "test" })), stream_id: StreamId::from(1u64), + commit_lsn: None, lsn: Some("0/16B3748".parse().unwrap()), } } @@ -148,6 +149,7 @@ async fn test_kafka_sink_uses_topic_from_metadata() { }), metadata: Some(serde_json::json!({ "topic": topic })), stream_id: StreamId::from(1u64), + commit_lsn: None, lsn: Some("0/16B3748".parse().unwrap()), }; diff --git a/tests/kinesis_sink_tests.rs b/tests/kinesis_sink_tests.rs index c9e8922..b6a3913 100644 --- a/tests/kinesis_sink_tests.rs +++ b/tests/kinesis_sink_tests.rs @@ -19,6 +19,7 @@ fn make_test_event(key: &str) -> TriggeredEvent { stream_id: StreamId::default(), payload: serde_json::json!({ "key": key, "value": "test_data" }), metadata: Some(serde_json::json!({ "source": "test" })), + commit_lsn: None, lsn: Some(PgLsn::from(12345u64)), } } @@ -231,6 +232,7 @@ async fn test_kinesis_sink_uses_stream_from_metadata() { stream_id: StreamId::default(), payload: serde_json::json!({ "action": "created" }), metadata: Some(serde_json::json!({ "stream": stream_name })), + commit_lsn: None, lsn: Some(PgLsn::from(99999u64)), }; diff --git a/tests/meilisearch_sink_tests.rs b/tests/meilisearch_sink_tests.rs index 86817b5..6e9ac39 100644 --- a/tests/meilisearch_sink_tests.rs +++ b/tests/meilisearch_sink_tests.rs @@ -22,6 +22,7 @@ fn make_test_event(key: &str) -> TriggeredEvent { stream_id: StreamId::default(), payload: serde_json::json!({ "id": id, "key": key, "value": "test_data" }), metadata: Some(serde_json::json!({ "source": "test" })), + commit_lsn: None, lsn: Some(PgLsn::from(12345u64)), } } @@ -119,6 +120,7 @@ async fn test_meilisearch_sink_indexes_only_payload() { stream_id: StreamId::default(), payload: serde_json::json!({ "id": event_id, "action": "created", "user": 456 }), metadata: Some(serde_json::json!({ "source": "api" })), + commit_lsn: None, lsn: Some(PgLsn::from(99999u64)), }; @@ -217,6 +219,7 @@ async fn test_meilisearch_sink_uses_index_from_metadata() { stream_id: StreamId::default(), payload: serde_json::json!({ "id": event_id, "routed": true }), metadata: Some(serde_json::json!({ "index": metadata_index })), + commit_lsn: None, lsn: None, }; diff --git a/tests/nats_sink_tests.rs b/tests/nats_sink_tests.rs index 2e1780b..a24bcef 100644 --- a/tests/nats_sink_tests.rs +++ b/tests/nats_sink_tests.rs @@ -21,6 +21,7 @@ fn make_test_event(id: &str) -> TriggeredEvent { }), metadata: Some(serde_json::json!({ "source": "test" })), stream_id: StreamId::from(1u64), + commit_lsn: None, lsn: Some("0/16B3748".parse().unwrap()), } } @@ -148,6 +149,7 @@ async fn test_nats_sink_uses_topic_from_metadata() { }), metadata: Some(serde_json::json!({ "topic": subject })), stream_id: StreamId::from(1u64), + commit_lsn: None, lsn: Some("0/16B3748".parse().unwrap()), }; diff --git a/tests/rabbitmq_sink_tests.rs b/tests/rabbitmq_sink_tests.rs index b7b23b1..7bc650b 100644 --- a/tests/rabbitmq_sink_tests.rs +++ b/tests/rabbitmq_sink_tests.rs @@ -21,6 +21,7 @@ fn make_test_event(id: &str) -> TriggeredEvent { }), metadata: Some(serde_json::json!({ "source": "test" })), stream_id: StreamId::from(1u64), + commit_lsn: None, lsn: Some("0/16B3748".parse().unwrap()), } } @@ -195,6 +196,7 @@ async fn test_rabbitmq_sink_uses_metadata_routing() { "routing_key": routing_key })), stream_id: StreamId::from(1u64), + commit_lsn: None, lsn: Some("0/16B3748".parse().unwrap()), }; diff --git a/tests/redis_streams_sink_tests.rs b/tests/redis_streams_sink_tests.rs index dee575c..f18e3ce 100644 --- a/tests/redis_streams_sink_tests.rs +++ b/tests/redis_streams_sink_tests.rs @@ -21,6 +21,7 @@ fn make_test_event(id: &str) -> TriggeredEvent { }), metadata: Some(serde_json::json!({ "source": "test" })), stream_id: StreamId::from(1u64), + commit_lsn: None, lsn: Some("0/16B3748".parse().unwrap()), } } @@ -172,6 +173,7 @@ async fn test_redis_streams_sink_uses_stream_from_metadata() { }), metadata: Some(serde_json::json!({ "stream": stream_name })), stream_id: StreamId::from(1u64), + commit_lsn: None, lsn: Some("0/16B3748".parse().unwrap()), }; diff --git a/tests/redis_strings_sink_tests.rs b/tests/redis_strings_sink_tests.rs index b007efe..383dd64 100644 --- a/tests/redis_strings_sink_tests.rs +++ b/tests/redis_strings_sink_tests.rs @@ -21,6 +21,7 @@ fn make_test_event(id: &str) -> TriggeredEvent { }), metadata: None, stream_id: StreamId::from(1u64), + commit_lsn: None, lsn: Some("0/16B3748".parse().unwrap()), } } @@ -149,6 +150,7 @@ async fn test_redis_strings_sink_uses_key_from_metadata() { }), metadata: Some(serde_json::json!({ "key": custom_key })), stream_id: StreamId::from(1u64), + commit_lsn: None, lsn: Some("0/16B3748".parse().unwrap()), }; diff --git a/tests/replay_client_tests.rs b/tests/replay_client_tests.rs index 2b26188..28fa2f7 100644 --- a/tests/replay_client_tests.rs +++ b/tests/replay_client_tests.rs @@ -4,7 +4,7 @@ use postgres_stream::store::StreamStore; use postgres_stream::test_utils::{ TestDatabase, create_postgres_store, insert_events_to_db, test_stream_config, }; -use postgres_stream::types::EventIdentifier; +use postgres_stream::types::{EventIdentifier, convert_events_from_table_rows}; use uuid::Uuid; // Connection tests @@ -188,8 +188,12 @@ async fn test_get_events_copy_stream_with_events() { .await .expect("Failed to connect"); - // Insert multiple events + // Insert multiple events with a trigger-time row LSN. let events = insert_events_to_db(&db, 5).await; + sqlx::query("UPDATE pgstream.events SET lsn = '0/100'::pg_lsn WHERE stream_id = 1") + .execute(&db.pool) + .await + .unwrap(); // Get table schema to parse the copy stream let config = test_stream_config(&db); @@ -214,14 +218,25 @@ async fn test_get_events_copy_stream_with_events() { let stream = TableCopyStream::wrap(stream, &table_schema.column_schemas, 1); pin!(stream); - let mut event_count = 0; + let mut table_rows = Vec::new(); while let Some(result) = stream.next().await { - let _row = result.expect("Should parse row successfully"); - event_count += 1; + table_rows.push(result.expect("Should parse row successfully")); } + let replayed_events = + convert_events_from_table_rows(table_rows, &table_schema.column_schemas).unwrap(); + // Should get events 1, 2, 3 (events between 0 and 4, exclusive) - assert_eq!(event_count, 3, "Should have received exactly 3 events"); + assert_eq!( + replayed_events.len(), + 3, + "Should have received exactly 3 events" + ); + assert!( + replayed_events + .iter() + .all(|event| event.lsn.is_some() && event.commit_lsn.is_none()) + ); } #[tokio::test(flavor = "multi_thread")] diff --git a/tests/sns_sink_tests.rs b/tests/sns_sink_tests.rs index 565316c..2ff7851 100644 --- a/tests/sns_sink_tests.rs +++ b/tests/sns_sink_tests.rs @@ -20,6 +20,7 @@ fn make_test_event(key: &str) -> TriggeredEvent { stream_id: StreamId::default(), payload: serde_json::json!({ "key": key, "value": "test_data" }), metadata: Some(serde_json::json!({ "source": "test" })), + commit_lsn: None, lsn: Some(PgLsn::from(12345u64)), } } @@ -266,6 +267,7 @@ async fn test_sns_sink_sends_only_payload() { stream_id: StreamId::default(), payload: serde_json::json!({ "action": "created", "data": "test" }), metadata: Some(serde_json::json!({ "user_id": 123, "source": "api" })), + commit_lsn: None, lsn: Some(PgLsn::from(99999u64)), }; diff --git a/tests/sqs_sink_tests.rs b/tests/sqs_sink_tests.rs index 50a2a94..d9f27a8 100644 --- a/tests/sqs_sink_tests.rs +++ b/tests/sqs_sink_tests.rs @@ -20,6 +20,7 @@ fn make_test_event(id: &str) -> TriggeredEvent { }), metadata: Some(serde_json::json!({ "source": "test" })), stream_id: StreamId::from(1u64), + commit_lsn: None, lsn: Some("0/16B3748".parse().unwrap()), } } @@ -190,6 +191,7 @@ async fn test_sqs_sink_queue_url_from_metadata() { "queue_url": queue_url })), stream_id: StreamId::from(1u64), + commit_lsn: None, lsn: None, }; diff --git a/tests/stream_tests.rs b/tests/stream_tests.rs index 80f164f..7b58655 100644 --- a/tests/stream_tests.rs +++ b/tests/stream_tests.rs @@ -1,6 +1,7 @@ use chrono::Duration; use etl::destination::Destination; use etl::store::both::postgres::PostgresStore; +use etl::types::{Cell, Event, PgLsn}; use postgres_stream::sink::memory::MemorySink; use postgres_stream::stream::PgStream; use postgres_stream::test_utils::{ @@ -37,10 +38,26 @@ async fn test_pgstream_write_events_via_destination_trait() { .await .expect("Failed to create PgStream"); - let events = vec![ - make_test_event(table_id, serde_json::json!({"id": 1, "name": "Alice"})), - make_test_event(table_id, serde_json::json!({"id": 2, "name": "Bob"})), - ]; + let row_lsn: PgLsn = "0/100".parse().unwrap(); + let commit_lsn: PgLsn = "0/200".parse().unwrap(); + let events = [ + serde_json::json!({"id": 1, "name": "Alice"}), + serde_json::json!({"id": 2, "name": "Bob"}), + ] + .into_iter() + .map(|payload| { + let mut event = make_test_event(table_id, payload); + let Event::Insert(insert_event) = &mut event else { + panic!("test event should be an insert"); + }; + insert_event.commit_lsn = commit_lsn; + insert_event + .table_row + .values + .push(Cell::String("0/100".to_string())); + event + }) + .collect(); stream .write_events(events) @@ -49,6 +66,11 @@ async fn test_pgstream_write_events_via_destination_trait() { let stored_events = sink.events().await; assert_eq!(stored_events.len(), 2); + assert!( + stored_events + .iter() + .all(|event| event.lsn == Some(row_lsn) && event.commit_lsn == Some(commit_lsn)) + ); } // Failover tests diff --git a/tests/transaction_lsn_tests.rs b/tests/transaction_lsn_tests.rs new file mode 100644 index 0000000..cb37fe0 --- /dev/null +++ b/tests/transaction_lsn_tests.rs @@ -0,0 +1,190 @@ +use etl::pipeline::Pipeline; +use etl::store::both::postgres::PostgresStore; +use postgres_stream::sink::memory::MemorySink; +use postgres_stream::stream::PgStream; +use postgres_stream::test_utils::{TestDatabase, test_stream_config_with_id, unique_pipeline_id}; +use std::collections::HashSet; +use std::time::Duration; + +#[tokio::test(flavor = "multi_thread")] +async fn test_insert_and_update_share_commit_lsn_distinct_from_row_lsns() { + let db = TestDatabase::spawn().await; + db.ensure_today_partition().await; + + sqlx::query( + r#" + create table public.users ( + id serial primary key, + name text not null + ) + "#, + ) + .execute(&db.pool) + .await + .unwrap(); + + let stream_config = test_stream_config_with_id(&db, unique_pipeline_id()); + let pipeline_id = stream_config.id; + + for (key, operation) in [("user_insert", "INSERT"), ("user_update", "UPDATE")] { + sqlx::query( + r#" + insert into pgstream.subscriptions + (key, stream_id, operation, schema_name, table_name, column_names, payload_extensions, metadata_extensions) + values + ($1, $2, $3::pgstream.operation_type, 'public', 'users', array['id', 'name'], '[]', '[]') + "#, + ) + .bind(key) + .bind(pipeline_id as i64) + .bind(operation) + .execute(&db.pool) + .await + .unwrap(); + } + + let sink = MemorySink::new(); + let state_store = PostgresStore::new(pipeline_id, db.config.clone()); + let pgstream = PgStream::create(stream_config.clone(), sink.clone(), state_store.clone()) + .await + .unwrap(); + let pipeline_config: etl::config::PipelineConfig = stream_config.into(); + let mut pipeline = Pipeline::new(pipeline_config, state_store, pgstream); + pipeline.start().await.unwrap(); + + let slot_name = format!("supabase_etl_apply_{pipeline_id}"); + tokio::time::timeout(Duration::from_secs(30), async { + loop { + let slot_exists: bool = sqlx::query_scalar( + "select exists(select 1 from pg_replication_slots where slot_name = $1)", + ) + .bind(&slot_name) + .fetch_one(&db.pool) + .await + .unwrap(); + let states: Vec = sqlx::query_scalar( + "select state::text from etl.replication_state where pipeline_id = $1 and is_current = true", + ) + .bind(pipeline_id as i64) + .fetch_all(&db.pool) + .await + .unwrap_or_default(); + + if slot_exists + && !states.is_empty() + && states + .iter() + .all(|state| state == "sync_done" || state == "ready") + { + break; + } + + tokio::time::sleep(Duration::from_millis(100)).await; + } + }) + .await + .expect("pipeline should initialize"); + + sqlx::query( + "insert into pgstream.events (payload, stream_id, lsn) values ($1, $2, pg_current_wal_lsn())", + ) + .bind(serde_json::json!({"pipeline_ready": true})) + .bind(pipeline_id as i64) + .execute(&db.pool) + .await + .unwrap(); + + tokio::time::timeout(Duration::from_secs(30), async { + loop { + let states: Vec = sqlx::query_scalar( + "select state::text from etl.replication_state where pipeline_id = $1 and is_current = true", + ) + .bind(pipeline_id as i64) + .fetch_all(&db.pool) + .await + .unwrap_or_default(); + let warmup_received = sink.events().await.iter().any(|event| { + event + .payload + .get("pipeline_ready") + .and_then(serde_json::Value::as_bool) + == Some(true) + }); + + if !states.is_empty() + && states.iter().all(|state| state == "ready") + && warmup_received + { + break; + } + + tokio::time::sleep(Duration::from_millis(100)).await; + } + }) + .await + .expect("pipeline should become ready"); + sink.clear().await; + + let mut transaction = db.pool.begin().await.unwrap(); + let user_id: i32 = + sqlx::query_scalar("insert into public.users (name) values ('before') returning id") + .fetch_one(&mut *transaction) + .await + .unwrap(); + sqlx::query("update public.users set name = 'after' where id = $1") + .bind(user_id) + .execute(&mut *transaction) + .await + .unwrap(); + transaction.commit().await.unwrap(); + + let transaction_events = tokio::time::timeout(Duration::from_secs(30), async { + loop { + let events = sink.events().await; + let transaction_events: Vec<_> = events + .into_iter() + .filter(|event| { + matches!( + event + .payload + .get("tg_op") + .and_then(serde_json::Value::as_str), + Some("INSERT" | "UPDATE") + ) + }) + .collect(); + + if transaction_events.len() == 2 { + break transaction_events; + } + + tokio::time::sleep(Duration::from_millis(100)).await; + } + }) + .await; + + let shutdown_tx = pipeline.shutdown_tx(); + shutdown_tx.shutdown().unwrap(); + pipeline.wait().await.unwrap(); + + let transaction_events = transaction_events.expect("transaction events should be replicated"); + let operations: HashSet<_> = transaction_events + .iter() + .filter_map(|event| { + event + .payload + .get("tg_op") + .and_then(serde_json::Value::as_str) + }) + .collect(); + assert_eq!(operations, HashSet::from(["INSERT", "UPDATE"])); + + let commit_lsn = transaction_events + .first() + .and_then(|event| event.commit_lsn) + .expect("live events should have a commit LSN"); + + assert!(transaction_events.iter().all(|event| { + event.commit_lsn == Some(commit_lsn) && event.lsn.is_some() && event.lsn != Some(commit_lsn) + })); +} diff --git a/tests/webhook_sink_tests.rs b/tests/webhook_sink_tests.rs index 512f8e9..1e40f11 100644 --- a/tests/webhook_sink_tests.rs +++ b/tests/webhook_sink_tests.rs @@ -21,6 +21,7 @@ fn make_test_event(id: &str) -> TriggeredEvent { }), metadata: Some(serde_json::json!({ "source": "test" })), stream_id: StreamId::from(1u64), + commit_lsn: None, lsn: Some("0/16B3748".parse().unwrap()), } } @@ -178,6 +179,7 @@ async fn test_webhook_sink_url_from_metadata() { "url": format!("{}/dynamic-endpoint", mock_server.uri()) })), stream_id: StreamId::from(1u64), + commit_lsn: None, lsn: None, }; @@ -223,6 +225,7 @@ async fn test_webhook_sink_headers_from_metadata() { } })), stream_id: StreamId::from(1u64), + commit_lsn: None, lsn: None, }; @@ -250,6 +253,7 @@ async fn test_webhook_sink_no_url_configured_fails() { payload: serde_json::json!({ "test": "data" }), metadata: None, stream_id: StreamId::from(1u64), + commit_lsn: None, lsn: None, };