From b265562a2081fb5b07df3d7ff5118a8f2c453493 Mon Sep 17 00:00:00 2001 From: Nicolas Burtey Date: Sun, 2 Aug 2026 10:59:36 -0600 Subject: [PATCH] feat(outbox): archive old outbox events to cold storage Port of GaloyMoney/lana-bank#7757 (obix event-archive wiring, GaloyMoney/obix#105) to the cala outbox, using the obix branch as a git dependency until the archive feature is released on crates.io: - cala-ledger: OutboxArchiveConfig { storage, retention_days = 3, path_prefix = "outbox-archive/cala/" } on CalaLedgerConfig; the consumer supplies an obix::EventArchiveStorage backend (GCS, S3, local fs) and settled days are swept out of postgres to JSONL chunks, with transparent read fallback for pre-watermark history - CalaLedger::register_outbox_archiver registers the "cala.outbox.archiver" job (one settled day per run, rescheduling while catching up) on the app's job executor - boundary: obix's age-based DailyRetentionBoundary over CalaMailboxTables - migration: cala_persistent_outbox_archive_chunks manifest folded into the cala obix setup migration (in-place, matching the recent pre-production edits of that migration) - .sqlx offline cache regenerated (5 new archive queries) Not yet: a released obix version (revert to crates.io once GaloyMoney/obix#105 ships); lana's dw-outbox-relay can then drop its obix-cala pin. --- Cargo.lock | 11 +- Cargo.toml | 2 +- ...f8d8e96eb7731327f5cd997fc1947cf3fe6a9.json | 48 +++++ ...a65ff458b0acbca857e90a385c537e8fb5e7f.json | 20 ++ ...a3b627f0aeef9449200ebfd49318e16991ec3.json | 24 +++ ...1e7fba0c357044122702c011839e50669c399.json | 29 +++ ...35b34c409e7504bbddc3ef5d0f1c9c7b9046f.json | 34 +++ .../20251204130226_cala_obix_setup.sql | 15 ++ cala-ledger/src/ledger/config.rs | 8 + cala-ledger/src/ledger/mod.rs | 21 +- cala-ledger/src/outbox/config.rs | 69 ++++++ cala-ledger/src/outbox/mod.rs | 2 + cala-ledger/src/outbox/publisher.rs | 9 +- cala-ledger/tests/outbox_archive.rs | 198 ++++++++++++++++++ 14 files changed, 480 insertions(+), 10 deletions(-) create mode 100644 cala-ledger/.sqlx/query-3f0fc13dc2b94fb2b41ca1d8c71f8d8e96eb7731327f5cd997fc1947cf3fe6a9.json create mode 100644 cala-ledger/.sqlx/query-6865c0a3671769b4171782c8c10a65ff458b0acbca857e90a385c537e8fb5e7f.json create mode 100644 cala-ledger/.sqlx/query-cab63371538c857e583492edabca3b627f0aeef9449200ebfd49318e16991ec3.json create mode 100644 cala-ledger/.sqlx/query-e9651d0da9d8ee98ff5344a5e201e7fba0c357044122702c011839e50669c399.json create mode 100644 cala-ledger/.sqlx/query-fa6f632029b2016f71a67592f6835b34c409e7504bbddc3ef5d0f1c9c7b9046f.json create mode 100644 cala-ledger/src/outbox/config.rs create mode 100644 cala-ledger/tests/outbox_archive.rs diff --git a/Cargo.lock b/Cargo.lock index 2d53a1b69..e71a9c25d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1611,11 +1611,11 @@ dependencies = [ [[package]] name = "obix" -version = "0.5.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7918bab03582f8ccb1f1a932d28b2220cab559339c6e48c731a1d92920bc830a" +version = "0.5.1-dev" +source = "git+https://github.com/GaloyMoney/obix.git?branch=feat-event-archive#e38bb75746931bbbe1fef5f05df92bed98b8b8a9" dependencies = [ "async-trait", + "bytes", "chrono", "derive_builder", "es-entity", @@ -1634,9 +1634,8 @@ dependencies = [ [[package]] name = "obix-macros" -version = "0.5.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d0d22945a31b5451e80efefcc19c5e995457f833a0771d76bd05bd5af1a0ad38" +version = "0.5.1-dev" +source = "git+https://github.com/GaloyMoney/obix.git?branch=feat-event-archive#e38bb75746931bbbe1fef5f05df92bed98b8b8a9" dependencies = [ "darling 0.23.0", "proc-macro2", diff --git a/Cargo.toml b/Cargo.toml index 3fa19e905..e86ece3b2 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -17,7 +17,7 @@ cala-ledger = { path = "cala-ledger", version = "0.20.2-dev" } cel = "0.14.1" es-entity = "0.11.11" job = { version = "0.6.36", features = ["es-entity"] } -obix = { version = "0.5.0", default-features = false } +obix = { git = "https://github.com/GaloyMoney/obix.git", branch = "feat-event-archive", default-features = false } # git dependency until GaloyMoney/obix#105 is released anyhow = "1.0.99" cached = { version = "2.0", features = ["async"] } diff --git a/cala-ledger/.sqlx/query-3f0fc13dc2b94fb2b41ca1d8c71f8d8e96eb7731327f5cd997fc1947cf3fe6a9.json b/cala-ledger/.sqlx/query-3f0fc13dc2b94fb2b41ca1d8c71f8d8e96eb7731327f5cd997fc1947cf3fe6a9.json new file mode 100644 index 000000000..f8d112dc2 --- /dev/null +++ b/cala-ledger/.sqlx/query-3f0fc13dc2b94fb2b41ca1d8c71f8d8e96eb7731327f5cd997fc1947cf3fe6a9.json @@ -0,0 +1,48 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT id, sequence AS \"sequence!: i64\", payload, tracing_context, recorded_at\n FROM cala_persistent_outbox_events\n WHERE sequence > $1\n AND sequence <= $2\n ORDER BY sequence\n LIMIT $3", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "id", + "type_info": "Uuid" + }, + { + "ordinal": 1, + "name": "sequence!: i64", + "type_info": "Int8" + }, + { + "ordinal": 2, + "name": "payload", + "type_info": "Jsonb" + }, + { + "ordinal": 3, + "name": "tracing_context", + "type_info": "Jsonb" + }, + { + "ordinal": 4, + "name": "recorded_at", + "type_info": "Timestamptz" + } + ], + "parameters": { + "Left": [ + "Int8", + "Int8", + "Int8" + ] + }, + "nullable": [ + false, + false, + true, + true, + false + ] + }, + "hash": "3f0fc13dc2b94fb2b41ca1d8c71f8d8e96eb7731327f5cd997fc1947cf3fe6a9" +} diff --git a/cala-ledger/.sqlx/query-6865c0a3671769b4171782c8c10a65ff458b0acbca857e90a385c537e8fb5e7f.json b/cala-ledger/.sqlx/query-6865c0a3671769b4171782c8c10a65ff458b0acbca857e90a385c537e8fb5e7f.json new file mode 100644 index 000000000..3a816afab --- /dev/null +++ b/cala-ledger/.sqlx/query-6865c0a3671769b4171782c8c10a65ff458b0acbca857e90a385c537e8fb5e7f.json @@ -0,0 +1,20 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT MAX(max_sequence) AS \"max_sequence?\" FROM cala_persistent_outbox_archive_chunks", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "max_sequence?", + "type_info": "Int8" + } + ], + "parameters": { + "Left": [] + }, + "nullable": [ + null + ] + }, + "hash": "6865c0a3671769b4171782c8c10a65ff458b0acbca857e90a385c537e8fb5e7f" +} diff --git a/cala-ledger/.sqlx/query-cab63371538c857e583492edabca3b627f0aeef9449200ebfd49318e16991ec3.json b/cala-ledger/.sqlx/query-cab63371538c857e583492edabca3b627f0aeef9449200ebfd49318e16991ec3.json new file mode 100644 index 000000000..88b7edf86 --- /dev/null +++ b/cala-ledger/.sqlx/query-cab63371538c857e583492edabca3b627f0aeef9449200ebfd49318e16991ec3.json @@ -0,0 +1,24 @@ +{ + "db_name": "PostgreSQL", + "query": "WITH deleted AS (\n DELETE FROM cala_persistent_outbox_events\n WHERE sequence >= $2 AND sequence <= $3\n RETURNING sequence\n )\n INSERT INTO cala_persistent_outbox_archive_chunks (path, min_sequence, max_sequence)\n SELECT $1, $2, $3\n ON CONFLICT (path) DO NOTHING\n RETURNING path", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "path", + "type_info": "Text" + } + ], + "parameters": { + "Left": [ + "Text", + "Int8", + "Int8" + ] + }, + "nullable": [ + false + ] + }, + "hash": "cab63371538c857e583492edabca3b627f0aeef9449200ebfd49318e16991ec3" +} diff --git a/cala-ledger/.sqlx/query-e9651d0da9d8ee98ff5344a5e201e7fba0c357044122702c011839e50669c399.json b/cala-ledger/.sqlx/query-e9651d0da9d8ee98ff5344a5e201e7fba0c357044122702c011839e50669c399.json new file mode 100644 index 000000000..1e4276d71 --- /dev/null +++ b/cala-ledger/.sqlx/query-e9651d0da9d8ee98ff5344a5e201e7fba0c357044122702c011839e50669c399.json @@ -0,0 +1,29 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT (recorded_at AT TIME ZONE 'UTC')::date AS \"day!\",\n MAX(sequence) AS \"max_sequence!: i64\"\n FROM cala_persistent_outbox_events\n WHERE sequence > $1\n AND recorded_at < $2\n GROUP BY 1\n ORDER BY 1", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "day!", + "type_info": "Date" + }, + { + "ordinal": 1, + "name": "max_sequence!: i64", + "type_info": "Int8" + } + ], + "parameters": { + "Left": [ + "Int8", + "Timestamptz" + ] + }, + "nullable": [ + null, + null + ] + }, + "hash": "e9651d0da9d8ee98ff5344a5e201e7fba0c357044122702c011839e50669c399" +} diff --git a/cala-ledger/.sqlx/query-fa6f632029b2016f71a67592f6835b34c409e7504bbddc3ef5d0f1c9c7b9046f.json b/cala-ledger/.sqlx/query-fa6f632029b2016f71a67592f6835b34c409e7504bbddc3ef5d0f1c9c7b9046f.json new file mode 100644 index 000000000..f99a6b1f4 --- /dev/null +++ b/cala-ledger/.sqlx/query-fa6f632029b2016f71a67592f6835b34c409e7504bbddc3ef5d0f1c9c7b9046f.json @@ -0,0 +1,34 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT path, min_sequence, max_sequence\n FROM cala_persistent_outbox_archive_chunks\n WHERE max_sequence > $1\n ORDER BY min_sequence", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "path", + "type_info": "Text" + }, + { + "ordinal": 1, + "name": "min_sequence", + "type_info": "Int8" + }, + { + "ordinal": 2, + "name": "max_sequence", + "type_info": "Int8" + } + ], + "parameters": { + "Left": [ + "Int8" + ] + }, + "nullable": [ + false, + false, + false + ] + }, + "hash": "fa6f632029b2016f71a67592f6835b34c409e7504bbddc3ef5d0f1c9c7b9046f" +} diff --git a/cala-ledger/migrations/20251204130226_cala_obix_setup.sql b/cala-ledger/migrations/20251204130226_cala_obix_setup.sql index 25cc3755c..89f6d9ee9 100644 --- a/cala-ledger/migrations/20251204130226_cala_obix_setup.sql +++ b/cala-ledger/migrations/20251204130226_cala_obix_setup.sql @@ -8,6 +8,21 @@ CREATE TABLE cala_persistent_outbox_events ( seen_at TIMESTAMPTZ NOT NULL DEFAULT NOW() ); +-- Archive manifest: one row per exported JSONL chunk of pruned +-- persistent outbox events. Chunks are contiguous — the next chunk starts +-- at max_sequence + 1 of the previous one. +-- Any grouping label (e.g. a calendar date) is encoded in a chunk's +-- path; grouping semantics belong to the deployment, not to obix. +CREATE TABLE cala_persistent_outbox_archive_chunks ( + path TEXT PRIMARY KEY, + min_sequence BIGINT NOT NULL, + max_sequence BIGINT NOT NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW() +); + +CREATE INDEX cala_idx_persistent_outbox_archive_chunks_max_sequence + ON cala_persistent_outbox_archive_chunks (max_sequence); + -- Ephemeral outbox events CREATE TABLE cala_ephemeral_outbox_events ( event_type VARCHAR NOT NULL UNIQUE, diff --git a/cala-ledger/src/ledger/config.rs b/cala-ledger/src/ledger/config.rs index 97c3f0e47..6855f24ea 100644 --- a/cala-ledger/src/ledger/config.rs +++ b/cala-ledger/src/ledger/config.rs @@ -1,6 +1,8 @@ use derive_builder::Builder; use es_entity::clock::{Clock, ClockHandle}; +use crate::outbox::OutboxArchiveConfig; + #[derive(Builder, Clone, Debug)] #[builder(build_fn(validate = "Self::validate"))] pub struct CalaLedgerConfig { @@ -14,6 +16,12 @@ pub struct CalaLedgerConfig { pub(super) pool: Option, #[builder(setter(into), default = "Clock::handle().clone()")] pub(super) clock: ClockHandle, + /// Cold-storage archiving of old outbox events. When set, settled + /// history is swept out of postgres by the archiver job (see + /// [`CalaLedger::register_outbox_archiver`](crate::ledger::CalaLedger::register_outbox_archiver)) + /// and pre-watermark reads fall back to the archive. + #[builder(setter(strip_option), default)] + pub(super) outbox_archive: Option, } impl CalaLedgerConfig { diff --git a/cala-ledger/src/ledger/mod.rs b/cala-ledger/src/ledger/mod.rs index b3e2a38e7..5a1e77484 100644 --- a/cala-ledger/src/ledger/mod.rs +++ b/cala-ledger/src/ledger/mod.rs @@ -63,7 +63,8 @@ impl CalaLedger { } let clock = config.clock; - let publisher = OutboxPublisher::init(&pool, &clock).await?; + let publisher = + OutboxPublisher::init(&pool, &clock, config.outbox_archive.as_ref()).await?; let accounts = Accounts::new(&pool, &publisher, &clock); let journals = Journals::new(&pool, &publisher, &clock); let tx_templates = TxTemplates::new(&pool, &publisher, &clock); @@ -223,6 +224,24 @@ impl CalaLedger { self.publisher.inner() } + /// Register the job sweeping settled spans of outbox history to the + /// configured archive storage (one span per run, rescheduling until + /// caught up). Requires `outbox_archive` to have been set at + /// [`init`](Self::init) — fails with `obix::ArchiveError::NotConfigured` + /// otherwise. + pub async fn register_outbox_archiver( + &self, + jobs: &mut job::Jobs, + ) -> Result<(), Box> { + self.publisher + .inner() + .register_event_archiver( + jobs, + obix::OutboxArchiverJobConfig::new(job::JobType::new("cala.outbox.archiver")), + ) + .await + } + pub fn register_outbox_listener( &self, start_after: Option, diff --git a/cala-ledger/src/outbox/config.rs b/cala-ledger/src/outbox/config.rs new file mode 100644 index 000000000..f0a3999dd --- /dev/null +++ b/cala-ledger/src/outbox/config.rs @@ -0,0 +1,69 @@ +use es_entity::clock::ClockHandle; + +use std::sync::Arc; + +use super::CalaMailboxTables; + +pub const DEFAULT_OUTBOX_ARCHIVE_RETENTION_DAYS: u32 = 3; +pub const DEFAULT_OUTBOX_ARCHIVE_PATH_PREFIX: &str = "outbox-archive/cala/"; + +/// Cold-storage archiving of old outbox events (obix archive). Archived +/// history is swept to object storage and pruned from postgres; reads of +/// pre-watermark events transparently fall back to the archive. +/// +/// The object-storage backend is supplied by the consumer via +/// [`obix::EventArchiveStorage`] (GCS, S3, local filesystem, ...); +/// [`obix::InMemoryArchiveStorage`] works for tests. +#[derive(Clone)] +pub struct OutboxArchiveConfig { + /// The object-storage backend chunks are written to / read from. + pub storage: Arc, + /// Days of history kept in postgres; older, fully-elapsed days are + /// swept to storage one day per archiver run. + pub retention_days: u32, + /// Prepended to every chunk path, e.g. `"outbox-archive/cala/"`. + pub path_prefix: String, +} + +impl OutboxArchiveConfig { + pub fn new(storage: Arc) -> Self { + Self { + storage, + retention_days: DEFAULT_OUTBOX_ARCHIVE_RETENTION_DAYS, + path_prefix: DEFAULT_OUTBOX_ARCHIVE_PATH_PREFIX.to_string(), + } + } + + pub fn with_retention_days(mut self, retention_days: u32) -> Self { + self.retention_days = retention_days; + self + } + + pub fn with_path_prefix(mut self, prefix: impl Into) -> Self { + self.path_prefix = prefix.into(); + self + } + + pub(super) fn build(&self, pool: &sqlx::PgPool, clock: &ClockHandle) -> obix::ArchiveConfig { + obix::ArchiveConfig::new( + self.storage.clone(), + Arc::new(obix::DailyRetentionBoundary::::new( + pool, + chrono::Duration::days(i64::from(self.retention_days)), + clock.clone(), + )), + ) + .with_path_prefix(self.path_prefix.clone()) + .with_clock(clock.clone()) + } +} + +impl std::fmt::Debug for OutboxArchiveConfig { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("OutboxArchiveConfig") + .field("storage", &"") + .field("retention_days", &self.retention_days) + .field("path_prefix", &self.path_prefix) + .finish() + } +} diff --git a/cala-ledger/src/outbox/mod.rs b/cala-ledger/src/outbox/mod.rs index 7c87c225a..c2e98f356 100644 --- a/cala-ledger/src/outbox/mod.rs +++ b/cala-ledger/src/outbox/mod.rs @@ -1,9 +1,11 @@ +mod config; mod publisher; mod event { pub use cala_types::outbox::*; } +pub use config::OutboxArchiveConfig; pub use event::*; pub use publisher::OutboxPublisher; diff --git a/cala-ledger/src/outbox/publisher.rs b/cala-ledger/src/outbox/publisher.rs index ea5ffaf57..78c71aa94 100644 --- a/cala-ledger/src/outbox/publisher.rs +++ b/cala-ledger/src/outbox/publisher.rs @@ -1,7 +1,7 @@ use cala_types::outbox::OutboxEventPayload; use es_entity::clock::ClockHandle; -use super::ObixOutbox; +use super::{ObixOutbox, OutboxArchiveConfig}; #[derive(Debug, Clone)] pub struct OutboxPublisher { @@ -9,11 +9,16 @@ pub struct OutboxPublisher { } impl OutboxPublisher { - pub async fn init(pool: &sqlx::PgPool, clock: &ClockHandle) -> Result { + pub async fn init( + pool: &sqlx::PgPool, + clock: &ClockHandle, + archive: Option<&OutboxArchiveConfig>, + ) -> Result { let config = obix::MailboxConfig::builder() .clock(clock.clone()) .event_buffer_size(50_000) .event_cache_size(10_000) + .archive(archive.map(|a| a.build(pool, clock))) .build() .expect("MailboxConfig"); let outbox = ObixOutbox::init(pool, config).await?; diff --git a/cala-ledger/tests/outbox_archive.rs b/cala-ledger/tests/outbox_archive.rs new file mode 100644 index 000000000..c7bf61d88 --- /dev/null +++ b/cala-ledger/tests/outbox_archive.rs @@ -0,0 +1,198 @@ +mod helpers; + +use chrono::{DateTime, NaiveDate, Utc}; +use es_entity::clock::ClockHandle; +use futures::StreamExt; +use rand::distr::{Alphanumeric, SampleString}; + +use std::{collections::HashSet, sync::Arc}; + +use cala_ledger::{outbox::OutboxArchiveConfig, *}; + +const DAY: std::time::Duration = std::time::Duration::from_secs(24 * 60 * 60); + +fn day(n: u32) -> DateTime { + // 2026-07-20 is "day 0"; return noon of day n to stay clear of + // midnight boundaries. + NaiveDate::from_ymd_opt(2026, 7, 20) + .unwrap() + .checked_add_days(chrono::Days::new(n as u64)) + .unwrap() + .and_hms_opt(12, 0, 0) + .unwrap() + .and_utc() +} + +/// A dedicated database for the test. The outbox tables are shared by +/// every test in this crate, and other tests write events with +/// past-dated (manual-clock) `recorded_at` — under a shared table the +/// date-bucketed archive spans and their sequence-range pruning cannot +/// be reasoned about, so this test needs a database of its own. +async fn init_isolated_pool() -> anyhow::Result<(sqlx::PgPool, String)> { + let pg_con = std::env::var("PG_CON")?; + let (base, _) = pg_con.rsplit_once('/').expect("PG_CON has a database path"); + + let admin = sqlx::PgPool::connect(&format!("{base}/postgres")).await?; + let db_name = format!( + "cala_outbox_archive_{}", + Alphanumeric + .sample_string(&mut rand::rng(), 8) + .to_lowercase() + ); + sqlx::query(&format!("CREATE DATABASE {db_name}")) + .execute(&admin) + .await?; + admin.close().await; + + let pool = sqlx::PgPool::connect(&format!("{base}/{db_name}")).await?; + sqlx::migrate!().run(&pool).await?; + Ok((pool, db_name)) +} + +async fn drop_database(db_name: &str) -> anyhow::Result<()> { + let pg_con = std::env::var("PG_CON")?; + let (base, _) = pg_con.rsplit_once('/').expect("PG_CON has a database path"); + let admin = sqlx::PgPool::connect(&format!("{base}/postgres")).await?; + sqlx::query(&format!("DROP DATABASE {db_name} WITH (FORCE)")) + .execute(&admin) + .await?; + Ok(()) +} + +async fn create_account(cala: &CalaLedger) -> account::Account { + let (account, _) = helpers::test_accounts(); + cala.accounts() + .create(account) + .await + .expect("create account") +} + +async fn old_event_count(pool: &sqlx::PgPool, before: DateTime) -> anyhow::Result { + let (count,): (i64,) = + sqlx::query_as("SELECT COUNT(*) FROM cala_persistent_outbox_events WHERE recorded_at < $1") + .bind(before) + .fetch_one(pool) + .await?; + Ok(count) +} + +#[tokio::test] +async fn outbox_archive_sweeps_settled_days_and_replays_across_seam() -> anyhow::Result<()> { + let (pool, db_name) = init_isolated_pool().await?; + let (clock, controller) = ClockHandle::manual_at(day(0)); + + let storage = Arc::new(obix::InMemoryArchiveStorage::new()); + let archive = OutboxArchiveConfig::new(storage.clone()).with_retention_days(2); + + let cala = CalaLedger::init( + CalaLedgerConfig::builder() + .pool(pool.clone()) + .exec_migrations(false) + .clock(clock) + .outbox_archive(archive) + .build()?, + ) + .await?; + + // Days 0 and 1 fall inside the archive window; day 3 is "today". + let journal = cala.journals().create(helpers::test_journal()).await?; + let day0_account = create_account(&cala).await; + controller.advance(DAY).await; + let day1_account = create_account(&cala).await; + controller.advance(DAY).await; + controller.advance(DAY).await; + let day3_account = create_account(&cala).await; + + let job_config = job::JobSvcConfig::builder() + .pool(pool.clone()) + .build() + .unwrap(); + let mut jobs = job::Jobs::init(job_config).await?; + cala.register_outbox_archiver(&mut jobs) + .await + .map_err(|e| anyhow::anyhow!(e.to_string()))?; + jobs.start_poll().await?; + + // The job sweeps one settled day per run, rescheduling while catching + // up: eventually nothing recorded before day 2 remains in postgres... + let start = std::time::Instant::now(); + while old_event_count(&pool, day(2)).await? > 0 { + if start.elapsed() > std::time::Duration::from_secs(30) { + panic!("archiver job did not sweep settled days within timeout"); + } + tokio::time::sleep(std::time::Duration::from_millis(100)).await; + } + // ...while the day 3 events stay. + let (recent_count,): (i64,) = sqlx::query_as( + "SELECT COUNT(*) FROM cala_persistent_outbox_events WHERE recorded_at >= $1", + ) + .bind(day(2)) + .fetch_one(&pool) + .await?; + assert!(recent_count > 0); + + // The manifest records the exported chunks under the configured prefix. + let paths: Vec<(String,)> = + sqlx::query_as("SELECT path FROM cala_persistent_outbox_archive_chunks") + .fetch_all(&pool) + .await?; + assert!(!paths.is_empty()); + assert!( + paths + .iter() + .all(|(path,)| path.starts_with("outbox-archive/cala/")), + "unexpected chunk paths: {paths:?}" + ); + assert!(!storage.list().is_empty()); + + // A listener resuming from the beginning is served from the archive + // first, then crosses into postgres mid-stream — day 0/1 events (now + // only in storage) and day 3 events (still in pg) all arrive, in + // contiguous sequence order. + let mut listener = cala.register_outbox_listener(Some(obix::EventSequence::BEGIN)); + let targets: HashSet = [ + journal.id().into(), + day0_account.id().into(), + day1_account.id().into(), + day3_account.id().into(), + ] + .into_iter() + .collect(); + let mut seen = HashSet::new(); + let mut expected_sequence = 1u64; + while seen.len() < targets.len() { + let event = tokio::time::timeout(std::time::Duration::from_secs(30), listener.next()) + .await + .expect("timed out waiting for replayed event") + .expect("stream ended during replay") + .expect("undecodable event during replay"); + assert_eq!( + u64::from(event.sequence), + expected_sequence, + "replay must be contiguous" + ); + expected_sequence += 1; + let id = match event.payload.as_ref() { + Some(outbox::OutboxEventPayload::JournalCreated { journal }) => { + Some(uuid::Uuid::from(journal.id)) + } + Some(outbox::OutboxEventPayload::AccountCreated { account }) => { + Some(uuid::Uuid::from(account.id)) + } + _ => None, + }; + if let Some(id) = id { + if targets.contains(&id) { + seen.insert(id); + } + } + } + assert_eq!(seen, targets); + + jobs.shutdown() + .await + .map_err(|e| anyhow::anyhow!(e.to_string()))?; + pool.close().await; + drop_database(&db_name).await?; + Ok(()) +}