replication: use SystemTime for timestamps at API boundary

Signed-off-by: Petros Angelatos <[email protected]>

Author: petrosagg

postgres-protocol/Cargo.toml

@@ -14,6 +14,7 @@ byteorder = "1.0"
 bytes = "1.0"
 fallible-iterator = "0.2"
 hmac = "0.10"
+lazy_static = "1.4"
 md-5 = "0.9"
 memchr = "2.0"
 rand = "0.8"

postgres-protocol/src/lib.rs

@@ -14,7 +14,9 @@
 
 use byteorder::{BigEndian, ByteOrder};
 use bytes::{BufMut, BytesMut};
+use lazy_static::lazy_static;
 use std::io;
+use std::time::{Duration, SystemTime, UNIX_EPOCH};
 
 pub mod authentication;
 pub mod escape;
@@ -28,6 +30,11 @@ pub type Oid = u32;
 /// A Postgres Log Sequence Number (LSN).
 pub type Lsn = u64;
 
+lazy_static! {
+    /// Postgres epoch is 2000-01-01T00:00:00Z
+    pub static ref PG_EPOCH: SystemTime = UNIX_EPOCH + Duration::from_secs(946_684_800);
+}
+
 /// An enum indicating if a value is `NULL` or not.
 pub enum IsNull {
     /// The value is `NULL`.

postgres-protocol/src/message/backend.rs

@@ -8,8 +8,9 @@ use std::cmp;
 use std::io::{self, Read};
 use std::ops::Range;
 use std::str;
+use std::time::{Duration, SystemTime};
 
-use crate::{Lsn, Oid};
+use crate::{Lsn, Oid, PG_EPOCH};
 
 // top-level message tags
 pub const PARSE_COMPLETE_TAG: u8 = b'1';
@@ -340,7 +341,12 @@ impl ReplicationMessage<Bytes> {
             XLOG_DATA_TAG => {
                 let wal_start = buf.read_u64::<BigEndian>()?;
                 let wal_end = buf.read_u64::<BigEndian>()?;
-                let timestamp = buf.read_i64::<BigEndian>()?;
+                let ts = buf.read_i64::<BigEndian>()?;
+                let timestamp = if ts > 0 {
+                    *PG_EPOCH + Duration::from_micros(ts as u64)
+                } else {
+                    *PG_EPOCH - Duration::from_micros(-ts as u64)
+                };
                 let data = buf.read_all();
                 ReplicationMessage::XLogData(XLogDataBody {
                     wal_start,
@@ -351,7 +357,12 @@ impl ReplicationMessage<Bytes> {
             }
             PRIMARY_KEEPALIVE_TAG => {
                 let wal_end = buf.read_u64::<BigEndian>()?;
-                let timestamp = buf.read_i64::<BigEndian>()?;
+                let ts = buf.read_i64::<BigEndian>()?;
+                let timestamp = if ts > 0 {
+                    *PG_EPOCH + Duration::from_micros(ts as u64)
+                } else {
+                    *PG_EPOCH - Duration::from_micros(-ts as u64)
+                };
                 let reply = buf.read_u8()?;
                 ReplicationMessage::PrimaryKeepAlive(PrimaryKeepAliveBody {
                     wal_end,
@@ -894,7 +905,7 @@ impl RowDescriptionBody {
 pub struct XLogDataBody<D> {
     wal_start: u64,
     wal_end: u64,
-    timestamp: i64,
+    timestamp: SystemTime,
     data: D,
 }
 
@@ -910,7 +921,7 @@ impl<D> XLogDataBody<D> {
     }
 
     #[inline]
-    pub fn timestamp(&self) -> i64 {
+    pub fn timestamp(&self) -> SystemTime {
         self.timestamp
     }
 
@@ -941,7 +952,7 @@ impl<D> XLogDataBody<D> {
 #[derive(Debug)]
 pub struct PrimaryKeepAliveBody {
     wal_end: u64,
-    timestamp: i64,
+    timestamp: SystemTime,
     reply: u8,
 }
 
@@ -952,7 +963,7 @@ impl PrimaryKeepAliveBody {
     }
 
     #[inline]
-    pub fn timestamp(&self) -> i64 {
+    pub fn timestamp(&self) -> SystemTime {
         self.timestamp
     }
 

tokio-postgres/src/replication.rs

@@ -10,6 +10,7 @@ use postgres_protocol::message::backend::ReplicationMessage;
 use postgres_types::PgLsn;
 use std::pin::Pin;
 use std::task::{Context, Poll};
+use std::time::SystemTime;
 
 const STANDBY_STATUS_UPDATE_TAG: u8 = b'r';
 const HOT_STANDBY_FEEDBACK_TAG: u8 = b'h';
@@ -38,17 +39,22 @@ impl ReplicationStream {
         write_lsn: PgLsn,
         flush_lsn: PgLsn,
         apply_lsn: PgLsn,
-        ts: i64,
+        timestamp: SystemTime,
         reply: u8,
     ) -> Result<(), Error> {
         let mut this = self.project();
 
+        let timestamp = match timestamp.duration_since(*PG_EPOCH) {
+            Ok(d) => d.as_micros() as i64,
+            Err(e) => -(e.duration().as_micros() as i64),
+        };
+
         let mut buf = BytesMut::new();
         buf.put_u8(STANDBY_STATUS_UPDATE_TAG);
         buf.put_u64(write_lsn.into());
         buf.put_u64(flush_lsn.into());
         buf.put_u64(apply_lsn.into());
-        buf.put_i64(ts);
+        buf.put_i64(timestamp);
         buf.put_u8(reply);
 
         this.stream.send(buf.freeze()).await
@@ -57,14 +63,19 @@ impl ReplicationStream {
     /// Send hot standby feedback message to server.
     pub async fn hot_standby_feedback(
         self: Pin<&mut Self>,
-        timestamp: i64,
+        timestamp: SystemTime,
         global_xmin: u32,
         global_xmin_epoch: u32,
         catalog_xmin: u32,
         catalog_xmin_epoch: u32,
     ) -> Result<(), Error> {
         let mut this = self.project();
 
+        let timestamp = match timestamp.duration_since(*PG_EPOCH) {
+            Ok(d) => d.as_micros() as i64,
+            Err(e) => -(e.duration().as_micros() as i64),
+        };
+
         let mut buf = BytesMut::new();
         buf.put_u8(HOT_STANDBY_FEEDBACK_TAG);
         buf.put_i64(timestamp);
@@ -118,19 +129,19 @@ impl LogicalReplicationStream {
         write_lsn: PgLsn,
         flush_lsn: PgLsn,
         apply_lsn: PgLsn,
-        ts: i64,
+        timestamp: SystemTime,
         reply: u8,
     ) -> Result<(), Error> {
         let this = self.project();
         this.stream
-            .standby_status_update(write_lsn, flush_lsn, apply_lsn, ts, reply)
+            .standby_status_update(write_lsn, flush_lsn, apply_lsn, timestamp, reply)
             .await
     }
 
     /// Send hot standby feedback message to server.
     pub async fn hot_standby_feedback(
         self: Pin<&mut Self>,
-        timestamp: i64,
+        timestamp: SystemTime,
         global_xmin: u32,
         global_xmin_epoch: u32,
         catalog_xmin: u32,

tokio-postgres/tests/test/replication.rs

@@ -1,5 +1,5 @@
 use futures::StreamExt;
-use std::time::{Duration, UNIX_EPOCH};
+use std::time::SystemTime;
 
 use postgres_protocol::message::backend::LogicalReplicationMessage::{Begin, Commit, Insert};
 use postgres_protocol::message::backend::ReplicationMessage::*;
@@ -130,13 +130,9 @@ async fn test_replication() {
 
     // Send a standby status update and require a keep alive response
     let lsn: PgLsn = lsn.parse().unwrap();
-
-    // Postgres epoch is 2000-01-01T00:00:00Z
-    let pg_epoch = UNIX_EPOCH + Duration::from_secs(946_684_800);
-    let ts = pg_epoch.elapsed().unwrap().as_micros() as i64;
     stream
         .as_mut()
-        .standby_status_update(lsn, lsn, lsn, ts, 1)
+        .standby_status_update(lsn, lsn, lsn, SystemTime::now(), 1)
         .await
         .unwrap();
     loop {