Reload stored events every 15 minutes (and after boot) so stats and rankings never lose data

app/main.py

@@ -39,6 +39,7 @@ MAX_BODY = 64 * 1024
 FRAME_ANCESTORS = os.environ.get("EXPLORER_FRAME_ANCESTORS", "'self' https://huggingface.co https://*.hf.space")
 DETAIL_TOP_N = int(os.environ.get("EXPLORER_DETAIL_TOP_N", "24"))
 PREWARM_N = int(os.environ.get("EXPLORER_PREWARM", "300"))
+RELOAD_SECONDS = float(os.environ.get("EXPLORER_RELOAD_SECONDS", "900"))
 
 
 class RecommendRequest(BaseModel):
@@ -67,7 +68,7 @@ class State:
         self.engine = HeuristicEngine()
         self.limiter = RateLimiter(rate_per_minute=float(os.environ.get("EXPLORER_RATE_PER_MIN", "120")), burst=60)
         self.existing_rows: list[dict] = []
-        self.stats = StatsCache(lambda: self.existing_rows + self.sink.all_rows)
+        self.stats = StatsCache(lambda: self.existing_rows + self.sink.live_rows())
         self.fetch_live = fetch_live
         self.collection_enabled = os.environ.get("EXPLORER_SINK", "local") != "off"
 
@@ -98,11 +99,26 @@ class State:
             self.catalogue.start_background_refresh()
         self.sink.start()
 
-        def load_existing():
-            self.existing_rows = self.sink.read_existing()
-            self.stats.invalidate()
-
-        threading.Thread(target=load_existing, daemon=True, name="load-existing").start()
+        def reload_loop():
+            # Load at boot, again shortly after (a restart can race the previous container's
+            # final flush), then regularly so shards written elsewhere or by compaction appear.
+            for delay in (0, 90):
+                time.sleep(delay)
+                self.reload_existing()
+            while True:
+                time.sleep(RELOAD_SECONDS)
+                self.reload_existing()
+
+        threading.Thread(target=reload_loop, daemon=True, name="reload-existing").start()
+
+    def reload_existing(self):
+        started = time.time()
+        rows = self.sink.read_existing()
+        if rows is None:
+            return
+        self.existing_rows = rows
+        self.sink.forget_flushed_before(started)
+        self.stats.invalidate()
 
 
 def create_app(state: State | None = None) -> FastAPI:

app/sink.py

@@ -66,7 +66,9 @@ class EventSink:
         self.last_flush_at: float | None = None
         self.last_error: str | None = None
         self.consecutive_failures = 0
-        self.all_rows: list[dict] = []  # everything accepted this process (for live stats)
+        # Rows accepted by this process that a reload of the stored dataset may not include yet:
+        # [row, flushed_at]. Stats read stored rows + these, so nothing is missed or double-counted.
+        self._recent: list[list] = []
         spool_dir.mkdir(parents=True, exist_ok=True)
         self._recover_spool()
 
@@ -90,9 +92,9 @@ class EventSink:
                 self._buf = self._buf[overflow:]
                 self.dropped += overflow
             self._buf.extend(rows)
-            self.all_rows.extend(rows)
-            if len(self.all_rows) > MAX_BUFFER:
-                self.all_rows = self.all_rows[-MAX_BUFFER:]
+            self._recent.extend([r, None] for r in rows)
+            if len(self._recent) > MAX_BUFFER:
+                self._recent = self._recent[-MAX_BUFFER:]
             try:
                 with open(self._spool, "a") as f:
                     for r in rows:
@@ -125,6 +127,10 @@ class EventSink:
             with self._lock:
                 self._buf = self._buf[len(rows):]
                 self._rewrite_spool()
+                written, now = {id(r) for r in rows}, time.time()
+                for entry in self._recent:
+                    if entry[1] is None and id(entry[0]) in written:
+                        entry[1] = now
             self.written_rows += len(rows)
             self.written_shards += 1
             self.last_flush_at = time.time()
@@ -142,6 +148,19 @@ class EventSink:
         except OSError:
             log.warning("spool rewrite failed", exc_info=True)
 
+    def live_rows(self) -> list[dict]:
+        with self._lock:
+            return [r for r, _ in self._recent]
+
+    def forget_flushed_before(self, t: float) -> None:
+        """Drop rows that were in a written shard before a reload started (the reload has them)."""
+        with self._lock:
+            self._recent = [e for e in self._recent if e[1] is None or e[1] >= t]
+
+    @property
+    def all_rows(self) -> list[dict]:
+        return self.live_rows()
+
     def _write_shard(self, path: str, data: bytes) -> None:
         raise NotImplementedError
 

tests/test_sink.py

@@ -93,3 +93,18 @@ def test_reads_monthly_and_incoming(tmp_path):
     s.add(rows(2))
     s.flush()
     assert len(s.read_existing()) == 6
+
+
+def test_reload_neither_misses_nor_double_counts(tmp_path):
+    import time
+    s = LocalParquetSink(tmp_path, flush_seconds=999)
+    s.add(rows(3))
+    stored = s.read_existing()
+    assert len(stored) + len(s.live_rows()) == 3          # accepted, not yet written
+    s.flush()
+    assert len(s.read_existing()) + len(s.live_rows()) == 6  # written but not yet reloaded: still visible (overlap is fine before a reload)
+    started = time.time()
+    stored = s.read_existing()
+    s.forget_flushed_before(started)
+    s.add(rows(2))
+    assert len(stored) + len(s.live_rows()) == 5          # after reload: 3 stored + 2 new, no duplicates