Merge pull request #207 from ikatson/fix-persistence-bugs

Fix persistence bugs

crates/librqbit/src/session.rs

@@ -1152,6 +1152,7 @@ impl Session {
             ));
             let handle = Arc::new(ManagedTorrent {
                 locked: RwLock::new(ManagedTorrentLocked {
+                    paused: opts.paused,
                     state: ManagedTorrentState::Initializing(initializing),
                     only_files,
                 }),
@@ -1327,6 +1328,7 @@ impl Session {
     }
 
     pub async fn pause(&self, handle: &ManagedTorrentHandle) -> anyhow::Result<()> {
+        handle.locked.write().paused = true;
         handle.pause()?;
         self.try_update_persistence_metadata(handle).await;
         Ok(())

crates/librqbit/src/session_persistence/json.rs

@@ -8,7 +8,6 @@ use crate::{
     storage::filesystem::FilesystemStorageFactory,
     torrent_state::ManagedTorrentHandle,
     type_aliases::BF,
-    ManagedTorrentState,
 };
 use anyhow::{bail, Context};
 use async_trait::async_trait;
@@ -86,6 +85,8 @@ impl JsonSessionPersistenceStore {
     }
 
     async fn flush(&self) -> anyhow::Result<()> {
+        // we don't need the write lock technically, but we need to stop concurrent modifications
+        let db_content = self.db_content.write().await;
         let tmp_filename = format!("{}.tmp", self.db_filename.to_str().unwrap());
         let mut tmp = tokio::fs::OpenOptions::new()
             .create(true)
@@ -97,8 +98,7 @@ impl JsonSessionPersistenceStore {
         trace!(?tmp_filename, "opened temp file");
 
         let mut buf = Vec::new();
-        serde_json::to_writer(&mut buf, &*self.db_content.read().await)
-            .context("error serializing")?;
+        serde_json::to_writer(&mut buf, &*db_content).context("error serializing")?;
 
         trace!(?tmp_filename, "serialized DB as JSON");
         tmp.write_all(&buf)
@@ -146,7 +146,7 @@ impl JsonSessionPersistenceStore {
             // we don't serialize this here, but to a file instead.
             torrent_bytes: Default::default(),
             only_files: torrent.only_files().clone(),
-            is_paused: torrent.with_state(|s| matches!(s, ManagedTorrentState::Paused(_))),
+            is_paused: torrent.is_paused(),
             output_folder: torrent.shared().options.output_folder.clone(),
         };
 

crates/librqbit/src/torrent_state/mod.rs

@@ -85,6 +85,11 @@ impl ManagedTorrentState {
 }
 
 pub(crate) struct ManagedTorrentLocked {
+    // The torrent might not be in "paused" state technically,
+    // but the intention might be for it to stay paused.
+    //
+    // This should change only on "unpause".
+    pub(crate) paused: bool,
     pub(crate) state: ManagedTorrentState,
     pub(crate) only_files: Option<Vec<usize>>,
 }
@@ -218,6 +223,7 @@ impl ManagedTorrent {
             .upgrade()
             .context("session is dead, cannot start torrent")?;
         let mut g = self.locked.write();
+        g.paused = start_paused;
         let cancellation_token = session.cancellation_token().child_token();
 
         let spawn_fatal_errors_receiver =
@@ -380,7 +386,7 @@ impl ManagedTorrent {
     }
 
     pub fn is_paused(&self) -> bool {
-        self.with_state(|s| matches!(s, ManagedTorrentState::Paused(..)))
+        self.locked.read().paused
     }
 
     /// Pause the torrent if it's live.