filip/Stalwart
1
0
Fork 0
Code Issues Pull Requests Actions 2 Packages Projects Releases Wiki Activity

Fix pagination by anchor for queued messages, tasks and metrics

Browse Source
This commit is contained in:
Maurus Decimus
2026-05-04 11:56:34 +02:00
parent ea32e5b9c3
commit 1b5e7bf771
9 changed files with 428 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

105
tests/src/smtp/management/queue.rs
View File

@@ -25,6 +25,7 @@ use registry::{
};
use serde_json::json;
use std::time::{Duration, Instant};
use types::id::Id;
#[tokio::test]
#[serial_test::serial]
@@ -335,6 +336,110 @@ async fn manage_queue() {
);
}
// Test pagination (forward and reverse)
let asc_order: Vec<Id> = admin
.registry_query_paginated(
ObjectType::QueuedMessage,
"due",
true,
None,
None,
None,
None,
false,
)
.await
.object_ids()
.collect();
assert_eq!(asc_order.len(), 6, "expected 6 messages, got {asc_order:?}");
let desc_order: Vec<Id> = asc_order.iter().rev().copied().collect();
for chunk_start in [0usize, 2, 4] {
let asc = admin
.registry_query_paginated(
ObjectType::QueuedMessage,
"due",
true,
Some(chunk_start as i32),
Some(2),
None,
None,
false,
)
.await
.object_ids()
.collect::<Vec<_>>();
assert_eq!(
asc,
asc_order[chunk_start..chunk_start + 2],
"ascending position={chunk_start} limit=2",
);
let desc = admin
.registry_query_paginated(
ObjectType::QueuedMessage,
"due",
false,
Some(chunk_start as i32),
Some(2),
None,
None,
false,
)
.await
.object_ids()
.collect::<Vec<_>>();
assert_eq!(
desc,
desc_order[chunk_start..chunk_start + 2],
"descending position={chunk_start} limit=2",
);
}
for anchor_idx in [1usize, 3] {
let asc = admin
.registry_query_paginated(
ObjectType::QueuedMessage,
"due",
true,
None,
Some(2),
Some(asc_order[anchor_idx]),
Some(1),
false,
)
.await
.object_ids()
.collect::<Vec<_>>();
assert_eq!(
asc,
asc_order[anchor_idx + 1..anchor_idx + 3],
"ascending anchor={} offset=1 limit=2",
asc_order[anchor_idx],
);
let desc = admin
.registry_query_paginated(
ObjectType::QueuedMessage,
"due",
false,
None,
Some(2),
Some(desc_order[anchor_idx]),
Some(1),
false,
)
.await
.object_ids()
.collect::<Vec<_>>();
assert_eq!(
desc,
desc_order[anchor_idx + 1..anchor_idx + 3],
"descending anchor={} offset=1 limit=2",
desc_order[anchor_idx],
);
}
// Retry delivery
admin
.registry_update_object(

136
tests/src/system/task.rs
View File

@@ -137,9 +137,145 @@ pub async fn test(test: &mut TestServer) {
.await
.assert_destroyed(&[task.id]);
pagination_test(test).await;
test.cleanup().await;
}
async fn pagination_test(test: &mut TestServer) {
println!("Running Task pagination tests...");
let admin = test.account("admin@example.org");
admin.assert_no_tasks().await;
let mut created = Vec::with_capacity(12);
for i in 0..12u64 {
created.push(admin.schedule_test_task(TASK_SUCCESS, 3600 + i).await);
}
let asc_order: Vec<Id> = admin
.registry_query_paginated(ObjectType::Task, "due", true, None, None, None, None, false)
.await
.object_ids()
.collect();
assert_eq!(asc_order.len(), 12, "expected 12 tasks, got {}", asc_order.len());
let desc_order: Vec<Id> = asc_order.iter().rev().copied().collect();
for chunk_start in [0usize, 5, 10] {
let chunk_size = std::cmp::min(5, 12 - chunk_start);
let asc = admin
.registry_query_paginated(
ObjectType::Task,
"due",
true,
Some(chunk_start as i32),
Some(5),
None,
None,
false,
)
.await
.object_ids()
.collect::<Vec<_>>();
assert_eq!(
asc,
asc_order[chunk_start..chunk_start + chunk_size],
"ascending position={chunk_start} limit=5",
);
let desc = admin
.registry_query_paginated(
ObjectType::Task,
"due",
false,
Some(chunk_start as i32),
Some(5),
None,
None,
false,
)
.await
.object_ids()
.collect::<Vec<_>>();
assert_eq!(
desc,
desc_order[chunk_start..chunk_start + chunk_size],
"descending position={chunk_start} limit=5",
);
}
for anchor_idx in [4usize, 9] {
let chunk_size = std::cmp::min(5, 12 - anchor_idx - 1);
let asc = admin
.registry_query_paginated(
ObjectType::Task,
"due",
true,
None,
Some(5),
Some(asc_order[anchor_idx]),
Some(1),
false,
)
.await
.object_ids()
.collect::<Vec<_>>();
assert_eq!(
asc,
asc_order[anchor_idx + 1..anchor_idx + 1 + chunk_size],
"ascending anchor={} offset=1 limit=5",
asc_order[anchor_idx],
);
let desc = admin
.registry_query_paginated(
ObjectType::Task,
"due",
false,
None,
Some(5),
Some(desc_order[anchor_idx]),
Some(1),
false,
)
.await
.object_ids()
.collect::<Vec<_>>();
assert_eq!(
desc,
desc_order[anchor_idx + 1..anchor_idx + 1 + chunk_size],
"descending anchor={} offset=1 limit=5",
desc_order[anchor_idx],
);
}
let response = admin
.registry_query_paginated(
ObjectType::Task,
"due",
true,
Some(0),
Some(5),
None,
None,
true,
)
.await;
let total = response
.pointer("/methodResponses/0/1/total")
.and_then(|v| v.as_u64());
assert_eq!(total, Some(12), "expected calculateTotal=12");
admin
.registry_destroy(ObjectType::Task, created.clone())
.await
.assert_destroyed(&created);
admin.assert_no_tasks().await;
}
impl Account {
async fn schedule_test_task(&self, test_type: u64, schedule_in: u64) -> Id {
self.registry_create_object(Task::StoreMaintenance(TaskStoreMaintenance {

108
tests/src/telemetry/metrics.rs
View File

@@ -72,6 +72,114 @@ pub async fn test(test: &TestServer) {
metric_ids.len()
);
// Test pagination (forward and reverse)
let asc_order: Vec<Id> = admin
.registry_query_paginated(
ObjectType::Metric,
"timestamp",
true,
None,
None,
None,
None,
false,
)
.await
.object_ids()
.collect();
assert!(asc_order.len() > 100, "expected >100 metrics, got {}", asc_order.len());
let desc_order: Vec<Id> = asc_order.iter().rev().copied().collect();
let total = asc_order.len();
let limit = 25usize;
for chunk_start in [0usize, limit, total - limit] {
let asc = admin
.registry_query_paginated(
ObjectType::Metric,
"timestamp",
true,
Some(chunk_start as i32),
Some(limit),
None,
None,
false,
)
.await
.object_ids()
.collect::<Vec<_>>();
assert_eq!(
asc,
asc_order[chunk_start..chunk_start + limit],
"ascending position={chunk_start} limit={limit}",
);
let desc = admin
.registry_query_paginated(
ObjectType::Metric,
"timestamp",
false,
Some(chunk_start as i32),
Some(limit),
None,
None,
false,
)
.await
.object_ids()
.collect::<Vec<_>>();
assert_eq!(
desc,
desc_order[chunk_start..chunk_start + limit],
"descending position={chunk_start} limit={limit}",
);
}
for anchor_idx in [limit - 1, total - limit - 1] {
let asc = admin
.registry_query_paginated(
ObjectType::Metric,
"timestamp",
true,
None,
Some(limit),
Some(asc_order[anchor_idx]),
Some(1),
false,
)
.await
.object_ids()
.collect::<Vec<_>>();
let asc_size = std::cmp::min(limit, total - anchor_idx - 1);
assert_eq!(
asc,
asc_order[anchor_idx + 1..anchor_idx + 1 + asc_size],
"ascending anchor={} offset=1 limit={limit}",
asc_order[anchor_idx],
);
let desc = admin
.registry_query_paginated(
ObjectType::Metric,
"timestamp",
false,
None,
Some(limit),
Some(desc_order[anchor_idx]),
Some(1),
false,
)
.await
.object_ids()
.collect::<Vec<_>>();
let desc_size = std::cmp::min(limit, total - anchor_idx - 1);
assert_eq!(
desc,
desc_order[anchor_idx + 1..anchor_idx + 1 + desc_size],
"descending anchor={} offset=1 limit={limit}",
desc_order[anchor_idx],
);
}
// Purge metrics and make sure they are gone
test.server
.metrics_store()

43
tests/src/utils/registry.rs
View File

@@ -144,6 +144,49 @@ impl Account {
.await
}
#[allow(clippy::too_many_arguments)]
pub async fn registry_query_paginated(
&self,
object: ObjectType,
sort_property: &str,
sort_ascending: bool,
position: Option<i32>,
limit: Option<usize>,
anchor: Option<Id>,
anchor_offset: Option<i32>,
calculate_total: bool,
) -> JmapResponse {
let name = object.as_str();
let mut args = serde_json::Map::new();
args.insert("filter".into(), json!({}));
args.insert(
"sort".into(),
json!([{ "property": sort_property, "isAscending": sort_ascending }]),
);
if let Some(p) = position {
args.insert("position".into(), json!(p));
}
if let Some(l) = limit {
args.insert("limit".into(), json!(l));
}
if let Some(a) = anchor {
args.insert("anchor".into(), json!(a.to_string()));
}
if let Some(ao) = anchor_offset {
args.insert("anchorOffset".into(), json!(ao));
}
if calculate_total {
args.insert("calculateTotal".into(), json!(true));
}
self.jmap_method_calls(json!([[
format!("x:{name}/query"),
Value::Object(args),
"0"
]]))
.await
}
pub async fn registry_destroy(
&self,
object: ObjectType,