From a11e3b6367d5b30d238c6edb7980d0f4f37d110d Mon Sep 17 00:00:00 2001 From: rays Date: Fri, 23 Oct 2026 16:04:32 +0000 Subject: [PATCH] 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 --- crates/ipx/CHANGELOG.md | 1 + crates/ipx/src/db.rs | 94 ++++++++++++++++++++++++++++++++++++++++ crates/ipx/src/entity.rs | 4 ++ crates/ipx/src/main.rs | 8 ++++ 4 files changed, 107 insertions(+) diff --git a/crates/ipx/CHANGELOG.md b/crates/ipx/CHANGELOG.md index 9f0cc51..5b658d7 100644 --- a/crates/ipx/CHANGELOG.md +++ b/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 diff --git a/crates/ipx/src/db.rs b/crates/ipx/src/db.rs index 66de3b3..ee4df67 100644 --- a/crates/ipx/src/db.rs +++ b/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 { + 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 { @@ -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 = 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 { + 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), diff --git a/crates/ipx/src/entity.rs b/crates/ipx/src/entity.rs index dfd6410..4b4a6cc 100644 --- a/crates/ipx/src/entity.rs +++ b/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, } owned_by_user!(); diff --git a/crates/ipx/src/main.rs b/crates/ipx/src/main.rs index 6c6ec22..6b3689a 100644 --- a/crates/ipx/src/main.rs +++ b/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 = 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?;