use super::{
AedbConfig, AedbError, AedbInstance, CallerContext, ColumnDef, ColumnType, ConsistencyMode,
DdlOperation, Mutation, Permission, REACTIVE_ACK_CACHE_MAX_ENTRIES,
ReactiveCheckpointAckCacheKey, ReactiveCheckpointAckState,
};
use crate::catalog::SYSTEM_PROJECT_ID;
use tempfile::tempdir;
#[tokio::test]
async fn event_outbox_and_reactive_processor_checkpoint_lag_work() {
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");
db.commit(Mutation::Ddl(DdlOperation::CreateTable {
owner_id: Some("system".into()),
if_not_exists: true,
project_id: SYSTEM_PROJECT_ID.into(),
scope_id: "app".into(),
table_name: "reactive_processor_checkpoints".into(),
columns: vec![
ColumnDef {
name: "processor_name".into(),
col_type: ColumnType::Text,
nullable: false,
},
ColumnDef {
name: "checkpoint_seq".into(),
col_type: ColumnType::Integer,
nullable: false,
},
ColumnDef {
name: "updated_at_micros".into(),
col_type: ColumnType::Timestamp,
nullable: false,
},
],
primary_key: vec!["processor_name".into()],
}))
.await
.expect("create processor checkpoint table");
let first = db
.emit_event(
"arcana",
"app",
"hand_settled",
"hand_1".into(),
r#"{"user_id":"u1","wager":100,"pnl":-100}"#.into(),
)
.await
.expect("emit first event");
let second = db
.emit_event(
"arcana",
"app",
"hand_settled",
"hand_2".into(),
r#"{"user_id":"u2","wager":75,"pnl":50}"#.into(),
)
.await
.expect("emit second event");
let page = db
.read_event_stream(Some("hand_settled"), 0, 10, ConsistencyMode::AtLatest)
.await
.expect("read stream");
assert_eq!(page.events.len(), 2);
assert_eq!(page.events[0].event_key, "hand_1");
assert_eq!(page.events[1].event_key, "hand_2");
assert_eq!(page.next_commit_seq, Some(second.commit_seq));
db.ack_reactive_processor_checkpoint("points_processor", first.commit_seq)
.await
.expect("ack checkpoint");
let lag = db
.reactive_processor_lag("points_processor", ConsistencyMode::AtLatest)
.await
.expect("lag");
assert_eq!(lag.processor_name, "points_processor");
assert_eq!(lag.checkpoint_seq, first.commit_seq);
assert!(lag.head_seq >= second.commit_seq);
assert_eq!(
lag.lag_commits,
lag.head_seq.saturating_sub(first.commit_seq)
);
}
#[tokio::test]
async fn reactive_processor_checkpoint_ack_batches_by_watermark() {
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");
db.commit(Mutation::Ddl(DdlOperation::CreateTable {
owner_id: Some("system".into()),
if_not_exists: true,
project_id: SYSTEM_PROJECT_ID.into(),
scope_id: "app".into(),
table_name: "reactive_processor_checkpoints".into(),
columns: vec![
ColumnDef {
name: "processor_name".into(),
col_type: ColumnType::Text,
nullable: false,
},
ColumnDef {
name: "checkpoint_seq".into(),
col_type: ColumnType::Integer,
nullable: false,
},
ColumnDef {
name: "updated_at_micros".into(),
col_type: ColumnType::Timestamp,
nullable: false,
},
],
primary_key: vec!["processor_name".into()],
}))
.await
.expect("create processor checkpoint table");
let first = db
.emit_event(
"arcana",
"app",
"hand_settled",
"hand_1".into(),
r#"{"user_id":"u1","wager":100,"pnl":-100}"#.into(),
)
.await
.expect("emit first event");
let second = db
.emit_event(
"arcana",
"app",
"hand_settled",
"hand_2".into(),
r#"{"user_id":"u2","wager":75,"pnl":50}"#.into(),
)
.await
.expect("emit second event");
let persisted = db
.ack_reactive_processor_checkpoint_batched("points_processor", first.commit_seq, 2)
.await
.expect("first batched ack");
assert!(persisted.is_some(), "first ack should persist baseline");
let deferred = db
.ack_reactive_processor_checkpoint_batched("points_processor", first.commit_seq + 1, 2)
.await
.expect("deferred batched ack");
assert!(deferred.is_none(), "ack below watermark should be deferred");
let lag_after_defer = db
.reactive_processor_lag("points_processor", ConsistencyMode::AtLatest)
.await
.expect("lag after deferred ack");
assert_eq!(lag_after_defer.checkpoint_seq, first.commit_seq);
let persisted_after_watermark = db
.ack_reactive_processor_checkpoint_batched("points_processor", first.commit_seq + 2, 2)
.await
.expect("persist on watermark");
assert!(
persisted_after_watermark.is_some(),
"watermark crossing should persist checkpoint"
);
let lag = db
.reactive_processor_lag("points_processor", ConsistencyMode::AtLatest)
.await
.expect("lag");
assert_eq!(lag.processor_name, "points_processor");
assert_eq!(lag.checkpoint_seq, first.commit_seq + 2);
assert!(lag.head_seq >= second.commit_seq);
}
#[tokio::test]
async fn reactive_processor_checkpoint_ack_rejects_regression() {
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");
db.commit(Mutation::Ddl(DdlOperation::CreateTable {
owner_id: Some("system".into()),
if_not_exists: true,
project_id: SYSTEM_PROJECT_ID.into(),
scope_id: "app".into(),
table_name: "reactive_processor_checkpoints".into(),
columns: vec![
ColumnDef {
name: "processor_name".into(),
col_type: ColumnType::Text,
nullable: false,
},
ColumnDef {
name: "checkpoint_seq".into(),
col_type: ColumnType::Integer,
nullable: false,
},
ColumnDef {
name: "updated_at_micros".into(),
col_type: ColumnType::Timestamp,
nullable: false,
},
],
primary_key: vec!["processor_name".into()],
}))
.await
.expect("create processor checkpoint table");
let first = db
.emit_event(
"arcana",
"app",
"hand_settled",
"hand_1".into(),
r#"{"user_id":"u1","wager":100,"pnl":-100}"#.into(),
)
.await
.expect("emit first event");
let second = db
.emit_event(
"arcana",
"app",
"hand_settled",
"hand_2".into(),
r#"{"user_id":"u2","wager":75,"pnl":50}"#.into(),
)
.await
.expect("emit second event");
db.ack_reactive_processor_checkpoint("points_processor", second.commit_seq)
.await
.expect("persist newer checkpoint");
let err = db
.ack_reactive_processor_checkpoint("points_processor", first.commit_seq)
.await
.expect_err("checkpoint regression must fail");
assert!(matches!(err, AedbError::Validation(_)));
let lag = db
.reactive_processor_lag("points_processor", ConsistencyMode::AtLatest)
.await
.expect("lag");
assert_eq!(lag.checkpoint_seq, second.commit_seq);
}
#[tokio::test]
async fn reactive_processor_checkpoint_batched_rejects_stale_persist_after_concurrent_advance() {
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");
db.commit(Mutation::Ddl(DdlOperation::CreateTable {
owner_id: Some("system".into()),
if_not_exists: true,
project_id: SYSTEM_PROJECT_ID.into(),
scope_id: "app".into(),
table_name: "reactive_processor_checkpoints".into(),
columns: vec![
ColumnDef {
name: "processor_name".into(),
col_type: ColumnType::Text,
nullable: false,
},
ColumnDef {
name: "checkpoint_seq".into(),
col_type: ColumnType::Integer,
nullable: false,
},
ColumnDef {
name: "updated_at_micros".into(),
col_type: ColumnType::Timestamp,
nullable: false,
},
],
primary_key: vec!["processor_name".into()],
}))
.await
.expect("create processor checkpoint table");
let first = db
.emit_event(
"arcana",
"app",
"hand_settled",
"hand_1".into(),
r#"{"user_id":"u1","wager":100,"pnl":-100}"#.into(),
)
.await
.expect("emit first event");
let second = db
.emit_event(
"arcana",
"app",
"hand_settled",
"hand_2".into(),
r#"{"user_id":"u2","wager":75,"pnl":50}"#.into(),
)
.await
.expect("emit second event");
let stale = db
.build_reactive_processor_checkpoint_envelope("points_processor", first.commit_seq)
.await
.expect("stale envelope");
db.ack_reactive_processor_checkpoint("points_processor", second.commit_seq)
.await
.expect("persist newer checkpoint");
let err = db
.commit_envelope_prevalidated_internal("stale_reactive_ack_test", stale)
.await
.expect_err("stale asserted ack must fail");
assert!(matches!(err, AedbError::Conflict(_)));
let lag = db
.reactive_processor_lag("points_processor", ConsistencyMode::AtLatest)
.await
.expect("lag");
assert_eq!(lag.checkpoint_seq, second.commit_seq);
}
#[tokio::test]
async fn reactive_processor_checkpoint_batched_as_isolated_by_caller() {
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");
db.emit_event("arcana", "app", "bootstrap", "evt-1".into(), "{}".into())
.await
.expect("bootstrap system scope");
db.commit(Mutation::Ddl(DdlOperation::CreateTable {
owner_id: Some("system".into()),
if_not_exists: true,
project_id: SYSTEM_PROJECT_ID.into(),
scope_id: "app".into(),
table_name: "reactive_processor_checkpoints".into(),
columns: vec![
ColumnDef {
name: "processor_name".into(),
col_type: ColumnType::Text,
nullable: false,
},
ColumnDef {
name: "checkpoint_seq".into(),
col_type: ColumnType::Integer,
nullable: false,
},
ColumnDef {
name: "updated_at_micros".into(),
col_type: ColumnType::Timestamp,
nullable: false,
},
],
primary_key: vec!["processor_name".into()],
}))
.await
.expect("create processor checkpoint table");
db.ack_reactive_processor_checkpoint("bootstrap", 1)
.await
.expect("bootstrap processor checkpoint table");
for caller in ["alice", "bob"] {
db.commit(Mutation::Ddl(DdlOperation::GrantPermission {
actor_id: None,
delegable: false,
caller_id: caller.into(),
permission: Permission::TableWrite {
project_id: SYSTEM_PROJECT_ID.into(),
scope_id: "app".into(),
table_name: "reactive_processor_checkpoints".into(),
},
}))
.await
.expect("grant system table write");
}
let first = db
.ack_reactive_processor_checkpoint_batched_as(
CallerContext::new("alice"),
"shared_processor",
42,
100,
)
.await
.expect("alice ack");
assert!(first.is_some(), "alice baseline ack should persist");
let second = db
.ack_reactive_processor_checkpoint_batched_as(
CallerContext::new("bob"),
"shared_processor",
42,
100,
)
.await
.expect("bob ack");
assert!(
second.is_some(),
"bob baseline ack should persist independently"
);
}
#[tokio::test]
async fn reactive_processor_checkpoint_batched_as_does_not_poison_cache_on_permission_failure() {
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");
db.emit_event("arcana", "app", "bootstrap", "evt-1".into(), "{}".into())
.await
.expect("bootstrap system scope");
db.commit(Mutation::Ddl(DdlOperation::CreateTable {
owner_id: Some("system".into()),
if_not_exists: true,
project_id: SYSTEM_PROJECT_ID.into(),
scope_id: "app".into(),
table_name: "reactive_processor_checkpoints".into(),
columns: vec![
ColumnDef {
name: "processor_name".into(),
col_type: ColumnType::Text,
nullable: false,
},
ColumnDef {
name: "checkpoint_seq".into(),
col_type: ColumnType::Integer,
nullable: false,
},
ColumnDef {
name: "updated_at_micros".into(),
col_type: ColumnType::Timestamp,
nullable: false,
},
],
primary_key: vec!["processor_name".into()],
}))
.await
.expect("create processor checkpoint table");
db.ack_reactive_processor_checkpoint("bootstrap", 1)
.await
.expect("bootstrap processor checkpoint table");
let denied = db
.ack_reactive_processor_checkpoint_batched_as(
CallerContext::new("mallory"),
"perm_test_processor",
100,
100,
)
.await
.expect_err("mallory should be denied before grant");
assert!(matches!(denied, AedbError::PermissionDenied(_)));
db.commit(Mutation::Ddl(DdlOperation::GrantPermission {
actor_id: None,
delegable: false,
caller_id: "mallory".into(),
permission: Permission::TableWrite {
project_id: SYSTEM_PROJECT_ID.into(),
scope_id: "app".into(),
table_name: "reactive_processor_checkpoints".into(),
},
}))
.await
.expect("grant system table write");
let persisted = db
.ack_reactive_processor_checkpoint_batched_as(
CallerContext::new("mallory"),
"perm_test_processor",
100,
100,
)
.await
.expect("mallory should succeed after grant");
assert!(
persisted.is_some(),
"failed pre-grant attempt must not suppress later successful persist"
);
}
#[tokio::test]
async fn reactive_processor_checkpoint_batched_cache_is_bounded() {
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");
db.emit_event("arcana", "app", "bootstrap", "evt-1".into(), "{}".into())
.await
.expect("bootstrap system scope");
db.commit(Mutation::Ddl(DdlOperation::CreateTable {
owner_id: Some("system".into()),
if_not_exists: true,
project_id: SYSTEM_PROJECT_ID.into(),
scope_id: "app".into(),
table_name: "reactive_processor_checkpoints".into(),
columns: vec![
ColumnDef {
name: "processor_name".into(),
col_type: ColumnType::Text,
nullable: false,
},
ColumnDef {
name: "checkpoint_seq".into(),
col_type: ColumnType::Integer,
nullable: false,
},
ColumnDef {
name: "updated_at_micros".into(),
col_type: ColumnType::Timestamp,
nullable: false,
},
],
primary_key: vec!["processor_name".into()],
}))
.await
.expect("create processor checkpoint table");
{
let mut cache = db.reactive_processor_ack_watermarks.lock();
for i in 0..(REACTIVE_ACK_CACHE_MAX_ENTRIES + 200) {
cache.insert(
ReactiveCheckpointAckCacheKey {
processor_name: format!("cache-seed-{i}"),
caller_id: None,
},
ReactiveCheckpointAckState {
last_persisted_seq: i as u64,
last_touch_micros: i as u64,
},
);
}
}
let persisted = db
.ack_reactive_processor_checkpoint_batched("cache-boundary", 1, 1)
.await
.expect("ack should succeed");
assert!(persisted.is_some());
let cache_len = db.reactive_processor_ack_watermarks.lock().len();
assert!(
cache_len <= REACTIVE_ACK_CACHE_MAX_ENTRIES,
"cache should be pruned to cap; got {}",
cache_len
);
}