Exclude closed session from get session ids method (payjoin#806)

Failing to exclude closed sessions causes payjoin-cli to resume a sender
or receiver sessions that has already reached their final state. This
update also extends the v2 e2e test to verify that no sessions are
resumed after completion.

Author: spacebear21

payjoin-cli/src/db/v2.rs

@@ -175,7 +175,12 @@ impl Database {
         let recv_tree = self.0.open_tree("recv_sessions")?;
         let mut session_ids = Vec::new();
         for item in recv_tree.iter() {
-            let (key, _) = item?;
+            let (key, value) = item?;
+            let wrapper: SessionWrapper<ReceiverSessionEvent> =
+                serde_json::from_slice(&value).map_err(Error::Deserialize)?;
+            if wrapper.completed_at.is_some() {
+                continue;
+            }
             session_ids.push(SessionId::new(u64::from_be_bytes(
                 key.as_ref().try_into().map_err(Error::TryFromSlice)?,
             )));
@@ -187,8 +192,15 @@ impl Database {
         let send_tree = self.0.open_tree("send_sessions")?;
         let mut session_ids = Vec::new();
         for item in send_tree.iter() {
-            let (key, _) = item?;
-            session_ids.push(SessionId::new(u64::from_be_bytes(key.as_ref().try_into().unwrap())));
+            let (key, value) = item?;
+            let wrapper: SessionWrapper<SenderSessionEvent> =
+                serde_json::from_slice(&value).map_err(Error::Deserialize)?;
+            if wrapper.completed_at.is_some() {
+                continue;
+            }
+            session_ids.push(SessionId::new(u64::from_be_bytes(
+                key.as_ref().try_into().map_err(Error::TryFromSlice)?,
+            )));
         }
         Ok(session_ids)
     }

payjoin-cli/tests/e2e.rs

@@ -276,6 +276,38 @@ mod e2e {
                 .spawn()
                 .expect("Failed to execute payjoin-cli");
             check_payjoin_sent(cli_send_resumer).await?;
+
+            // Check that neither the sender or the receiver have sessions to resume
+            let cli_receive_resumer = Command::new(payjoin_cli)
+                .arg("--rpchost")
+                .arg(&receiver_rpchost)
+                .arg("--cookie-file")
+                .arg(cookie_file)
+                .arg("--db-path")
+                .arg(&receiver_db_path)
+                .arg("--ohttp-relays")
+                .arg(ohttp_relay)
+                .arg("resume")
+                .stdout(Stdio::piped())
+                .stderr(Stdio::inherit())
+                .spawn()
+                .expect("Failed to execute payjoin-cli");
+            check_resume_has_no_sessions(cli_receive_resumer).await?;
+            let cli_send_resumer = Command::new(payjoin_cli)
+                .arg("--rpchost")
+                .arg(&sender_rpchost)
+                .arg("--cookie-file")
+                .arg(cookie_file)
+                .arg("--db-path")
+                .arg(&sender_db_path)
+                .arg("--ohttp-relays")
+                .arg(ohttp_relay)
+                .arg("resume")
+                .stdout(Stdio::piped())
+                .stderr(Stdio::inherit())
+                .spawn()
+                .expect("Failed to execute payjoin-cli");
+            check_resume_has_no_sessions(cli_send_resumer).await?;
             Ok(())
         }
 
@@ -338,6 +370,21 @@ mod e2e {
             Ok(())
         }
 
+        async fn check_resume_has_no_sessions(mut cli_resumer: Child) -> Result<()> {
+            let mut stdout =
+                cli_resumer.stdout.take().expect("Failed to take stdout of child process");
+            let timeout = tokio::time::Duration::from_secs(10);
+            let res = tokio::time::timeout(
+                timeout,
+                wait_for_stdout_match(&mut stdout, |line| line.contains("No sessions to resume.")),
+            )
+            .await?;
+
+            terminate(cli_resumer).await.expect("Failed to kill payjoin-cli");
+            assert!(res.is_some(), "Expected no sessions to resume");
+            Ok(())
+        }
+
         /// Read lines from `child_stdout` until `match_pattern` is found and the corresponding
         /// line is returned.
         /// Also writes every read line to tokio::io::stdout();