1mod contracts;
4mod coordinated_commit;
5mod recency;
6mod sink;
7mod source;
8mod source_batch;
9mod task_tracking;
10
11pub use contracts::{
12 DeliveryGuarantee, SinkConsistency, SinkContract, SinkInputMode, SinkTopology,
13 SourceConsistency, SourceContract, SourceInputMode, SourceReplayOrder,
14 SourceRowPositionCapability, SourceTopology,
15};
16pub use coordinated_commit::{
17 CoordinatedAbortBatch, CoordinatedAbortCleaner, CoordinatedAbortDescriptor,
18 CoordinatedAbortEntry, CoordinatedCommitBatch, CoordinatedCommitContext,
19 CoordinatedCommitCursor, CoordinatedCommitNamespace, CoordinatedCommitPayload,
20 CoordinatedCommitter, MAX_COORDINATED_COMMIT_BATCH_BYTES, MAX_COORDINATED_COMMIT_BATCH_ENTRIES,
21 MAX_COORDINATED_COMMIT_PAYLOAD_BYTES,
22};
23pub use recency::{RecencyTarget, SourceRecency};
24pub use sink::{SinkConnector, SinkRuntimeContext, WriteResult};
25pub use source::{
26 SourceCheckpointUnavailablePolicy, SourceConnector, SourceDrainOutcome, SourceDrainRequest,
27 SourceDrainResolution, SourcePosition, SourceStart,
28};
29pub use source_batch::{
30 schema_with_source_mutations_and_row_positions, schema_with_source_row_positions,
31 source_mutations, source_mutations_routed, source_row_positions, strip_source_mutations,
32 strip_source_mutations_routed, strip_source_row_positions, SourceBatch, SourceBatchCursor,
33 SourceMutation, SourceMutationView, SourceRowPositionRef, SourceRowPositionView,
34 SourceRowPositions, SOURCE_MUTATION_COLUMN, SOURCE_ORDER_KEY_COLUMN, SOURCE_PARTITION_COLUMN,
35 SOURCE_SUB_OFFSET_COLUMN,
36};
37pub use task_tracking::{
38 ConnectorCancellationPolicy, ConnectorTaskAdmission, ConnectorTaskGuard, ConnectorTaskOwner,
39 ConnectorTaskTracker,
40};
41
42#[cfg(test)]
43#[allow(clippy::cast_possible_wrap)]
44mod tests {
45 use super::*;
46 use arrow_array::{ArrayRef, BinaryArray, Int64Array, RecordBatch, UInt32Array, UInt8Array};
47 use arrow_schema::{DataType, Field, Schema, SchemaRef};
48 use async_trait::async_trait;
49 use std::sync::Arc;
50
51 use crate::checkpoint::SourceCheckpoint;
52 use crate::config::ConnectorConfig;
53 use crate::error::ConnectorError;
54
55 fn test_schema() -> SchemaRef {
56 Arc::new(Schema::new(vec![Field::new("id", DataType::Int64, false)]))
57 }
58
59 fn test_batch(n: usize) -> RecordBatch {
60 #[allow(clippy::cast_possible_wrap)]
61 let ids: Vec<i64> = (0..n as i64).collect();
62 RecordBatch::try_new(test_schema(), vec![Arc::new(Int64Array::from(ids))]).unwrap()
63 }
64
65 #[test]
66 fn connector_task_generation_terminates_after_owner_and_every_guard() {
67 let (owner, tracker) = ConnectorTaskOwner::new();
68 let first = owner.track().expect("live generation");
69 let second = owner.track().expect("live generation");
70
71 assert!(!tracker.is_terminated());
72 drop(owner);
73 assert!(!tracker.is_terminated());
74 drop(first);
75 assert!(!tracker.is_terminated());
76 drop(second);
77 assert!(tracker.is_terminated());
78 }
79
80 #[test]
81 fn connector_task_admission_is_sealed_by_owner_drop() {
82 let (owner, tracker) = ConnectorTaskOwner::new();
83 let admission = owner.admission();
84 let admission_clone = admission.clone();
85 let admitted = admission.track().expect("live generation");
86
87 drop(owner);
88
89 assert!(admission.track().is_none());
90 assert!(admission_clone.track().is_none());
91 assert!(!tracker.is_terminated());
92 drop(admitted);
93 assert!(tracker.is_terminated());
94 }
95
96 #[test]
97 fn connector_task_admission_does_not_retain_generation_state() {
98 let (owner, tracker) = ConnectorTaskOwner::new();
99 let admission = owner.admission();
100
101 drop(owner);
102 drop(tracker);
103
104 assert!(admission.inner.upgrade().is_none());
105 assert!(admission.track().is_none());
106 }
107
108 #[tokio::test]
109 async fn connector_task_wait_wakes_every_tracker_clone() {
110 let (owner, tracker) = ConnectorTaskOwner::new();
111 let guard = owner.track().expect("live generation");
112 let first = tokio::spawn({
113 let tracker = tracker.clone();
114 async move { tracker.wait_terminated().await }
115 });
116 let second = tokio::spawn({
117 let tracker = tracker.clone();
118 async move { tracker.wait_terminated().await }
119 });
120
121 drop(owner);
122 assert!(!tracker.is_terminated());
123 drop(guard);
124
125 tokio::time::timeout(std::time::Duration::from_secs(1), async {
126 first.await.expect("first waiter task");
127 second.await.expect("second waiter task");
128 })
129 .await
130 .expect("tracker waiters must wake");
131 tracker.wait_terminated().await;
132 }
133
134 #[test]
135 fn test_source_batch() {
136 let batch = SourceBatch::new(test_batch(10));
137 assert_eq!(batch.num_rows(), 10);
138 assert!(batch.row_positions().is_none());
139 assert!(batch.mutations().is_none());
140 }
141
142 #[test]
143 fn source_row_positions_reject_nulls_and_misalignment() {
144 let null_partition = BinaryArray::from(vec![Some(&b"p0"[..]), None]);
145 let order = BinaryArray::from(vec![&b"0"[..], &b"1"[..]]);
146 let sub_offset = UInt32Array::from(vec![0, 0]);
147 assert!(
148 SourceRowPositions::try_new(null_partition, order.clone(), sub_offset.clone()).is_err()
149 );
150
151 let positions = SourceRowPositions::try_new(
152 BinaryArray::from(vec![&b"p0"[..], &b"p0"[..]]),
153 order,
154 sub_offset,
155 )
156 .unwrap();
157 assert!(SourceBatch::positioned(test_batch(1), positions).is_err());
158 }
159
160 #[test]
161 fn source_batch_validates_and_canonicalizes_mutations() {
162 let mixed = SourceBatch::new(test_batch(2))
163 .with_mutations(vec![SourceMutation::Put, SourceMutation::Tombstone])
164 .unwrap();
165 assert_eq!(
166 mixed.mutations(),
167 Some(&[SourceMutation::Put, SourceMutation::Tombstone][..])
168 );
169
170 let puts = SourceBatch::new(test_batch(2))
171 .with_mutations(vec![SourceMutation::Put; 2])
172 .unwrap();
173 assert!(puts.mutations().is_none());
174 assert!(SourceBatch::new(test_batch(2))
175 .with_mutations(vec![SourceMutation::Tombstone])
176 .is_err());
177 }
178
179 #[test]
180 fn source_metadata_round_trip_is_sparse_and_zero_copy() {
181 let records = test_batch(2);
182 assert!(source_row_positions(&records).unwrap().is_none());
183 let positioned_schema = schema_with_source_row_positions(&records.schema()).unwrap();
184 let mutation_schema =
185 schema_with_source_mutations_and_row_positions(&records.schema()).unwrap();
186 let positions = SourceRowPositions::try_new(
187 BinaryArray::from(vec![&b"p0"[..], &b"p0"[..]]),
188 BinaryArray::from(vec![&b"0"[..], &b"1"[..]]),
189 UInt32Array::from(vec![0, 0]),
190 )
191 .unwrap();
192 let encoded = SourceBatch::positioned(records.clone(), positions.clone())
193 .unwrap()
194 .with_mutations(vec![SourceMutation::Put, SourceMutation::Tombstone])
195 .unwrap()
196 .into_records_with_metadata(
197 SourceRowPositionCapability::OrderedDeterministic,
198 &positioned_schema,
199 &mutation_schema,
200 )
201 .unwrap();
202 let mutations = source_mutations(&encoded).unwrap().unwrap();
203 assert_eq!(mutations.len(), 2);
204 assert!(!mutations.is_empty());
205 assert_eq!(mutations.get(0), Some(SourceMutation::Put));
206 assert_eq!(mutations.get(1), Some(SourceMutation::Tombstone));
207 let row_positions = source_row_positions(&encoded).unwrap().unwrap();
208 assert_eq!(row_positions.len(), 2);
209 assert!(!row_positions.is_empty());
210 assert_eq!(
211 row_positions.get(1),
212 Some(SourceRowPositionRef {
213 partition: b"p0",
214 order_key: b"1",
215 sub_offset: 0,
216 })
217 );
218 assert_eq!(row_positions.get(2), None);
219 assert_eq!(
220 encoded.schema().field(records.num_columns()).name(),
221 SOURCE_MUTATION_COLUMN
222 );
223
224 let positioned = strip_source_mutations(&encoded).unwrap();
225 assert_eq!(positioned.schema(), positioned_schema);
226 assert!(Arc::ptr_eq(positioned.column(0), records.column(0)));
227
228 let routed_put = encoded.slice(0, 1);
229 assert!(source_mutations(&routed_put).is_err());
230 assert_eq!(
231 source_mutations_routed(&routed_put)
232 .unwrap()
233 .unwrap()
234 .get(0),
235 Some(SourceMutation::Put)
236 );
237 let routed_visible = Arc::clone(routed_put.column(0));
238 let routed_positioned = strip_source_mutations_routed(&routed_put).unwrap();
239 assert!(Arc::ptr_eq(&routed_visible, routed_positioned.column(0)));
240
241 let stripped = strip_source_row_positions(&encoded).unwrap();
242 assert_eq!(stripped.schema(), records.schema());
243 assert_eq!(stripped.num_rows(), records.num_rows());
244 assert!(Arc::ptr_eq(stripped.column(0), records.column(0)));
245
246 let puts = SourceBatch::positioned(records.clone(), positions)
247 .unwrap()
248 .with_mutations(vec![SourceMutation::Put; 2])
249 .unwrap()
250 .into_records_with_metadata(
251 SourceRowPositionCapability::OrderedDeterministic,
252 &positioned_schema,
253 &mutation_schema,
254 )
255 .unwrap();
256 assert!(Arc::ptr_eq(&puts.schema(), &positioned_schema));
257 assert!(puts.column_by_name(SOURCE_MUTATION_COLUMN).is_none());
258 assert!(source_mutations(&puts).unwrap().is_none());
259 }
260
261 #[test]
262 fn source_metadata_rejects_collisions_and_malformed_batches() {
263 let collision = Arc::new(Schema::new(vec![Field::new(
264 "__SOURCE_MUTATION",
265 DataType::UInt8,
266 false,
267 )]));
268 assert!(schema_with_source_row_positions(&collision).is_err());
269 assert!(schema_with_source_mutations_and_row_positions(&collision).is_err());
270
271 let records = test_batch(2);
272 let positioned_schema = schema_with_source_row_positions(&records.schema()).unwrap();
273 let mutation_schema =
274 schema_with_source_mutations_and_row_positions(&records.schema()).unwrap();
275 let positions = SourceRowPositions::try_new(
276 BinaryArray::from(vec![&b"p0"[..], &b"p0"[..]]),
277 BinaryArray::from(vec![&b"0"[..], &b"1"[..]]),
278 UInt32Array::from(vec![0, 0]),
279 )
280 .unwrap();
281 let encoded = SourceBatch::positioned(records.clone(), positions)
282 .unwrap()
283 .with_mutations(vec![SourceMutation::Put, SourceMutation::Tombstone])
284 .unwrap()
285 .into_records_with_metadata(
286 SourceRowPositionCapability::OrderedDeterministic,
287 &positioned_schema,
288 &mutation_schema,
289 )
290 .unwrap();
291 let mutation_index = records.num_columns();
292
293 let malformed = |field: Field, array: ArrayRef| {
294 let mut fields = encoded.schema().fields().to_vec();
295 fields[mutation_index] = Arc::new(field);
296 let mut columns = encoded.columns().to_vec();
297 columns[mutation_index] = array;
298 RecordBatch::try_new(Arc::new(Schema::new(fields)), columns).unwrap()
299 };
300 let wrong_type = malformed(
301 Field::new(SOURCE_MUTATION_COLUMN, DataType::Int64, false),
302 Arc::new(Int64Array::from(vec![0, 1])),
303 );
304 assert!(source_mutations(&wrong_type).is_err());
305
306 let null = malformed(
307 Field::new(SOURCE_MUTATION_COLUMN, DataType::UInt8, true),
308 Arc::new(UInt8Array::from(vec![Some(0), None])),
309 );
310 assert!(strip_source_mutations(&null).is_err());
311
312 let unknown = malformed(
313 Field::new(SOURCE_MUTATION_COLUMN, DataType::UInt8, false),
314 Arc::new(UInt8Array::from(vec![0, 2])),
315 );
316 assert!(source_mutations(&unknown).is_err());
317 assert!(strip_source_mutations(&unknown).is_err());
318
319 let all_put = malformed(
320 Field::new(SOURCE_MUTATION_COLUMN, DataType::UInt8, false),
321 Arc::new(UInt8Array::from(vec![0, 0])),
322 );
323 assert!(strip_source_mutations(&all_put).is_err());
324
325 let mut fields = encoded.schema().fields().to_vec();
326 let mutation_field = fields.remove(mutation_index);
327 fields.push(mutation_field);
328 let mut columns = encoded.columns().to_vec();
329 let mutation_column = columns.remove(mutation_index);
330 columns.push(mutation_column);
331 let misplaced = RecordBatch::try_new(Arc::new(Schema::new(fields)), columns).unwrap();
332 assert!(source_mutations(&misplaced).is_err());
333 }
334
335 #[test]
336 fn test_write_result() {
337 let result = WriteResult::new(100, 5000);
338 assert_eq!(result.records_written, 100);
339 assert_eq!(result.bytes_written, 5000);
340 }
341
342 #[test]
343 fn source_drain_request_requires_canonical_round() {
344 let round = laminar_core::checkpoint::AssignmentDrainId {
345 predecessor_version: 7,
346 target_version: 8,
347 digest: [9; 32],
348 };
349 assert_eq!(SourceDrainRequest::new(round).unwrap().round, round);
350 assert!(
351 SourceDrainRequest::new(laminar_core::checkpoint::AssignmentDrainId {
352 predecessor_version: 8,
353 target_version: 8,
354 digest: [9; 32],
355 })
356 .is_err()
357 );
358 }
359
360 #[test]
361 fn source_contract_defaults_fail_closed() {
362 let contract = SourceContract::default();
363 assert_eq!(contract.consistency, SourceConsistency::Ephemeral);
364 assert_eq!(contract.topology, SourceTopology::Singleton);
365 assert_eq!(contract.input_mode, SourceInputMode::AppendOnly);
366 assert_eq!(
367 contract.row_positions,
368 SourceRowPositionCapability::Unavailable
369 );
370 assert!(!contract.supports_replay());
371 assert_eq!(contract.replay_order, SourceReplayOrder::Unspecified);
372 assert!(!contract.requires_checkpointing());
373 assert!(!contract.is_exact_delivery_certified());
374 }
375
376 #[test]
377 fn commit_coupled_sources_are_replayable_and_require_checkpoints() {
378 let contract = SourceContract::new(
379 SourceConsistency::CommitCoupled,
380 SourceTopology::NodeLocalIngress,
381 SourceInputMode::FullChangelog,
382 );
383 assert!(contract.supports_replay());
384 assert!(contract.requires_checkpointing());
385 assert_eq!(contract.input_mode, SourceInputMode::FullChangelog);
386 }
387
388 #[test]
389 fn source_start_rejects_split_and_zero_resume_before_connector_start() {
390 use laminar_core::checkpoint::CheckpointAttempt;
391
392 for attempt in [CheckpointAttempt::new(7, 8), CheckpointAttempt::new(0, 0)] {
393 let error = SourceStart::new(
394 ConnectorConfig::new("test"),
395 SourcePosition::Resume {
396 attempt,
397 checkpoint: SourceCheckpoint::new(),
398 },
399 DeliveryGuarantee::AtLeastOnce,
400 )
401 .unwrap_err();
402 assert!(matches!(
403 error,
404 ConnectorError::ConfigurationError(message)
405 if message.contains("one nonzero canonical checkpoint ID")
406 ));
407 }
408 }
409
410 #[test]
411 fn source_start_accepts_initial_and_exposes_validated_parts() {
412 let mut config = ConnectorConfig::new("test");
413 config.set("endpoint", "local");
414 let request = SourceStart::new(
415 config,
416 SourcePosition::Initial,
417 DeliveryGuarantee::BestEffort,
418 )
419 .unwrap();
420
421 let (config, position, delivery) = request.into_parts();
422 assert_eq!(config.get("endpoint"), Some("local"));
423 assert!(matches!(position, SourcePosition::Initial));
424 assert_eq!(delivery, DeliveryGuarantee::BestEffort);
425 }
426
427 #[test]
428 fn topology_start_initialized_has_no_attempt_and_rejects_assignment_ownership() {
429 let mut checkpoint = SourceCheckpoint::new();
430 checkpoint.set_offset("partition-0-next", "91");
431 let start = SourceStart::new(
432 ConnectorConfig::new("test"),
433 SourcePosition::Initialized {
434 checkpoint: checkpoint.clone(),
435 },
436 DeliveryGuarantee::AtLeastOnce,
437 )
438 .unwrap();
439 assert!(
440 matches!(start.into_parts().1, SourcePosition::Initialized { checkpoint }
441 if checkpoint.get_offset("partition-0-next") == Some("91")
442 && checkpoint.assignment_version().is_none())
443 );
444 checkpoint.bind_assignment_version(std::num::NonZeroU64::MIN);
445 assert!(matches!(SourceStart::new(
446 ConnectorConfig::new("test"), SourcePosition::Initialized { checkpoint },
447 DeliveryGuarantee::AtLeastOnce,
448 ), Err(ConnectorError::ConfigurationError(message)) if message.contains("unowned global")));
449 assert!(matches!(SourceStart::new(ConnectorConfig::new("test"),
450 SourcePosition::Initialized { checkpoint: SourceCheckpoint::new() },
451 DeliveryGuarantee::BestEffort), Err(ConnectorError::ConfigurationError(message))
452 if message.contains("requires at-least-once")));
453 }
454
455 #[test]
456 fn sink_contract_defaults_fail_closed() {
457 let contract = SinkContract::default();
458 assert_eq!(contract.consistency, SinkConsistency::Ephemeral);
459 assert_eq!(contract.topology, SinkTopology::Singleton);
460 assert_eq!(contract.input_mode, SinkInputMode::AppendOnly);
461 assert!(!contract.input_mode.accepts_full_changelog());
462 }
463
464 #[test]
465 fn coordinated_namespace_is_bounded_stable_and_sink_scoped() {
466 use laminar_core::checkpoint::checkpoint_manifest::PipelineIdentity;
467 const DEPLOYMENT: &str = "018f0000-0000-7000-8000-000000000001";
468
469 let first =
470 CoordinatedCommitNamespace::try_new(PipelineIdentity::empty(), DEPLOYMENT, "orders")
471 .unwrap();
472 let same =
473 CoordinatedCommitNamespace::try_new(PipelineIdentity::empty(), DEPLOYMENT, "orders")
474 .unwrap();
475 let other =
476 CoordinatedCommitNamespace::try_new(PipelineIdentity::empty(), DEPLOYMENT, "audit")
477 .unwrap();
478 let other_deployment = CoordinatedCommitNamespace::try_new(
479 PipelineIdentity::empty(),
480 "018f0000-0000-7000-8000-000000000002",
481 "orders",
482 )
483 .unwrap();
484
485 assert_eq!(first.external_key(), same.external_key());
486 assert_eq!(
487 first.external_key(),
488 "ldb-c3-d287aeafc8c53e0d469ee0f2d853e4e24481d12b2ab88068ca9c83522d29a3c3"
489 );
490 assert_ne!(first.external_key(), other.external_key());
491 assert_ne!(first.external_key(), other_deployment.external_key());
492 assert_eq!(first.external_key().len(), "ldb-c3-".len() + 64);
493 assert!(first
494 .external_key()
495 .bytes()
496 .all(|byte| byte.is_ascii_lowercase() || byte.is_ascii_digit() || byte == b'-'));
497 }
498
499 #[test]
500 fn coordinated_namespace_rejects_ambiguous_identity() {
501 const DEPLOYMENT: &str = "018f0000-0000-7000-8000-000000000001";
502
503 use laminar_core::checkpoint::checkpoint_manifest::{
504 PipelineIdentity, PIPELINE_IDENTITY_VERSION,
505 };
506 let malformed = PipelineIdentity {
507 canonical_version: PIPELINE_IDENTITY_VERSION,
508 sha256: "NOT-A-DIGEST".into(),
509 };
510 assert!(CoordinatedCommitNamespace::try_new(malformed, DEPLOYMENT, "orders").is_err());
511 let future_version = PipelineIdentity {
512 canonical_version: PIPELINE_IDENTITY_VERSION + 1,
513 sha256: PipelineIdentity::empty().sha256,
514 };
515 assert!(CoordinatedCommitNamespace::try_new(future_version, DEPLOYMENT, "orders").is_err());
516 assert!(
517 CoordinatedCommitNamespace::try_new(PipelineIdentity::empty(), DEPLOYMENT, "").is_err()
518 );
519 assert!(CoordinatedCommitNamespace::try_new(
520 PipelineIdentity::empty(),
521 "not-a-uuid",
522 "orders"
523 )
524 .is_err());
525 }
526
527 #[test]
528 fn coordinated_batch_fingerprint_covers_the_exact_ordered_cut() {
529 use laminar_core::checkpoint::checkpoint_manifest::PipelineIdentity;
530 use laminar_core::checkpoint::CheckpointAttempt;
531
532 let namespace = CoordinatedCommitNamespace::try_new(
533 PipelineIdentity::empty(),
534 "018f0000-0000-7000-8000-000000000001",
535 "orders",
536 )
537 .unwrap();
538 let attempt = CheckpointAttempt::new(8, 108);
539 let batch = CoordinatedCommitBatch {
540 namespace,
541 expected_predecessor: CoordinatedCommitCursor {
542 checkpoint_id: 107,
543 fencing_token: 3,
544 },
545 fencing_token: 4,
546 target: attempt,
547 entries: vec![CoordinatedCommitPayload {
548 attempt,
549 participant_id: 7,
550 payload: None,
551 }],
552 };
553 let expected = batch.exact_fingerprint();
554 assert_eq!(expected, batch.clone().exact_fingerprint());
555 assert_eq!(
556 expected,
557 [
558 0x4c, 0x2a, 0x38, 0x28, 0x25, 0x99, 0xd9, 0x73, 0xc5, 0xd6, 0x66, 0x92, 0x4f, 0x74,
559 0x0c, 0x41, 0x74, 0xa0, 0x0b, 0xd4, 0x59, 0xfb, 0x6b, 0xb7, 0x2f, 0xeb, 0xe6, 0x00,
560 0xe1, 0x7d, 0xac, 0x46,
561 ]
562 );
563
564 let mut variants = Vec::new();
565 let mut variant = batch.clone();
566 variant.namespace = CoordinatedCommitNamespace::try_new(
567 PipelineIdentity::empty(),
568 "018f0000-0000-7000-8000-000000000001",
569 "audit",
570 )
571 .unwrap();
572 variants.push(variant);
573 let mut variant = batch.clone();
574 variant.expected_predecessor.checkpoint_id -= 1;
575 variants.push(variant);
576 let mut variant = batch.clone();
577 variant.fencing_token += 1;
578 variants.push(variant);
579 let mut variant = batch.clone();
580 variant.target.epoch += 1;
581 variants.push(variant);
582 let mut variant = batch.clone();
583 variant.entries[0].attempt.checkpoint_id += 1;
584 variants.push(variant);
585 let mut variant = batch.clone();
586 variant.entries[0].participant_id += 1;
587 variants.push(variant);
588 let mut variant = batch;
589 variant.entries[0].payload = Some(Vec::new());
590 variants.push(variant);
591
592 assert!(variants
593 .into_iter()
594 .all(|variant| variant.exact_fingerprint() != expected));
595 }
596
597 fn valid_coordinated_batch() -> CoordinatedCommitBatch {
598 use laminar_core::checkpoint::checkpoint_manifest::PipelineIdentity;
599 use laminar_core::checkpoint::CheckpointAttempt;
600
601 let target = CheckpointAttempt::canonical(102);
602 CoordinatedCommitBatch {
603 namespace: CoordinatedCommitNamespace::try_new(
604 PipelineIdentity::empty(),
605 "018f0000-0000-7000-8000-000000000001",
606 "orders",
607 )
608 .unwrap(),
609 expected_predecessor: CoordinatedCommitCursor {
610 checkpoint_id: 101,
611 fencing_token: 1,
612 },
613 fencing_token: 2,
614 target,
615 entries: vec![CoordinatedCommitPayload {
616 attempt: target,
617 participant_id: 1,
618 payload: None,
619 }],
620 }
621 }
622
623 #[test]
624 fn coordinated_abort_batch_accepts_only_one_exact_attempt() {
625 use laminar_core::checkpoint::CheckpointAttempt;
626
627 let commit = valid_coordinated_batch();
628 let mut batch = CoordinatedAbortBatch {
629 namespace: commit.namespace,
630 fencing_token: commit.fencing_token,
631 target: commit.target,
632 entries: commit
633 .entries
634 .into_iter()
635 .map(|entry| CoordinatedAbortEntry {
636 attempt: entry.attempt,
637 participant_id: entry.participant_id,
638 descriptor: CoordinatedAbortDescriptor::Prepared(entry.payload),
639 artifact_intent: None,
640 })
641 .collect(),
642 };
643 batch.validate_shape().unwrap();
644
645 batch.entries.insert(
646 0,
647 CoordinatedAbortEntry {
648 attempt: CheckpointAttempt::canonical(101),
649 participant_id: 1,
650 descriptor: CoordinatedAbortDescriptor::Open,
651 artifact_intent: None,
652 },
653 );
654 assert!(batch
655 .validate_shape()
656 .unwrap_err()
657 .contains("exact target attempt"));
658 batch.entries.remove(0);
659 batch.fencing_token = 0;
660 assert!(batch
661 .validate_shape()
662 .unwrap_err()
663 .contains("fencing token"));
664 }
665
666 #[test]
667 fn coordinated_batch_rejects_noncanonical_target_before_other_shape_checks() {
668 use laminar_core::checkpoint::CheckpointAttempt;
669
670 for target in [
671 CheckpointAttempt::new(102, 103),
672 CheckpointAttempt::new(0, 0),
673 ] {
674 let mut batch = valid_coordinated_batch();
675 batch.target = target;
676 let error = batch.validate_shape().unwrap_err();
677 assert!(
678 error.contains("target must use one nonzero canonical checkpoint ID"),
679 "unexpected validation error: {error}"
680 );
681 }
682 }
683
684 #[test]
685 fn coordinated_batch_rejects_a_mutated_namespace() {
686 let mut batch = valid_coordinated_batch();
687 batch.namespace.sink_id.clear();
688
689 assert!(batch
690 .validate_shape()
691 .unwrap_err()
692 .contains("sink id cannot be empty"));
693 }
694
695 #[test]
696 fn coordinated_batch_rejects_noncanonical_entry_before_other_shape_checks() {
697 use laminar_core::checkpoint::CheckpointAttempt;
698
699 for attempt in [
700 CheckpointAttempt::new(101, 102),
701 CheckpointAttempt::new(0, 0),
702 ] {
703 let mut batch = valid_coordinated_batch();
704 batch.entries[0].attempt = attempt;
705 let error = batch.validate_shape().unwrap_err();
706 assert!(
707 error.contains("canonical checkpoint ID"),
708 "unexpected validation error: {error}"
709 );
710 }
711 let mut batch = valid_coordinated_batch();
712 batch.entries[0].participant_id = 0;
713 assert!(batch
714 .validate_shape()
715 .unwrap_err()
716 .contains("nonzero participant"));
717 }
718
719 #[test]
720 fn coordinated_batch_rejects_cursor_rollback_and_unproven_overlap() {
721 use laminar_core::checkpoint::checkpoint_manifest::PipelineIdentity;
722 use laminar_core::checkpoint::CheckpointAttempt;
723
724 let first = CheckpointAttempt::canonical(108);
725 let target = CheckpointAttempt::canonical(110);
726 let batch = CoordinatedCommitBatch {
727 namespace: CoordinatedCommitNamespace::try_new(
728 PipelineIdentity::empty(),
729 "018f0000-0000-7000-8000-000000000001",
730 "orders",
731 )
732 .unwrap(),
733 expected_predecessor: CoordinatedCommitCursor {
734 checkpoint_id: 107,
735 fencing_token: 3,
736 },
737 fencing_token: 4,
738 target,
739 entries: vec![
740 CoordinatedCommitPayload {
741 attempt: first,
742 participant_id: 1,
743 payload: None,
744 },
745 CoordinatedCommitPayload {
746 attempt: target,
747 participant_id: 1,
748 payload: None,
749 },
750 ],
751 };
752
753 let cursor = |checkpoint_id, fencing_token| {
754 Some(CoordinatedCommitCursor {
755 checkpoint_id,
756 fencing_token,
757 })
758 };
759 assert!(batch.validate_observed_cursor(cursor(106, 3)).is_err());
760 assert!(batch.validate_observed_cursor(cursor(109, 3)).is_err());
761 assert!(batch.validate_observed_cursor(cursor(107, 2)).is_err());
762 assert!(batch.validate_observed_cursor(cursor(107, 3)).is_ok());
763 assert!(batch.validate_observed_cursor(cursor(108, 3)).is_ok());
764 assert!(batch.validate_observed_cursor(cursor(110, 4)).is_ok());
765 assert!(batch.validate_observed_cursor(cursor(110, 3)).is_err());
766 assert!(batch.validate_observed_cursor(cursor(108, 5)).is_err());
767 }
768
769 #[test]
770 fn coordinated_batch_requires_unique_canonical_attempt_participants() {
771 use laminar_core::checkpoint::checkpoint_manifest::PipelineIdentity;
772 use laminar_core::checkpoint::CheckpointAttempt;
773
774 let namespace = CoordinatedCommitNamespace::try_new(
775 PipelineIdentity::empty(),
776 "018f0000-0000-7000-8000-000000000001",
777 "orders",
778 )
779 .unwrap();
780 let target = CheckpointAttempt::canonical(102);
781 let batch = |entries| CoordinatedCommitBatch {
782 namespace: namespace.clone(),
783 expected_predecessor: CoordinatedCommitCursor {
784 checkpoint_id: 100,
785 fencing_token: 1,
786 },
787 fencing_token: 2,
788 target,
789 entries,
790 };
791 let payload = |attempt, participant_id| CoordinatedCommitPayload {
792 attempt,
793 participant_id,
794 payload: None,
795 };
796
797 let duplicate = batch(vec![payload(target, 1), payload(target, 1)]);
798 assert!(duplicate
799 .validate_shape()
800 .unwrap_err()
801 .contains("duplicate"));
802
803 let out_of_order = batch(vec![payload(target, 2), payload(target, 1)]);
804 assert!(out_of_order
805 .validate_shape()
806 .unwrap_err()
807 .contains("out-of-order"));
808
809 let noncanonical = batch(vec![
810 payload(CheckpointAttempt::new(3, 101), 1),
811 payload(target, 2),
812 ]);
813 assert!(noncanonical
814 .validate_shape()
815 .unwrap_err()
816 .contains("canonical checkpoint ID"));
817 }
818
819 #[test]
820 fn coordinated_batch_entry_limit_accepts_max_and_rejects_max_plus_one() {
821 use laminar_core::checkpoint::checkpoint_manifest::PipelineIdentity;
822 use laminar_core::checkpoint::CheckpointAttempt;
823
824 let namespace = CoordinatedCommitNamespace::try_new(
825 PipelineIdentity::empty(),
826 "018f0000-0000-7000-8000-000000000001",
827 "orders",
828 )
829 .unwrap();
830 let target = CheckpointAttempt::canonical(101);
831 let make_batch = |count: usize| CoordinatedCommitBatch {
832 namespace: namespace.clone(),
833 expected_predecessor: CoordinatedCommitCursor {
834 checkpoint_id: 0,
835 fencing_token: 0,
836 },
837 fencing_token: 1,
838 target,
839 entries: (1..=count)
840 .map(|participant_id| CoordinatedCommitPayload {
841 attempt: target,
842 participant_id: participant_id as u64,
843 payload: None,
844 })
845 .collect(),
846 };
847
848 assert!(make_batch(MAX_COORDINATED_COMMIT_BATCH_ENTRIES - 1)
849 .validate_shape()
850 .is_ok());
851 assert!(make_batch(MAX_COORDINATED_COMMIT_BATCH_ENTRIES)
852 .validate_shape()
853 .is_ok());
854 assert!(make_batch(MAX_COORDINATED_COMMIT_BATCH_ENTRIES + 1)
855 .validate_shape()
856 .is_err());
857 }
858
859 struct DefaultPreCommitSink {
860 coordinated: bool,
861 }
862
863 #[async_trait]
864 impl SinkConnector for DefaultPreCommitSink {
865 async fn open(&mut self, _config: &ConnectorConfig) -> Result<(), ConnectorError> {
866 Ok(())
867 }
868 async fn write_batch(
869 &mut self,
870 _batch: &RecordBatch,
871 ) -> Result<WriteResult, ConnectorError> {
872 Ok(WriteResult::new(0, 0))
873 }
874 fn schema(&self) -> SchemaRef {
875 test_schema()
876 }
877 fn suggested_write_timeout(&self) -> std::time::Duration {
878 std::time::Duration::from_secs(5)
879 }
880 fn as_coordinated_committer(&self) -> Option<&dyn CoordinatedCommitter> {
881 self.coordinated
882 .then_some(self as &dyn CoordinatedCommitter)
883 }
884 async fn close(&mut self) -> Result<(), ConnectorError> {
885 Ok(())
886 }
887 }
888
889 #[async_trait]
890 impl CoordinatedCommitter for DefaultPreCommitSink {
891 async fn commit_aggregated(
892 &self,
893 _batch: CoordinatedCommitBatch,
894 _context: CoordinatedCommitContext,
895 ) -> Result<(), ConnectorError> {
896 Ok(())
897 }
898
899 async fn committed_cursor(
900 &self,
901 _namespace: &CoordinatedCommitNamespace,
902 ) -> Result<Option<CoordinatedCommitCursor>, ConnectorError> {
903 Ok(None)
904 }
905 }
906
907 #[tokio::test]
908 async fn default_pre_commit_rejects_coordinated_sink() {
909 let mut sink = DefaultPreCommitSink { coordinated: true };
910 assert!(matches!(
911 sink.pre_commit(1).await,
912 Err(ConnectorError::ConfigurationError(_))
913 ));
914 }
915
916 #[tokio::test]
917 async fn default_pre_commit_ok_for_non_coordinated_sink() {
918 let mut sink = DefaultPreCommitSink { coordinated: false };
919 assert!(matches!(sink.pre_commit(1).await, Ok(None)));
920 }
921}