Merge pull request #332 from NordSecurity/msz/FILE-601-receiver-does-not-reports-events

matszczygiel · web-flow · commit 024e24d4bfb5 · 2024-05-22T12:19:39.000+02:00
FILE-601 Receiver does not reports events
diff --git a/changelog.md b/changelog.md
@@ -6,6 +6,7 @@ This means the whole API has changed. Even though semantics are the same
 the mechanism is now different and requires new implementation to use properly.
 * Split checksum events into finalize and verify
 * Add `base_dir` field in the `RequestQueued` event files
+* Fix rare issue of missing receiver's in-progress events
 
 ---
 <br>
diff --git a/drop-transfer/src/ws/server/handler.rs b/drop-transfer/src/ws/server/handler.rs
@@ -60,6 +60,7 @@ pub trait HandlerLoop {
     async fn on_bin_msg(&mut self, ws: &mut WebSocket, bytes: Vec<u8>) -> anyhow::Result<()>;
 
     async fn finalize_success(self);
+    async fn finalize_failure(self);
 }
 
 pub trait Request {
diff --git a/drop-transfer/src/ws/server/mod.rs b/drop-transfer/src/ws/server/mod.rs
@@ -564,6 +564,7 @@ impl RunContext<'_> {
                 "WS connection broke for {}: {err:?}",
                 xfer.id()
             );
+            handler.finalize_failure().await;
         } else {
             info!(self.logger, "Sucesfully finalizing transfer loop");
             handler.finalize_success().await;
diff --git a/drop-transfer/src/ws/server/v2.rs b/drop-transfer/src/ws/server/v2.rs
@@ -226,6 +226,18 @@ impl<const PING: bool> HandlerLoop<'_, PING> {
             self.stop_task(&file_id, Status::FileRejected).await;
         }
     }
+
+    fn take_pause_futures(&mut self) -> impl Future<Output = ()> {
+        let jobs = std::mem::take(&mut self.jobs);
+
+        async move {
+            let tasks = jobs.into_values().map(|task| async move {
+                task.events.pause().await;
+            });
+
+            futures::future::join_all(tasks).await;
+        }
+    }
 }
 
 #[async_trait::async_trait]
@@ -359,20 +371,18 @@ impl<const PING: bool> handler::HandlerLoop for HandlerLoop<'_, PING> {
     async fn finalize_success(mut self) {
         debug!(self.logger, "Finalizing");
     }
+
+    // While the destructor ensures the events are paused this function waits for
+    // the execution to be finished
+    async fn finalize_failure(mut self) {
+        self.take_pause_futures().await;
+    }
 }
 
 impl<const PING: bool> Drop for HandlerLoop<'_, PING> {
     fn drop(&mut self) {
         debug!(self.logger, "Stopping server handler");
-
-        let jobs = std::mem::take(&mut self.jobs);
-        tokio::spawn(async move {
-            let tasks = jobs.into_values().map(|task| async move {
-                task.events.pause().await;
-            });
-
-            futures::future::join_all(tasks).await;
-        });
+        tokio::spawn(self.take_pause_futures());
     }
 }
 
diff --git a/drop-transfer/src/ws/server/v4.rs b/drop-transfer/src/ws/server/v4.rs
@@ -325,6 +325,18 @@ impl HandlerLoop<'_> {
             }
         }
     }
+
+    fn take_pause_futures(&mut self) -> impl Future<Output = ()> {
+        let jobs = std::mem::take(&mut self.jobs);
+
+        async move {
+            let tasks = jobs.into_values().map(|task| async move {
+                task.events.pause().await;
+            });
+
+            futures::future::join_all(tasks).await;
+        }
+    }
 }
 
 #[async_trait::async_trait]
@@ -476,20 +488,18 @@ impl handler::HandlerLoop for HandlerLoop<'_> {
                 .map(|tmp| (tmp.base_path, FileId::from(tmp.file_id))),
         );
     }
+
+    // While the destructor ensures the events are paused this function waits for
+    // the execution to be finished
+    async fn finalize_failure(mut self) {
+        self.take_pause_futures().await;
+    }
 }
 
 impl Drop for HandlerLoop<'_> {
     fn drop(&mut self) {
         debug!(self.logger, "Stopping server handler");
-
-        let jobs = std::mem::take(&mut self.jobs);
-        tokio::spawn(async move {
-            let tasks = jobs.into_values().map(|task| async move {
-                task.events.pause().await;
-            });
-
-            futures::future::join_all(tasks).await;
-        });
+        tokio::spawn(self.take_pause_futures());
     }
 }
 
diff --git a/drop-transfer/src/ws/server/v6.rs b/drop-transfer/src/ws/server/v6.rs
@@ -370,6 +370,18 @@ impl HandlerLoop<'_> {
             }
         }
     }
+
+    fn take_pause_futures(&mut self) -> impl Future<Output = ()> {
+        let jobs = std::mem::take(&mut self.jobs);
+
+        async move {
+            let tasks = jobs.into_values().map(|task| async move {
+                task.events.pause().await;
+            });
+
+            futures::future::join_all(tasks).await;
+        }
+    }
 }
 
 #[async_trait::async_trait]
@@ -535,20 +547,18 @@ impl handler::HandlerLoop for HandlerLoop<'_> {
                 .map(|tmp| (tmp.base_path, FileId::from(tmp.file_id))),
         );
     }
+
+    // While the destructor ensures the events are paused this function waits for
+    // the execution to be finished
+    async fn finalize_failure(mut self) {
+        self.take_pause_futures().await;
+    }
 }
 
 impl Drop for HandlerLoop<'_> {
     fn drop(&mut self) {
         debug!(self.logger, "Stopping server handler");
-
-        let jobs = std::mem::take(&mut self.jobs);
-        tokio::spawn(async move {
-            let tasks = jobs.into_values().map(|task| async move {
-                task.events.pause().await;
-            });
-
-            futures::future::join_all(tasks).await;
-        });
+        tokio::spawn(self.take_pause_futures());
     }
 }
 
diff --git a/test/drop_test/action.py b/test/drop_test/action.py
@@ -682,6 +682,17 @@ def __str__(self):
         return f"DrainEvents({self._count})"
 
 
+class ClearEventQueue(Action):
+    def __init__(self):
+        pass
+
+    async def run(self, drop: ffi.Drop):
+        _ = await drop._events.gather_all(0)
+
+    def __str__(self):
+        return "ClearEventQueue()"
+
+
 class NoEvent(Action):
     def __init__(self, duration: int = 3):
         self._duration = duration
diff --git a/test/scenarios.py b/test/scenarios.py
@@ -2664,6 +2664,202 @@
             ),
         },
     ),
+    Scenario(
+        "scenario11-4",
+        "Initate a transfers with multiple files. Wait for them to start and then stop the sender. Then restart sender and expect transfer to finish successfuly",
+        {
+            "DROP_PEER_REN": ActionList(
+                [
+                    action.WaitForAnotherPeer("DROP_PEER_STIMPY"),
+                    action.Start("DROP_PEER_REN", dbpath="/tmp/db/11-4-ren.sqlite"),
+                    action.NewTransfer(
+                        "DROP_PEER_STIMPY",
+                        [
+                            "/tmp/testfile-bulk-01",
+                            "/tmp/testfile-bulk-02",
+                            "/tmp/testfile-bulk-03",
+                            "/tmp/testfile-bulk-04",
+                            "/tmp/testfile-bulk-05",
+                            "/tmp/testfile-bulk-06",
+                            "/tmp/testfile-bulk-07",
+                            "/tmp/testfile-bulk-08",
+                            "/tmp/testfile-bulk-09",
+                            "/tmp/testfile-bulk-10",
+                        ],
+                    ),
+                    # fmt: off
+                    action.Wait(
+                        event.Queued(0, "DROP_PEER_STIMPY", [
+                            norddrop.QueuedFile(FILES["testfile-bulk-01"].id, "testfile-bulk-01", 10485760, "/tmp"),
+                            norddrop.QueuedFile(FILES["testfile-bulk-02"].id, "testfile-bulk-02", 10485760, "/tmp"),
+                            norddrop.QueuedFile(FILES["testfile-bulk-03"].id, "testfile-bulk-03", 10485760, "/tmp"),
+                            norddrop.QueuedFile(FILES["testfile-bulk-04"].id, "testfile-bulk-04", 10485760, "/tmp"),
+                            norddrop.QueuedFile(FILES["testfile-bulk-05"].id, "testfile-bulk-05", 10485760, "/tmp"),
+                            norddrop.QueuedFile(FILES["testfile-bulk-06"].id, "testfile-bulk-06", 10485760, "/tmp"),
+                            norddrop.QueuedFile(FILES["testfile-bulk-07"].id, "testfile-bulk-07", 10485760, "/tmp"),
+                            norddrop.QueuedFile(FILES["testfile-bulk-08"].id, "testfile-bulk-08", 10485760, "/tmp"),
+                            norddrop.QueuedFile(FILES["testfile-bulk-09"].id, "testfile-bulk-09", 10485760, "/tmp"),
+                            norddrop.QueuedFile(FILES["testfile-bulk-10"].id, "testfile-bulk-10", 10485760, "/tmp"),
+                        ]),
+                    ),
+                    # Wait for some of the files and the stop
+                    action.Repeated([action.WaitForOneOf([
+                        event.Start(0, FILES["testfile-bulk-01"].id),
+                        event.Start(0, FILES["testfile-bulk-02"].id),
+                        event.Start(0, FILES["testfile-bulk-03"].id),
+                        event.Start(0, FILES["testfile-bulk-04"].id),
+                        event.Start(0, FILES["testfile-bulk-05"].id),
+                        event.Start(0, FILES["testfile-bulk-06"].id),
+                        event.Start(0, FILES["testfile-bulk-07"].id),
+                        event.Start(0, FILES["testfile-bulk-08"].id),
+                        event.Start(0, FILES["testfile-bulk-09"].id),
+                        event.Start(0, FILES["testfile-bulk-10"].id),
+                    ])], 4),
+                    # fmt: on
+                    action.Stop(),
+                    # fmt: off
+                    # Wait of paused events
+                    action.WaitAndIgnoreExcept([
+                        event.Paused(0, FILES["testfile-bulk-01"].id),
+                        event.Paused(0, FILES["testfile-bulk-02"].id),
+                        event.Paused(0, FILES["testfile-bulk-03"].id),
+                        event.Paused(0, FILES["testfile-bulk-04"].id),
+                        event.Paused(0, FILES["testfile-bulk-05"].id),
+                        event.Paused(0, FILES["testfile-bulk-06"].id),
+                        event.Paused(0, FILES["testfile-bulk-07"].id),
+                        event.Paused(0, FILES["testfile-bulk-08"].id),
+                        event.Paused(0, FILES["testfile-bulk-09"].id),
+                        event.Paused(0, FILES["testfile-bulk-10"].id),
+                    ]),
+                    # fmt: on
+                    action.Sleep(2),
+                    action.ClearEventQueue(),
+                    action.Start("DROP_PEER_REN", dbpath="/tmp/db/11-4-ren.sqlite"),
+                    # fmt: off
+                    # Wait for some of the files and the stop
+                    action.WaitRacy([
+                        event.Start(0, FILES["testfile-bulk-01"].id, None),
+                        event.Start(0, FILES["testfile-bulk-02"].id, None),
+                        event.Start(0, FILES["testfile-bulk-03"].id, None),
+                        event.Start(0, FILES["testfile-bulk-04"].id, None),
+                        event.Start(0, FILES["testfile-bulk-05"].id, None),
+                        event.Start(0, FILES["testfile-bulk-06"].id, None),
+                        event.Start(0, FILES["testfile-bulk-07"].id, None),
+                        event.Start(0, FILES["testfile-bulk-08"].id, None),
+                        event.Start(0, FILES["testfile-bulk-09"].id, None),
+                        event.Start(0, FILES["testfile-bulk-10"].id, None),
+                        event.FinishFileUploaded(0, FILES["testfile-bulk-01"].id),
+                        event.FinishFileUploaded(0, FILES["testfile-bulk-02"].id),
+                        event.FinishFileUploaded(0, FILES["testfile-bulk-03"].id),
+                        event.FinishFileUploaded(0, FILES["testfile-bulk-04"].id),
+                        event.FinishFileUploaded(0, FILES["testfile-bulk-05"].id),
+                        event.FinishFileUploaded(0, FILES["testfile-bulk-06"].id),
+                        event.FinishFileUploaded(0, FILES["testfile-bulk-07"].id),
+                        event.FinishFileUploaded(0, FILES["testfile-bulk-08"].id),
+                        event.FinishFileUploaded(0, FILES["testfile-bulk-09"].id),
+                        event.FinishFileUploaded(0, FILES["testfile-bulk-10"].id),
+                    ]),
+                    # fmt: on
+                    action.ExpectCancel([0], True),
+                ]
+            ),
+            "DROP_PEER_STIMPY": ActionList(
+                [
+                    action.Start("DROP_PEER_STIMPY"),
+                    # fmt: off
+                    action.Wait(
+                        event.Receive(0, "DROP_PEER_REN", [ 
+                            norddrop.ReceivedFile(FILES["testfile-bulk-01"].id, "testfile-bulk-01", 10485760),
+                            norddrop.ReceivedFile(FILES["testfile-bulk-02"].id, "testfile-bulk-02", 10485760),
+                            norddrop.ReceivedFile(FILES["testfile-bulk-03"].id, "testfile-bulk-03", 10485760),
+                            norddrop.ReceivedFile(FILES["testfile-bulk-04"].id, "testfile-bulk-04", 10485760),
+                            norddrop.ReceivedFile(FILES["testfile-bulk-05"].id, "testfile-bulk-05", 10485760),
+                            norddrop.ReceivedFile(FILES["testfile-bulk-06"].id, "testfile-bulk-06", 10485760),
+                            norddrop.ReceivedFile(FILES["testfile-bulk-07"].id, "testfile-bulk-07", 10485760),
+                            norddrop.ReceivedFile(FILES["testfile-bulk-08"].id, "testfile-bulk-08", 10485760),
+                            norddrop.ReceivedFile(FILES["testfile-bulk-09"].id, "testfile-bulk-09", 10485760),
+                            norddrop.ReceivedFile(FILES["testfile-bulk-10"].id, "testfile-bulk-10", 10485760),
+                        ]),
+                    ),
+                    # fmt: on
+                    # fmt: off
+                    action.Download(0, FILES["testfile-bulk-01"].id, "/tmp/received/11-4"),
+                    action.Download(0, FILES["testfile-bulk-02"].id, "/tmp/received/11-4"),
+                    action.Download(0, FILES["testfile-bulk-03"].id, "/tmp/received/11-4"),
+                    action.Download(0, FILES["testfile-bulk-04"].id, "/tmp/received/11-4"),
+                    action.Download(0, FILES["testfile-bulk-05"].id, "/tmp/received/11-4"),
+                    action.Download(0, FILES["testfile-bulk-06"].id, "/tmp/received/11-4"),
+                    action.Download(0, FILES["testfile-bulk-07"].id, "/tmp/received/11-4"),
+                    action.Download(0, FILES["testfile-bulk-08"].id, "/tmp/received/11-4"),
+                    action.Download(0, FILES["testfile-bulk-09"].id, "/tmp/received/11-4"),
+                    action.Download(0, FILES["testfile-bulk-10"].id, "/tmp/received/11-4"),
+                    # Wait for all 10 pending events
+                    action.Repeated([action.WaitForOneOf([
+                        event.Pending(0, FILES["testfile-bulk-01"].id),
+                        event.Pending(0, FILES["testfile-bulk-02"].id),
+                        event.Pending(0, FILES["testfile-bulk-03"].id),
+                        event.Pending(0, FILES["testfile-bulk-04"].id),
+                        event.Pending(0, FILES["testfile-bulk-05"].id),
+                        event.Pending(0, FILES["testfile-bulk-06"].id),
+                        event.Pending(0, FILES["testfile-bulk-07"].id),
+                        event.Pending(0, FILES["testfile-bulk-08"].id),
+                        event.Pending(0, FILES["testfile-bulk-09"].id),
+                        event.Pending(0, FILES["testfile-bulk-10"].id),
+                        event.Start(0, FILES["testfile-bulk-01"].id),
+                        event.Start(0, FILES["testfile-bulk-02"].id),
+                        event.Start(0, FILES["testfile-bulk-03"].id),
+                        event.Start(0, FILES["testfile-bulk-04"].id),
+                        event.Start(0, FILES["testfile-bulk-05"].id),
+                        event.Start(0, FILES["testfile-bulk-06"].id),
+                        event.Start(0, FILES["testfile-bulk-07"].id),
+                        event.Start(0, FILES["testfile-bulk-08"].id),
+                        event.Start(0, FILES["testfile-bulk-09"].id),
+                        event.Start(0, FILES["testfile-bulk-10"].id),
+                        ]),
+                    ], 10),
+                    action.WaitAndIgnoreExcept([
+                        event.Paused(0, FILES["testfile-bulk-01"].id),
+                        event.Paused(0, FILES["testfile-bulk-02"].id),
+                        event.Paused(0, FILES["testfile-bulk-03"].id),
+                        event.Paused(0, FILES["testfile-bulk-04"].id),
+                        event.Paused(0, FILES["testfile-bulk-05"].id),
+                        event.Paused(0, FILES["testfile-bulk-06"].id),
+                        event.Paused(0, FILES["testfile-bulk-07"].id),
+                        event.Paused(0, FILES["testfile-bulk-08"].id),
+                        event.Paused(0, FILES["testfile-bulk-09"].id),
+                        event.Paused(0, FILES["testfile-bulk-10"].id),
+                    ]),
+                    # The sender is stopped
+                    action.WaitRacy([
+                        event.Start(0, FILES["testfile-bulk-01"].id, None),
+                        event.Start(0, FILES["testfile-bulk-02"].id, None),
+                        event.Start(0, FILES["testfile-bulk-03"].id, None),
+                        event.Start(0, FILES["testfile-bulk-04"].id, None),
+                        event.Start(0, FILES["testfile-bulk-05"].id, None),
+                        event.Start(0, FILES["testfile-bulk-06"].id, None),
+                        event.Start(0, FILES["testfile-bulk-07"].id, None),
+                        event.Start(0, FILES["testfile-bulk-08"].id, None),
+                        event.Start(0, FILES["testfile-bulk-09"].id, None),
+                        event.Start(0, FILES["testfile-bulk-10"].id, None),
+                        event.FinishFileDownloaded(0, FILES["testfile-bulk-01"].id, "/tmp/received/11-4/testfile-bulk-01"),
+                        event.FinishFileDownloaded(0, FILES["testfile-bulk-02"].id, "/tmp/received/11-4/testfile-bulk-02"),
+                        event.FinishFileDownloaded(0, FILES["testfile-bulk-03"].id, "/tmp/received/11-4/testfile-bulk-03"),
+                        event.FinishFileDownloaded(0, FILES["testfile-bulk-04"].id, "/tmp/received/11-4/testfile-bulk-04"),
+                        event.FinishFileDownloaded(0, FILES["testfile-bulk-05"].id, "/tmp/received/11-4/testfile-bulk-05"),
+                        event.FinishFileDownloaded(0, FILES["testfile-bulk-06"].id, "/tmp/received/11-4/testfile-bulk-06"),
+                        event.FinishFileDownloaded(0, FILES["testfile-bulk-07"].id, "/tmp/received/11-4/testfile-bulk-07"),
+                        event.FinishFileDownloaded(0, FILES["testfile-bulk-08"].id, "/tmp/received/11-4/testfile-bulk-08"),
+                        event.FinishFileDownloaded(0, FILES["testfile-bulk-09"].id, "/tmp/received/11-4/testfile-bulk-09"),
+                        event.FinishFileDownloaded(0, FILES["testfile-bulk-10"].id, "/tmp/received/11-4/testfile-bulk-10"),
+                    ]),
+                    # fmt: on
+                    action.CancelTransferRequest([0]),
+                    action.ExpectCancel([0], False),
+                    action.Stop(),
+                ]
+            ),
+        },
+    ),
     Scenario(
         "scenario12-1",
         "Transfer file to two clients simultaneously",

Original file line number	Diff line number	Diff line change
`@@ -60,6 +60,7 @@ pub trait HandlerLoop {`
`60`	`60`	`async fn on_bin_msg(&mut self, ws: &mut WebSocket, bytes: Vec<u8>) -> anyhow::Result<()>;`
`61`	`61`
`62`	`62`	`async fn finalize_success(self);`
	`63`	`+ async fn finalize_failure(self);`
`63`	`64`	`}`
`64`	`65`
`65`	`66`	`pub trait Request {`
Original file line number	Diff line number	Diff line change
`@@ -325,6 +325,18 @@ impl HandlerLoop<'_> {`
`325`	`325`	`}`
`326`	`326`	`}`
`327`	`327`	`}`
	`328`	`+`
	`329`	`+ fn take_pause_futures(&mut self) -> impl Future<Output = ()> {`
	`330`	`+ let jobs = std::mem::take(&mut self.jobs);`
	`331`	`+`
	`332`	`+ async move {`
	`333`	`+ let tasks = jobs.into_values().map(\|task\| async move {`
	`334`	`+ task.events.pause().await;`
	`335`	`+ });`
	`336`	`+`
	`337`	`+ futures::future::join_all(tasks).await;`
	`338`	`+ }`
	`339`	`+ }`
`328`	`340`	`}`
`329`	`341`
`330`	`342`	`#[async_trait::async_trait]`
`@@ -476,20 +488,18 @@ impl handler::HandlerLoop for HandlerLoop<'_> {`
`476`	`488`	`.map(\|tmp\| (tmp.base_path, FileId::from(tmp.file_id))),`
`477`	`489`	`);`
`478`	`490`	`}`
	`491`	`+`
	`492`	`+ // While the destructor ensures the events are paused this function waits for`
	`493`	`+ // the execution to be finished`
	`494`	`+ async fn finalize_failure(mut self) {`
	`495`	`+ self.take_pause_futures().await;`
	`496`	`+ }`
`479`	`497`	`}`
`480`	`498`
`481`	`499`	`impl Drop for HandlerLoop<'_> {`
`482`	`500`	`fn drop(&mut self) {`
`483`	`501`	`debug!(self.logger, "Stopping server handler");`
`484`		`-`
`485`		`- let jobs = std::mem::take(&mut self.jobs);`
`486`		`- tokio::spawn(async move {`
`487`		`- let tasks = jobs.into_values().map(\|task\| async move {`
`488`		`- task.events.pause().await;`
`489`		`- });`
`490`		`-`
`491`		`- futures::future::join_all(tasks).await;`
`492`		`- });`
	`502`	`+ tokio::spawn(self.take_pause_futures());`
`493`	`503`	`}`
`494`	`504`	`}`
`495`	`505`
Original file line number	Diff line number	Diff line change
`@@ -370,6 +370,18 @@ impl HandlerLoop<'_> {`
`370`	`370`	`}`
`371`	`371`	`}`
`372`	`372`	`}`
	`373`	`+`
	`374`	`+ fn take_pause_futures(&mut self) -> impl Future<Output = ()> {`
	`375`	`+ let jobs = std::mem::take(&mut self.jobs);`
	`376`	`+`
	`377`	`+ async move {`
	`378`	`+ let tasks = jobs.into_values().map(\|task\| async move {`
	`379`	`+ task.events.pause().await;`
	`380`	`+ });`
	`381`	`+`
	`382`	`+ futures::future::join_all(tasks).await;`
	`383`	`+ }`
	`384`	`+ }`
`373`	`385`	`}`
`374`	`386`
`375`	`387`	`#[async_trait::async_trait]`
`@@ -535,20 +547,18 @@ impl handler::HandlerLoop for HandlerLoop<'_> {`
`535`	`547`	`.map(\|tmp\| (tmp.base_path, FileId::from(tmp.file_id))),`
`536`	`548`	`);`
`537`	`549`	`}`
	`550`	`+`
	`551`	`+ // While the destructor ensures the events are paused this function waits for`
	`552`	`+ // the execution to be finished`
	`553`	`+ async fn finalize_failure(mut self) {`
	`554`	`+ self.take_pause_futures().await;`
	`555`	`+ }`
`538`	`556`	`}`
`539`	`557`
`540`	`558`	`impl Drop for HandlerLoop<'_> {`
`541`	`559`	`fn drop(&mut self) {`
`542`	`560`	`debug!(self.logger, "Stopping server handler");`
`543`		`-`
`544`		`- let jobs = std::mem::take(&mut self.jobs);`
`545`		`- tokio::spawn(async move {`
`546`		`- let tasks = jobs.into_values().map(\|task\| async move {`
`547`		`- task.events.pause().await;`
`548`		`- });`
`549`		`-`
`550`		`- futures::future::join_all(tasks).await;`
`551`		`- });`
	`561`	`+ tokio::spawn(self.take_pause_futures());`
`552`	`562`	`}`
`553`	`563`	`}`
`554`	`564`