From 7ff6f86e7b60e50f2ad4375bde1b1877ec33cfeb Mon Sep 17 00:00:00 2001 From: rays Date: Mon, 12 Oct 2026 15:31:24 +0000 Subject: [PATCH] Share with the directory only when the snapshot changed, or half-hourly (#211) A scan wakes the sharer, and a scan runs every half minute or so since each checks only the feeds that are due: Tower sent the same 1,471 feeds every 33 seconds and the directory rewrote its rows each time. The snapshot, with the address and token it goes to, is compared with what last went; the same again waits for SHARE_EVERY. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 4 +++ crates/ipx/src/directory.rs | 64 ++++++++++++++++++++++++++++++------- 2 files changed, 56 insertions(+), 12 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index e4f2552..f007589 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -20,6 +20,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - `docker-compose.yml` is a sample to start from, ipx alone on SQLite with its folders beside the file, instead of a copy of the author's server (#208). - The source is a Cargo workspace: the daemon in `crates/ipx`, feed fetching and parsing in `crates/feed`. `cargo build --release` still makes `target/release/ipx` (#202). +### Fixed + +- ipx sends the directory its feeds when they change, and every half hour, instead of after every scan, which was every half minute or so (#211). + ## [0.12.0] - 2026-10-11 ### Added diff --git a/crates/ipx/src/directory.rs b/crates/ipx/src/directory.rs index 8d77d3f..f5c53f3 100644 --- a/crates/ipx/src/directory.rs +++ b/crates/ipx/src/directory.rs @@ -120,11 +120,14 @@ pub fn snapshot_entries(catalogue: &[Sub], counts: &HashMap) -> Vec } /// Sends the snapshot now. -pub async fn share(ctx: &Ctx) -> Result { +async fn snapshot(ctx: &Ctx) -> Result, DirError> { let catalogue = crate::subscriptions(ctx).await.map_err(|e| DirError::Unreachable(format!("{e:#}")))?; let counts = ctx.db.subscriber_counts().await.map_err(|e| DirError::Unreachable(format!("{e:#}")))?; - let entries = snapshot_entries(&catalogue, &counts); - let r: SnapshotResult = call(ctx, reqwest::Method::POST, "/v1/snapshot", Some(&entries)).await?; + Ok(snapshot_entries(&catalogue, &counts)) +} + +pub async fn share(ctx: &Ctx, entries: &[SnapshotEntry]) -> Result { + let r: SnapshotResult = call(ctx, reqwest::Method::POST, "/v1/snapshot", Some(entries)).await?; if r.refused_private > 0 { // The directory caught what this side should have: worth knowing which check differs. tracing::warn!(refused = r.refused_private, "the directory refused feeds as private that ipx sent"); @@ -134,19 +137,38 @@ pub async fn share(ctx: &Ctx) -> Result { /// Sends the snapshot every half hour while sharing is on, and at once when the settings change. pub async fn run(ctx: std::sync::Arc, mut stop: tokio::sync::watch::Receiver) { + // What went last, where, and when: a scan wakes this every half minute or so, and the same + // 1,500 feeds went each time, the directory rewriting its rows for nothing (#211). + let mut last: Option<(Vec, std::time::Instant)> = None; loop { let cfg = ctx.cfg(); if cfg.general.directory_share && configured(&cfg) { - let got = share(&ctx).await; - match &got { - Ok(r) => tracing::info!(accepted = r.accepted, refused_private = r.refused_private, refused_blocked = r.refused_blocked, "shared with the directory"), - Err(e) => tracing::warn!(error = %e, "could not share with the directory"), - } - let (result, error) = match got { - Ok(r) => (Some(r), None), - Err(e) => (None, Some(e.to_string())), + let got = match snapshot(&ctx).await { + Ok(entries) => { + let key = serde_json::to_vec(&(target(&cfg), &entries)).unwrap_or_default(); + if resend(last.as_ref(), &key) { + let got = share(&ctx, &entries).await; + if got.is_ok() { + last = Some((key, std::time::Instant::now())); + } + Some(got) + } else { + None + } + } + Err(e) => Some(Err(e)), }; - *ctx.directory_shared.lock().unwrap() = Some(Shared { at: crate::db::now(), result, error }); + if let Some(got) = got { + match &got { + Ok(r) => tracing::info!(accepted = r.accepted, refused_private = r.refused_private, refused_blocked = r.refused_blocked, "shared with the directory"), + Err(e) => tracing::warn!(error = %e, "could not share with the directory"), + } + let (result, error) = match got { + Ok(r) => (Some(r), None), + Err(e) => (None, Some(e.to_string())), + }; + *ctx.directory_shared.lock().unwrap() = Some(Shared { at: crate::db::now(), result, error }); + } } tokio::select! { _ = tokio::time::sleep(SHARE_EVERY) => {} @@ -156,6 +178,13 @@ pub async fn run(ctx: std::sync::Arc, mut stop: tokio::sync::watch::Receive } } +/// Whether to send the snapshot `key` stands for: one that changed, or went to another directory +/// or with another token, goes now; the same again only once SHARE_EVERY has passed, so a +/// directory that lost it, or dropped this server's counts, has it back within the half hour. +fn resend(last: Option<&(Vec, std::time::Instant)>, key: &[u8]) -> bool { + last.is_none_or(|(k, at)| k != key || at.elapsed() >= SHARE_EVERY) +} + #[cfg(test)] mod tests { use super::*; @@ -198,4 +227,15 @@ mod tests { cfg.general.directory_token = "t".into(); assert_eq!(target(&cfg), Some(("https://ipx-dir.example".into(), "t".into()))); } + + #[test] + fn the_same_snapshot_is_not_sent_again_within_the_half_hour() { + let now = std::time::Instant::now(); + assert!(resend(None, b"a")); + assert!(!resend(Some(&(b"a".to_vec(), now)), b"a")); + assert!(resend(Some(&(b"a".to_vec(), now)), b"b")); + if let Some(long_ago) = now.checked_sub(SHARE_EVERY) { + assert!(resend(Some(&(b"a".to_vec(), long_ago)), b"a")); + } + } }