Skip to content

Unify mailbox TTL to a single value #1457

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Arowolokehinde wants to merge 1 commit into
base: master
Choose a base branch
from
+74 −150
Open

Unify mailbox TTL to a single value #1457

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions payjoin-mailroom/src/config.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,10 @@ pub struct Config {
pub storage_dir: PathBuf,
#[serde(deserialize_with = "deserialize_duration_secs")]
pub timeout: Duration,
/// Mailbox lifetime in seconds. Together with `DEFAULT_CAPACITY` in
/// `db::files`, this bounds both disk usage and write throughput.
Copy link
Copy Markdown
Contributor

@nothingmuch nothingmuch Apr 7, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

this could bound write throughput but doesn't do so currently except in the trivial sense that once storage is full no new writes are accepted, see tracking tasks in #941, this could be done with tower middleware

#[serde(deserialize_with = "deserialize_duration_secs")]
pub mailbox_ttl: Duration,
pub v1: Option<V1Config>,
#[cfg(feature = "telemetry")]
pub telemetry: Option<TelemetryConfig>,
Expand Down Expand Up @@ -85,6 +89,7 @@ impl Default for Config {
listener: "[::]:8080".parse().expect("valid default listener address"),
storage_dir: PathBuf::from("./data"),
timeout: Duration::from_secs(30),
mailbox_ttl: Duration::from_secs(60 * 60 * 24 * 7), // 1 week
v1: None,
#[cfg(feature = "telemetry")]
telemetry: None,
Expand Down Expand Up @@ -115,6 +120,7 @@ impl Config {
listener,
storage_dir,
timeout,
mailbox_ttl: Duration::from_secs(60 * 60 * 24 * 7), // 1 week
v1,
#[cfg(feature = "telemetry")]
telemetry: None,
Expand Down
204 changes: 57 additions & 147 deletions payjoin-mailroom/src/db/files.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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;
Expand All @@ -11,7 +11,7 @@ use rand::RngCore;
use tokio::fs::{self, File};
use tokio::io::{self, AsyncReadExt, AsyncWriteExt};
use tokio::sync::{oneshot, Mutex};
use tracing::trace;
use tracing::{trace, warn};

use crate::db::{Db as DbTrait, Error as DbError};

Expand All @@ -21,14 +21,6 @@ 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

#[derive(Debug)]
struct V2WaitMapEntry {
receiver: future::Shared<oneshot::Receiver<Arc<Vec<u8>>>>,
Expand All @@ -50,12 +42,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,
early_removal_count: usize,
ttl: Duration,
}

#[derive(Debug)]
Expand Down Expand Up @@ -200,7 +187,7 @@ impl DiskStorage {
}

impl Mailboxes {
async fn init(dir: PathBuf) -> io::Result<Self> {
async fn init(dir: PathBuf, ttl: Duration) -> io::Result<Self> {
let storage = DiskStorage::init(dir).await?;
let insert_order = storage.insert_order().await?.into();
Ok(Self {
Expand All @@ -209,12 +196,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,
early_removal_count: 0,
ttl,
})
}
}
Expand All @@ -226,8 +208,8 @@ pub struct FilesDb {
}

impl FilesDb {
pub async fn init(timeout: Duration, path: PathBuf) -> io::Result<Self> {
Ok(Self { timeout, mailboxes: Arc::new(Mutex::new(Mailboxes::init(path).await?)) })
pub async fn init(timeout: Duration, path: PathBuf, ttl: Duration) -> io::Result<Self> {
Ok(Self { timeout, mailboxes: Arc::new(Mutex::new(Mailboxes::init(path, ttl).await?)) })
}

pub async fn prune(&self) -> io::Result<Duration> { self.mailboxes.lock().await.prune().await }
Expand Down Expand Up @@ -337,19 +319,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)
Expand Down Expand Up @@ -399,12 +374,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))
Expand Down Expand Up @@ -444,7 +417,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
}

Expand All @@ -457,9 +429,7 @@ impl Mailboxes {
}

fn len(&self) -> usize {
(self.insert_order.len() - self.early_removal_count)
+ self.pending_v1.len()
+ self.pending_v2.len()
self.insert_order.len() + self.pending_v1.len() + self.pending_v2.len()
}

async fn maybe_prune(&mut self) -> io::Result<Duration> {
Expand All @@ -479,86 +449,37 @@ 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
Copy link
Copy Markdown
Contributor

@nothingmuch nothingmuch Apr 7, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit:

Suggested change
// Prune any fully expired mailboxes
// Prune any expired mailboxes

while let Some((created, id)) = self.insert_order.front().cloned() {
if created + self.unread_ttl_below_capacity < now {
debug_assert!(self.insert_order.len() >= self.early_removal_count);
if created + self.ttl < now {
_ = self.insert_order.pop_front();
if self.remove(&id).await?.is_none() {
self.early_removal_count = self
.early_removal_count
.checked_sub(1)
.expect("early removal adjustment should never underflow");
}
debug_assert!(self.insert_order.len() >= self.early_removal_count);
trace!("Pruned old mailbox {id}");
} else {
break;
}
}

// 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);
warn!("Mailbox file missing during prune; possible external deletion or disk error");
} else {
trace!("Pruned old mailbox {id}");
}
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
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 {
_ = self.insert_order.pop_front();
self.remove(&id).await?;
trace!("Pruned unread mailbox {id} to make room");
} else {
trace!("Nothing to prune, {} entries remain", self.len());
}
}
}

Ok(self.next_prune())
}

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)
}
}

Expand Down Expand Up @@ -754,9 +675,13 @@ async fn test_disk_storage_mailboxes() -> std::io::Result<()> {
async fn test_mailbox_storage() -> std::io::Result<()> {
let dir = tempfile::tempdir()?;

let db = FilesDb::init(Duration::from_millis(10), dir.path().to_owned())
.await
.expect("initializing mailbox database should succeed");
let db = FilesDb::init(
Duration::from_millis(10),
dir.path().to_owned(),
Duration::from_secs(60 * 60 * 24 * 7),
)
.await
.expect("initializing mailbox database should succeed");

let id = ShortId([0u8; 8]);
let contents = b"foo bar";
Expand All @@ -775,9 +700,13 @@ async fn test_mailbox_storage() -> std::io::Result<()> {
async fn test_v2_wait() -> std::io::Result<()> {
let dir = tempfile::tempdir()?;

let db = FilesDb::init(Duration::from_millis(1), dir.path().to_owned())
.await
.expect("initializing mailbox database should succeed");
let db = FilesDb::init(
Duration::from_millis(1),
dir.path().to_owned(),
Duration::from_secs(60 * 60 * 24 * 7),
)
.await
.expect("initializing mailbox database should succeed");

let id = ShortId([0u8; 8]);
let contents = b"foo bar";
Expand Down Expand Up @@ -832,9 +761,13 @@ async fn test_v1_wait() -> std::io::Result<()> {
let dir = tempfile::tempdir()?;

let db = Arc::new(
FilesDb::init(Duration::from_millis(1), dir.path().to_owned())
.await
.expect("initializing mailbox database should succeed"),
FilesDb::init(
Duration::from_millis(1),
dir.path().to_owned(),
Duration::from_secs(60 * 60 * 24 * 7),
)
.await
.expect("initializing mailbox database should succeed"),
);

let id = ShortId([0u8; 8]);
Expand Down Expand Up @@ -879,9 +812,13 @@ async fn test_v1_data_minimization() -> std::io::Result<()> {
let dir = tempfile::tempdir()?;

let db = Arc::new(
FilesDb::init(Duration::from_millis(500), dir.path().to_owned())
.await
.expect("initializing mailbox database should succeed"),
FilesDb::init(
Duration::from_millis(500),
dir.path().to_owned(),
Duration::from_secs(60 * 60 * 24 * 7),
)
.await
.expect("initializing mailbox database should succeed"),
);

let id = ShortId([0u8; 8]);
Expand Down Expand Up @@ -934,20 +871,20 @@ async fn test_v1_data_minimization() -> std::io::Result<()> {
async fn test_prune() -> std::io::Result<()> {
let dir = tempfile::tempdir()?;

let db = FilesDb::init(Duration::from_millis(2), dir.path().to_owned())
.await
.expect("initializing mailbox database should succeed");
let db = FilesDb::init(
Duration::from_millis(2),
dir.path().to_owned(),
Duration::from_secs(60 * 60 * 24 * 7),
)
.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);
Expand Down Expand Up @@ -984,38 +921,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);
}
}

Expand Down
Loading