Unify mailbox TTL to a single value

Replace three separate TTL constants (read: 10 min,
unread at capacity: 1 day, unread below capacity: 7 days)
with a single DEFAULT_TTL of 24 hours based on creation
time only. Removes read_order, read_mailbox_ids,
mark_read(), and the read-TTL pruning pass. Capacity
eviction remains as a storage pressure mechanism.

Author: Arowolokehinde

payjoin-mailroom/src/db/files.rs

@@ -1,4 +1,4 @@
-use std::collections::{HashMap, HashSet, VecDeque};
+use std::collections::{HashMap, VecDeque};
 use std::path::PathBuf;
 use std::str::FromStr;
 use std::sync::Arc;
@@ -21,13 +21,11 @@ use crate::db::{Db as DbTrait, Error as DbError};
 /// mailboxes/tx, ~4K txs/block, and ~144 blocks/24h.
 const DEFAULT_CAPACITY: usize = 1 << (1 + 12 + 8);
 
-const DEFAULT_UNREAD_TTL_AT_CAPACITY: Duration = Duration::from_secs(60 * 60 * 24); // 1 day
-const DEFAULT_UNREAD_TTL_BELOW_CAPACITY: Duration = Duration::from_secs(60 * 60 * 24 * 7); // 1 week
-
-/// How long read messages should be kept in mailboxes. Defaults to a 10 minute
-/// grace period from first read attempt, in case of intermittent network or
-/// relay errors.
-const DEFAULT_READ_TTL: Duration = Duration::from_secs(60 * 10); // 10 minutes
+/// How long a mailbox is retained before being pruned, regardless of
+/// whether it has been read.
+/// Matches the default receiver session lifetime
+/// (`TWENTY_FOUR_HOURS_DEFAULT_EXPIRATION` in `payjoin::receive::v2`)
+const DEFAULT_TTL: Duration = Duration::from_secs(60 * 60 * 24); // 24 hours
 
 #[derive(Debug)]
 struct V2WaitMapEntry {
@@ -50,11 +48,7 @@ pub(crate) struct Mailboxes {
     pending_v1: HashMap<ShortId, V1WaitMapEntry>,
     pending_v2: HashMap<ShortId, V2WaitMapEntry>,
     insert_order: VecDeque<(SystemTime, ShortId)>,
-    read_order: VecDeque<(SystemTime, ShortId)>,
-    read_mailbox_ids: HashSet<ShortId>,
-    unread_ttl_below_capacity: Duration,
-    unread_ttl_at_capacity: Duration,
-    read_ttl: Duration,
+    ttl: Duration,
     early_removal_count: usize,
 }
 
@@ -209,11 +203,7 @@ impl Mailboxes {
             capacity: DEFAULT_CAPACITY,
             pending_v1: HashMap::default(),
             pending_v2: HashMap::default(),
-            read_order: VecDeque::default(),
-            read_mailbox_ids: HashSet::default(),
-            unread_ttl_below_capacity: DEFAULT_UNREAD_TTL_BELOW_CAPACITY,
-            unread_ttl_at_capacity: DEFAULT_UNREAD_TTL_AT_CAPACITY,
-            read_ttl: DEFAULT_READ_TTL,
+            ttl: DEFAULT_TTL,
             early_removal_count: 0,
         })
     }
@@ -337,19 +327,12 @@ impl Mailboxes {
 
         // V2 requests are stored on disk
         if let Some((_created, payload)) = self.persistent_storage.get(id).await? {
-            self.mark_read(id);
             return Ok(Some(Arc::new(payload)));
         }
 
         Ok(None)
     }
 
-    fn mark_read(&mut self, id: &ShortId) {
-        if self.read_mailbox_ids.insert(*id) {
-            self.read_order.push_back((SystemTime::now(), *id));
-        }
-    }
-
     async fn has_capacity(&mut self) -> io::Result<bool> {
         self.maybe_prune().await?;
         Ok(self.len() < self.capacity)
@@ -399,12 +382,10 @@ impl Mailboxes {
 
         self.insert_order.push_back((created, *id));
 
-        // If there are pending readers, satisfy them and mark the payload as read
+        // If there are pending readers, satisfy them
         if let Some(pending) = self.pending_v2.remove(id) {
             trace!("notifying pending readers for {}", id);
 
-            self.mark_read(id);
-
             pending
                 .sender
                 .send(Arc::new(payload))
@@ -444,7 +425,6 @@ impl Mailboxes {
     }
 
     async fn remove(&mut self, id: &ShortId) -> io::Result<Option<()>> {
-        self.read_mailbox_ids.remove(id);
         self.persistent_storage.remove(id).await
     }
 
@@ -479,16 +459,14 @@ impl Mailboxes {
         trace!("pruning");
         let now = SystemTime::now();
 
-        debug_assert!(self.read_ttl < self.unread_ttl_at_capacity);
-        debug_assert!(self.unread_ttl_at_capacity < self.unread_ttl_below_capacity);
         debug_assert!(self.pending_v1.iter().all(|(_, v)| !v.sender.is_closed()));
 
         // Prune in flight requests, these can persist in the case of an incomplete session
         self.pending_v2.retain(|_, v| v.receiver.strong_count().unwrap_or(0) > 1);
 
-        // Prune any fully expired mailboxes, whether read or unread
+        // Prune any fully expired mailboxes
         while let Some((created, id)) = self.insert_order.front().cloned() {
-            if created + self.unread_ttl_below_capacity < now {
+            if created + self.ttl < now {
                 debug_assert!(self.insert_order.len() >= self.early_removal_count);
                 _ = self.insert_order.pop_front();
                 if self.remove(&id).await?.is_none() {
@@ -504,30 +482,15 @@ impl Mailboxes {
             }
         }
 
-        // So long as there expired read mailboxes, prune those. Stop when a
-        // mailbox within the TTL is encountered.
-        while let Some((read, id)) = self.read_order.front().cloned() {
-            if read + self.read_ttl < now {
-                _ = self.read_order.pop_front();
-                if self.remove(&id).await?.is_some() {
-                    self.early_removal_count += 1;
-                    debug_assert!(self.insert_order.len() >= self.early_removal_count);
-                }
-                trace!("Pruned read mailbox {id}");
-            } else {
-                break;
-            }
-        }
-
-        // If no room was created, try to prune the oldest unread mailbox if
-        // it's over the minimum TTL
+        // If no room was created, try to prune the oldest mailbox if
+        // it's over the TTL
         debug_assert!(self.len() <= self.capacity);
         if self.len() == self.capacity {
             if let Some((created, id)) = self.insert_order.front().cloned() {
-                if created + self.unread_ttl_at_capacity < now {
+                if created + self.ttl < now {
                     _ = self.insert_order.pop_front();
                     self.remove(&id).await?;
-                    trace!("Pruned unread mailbox {id} to make room");
+                    trace!("Pruned mailbox {id} to make room");
                 } else {
                     trace!("Nothing to prune, {} entries remain", self.len());
                 }
@@ -538,27 +501,14 @@ impl Mailboxes {
     }
 
     fn next_prune(&mut self) -> Duration {
-        let earliest_read_prune_opportunity = self
-            .read_order
-            .front()
-            .map(|(read, _id)| {
-                self.read_ttl
-                    .checked_sub(read.elapsed().expect("system clock moved back"))
-                    .unwrap_or(self.read_ttl)
-            })
-            .unwrap_or_else(|| self.read_ttl);
-
-        let earliest_unread_prune_opportunity = self
-            .insert_order
+        self.insert_order
             .front()
             .map(|(created, _id)| {
-                self.unread_ttl_at_capacity
+                self.ttl
                     .checked_sub(created.elapsed().expect("system clock moved back"))
-                    .unwrap_or(self.unread_ttl_at_capacity)
+                    .unwrap_or(self.ttl)
             })
-            .unwrap_or_else(|| self.unread_ttl_at_capacity);
-
-        std::cmp::min(earliest_read_prune_opportunity, earliest_unread_prune_opportunity)
+            .unwrap_or(self.ttl)
     }
 }
 
@@ -938,16 +888,12 @@ async fn test_prune() -> std::io::Result<()> {
         .await
         .expect("initializing mailbox database should succeed");
 
-    let read_ttl = Duration::from_secs(60);
-    let unread_ttl_at_capacity = Duration::from_secs(600);
-    let unread_ttl_below_capacity = Duration::from_secs(3600);
+    let ttl = Duration::from_secs(600);
 
     {
         let mut guard = db.mailboxes.lock().await;
         guard.capacity = 2;
-        guard.read_ttl = read_ttl;
-        guard.unread_ttl_at_capacity = unread_ttl_at_capacity;
-        guard.unread_ttl_below_capacity = unread_ttl_below_capacity;
+        guard.ttl = ttl;
     }
 
     assert_eq!(db.mailboxes.lock().await.len(), 0);
@@ -984,38 +930,11 @@ async fn test_prune() -> std::io::Result<()> {
     db.prune().await.expect("pruning should not fail");
     assert_eq!(db.mailboxes.lock().await.len(), 1);
 
-    // Shift insert timestamps past unread_ttl_below_capacity
+    // Shift insert timestamps past ttl
     {
         let mut guard = db.mailboxes.lock().await;
         for (ts, _) in guard.insert_order.iter_mut() {
-            *ts -= unread_ttl_below_capacity + Duration::from_secs(1);
-        }
-    }
-
-    assert_eq!(db.mailboxes.lock().await.len(), 1);
-    db.prune().await.expect("pruning should not fail");
-    assert_eq!(db.mailboxes.lock().await.len(), 0);
-
-    // Post again, read it, then verify read TTL pruning
-    db.post_v2_payload(&id, contents.to_vec())
-        .await
-        .expect("posting payload should succeed")
-        .expect("contents should be accepted");
-
-    assert_eq!(db.mailboxes.lock().await.len(), 1);
-
-    // Mark the mailbox as read
-    _ = db.wait_for_v2_payload(&id).await.expect("waiting for payload should succeed");
-
-    assert_eq!(db.mailboxes.lock().await.len(), 1);
-    db.prune().await.expect("pruning should not fail");
-    assert_eq!(db.mailboxes.lock().await.len(), 1);
-
-    // Shift read timestamps past read_ttl
-    {
-        let mut guard = db.mailboxes.lock().await;
-        for (ts, _) in guard.read_order.iter_mut() {
-            *ts -= read_ttl + Duration::from_secs(1);
+            *ts -= ttl + Duration::from_secs(1);
         }
     }