#[path = "support/logs.rs"]
mod logs;
#[path = "support/postgres_fixture.rs"]
mod postgres_fixture;
use postgres_fixture::{Product, count, database, fixture};
use baukit_erasure::{
ErasureError, ErasurePreflight, ErasureService, ErasureState, IdentityDeletionError,
IdentityDeletionHandler, IdentityRetention, PostgresErasureStore, reject_fenced_subject,
};
use baukit_jobs::{JobStore, PostgresJobStore, WorkerConfig, WorkerRunner};
use baukit_runtime::ShutdownToken;
use baukit_test::FakeIdentityAccountDeleter;
use std::{error::Error, sync::Arc, time::Duration};
use tracing::instrument::WithSubscriber as _;
#[tokio::test]
#[ignore = "requires Docker PostgreSQL"]
async fn preflight_proceeds_with_read_only_connections_and_no_subject_lock()
-> Result<(), Box<dyn Error>> {
let (_container, pool, store) = database().await?;
let read_only_pool = sqlx::postgres::PgPoolOptions::new()
.max_connections(1)
.after_connect(|connection, _| {
Box::pin(async move {
sqlx::query("SET default_transaction_read_only = on")
.execute(connection)
.await?;
Ok(())
})
})
.connect(_container.connection_url())
.await?;
let read_only_store = PostgresErasureStore::new(
read_only_pool.clone(),
baukit_config::Secret::new("test-key-with-at-least-32-bytes-of-entropy".into()),
)?;
let service = ErasureService::new(read_only_store, IdentityRetention::Retain)?;
let mut transaction = pool.begin().await?;
store.guard_subject(&mut transaction, "alice", None).await?;
assert_eq!(
tokio::time::timeout(
Duration::from_secs(1),
service.preflight("alice", None, "preflight-proceed-01"),
)
.await??,
ErasurePreflight::Proceed
);
transaction.rollback().await?;
assert_eq!(count(&pool, "profiles").await?, 2);
assert_eq!(count(&pool, "erasure_operations").await?, 0);
assert_eq!(count(&pool, "erasure_fences").await?, 0);
assert_eq!(count(&pool, "job_outbox").await?, 0);
read_only_pool.close().await;
Ok(())
}
#[tokio::test]
#[ignore = "requires Docker PostgreSQL"]
async fn preflight_replays_each_stored_outcome_before_checking_the_fence()
-> Result<(), Box<dyn Error>> {
let (_container, pool, store, fake, service) = fixture().await?;
fake.push_outcome(Err(IdentityDeletionError::Retryable));
let mut outcome = service
.erase("alice", None, "preflight-replay-01", &Product)
.await?;
assert!(store.is_fenced("alice", None).await?);
for state in [
ErasureState::Pending,
ErasureState::Failed,
ErasureState::Completed,
] {
outcome.status = state;
outcome.completed_at = (state == ErasureState::Completed).then(chrono::Utc::now);
sqlx::query("UPDATE erasure_operations SET state = $2, response = $3, completed_at = $4 WHERE id = $1")
.bind(outcome.operation_id)
.bind(serde_json::to_value(state)?.as_str().ok_or("state")?)
.bind(serde_json::to_value(&outcome)?)
.bind(outcome.completed_at)
.execute(&pool)
.await?;
assert_eq!(
service
.preflight("alice", None, "preflight-replay-01")
.await?,
ErasurePreflight::Replay(outcome.clone())
);
assert_eq!(
store.status("alice", outcome.operation_id).await?,
Some(outcome.clone())
);
}
assert_eq!(count(&pool, "erasure_operations").await?, 1);
assert_eq!(count(&pool, "erasure_fences").await?, 1);
assert_eq!(fake.calls(), ["alice"]);
Ok(())
}
#[tokio::test]
#[ignore = "requires Docker PostgreSQL"]
async fn preflight_conflict_precedes_receipt_decoding_and_the_subject_fence()
-> Result<(), Box<dyn Error>> {
let (_container, pool, store) = database().await?;
let service = ErasureService::new(store.clone(), IdentityRetention::Retain)?;
let alice = service
.erase("alice", None, "preflight-alice-01", &Product)
.await?;
service
.erase("bob", None, "preflight-bob-key-01", &Product)
.await?;
assert!(store.is_fenced("bob", None).await?);
sqlx::query("UPDATE erasure_operations SET response = '{}'::jsonb WHERE id = $1")
.bind(alice.operation_id)
.execute(&pool)
.await?;
assert!(matches!(
service.preflight("bob", None, "preflight-alice-01").await,
Err(ErasureError::Conflict)
));
assert!(matches!(
service.preflight("alice", None, "preflight-alice-01").await,
Err(ErasureError::Receipt(_))
));
assert_eq!(count(&pool, "erasure_operations").await?, 2);
Ok(())
}
#[tokio::test]
#[ignore = "requires Docker PostgreSQL"]
async fn erase_rechecks_replay_and_fence_after_preflight_proceeds() -> Result<(), Box<dyn Error>> {
let (_container, pool, store) = database().await?;
let service = ErasureService::new(store, IdentityRetention::Retain)?;
for key in ["preflight-race-key-01", "preflight-race-key-02"] {
assert_eq!(
service.preflight("alice", None, key).await?,
ErasurePreflight::Proceed
);
}
let first = service
.erase("alice", None, "preflight-race-key-01", &Product)
.await?;
assert_eq!(
service
.erase("alice", None, "preflight-race-key-01", &Product)
.await?,
first
);
assert!(matches!(
service
.erase("alice", None, "preflight-race-key-02", &Product)
.await,
Err(ErasureError::ProfileErased)
));
assert_eq!(count(&pool, "profiles").await?, 1);
assert_eq!(count(&pool, "erasure_operations").await?, 1);
Ok(())
}
#[tokio::test]
#[ignore = "requires Docker PostgreSQL"]
async fn retained_identity_completes_replays_and_fences_without_a_job() -> Result<(), Box<dyn Error>>
{
let (_container, pool, store) = database().await?;
let service = ErasureService::new(store.clone(), IdentityRetention::Retain)?;
let first = service
.erase("alice", None, "retain-replay-key-01", &Product)
.await?;
assert_eq!(first.status, ErasureState::Completed);
assert_eq!(first.status_code(), axum::http::StatusCode::OK);
assert!(first.completed_at.is_some());
let row: (String, chrono::DateTime<chrono::Utc>, serde_json::Value) = sqlx::query_as(
"SELECT state, completed_at, response FROM erasure_operations WHERE id = $1",
)
.bind(first.operation_id)
.fetch_one(&pool)
.await?;
assert_eq!(row.0, "completed");
assert_eq!(
row.1.timestamp_micros(),
first.completed_at.expect("completed at").timestamp_micros()
);
assert_eq!(row.2, serde_json::to_value(&first)?);
assert_eq!(count(&pool, "profiles").await?, 1);
assert_eq!(count(&pool, "erasure_operations").await?, 1);
assert_eq!(count(&pool, "erasure_fences").await?, 1);
assert_eq!(count(&pool, "job_outbox").await?, 0);
assert!(store.is_fenced("alice", None).await?);
assert!(!store.is_fenced("bob", None).await?);
assert_eq!(
store.status("alice", first.operation_id).await?,
Some(first.clone())
);
assert!(store.status("bob", first.operation_id).await?.is_none());
assert!(store.status("alice", uuid::Uuid::now_v7()).await?.is_none());
sqlx::raw_sql("CREATE FUNCTION reject_product_delete() RETURNS trigger LANGUAGE plpgsql AS $$ BEGIN RAISE EXCEPTION 'product erasure must not run on replay'; END; $$; CREATE TRIGGER reject_product_delete BEFORE DELETE ON profiles FOR EACH STATEMENT EXECUTE FUNCTION reject_product_delete();").execute(&pool).await?;
assert_eq!(
service
.erase("alice", None, "retain-replay-key-01", &Product)
.await?,
first
);
assert!(matches!(
service
.erase("bob", None, "retain-replay-key-01", &Product)
.await,
Err(ErasureError::Conflict)
));
assert!(matches!(
service
.erase("alice", None, "retain-second-key-01", &Product)
.await,
Err(ErasureError::ProfileErased)
));
assert!(matches!(
reject_fenced_subject(&store, "alice", None).await,
Err(ErasureError::ProfileErased)
));
let mut transaction = pool.begin().await?;
assert!(matches!(
store.guard_subject(&mut transaction, "alice", None).await,
Err(ErasureError::ProfileErased)
));
transaction.rollback().await?;
assert_eq!(count(&pool, "erasure_operations").await?, 1);
assert_eq!(count(&pool, "job_outbox").await?, 0);
Ok(())
}
#[tokio::test]
#[ignore = "requires Docker PostgreSQL"]
async fn retained_identity_completion_failure_rolls_back_product_receipt_and_fence()
-> Result<(), Box<dyn Error>> {
let (_container, pool, store) = database().await?;
let service = ErasureService::new(store.clone(), IdentityRetention::Retain)?;
sqlx::raw_sql("CREATE FUNCTION reject_completion() RETURNS trigger LANGUAGE plpgsql AS $$ BEGIN RAISE EXCEPTION 'injected completion failure'; END; $$; CREATE TRIGGER reject_completion BEFORE UPDATE ON erasure_operations FOR EACH ROW EXECUTE FUNCTION reject_completion();").execute(&pool).await?;
assert!(matches!(
service
.erase("alice", None, "retain-rollback-key-01", &Product)
.await,
Err(ErasureError::Database(_))
));
assert_eq!(count(&pool, "profiles").await?, 2);
assert_eq!(count(&pool, "erasure_operations").await?, 0);
assert_eq!(count(&pool, "erasure_fences").await?, 0);
assert_eq!(count(&pool, "job_outbox").await?, 0);
assert!(!store.is_fenced("alice", None).await?);
Ok(())
}
#[tokio::test]
#[ignore = "requires Docker PostgreSQL"]
async fn retained_identity_validates_keys_and_subject_before_erasure() -> Result<(), Box<dyn Error>>
{
let (_container, pool, store) = database().await?;
let service = ErasureService::new(store, IdentityRetention::Retain)?;
for key in [
"short".to_owned(),
"a".repeat(129),
"invalid key with space".to_owned(),
"invalid-key-0000\n".to_owned(),
] {
assert!(matches!(
service.erase("alice", None, &key, &Product).await,
Err(ErasureError::InvalidKey)
));
}
assert!(matches!(
service
.erase("", None, "retain-valid-key-01", &Product)
.await,
Err(ErasureError::Configuration)
));
assert_eq!(count(&pool, "profiles").await?, 2);
assert_eq!(count(&pool, "erasure_operations").await?, 0);
assert_eq!(count(&pool, "erasure_fences").await?, 0);
assert_eq!(count(&pool, "job_outbox").await?, 0);
Ok(())
}
#[tokio::test]
#[ignore = "requires Docker PostgreSQL"]
async fn retained_identity_concurrent_same_key_returns_the_same_completed_receipt()
-> Result<(), Box<dyn Error>> {
let (_container, pool, store) = database().await?;
let service = ErasureService::new(store, IdentityRetention::Retain)?;
let barrier = Arc::new(tokio::sync::Barrier::new(9));
let mut tasks = Vec::new();
for _ in 0..8 {
let service = service.clone();
let barrier = barrier.clone();
tasks.push(tokio::spawn(async move {
barrier.wait().await;
service
.erase("alice", None, "retain-concurrent-01", &Product)
.await
}));
}
barrier.wait().await;
let mut receipt = None;
for task in tasks {
let outcome = task.await??;
assert_eq!(outcome.status, ErasureState::Completed);
assert!(outcome.completed_at.is_some());
if let Some(first) = &receipt {
assert_eq!(&outcome, first);
}
receipt = Some(outcome);
}
assert_eq!(count(&pool, "profiles").await?, 1);
assert_eq!(count(&pool, "erasure_operations").await?, 1);
assert_eq!(count(&pool, "erasure_fences").await?, 1);
assert_eq!(count(&pool, "job_outbox").await?, 0);
Ok(())
}
#[tokio::test]
#[ignore = "requires Docker PostgreSQL"]
async fn atomic_rollback_after_product_callback() -> Result<(), Box<dyn Error>> {
let (_container, pool, store, fake, service) = fixture().await?;
sqlx::raw_sql("CREATE FUNCTION reject_enqueue() RETURNS trigger LANGUAGE plpgsql AS $$ BEGIN RAISE EXCEPTION 'injected failure'; END; $$; CREATE TRIGGER reject_enqueue BEFORE INSERT ON job_outbox FOR EACH ROW EXECUTE FUNCTION reject_enqueue();").execute(&pool).await?;
assert!(
service
.erase("alice", None, "atomic-key-0000001", &Product)
.await
.is_err()
);
assert_eq!(count(&pool, "profiles").await?, 2);
assert_eq!(count(&pool, "erasure_operations").await?, 0);
assert_eq!(count(&pool, "erasure_fences").await?, 0);
assert_eq!(count(&pool, "job_outbox").await?, 0);
assert!(!store.is_fenced("alice", None).await?);
assert!(fake.calls().is_empty());
Ok(())
}
#[tokio::test]
#[ignore = "requires Docker PostgreSQL"]
async fn completed_replay_conflict_and_fence() -> Result<(), Box<dyn Error>> {
let (_container, pool, store, fake, service) = fixture().await?;
let first = service
.erase("alice", None, "replay-key-0000001", &Product)
.await?;
assert_eq!(first.status, ErasureState::Completed);
assert_eq!(
service
.erase("alice", None, "replay-key-0000001", &Product)
.await?,
first
);
assert!(matches!(
service
.erase("bob", None, "replay-key-0000001", &Product)
.await,
Err(ErasureError::Conflict)
));
assert!(matches!(
service
.erase("alice", None, "different-key-0001", &Product)
.await,
Err(ErasureError::ProfileErased)
));
assert!(store.status("bob", first.operation_id).await?.is_none());
assert_eq!(
store.status("alice", first.operation_id).await?,
Some(first)
);
let mut transaction = pool.begin().await?;
assert!(matches!(
store.guard_subject(&mut transaction, "alice", None).await,
Err(ErasureError::ProfileErased)
));
assert_eq!(count(&pool, "profiles").await?, 1);
assert_eq!(count(&pool, "job_outbox").await?, 0);
assert_eq!(fake.calls(), ["alice"]);
Ok(())
}
#[tokio::test]
#[ignore = "requires Docker PostgreSQL"]
async fn concurrent_same_key_only_erases_once() -> Result<(), Box<dyn Error>> {
let (_container, pool, _store, fake, service) = fixture().await?;
let barrier = Arc::new(tokio::sync::Barrier::new(9));
let mut tasks = Vec::new();
for _ in 0..8 {
let service = service.clone();
let barrier = barrier.clone();
tasks.push(tokio::spawn(async move {
barrier.wait().await;
service
.erase("alice", None, "concurrent-key-001", &Product)
.await
}));
}
barrier.wait().await;
let mut operation = None;
for task in tasks {
let outcome = task.await??;
if let Some(id) = operation {
assert_eq!(outcome.operation_id, id);
}
operation = Some(outcome.operation_id);
}
assert_eq!(count(&pool, "erasure_operations").await?, 1);
assert_eq!(count(&pool, "erasure_fences").await?, 1);
assert_eq!(fake.calls(), ["alice"]);
Ok(())
}
async fn run_until(
store: PostgresErasureStore,
fake: Arc<FakeIdentityAccountDeleter>,
subject: &str,
operation: uuid::Uuid,
expected: ErasureState,
) -> Result<(), Box<dyn Error>> {
let runner = WorkerRunner::new(
Arc::new(PostgresJobStore::new(store.pool().clone())),
Arc::new(IdentityDeletionHandler::new(
store.clone(),
"test".into(),
fake,
)),
WorkerConfig {
poll_interval: Duration::from_millis(10),
retry_initial: Duration::from_millis(10),
retry_max: Duration::from_millis(10),
..WorkerConfig::default()
},
)?;
let shutdown = ShutdownToken::new(Duration::from_secs(1));
let worker = tokio::spawn(runner.run(shutdown.clone()));
let result = tokio::time::timeout(Duration::from_secs(5), async {
loop {
if store
.status(subject, operation)
.await?
.is_some_and(|receipt| receipt.status == expected)
{
break Ok::<_, ErasureError>(());
}
tokio::time::sleep(Duration::from_millis(10)).await;
}
})
.await;
shutdown.trigger();
worker.await??;
result??;
Ok(())
}
#[tokio::test]
#[ignore = "requires Docker PostgreSQL"]
async fn worker_retries_completes_and_removes_subject_payload() -> Result<(), Box<dyn Error>> {
let (_container, pool, store, fake, service) = fixture().await?;
fake.push_outcome(Err(IdentityDeletionError::Retryable));
fake.push_outcome(Err(IdentityDeletionError::Retryable));
let pending = service
.erase("alice", None, "worker-retry-key-01", &Product)
.await?;
assert_eq!(pending.status, ErasureState::Pending);
assert_eq!(count(&pool, "profiles").await?, 1);
assert_eq!(count(&pool, "job_outbox").await?, 1);
assert!(store.is_fenced("alice", None).await?);
assert_eq!(
service
.erase("alice", None, "worker-retry-key-01", &Product)
.await?,
pending
);
run_until(
store.clone(),
fake.clone(),
"alice",
pending.operation_id,
ErasureState::Completed,
)
.await?;
assert_eq!(count(&pool, "job_outbox").await?, 0);
assert_eq!(fake.calls(), ["alice", "alice", "alice"]);
let raw: i64 = sqlx::query_scalar(
"SELECT count(*) FROM erasure_operations WHERE response::text LIKE '%alice%'",
)
.fetch_one(&pool)
.await?;
assert_eq!(raw, 0);
Ok(())
}
#[tokio::test]
#[ignore = "requires Docker PostgreSQL"]
async fn permanent_and_exhausted_failures_remain_rerunnable() -> Result<(), Box<dyn Error>> {
for failure in [
IdentityDeletionError::Permanent,
IdentityDeletionError::Retryable,
] {
let (_container, pool, store, fake, service) = fixture().await?;
for _ in 0..3 {
fake.push_outcome(Err(failure));
}
let pending = service
.erase("alice", None, "failed-worker-key-01", &Product)
.await?;
run_until(
store.clone(),
fake.clone(),
"alice",
pending.operation_id,
ErasureState::Failed,
)
.await?;
assert!(store.is_fenced("alice", None).await?);
let state: String = sqlx::query_scalar("SELECT status FROM job_outbox")
.fetch_one(&pool)
.await?;
assert_eq!(state, "failed");
let calls_before_replay = fake.calls().len();
let replay = service
.erase("alice", None, "failed-worker-key-01", &Product)
.await?;
assert_eq!(replay.status, ErasureState::Failed);
assert_eq!(replay.operation_id, pending.operation_id);
assert_eq!(replay.status_code(), axum::http::StatusCode::OK);
assert_eq!(fake.calls().len(), calls_before_replay);
assert_eq!(count(&pool, "erasure_operations").await?, 1);
if failure == IdentityDeletionError::Permanent {
assert_eq!(fake.calls().len(), 2);
} else {
assert_eq!(fake.calls().len(), 3);
}
let restored_fake = Arc::new(FakeIdentityAccountDeleter::default());
sqlx::query("UPDATE job_outbox SET status = 'pending', attempts = 0, failure_reason = NULL, run_after = now()").execute(&pool).await?;
run_until(
store,
restored_fake.clone(),
"alice",
pending.operation_id,
ErasureState::Completed,
)
.await?;
assert_eq!(count(&pool, "job_outbox").await?, 0);
assert_eq!(restored_fake.calls(), ["alice"]);
}
Ok(())
}
#[tokio::test]
#[ignore = "requires Docker PostgreSQL"]
async fn final_expired_lease_fails_operation() -> Result<(), Box<dyn Error>> {
let (_container, pool, store, fake, service) = fixture().await?;
fake.push_outcome(Err(IdentityDeletionError::Retryable));
let pending = service
.erase("alice", None, "expired-lease-key-01", &Product)
.await?;
sqlx::query("UPDATE job_outbox SET status = 'running', attempts = max_attempts, locked_by = 'dead-worker', locked_until = now() - interval '1 second'").execute(&pool).await?;
PostgresJobStore::new(pool)
.claim(
"replacement",
&[baukit_erasure::IDENTITY_DELETE_JOB_TYPE],
chrono::Utc::now(),
Duration::from_secs(5),
)
.await?;
assert_eq!(
store
.status("alice", pending.operation_id)
.await?
.expect("receipt")
.status,
ErasureState::Failed
);
Ok(())
}
#[tokio::test]
#[ignore = "requires Docker PostgreSQL"]
async fn inline_completion_database_failure_returns_durable_acceptance()
-> Result<(), Box<dyn Error>> {
let (_container, pool, store, fake, service) = fixture().await?;
sqlx::raw_sql("CREATE FUNCTION reject_completion() RETURNS trigger LANGUAGE plpgsql AS $$ BEGIN RAISE EXCEPTION 'injected receipt failure'; END; $$; CREATE TRIGGER reject_completion BEFORE UPDATE ON erasure_operations FOR EACH ROW EXECUTE FUNCTION reject_completion();").execute(&pool).await?;
let logs = logs::Logs::default();
let pending = service
.erase("alice", None, "inline-db-failure-01", &Product)
.with_subscriber(logs.subscriber())
.await?;
let text = logs.text();
assert!(text.contains(&pending.operation_id.to_string()), "{text}");
assert!(text.contains("error_class=\"database\""), "{text}");
assert!(!text.contains("alice"), "{text}");
assert_eq!(pending.status, ErasureState::Pending);
assert_eq!(count(&pool, "profiles").await?, 1);
assert_eq!(count(&pool, "job_outbox").await?, 1);
assert!(store.is_fenced("alice", None).await?);
assert_eq!(fake.calls(), ["alice"]);
sqlx::query("DROP TRIGGER reject_completion ON erasure_operations")
.execute(&pool)
.await?;
run_until(
store,
fake.clone(),
"alice",
pending.operation_id,
ErasureState::Completed,
)
.await?;
assert_eq!(count(&pool, "job_outbox").await?, 0);
assert_eq!(fake.calls(), ["alice", "alice"]);
Ok(())
}
#[tokio::test]
#[ignore = "requires Docker PostgreSQL"]
async fn inline_failures_log_operation_and_class_without_subjects() -> Result<(), Box<dyn Error>> {
let (_container, _pool, _store, fake, service) = fixture().await?;
for (subject, key, error, class) in [
(
"alice",
"warn-retry-key-001",
IdentityDeletionError::Retryable,
"retryable",
),
(
"bob",
"warn-permanent-001",
IdentityDeletionError::Permanent,
"permanent",
),
] {
fake.push_outcome(Err(error));
let logs = logs::Logs::default();
let outcome = service
.erase(subject, None, key, &Product)
.with_subscriber(logs.subscriber())
.await?;
assert_eq!(outcome.status, ErasureState::Pending);
let text = logs.text();
assert!(text.contains(&outcome.operation_id.to_string()), "{text}");
assert!(text.contains(&format!("error_class=\"{class}\"")), "{text}");
assert!(!text.contains(subject), "{text}");
assert!(!text.contains(key), "{text}");
}
Ok(())
}
async fn assert_session_checks(
store: &PostgresErasureStore,
subject: &str,
authenticated_at: Option<chrono::DateTime<chrono::Utc>>,
rejected: bool,
) -> Result<(), Box<dyn Error>> {
assert_eq!(store.is_fenced(subject, authenticated_at).await?, rejected);
let service = ErasureService::new(store.clone(), IdentityRetention::Retain)?;
let result = service
.preflight(subject, authenticated_at, "preflight-fence-key-01")
.await;
if rejected {
assert!(matches!(result, Err(ErasureError::ProfileErased)));
} else {
assert_eq!(result?, ErasurePreflight::Proceed);
}
let result = reject_fenced_subject(store, subject, authenticated_at).await;
if rejected {
assert!(matches!(result, Err(ErasureError::ProfileErased)));
} else {
result?;
}
let mut transaction = store.pool().begin().await?;
let result = store
.guard_subject(&mut transaction, subject, authenticated_at)
.await;
if rejected {
assert!(matches!(result, Err(ErasureError::ProfileErased)));
} else {
result?;
}
transaction.rollback().await?;
Ok(())
}
#[tokio::test]
#[ignore = "requires Docker PostgreSQL"]
async fn all_fence_checks_require_a_later_login_and_every_operation_completed()
-> Result<(), Box<dyn Error>> {
let (_container, pool, store, fake, service) = fixture().await?;
for authenticated_at in [None, chrono::DateTime::from_timestamp(0, 0)] {
assert_session_checks(&store, "alice", authenticated_at, false).await?;
}
fake.push_outcome(Err(IdentityDeletionError::Retryable));
let operation = service
.erase("alice", None, "fence-matrix-key-01", &Product)
.await?;
assert_eq!(operation.status, ErasureState::Pending);
let erased_at: chrono::DateTime<chrono::Utc> =
sqlx::query_scalar("SELECT erased_at FROM erasure_fences WHERE subject_hash = $1")
.bind(store.subject_hash("alice"))
.fetch_one(&pool)
.await?;
let sessions = [
(None, true),
(Some(erased_at - chrono::Duration::microseconds(1)), true),
(Some(erased_at), true),
(Some(erased_at + chrono::Duration::microseconds(1)), false),
];
for state in ["pending", "failed", "completed"] {
sqlx::query("UPDATE erasure_operations SET state = $2, completed_at = CASE WHEN $2 = 'completed' THEN now() ELSE NULL END WHERE id = $1")
.bind(operation.operation_id).bind(state).execute(&pool).await?;
for (authenticated_at, rejected_after_completion) in sessions {
assert_session_checks(
&store,
"alice",
authenticated_at,
state != "completed" || rejected_after_completion,
)
.await?;
assert_session_checks(&store, "bob", authenticated_at, false).await?;
}
}
sqlx::query("INSERT INTO erasure_operations (id, subject_hash, key_hash, state, response) SELECT $1, subject_hash, $2, 'failed', response FROM erasure_operations WHERE id = $3")
.bind(uuid::Uuid::now_v7()).bind(vec![42_u8; 32]).bind(operation.operation_id).execute(&pool).await?;
assert_session_checks(&store, "alice", sessions[3].0, true).await?;
Ok(())
}
#[tokio::test]
#[ignore = "requires Docker PostgreSQL"]
async fn reonboarding_can_erase_again_and_old_keys_keep_their_receipts()
-> Result<(), Box<dyn Error>> {
let (_container, pool, store) = database().await?;
let service = ErasureService::new(store.clone(), IdentityRetention::Retain)?;
let first = service
.erase("alice", None, "first-erasure-key-01", &Product)
.await?;
let first_fence: chrono::DateTime<chrono::Utc> =
sqlx::query_scalar("SELECT erased_at FROM erasure_fences WHERE subject_hash = $1")
.bind(store.subject_hash("alice"))
.fetch_one(&pool)
.await?;
let fresh: chrono::DateTime<chrono::Utc> = sqlx::query_scalar("SELECT clock_timestamp()")
.fetch_one(&pool)
.await?;
assert!(fresh > first_fence);
let mut transaction = pool.begin().await?;
store
.guard_subject(&mut transaction, "alice", Some(fresh))
.await?;
sqlx::query("INSERT INTO profiles VALUES ('alice')")
.execute(&mut *transaction)
.await?;
transaction.commit().await?;
let second = service
.erase("alice", Some(fresh), "second-erasure-key-01", &Product)
.await?;
assert_ne!(first.operation_id, second.operation_id);
assert_eq!(second.status, ErasureState::Completed);
assert_eq!(count(&pool, "profiles").await?, 1);
assert_eq!(count(&pool, "erasure_operations").await?, 2);
assert_eq!(count(&pool, "erasure_fences").await?, 1);
let second_fence: chrono::DateTime<chrono::Utc> =
sqlx::query_scalar("SELECT erased_at FROM erasure_fences WHERE subject_hash = $1")
.bind(store.subject_hash("alice"))
.fetch_one(&pool)
.await?;
assert!(second_fence > fresh);
assert_session_checks(&store, "alice", Some(fresh), true).await?;
let newer: chrono::DateTime<chrono::Utc> = sqlx::query_scalar("SELECT clock_timestamp()")
.fetch_one(&pool)
.await?;
assert_session_checks(&store, "alice", Some(newer), false).await?;
for authenticated_at in [None, Some(fresh), Some(newer)] {
assert_eq!(
service
.erase("alice", authenticated_at, "first-erasure-key-01", &Product)
.await?,
first
);
assert_eq!(
service
.erase("alice", authenticated_at, "second-erasure-key-01", &Product)
.await?,
second
);
}
let unchanged: chrono::DateTime<chrono::Utc> =
sqlx::query_scalar("SELECT erased_at FROM erasure_fences WHERE subject_hash = $1")
.bind(store.subject_hash("alice"))
.fetch_one(&pool)
.await?;
assert_eq!(unchanged, second_fence);
Ok(())
}