1use std::collections::BTreeMap;
2
3use crate::model::{
4 ColumnSchema, ColumnValue, EntityDifference, EntityKey, Operation, ScalarValue, TableSchema,
5};
6use crate::{PgLsn, PostgresConnectError};
7
8#[derive(Debug, Clone, PartialEq, Eq)]
9pub struct RelationEvent {
10 pub relation_id: u32,
11 pub schema: String,
12 pub table: String,
13 pub columns: Vec<ColumnSchema>,
14}
15
16#[derive(Debug, Clone, Copy, PartialEq, Eq)]
17pub enum RowTupleKind {
18 Full,
19 Key,
20}
21
22#[derive(Debug, Clone, PartialEq, Eq)]
23pub enum RowEventKind {
24 Insert,
25 Update {
26 old: Vec<ColumnValue>,
27 old_tuple_kind: RowTupleKind,
28 },
29 Delete {
30 tuple_kind: RowTupleKind,
31 },
32}
33
34#[derive(Debug, Clone, PartialEq, Eq)]
35pub struct RowEvent {
36 pub relation_id: u32,
37 pub lsn: PgLsn,
38 pub commit_lsn: Option<PgLsn>,
39 pub txid: Option<i64>,
40 pub commit_timestamp_ms: Option<i64>,
41 pub kind: RowEventKind,
42 pub values: Vec<ColumnValue>,
43}
44
45#[derive(Debug, Clone, PartialEq, Eq)]
46pub enum DecodedMessage {
47 Begin {
48 final_lsn: PgLsn,
49 xid: i64,
50 },
51 Commit {
52 commit_lsn: PgLsn,
53 end_lsn: PgLsn,
54 commit_timestamp_ms: i64,
55 },
56 Relation(RelationEvent),
57 Row(RowEvent),
58 Keepalive,
59}
60
61pub fn decode_pgoutput_message(
62 bytes: &[u8],
63 lsn: PgLsn,
64 txid: Option<i64>,
65) -> Result<DecodedMessage, PostgresConnectError> {
66 let mut reader = PgOutputReader::new(bytes);
67 let tag = reader.read_u8("message tag")?;
68
69 let message = match tag {
70 b'B' => {
71 let final_lsn = PgLsn(reader.read_u64("begin final_lsn")?);
72 let _commit_time = reader.read_i64("begin commit_time")?;
73 let xid = i64::from(reader.read_u32("begin xid")?);
74 DecodedMessage::Begin { final_lsn, xid }
75 }
76 b'C' => {
77 let _flags = reader.read_u8("commit flags")?;
78 let commit_lsn = PgLsn(reader.read_u64("commit commit_lsn")?);
79 let end_lsn = PgLsn(reader.read_u64("commit end_lsn")?);
80 let commit_timestamp_ms =
81 postgres_timestamp_micros_to_unix_ms(reader.read_i64("commit commit_time")?);
82 DecodedMessage::Commit {
83 commit_lsn,
84 end_lsn,
85 commit_timestamp_ms,
86 }
87 }
88 b'R' => DecodedMessage::Relation(decode_relation(&mut reader)?),
89 b'I' => DecodedMessage::Row(decode_insert(&mut reader, lsn, txid)?),
90 b'U' => DecodedMessage::Row(decode_update(&mut reader, lsn, txid)?),
91 b'D' => DecodedMessage::Row(decode_delete(&mut reader, lsn, txid)?),
92 tag => {
93 return Err(PostgresConnectError::Backend(format!(
94 "unsupported pgoutput message tag {:?}",
95 char::from(tag)
96 )));
97 }
98 };
99
100 reader.finish()?;
101 Ok(message)
102}
103
104fn decode_relation(reader: &mut PgOutputReader<'_>) -> Result<RelationEvent, PostgresConnectError> {
105 let relation_id = reader.read_u32("relation id")?;
106 let schema = reader.read_cstr("relation namespace")?;
107 let table = reader.read_cstr("relation name")?;
108 let _replica_identity = reader.read_u8("relation replica identity")?;
109 let column_count = reader.read_count("relation column count")?;
110 let mut columns = Vec::with_capacity(column_count);
111
112 for index in 0..column_count {
113 let flags = reader.read_u8("relation column flags")?;
114 let name = reader.read_cstr("relation column name")?;
115 let type_oid = reader.read_i32("relation column type oid")?;
116 let _atttypmod = reader.read_i32("relation column atttypmod")?;
117 columns.push(ColumnSchema {
118 name,
119 type_name: type_name(type_oid).to_owned(),
120 key: flags & 1 == 1,
121 });
122
123 if columns[index].name.is_empty() {
124 return Err(PostgresConnectError::Backend(format!(
125 "invalid pgoutput relation column {index}: empty name"
126 )));
127 }
128 }
129
130 Ok(RelationEvent {
131 relation_id,
132 schema,
133 table,
134 columns,
135 })
136}
137
138fn decode_insert(
139 reader: &mut PgOutputReader<'_>,
140 lsn: PgLsn,
141 txid: Option<i64>,
142) -> Result<RowEvent, PostgresConnectError> {
143 let relation_id = reader.read_u32("insert relation id")?;
144 reader.expect_u8(b'N', "insert tuple tag N")?;
145 let values = decode_tuple(reader)?;
146
147 Ok(RowEvent {
148 relation_id,
149 lsn,
150 commit_lsn: None,
151 txid,
152 commit_timestamp_ms: None,
153 kind: RowEventKind::Insert,
154 values,
155 })
156}
157
158fn decode_update(
159 reader: &mut PgOutputReader<'_>,
160 lsn: PgLsn,
161 txid: Option<i64>,
162) -> Result<RowEvent, PostgresConnectError> {
163 let relation_id = reader.read_u32("update relation id")?;
164 let tag = reader.read_u8("update tuple tag")?;
165 let (old, old_tuple_kind, new_tag) = match tag {
166 b'K' | b'O' => (
167 decode_tuple(reader)?,
168 tuple_kind(tag),
169 reader.read_u8("update new tuple tag")?,
170 ),
171 b'N' => (Vec::new(), RowTupleKind::Full, b'N'),
172 tag => {
173 return Err(PostgresConnectError::Backend(format!(
174 "invalid pgoutput update tuple tag {:?}",
175 char::from(tag)
176 )));
177 }
178 };
179
180 if new_tag != b'N' {
181 return Err(PostgresConnectError::Backend(format!(
182 "invalid pgoutput update new tuple tag {:?}",
183 char::from(new_tag)
184 )));
185 }
186
187 let values = decode_tuple(reader)?;
188 Ok(RowEvent {
189 relation_id,
190 lsn,
191 commit_lsn: None,
192 txid,
193 commit_timestamp_ms: None,
194 kind: RowEventKind::Update {
195 old,
196 old_tuple_kind,
197 },
198 values,
199 })
200}
201
202fn decode_delete(
203 reader: &mut PgOutputReader<'_>,
204 lsn: PgLsn,
205 txid: Option<i64>,
206) -> Result<RowEvent, PostgresConnectError> {
207 let relation_id = reader.read_u32("delete relation id")?;
208 let tag = reader.read_u8("delete tuple tag")?;
209 if !matches!(tag, b'K' | b'O') {
210 return Err(PostgresConnectError::Backend(format!(
211 "invalid pgoutput delete tuple tag {:?}",
212 char::from(tag)
213 )));
214 }
215 let values = decode_tuple(reader)?;
216
217 Ok(RowEvent {
218 relation_id,
219 lsn,
220 commit_lsn: None,
221 txid,
222 commit_timestamp_ms: None,
223 kind: RowEventKind::Delete {
224 tuple_kind: tuple_kind(tag),
225 },
226 values,
227 })
228}
229
230fn tuple_kind(tag: u8) -> RowTupleKind {
231 if tag == b'K' {
232 RowTupleKind::Key
233 } else {
234 RowTupleKind::Full
235 }
236}
237
238fn decode_tuple(reader: &mut PgOutputReader<'_>) -> Result<Vec<ColumnValue>, PostgresConnectError> {
239 let value_count = reader.read_count("tuple value count")?;
240 let mut values = Vec::with_capacity(value_count);
241
242 for index in 0..value_count {
243 let value_tag = reader.read_u8("tuple value tag")?;
244 let value = match value_tag {
245 b'n' => ScalarValue::Null,
246 b'u' => ScalarValue::UnchangedToast,
247 b't' => {
248 let bytes = reader.read_len_bytes("text tuple value")?;
249 let value = std::str::from_utf8(bytes).map_err(|error| {
250 PostgresConnectError::Backend(format!(
251 "invalid utf8 in pgoutput text tuple value: {error}"
252 ))
253 })?;
254 ScalarValue::Text(value.to_owned())
255 }
256 b'b' => ScalarValue::Bytes(reader.read_len_bytes("binary tuple value")?.to_vec()),
257 tag => {
258 return Err(PostgresConnectError::Backend(format!(
259 "unsupported pgoutput tuple value tag {:?}",
260 char::from(tag)
261 )));
262 }
263 };
264
265 values.push(ColumnValue {
266 name: format!("col{index}"),
267 value,
268 });
269 }
270
271 Ok(values)
272}
273
274fn type_name(type_oid: i32) -> &'static str {
275 match type_oid {
276 16 => "bool",
277 17 => "bytea",
278 20 => "int8",
279 21 => "int2",
280 23 => "int4",
281 25 => "text",
282 700 => "float4",
283 701 => "float8",
284 1043 => "varchar",
285 1700 => "numeric",
286 _ => "unknown",
287 }
288}
289
290fn postgres_timestamp_micros_to_unix_ms(value: i64) -> i64 {
291 const POSTGRES_EPOCH_UNIX_MS: i64 = 946_684_800_000;
292 POSTGRES_EPOCH_UNIX_MS + value.div_euclid(1_000)
293}
294
295struct PgOutputReader<'a> {
296 bytes: &'a [u8],
297 position: usize,
298}
299
300impl<'a> PgOutputReader<'a> {
301 fn new(bytes: &'a [u8]) -> Self {
302 Self { bytes, position: 0 }
303 }
304
305 fn finish(&self) -> Result<(), PostgresConnectError> {
306 if self.position == self.bytes.len() {
307 Ok(())
308 } else {
309 Err(PostgresConnectError::Backend(format!(
310 "invalid pgoutput message: {} trailing bytes",
311 self.bytes.len() - self.position
312 )))
313 }
314 }
315
316 fn read_u8(&mut self, field: &str) -> Result<u8, PostgresConnectError> {
317 let bytes = self.read_exact(1, field)?;
318 Ok(bytes[0])
319 }
320
321 fn expect_u8(&mut self, expected: u8, field: &str) -> Result<(), PostgresConnectError> {
322 let actual = self.read_u8(field)?;
323 if actual == expected {
324 Ok(())
325 } else {
326 Err(PostgresConnectError::Backend(format!(
327 "invalid pgoutput {field}: expected {:?}, got {:?}",
328 char::from(expected),
329 char::from(actual)
330 )))
331 }
332 }
333
334 fn read_i16(&mut self, field: &str) -> Result<i16, PostgresConnectError> {
335 let bytes = self.read_exact(2, field)?;
336 Ok(i16::from_be_bytes([bytes[0], bytes[1]]))
337 }
338
339 fn read_count(&mut self, field: &str) -> Result<usize, PostgresConnectError> {
340 let count = self.read_i16(field)?;
341 usize::try_from(count).map_err(|_| {
342 PostgresConnectError::Backend(format!("invalid pgoutput {field}: negative count"))
343 })
344 }
345
346 fn read_i32(&mut self, field: &str) -> Result<i32, PostgresConnectError> {
347 let bytes = self.read_exact(4, field)?;
348 Ok(i32::from_be_bytes([bytes[0], bytes[1], bytes[2], bytes[3]]))
349 }
350
351 fn read_u32(&mut self, field: &str) -> Result<u32, PostgresConnectError> {
352 let bytes = self.read_exact(4, field)?;
353 Ok(u32::from_be_bytes([bytes[0], bytes[1], bytes[2], bytes[3]]))
354 }
355
356 fn read_i64(&mut self, field: &str) -> Result<i64, PostgresConnectError> {
357 let bytes = self.read_exact(8, field)?;
358 Ok(i64::from_be_bytes([
359 bytes[0], bytes[1], bytes[2], bytes[3], bytes[4], bytes[5], bytes[6], bytes[7],
360 ]))
361 }
362
363 fn read_u64(&mut self, field: &str) -> Result<u64, PostgresConnectError> {
364 let bytes = self.read_exact(8, field)?;
365 Ok(u64::from_be_bytes([
366 bytes[0], bytes[1], bytes[2], bytes[3], bytes[4], bytes[5], bytes[6], bytes[7],
367 ]))
368 }
369
370 fn read_len_bytes(&mut self, field: &str) -> Result<&'a [u8], PostgresConnectError> {
371 let length = self.read_i32(field)?;
372 let length = usize::try_from(length).map_err(|_| {
373 PostgresConnectError::Backend(format!("invalid pgoutput {field}: negative length"))
374 })?;
375 self.read_exact(length, field)
376 }
377
378 fn read_cstr(&mut self, field: &str) -> Result<String, PostgresConnectError> {
379 let remaining = &self.bytes[self.position..];
380 let Some(length) = remaining.iter().position(|byte| *byte == 0) else {
381 return Err(PostgresConnectError::Backend(format!(
382 "truncated pgoutput {field}: missing null terminator"
383 )));
384 };
385 let bytes = &remaining[..length];
386 self.position += length + 1;
387 let value = std::str::from_utf8(bytes).map_err(|error| {
388 PostgresConnectError::Backend(format!("invalid utf8 in pgoutput {field}: {error}"))
389 })?;
390 Ok(value.to_owned())
391 }
392
393 fn read_exact(&mut self, length: usize, field: &str) -> Result<&'a [u8], PostgresConnectError> {
394 let end = self.position.checked_add(length).ok_or_else(|| {
395 PostgresConnectError::Backend(format!("invalid pgoutput {field}: length overflow"))
396 })?;
397 if end > self.bytes.len() {
398 return Err(PostgresConnectError::Backend(format!(
399 "truncated pgoutput {field}: needed {length} bytes, have {}",
400 self.bytes.len().saturating_sub(self.position)
401 )));
402 }
403
404 let bytes = &self.bytes[self.position..end];
405 self.position = end;
406 Ok(bytes)
407 }
408}
409
410#[derive(Debug, Clone, Default)]
411pub struct RelationCache {
412 relations: BTreeMap<u32, TableSchema>,
413}
414
415impl RelationCache {
416 pub fn apply_relation(&mut self, event: RelationEvent) {
417 self.relations.insert(
418 event.relation_id,
419 TableSchema {
420 schema: event.schema,
421 table: event.table,
422 columns: event.columns,
423 },
424 );
425 }
426
427 pub fn translate(&self, event: RowEvent) -> Result<EntityDifference, PostgresConnectError> {
428 let schema = self.relations.get(&event.relation_id).ok_or_else(|| {
429 PostgresConnectError::Backend(format!(
430 "missing relation metadata for relation id {}",
431 event.relation_id
432 ))
433 })?;
434 let table = format!("{}.{}", schema.schema, schema.table);
435 let lsn = event.commit_lsn.unwrap_or(event.lsn);
436
437 let (op, before, after, key_columns) = match event.kind {
438 RowEventKind::Insert => {
439 let values =
440 normalize_values(schema, event.values, RowTupleKind::Full, "row values")?;
441 let key_columns = extract_key_columns(schema, &values)?;
442 (Operation::Insert, Vec::new(), values, key_columns)
443 }
444 RowEventKind::Update {
445 old,
446 old_tuple_kind,
447 } => {
448 let values =
449 normalize_values(schema, event.values, RowTupleKind::Full, "row values")?;
450 let key_columns = extract_key_columns(schema, &values)?;
451 let old = if old.is_empty() {
452 old
453 } else {
454 normalize_values(schema, old, old_tuple_kind, "old row values")?
455 };
456 if has_any_key_column(schema, &old) {
457 let old_key_columns = extract_key_columns(schema, &old)?;
458 if old_key_columns != key_columns {
459 return Err(PostgresConnectError::Backend(
460 "key-changing updates are not supported".to_owned(),
461 ));
462 }
463 }
464
465 (Operation::Update, old, values, key_columns)
466 }
467 RowEventKind::Delete { tuple_kind } => {
468 let values = normalize_values(schema, event.values, tuple_kind, "row values")?;
469 let key_columns = extract_key_columns(schema, &values)?;
470 (Operation::Delete, values, Vec::new(), key_columns)
471 }
472 };
473 let key = EntityKey {
474 table: table.clone(),
475 columns: key_columns,
476 };
477
478 Ok(EntityDifference {
479 table,
480 key,
481 op,
482 before,
483 after,
484 lsn,
485 txid: event.txid,
486 commit_timestamp_ms: event.commit_timestamp_ms,
487 schema: schema.clone(),
488 })
489 }
490}
491
492fn normalize_values(
493 schema: &TableSchema,
494 values: Vec<ColumnValue>,
495 tuple_kind: RowTupleKind,
496 label: &str,
497) -> Result<Vec<ColumnValue>, PostgresConnectError> {
498 let expected_columns: Vec<&ColumnSchema> = match tuple_kind {
499 RowTupleKind::Full => schema.columns.iter().collect(),
500 RowTupleKind::Key => schema.columns.iter().filter(|column| column.key).collect(),
501 };
502
503 if values.len() != expected_columns.len() {
504 return Err(PostgresConnectError::Backend(format!(
505 "column count mismatch for {}.{} {label}: decoded {} values but expected {} columns",
506 schema.schema,
507 schema.table,
508 values.len(),
509 expected_columns.len()
510 )));
511 }
512
513 if values.iter().all(|value| {
514 expected_columns
515 .iter()
516 .any(|column| column.name == value.name)
517 }) {
518 return values
519 .into_iter()
520 .map(|value| {
521 let column = expected_columns
522 .iter()
523 .find(|column| column.name == value.name)
524 .expect("validated column name should be present");
525 Ok(ColumnValue {
526 name: value.name,
527 value: coerce_value(value.value, &column.type_name)?,
528 })
529 })
530 .collect();
531 }
532
533 values
534 .into_iter()
535 .zip(expected_columns)
536 .map(|(value, column)| {
537 Ok(ColumnValue {
538 name: column.name.clone(),
539 value: coerce_value(value.value, &column.type_name)?,
540 })
541 })
542 .collect::<Result<Vec<_>, _>>()
543}
544
545fn coerce_value(value: ScalarValue, type_name: &str) -> Result<ScalarValue, PostgresConnectError> {
546 let ScalarValue::Text(text) = value else {
547 return Ok(value);
548 };
549
550 match type_name {
551 "bool" => coerce_bool(&text),
552 "int2" | "int4" | "int8" => text.parse::<i64>().map(ScalarValue::Int).map_err(|error| {
553 PostgresConnectError::Backend(format!(
554 "invalid {type_name} pgoutput text value {text:?}: {error}"
555 ))
556 }),
557 "float4" | "float8" | "numeric" => Ok(ScalarValue::Float(text)),
558 "bytea" => Ok(ScalarValue::Bytes(coerce_bytea(&text)?)),
559 _ => Ok(ScalarValue::Text(text)),
560 }
561}
562
563fn coerce_bool(text: &str) -> Result<ScalarValue, PostgresConnectError> {
564 match text {
565 "t" | "true" | "1" => Ok(ScalarValue::Bool(true)),
566 "f" | "false" | "0" => Ok(ScalarValue::Bool(false)),
567 _ => Err(PostgresConnectError::Backend(format!(
568 "invalid bool pgoutput text value {text:?}"
569 ))),
570 }
571}
572
573fn coerce_bytea(text: &str) -> Result<Vec<u8>, PostgresConnectError> {
574 let Some(hex) = text.strip_prefix("\\x") else {
575 return Ok(text.as_bytes().to_vec());
576 };
577 if hex.len() % 2 != 0 {
578 return Err(PostgresConnectError::Backend(format!(
579 "invalid bytea pgoutput text value {text:?}: odd hex length"
580 )));
581 }
582
583 hex.as_bytes()
584 .chunks_exact(2)
585 .map(|chunk| {
586 let pair = std::str::from_utf8(chunk).map_err(|error| {
587 PostgresConnectError::Backend(format!(
588 "invalid bytea pgoutput text value {text:?}: {error}"
589 ))
590 })?;
591 u8::from_str_radix(pair, 16).map_err(|error| {
592 PostgresConnectError::Backend(format!(
593 "invalid bytea pgoutput text value {text:?}: {error}"
594 ))
595 })
596 })
597 .collect()
598}
599
600fn extract_key_columns(
601 schema: &TableSchema,
602 values: &[ColumnValue],
603) -> Result<Vec<ColumnValue>, PostgresConnectError> {
604 schema
605 .columns
606 .iter()
607 .filter(|column| column.key)
608 .map(|column| {
609 values
610 .iter()
611 .find(|value| value.name == column.name)
612 .cloned()
613 .ok_or_else(|| {
614 PostgresConnectError::Backend(format!(
615 "missing key column {:?} for {}.{}",
616 column.name, schema.schema, schema.table
617 ))
618 })
619 })
620 .collect()
621}
622
623fn has_any_key_column(schema: &TableSchema, values: &[ColumnValue]) -> bool {
624 schema
625 .columns
626 .iter()
627 .filter(|column| column.key)
628 .any(|column| values.iter().any(|value| value.name == column.name))
629}
630
631#[cfg(test)]
632mod tests {
633 use assert2::check;
634
635 use super::{
636 RelationCache, RelationEvent, RowEvent, RowEventKind, RowTupleKind, coerce_bytea,
637 coerce_value,
638 };
639 use crate::PgLsn;
640 use crate::PostgresConnectError;
641 use crate::model::{ColumnSchema, ColumnValue, Operation, ScalarValue, TableSchema};
642
643 fn orders_relation(type_name: &str) -> RelationEvent {
644 RelationEvent {
645 relation_id: 7,
646 schema: "public".to_owned(),
647 table: "orders".to_owned(),
648 columns: vec![
649 ColumnSchema {
650 name: "id".to_owned(),
651 type_name: "int8".to_owned(),
652 key: true,
653 },
654 ColumnSchema {
655 name: "status".to_owned(),
656 type_name: type_name.to_owned(),
657 key: false,
658 },
659 ],
660 }
661 }
662
663 fn id(value: i64) -> ColumnValue {
664 ColumnValue {
665 name: "id".to_owned(),
666 value: ScalarValue::Int(value),
667 }
668 }
669
670 fn status(value: &str) -> ColumnValue {
671 ColumnValue {
672 name: "status".to_owned(),
673 value: ScalarValue::Text(value.to_owned()),
674 }
675 }
676
677 fn tenant_id(value: i64) -> ColumnValue {
678 ColumnValue {
679 name: "tenant_id".to_owned(),
680 value: ScalarValue::Int(value),
681 }
682 }
683
684 fn composite_key_orders_relation() -> RelationEvent {
685 RelationEvent {
686 relation_id: 7,
687 schema: "public".to_owned(),
688 table: "orders".to_owned(),
689 columns: vec![
690 ColumnSchema {
691 name: "tenant_id".to_owned(),
692 type_name: "int8".to_owned(),
693 key: true,
694 },
695 ColumnSchema {
696 name: "id".to_owned(),
697 type_name: "int8".to_owned(),
698 key: true,
699 },
700 ColumnSchema {
701 name: "status".to_owned(),
702 type_name: "text".to_owned(),
703 key: false,
704 },
705 ],
706 }
707 }
708
709 #[test]
710 fn insert_translates_to_entity_difference_with_key() {
711 let mut cache = RelationCache::default();
712 cache.apply_relation(orders_relation("text"));
713
714 let values = vec![id(42), status("paid")];
715 let difference = cache
716 .translate(RowEvent {
717 relation_id: 7,
718 lsn: PgLsn(0x16_b374_d848),
719 commit_lsn: None,
720 txid: Some(99),
721 commit_timestamp_ms: Some(1_700_000_000_000),
722 kind: RowEventKind::Insert,
723 values: values.clone(),
724 })
725 .expect("relation should translate");
726
727 check!(difference.table == "public.orders");
728 check!(difference.key.table == "public.orders");
729 check!(difference.key.columns == vec![id(42)]);
730 check!(difference.op == Operation::Insert);
731 check!(difference.before == Vec::new());
732 check!(difference.after == values);
733 check!(difference.lsn == PgLsn(0x16_b374_d848));
734 check!(difference.txid == Some(99));
735 check!(difference.commit_timestamp_ms == Some(1_700_000_000_000));
736 check!(difference.schema.table == "orders");
737 check!(difference.schema.columns[0].key);
738 }
739
740 #[test]
741 fn delete_translates_to_before_only_difference() {
742 let mut cache = RelationCache::default();
743 cache.apply_relation(orders_relation("text"));
744
745 let values = vec![id(42), status("cancelled")];
746 let difference = cache
747 .translate(RowEvent {
748 relation_id: 7,
749 lsn: PgLsn(0x2a),
750 commit_lsn: None,
751 txid: None,
752 commit_timestamp_ms: None,
753 kind: RowEventKind::Delete {
754 tuple_kind: RowTupleKind::Full,
755 },
756 values: values.clone(),
757 })
758 .expect("relation should translate");
759
760 check!(difference.table == "public.orders");
761 check!(difference.key.columns == vec![id(42)]);
762 check!(difference.op == Operation::Delete);
763 check!(difference.before == values);
764 check!(difference.after == Vec::new());
765 }
766
767 #[test]
768 fn relation_refresh_changes_table_schema() {
769 let mut cache = RelationCache::default();
770 cache.apply_relation(orders_relation("text"));
771 cache.apply_relation(orders_relation("varchar"));
772
773 let difference = cache
774 .translate(RowEvent {
775 relation_id: 7,
776 lsn: PgLsn(0x2b),
777 commit_lsn: None,
778 txid: None,
779 commit_timestamp_ms: None,
780 kind: RowEventKind::Insert,
781 values: vec![id(7), status("new")],
782 })
783 .expect("relation should translate");
784
785 check!(difference.schema.columns[1].type_name == "varchar");
786 }
787
788 #[test]
789 fn composite_key_output_follows_schema_order_even_if_row_values_are_out_of_order() {
790 let mut cache = RelationCache::default();
791 cache.apply_relation(composite_key_orders_relation());
792
793 let difference = cache
794 .translate(RowEvent {
795 relation_id: 7,
796 lsn: PgLsn(0x2c),
797 commit_lsn: None,
798 txid: None,
799 commit_timestamp_ms: None,
800 kind: RowEventKind::Insert,
801 values: vec![status("new"), id(7), tenant_id(3)],
802 })
803 .expect("relation should translate");
804
805 check!(difference.key.columns == vec![tenant_id(3), id(7)]);
806 }
807
808 #[test]
809 fn missing_key_column_is_an_error() {
810 let mut cache = RelationCache::default();
811 cache.apply_relation(orders_relation("text"));
812
813 let result = cache.translate(RowEvent {
814 relation_id: 7,
815 lsn: PgLsn(0x2d),
816 commit_lsn: None,
817 txid: None,
818 commit_timestamp_ms: None,
819 kind: RowEventKind::Insert,
820 values: vec![status("new"), status("paid")],
821 });
822
823 let error = result.expect_err("missing key should fail");
824 match error {
825 PostgresConnectError::Backend(message) => {
826 check!(message.contains("missing key column"));
827 check!(message.contains("id"));
828 }
829 error => panic!("expected backend error, got {error:?}"),
830 }
831 }
832
833 #[test]
834 fn non_key_update_translates_with_before_after_and_stable_key() {
835 let mut cache = RelationCache::default();
836 cache.apply_relation(orders_relation("text"));
837
838 let old = vec![id(42), status("new")];
839 let new = vec![id(42), status("paid")];
840 let difference = cache
841 .translate(RowEvent {
842 relation_id: 7,
843 lsn: PgLsn(0x2e),
844 commit_lsn: None,
845 txid: Some(100),
846 commit_timestamp_ms: Some(1_700_000_000_001),
847 kind: RowEventKind::Update {
848 old: old.clone(),
849 old_tuple_kind: RowTupleKind::Full,
850 },
851 values: new.clone(),
852 })
853 .expect("relation should translate");
854
855 check!(difference.op == Operation::Update);
856 check!(difference.before == old);
857 check!(difference.after == new);
858 check!(difference.key.columns == vec![id(42)]);
859 }
860
861 #[test]
862 fn key_changing_update_is_an_error() {
863 let mut cache = RelationCache::default();
864 cache.apply_relation(orders_relation("text"));
865
866 let result = cache.translate(RowEvent {
867 relation_id: 7,
868 lsn: PgLsn(0x2f),
869 commit_lsn: None,
870 txid: None,
871 commit_timestamp_ms: None,
872 kind: RowEventKind::Update {
873 old: vec![id(41), status("new")],
874 old_tuple_kind: RowTupleKind::Full,
875 },
876 values: vec![id(42), status("paid")],
877 });
878
879 let error = result.expect_err("key-changing update should fail");
880 match error {
881 PostgresConnectError::Backend(message) => {
882 check!(message == "key-changing updates are not supported");
883 }
884 error => panic!("expected backend error, got {error:?}"),
885 }
886 }
887
888 #[test]
889 fn decoded_placeholder_values_bind_to_schema_names_before_key_extraction() {
890 let mut cache = RelationCache::default();
891 cache.apply_relation(orders_relation("text"));
892
893 let difference = cache
894 .translate(RowEvent {
895 relation_id: 7,
896 lsn: PgLsn(0x30),
897 commit_lsn: None,
898 txid: None,
899 commit_timestamp_ms: None,
900 kind: RowEventKind::Insert,
901 values: vec![
902 ColumnValue {
903 name: "col0".to_owned(),
904 value: ScalarValue::Text("42".to_owned()),
905 },
906 ColumnValue {
907 name: "col1".to_owned(),
908 value: ScalarValue::Text("paid".to_owned()),
909 },
910 ],
911 })
912 .expect("decoded placeholders should bind to relation schema");
913
914 check!(difference.key.columns[0].name == "id");
915 check!(difference.key.columns[0].value == ScalarValue::Int(42));
916 check!(difference.after[1].name == "status");
917 }
918
919 #[test]
920 fn decoded_placeholder_count_mismatch_is_an_error() {
921 let mut cache = RelationCache::default();
922 cache.apply_relation(orders_relation("text"));
923
924 let error = cache
925 .translate(RowEvent {
926 relation_id: 7,
927 lsn: PgLsn(0x31),
928 commit_lsn: None,
929 txid: None,
930 commit_timestamp_ms: None,
931 kind: RowEventKind::Insert,
932 values: vec![ColumnValue {
933 name: "col0".to_owned(),
934 value: ScalarValue::Text("42".to_owned()),
935 }],
936 })
937 .expect_err("decoded placeholder count should match relation schema");
938
939 match error {
940 PostgresConnectError::Backend(message) => {
941 check!(message.contains("column count mismatch"));
942 check!(message.contains("public.orders"));
943 }
944 error => panic!("expected backend error, got {error:?}"),
945 }
946 }
947
948 #[test]
949 fn named_partial_full_row_values_are_a_count_mismatch() {
950 let mut cache = RelationCache::default();
951 cache.apply_relation(orders_relation("text"));
952
953 let error = cache
954 .translate(RowEvent {
955 relation_id: 7,
956 lsn: PgLsn(0x34),
957 commit_lsn: None,
958 txid: None,
959 commit_timestamp_ms: None,
960 kind: RowEventKind::Insert,
961 values: vec![id(42)],
962 })
963 .expect_err("partial full row should fail count validation");
964
965 match error {
966 PostgresConnectError::Backend(message) => {
967 check!(message.contains("column count mismatch"));
968 check!(message.contains("public.orders"));
969 }
970 error => panic!("expected backend error, got {error:?}"),
971 }
972 }
973
974 #[test]
975 fn pgoutput_text_scalars_are_coerced_to_declared_column_types() {
976 check!(
977 coerce_value(ScalarValue::Text("t".to_owned()), "bool").expect("true bool")
978 == ScalarValue::Bool(true)
979 );
980 check!(
981 coerce_value(ScalarValue::Text("f".to_owned()), "bool").expect("false bool")
982 == ScalarValue::Bool(false)
983 );
984 check!(
985 coerce_value(ScalarValue::Text("12.50".to_owned()), "numeric").expect("numeric")
986 == ScalarValue::Float("12.50".to_owned())
987 );
988 check!(
989 coerce_value(ScalarValue::Text("\\xDEad".to_owned()), "bytea").expect("bytea")
990 == ScalarValue::Bytes(vec![0xde, 0xad])
991 );
992 }
993
994 #[test]
995 fn invalid_bool_text_is_rejected() {
996 let error = coerce_value(ScalarValue::Text("yes".to_owned()), "bool")
997 .expect_err("invalid bool should fail");
998
999 match error {
1000 PostgresConnectError::Backend(message) => {
1001 check!(message.contains("invalid bool pgoutput text value"));
1002 check!(message.contains("yes"));
1003 }
1004 error => panic!("expected backend error, got {error:?}"),
1005 }
1006 }
1007
1008 #[test]
1009 fn bytea_text_supports_escape_and_raw_forms() {
1010 check!(coerce_bytea("\\x0001ff").expect("hex bytea") == vec![0, 1, 255]);
1011 check!(coerce_bytea("raw").expect("raw bytea") == b"raw".to_vec());
1012 }
1013
1014 #[test]
1015 fn bytea_hex_rejects_odd_length_and_invalid_pairs() {
1016 let odd = coerce_bytea("\\x0").expect_err("odd-length bytea should fail");
1017 let invalid = coerce_bytea("\\xzz").expect_err("invalid bytea should fail");
1018
1019 match odd {
1020 PostgresConnectError::Backend(message) => {
1021 check!(message.contains("odd hex length"));
1022 }
1023 error => panic!("expected backend error, got {error:?}"),
1024 }
1025 match invalid {
1026 PostgresConnectError::Backend(message) => {
1027 check!(message.contains("invalid bytea pgoutput text value"));
1028 check!(message.contains("zz"));
1029 }
1030 error => panic!("expected backend error, got {error:?}"),
1031 }
1032 }
1033
1034 #[test]
1035 fn update_without_old_tuple_does_not_trigger_key_change_check() {
1036 let mut cache = RelationCache::default();
1037 cache.apply_relation(orders_relation("text"));
1038
1039 let difference = cache
1040 .translate(RowEvent {
1041 relation_id: 7,
1042 lsn: PgLsn(0x35),
1043 commit_lsn: None,
1044 txid: None,
1045 commit_timestamp_ms: None,
1046 kind: RowEventKind::Update {
1047 old: Vec::new(),
1048 old_tuple_kind: RowTupleKind::Full,
1049 },
1050 values: vec![id(42), status("paid")],
1051 })
1052 .expect("missing old tuple should translate");
1053
1054 check!(difference.key.columns == vec![id(42)]);
1055 check!(difference.before == Vec::new());
1056 }
1057
1058 #[test]
1059 fn key_column_presence_ignores_non_key_columns() {
1060 let schema = TableSchema {
1061 schema: "public".to_owned(),
1062 table: "orders".to_owned(),
1063 columns: orders_relation("text").columns,
1064 };
1065
1066 check!(super::has_any_key_column(&schema, &[id(42)]));
1067 check!(!super::has_any_key_column(&schema, &[status("paid")]));
1068 check!(!super::has_any_key_column(&schema, &[]));
1069 }
1070
1071 #[test]
1072 fn type_oids_map_to_supported_scalar_names() {
1073 for (oid, name) in [
1074 (16, "bool"),
1075 (17, "bytea"),
1076 (20, "int8"),
1077 (21, "int2"),
1078 (23, "int4"),
1079 (25, "text"),
1080 (700, "float4"),
1081 (701, "float8"),
1082 (1043, "varchar"),
1083 (1700, "numeric"),
1084 (999_999, "unknown"),
1085 ] {
1086 check!(super::type_name(oid) == name);
1087 }
1088 }
1089}
1090
1091#[cfg(test)]
1092mod decode_tests {
1093 use assert2::check;
1094
1095 use super::{DecodedMessage, RowEventKind, RowTupleKind, decode_pgoutput_message};
1096 use crate::model::{ColumnSchema, ScalarValue};
1097 use crate::{PgLsn, PostgresConnectError};
1098
1099 fn put_i16(bytes: &mut Vec<u8>, value: i16) {
1100 bytes.extend_from_slice(&value.to_be_bytes());
1101 }
1102
1103 fn put_i32(bytes: &mut Vec<u8>, value: i32) {
1104 bytes.extend_from_slice(&value.to_be_bytes());
1105 }
1106
1107 fn put_u32(bytes: &mut Vec<u8>, value: u32) {
1108 bytes.extend_from_slice(&value.to_be_bytes());
1109 }
1110
1111 fn put_i64(bytes: &mut Vec<u8>, value: i64) {
1112 bytes.extend_from_slice(&value.to_be_bytes());
1113 }
1114
1115 fn put_cstr(bytes: &mut Vec<u8>, value: &str) {
1116 bytes.extend_from_slice(value.as_bytes());
1117 bytes.push(0);
1118 }
1119
1120 fn put_text_value(bytes: &mut Vec<u8>, value: &str) {
1121 bytes.push(b't');
1122 let length = i32::try_from(value.len()).expect("test value length should fit i32");
1123 put_i32(bytes, length);
1124 bytes.extend_from_slice(value.as_bytes());
1125 }
1126
1127 fn orders_relation_message() -> super::RelationEvent {
1128 super::RelationEvent {
1129 relation_id: 7,
1130 schema: "public".to_owned(),
1131 table: "orders".to_owned(),
1132 columns: vec![
1133 ColumnSchema {
1134 name: "id".to_owned(),
1135 type_name: "int8".to_owned(),
1136 key: true,
1137 },
1138 ColumnSchema {
1139 name: "status".to_owned(),
1140 type_name: "text".to_owned(),
1141 key: false,
1142 },
1143 ],
1144 }
1145 }
1146
1147 #[test]
1148 fn decodes_relation_message() {
1149 let mut bytes = vec![b'R'];
1150 put_i32(&mut bytes, 7);
1151 put_cstr(&mut bytes, "public");
1152 put_cstr(&mut bytes, "orders");
1153 bytes.push(b'd');
1154 put_i16(&mut bytes, 2);
1155 bytes.push(1);
1156 put_cstr(&mut bytes, "id");
1157 put_i32(&mut bytes, 20);
1158 put_i32(&mut bytes, -1);
1159 bytes.push(0);
1160 put_cstr(&mut bytes, "status");
1161 put_i32(&mut bytes, 25);
1162 put_i32(&mut bytes, -1);
1163
1164 let decoded = decode_pgoutput_message(&bytes, PgLsn(0x2a), Some(99))
1165 .expect("relation message should decode");
1166
1167 let DecodedMessage::Relation(relation) = decoded else {
1168 panic!("expected relation message");
1169 };
1170 check!(relation.relation_id == 7);
1171 check!(relation.schema == "public");
1172 check!(relation.table == "orders");
1173 check!(relation.columns.len() == 2);
1174 check!(relation.columns[0].name == "id");
1175 check!(relation.columns[0].type_name == "int8");
1176 check!(relation.columns[0].key);
1177 check!(relation.columns[1].name == "status");
1178 check!(relation.columns[1].type_name == "text");
1179 check!(!relation.columns[1].key);
1180 }
1181
1182 #[test]
1183 fn decodes_insert_message_to_row_event() {
1184 let mut bytes = vec![b'I'];
1185 put_i32(&mut bytes, 7);
1186 bytes.push(b'N');
1187 put_i16(&mut bytes, 2);
1188 put_text_value(&mut bytes, "42");
1189 put_text_value(&mut bytes, "paid");
1190
1191 let decoded = decode_pgoutput_message(&bytes, PgLsn(0x2a), Some(99))
1192 .expect("insert message should decode");
1193
1194 let DecodedMessage::Row(row) = decoded else {
1195 panic!("expected row message");
1196 };
1197 check!(row.relation_id == 7);
1198 check!(row.lsn == PgLsn(0x2a));
1199 check!(row.txid == Some(99));
1200 check!(row.commit_timestamp_ms == None);
1201 check!(row.kind == RowEventKind::Insert);
1202 check!(row.values.len() == 2);
1203 check!(row.values[0].name == "col0");
1204 check!(row.values[0].value == ScalarValue::Text("42".to_owned()));
1205 }
1206
1207 #[test]
1208 fn decoded_relation_and_insert_translate_with_schema_column_names() {
1209 let mut relation_bytes = vec![b'R'];
1210 put_i32(&mut relation_bytes, 7);
1211 put_cstr(&mut relation_bytes, "public");
1212 put_cstr(&mut relation_bytes, "orders");
1213 relation_bytes.push(b'd');
1214 put_i16(&mut relation_bytes, 2);
1215 relation_bytes.push(1);
1216 put_cstr(&mut relation_bytes, "id");
1217 put_i32(&mut relation_bytes, 20);
1218 put_i32(&mut relation_bytes, -1);
1219 relation_bytes.push(0);
1220 put_cstr(&mut relation_bytes, "status");
1221 put_i32(&mut relation_bytes, 25);
1222 put_i32(&mut relation_bytes, -1);
1223
1224 let mut cache = super::RelationCache::default();
1225 let DecodedMessage::Relation(relation) =
1226 decode_pgoutput_message(&relation_bytes, PgLsn(0), None)
1227 .expect("relation should decode")
1228 else {
1229 panic!("expected relation");
1230 };
1231 cache.apply_relation(relation);
1232
1233 let mut insert_bytes = vec![b'I'];
1234 put_i32(&mut insert_bytes, 7);
1235 insert_bytes.push(b'N');
1236 put_i16(&mut insert_bytes, 2);
1237 put_text_value(&mut insert_bytes, "42");
1238 put_text_value(&mut insert_bytes, "paid");
1239
1240 let DecodedMessage::Row(row) =
1241 decode_pgoutput_message(&insert_bytes, PgLsn(0x32), Some(101))
1242 .expect("insert should decode")
1243 else {
1244 panic!("expected row");
1245 };
1246 let difference = cache
1247 .translate(row)
1248 .expect("decoded insert should translate");
1249
1250 check!(difference.key.columns[0].name == "id");
1251 check!(difference.key.columns[0].value == ScalarValue::Int(42));
1252 check!(difference.after[1].name == "status");
1253 }
1254
1255 #[test]
1256 fn decoded_int8_text_value_translates_to_int_scalar_for_key() {
1257 let mut cache = super::RelationCache::default();
1258 cache.apply_relation(orders_relation_message());
1259
1260 let mut insert_bytes = vec![b'I'];
1261 put_i32(&mut insert_bytes, 7);
1262 insert_bytes.push(b'N');
1263 put_i16(&mut insert_bytes, 2);
1264 put_text_value(&mut insert_bytes, "42");
1265 put_text_value(&mut insert_bytes, "paid");
1266
1267 let DecodedMessage::Row(row) =
1268 decode_pgoutput_message(&insert_bytes, PgLsn(0x32), Some(101))
1269 .expect("insert should decode")
1270 else {
1271 panic!("expected row");
1272 };
1273 let difference = cache
1274 .translate(row)
1275 .expect("decoded insert should translate");
1276
1277 check!(difference.key.columns[0].name == "id");
1278 check!(difference.key.columns[0].value == ScalarValue::Int(42));
1279 }
1280
1281 #[test]
1282 fn decoded_delete_key_tuple_translates_with_key_column_name() {
1283 let mut cache = super::RelationCache::default();
1284 cache.apply_relation(orders_relation_message());
1285
1286 let mut delete_bytes = vec![b'D'];
1287 put_i32(&mut delete_bytes, 7);
1288 delete_bytes.push(b'K');
1289 put_i16(&mut delete_bytes, 1);
1290 put_text_value(&mut delete_bytes, "42");
1291
1292 let DecodedMessage::Row(row) =
1293 decode_pgoutput_message(&delete_bytes, PgLsn(0x35), Some(102))
1294 .expect("delete should decode")
1295 else {
1296 panic!("expected row");
1297 };
1298 let difference = cache
1299 .translate(row)
1300 .expect("decoded delete should translate");
1301
1302 check!(difference.key.columns[0].name == "id");
1303 check!(difference.before[0].name == "id");
1304 check!(difference.before[0].value == ScalarValue::Int(42));
1305 }
1306
1307 #[test]
1308 fn decoded_update_key_old_tuple_translates_for_non_key_update() {
1309 let mut cache = super::RelationCache::default();
1310 cache.apply_relation(orders_relation_message());
1311
1312 let mut update_bytes = vec![b'U'];
1313 put_i32(&mut update_bytes, 7);
1314 update_bytes.push(b'K');
1315 put_i16(&mut update_bytes, 1);
1316 put_text_value(&mut update_bytes, "42");
1317 update_bytes.push(b'N');
1318 put_i16(&mut update_bytes, 2);
1319 put_text_value(&mut update_bytes, "42");
1320 put_text_value(&mut update_bytes, "paid");
1321
1322 let DecodedMessage::Row(row) =
1323 decode_pgoutput_message(&update_bytes, PgLsn(0x36), Some(103))
1324 .expect("update should decode")
1325 else {
1326 panic!("expected row");
1327 };
1328 let difference = cache
1329 .translate(row)
1330 .expect("decoded update should translate");
1331
1332 check!(difference.before[0].name == "id");
1333 check!(difference.before[0].value == ScalarValue::Int(42));
1334 check!(difference.after[1].name == "status");
1335 check!(difference.after[1].value == ScalarValue::Text("paid".to_owned()));
1336 }
1337
1338 #[test]
1339 fn decodes_delete_message_to_row_event() {
1340 let mut bytes = vec![b'D'];
1341 put_i32(&mut bytes, 7);
1342 bytes.push(b'K');
1343 put_i16(&mut bytes, 1);
1344 put_text_value(&mut bytes, "42");
1345
1346 let decoded = decode_pgoutput_message(&bytes, PgLsn(0x2b), None)
1347 .expect("delete message should decode");
1348
1349 let DecodedMessage::Row(row) = decoded else {
1350 panic!("expected row message");
1351 };
1352 check!(row.relation_id == 7);
1353 check!(row.lsn == PgLsn(0x2b));
1354 check!(row.txid == None);
1355 check!(
1356 row.kind
1357 == RowEventKind::Delete {
1358 tuple_kind: RowTupleKind::Key,
1359 }
1360 );
1361 check!(row.values.len() == 1);
1362 check!(row.values[0].value == ScalarValue::Text("42".to_owned()));
1363 }
1364
1365 #[test]
1366 fn decodes_update_message_to_row_event() {
1367 let mut bytes = vec![b'U'];
1368 put_i32(&mut bytes, 7);
1369 bytes.push(b'K');
1370 put_i16(&mut bytes, 1);
1371 put_text_value(&mut bytes, "41");
1372 bytes.push(b'N');
1373 put_i16(&mut bytes, 2);
1374 put_text_value(&mut bytes, "42");
1375 put_text_value(&mut bytes, "paid");
1376
1377 let decoded = decode_pgoutput_message(&bytes, PgLsn(0x2c), Some(100))
1378 .expect("update message should decode");
1379
1380 let DecodedMessage::Row(row) = decoded else {
1381 panic!("expected row message");
1382 };
1383 check!(row.relation_id == 7);
1384 let RowEventKind::Update {
1385 old,
1386 old_tuple_kind,
1387 } = row.kind
1388 else {
1389 panic!("expected update event");
1390 };
1391 check!(old_tuple_kind == RowTupleKind::Key);
1392 check!(old.len() == 1);
1393 check!(old[0].value == ScalarValue::Text("41".to_owned()));
1394 check!(row.values.len() == 2);
1395 check!(row.values[0].value == ScalarValue::Text("42".to_owned()));
1396 }
1397
1398 #[test]
1399 fn decodes_unchanged_toast_as_distinct_scalar_value() {
1400 let mut bytes = vec![b'U'];
1401 put_i32(&mut bytes, 7);
1402 bytes.push(b'N');
1403 put_i16(&mut bytes, 2);
1404 put_text_value(&mut bytes, "42");
1405 bytes.push(b'u');
1406
1407 let decoded = decode_pgoutput_message(&bytes, PgLsn(0x33), Some(100))
1408 .expect("update message should decode");
1409
1410 let DecodedMessage::Row(row) = decoded else {
1411 panic!("expected row message");
1412 };
1413 check!(row.values[1].value == ScalarValue::UnchangedToast);
1414 }
1415
1416 #[test]
1417 fn malformed_message_returns_backend_error() {
1418 let error = decode_pgoutput_message(&[b'R', 0, 0], PgLsn(0), None)
1419 .expect_err("truncated message should fail");
1420
1421 match error {
1422 PostgresConnectError::Backend(message) => {
1423 check!(message.contains("truncated"));
1424 }
1425 error => panic!("expected backend error, got {error:?}"),
1426 }
1427 }
1428
1429 #[test]
1430 fn trailing_message_bytes_are_rejected() {
1431 let mut bytes = vec![b'I'];
1432 put_i32(&mut bytes, 7);
1433 bytes.push(b'N');
1434 put_i16(&mut bytes, 2);
1435 put_text_value(&mut bytes, "42");
1436 put_text_value(&mut bytes, "paid");
1437 bytes.push(0xff);
1438
1439 let error = decode_pgoutput_message(&bytes, PgLsn(0), None)
1440 .expect_err("trailing bytes should fail");
1441
1442 match error {
1443 PostgresConnectError::Backend(message) => {
1444 check!(message.contains("trailing bytes"));
1445 }
1446 error => panic!("expected backend error, got {error:?}"),
1447 }
1448 }
1449
1450 #[test]
1451 fn decodes_begin_and_commit_messages() {
1452 let mut begin = vec![b'B'];
1453 put_i64(&mut begin, 0x0102_0304);
1454 put_i64(&mut begin, 1_700_000_000);
1455 put_i32(&mut begin, 123);
1456
1457 let decoded = decode_pgoutput_message(&begin, PgLsn(0), None).expect("begin should decode");
1458 check!(
1459 decoded
1460 == DecodedMessage::Begin {
1461 final_lsn: PgLsn(0x0102_0304),
1462 xid: 123,
1463 }
1464 );
1465
1466 let mut commit = vec![b'C', 0];
1467 put_i64(&mut commit, 0x0102_0305);
1468 put_i64(&mut commit, 0x0102_0306);
1469 put_i64(&mut commit, 1_000);
1470
1471 let decoded =
1472 decode_pgoutput_message(&commit, PgLsn(0), None).expect("commit should decode");
1473 check!(
1474 decoded
1475 == DecodedMessage::Commit {
1476 commit_lsn: PgLsn(0x0102_0305),
1477 end_lsn: PgLsn(0x0102_0306),
1478 commit_timestamp_ms: 946_684_800_001,
1479 }
1480 );
1481 }
1482
1483 #[test]
1484 fn decodes_begin_xid_as_unsigned_u32() {
1485 let mut begin = vec![b'B'];
1486 put_i64(&mut begin, 0x0102_0304);
1487 put_i64(&mut begin, 1_700_000_000);
1488 put_u32(&mut begin, 0x8000_0000);
1489
1490 let decoded = decode_pgoutput_message(&begin, PgLsn(0), None).expect("begin should decode");
1491 check!(
1492 decoded
1493 == DecodedMessage::Begin {
1494 final_lsn: PgLsn(0x0102_0304),
1495 xid: 2_147_483_648,
1496 }
1497 );
1498 }
1499}