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 <noreply@anthropic.com>

crates/ipx/src/directory.rs

@@ -120,11 +120,14 @@ pub fn snapshot_entries(catalogue: &[Sub], counts: &HashMap<String, i64>) -> Vec
 }
 
 /// Sends the snapshot now.
-pub async fn share(ctx: &Ctx) -> Result<SnapshotResult, DirError> {
+async fn snapshot(ctx: &Ctx) -> Result<Vec<SnapshotEntry>, 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<SnapshotResult, DirError> {
+    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<SnapshotResult, DirError> {
 
 /// 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<Ctx>, mut stop: tokio::sync::watch::Receiver<bool>) {
+    // 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<u8>, 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<Ctx>, 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<u8>, 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"));
+        }
+    }
 }