mirror of
https://github.com/block/buzz.git
synced 2026-10-27 12:40:30 +00:00
fix(read-state): bound the follow refresh interval by the dormancy window
Ingest refreshed a live follow's active_at at most daily, so with a dormancy window shorter than a day a mention could not revive a follow that had gone dormant. The refresh interval is now min(1 day, window/4), from the window the relay sets at startup, so it never exceeds the window. Adds codex00's repro: a 1-hour window, a follow last active 2 hours ago, and a fresh mention revives it in the sidebar and the threads view; it fails against the fixed one-day throttle. Co-authored-by: murderbot <3754f8729004d95654c46dbab3129e4ab9ef05cc2534e2a3fbfc155983bd637b@buzz.block.builderlab.xyz> Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Signed-off-by: Bradley Axen <baxen@squareup.com>
This commit is contained in:
@@ -1,11 +1,30 @@
|
||||
//! Read state that ingest writes. A thread frontier row exists for a thread
|
||||
//! its actor follows or has unfollowed; ingest and follow intents create one.
|
||||
use super::model::ELIGIBLE_KINDS;
|
||||
use super::model::{DEFAULT_RETENTION_SECONDS, ELIGIBLE_KINDS};
|
||||
use crate::{AdmittedTx, Result};
|
||||
use chrono::{DateTime, Utc};
|
||||
use nostr::Event;
|
||||
use std::sync::atomic::{AtomicU32, Ordering};
|
||||
use uuid::Uuid;
|
||||
|
||||
/// The relay's follow dormancy window, in seconds (`BUZZ_V1_RETENTION_SECONDS`).
|
||||
static FOLLOW_WINDOW_SECONDS: AtomicU32 = AtomicU32::new(DEFAULT_RETENTION_SECONDS);
|
||||
|
||||
/// Set the follow dormancy window ingest refreshes against: the same window
|
||||
/// the read projections are given. Process-wide; the relay sets it once at
|
||||
/// startup.
|
||||
pub fn set_follow_window_seconds(seconds: u32) {
|
||||
FOLLOW_WINDOW_SECONDS.store(seconds, Ordering::Relaxed);
|
||||
}
|
||||
|
||||
/// How stale a live follow's involvement may grow before an involvement that
|
||||
/// does not otherwise write its row refreshes it: a day, or a quarter of the
|
||||
/// window when that is shorter. Never longer than the window, so such an
|
||||
/// involvement always revives a dormant follow.
|
||||
fn refresh_after_seconds() -> f64 {
|
||||
f64::from((FOLLOW_WINDOW_SECONDS.load(Ordering::Relaxed) / 4).min(24 * 60 * 60))
|
||||
}
|
||||
|
||||
/// Where a newly stored channel message sits.
|
||||
#[derive(Clone, Copy)]
|
||||
pub(crate) enum Place<'a> {
|
||||
@@ -133,7 +152,8 @@ pub(crate) async fn record_message(
|
||||
AND c.channel_type='dm' AND cm.pubkey<>$4
|
||||
), rows AS (
|
||||
-- An existing follower keeps their position, so is only touched
|
||||
-- to keep the follow live, at most daily rather than per message;
|
||||
-- to keep the follow live, at most once per refresh interval ($12)
|
||||
-- rather than per message;
|
||||
-- an unfollow is not touched unless this message re-follows (the
|
||||
-- author posting, or a mention). Existence probes are LIMIT 1
|
||||
-- laterals: always one index lookup.
|
||||
@@ -148,7 +168,7 @@ pub(crate) async fn record_message(
|
||||
AND (i.message IS NOT NULL OR f.unfollowed IS NULL
|
||||
OR (i.refollow AND f.unfollowed)
|
||||
OR (i.root_id<>''::bytea AND NOT f.unfollowed
|
||||
AND f.active_at < now()-interval '1 day'))
|
||||
AND f.active_at < now()-make_interval(secs=>$12)))
|
||||
ORDER BY i.actor, i.root_id, i.message IS NULL, NOT i.refollow
|
||||
), accounts AS (
|
||||
INSERT INTO personal_read_accounts (community_id, actor)
|
||||
@@ -184,7 +204,7 @@ pub(crate) async fn record_message(
|
||||
WHERE excluded.through_message_id IS NOT NULL OR (personal_read_frontiers.unfollowed
|
||||
AND excluded.through_timestamp > personal_read_frontiers.through_timestamp)
|
||||
OR (excluded.root_id<>''::bytea AND NOT personal_read_frontiers.unfollowed
|
||||
AND personal_read_frontiers.active_at < now()-interval '1 day')",
|
||||
AND personal_read_frontiers.active_at < now()-make_interval(secs=>$12))",
|
||||
)
|
||||
.bind(community.as_uuid())
|
||||
.bind(channel)
|
||||
@@ -197,6 +217,7 @@ pub(crate) async fn record_message(
|
||||
.bind(reply)
|
||||
.bind(root_created_at)
|
||||
.bind(broadcast)
|
||||
.bind(refresh_after_seconds())
|
||||
.execute(tx.conn())
|
||||
.await?;
|
||||
Ok(())
|
||||
|
||||
@@ -866,3 +866,75 @@ async fn an_agent_keeps_no_read_state() {
|
||||
.await
|
||||
.is_some());
|
||||
}
|
||||
|
||||
/// How many threads the sidebar and the threads view list under `window`.
|
||||
async fn listed_within(
|
||||
db: &Db,
|
||||
community: CommunityId,
|
||||
channel: Uuid,
|
||||
actor: &Keys,
|
||||
window: u32,
|
||||
) -> (usize, usize) {
|
||||
let sidebar = db
|
||||
.personal_read_sidebar_channels(community, &actor.public_key(), window, &[channel])
|
||||
.await
|
||||
.unwrap()
|
||||
.channels
|
||||
.remove(0)
|
||||
.threads
|
||||
.len();
|
||||
let view = db
|
||||
.personal_read_threads(community, &actor.public_key(), window, MAX_THREADS, None)
|
||||
.await
|
||||
.unwrap()
|
||||
.threads
|
||||
.len();
|
||||
(sidebar, view)
|
||||
}
|
||||
|
||||
/// A window shorter than a day still revives on a mention: ingest refreshes
|
||||
/// against the same window the projections read.
|
||||
#[tokio::test]
|
||||
#[ignore = "requires Postgres"]
|
||||
async fn a_mention_revives_a_follow_under_a_sub_day_window() {
|
||||
const HOUR: u32 = 60 * 60;
|
||||
set_follow_window_seconds(HOUR);
|
||||
let (db, pool, community, channel, actor, root) = fixture().await;
|
||||
reply(
|
||||
&db,
|
||||
community,
|
||||
channel,
|
||||
&root,
|
||||
&Keys::generate(),
|
||||
mention(&actor),
|
||||
)
|
||||
.await;
|
||||
sqlx::query(
|
||||
"UPDATE personal_read_frontiers SET active_at=now()-interval '2 hours'
|
||||
WHERE community_id=$1 AND actor=$2",
|
||||
)
|
||||
.bind(community.as_uuid())
|
||||
.bind(actor.public_key().to_bytes().as_slice())
|
||||
.execute(&pool)
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(
|
||||
listed_within(&db, community, channel, &actor, HOUR).await,
|
||||
(0, 0),
|
||||
"dormant under the window"
|
||||
);
|
||||
reply(
|
||||
&db,
|
||||
community,
|
||||
channel,
|
||||
&root,
|
||||
&Keys::generate(),
|
||||
mention(&actor),
|
||||
)
|
||||
.await;
|
||||
assert_eq!(
|
||||
listed_within(&db, community, channel, &actor, HOUR).await,
|
||||
(1, 1),
|
||||
"a fresh mention revives it"
|
||||
);
|
||||
}
|
||||
|
||||
@@ -12,6 +12,7 @@ mod model;
|
||||
mod projection;
|
||||
mod writes;
|
||||
|
||||
pub use follows::set_follow_window_seconds;
|
||||
pub(crate) use follows::{place, record_message};
|
||||
pub use model::*;
|
||||
|
||||
|
||||
@@ -388,6 +388,7 @@ async fn run_relay_main(boot: BootTracker) -> anyhow::Result<()> {
|
||||
anyhow::anyhow!("Configuration error: {error}")
|
||||
})?;
|
||||
boot = next_boot;
|
||||
buzz_db::personal_read::set_follow_window_seconds(config.buzz_v1_retention_seconds);
|
||||
|
||||
let key_failure = if config.relay_private_key.is_some() {
|
||||
LifecycleReason::RequiredInvalid
|
||||
|
||||
@@ -150,8 +150,8 @@ thread is neither counted nor listed, but keeps its row and position, and any
|
||||
of those involvements makes it live again at that position. Replies that do not
|
||||
involve you never revive it. A follow made by the `follow` intent never goes
|
||||
dormant; an unfollow ends that. Being mentioned, a reply to your root and DM
|
||||
replies refresh a live follow at most daily, so dormancy is measured to the
|
||||
day.
|
||||
replies refresh a live follow at most once a day, or once a quarter of the
|
||||
window when that is shorter, so dormancy is measured to that interval.
|
||||
|
||||
## Threads view
|
||||
|
||||
|
||||
Reference in New Issue
Block a user