fix: reset fetch_notes cursor stranded above the seq high-water

crates/node/src/database/mod.rs

@@ -34,11 +34,21 @@ pub trait DatabaseBackend: Send + Sync {
     /// This is the preferred multi-tag query — running per-tag queries back
     /// to back reopens a race where a concurrent INSERT can land between two
     /// per-tag queries and get leapfrogged by the cursor advance.
+    ///
+    /// Returns the matching notes and the *effective* cursor used for the
+    /// `seq > ?` filter — 0 if the requested cursor was reset (legacy µs cursor,
+    /// or stranded above the seq high-water), otherwise the requested cursor.
+    /// Callers should base their response cursor on this effective value so a
+    /// reset heals the client rather than re-triggering every poll.
+    ///
+    /// An empty tag set short-circuits before the stranded-cursor check and
+    /// returns the post-legacy cursor unchanged (there are no notes to deliver,
+    /// so there is nothing to heal).
     async fn fetch_notes_by_tags(
         &self,
         tags: &[NoteTag],
         cursor: u64,
-    ) -> Result<Vec<StoredNote>, DatabaseError>;
+    ) -> Result<(Vec<StoredNote>, u64), DatabaseError>;
 
     /// Get statistics about the database
     async fn get_stats(&self) -> Result<(u64, u64), DatabaseError>;
@@ -99,11 +109,14 @@ impl Database {
     }
 
     /// Fetch notes matching ANY of a set of tags, in a single DB snapshot.
+    ///
+    /// Returns the notes plus the effective cursor used (see
+    /// [`DatabaseBackend::fetch_notes_by_tags`]).
     pub async fn fetch_notes_by_tags(
         &self,
         tags: &[NoteTag],
         cursor: u64,
-    ) -> Result<Vec<StoredNote>, DatabaseError> {
+    ) -> Result<(Vec<StoredNote>, u64), DatabaseError> {
         self.backend.fetch_notes_by_tags(tags, cursor).await
     }
 
@@ -430,7 +443,7 @@ mod tests {
         );
 
         // === The fix: single-snapshot multi-tag query ===
-        let snapshot = db.fetch_notes_by_tags(&[TAG_A.into(), TAG_B.into()], 0).await.unwrap();
+        let (snapshot, _) = db.fetch_notes_by_tags(&[TAG_A.into(), TAG_B.into()], 0).await.unwrap();
         assert_eq!(
             snapshot.len(),
             3,
@@ -472,10 +485,68 @@ mod tests {
             "legacy microsecond cursor should be reset to 0, returning the note"
         );
 
-        // Sanity check: a non-legacy cursor above the note's seq should NOT trigger the reset.
-        let normal_cursor: u64 = 1_000;
-        let empty = db.fetch_notes(TAG_LOCAL_ANY.into(), normal_cursor).await.unwrap();
-        assert_eq!(empty.len(), 0, "normal cursor > seq should filter correctly");
+        // Sanity check: a caught-up client's cursor (== the note's own seq, i.e.
+        // the current high-water) is NOT reset and filters correctly to empty.
+        // Note: a cursor STRICTLY above the high-water is treated as stranded and
+        // reset — see `test_fetch_notes_resets_cursor_stranded_above_high_water`.
+        let all = db.fetch_notes(TAG_LOCAL_ANY.into(), 0).await.unwrap();
+        let caught_up_cursor = u64::try_from(all[0].seq).expect("seq is non-negative");
+        let empty = db.fetch_notes(TAG_LOCAL_ANY.into(), caught_up_cursor).await.unwrap();
+        assert_eq!(
+            empty.len(),
+            0,
+            "caught-up cursor (== high-water) filters correctly and does not reset"
+        );
+    }
+
+    /// A cursor STRICTLY ABOVE the current `seq` high-water can only exist if the
+    /// server's seq space regressed beneath a client's stored cursor (the backing
+    /// DB was recreated and AUTOINCREMENT restarted low). Such a cursor is below
+    /// the legacy-timestamp threshold, so the legacy guard never fires; without
+    /// the stranded reset it would match `seq > cursor` = nothing, forever.
+    #[tokio::test]
+    async fn test_fetch_notes_resets_cursor_stranded_above_high_water() {
+        let db = Database::connect(DatabaseConfig::default(), Metrics::default().db)
+            .await
+            .unwrap();
+
+        // Populate the current epoch with two notes (high-water becomes their max seq).
+        for details in [vec![1u8], vec![2u8]] {
+            db.store_note(&StoredNote {
+                header: test_note_header(),
+                details,
+                created_at: Utc::now(),
+                seq: 0,
+                after_block_num: None,
+            })
+            .await
+            .unwrap();
+        }
+
+        let all = db.fetch_notes(TAG_LOCAL_ANY.into(), 0).await.unwrap();
+        assert_eq!(all.len(), 2, "sanity: both notes present in this epoch");
+        let high_water = u64::try_from(all[1].seq).expect("seq is non-negative");
+
+        // Caught-up client (cursor == high-water) must NOT reset — else it would
+        // re-deliver the whole backlog on every steady-state poll.
+        let caught_up = db.fetch_notes(TAG_LOCAL_ANY.into(), high_water).await.unwrap();
+        assert_eq!(caught_up.len(), 0, "caught-up cursor must not be reset");
+
+        // Stranded cursor (well above high-water, far below the 1e12 legacy
+        // threshold) resets to 0, recovers the epoch, and reports effective 0 so
+        // the caller can heal the client's stored cursor.
+        let stranded = high_water + 5_000;
+        let (recovered, effective) =
+            db.fetch_notes_by_tags(&[TAG_LOCAL_ANY.into()], stranded).await.unwrap();
+        assert_eq!(
+            recovered.len(),
+            2,
+            "cursor above high-water must reset to 0 and recover the epoch"
+        );
+        assert_eq!(
+            effective, 0,
+            "effective cursor must be 0 so the echoed response cursor heals the client"
+        );
     }
 
     /// Pagination: a response is capped at `FETCH_NOTES_BATCH_SIZE` rows. A

crates/node/src/database/sqlite/mod.rs

@@ -31,6 +31,43 @@ pub(crate) const FETCH_NOTES_BATCH_SIZE: i64 = 500;
 /// and two orders of magnitude below any microsecond timestamp this decade.
 const LEGACY_CURSOR_THRESHOLD: u64 = 1_000_000_000_000;
 
+/// AUTOINCREMENT high-water mark for the `notes` table (`sqlite_sequence.seq`):
+/// the maximum `seq` ever allocated in this DB file's lifetime.
+///
+/// AUTOINCREMENT guarantees this never decreases within a single lifetime — it
+/// survives `DELETE`, `VACUUM` and `cleanup_old_notes` — so a client cursor that
+/// is strictly above it can only have come from a *different* (larger) seq space
+/// that no longer exists, i.e. the backing DB was recreated. See
+/// [`SqliteDatabase::fetch_notes_by_tags`] for how that stranded cursor is reset.
+///
+/// Returns `None` when the high-water can't be determined (no note ever inserted,
+/// or the sequence bookkeeping is unavailable). The caller then SKIPS the
+/// stranded-cursor check entirely, leaving the cursor unchanged — fail-safe, so a
+/// transient unavailability of `sqlite_sequence` can never fail a fetch or falsely
+/// reset a live client.
+fn high_water_seq(conn: &mut SqliteConnection) -> Option<i64> {
+    #[derive(diesel::QueryableByName)]
+    struct HighWater {
+        #[diesel(sql_type = diesel::sql_types::BigInt)]
+        seq: i64,
+    }
+
+    // `sqlite_sequence` is SQLite bookkeeping, not part of the Diesel schema, so
+    // query it with raw SQL. `.optional()` maps "no row for notes yet" (nothing
+    // inserted) to `None`; any other error also degrades to `None` so this check
+    // is purely advisory and can never break a fetch.
+    match diesel::sql_query("SELECT seq FROM sqlite_sequence WHERE name = 'notes'")
+        .get_result::<HighWater>(conn)
+        .optional()
+    {
+        Ok(row) => row.map(|r| r.seq),
+        Err(err) => {
+            tracing::debug!(?err, "sqlite_sequence unreadable; skipping stranded-cursor check");
+            None
+        },
+    }
+}
+
 /// `SQLite` implementation of the database backend
 pub struct SqliteDatabase {
     pool: deadpool_diesel::Pool<ConnectionManager, deadpool::managed::Object<ConnectionManager>>,
@@ -137,7 +174,9 @@ impl DatabaseBackend for SqliteDatabase {
         tag: NoteTag,
         cursor: u64,
     ) -> Result<Vec<StoredNote>, DatabaseError> {
-        self.fetch_notes_by_tags(&[tag], cursor).await
+        // Single-tag convenience: drop the effective cursor (only the gRPC
+        // fetch handler needs it, to base its response cursor on the reset).
+        Ok(self.fetch_notes_by_tags(&[tag], cursor).await?.0)
     }
 
     #[tracing::instrument(skip(self, tags), fields(
@@ -150,28 +189,27 @@ impl DatabaseBackend for SqliteDatabase {
         &self,
         tags: &[NoteTag],
         cursor: u64,
-    ) -> Result<Vec<StoredNote>, DatabaseError> {
+    ) -> Result<(Vec<StoredNote>, u64), DatabaseError> {
         let timer = self.metrics.db_fetch_notes();
 
         // Legacy cursor detection: clients upgraded from the pre-`seq` schema
         // carry microsecond-timestamp cursors; interpret those as 0 so they
         // don't stall forever waiting for `seq` to catch up. Record a metric
         // so operators can see when pre-migration clients are being reset.
-        let effective_cursor = if cursor > LEGACY_CURSOR_THRESHOLD {
+        let legacy_reset = cursor > LEGACY_CURSOR_THRESHOLD;
+        if legacy_reset {
             self.metrics.db_fetch_notes_legacy_cursor_reset();
             tracing::info!(original_cursor = cursor, "Legacy cursor reset to 0");
-            0
-        } else {
-            cursor
-        };
+        }
+        let post_legacy_cursor = if legacy_reset { 0 } else { cursor };
 
-        let cursor_i64: i64 = effective_cursor.try_into().map_err(|_| {
+        let cursor_i64: i64 = post_legacy_cursor.try_into().map_err(|_| {
             DatabaseError::QueryExecution("Cursor too large for SQLite".to_string())
         })?;
 
         if tags.is_empty() {
             timer.finish("ok");
-            return Ok(Vec::new());
+            return Ok((Vec::new(), post_legacy_cursor));
         }
 
         let tag_values: Vec<i64> = tags.iter().map(|t| i64::from(t.as_u32())).collect();
@@ -183,22 +221,65 @@ impl DatabaseBackend for SqliteDatabase {
         //
         // LIMIT caps response size; a backlogged client paginates by re-calling
         // with the returned cursor until the response is smaller than the limit.
-        let notes: Vec<Note> = self
+        //
+        // Stranded-cursor detection runs in the SAME snapshot so the high-water
+        // is consistent with the rows read: a non-zero cursor strictly above the
+        // AUTOINCREMENT high-water can only come from a regressed seq space (the
+        // backing DB was recreated), and — unlike a legacy µs cursor — it is a
+        // plausible small `seq`, so `LEGACY_CURSOR_THRESHOLD` never catches it.
+        // Reset it to 0 so the client re-scans the current epoch; the effective
+        // cursor returned below lets the caller heal the client's stored position
+        // (see the gRPC fetch handler) instead of re-triggering the reset forever.
+        let (notes, effective_i64): (Vec<Note>, i64) = self
             .transact("fetch notes by tags", move |conn| {
                 use schema::notes::dsl::{notes, seq, tag};
+
+                // Only a non-zero cursor can be stranded. This guard also makes
+                // the heal converge: a reset lands the client (and, on the push
+                // path, the stored subscription cursor) at 0, and cursor 0 never
+                // re-enters this branch — so the reset fires at most once per
+                // stranding event, not every poll.
+                let mut effective = cursor_i64;
+                if effective > 0 {
+                    if let Some(high_water) = high_water_seq(conn) {
+                        // Strict `>` is deliberate: a caught-up client sits at
+                        // `cursor == high_water` and must NOT be reset (that would
+                        // re-deliver the whole epoch every steady-state poll). The
+                        // accepted residual edge is that if a recreated epoch grows
+                        // to exactly the stranded cursor before the client's next
+                        // poll, the reset is skipped and the client misses that
+                        // epoch's `1..=cursor` backlog — a one-insert-wide
+                        // coincidence with bounded impact.
+                        if effective > high_water {
+                            effective = 0;
+                        }
+                    }
+                }
+
                 let fetched_notes = notes
                     .filter(tag.eq_any(&tag_values))
-                    .filter(seq.gt(cursor_i64))
+                    .filter(seq.gt(effective))
                     .order(seq.asc())
                     .limit(FETCH_NOTES_BATCH_SIZE)
                     // Name-based column selection (via `Selectable`) so a future
                     // mid-table column insert can't silently misalign fields.
                     .select(Note::as_select())
                     .load(conn)?;
-                Ok(fetched_notes)
+                Ok((fetched_notes, effective))
             })
             .await?;
 
+        // A post-legacy cursor that came back as 0 was reset by the high-water
+        // check above (a legacy reset already zeroed `post_legacy_cursor`, so it
+        // is excluded by the `> 0` guard and never counted here).
+        if post_legacy_cursor > 0 && effective_i64 == 0 {
+            self.metrics.db_fetch_notes_stranded_cursor_reset();
+            tracing::warn!(
+                original_cursor = cursor,
+                "Cursor above seq high-water reset to 0 (server seq space regressed; DB recreated)"
+            );
+        }
+
         let mut stored_notes = Vec::new();
         for note in notes {
             let stored_note = StoredNote::try_from(note).map_err(|e| {
@@ -210,7 +291,8 @@ impl DatabaseBackend for SqliteDatabase {
         tracing::Span::current().record("notes_returned", stored_notes.len());
         timer.finish("ok");
 
-        Ok(stored_notes)
+        let effective_cursor: u64 = effective_i64.try_into().unwrap_or(0);
+        Ok((stored_notes, effective_cursor))
     }
 
     async fn get_stats(&self) -> Result<(u64, u64), DatabaseError> {

crates/node/src/metrics.rs

@@ -39,6 +39,8 @@ pub struct MetricsDatabase {
     fetch_notes_duration: Histogram<f64>,
     // legacy cursor reset (pre-seq-migration clients)
     fetch_notes_legacy_cursor_reset_count: Counter<u64>,
+    // stranded cursor reset (cursor above the seq high-water → DB recreated)
+    fetch_notes_stranded_cursor_reset_count: Counter<u64>,
     // Maintenance
     maintenance_cleanup_notes_count: Counter<u64>,
     maintenance_cleanup_notes_duration: Histogram<f64>,
@@ -178,6 +180,16 @@ impl MetricsDatabase {
             )
             .build();
 
+        let fetch_notes_stranded_cursor_reset_count = meter
+            .u64_counter("db_fetch_notes_stranded_cursor_reset_count")
+            .with_description(
+                "Number of fetch_notes() requests where the client's cursor was at or \
+                 below the legacy threshold but strictly above the current seq \
+                 high-water and reset to 0 (server seq space regressed — backing DB \
+                 recreated)",
+            )
+            .build();
+
         let maintenance_cleanup_notes_count = meter
             .u64_counter("db_maintenance_cleanup_notes_count")
             .with_description("Total number of DB maintenance cleanup_old_notes() requests")
@@ -195,6 +207,7 @@ impl MetricsDatabase {
             fetch_notes_count,
             fetch_notes_duration,
             fetch_notes_legacy_cursor_reset_count,
+            fetch_notes_stranded_cursor_reset_count,
             maintenance_cleanup_notes_count,
             maintenance_cleanup_notes_duration,
         }
@@ -227,6 +240,12 @@ impl MetricsDatabase {
         self.fetch_notes_legacy_cursor_reset_count.add(1, &[]);
     }
 
+    /// Record a stranded-cursor reset: the client's cursor was above the current
+    /// seq high-water and reset to 0 (server seq space regressed — DB recreated).
+    pub fn db_fetch_notes_stranded_cursor_reset(&self) {
+        self.fetch_notes_stranded_cursor_reset_count.add(1, &[]);
+    }
+
     /// Measure a DB maintenance cleanup-old-notes procedure
     ///
     /// Increases the request counter and measures request duration.