Brummel/doctate
1
0
Fork 0
Code Issues 9 Pull Requests Actions Packages Projects Releases Wiki Activity

Formatting

Browse Source
This commit is contained in:
Brummel
2026-04-19 15:35:10 +02:00
parent 041f9015ca
commit 0d5c2f5888
42 changed files with 612 additions and 357 deletions
Download Patch File Download Diff File Expand all files Collapse all files
doctate-client-core/src/case_store.rs
+146 -63
View File
@@ -27,7 +27,7 @@ use serde::{Deserialize, Serialize};
use thiserror::Error;
use time::format_description::well_known::Rfc3339;
use time::{Duration, OffsetDateTime};
use tokio::sync::{watch, Mutex};
use tokio::sync::{Mutex, watch};
use tracing::{debug, info, warn};
use uuid::Uuid;
@@ -93,8 +93,7 @@ impl CaseStore {
})?;
let initial = read_all_markers(&dir).await?;
let map: HashMap<Uuid, CaseMarker> =
initial.into_iter().map(|m| (m.case_id, m)).collect();
let map: HashMap<Uuid, CaseMarker> = initial.into_iter().map(|m| (m.case_id, m)).collect();
let snapshot = sorted_snapshot(&map);
let (tx, rx) = watch::channel(snapshot);
@@ -278,7 +277,9 @@ impl CaseStore {
for id in &to_remove {
let path = self.dir.join(format!("{id}.json"));
match tokio::fs::remove_file(&path).await {
Ok(_) => info!(case_id = %id, "marker removed (server reconcile: case missing from snapshot)"),
Ok(_) => {
info!(case_id = %id, "marker removed (server reconcile: case missing from snapshot)")
}
Err(e) => warn!(case_id = %id, error = %e, "failed to remove reconciled marker"),
}
state.remove(id);
@@ -298,10 +299,7 @@ impl CaseStore {
/// Returns the number of markers removed. Intended as an
/// app-startup pass so the data-minimization rule holds: metadata
/// ages out after a few days even without an explicit UI action.
pub async fn cleanup_stale(
&self,
cutoff: OffsetDateTime,
) -> Result<usize, CaseStoreError> {
pub async fn cleanup_stale(&self, cutoff: OffsetDateTime) -> Result<usize, CaseStoreError> {
let cutoff_str = format_utc(cutoff)?;
let mut state = self.state.lock().await;
let to_remove: Vec<Uuid> = state
@@ -313,7 +311,9 @@ impl CaseStore {
for id in &to_remove {
let path = self.dir.join(format!("{id}.json"));
match tokio::fs::remove_file(&path).await {
Ok(_) => tracing::info!(case_id = %id, path = %path.display(), "stale marker removed"),
Ok(_) => {
tracing::info!(case_id = %id, path = %path.display(), "stale marker removed")
}
Err(e) => warn!(case_id = %id, error = %e, "failed to remove stale marker"),
}
state.remove(id);
@@ -328,13 +328,12 @@ impl CaseStore {
/// (new API key = potentially different user).
pub async fn clear_all(&self) -> Result<(), CaseStoreError> {
let mut state = self.state.lock().await;
let mut entries =
tokio::fs::read_dir(&self.dir)
.await
.map_err(|e| CaseStoreError::Io {
path: self.dir.clone(),
source: e,
})?;
let mut entries = tokio::fs::read_dir(&self.dir)
.await
.map_err(|e| CaseStoreError::Io {
path: self.dir.clone(),
source: e,
})?;
while let Ok(Some(entry)) = entries.next_entry().await {
let path = entry.path();
if path.extension().and_then(|s| s.to_str()) != Some("json") {
@@ -378,10 +377,12 @@ fn sorted_vec(state: &HashMap<Uuid, CaseMarker>) -> Vec<CaseMarker> {
async fn read_all_markers(dir: &Path) -> Result<Vec<CaseMarker>, CaseStoreError> {
let mut out = Vec::new();
let mut entries = tokio::fs::read_dir(dir).await.map_err(|e| CaseStoreError::Io {
path: dir.to_owned(),
source: e,
})?;
let mut entries = tokio::fs::read_dir(dir)
.await
.map_err(|e| CaseStoreError::Io {
path: dir.to_owned(),
source: e,
})?;
while let Ok(Some(entry)) = entries.next_entry().await {
let path = entry.path();
if path.extension().and_then(|s| s.to_str()) != Some("json") {
@@ -400,10 +401,12 @@ async fn read_all_markers(dir: &Path) -> Result<Vec<CaseMarker>, CaseStoreError>
}
async fn read_marker_file(path: &Path) -> Result<CaseMarker, CaseStoreError> {
let bytes = tokio::fs::read(path).await.map_err(|e| CaseStoreError::Io {
path: path.to_owned(),
source: e,
})?;
let bytes = tokio::fs::read(path)
.await
.map_err(|e| CaseStoreError::Io {
path: path.to_owned(),
source: e,
})?;
Ok(serde_json::from_slice(&bytes)?)
}
@@ -465,7 +468,10 @@ mod tests {
assert!(rx.borrow().is_empty());
let id = Uuid::new_v4();
let marker = store.create_local(id, t("2026-04-18T10:32:00Z")).await.unwrap();
let marker = store
.create_local(id, t("2026-04-18T10:32:00Z"))
.await
.unwrap();
assert_eq!(marker.case_id, id);
assert!(!marker.synced_to_server);
assert!(marker.oneliner.is_none());
@@ -486,8 +492,14 @@ mod tests {
let tmp = TempDir::new().unwrap();
let (store, _rx) = CaseStore::open(tmp.path().to_owned()).await.unwrap();
let id = Uuid::new_v4();
store.create_local(id, t("2026-04-18T10:00:00Z")).await.unwrap();
store.mark_activity(id, t("2026-04-18T10:45:00Z")).await.unwrap();
store
.create_local(id, t("2026-04-18T10:00:00Z"))
.await
.unwrap();
store
.mark_activity(id, t("2026-04-18T10:45:00Z"))
.await
.unwrap();
let list = store.list().await;
assert_eq!(list.len(), 1);
@@ -501,7 +513,10 @@ mod tests {
let tmp = TempDir::new().unwrap();
let (store, _rx) = CaseStore::open(tmp.path().to_owned()).await.unwrap();
let id = Uuid::new_v4();
store.mark_activity(id, t("2026-04-18T10:00:00Z")).await.unwrap();
store
.mark_activity(id, t("2026-04-18T10:00:00Z"))
.await
.unwrap();
assert_eq!(store.list().await.len(), 1);
}
@@ -529,7 +544,10 @@ mod tests {
let tmp = TempDir::new().unwrap();
let (store, _rx) = CaseStore::open(tmp.path().to_owned()).await.unwrap();
let id = Uuid::new_v4();
store.create_local(id, t("2026-04-18T10:00:00Z")).await.unwrap();
store
.create_local(id, t("2026-04-18T10:00:00Z"))
.await
.unwrap();
let snap = server_snapshot(vec![server_entry(
id,
@@ -551,7 +569,10 @@ mod tests {
let tmp = TempDir::new().unwrap();
let (store, _rx) = CaseStore::open(tmp.path().to_owned()).await.unwrap();
let id = Uuid::new_v4();
store.create_local(id, t("2026-04-18T10:00:00Z")).await.unwrap();
store
.create_local(id, t("2026-04-18T10:00:00Z"))
.await
.unwrap();
let snap = server_snapshot(vec![]);
store.merge_server_snapshot(&snap).await.unwrap();
@@ -566,16 +587,19 @@ mod tests {
let tmp = TempDir::new().unwrap();
let (store, _rx) = CaseStore::open(tmp.path().to_owned()).await.unwrap();
let id = Uuid::new_v4();
store.create_local(id, t("2026-04-18T10:00:00Z")).await.unwrap();
let synced_snap = server_snapshot(vec![server_entry(
id,
Some("old"),
"2026-04-18T10:00:00Z",
)]);
store
.create_local(id, t("2026-04-18T10:00:00Z"))
.await
.unwrap();
let synced_snap =
server_snapshot(vec![server_entry(id, Some("old"), "2026-04-18T10:00:00Z")]);
store.merge_server_snapshot(&synced_snap).await.unwrap();
// Next poll: server window moved, case fell out.
store.merge_server_snapshot(&server_snapshot(vec![])).await.unwrap();
store
.merge_server_snapshot(&server_snapshot(vec![]))
.await
.unwrap();
let list = store.list().await;
assert_eq!(list.len(), 1);
@@ -588,15 +612,17 @@ mod tests {
let tmp = TempDir::new().unwrap();
let (store, _rx) = CaseStore::open(tmp.path().to_owned()).await.unwrap();
let id = Uuid::new_v4();
store.create_local(id, t("2026-04-18T09:00:00Z")).await.unwrap();
store.mark_activity(id, t("2026-04-18T11:00:00Z")).await.unwrap();
store
.create_local(id, t("2026-04-18T09:00:00Z"))
.await
.unwrap();
store
.mark_activity(id, t("2026-04-18T11:00:00Z"))
.await
.unwrap();
// Server says last activity was 10:00 (stale) — local 11:00 must win.
let snap = server_snapshot(vec![server_entry(
id,
Some("x"),
"2026-04-18T10:00:00Z",
)]);
let snap = server_snapshot(vec![server_entry(id, Some("x"), "2026-04-18T10:00:00Z")]);
store.merge_server_snapshot(&snap).await.unwrap();
let list = store.list().await;
@@ -607,8 +633,14 @@ mod tests {
async fn clear_all_removes_everything() {
let tmp = TempDir::new().unwrap();
let (store, _rx) = CaseStore::open(tmp.path().to_owned()).await.unwrap();
store.create_local(Uuid::new_v4(), t("2026-04-18T10:00:00Z")).await.unwrap();
store.create_local(Uuid::new_v4(), t("2026-04-18T11:00:00Z")).await.unwrap();
store
.create_local(Uuid::new_v4(), t("2026-04-18T10:00:00Z"))
.await
.unwrap();
store
.create_local(Uuid::new_v4(), t("2026-04-18T11:00:00Z"))
.await
.unwrap();
assert_eq!(store.list().await.len(), 2);
store.clear_all().await.unwrap();
@@ -685,7 +717,10 @@ mod tests {
let old_unsynced = Uuid::new_v4();
// old_synced: created 4 days ago, then merged in → synced=true, old
store.create_local(old_synced, t("2026-04-14T10:00:00Z")).await.unwrap();
store
.create_local(old_synced, t("2026-04-14T10:00:00Z"))
.await
.unwrap();
store
.merge_server_snapshot(&server_snapshot(vec![OnelinerEntry {
case_id: old_synced.to_string(),
@@ -698,7 +733,10 @@ mod tests {
.unwrap();
// young_synced: created today, synced
store.create_local(young_synced, t("2026-04-18T10:00:00Z")).await.unwrap();
store
.create_local(young_synced, t("2026-04-18T10:00:00Z"))
.await
.unwrap();
store
.merge_server_snapshot(&server_snapshot(vec![OnelinerEntry {
case_id: young_synced.to_string(),
@@ -711,7 +749,10 @@ mod tests {
.unwrap();
// old_unsynced: created 4 days ago, NEVER merged in → pending upload
store.create_local(old_unsynced, t("2026-04-14T10:00:00Z")).await.unwrap();
store
.create_local(old_unsynced, t("2026-04-14T10:00:00Z"))
.await
.unwrap();
// Cutoff: 72h before the "today" markers, i.e. 2026-04-15.
let cutoff = t("2026-04-15T10:00:00Z");
@@ -729,15 +770,24 @@ mod tests {
// Disk: old_synced's file deleted.
let old_path = tmp.path().join(format!("{old_synced}.json"));
assert!(!old_path.exists(), "marker file for old_synced must be deleted");
assert!(
!old_path.exists(),
"marker file for old_synced must be deleted"
);
}
#[tokio::test]
async fn cleanup_stale_is_noop_when_nothing_old() {
let tmp = TempDir::new().unwrap();
let (store, _rx) = CaseStore::open(tmp.path().to_owned()).await.unwrap();
store.create_local(Uuid::new_v4(), t("2026-04-18T10:00:00Z")).await.unwrap();
let removed = store.cleanup_stale(t("2026-01-01T00:00:00Z")).await.unwrap();
store
.create_local(Uuid::new_v4(), t("2026-04-18T10:00:00Z"))
.await
.unwrap();
let removed = store
.cleanup_stale(t("2026-01-01T00:00:00Z"))
.await
.unwrap();
assert_eq!(removed, 0);
assert_eq!(store.list().await.len(), 1);
}
@@ -761,7 +811,10 @@ mod tests {
let id = Uuid::new_v4();
// Seed a synced marker with recent activity.
store.create_local(id, t("2026-04-18T09:00:00Z")).await.unwrap();
store
.create_local(id, t("2026-04-18T09:00:00Z"))
.await
.unwrap();
store.mark_synced(id).await.unwrap();
// Snapshot authoritative for [2026-04-15T10Z, 2026-04-18T10Z],
@@ -784,7 +837,10 @@ mod tests {
let id = Uuid::new_v4();
// Pending: synced_to_server = false (default after create_local).
store.create_local(id, t("2026-04-18T09:00:00Z")).await.unwrap();
store
.create_local(id, t("2026-04-18T09:00:00Z"))
.await
.unwrap();
let snap = snapshot_as_of("2026-04-18T10:00:00Z", vec![]);
let removed = store.reconcile_with_server_snapshot(&snap).await.unwrap();
@@ -804,7 +860,10 @@ mod tests {
// last_activity_at = 2026-04-10 (8 days ago); window = 72h anchor 2026-04-18T10Z
// → window_start = 2026-04-15T10Z. Marker is outside.
store.create_local(id, t("2026-04-10T10:00:00Z")).await.unwrap();
store
.create_local(id, t("2026-04-10T10:00:00Z"))
.await
.unwrap();
store.mark_synced(id).await.unwrap();
let snap = snapshot_as_of("2026-04-18T10:00:00Z", vec![]);
@@ -823,8 +882,14 @@ mod tests {
let (store, _rx) = CaseStore::open(tmp.path().to_owned()).await.unwrap();
let id_a = Uuid::new_v4();
let id_b = Uuid::new_v4();
store.create_local(id_a, t("2026-04-18T09:00:00Z")).await.unwrap();
store.create_local(id_b, t("2026-04-18T09:30:00Z")).await.unwrap();
store
.create_local(id_a, t("2026-04-18T09:00:00Z"))
.await
.unwrap();
store
.create_local(id_b, t("2026-04-18T09:30:00Z"))
.await
.unwrap();
store.mark_synced(id_a).await.unwrap();
store.mark_synced(id_b).await.unwrap();
@@ -852,7 +917,10 @@ mod tests {
let id = Uuid::new_v4();
// Marker last_activity_at 2026-04-18T09:00Z, synced.
store.create_local(id, t("2026-04-18T09:00:00Z")).await.unwrap();
store
.create_local(id, t("2026-04-18T09:00:00Z"))
.await
.unwrap();
store.mark_synced(id).await.unwrap();
// Server reports as_of 2026-04-18T10:00Z (1h after marker,
@@ -866,7 +934,10 @@ mod tests {
// Seed a second case for the symmetric assertion.
let id2 = Uuid::new_v4();
store.create_local(id2, t("2026-04-18T09:00:00Z")).await.unwrap();
store
.create_local(id2, t("2026-04-18T09:00:00Z"))
.await
.unwrap();
store.mark_synced(id2).await.unwrap();
// Now the server reports a much later as_of. The marker's
@@ -885,7 +956,10 @@ mod tests {
let tmp = TempDir::new().unwrap();
let (store, _rx) = CaseStore::open(tmp.path().to_owned()).await.unwrap();
let id = Uuid::new_v4();
store.create_local(id, t("2026-04-18T10:00:00Z")).await.unwrap();
store
.create_local(id, t("2026-04-18T10:00:00Z"))
.await
.unwrap();
assert!(!store.list().await[0].synced_to_server);
store.mark_synced(id).await.unwrap();
@@ -908,7 +982,10 @@ mod tests {
let tmp = TempDir::new().unwrap();
let (store, _rx) = CaseStore::open(tmp.path().to_owned()).await.unwrap();
let id = Uuid::new_v4();
store.create_local(id, t("2026-04-18T10:00:00Z")).await.unwrap();
store
.create_local(id, t("2026-04-18T10:00:00Z"))
.await
.unwrap();
store.mark_synced(id).await.unwrap();
store.mark_synced(id).await.unwrap(); // second call: no disk churn
assert!(store.list().await[0].synced_to_server);
@@ -920,8 +997,14 @@ mod tests {
let (store, _rx) = CaseStore::open(tmp.path().to_owned()).await.unwrap();
let older = Uuid::new_v4();
let newer = Uuid::new_v4();
store.create_local(older, t("2026-04-18T08:00:00Z")).await.unwrap();
store.create_local(newer, t("2026-04-18T10:00:00Z")).await.unwrap();
store
.create_local(older, t("2026-04-18T08:00:00Z"))
.await
.unwrap();
store
.create_local(newer, t("2026-04-18T10:00:00Z"))
.await
.unwrap();
let list = store.list().await;
assert_eq!(list[0].case_id, newer);
doctate-client-core/src/footer_status.rs
+16 -4
View File
@@ -59,8 +59,8 @@ pub fn pick_footer_status(
WorkerState::Polling => FooterStatus::Synchronizing,
WorkerState::Idle => {
let failure_more_recent = match (last_success_age, last_failure_age) {
(Some(s), Some(f)) => f < s, // smaller age = younger
(None, Some(_)) => true, // only a failure on record
(Some(s), Some(f)) => f < s, // smaller age = younger
(None, Some(_)) => true, // only a failure on record
_ => false,
};
if failure_more_recent {
@@ -88,7 +88,14 @@ mod tests {
queue_len: usize,
last_success_age: Option<Duration>,
) -> FooterStatus {
pick_footer_status(finalizing, worker_state, queue_len, last_success_age, None, THRESHOLD)
pick_footer_status(
finalizing,
worker_state,
queue_len,
last_success_age,
None,
THRESHOLD,
)
}
#[test]
@@ -128,7 +135,12 @@ mod tests {
#[test]
fn synchronizing_when_polling_and_no_queue() {
let s = status(false, WorkerState::Polling, 0, Some(Duration::from_millis(1)));
let s = status(
false,
WorkerState::Polling,
0,
Some(Duration::from_millis(1)),
);
assert_eq!(s, FooterStatus::Synchronizing);
}
doctate-client-core/src/lib.rs
+8 -4
View File
@@ -23,10 +23,14 @@ pub mod startup;
pub mod upload;
pub use case_store::{CaseMarker, CaseStore, CaseStoreError};
pub use config::{Config, ConfigError, DEFAULT_ONELINER_WINDOW_HOURS, DEFAULT_POLL_INTERVAL_SECONDS};
pub use footer_status::{pick_footer_status, FooterStatus};
pub use config::{
Config, ConfigError, DEFAULT_ONELINER_WINDOW_HOURS, DEFAULT_POLL_INTERVAL_SECONDS,
};
pub use footer_status::{FooterStatus, pick_footer_status};
pub use pending_cleanup::cleanup_orphan_audio;
pub use server_sync::{ServerSync, SyncConfig, WorkerSnapshot, WorkerState};
pub use snapshot_cache::{CachedSnapshot, SnapshotCache};
pub use startup::{run_startup_cleanup, DEFAULT_MARKER_RETENTION, DEFAULT_ORPHAN_AUDIO_RETENTION};
pub use upload::{scan_pending_dir, sidecar_path_for, write_sidecar, PendingUpload, UploadEvent, UploadIoError};
pub use startup::{DEFAULT_MARKER_RETENTION, DEFAULT_ORPHAN_AUDIO_RETENTION, run_startup_cleanup};
pub use upload::{
PendingUpload, UploadEvent, UploadIoError, scan_pending_dir, sidecar_path_for, write_sidecar,
};
doctate-client-core/src/server_sync.rs
+13 -9
View File
@@ -27,7 +27,7 @@ use doctate_common::constants::{
API_KEY_HEADER, CONTENT_TYPE_AUDIO_MP4, FIELD_AUDIO, FIELD_CASE_ID, FIELD_RECORDED_AT,
UPLOAD_PATH,
};
use doctate_common::oneliners::{OnelinersResponse, ONELINERS_PATH};
use doctate_common::oneliners::{ONELINERS_PATH, OnelinersResponse};
use reqwest::StatusCode;
use tokio::sync::{mpsc, watch};
use tokio::task::AbortHandle;
@@ -36,7 +36,7 @@ use uuid::Uuid;
use crate::case_store::CaseStore;
use crate::snapshot_cache::{CachedSnapshot, SnapshotCache};
use crate::upload::{scan_pending_dir, sidecar_path_for, PendingUpload, UploadEvent};
use crate::upload::{PendingUpload, UploadEvent, scan_pending_dir, sidecar_path_for};
const INITIAL_BACKOFF: Duration = Duration::from_secs(2);
const MAX_BACKOFF: Duration = Duration::from_secs(60);
@@ -233,8 +233,7 @@ async fn run_loop(
let mut files = scan_pending_dir(&pending_dir);
files.sort_by(|a, b| a.recorded_at.cmp(&b.recorded_at)); // FIFO
let queue_len = files.len();
let pending_ids: Arc<HashSet<Uuid>> =
Arc::new(files.iter().map(|u| u.case_id).collect());
let pending_ids: Arc<HashSet<Uuid>> = Arc::new(files.iter().map(|u| u.case_id).collect());
let next = files.into_iter().next();
if let Some(upload) = next {
@@ -565,7 +564,10 @@ mod tests {
let http = reqwest::Client::new();
let outcome = post_upload(&http, &cfg(&server), &upload).await;
assert!(matches!(outcome, UploadOutcome::Succeeded(AckStatus::Received)));
assert!(matches!(
outcome,
UploadOutcome::Succeeded(AckStatus::Received)
));
}
#[tokio::test]
@@ -710,7 +712,8 @@ mod tests {
sample_pending(&pending_dir, case_id, "2026-04-18T10-00-00Z");
let (store, cache) = make_store_and_cache(&tmp).await;
store.create_local(case_id, time::OffsetDateTime::now_utc())
store
.create_local(case_id, time::OffsetDateTime::now_utc())
.await
.unwrap();
@@ -914,10 +917,11 @@ mod tests {
async fn fifo_order_by_recorded_at() {
let server = MockServer::start().await;
// Record which case_ids we saw, in order.
let seen: Arc<tokio::sync::Mutex<Vec<Uuid>>> = Arc::new(tokio::sync::Mutex::new(Vec::new()));
let seen: Arc<tokio::sync::Mutex<Vec<Uuid>>> =
Arc::new(tokio::sync::Mutex::new(Vec::new()));
let seen_for_mock = seen.clone();
let responder = wiremock::ResponseTemplate::new(200)
.set_body_json(ack_body(AckStatus::Received));
let responder =
wiremock::ResponseTemplate::new(200).set_body_json(ack_body(AckStatus::Received));
// Simpler: just let both POSTs succeed; infer order from event
// stream on our side.
let _ = seen_for_mock;
doctate-client-core/src/snapshot_cache.rs
+4 -1
View File
@@ -123,7 +123,10 @@ mod tests {
let loaded = cache.load().await.expect("cache present");
assert_eq!(loaded.etag, s.etag);
assert_eq!(loaded.response.oneliners.len(), 1);
assert_eq!(loaded.response.oneliners[0].oneliner.as_deref(), Some("Kniegelenk"));
assert_eq!(
loaded.response.oneliners[0].oneliner.as_deref(),
Some("Kniegelenk")
);
}
#[tokio::test]
doctate-client-core/src/startup.rs
+6 -6
View File
@@ -43,8 +43,8 @@ pub async fn run_startup_cleanup(
marker_retention: Duration,
orphan_audio_retention: Duration,
) {
let marker_cutoff = OffsetDateTime::now_utc()
- time::Duration::seconds(marker_retention.as_secs() as i64);
let marker_cutoff =
OffsetDateTime::now_utc() - time::Duration::seconds(marker_retention.as_secs() as i64);
match store.cleanup_stale(marker_cutoff).await {
Ok(n) if n > 0 => info!(count = n, "startup cleanup: stale markers removed"),
Ok(_) => {}
@@ -61,8 +61,8 @@ pub async fn run_startup_cleanup(
#[cfg(test)]
mod tests {
use super::*;
use crate::upload::{write_sidecar, PendingUpload};
use filetime::{set_file_mtime, FileTime};
use crate::upload::{PendingUpload, write_sidecar};
use filetime::{FileTime, set_file_mtime};
use tempfile::TempDir;
use uuid::Uuid;
@@ -114,8 +114,8 @@ mod tests {
run_startup_cleanup(
&store,
&pending_dir,
Duration::from_secs(3600), // markers older than 1h: nuke
Duration::from_secs(3600), // orphans older than 1h: nuke
Duration::from_secs(3600), // markers older than 1h: nuke
Duration::from_secs(3600), // orphans older than 1h: nuke
)
.await;