From dd6196a62e9a8949b2cdee6ea04cf01c6a3a46f9 Mon Sep 17 00:00:00 2001 From: Alexey <247128645+axkurcom@users.noreply.github.com> Date: Wed, 21 Oct 2026 21:06:51 +0300 Subject: [PATCH] Retired Quota State retention from surviving user recreation --- src/api/users/create.rs | 7 ++ src/api/users/period_tests.rs | 2 + src/api/users/period_tests/recreation.rs | 112 +++++++++++++++++++++++ src/quota_state.rs | 4 + src/quota_state/recreation.rs | 24 +++++ src/quota_state/recreation_tests.rs | 111 ++++++++++++++++++++++ 6 files changed, 260 insertions(+) create mode 100644 src/api/users/period_tests/recreation.rs create mode 100644 src/quota_state/recreation.rs create mode 100644 src/quota_state/recreation_tests.rs diff --git a/src/api/users/create.rs b/src/api/users/create.rs index be33008..cebf8d2 100644 --- a/src/api/users/create.rs +++ b/src/api/users/create.rs @@ -1,5 +1,6 @@ use super::*; +/// Completes an accepted user creation even if the requesting client disconnects. pub(in crate::api) async fn create_user( body: CreateUserRequest, expected_revision: Option, @@ -147,6 +148,12 @@ async fn create_user_to_completion( touched_sections.push(AccessSection::UserEnabled); } + // Clear any retired durable identity before the username can survive restart. + let quota_guard = shared + .quota_state + .prepare_user_creation_locked(&body.username, quota_guard) + .await + .map_err(|error| ApiFailure::internal(format!("failed to prepare user quota: {error}")))?; let revision = save_access_sections_to_disk_if_revision( &shared.config_path, &cfg, diff --git a/src/api/users/period_tests.rs b/src/api/users/period_tests.rs index bb6b09d..1ef0000 100644 --- a/src/api/users/period_tests.rs +++ b/src/api/users/period_tests.rs @@ -13,6 +13,8 @@ use tokio::sync::{Mutex, watch}; // Period-only mutations must fence already attached accounts as well as new ones. mod freshness; +// Recreation must remove retired durable state before committing a new account. +mod recreation; struct Fixture { _dir: tempfile::TempDir, diff --git a/src/api/users/period_tests/recreation.rs b/src/api/users/period_tests/recreation.rs new file mode 100644 index 0000000..23b59b2 --- /dev/null +++ b/src/api/users/period_tests/recreation.rs @@ -0,0 +1,112 @@ +use super::*; + +#[tokio::test] +async fn quota_period_failed_delete_recreate_restart_must_not_inherit_retired_calendar() { + for secret in ["0123456789abcdef0123456789abcdef", "fedcba9876543210fedcba9876543210"] { + quota_period_delete_recreate_restart(true, secret).await; + } +} + +#[tokio::test] +async fn quota_period_successful_delete_recreate_restart_has_fresh_calendar() { + for secret in ["0123456789abcdef0123456789abcdef", "fedcba9876543210fedcba9876543210"] { + quota_period_delete_recreate_restart(false, secret).await; + } +} + +async fn quota_period_delete_recreate_restart(fail_deletion: bool, secret: &str) { + let fixture = Fixture::new(); + let create = |secret: &str| serde_json::from_value(serde_json::json!({ + "username":"alice", "secret":secret, + "data_quota_bytes":1024, "quota_period":"monthly" + })).unwrap(); + create_user(create("0123456789abcdef0123456789abcdef"), None, &fixture.shared).await.unwrap(); + fixture.publish().await; + let now = crate::quota_state::now_epoch_secs(); + // Model a previously processed future boundary followed by clock rollback. + let next = UserQuotaPeriod::Monthly.next(now).unwrap(); + let future = UserQuotaPeriod::Monthly.next(next).unwrap(); + fixture.shared.quota_state.process_schedules(future, false).await.unwrap(); + fixture.store.current_handle("alice").unwrap().charge(700); + fixture.shared.quota_state.save(&std::collections::BTreeSet::from(["alice".into(), "admin".into()])).await.unwrap(); + let old_checkpoint = fixture.store.current_handle("alice").unwrap().schedule_checkpoint(); + let quota_path = fixture.shared.quota_state.path().to_owned(); + let retired_file = quota_path.with_extension("retired"); + if fail_deletion { + std::fs::rename("a_path, &retired_file).unwrap(); + std::fs::create_dir("a_path).unwrap(); + } + // Config deletion succeeds but the quota file cannot replace a directory. + delete_user("alice", None, &fixture.shared).await.unwrap(); + assert!(fixture.store.current_handle("alice").is_none()); + if fail_deletion { + let config_before = std::fs::read(&fixture.shared.config_path).unwrap(); + assert!(create_user(create(secret), None, &fixture.shared).await.is_err()); + assert_eq!(std::fs::read(&fixture.shared.config_path).unwrap(), config_before); + assert!(fixture.store.current_handle("alice").is_none()); + // Restore the old durable bytes to model storage recovery before retry. + std::fs::remove_dir("a_path).unwrap(); + std::fs::rename(&retired_file, "a_path).unwrap(); + } + create_user(create(secret), None, &fixture.shared).await.unwrap(); + assert_eq!(fixture.store.used("alice"), 0); + assert_eq!(fixture.store.current_handle("alice").unwrap().schedule_checkpoint().period_cursor_epoch_secs, 0); + let cfg = Arc::new(ProxyConfig::load(&fixture.shared.config_path).unwrap()); + let restart_store = Arc::new(QuotaStore::default()); + let restart = QuotaStateOwner::new(quota_path, restart_store.clone()); + restart.load_for_config(&cfg).await.unwrap(); + let authority = UserAdmissionAuthority::new_with_quota_store(restart_store.clone()); + let gate = restart.lock().await; + authority.activate_config_source(1, None, &cfg.access.users, &cfg.access.user_enabled).unwrap(); + authority.publish_quota_config(1, &cfg, true); + drop(gate); + restart.process_schedules(now, true).await.unwrap(); + let actual = restart_store.current_handle("alice").unwrap().schedule_checkpoint(); + assert!(actual.period_cursor_epoch_secs < old_checkpoint.period_cursor_epoch_secs); + assert_eq!(restart_store.schedule_deadline(), Some(next), "recreated account inherited a retired calendar watermark across restart"); + assert_eq!(restart_store.used("alice"), 0, "recreated account inherited retired usage"); + fixture.finish().await; +} + +#[tokio::test] +async fn quota_period_create_rejects_unreadable_checkpoint_without_publishing_user() { + for corrupt in [false, true] { + let fixture = Fixture::new(); + let path = fixture.shared.quota_state.path(); + if corrupt { + std::fs::write(path, b"invalid checkpoint").unwrap(); + } else { + std::fs::create_dir(path).unwrap(); + } + let before = std::fs::read(&fixture.shared.config_path).unwrap(); + let result = create_user( + serde_json::from_value(serde_json::json!({ + "username":"alice", "data_quota_bytes":1024, "quota_period":"monthly" + })).unwrap(), + None, + &fixture.shared, + ).await; + assert!(result.is_err(), "CREATE must reject untrusted durable quota state"); + assert_eq!(std::fs::read(&fixture.shared.config_path).unwrap(), before); + assert!(fixture.store.current_handle("alice").is_none()); + fixture.finish().await; + } +} + +#[tokio::test] +async fn quota_period_create_revision_conflict_does_not_touch_retired_checkpoint() { + let fixture = Fixture::new(); + let retired = b"{\"last_reset_epoch_secs\":123,\"users\":{\"alice\":{\"used_bytes\":700,\"last_reset_epoch_secs\":123}}}"; + std::fs::write(fixture.shared.quota_state.path(), retired).unwrap(); + let before = std::fs::read(&fixture.shared.config_path).unwrap(); + let result = create_user( + serde_json::from_value(serde_json::json!({"username":"alice"})).unwrap(), + Some("outdated-revision".into()), + &fixture.shared, + ).await; + assert_eq!(result.err().unwrap().code, "revision_conflict"); + assert_eq!(std::fs::read(&fixture.shared.config_path).unwrap(), before); + assert_eq!(std::fs::read(fixture.shared.quota_state.path()).unwrap(), retired); + assert!(fixture.store.current_handle("alice").is_none()); + fixture.finish().await; +} diff --git a/src/quota_state.rs b/src/quota_state.rs index fdd2265..58af1a5 100644 --- a/src/quota_state.rs +++ b/src/quota_state.rs @@ -11,6 +11,10 @@ use crate::stats::{QuotaScheduleCheckpoint, QuotaStore, UserQuotaSnapshot}; // UTC period transactions share the existing bounded durable checkpoint owner. mod periods; +// Account creation must retire durable state before publishing new credentials. +mod recreation; +#[cfg(test)] +mod recreation_tests; #[cfg(test)] mod test_io; diff --git a/src/quota_state/recreation.rs b/src/quota_state/recreation.rs new file mode 100644 index 0000000..235710f --- /dev/null +++ b/src/quota_state/recreation.rs @@ -0,0 +1,24 @@ +use super::*; + +impl QuotaStateOwner { + /// Removes an absent account's old checkpoint before CREATE commits its config. + /// Returns the same gate so config persistence and admission remain serialized. + pub(crate) async fn prepare_user_creation_locked( + &self, + user: &str, + guard: tokio::sync::OwnedMutexGuard<()>, + ) -> std::io::Result> { + let Some(mut state) = read_state_file(&self.path).await? else { + return Ok(guard); + }; + state.users.remove(user); + // Recommit even an absent row: a previous rename may have succeeded while + // directory sync failed. CREATE needs a confirmed durable barrier on retry. + let path = self.path.clone(); + let task = tokio::task::spawn_blocking(move || { + write_state_file_blocking(&path, &state)?; + Ok(guard) + }); + wait_for_blocking_io(task).await + } +} diff --git a/src/quota_state/recreation_tests.rs b/src/quota_state/recreation_tests.rs new file mode 100644 index 0000000..b32e396 --- /dev/null +++ b/src/quota_state/recreation_tests.rs @@ -0,0 +1,111 @@ +use super::*; +use std::sync::atomic::Ordering; + +async fn fixture() -> (tempfile::TempDir, Arc) { + let dir = tempfile::tempdir().unwrap(); + let owner = QuotaStateOwner::new(dir.path().join("quota.json"), Arc::new(QuotaStore::default())); + let mut state = QuotaStateFile::default(); + state.last_reset_epoch_secs = 123; + state.users.insert("alice".into(), quota_user_state(UserQuotaSnapshot { + used_bytes: 700, + last_reset_epoch_secs: 123, + })); + state.users.insert("bob".into(), QuotaUserState { + used_bytes: 321, + last_reset_epoch_secs: 111, + schedule: QuotaScheduleCheckpoint { + quota_period: Some(crate::config::UserQuotaPeriod::Monthly), + period_cursor_epoch_secs: 1_793_491_200, + }, + }); + write_state_file(owner.path(), state).await.unwrap(); + (dir, owner) +} + +#[tokio::test] +async fn creation_removes_only_retired_row_and_returns_owned_gate() { + let (_dir, owner) = fixture().await; + let before = read_state_file(owner.path()).await.unwrap().unwrap(); + let guard = owner.lock().await; + let guard = owner.prepare_user_creation_locked("alice", guard).await.unwrap(); + assert!(owner.mutation.try_lock().is_err()); + let after = read_state_file(owner.path()).await.unwrap().unwrap(); + assert!(!after.users.contains_key("alice")); + assert_eq!(serde_json::to_value(&after.users["bob"]).unwrap(), + serde_json::to_value(&before.users["bob"]).unwrap()); + assert_eq!(after.last_reset_epoch_secs, before.last_reset_epoch_secs); + drop(guard); + assert!(owner.mutation.try_lock().is_ok()); +} + +#[tokio::test] +async fn creation_without_checkpoint_does_not_require_a_write() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("missing/quota.json"); + let owner = QuotaStateOwner::new(path.clone(), Arc::new(QuotaStore::default())); + let guard = owner.lock().await; + let _guard = owner.prepare_user_creation_locked("alice", guard).await.unwrap(); + assert!(!path.exists()); + assert!(owner.mutation.try_lock().is_err()); +} + +#[tokio::test] +async fn creation_retries_durability_after_ambiguous_removal() { + let (_dir, owner) = fixture().await; + let (hook, _) = test_io::Hook::install(owner.path(), false); + hook.fail_after.store(true, Ordering::Relaxed); + let guard = owner.lock().await; + assert!(owner.prepare_user_creation_locked("alice", guard).await.is_err()); + assert!(!read_state_file(owner.path()).await.unwrap().unwrap().users.contains_key("alice")); + // The absent row does not prove that the previous directory sync succeeded. + let guard = owner.lock().await; + assert!(owner.prepare_user_creation_locked("alice", guard).await.is_err()); + assert_eq!(hook.writes.load(Ordering::Relaxed), 2); + hook.fail_after.store(false, Ordering::Relaxed); + let guard = owner.lock().await; + let _guard = owner.prepare_user_creation_locked("alice", guard).await.unwrap(); + assert_eq!(hook.writes.load(Ordering::Relaxed), 3); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn creation_waiter_abort_keeps_gate_until_durable_removal_finishes() { + let (_dir, owner) = fixture().await; + let (hook, started) = test_io::Hook::install(owner.path(), true); + let operation = owner.clone(); + let waiter = tokio::spawn(async move { + let guard = operation.lock().await; + operation.prepare_user_creation_locked("alice", guard).await + }); + started.await.unwrap(); + waiter.abort(); + assert!(waiter.await.unwrap_err().is_cancelled()); + assert!(owner.mutation.try_lock().is_err()); + assert!(read_state_file(owner.path()).await.unwrap().unwrap().users.contains_key("alice")); + hook.release(); + let _guard = tokio::time::timeout(std::time::Duration::from_secs(2), owner.lock()).await.unwrap(); + assert!(!read_state_file(owner.path()).await.unwrap().unwrap().users.contains_key("alice")); + assert_eq!(hook.writes.load(Ordering::Relaxed), 1); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn creation_write_failure_keeps_retired_checkpoint_for_safe_retry() { + let (_dir, owner) = fixture().await; + let retired = owner.path().with_extension("retired"); + let (hook, started) = test_io::Hook::install(owner.path(), true); + let operation = owner.clone(); + let writer = tokio::spawn(async move { + let guard = operation.lock().await; + operation.prepare_user_creation_locked("alice", guard).await + }); + started.await.unwrap(); + std::fs::rename(owner.path(), &retired).unwrap(); + std::fs::create_dir(owner.path()).unwrap(); + hook.release(); + assert!(writer.await.unwrap().is_err()); + std::fs::remove_dir(owner.path()).unwrap(); + std::fs::rename(retired, owner.path()).unwrap(); + assert!(read_state_file(owner.path()).await.unwrap().unwrap().users.contains_key("alice")); + let guard = owner.lock().await; + let _guard = owner.prepare_user_creation_locked("alice", guard).await.unwrap(); + assert!(!read_state_file(owner.path()).await.unwrap().unwrap().users.contains_key("alice")); +}