Unread lists start from each person's oldest unread item (#347)

All Subscriptions' Unread page, its count and the feed list's unread
counts read every item of the person's feeds and hashed all their
entry_state rows to find the few unread: 11,727 items and 22,949 rows
for 44, 27-32 ms on production, growing with the back catalogue. A
NOT EXISTS rewrite of the same scan was no faster.

subscriptions.unread_from says every item of the feed first seen
before it is read by this person. Unread queries join the subscription
on e.first_seen >= coalesce(unread_from, 0), and the new index
entries_seen (feed_id, first_seen) makes that a range per feed.

Nothing has to keep a list in step, so nothing can drift: the reap job
and daemon start move it up to the oldest unread item (hidden ones
too, since a block list can be lifted), never past an hour ago, so an
item a scan is storing meanwhile is not passed over. The one way an
older item becomes unread, marking it so, clears the person's value
for that feed. NULL, or stale, only means reading more.

The filter test now subscribes its reader, as every list's reader is.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

crates/ipx/CHANGELOG.md

@@ -19,6 +19,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - A page of items asks for its count beside the items, and the feed list asks its five questions at once, where each waited for the one before (#346).
 - A Signal's count of items and of unread ones are worked out in one pass, not two (#348).
 - Artwork kept on disk is read and written off the threads that answer requests, and two requests for the same missing artwork fetch it once (#345, #349).
+- Unread lists and counts read a feed's items from each person's oldest unread one on, which the daily cleanup works out, not the whole back catalogue (#347).
 
 ### Fixed
 

crates/ipx/src/db.rs

@@ -86,6 +86,8 @@ async fn create_missing(orm: &sea_orm::DatabaseConnection) -> Result<()> {
         // the expression it is asked by. Without it, each wake read all of entries, 605 MB, 566 ms
         // to find nothing (#274).
         "CREATE INDEX IF NOT EXISTS entries_recent ON entries (feed_id, (coalesce(published, first_seen)))",
+        // Unread lists read a feed's items from `subscriptions.unread_from` on (#347).
+        "CREATE INDEX IF NOT EXISTS entries_seen ON entries (feed_id, first_seen)",
     ] {
         orm.execute_unprepared(sql).await.with_context(|| sql.to_owned())?;
     }
@@ -166,6 +168,10 @@ async fn create_missing(orm: &sea_orm::DatabaseConnection) -> Result<()> {
         let sql = "ALTER TABLE signals ADD COLUMN icon TEXT";
         orm.execute_unprepared(sql).await.with_context(|| sql.to_owned())?;
     }
+    if orm.execute_unprepared("SELECT unread_from FROM subscriptions LIMIT 0").await.is_err() {
+        let sql = "ALTER TABLE subscriptions ADD COLUMN unread_from BIGINT";
+        orm.execute_unprepared(sql).await.with_context(|| sql.to_owned())?;
+    }
     if orm.execute_unprepared("SELECT pause_ms FROM deadair_prefs LIMIT 0").await.is_err() {
         let sql = "ALTER TABLE deadair_prefs ADD COLUMN pause_ms BIGINT";
         orm.execute_unprepared(sql).await.with_context(|| sql.to_owned())?;
@@ -541,6 +547,7 @@ impl Db {
             "unread",
             "SELECT e.feed_id, count(*) AS n FROM entries e
                JOIN subscriptions sub ON sub.user_id = $1 AND sub.feed_id = e.feed_id
+                                     AND e.first_seen >= coalesce(sub.unread_from, 0)
                LEFT JOIN entry_state s
                       ON s.user_id = $1 AND s.feed_id = e.feed_id AND s.guid = e.guid
               WHERE NOT coalesce(s.read, false)
@@ -1236,8 +1243,20 @@ fn entries_from(
         }
         None => "true".into(),
     };
+    // Unread starts from where this person's unread begins in each feed, so it reads the few
+    // items since, not every one: 11,727 read for 44 unread, 32 ms (#347). Every list is of
+    // subscribed feeds, so the join takes nothing away.
+    let since = if filter == Filter::Unread {
+        format!(
+            "JOIN subscriptions w ON w.user_id = {user} AND w.feed_id = e.feed_id
+                               AND e.first_seen >= coalesce(w.unread_from, 0)"
+        )
+    } else {
+        String::new()
+    };
     format!(
         "FROM entries e
+         {since}
          LEFT JOIN entry_state s ON s.user_id = {user} AND s.feed_id = e.feed_id AND s.guid = e.guid
          WHERE {scope} AND {} AND {search} AND {}",
         filter.sql(),
@@ -2278,9 +2297,37 @@ impl Db {
             vec![user_id.into(), feed_id.into(), guid.into(), on.into()],
         )
         .await?;
+        // An item marked unread again may be older than where this person's unread was known to
+        // start, so that is forgotten until the next reap works it out (`advance_unread_from`).
+        if matches!(field, EntryFlag::Read) && !on {
+            self.exec(
+                "UPDATE subscriptions SET unread_from = NULL
+                  WHERE user_id = $1 AND feed_id = $2 AND unread_from IS NOT NULL",
+                vec![user_id.into(), feed_id.into()],
+            )
+            .await?;
+        }
         Ok(())
     }
 
+    /// Moves each subscription's `unread_from` up to its oldest unread item, hidden ones included,
+    /// since a block list can be lifted. No more than an hour ago: an item a scan is storing now,
+    /// not yet visible to this statement, is first seen about now. Returns how many moved.
+    pub async fn advance_unread_from(&self) -> Result<u64> {
+        let cutoff = now() - 3600;
+        self.exec(
+            "UPDATE subscriptions SET unread_from = (
+                 SELECT CASE WHEN min(e.first_seen) < $1 THEN min(e.first_seen) ELSE $1 END
+                   FROM entries e
+                   LEFT JOIN entry_state s
+                          ON s.user_id = subscriptions.user_id AND s.feed_id = e.feed_id AND s.guid = e.guid
+                  WHERE e.feed_id = subscriptions.feed_id AND NOT coalesce(s.read, false))
+              WHERE true",
+            vec![cutoff.into()],
+        )
+        .await
+    }
+
     /// How many files this feed has on disk. Decides whether a feed dropped from an OPML
     /// can be removed or must be kept.
     pub async fn downloaded_count(&self, feed_id: &str) -> Result<i64> {
@@ -3019,6 +3066,51 @@ pub fn now() -> i64 {
 mod tests {
     use super::*;
 
+    #[tokio::test]
+    async fn unread_starts_where_it_was_worked_out_and_loses_nothing() {
+        let db = Db::memory().await.unwrap();
+        let me = db.create_user("me", None, false).await.unwrap();
+        db.exec_for_test(
+            "INSERT INTO feeds (id, url) VALUES ('f', 'http://x/f.xml');
+             INSERT INTO entries (feed_id, guid, first_seen) VALUES ('f','old',100),('f','mid',200),('f','new',300);",
+        )
+        .await
+        .unwrap();
+        db.subscribe(me, "f").await.unwrap();
+        db.set_entry_flag(me, "f", "old", EntryFlag::Read, true).await.unwrap();
+        let unread = async || {
+            let order = order_sql("published", "desc");
+            let rows = db.entries_of(me, None, Filter::Unread, None, 0, 50, &order).await.unwrap();
+            let mut guids: Vec<String> = rows.into_iter().map(|r| r.guid).collect();
+            guids.sort();
+            let count = db.count_of(me, None, Filter::Unread, None).await.unwrap();
+            let listed = db.feed_list(me, None).await.unwrap()["f"].unread;
+            assert_eq!(count, guids.len() as i64);
+            assert_eq!(listed, count, "the feed list counts what the list shows");
+            guids
+        };
+        assert_eq!(unread().await, ["mid", "new"]);
+
+        db.advance_unread_from().await.unwrap();
+        let from = async || -> Option<i64> {
+            db.rows("SELECT unread_from FROM subscriptions", vec![]).await.unwrap()[0]
+                .try_get("", "unread_from")
+                .unwrap()
+        };
+        assert_eq!(from().await, Some(200), "the oldest unread");
+        assert_eq!(unread().await, ["mid", "new"], "nothing unread goes missing (#347)");
+
+        // An old item marked unread again comes back, though it is older than where unread began.
+        db.set_entry_flag(me, "f", "old", EntryFlag::Read, false).await.unwrap();
+        assert_eq!(unread().await, ["mid", "new", "old"]);
+
+        // With everything read, it starts no later than an hour ago.
+        db.mark_all_read(me, &["f".into()]).await.unwrap();
+        db.advance_unread_from().await.unwrap();
+        assert!(from().await.is_some_and(|t| t <= now() - 3600));
+        assert!(unread().await.is_empty());
+    }
+
     #[tokio::test]
     async fn a_new_password_signs_the_account_out_everywhere_but_keeps_its_tokens() {
         let db = Db::memory().await.unwrap();
@@ -3620,6 +3712,8 @@ mod tests {
                 (1,'f','b','u1','/tmp/b','done');
              -- Read and starred belong to a person now, so say which one.
              INSERT INTO users (id, name, is_admin) VALUES (7,'reader',true);
+             -- Every list is of feeds its reader subscribes to; Unread joins the subscription.
+             INSERT INTO subscriptions (user_id, feed_id) VALUES (7,'f');
              INSERT INTO entry_state (user_id, feed_id, guid, read, flagged, position) VALUES
                 (7,'f','b',true,false,0),
                 (7,'f','c',true,true,0),

crates/ipx/src/entity.rs

@@ -230,6 +230,10 @@ pub mod subscriptions {
         /// Left out of every Signal, and never put to Jev for them, as `hide_from_all` (#289).
         #[sea_orm(default_value = false)]
         pub hide_from_signals: bool,
+        /// Every item of the feed first seen before this is read by this person (#347): unread
+        /// lists start here instead of reading the whole back catalogue. NULL knows nothing, and
+        /// the daily reap moves it up (`Db::advance_unread_from`).
+        pub unread_from: Option<i64>,
     }
 
     owned_by_user!();

crates/ipx/src/main.rs

@@ -651,6 +651,11 @@ async fn daemon(
         );
     }
 
+    // Unread lists start from here, and a reap may be a day off (#347). Slower without, never wrong.
+    if let Err(e) = ctx.db.advance_unread_from().await {
+        tracing::warn!(error = %format!("{e:#}"), "could not work out where unread items start");
+    }
+
     if let Some(admin) = ctx.db.users().await?.into_iter().find(|u| u.is_admin) {
         let catalogue: Vec<String> = ctx.cfg().feeds.keys().cloned().collect();
         match ctx.db.adopt_catalogue(admin.id, &catalogue).await {
@@ -1238,6 +1243,9 @@ async fn reap(ctx: &Ctx, dry_run: bool, standalone: bool) -> Result<()> {
     span.record("entries", r.entries_pruned);
     if !dry_run {
         tracing::Instrument::instrument(clean_directory(ctx), db::step("clean_directory")).await?;
+        let moved =
+            tracing::Instrument::instrument(ctx.db.advance_unread_from(), db::step("advance_unread_from")).await?;
+        tracing::debug!(moved, "unread lists start from each person's oldest unread item");
         // Off the async threads: it walks every file kept, on FUSE (#345).
         let (span, limit) = (db::step("trim_artwork"), ctx.cfg().general.art_cache_mb * 1_048_576);
         let gone = tokio::task::spawn_blocking(move || span.in_scope(|| art::trim(&art::dir(), limit))).await?;