From e988fb85e624a1167314ead224b82d5284f0ca06 Mon Sep 17 00:00:00 2001
From: Theo Butler <theodusbutler@gmail.com>
Date: Thu, 17 Oct 2024 13:08:12 -0400
Subject: [PATCH] fix: start timestamp

---
 src/bin/main.rs | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/src/bin/main.rs b/src/bin/main.rs
index 388559f..4d0551e 100644
--- a/src/bin/main.rs
+++ b/src/bin/main.rs
@@ -66,7 +66,8 @@ async fn run() -> anyhow::Result<()> {
     )?
     .create()?;
 
-    let start_timestamp = latest_sink_timestamp(&consumer).await?.unwrap_or(0);
+    let start_timestamp =
+        latest_sink_timestamp(&consumer).await?.unwrap_or(0) + AGGREGATION_INTERVAL_MILLIS;
     consumer.unassign().context("unassign sinks")?;
     tracing::info!(start_timestamp = print_unix_millis(start_timestamp)?);