use super::{
AedbConfig, AedbError, AedbInstance, CallerContext, ConsistencyMode, DdlOperation, Expr,
LifecycleEvent, LifecycleHook, Mutation, Permission, Query, QueryOptions,
ReactiveProcessorOptions, RecordingLifecycleHook, Value,
};
use crate::catalog::SYSTEM_PROJECT_ID;
use std::sync::Arc;
use std::time::{Duration, Instant};
use tempfile::tempdir;
#[tokio::test]
async fn event_stream_and_processor_lag_as_require_explicit_permissions() {
let dir = tempdir().expect("temp");
let db = AedbInstance::open_production(AedbConfig::production([3u8; 32]), dir.path())
.expect("open secure");
let caller = CallerContext::new("reader_1");
let stream_err = db
.read_event_stream_as(&caller, None, 0, 10, ConsistencyMode::AtLatest)
.await
.expect_err("event stream read without permission must fail");
assert!(matches!(stream_err, AedbError::PermissionDenied(_)));
let lag_err = db
.reactive_processor_lag_as(&caller, "points_processor", ConsistencyMode::AtLatest)
.await
.expect_err("reactive processor lag without permission must fail");
assert!(matches!(lag_err, AedbError::PermissionDenied(_)));
}
#[tokio::test]
async fn lifecycle_hooks_receive_post_commit_events_for_applied_ddl() {
let dir = tempdir().expect("temp");
let db = AedbInstance::open_anonymous(AedbConfig::default(), dir.path()).expect("open");
let events = Arc::new(std::sync::Mutex::new(Vec::new()));
let hook: Arc<dyn LifecycleHook> = Arc::new(RecordingLifecycleHook {
events: Arc::clone(&events),
});
db.add_lifecycle_hook(Arc::clone(&hook));
db.commit_ddl(DdlOperation::CreateProject {
owner_id: None,
project_id: "arcana".into(),
if_not_exists: true,
})
.await
.expect("create project");
db.commit_ddl(DdlOperation::CreateProject {
owner_id: None,
project_id: "arcana".into(),
if_not_exists: true,
})
.await
.expect("idempotent create");
let deadline = Instant::now() + Duration::from_secs(2);
while Instant::now() < deadline {
if !events.lock().expect("events lock").is_empty() {
break;
}
tokio::time::sleep(Duration::from_millis(10)).await;
}
let seen = events.lock().expect("events lock");
assert_eq!(seen.len(), 1);
assert!(matches!(
&seen[0],
LifecycleEvent::ProjectCreated { project_id, .. } if project_id == "arcana"
));
}
#[tokio::test]
async fn removing_last_lifecycle_hook_disables_dispatch() {
let dir = tempdir().expect("temp");
let db = AedbInstance::open_anonymous(AedbConfig::default(), dir.path()).expect("open");
let events = Arc::new(std::sync::Mutex::new(Vec::new()));
let hook: Arc<dyn LifecycleHook> = Arc::new(RecordingLifecycleHook {
events: Arc::clone(&events),
});
db.add_lifecycle_hook(Arc::clone(&hook));
db.remove_lifecycle_hook(&hook);
db.commit_ddl(DdlOperation::CreateProject {
owner_id: None,
project_id: "arcana".into(),
if_not_exists: true,
})
.await
.expect("create project");
tokio::time::sleep(Duration::from_millis(25)).await;
assert!(events.lock().expect("events lock").is_empty());
}
#[tokio::test]
async fn lifecycle_outbox_persists_applied_events() {
let dir = tempdir().expect("temp");
let db = AedbInstance::open_anonymous(AedbConfig::default(), dir.path()).expect("open");
let created = db
.commit_ddl(DdlOperation::CreateProject {
owner_id: None,
project_id: "arcana".into(),
if_not_exists: true,
})
.await
.expect("create project");
let outbox = db
.query_no_auth(
"_system",
"app",
Query::select(&["event_count", "events"])
.from("lifecycle_outbox")
.where_(Expr::Eq(
"commit_seq".into(),
Value::Integer(created.seq as i64),
))
.limit(1),
QueryOptions::default(),
)
.await
.expect("query lifecycle outbox");
assert_eq!(outbox.rows.len(), 1, "expected lifecycle outbox row");
assert_eq!(outbox.rows[0].values[0], Value::Integer(1));
let Value::Json(payload) = &outbox.rows[0].values[1] else {
panic!("expected json payload");
};
let events: Vec<LifecycleEvent> =
serde_json::from_str(payload.as_str()).expect("decode lifecycle payload");
assert!(matches!(
events.first(),
Some(LifecycleEvent::ProjectCreated { project_id, seq })
if project_id == "arcana" && *seq == created.seq
));
}
#[tokio::test]
async fn lifecycle_outbox_includes_app_emit_events() {
let dir = tempdir().expect("temp");
let db = AedbInstance::open_anonymous(AedbConfig::default(), dir.path()).expect("open");
db.create_project("arcana").await.expect("project");
db.create_scope("arcana", "app").await.expect("scope");
let committed = db
.emit_event(
"arcana",
"app",
"hand_settled",
"hand_1".into(),
r#"{"user_id":"u1","wager":100,"pnl":-25}"#.into(),
)
.await
.expect("emit");
let outbox = db
.query_no_auth(
"_system",
"app",
Query::select(&["events"])
.from("lifecycle_outbox")
.where_(Expr::Eq(
"commit_seq".into(),
Value::Integer(committed.commit_seq as i64),
))
.limit(1),
QueryOptions::default(),
)
.await
.expect("query lifecycle outbox");
assert_eq!(outbox.rows.len(), 1, "expected lifecycle outbox row");
let Value::Json(payload) = &outbox.rows[0].values[0] else {
panic!("expected json payload");
};
let events: Vec<LifecycleEvent> =
serde_json::from_str(payload.as_str()).expect("decode lifecycle payload");
assert!(events.iter().any(|evt| matches!(
evt,
LifecycleEvent::AppEventEmitted {
project_id,
scope_id,
topic,
event_key,
payload_json,
..
} if project_id == "arcana"
&& scope_id == "app"
&& topic == "hand_settled"
&& event_key == "hand_1"
&& payload_json.contains("\"user_id\":\"u1\"")
)));
}
#[tokio::test]
async fn reactive_processor_scheduler_runs_with_period_and_batch_limits() {
let dir = tempdir().expect("temp");
let db =
Arc::new(AedbInstance::open_anonymous(AedbConfig::default(), dir.path()).expect("open"));
db.create_project("arcana").await.expect("project");
db.create_scope("arcana", "app").await.expect("scope");
let seen_batches = Arc::new(std::sync::Mutex::new(Vec::<usize>::new()));
let seen_batches_handler = Arc::clone(&seen_batches);
db.start_reactive_processor(
"sched_processor",
ReactiveProcessorOptions {
caller_id: None,
topic_filter: Some("hand_settled".into()),
run_on_interval: false,
max_allowed_lag_commits: None,
max_allowed_stall_ms: None,
max_events_per_run: 2,
max_bytes_per_run: 1_000_000,
max_run_duration_ms: 250,
run_interval_ms: 10,
idle_backoff_ms: 10,
checkpoint_watermark_commits: 1,
max_retries: 3,
retry_backoff_ms: 5,
},
move |_db, events| {
let seen_batches_handler = Arc::clone(&seen_batches_handler);
async move {
seen_batches_handler
.lock()
.expect("batch lock")
.push(events.len());
Ok(())
}
},
)
.await
.expect("start processor");
let mut last_seq = 0u64;
for i in 0..5 {
let commit = db
.emit_event(
"arcana",
"app",
"hand_settled",
format!("hand-{i}"),
format!(r#"{{"hand":"{i}"}}"#),
)
.await
.expect("emit event");
last_seq = last_seq.max(commit.commit_seq);
}
let deadline = Instant::now() + Duration::from_secs(3);
loop {
let lag = db
.reactive_processor_lag("sched_processor", ConsistencyMode::AtLatest)
.await
.expect("lag");
if lag.checkpoint_seq >= last_seq {
break;
}
if Instant::now() >= deadline {
panic!("processor did not catch up before deadline");
}
tokio::time::sleep(Duration::from_millis(20)).await;
}
let status = db
.reactive_processor_runtime_status("sched_processor")
.await
.expect("runtime status");
assert!(status.processed_events_total >= 5);
let batches = seen_batches.lock().expect("batches").clone();
assert!(!batches.is_empty());
assert!(batches.iter().all(|size| *size <= 2));
db.stop_reactive_processor("sched_processor")
.await
.expect("stop processor");
assert!(
db.reactive_processor_runtime_status("sched_processor")
.await
.is_none()
);
}
#[tokio::test]
async fn reactive_processor_registry_persists_and_auto_resumes_on_handler_registration() {
let dir = tempdir().expect("temp");
let db =
Arc::new(AedbInstance::open_anonymous(AedbConfig::default(), dir.path()).expect("open"));
db.create_project("arcana").await.expect("project");
db.create_scope("arcana", "app").await.expect("scope");
let processed = Arc::new(std::sync::atomic::AtomicU64::new(0));
let processed_first = Arc::clone(&processed);
db.start_reactive_processor(
"resume_processor",
ReactiveProcessorOptions {
topic_filter: Some("hand_settled".into()),
checkpoint_watermark_commits: 1,
..ReactiveProcessorOptions::default()
},
move |_db, events| {
let processed_first = Arc::clone(&processed_first);
async move {
processed_first
.fetch_add(events.len() as u64, std::sync::atomic::Ordering::Relaxed);
Ok(())
}
},
)
.await
.expect("start processor");
let first = db
.emit_event("arcana", "app", "hand_settled", "h1".into(), "{}".into())
.await
.expect("emit first");
let deadline = Instant::now() + Duration::from_secs(2);
loop {
let lag = db
.reactive_processor_lag("resume_processor", ConsistencyMode::AtLatest)
.await
.expect("lag");
if lag.checkpoint_seq >= first.commit_seq {
break;
}
if Instant::now() >= deadline {
panic!("processor failed to checkpoint first event");
}
tokio::time::sleep(Duration::from_millis(20)).await;
}
db.shutdown().await.expect("graceful shutdown");
drop(db);
let db2 =
Arc::new(AedbInstance::open_anonymous(AedbConfig::default(), dir.path()).expect("reopen"));
let processed_resume = Arc::clone(&processed);
let resumed = db2
.register_reactive_processor_handler("resume_processor", move |_db, events| {
let processed = Arc::clone(&processed_resume);
async move {
processed.fetch_add(events.len() as u64, std::sync::atomic::Ordering::Relaxed);
Ok(())
}
})
.await
.expect("register handler");
assert!(resumed, "enabled processor should auto-resume");
let second = db2
.emit_event("arcana", "app", "hand_settled", "h2".into(), "{}".into())
.await
.expect("emit second");
let deadline = Instant::now() + Duration::from_secs(2);
loop {
let lag = db2
.reactive_processor_lag("resume_processor", ConsistencyMode::AtLatest)
.await
.expect("lag after resume");
if lag.checkpoint_seq >= second.commit_seq {
break;
}
if Instant::now() >= deadline {
panic!("processor failed to checkpoint second event after resume");
}
tokio::time::sleep(Duration::from_millis(20)).await;
}
db2.stop_reactive_processor("resume_processor")
.await
.expect("stop resumed");
assert!(
processed.load(std::sync::atomic::Ordering::Relaxed) >= 2,
"expected both events processed"
);
}
#[tokio::test]
async fn reactive_processor_scheduler_retries_and_then_succeeds() {
let dir = tempdir().expect("temp");
let db =
Arc::new(AedbInstance::open_anonymous(AedbConfig::default(), dir.path()).expect("open"));
db.create_project("arcana").await.expect("project");
db.create_scope("arcana", "app").await.expect("scope");
let attempts = Arc::new(std::sync::atomic::AtomicU64::new(0));
let attempts_handler = Arc::clone(&attempts);
db.start_reactive_processor(
"retry_processor",
ReactiveProcessorOptions {
caller_id: None,
topic_filter: Some("hand_settled".into()),
run_on_interval: false,
max_allowed_lag_commits: None,
max_allowed_stall_ms: None,
checkpoint_watermark_commits: 1,
max_events_per_run: 16,
max_bytes_per_run: 1_000_000,
max_run_duration_ms: 250,
run_interval_ms: 10,
idle_backoff_ms: 10,
max_retries: 3,
retry_backoff_ms: 10,
},
move |_db, _events| {
let attempts_handler = Arc::clone(&attempts_handler);
async move {
let n = attempts_handler.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
if n == 0 {
return Err(AedbError::Validation("injected retryable failure".into()));
}
Ok(())
}
},
)
.await
.expect("start retry processor");
let evt = db
.emit_event(
"arcana",
"app",
"hand_settled",
"retry-hand".into(),
"{}".into(),
)
.await
.expect("emit");
let deadline = Instant::now() + Duration::from_secs(3);
loop {
let lag = db
.reactive_processor_lag("retry_processor", ConsistencyMode::AtLatest)
.await
.expect("lag");
if lag.checkpoint_seq >= evt.commit_seq {
break;
}
if Instant::now() >= deadline {
panic!("retry processor did not checkpoint");
}
tokio::time::sleep(Duration::from_millis(20)).await;
}
let status = db
.reactive_processor_runtime_status("retry_processor")
.await
.expect("status");
assert!(status.retries_total >= 1, "expected at least one retry");
assert_eq!(status.dead_lettered_total, 0);
db.stop_reactive_processor("retry_processor")
.await
.expect("stop");
}
#[tokio::test]
async fn reactive_processor_scheduler_dead_letters_after_retry_exhaustion() {
let dir = tempdir().expect("temp");
let db =
Arc::new(AedbInstance::open_anonymous(AedbConfig::default(), dir.path()).expect("open"));
db.create_project("arcana").await.expect("project");
db.create_scope("arcana", "app").await.expect("scope");
db.start_reactive_processor(
"dlq_processor",
ReactiveProcessorOptions {
caller_id: None,
topic_filter: Some("hand_settled".into()),
run_on_interval: false,
max_allowed_lag_commits: None,
max_allowed_stall_ms: None,
checkpoint_watermark_commits: 1,
max_events_per_run: 16,
max_bytes_per_run: 1_000_000,
max_run_duration_ms: 250,
run_interval_ms: 10,
idle_backoff_ms: 10,
max_retries: 1,
retry_backoff_ms: 10,
},
move |_db, _events| async move { Err(AedbError::Validation("permanent fail".into())) },
)
.await
.expect("start dlq processor");
let evt = db
.emit_event(
"arcana",
"app",
"hand_settled",
"dlq-hand".into(),
"{}".into(),
)
.await
.expect("emit");
let deadline = Instant::now() + Duration::from_secs(4);
loop {
let lag = db
.reactive_processor_lag("dlq_processor", ConsistencyMode::AtLatest)
.await
.expect("lag");
if lag.checkpoint_seq >= evt.commit_seq {
break;
}
if Instant::now() >= deadline {
panic!("dlq processor did not advance checkpoint");
}
tokio::time::sleep(Duration::from_millis(20)).await;
}
let dlq = db
.query_no_auth(
SYSTEM_PROJECT_ID,
"app",
Query::select(&["processor_name", "event_key", "attempts"])
.from("reactive_processor_dead_letters")
.where_(Expr::Eq(
"processor_name".into(),
Value::Text("dlq_processor".into()),
))
.limit(10),
QueryOptions::default(),
)
.await
.expect("query dlq");
assert_eq!(dlq.rows.len(), 1);
assert_eq!(dlq.rows[0].values[1], Value::Text("dlq-hand".into()));
assert_eq!(dlq.rows[0].values[2], Value::Integer(2));
let status = db
.reactive_processor_runtime_status("dlq_processor")
.await
.expect("status");
assert!(status.dead_lettered_total >= 1);
db.stop_reactive_processor("dlq_processor")
.await
.expect("stop");
}
#[tokio::test]
async fn reactive_processor_scheduler_requires_caller_id_in_secure_mode() {
let dir = tempdir().expect("temp");
let db = Arc::new(
AedbInstance::open_secure(AedbConfig::production([11u8; 32]), dir.path())
.expect("open secure"),
);
let err = db
.start_reactive_processor(
"secure_proc",
ReactiveProcessorOptions {
caller_id: None,
..ReactiveProcessorOptions::default()
},
move |_db, _events| async move { Ok(()) },
)
.await
.expect_err("secure mode should require caller_id");
assert!(matches!(err, AedbError::PermissionDenied(_)));
}
#[tokio::test]
async fn reactive_processor_scheduler_uses_explicit_caller_permissions() {
let dir = tempdir().expect("temp");
let db = Arc::new(
AedbInstance::open_secure(AedbConfig::production([12u8; 32]), dir.path())
.expect("open secure"),
);
let system = CallerContext::system_internal();
db.commit_as(
system.clone(),
Mutation::Ddl(DdlOperation::CreateProject {
owner_id: None,
if_not_exists: true,
project_id: "arcana".into(),
}),
)
.await
.expect("create project");
db.commit_as(
system.clone(),
Mutation::Ddl(DdlOperation::CreateScope {
owner_id: None,
if_not_exists: true,
project_id: "arcana".into(),
scope_id: "app".into(),
}),
)
.await
.expect("create scope");
db.commit_as(
system.clone(),
Mutation::Ddl(DdlOperation::GrantPermission {
actor_id: Some("system".into()),
delegable: false,
caller_id: "proc_sched".into(),
permission: Permission::TableWrite {
project_id: SYSTEM_PROJECT_ID.into(),
scope_id: "app".into(),
table_name: "reactive_processor_registry".into(),
},
}),
)
.await
.expect("grant registry write");
let processed = Arc::new(std::sync::atomic::AtomicU64::new(0));
let processed_handler = Arc::clone(&processed);
db.start_reactive_processor(
"secure_sched",
ReactiveProcessorOptions {
caller_id: Some("proc_sched".into()),
topic_filter: Some("hand_settled".into()),
checkpoint_watermark_commits: 1,
run_interval_ms: 10,
idle_backoff_ms: 10,
max_retries: 0,
..ReactiveProcessorOptions::default()
},
move |_db, events| {
let processed_handler = Arc::clone(&processed_handler);
async move {
processed_handler
.fetch_add(events.len() as u64, std::sync::atomic::Ordering::Relaxed);
Ok(())
}
},
)
.await
.expect("start processor");
let deadline = Instant::now() + Duration::from_secs(2);
loop {
let status = db
.reactive_processor_runtime_status("secure_sched")
.await
.expect("status");
if status.failures_total > 0 {
break;
}
if Instant::now() >= deadline {
panic!("expected scheduler permission failures before grants");
}
tokio::time::sleep(Duration::from_millis(20)).await;
}
for permission in [
Permission::TableRead {
project_id: SYSTEM_PROJECT_ID.into(),
scope_id: "app".into(),
table_name: "event_outbox".into(),
},
Permission::TableRead {
project_id: SYSTEM_PROJECT_ID.into(),
scope_id: "app".into(),
table_name: "reactive_processor_checkpoints".into(),
},
Permission::TableWrite {
project_id: SYSTEM_PROJECT_ID.into(),
scope_id: "app".into(),
table_name: "reactive_processor_checkpoints".into(),
},
] {
db.commit_as(
system.clone(),
Mutation::Ddl(DdlOperation::GrantPermission {
actor_id: Some("system".into()),
delegable: false,
caller_id: "proc_sched".into(),
permission,
}),
)
.await
.expect("grant processor permission");
}
let evt = db
.emit_event_as(
system,
"arcana",
"app",
"hand_settled",
"secure-hand".into(),
"{}".into(),
)
.await
.expect("emit secure event");
let proc_caller = CallerContext::new("proc_sched");
let deadline = Instant::now() + Duration::from_secs(3);
loop {
let lag = db
.reactive_processor_lag_as(&proc_caller, "secure_sched", ConsistencyMode::AtLatest)
.await
.expect("lag");
if lag.checkpoint_seq >= evt.commit_seq {
break;
}
if Instant::now() >= deadline {
panic!("processor failed to progress after grants");
}
tokio::time::sleep(Duration::from_millis(20)).await;
}
assert!(
processed.load(std::sync::atomic::Ordering::Relaxed) >= 1,
"expected at least one processed event"
);
db.stop_reactive_processor("secure_sched")
.await
.expect("stop processor");
}
#[tokio::test]
async fn reactive_processor_pause_resume_list_and_health_work() {
let dir = tempdir().expect("temp");
let db =
Arc::new(AedbInstance::open_anonymous(AedbConfig::default(), dir.path()).expect("open"));
db.create_project("arcana").await.expect("project");
db.create_scope("arcana", "app").await.expect("scope");
db.start_reactive_processor(
"lifecycle_processor",
ReactiveProcessorOptions {
caller_id: None,
topic_filter: Some("hand_settled".into()),
checkpoint_watermark_commits: 1,
run_interval_ms: 10,
idle_backoff_ms: 10,
..ReactiveProcessorOptions::default()
},
move |_db, _events| async move { Ok(()) },
)
.await
.expect("start");
let first = db
.emit_event(
"arcana",
"app",
"hand_settled",
"life-1".into(),
"{}".into(),
)
.await
.expect("emit first");
let deadline = Instant::now() + Duration::from_secs(3);
loop {
let lag = db
.reactive_processor_lag("lifecycle_processor", ConsistencyMode::AtLatest)
.await
.expect("lag");
if lag.checkpoint_seq >= first.commit_seq {
break;
}
if Instant::now() >= deadline {
panic!("lifecycle processor did not checkpoint first event");
}
tokio::time::sleep(Duration::from_millis(20)).await;
}
let listed = db
.list_reactive_processors(ConsistencyMode::AtLatest)
.await
.expect("list processors");
let row = listed
.into_iter()
.find(|r| r.processor_name == "lifecycle_processor")
.expect("processor listed");
assert!(row.enabled);
assert!(row.running);
let health = db
.reactive_processor_health("lifecycle_processor", ConsistencyMode::AtLatest)
.await
.expect("health");
assert!(health.running);
assert!(health.enabled);
assert!(health.processed_events_total >= 1);
assert!(health.last_run_completed_micros.is_some());
db.pause_reactive_processor("lifecycle_processor")
.await
.expect("pause");
let listed = db
.list_reactive_processors(ConsistencyMode::AtLatest)
.await
.expect("list processors paused");
let row = listed
.into_iter()
.find(|r| r.processor_name == "lifecycle_processor")
.expect("processor listed paused");
assert!(!row.enabled);
assert!(!row.running);
let paused_health = db
.reactive_processor_health("lifecycle_processor", ConsistencyMode::AtLatest)
.await
.expect("paused health");
assert!(!paused_health.running);
assert!(!paused_health.enabled);
db.resume_reactive_processor("lifecycle_processor")
.await
.expect("resume");
let second = db
.emit_event(
"arcana",
"app",
"hand_settled",
"life-2".into(),
"{}".into(),
)
.await
.expect("emit second");
let deadline = Instant::now() + Duration::from_secs(3);
loop {
let lag = db
.reactive_processor_lag("lifecycle_processor", ConsistencyMode::AtLatest)
.await
.expect("lag resumed");
if lag.checkpoint_seq >= second.commit_seq {
break;
}
if Instant::now() >= deadline {
panic!("lifecycle processor did not checkpoint second event");
}
tokio::time::sleep(Duration::from_millis(20)).await;
}
db.stop_reactive_processor("lifecycle_processor")
.await
.expect("stop");
}
#[tokio::test]
async fn reactive_processor_resume_requires_registered_handler() {
let dir = tempdir().expect("temp");
let db =
Arc::new(AedbInstance::open_anonymous(AedbConfig::default(), dir.path()).expect("open"));
db.create_project("arcana").await.expect("project");
db.create_scope("arcana", "app").await.expect("scope");
db.start_reactive_processor(
"resume_requires_handler",
ReactiveProcessorOptions {
caller_id: None,
topic_filter: Some("hand_settled".into()),
..ReactiveProcessorOptions::default()
},
move |_db, _events| async move { Ok(()) },
)
.await
.expect("start");
db.pause_reactive_processor("resume_requires_handler")
.await
.expect("pause");
db.shutdown().await.expect("shutdown");
drop(db);
let db2 =
Arc::new(AedbInstance::open_anonymous(AedbConfig::default(), dir.path()).expect("reopen"));
let err = db2
.resume_reactive_processor("resume_requires_handler")
.await
.expect_err("resume without handler should fail");
assert!(matches!(err, AedbError::Validation(_)));
}
#[tokio::test]
async fn reactive_processor_scheduler_can_run_periodically_without_events() {
let dir = tempdir().expect("temp");
let db =
Arc::new(AedbInstance::open_anonymous(AedbConfig::default(), dir.path()).expect("open"));
db.create_project("arcana").await.expect("project");
db.create_scope("arcana", "app").await.expect("scope");
let ticks = Arc::new(std::sync::atomic::AtomicU64::new(0));
let ticks_handler = Arc::clone(&ticks);
db.start_reactive_processor(
"periodic_processor",
ReactiveProcessorOptions {
caller_id: None,
topic_filter: Some("never_emitted".into()),
run_on_interval: true,
run_interval_ms: 10,
idle_backoff_ms: 10,
max_retries: 0,
..ReactiveProcessorOptions::default()
},
move |_db, events| {
let ticks_handler = Arc::clone(&ticks_handler);
async move {
assert!(events.is_empty(), "periodic run should be empty-batch");
ticks_handler.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
Ok(())
}
},
)
.await
.expect("start periodic processor");
let deadline = Instant::now() + Duration::from_secs(2);
loop {
let n = ticks.load(std::sync::atomic::Ordering::Relaxed);
if n >= 3 {
break;
}
if Instant::now() >= deadline {
panic!("periodic processor did not run enough ticks");
}
tokio::time::sleep(Duration::from_millis(20)).await;
}
db.stop_reactive_processor("periodic_processor")
.await
.expect("stop periodic");
}
#[tokio::test]
async fn reactive_processor_slo_status_and_enforcement_detect_breaches() {
let dir = tempdir().expect("temp");
let db =
Arc::new(AedbInstance::open_anonymous(AedbConfig::default(), dir.path()).expect("open"));
db.create_project("arcana").await.expect("project");
db.create_scope("arcana", "app").await.expect("scope");
db.start_reactive_processor(
"slo_breach_processor",
ReactiveProcessorOptions {
caller_id: None,
topic_filter: Some("hand_settled".into()),
run_on_interval: false,
max_allowed_lag_commits: Some(0),
max_allowed_stall_ms: None,
run_interval_ms: 60_000,
idle_backoff_ms: 60_000,
..ReactiveProcessorOptions::default()
},
move |_db, _events| async move { Ok(()) },
)
.await
.expect("start processor");
db.emit_event(
"arcana",
"app",
"hand_settled",
"slo-hand".into(),
"{}".into(),
)
.await
.expect("emit");
tokio::time::sleep(Duration::from_millis(20)).await;
let status = db
.reactive_processor_slo_status("slo_breach_processor", ConsistencyMode::AtLatest)
.await
.expect("slo status");
assert!(status.breached, "expected lag SLO breach");
assert!(!status.reasons.is_empty());
let statuses = db
.list_reactive_processor_slo_statuses(ConsistencyMode::AtLatest)
.await
.expect("list slo statuses");
assert!(
statuses
.iter()
.any(|s| s.processor_name == "slo_breach_processor" && s.breached)
);
let enforce = db
.enforce_reactive_processor_slos(ConsistencyMode::AtLatest)
.await;
assert!(matches!(enforce, Err(AedbError::Unavailable { .. })));
db.stop_reactive_processor("slo_breach_processor")
.await
.expect("stop");
}
#[tokio::test]
async fn reactive_processor_slo_status_healthy_when_within_thresholds() {
let dir = tempdir().expect("temp");
let db =
Arc::new(AedbInstance::open_anonymous(AedbConfig::default(), dir.path()).expect("open"));
db.create_project("arcana").await.expect("project");
db.create_scope("arcana", "app").await.expect("scope");
let ticks = Arc::new(std::sync::atomic::AtomicU64::new(0));
let ticks_handler = Arc::clone(&ticks);
db.start_reactive_processor(
"slo_ok_processor",
ReactiveProcessorOptions {
caller_id: None,
topic_filter: Some("none".into()),
run_on_interval: true,
max_allowed_lag_commits: Some(10),
max_allowed_stall_ms: Some(5_000),
run_interval_ms: 10,
idle_backoff_ms: 10,
..ReactiveProcessorOptions::default()
},
move |_db, _events| {
let ticks_handler = Arc::clone(&ticks_handler);
async move {
ticks_handler.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
Ok(())
}
},
)
.await
.expect("start");
let deadline = Instant::now() + Duration::from_secs(2);
loop {
let n = ticks.load(std::sync::atomic::Ordering::Relaxed);
if n >= 2 {
break;
}
if Instant::now() >= deadline {
panic!("processor did not tick");
}
tokio::time::sleep(Duration::from_millis(20)).await;
}
let status = db
.reactive_processor_slo_status("slo_ok_processor", ConsistencyMode::AtLatest)
.await
.expect("slo status");
assert!(!status.breached, "expected no SLO breach");
db.enforce_reactive_processor_slos(ConsistencyMode::AtLatest)
.await
.expect("slo enforce should pass");
db.stop_reactive_processor("slo_ok_processor")
.await
.expect("stop");
}
#[tokio::test]
async fn emit_event_respects_max_event_payload_bytes() {
let dir = tempdir().expect("temp");
let config = AedbConfig {
max_event_payload_bytes: 16,
..AedbConfig::default()
};
let db = AedbInstance::open_anonymous(config, dir.path()).expect("open");
db.create_project("p").await.expect("project");
db.create_scope("p", "app").await.expect("scope");
let err = db
.emit_event(
"p",
"app",
"topic",
"event-1".into(),
"{\"payload\":\"1234567890\"}".into(),
)
.await
.expect_err("event payload should be bounded");
assert!(
matches!(err, AedbError::Validation(ref msg) if msg.contains("max_event_payload_bytes"))
);
}