Retired Quota State retention from surviving user recreation

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<String>,
@@ -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,

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,

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(&quota_path, &retired_file).unwrap();
+        std::fs::create_dir(&quota_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(&quota_path).unwrap();
+        std::fs::rename(&retired_file, &quota_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;
+}

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;
 

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<tokio::sync::OwnedMutexGuard<()>> {
+        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
+    }
+}

src/quota_state/recreation_tests.rs

@@ -0,0 +1,111 @@
+use super::*;
+use std::sync::atomic::Ordering;
+
+async fn fixture() -> (tempfile::TempDir, Arc<QuotaStateOwner>) {
+    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"));
+}