diff --git a/Cargo.lock b/Cargo.lock
index 7758eae..e963f59 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -542,6 +542,7 @@ version = "0.1.0"
 dependencies = [
  "anyhow",
  "bincode",
+ "chrono",
  "jetstream",
  "log",
  "serde",
diff --git a/packages-rs/drainpipe-store/Cargo.toml b/packages-rs/drainpipe-store/Cargo.toml
index fdb5a07..7bb0a6a 100644
--- a/packages-rs/drainpipe-store/Cargo.toml
+++ b/packages-rs/drainpipe-store/Cargo.toml
@@ -6,6 +6,7 @@ edition = "2021"
 [dependencies]
 anyhow = "1.0.93"
 bincode = "1.3.3"
+chrono = "0.4.42"
 jetstream = { path = "../jetstream" }
 log = "0.4.22"
 serde = { version = "1.0.215", features = ["derive"] }
diff --git a/packages-rs/drainpipe-store/src/lib.rs b/packages-rs/drainpipe-store/src/lib.rs
index 690d994..8c3b9ce 100644
--- a/packages-rs/drainpipe-store/src/lib.rs
+++ b/packages-rs/drainpipe-store/src/lib.rs
@@ -14,6 +14,7 @@ pub struct Store {
 #[derive(Serialize, Deserialize, Debug)]
 enum CursorInner {
     V1(u64),
+    V2 { value: u64, recorded_at: i64 },
 }
 
 #[derive(Serialize, Deserialize, Debug)]
@@ -21,12 +22,23 @@ pub struct Cursor(CursorInner);
 
 impl Cursor {
     pub fn new(value: u64) -> Self {
-        Self(CursorInner::V1(value))
+        Self(CursorInner::V2 {
+            value,
+            recorded_at: chrono::Utc::now().timestamp_micros(),
+        })
     }
 
     pub fn value(&self) -> u64 {
         match self.0 {
             CursorInner::V1(value) => value,
+            CursorInner::V2 { value, .. } => value,
+        }
+    }
+
+    pub fn lag_micros(&self) -> Option<i64> {
+        match self.0 {
+            CursorInner::V1(_) => None,
+            CursorInner::V2 { recorded_at, value } => Some(recorded_at - value as i64),
         }
     }
 }
@@ -88,6 +100,19 @@ impl Store {
             .transpose()
     }
 
+    pub fn get_cursor_lag_micros(&self) -> anyhow::Result<Option<i64>> {
+        self.cursor_tree
+            .get("cursor")
+            .context("Failed to get cursor")?
+            .map(|cursor_bytes| {
+                bincode::deserialize::<Cursor>(&cursor_bytes)
+                    .context("Failed to deserialize cursor")
+                    .map(|c| c.lag_micros())
+            })
+            .transpose()
+            .map(|l| l.flatten())
+    }
+
     fn record_dead_letter(&self, commit_json: String, error_message: String) -> anyhow::Result<()> {
         let key = self.db.generate_id()?.to_string();
         self.dead_letter_tree.insert(
diff --git a/packages-rs/drainpipe/src/main.rs b/packages-rs/drainpipe/src/main.rs
index c0a6cc0..67b6654 100644
--- a/packages-rs/drainpipe/src/main.rs
+++ b/packages-rs/drainpipe/src/main.rs
@@ -70,12 +70,11 @@ async fn main() -> anyhow::Result<()> {
             let store = store.clone();
             tokio::spawn(async move {
                 for interval in metrics_monitor.intervals() {
-                    let lag_output = store.get_cursor().unwrap_or(None).and_then(|c| {
-                        let cursor_time = Utc.timestamp_micros(c as i64).earliest()?;
-                        let now = Utc::now();
-                        let lag = now.signed_duration_since(cursor_time);
-                        Some(format!("Cursor lag: {}ms", lag.num_milliseconds()))
-                    });
+                    let lag_output = store
+                        .get_cursor_lag_micros()
+                        .ok()
+                        .flatten()
+                        .map(|lag| format!("Cursor lag: {}ms", lag / 1000));
                     if let Some(lag) = lag_output {
                         log::info!(
                             "{:?} per second; {}",