Keep two changes to the catalogue made at once from undoing each other (#153)

Each change to the catalogue or the server settings stored the whole thing from a copy it took
earlier, and nothing kept two from overlapping. Adding a feed copied it first and then spent up to
a minute finding and reading the feed, so another person's new feed, an unsubscribe, an admin's
settings, or the scan's own changes (a moved feed's address, a category) made in that minute were
stored over when it finished, with no error. A feed dropped that way left its subscriber's list.

Every change now takes its copy through Ctx::edit_cfg, which holds a lock until store_cfg keeps it,
and store_cfg takes nothing else, so no writer can skip it. Adding a feed takes its copy once it
has found the feed, and before the duplicate check, so two people adding one feed add it once.
reload takes the lock too. `ipx add`, `rm` and `import`, in a process of their own, are outside it,
as the comment on the lock says.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

src/main.rs

@@ -145,6 +145,32 @@ pub struct Ctx {
     /// failure that says nothing about the feed, kept off it and out of the database (#143).
     /// Forgotten on a restart, which only makes the retry sooner.
     pub retry: std::sync::Mutex<std::collections::HashMap<String, i64>>,
+    /// Held from taking a copy of the config to storing it (`edit_cfg`). Each change stores the
+    /// whole catalogue from its copy, and two at once stored over each other: a feed added while
+    /// another add spent a minute fetching was gone once that one stored (#153).
+    // ponytail: this process only. `ipx add`, `rm` and `import` write from a process of their
+    // own, and a change in flight here can still store over theirs; take the copy from the
+    // database rather than memory if that ever happens.
+    pub cfg_write: tokio::sync::Mutex<()>,
+}
+
+/// A copy of the config to change, holding off every other change until `store_cfg` keeps it.
+pub struct CfgEdit<'a> {
+    cfg: config::Config,
+    _held: tokio::sync::MutexGuard<'a, ()>,
+}
+
+impl std::ops::Deref for CfgEdit<'_> {
+    type Target = config::Config;
+    fn deref(&self) -> &config::Config {
+        &self.cfg
+    }
+}
+
+impl std::ops::DerefMut for CfgEdit<'_> {
+    fn deref_mut(&mut self) -> &mut config::Config {
+        &mut self.cfg
+    }
 }
 
 /// How soon a feed whose host could not be looked up is tried again (#143).
@@ -177,12 +203,18 @@ impl Ctx {
         self.cfg.read().unwrap().clone()
     }
 
-    /// Re-reads config.toml into the live snapshot.
+    /// The config as it stands, to change and then keep with `store_cfg`; any other change waits
+    /// until then.
+    pub async fn edit_cfg(&self) -> CfgEdit<'_> {
+        let held = self.cfg_write.lock().await;
+        CfgEdit { cfg: (*self.cfg()).clone(), _held: held }
+    }
+
     /// Keeps a changed catalogue or server settings: in the database, and for everything running
     /// here from now on. What config.toml's save and reload did, before the database held them.
-    pub async fn store_cfg(&self, cfg: config::Config) -> Result<()> {
-        self.db.store_config(&cfg).await?;
-        self.set_cfg(cfg);
+    pub async fn store_cfg(&self, edit: CfgEdit<'_>) -> Result<()> {
+        self.db.store_config(&edit.cfg).await?;
+        self.set_cfg(edit.cfg);
         Ok(())
     }
 
@@ -308,6 +340,7 @@ async fn main() -> Result<()> {
         config_path: config_path.clone(),
         detach_torrents: is_daemon,
         retry: Default::default(),
+        cfg_write: Default::default(),
     });
 
     let result = match cli.command {
@@ -502,6 +535,7 @@ async fn run(ctx: &Arc<Ctx>, cmd: Cmd) -> Result<()> {
 /// memory, for a daemon told another ipx changed them (#120); config.toml, read again, for where
 /// things are and who gets in.
 async fn reload(ctx: &Ctx) -> Result<()> {
+    let _held = ctx.cfg_write.lock().await;
     let cfg = assemble_config(&ctx.db, config::Config::load(&ctx.config_path)?, &ctx.config_path).await?;
     ctx.set_cfg(cfg);
     Ok(())
@@ -754,7 +788,7 @@ async fn add(
     category: Option<String>,
     list: bool,
 ) -> Result<()> {
-    let mut cfg = (*ctx.cfg()).clone();
+    let mut cfg = ctx.edit_cfg().await;
     let url = &feed::find_feed(&ctx.client, &feed::expand_input(url)).await?;
     // Includes feeds derived from an OPML, or the same show could be added twice.
     if let Some(existing) = subscriptions(ctx).await?.iter().find(|s| feed::same_feed(&s.cfg.url, url)) {
@@ -852,7 +886,7 @@ fn url_stem(url: &str) -> String {
 }
 
 async fn rm(ctx: &Ctx, feed: &str) -> Result<()> {
-    let mut cfg = (*ctx.cfg()).clone();
+    let mut cfg = ctx.edit_cfg().await;
     if cfg.feeds.remove(feed).is_none() {
         // Derived from an OPML: drop it here, though the subscription will list it again
         // on the next read unless the OPML itself goes.
@@ -902,8 +936,8 @@ pub async fn subscribe_opml(
     let mut found = vec![];
     collect_outlines(&doc.body.outlines, &mut found);
 
+    let mut cfg = ctx.edit_cfg().await;
     let known = subscriptions(ctx).await?;
-    let mut cfg = (*ctx.cfg()).clone();
     let mut ids = vec![];
     let mut grew = false;
     for (title, url) in found {
@@ -1057,7 +1091,7 @@ async fn clean_directory(ctx: &Ctx) -> Result<()> {
     if stale.is_empty() {
         return Ok(());
     }
-    let mut cfg = (*ctx.cfg()).clone();
+    let mut cfg = ctx.edit_cfg().await;
     if stale.iter().fold(false, |any, id| cfg.feeds.remove(id).is_some() || any) {
         ctx.store_cfg(cfg).await?;
     }
@@ -1441,7 +1475,7 @@ async fn follow_move(ctx: &Ctx, id: &str, to: &str) -> Result<()> {
         tracing::info!(feed = id, to, "the feed moved to an address another feed already has; leaving it");
         return Ok(());
     }
-    let mut cfg = (*ctx.cfg()).clone();
+    let mut cfg = ctx.edit_cfg().await;
     let Some(f) = cfg.feeds.get_mut(id) else { return Ok(()) };
     let from = std::mem::replace(&mut f.url, to.to_owned());
     ctx.store_cfg(cfg).await?;
@@ -1608,7 +1642,7 @@ async fn scan_one(
     {
         match feed::classify(&ctx.client, &key, &parsed).await {
             Ok(category) => {
-                let mut cfg = (*ctx.cfg()).clone();
+                let mut cfg = ctx.edit_cfg().await;
                 if let Some(f) = cfg.feeds.get_mut(id).filter(|f| f.category.is_none()) {
                     tracing::info!(feed = id, category = %category, "categorised");
                     f.category = Some(category);
@@ -2319,9 +2353,33 @@ mod tests {
             config_path: PathBuf::new(),
             detach_torrents: false,
             retry: Default::default(),
+            cfg_write: Default::default(),
         }
     }
 
+    #[tokio::test]
+    async fn two_changes_to_the_catalogue_at_once_both_keep() {
+        let ctx = Arc::new(test_ctx(config::Config::default()).await);
+        // A slow change, as adding a feed is while it fetches, and a quick one made meanwhile.
+        let slow = {
+            let ctx = ctx.clone();
+            tokio::spawn(async move {
+                let mut cfg = ctx.edit_cfg().await;
+                tokio::time::sleep(std::time::Duration::from_millis(50)).await;
+                cfg.feeds.insert("a".into(), feed());
+                ctx.store_cfg(cfg).await.unwrap();
+            })
+        };
+        tokio::time::sleep(std::time::Duration::from_millis(10)).await;
+        let mut cfg = ctx.edit_cfg().await;
+        cfg.feeds.insert("b".into(), feed());
+        ctx.store_cfg(cfg).await.unwrap();
+        slow.await.unwrap();
+        assert_eq!(ctx.cfg().feeds.keys().collect::<Vec<_>>(), ["a", "b"], "neither stored over the other (#153)");
+        let (_, stored) = ctx.db.stored_config().await.unwrap().unwrap();
+        assert_eq!(stored.keys().collect::<Vec<_>>(), ["a", "b"]);
+    }
+
     #[tokio::test]
     async fn a_moments_lookup_failure_is_tried_again_soon_and_not_held_against_the_feed() {
         let ctx = test_ctx(config::Config::default()).await;

src/web.rs

@@ -1457,12 +1457,13 @@ async fn add_feed(
     user: crate::db::User,
     Json(body): Json<NewFeed>,
 ) -> Result<Json<serde_json::Value>, ApiError> {
-    let mut cfg = (*state.ctx.cfg()).clone();
     // Before the duplicate check, so a site's page finds the feed someone already has. What is
     // not a feed and links none is refused here, with why, and nothing is added.
     let url = crate::feed::find_feed(&state.ctx.client, &crate::feed::expand_input(&body.url))
         .await
         .map_err(|e| ApiError::bad_request(format!("{e:#}")))?;
+    // Taken before the duplicate check, so two people adding the same feed at once add it once.
+    let mut cfg = state.ctx.edit_cfg().await;
     // Someone else may already have it. Then adding costs nothing: no second fetch, no
     // second copy on disk, just another name against the same feed.
     if let Some(existing) = crate::subscriptions(&state.ctx).await?
@@ -1600,7 +1601,7 @@ async fn patch_feed(
             "the feed's address, folder, schedule and category are the same for everyone, so only an admin changes them",
         ));
     }
-    let mut cfg = (*state.ctx.cfg()).clone();
+    let mut cfg = state.ctx.edit_cfg().await;
 
     // Derived feeds have no config entry. Editing one is the moment it earns a real
     // entry: promote it, so the config holds your decisions and nothing else.
@@ -1677,7 +1678,7 @@ async fn remove_feed(
     // Nobody is left: the feed stops being scanned. Its files and history stay, so if
     // someone subscribes again they do not pull the back catalogue a second time. A feed an
     // admin listed stays in the Directory, for the next person.
-    let mut cfg = (*state.ctx.cfg()).clone();
+    let mut cfg = state.ctx.edit_cfg().await;
     if cfg.feeds.get(&id).is_some_and(|f| f.listed) {
         return Ok(StatusCode::NO_CONTENT);
     }
@@ -2156,7 +2157,7 @@ async fn patch_settings(
     if !user.is_admin {
         return Err(ApiError::forbidden("only an admin changes these settings"));
     }
-    let mut cfg = (*state.ctx.cfg()).clone();
+    let mut cfg = state.ctx.edit_cfg().await;
     if let Some(sched) = body.schedule {
         let sched = sched.trim().to_owned();
         if crate::config::parse_interval(&sched).is_none() {