rays/ipx
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 20 Wiki Activity

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

Browse Source 
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>
This commit is contained in:
Ray Slakinski
2026-10-23 16:04:32 +00:00
parent cb4ee5a570
commit a11e3b6367
4 changed files with 107 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

94
crates/ipx/src/db.rs
View File

@@ -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),