Rustfmt

src/quota_state/period_tests.rs

@@ -0,0 +1,297 @@
+use super::*;
+use crate::config::{ProxyConfig, UserQuotaPeriod};
+use crate::proxy::user_admission::{UserAdmissionAuthority, credential_id_from_hex};
+
+const SECRET: &str = "0123456789abcdef0123456789abcdef";
+const DAY: u64 = 86400;
+const START: u64 = 1_735_689_600;
+
+fn config(period: Option<UserQuotaPeriod>) -> Arc<ProxyConfig> {
+    let mut config = ProxyConfig::default();
+    config.access.users.clear();
+    config.access.users.insert("alice".into(), SECRET.into());
+    config.access.user_data_quota.insert("alice".into(), 1024);
+    if let Some(period) = period {
+        config
+            .access
+            .user_quota_periods
+            .insert("alice".into(), period);
+    }
+    Arc::new(config)
+}
+
+fn request(store: &QuotaStore, config: &Arc<ProxyConfig>, now: u64) {
+    let target = store.current_handle("alice").unwrap();
+    store.request_schedule(
+        "alice",
+        target.incarnation(),
+        credential_id_from_hex(&config.access.users["alice"]).unwrap(),
+        config,
+        now,
+        true,
+    );
+}
+
+fn fixture() -> (tempfile::TempDir, Arc<QuotaStore>, Arc<QuotaStateOwner>) {
+    let dir = tempfile::tempdir().unwrap();
+    let store = Arc::new(QuotaStore::default());
+    store.activate_fresh("alice", 1);
+    let owner = QuotaStateOwner::new(dir.path().join("quota.json"), store.clone());
+    (dir, store, owner)
+}
+
+#[tokio::test]
+async fn quota_period_activation_preserves_usage_and_boundaries_are_idempotent() {
+    let (_dir, store, owner) = fixture();
+    let cfg = config(Some(UserQuotaPeriod::Daily));
+    store.current_handle("alice").unwrap().charge(600);
+    request(&store, &cfg, START + 100);
+    assert_eq!(
+        store
+            .schedule_status("alice", Some(UserQuotaPeriod::Daily), START)
+            .quota_schedule_state,
+        "pending"
+    );
+    owner.process_schedules(START + 100, false).await.unwrap();
+    assert_eq!(store.used("alice"), 600);
+    assert_eq!(store.schedule_deadline(), Some(START + DAY));
+    owner
+        .process_schedules(START + DAY - 1, false)
+        .await
+        .unwrap();
+    assert_eq!(store.used("alice"), 600);
+    owner.process_schedules(START + DAY, false).await.unwrap();
+    assert_eq!(store.used("alice"), 0);
+    store.current_handle("alice").unwrap().charge(17);
+    for now in [START, START + DAY, START + DAY + 10] {
+        owner.process_schedules(now, false).await.unwrap();
+        assert_eq!(store.used("alice"), 17);
+    }
+    owner
+        .process_schedules(START + 50 * DAY + 1, false)
+        .await
+        .unwrap();
+    assert_eq!(store.used("alice"), 0);
+    assert_eq!(store.schedule_deadline(), Some(START + 51 * DAY));
+    let state = read_state_file(owner.path()).await.unwrap().unwrap();
+    assert_eq!(
+        state.users["alice"].schedule.period_cursor_epoch_secs,
+        START + 50 * DAY
+    );
+}
+
+#[tokio::test]
+async fn quota_period_change_remove_reenable_and_rotation_preserve_watermark() {
+    let (_dir, store, owner) = fixture();
+    request(&store, &config(Some(UserQuotaPeriod::Daily)), START + 100);
+    owner.process_schedules(START + 100, false).await.unwrap();
+    owner
+        .process_schedules(START + 2 * DAY, false)
+        .await
+        .unwrap();
+    store.current_handle("alice").unwrap().charge(600);
+    let before = store.current_handle("alice").unwrap().schedule_checkpoint();
+    store.advance_preserving_usage("alice", 2);
+    assert_eq!(
+        store.current_handle("alice").unwrap().schedule_checkpoint(),
+        before
+    );
+    for cfg in [
+        config(None),
+        config(Some(UserQuotaPeriod::Weekly)),
+        config(Some(UserQuotaPeriod::Daily)),
+    ] {
+        request(&store, &cfg, START);
+        owner.process_schedules(START, false).await.unwrap();
+        assert_eq!(store.used("alice"), 600);
+        assert!(
+            store
+                .current_handle("alice")
+                .unwrap()
+                .schedule_checkpoint()
+                .period_cursor_epoch_secs
+                >= START + 2 * DAY
+        );
+    }
+    assert_eq!(store.schedule_deadline(), Some(START + 3 * DAY));
+}
+
+#[tokio::test]
+async fn quota_period_failure_keeps_usage_and_cursor_until_retry() {
+    let dir = tempfile::tempdir().unwrap();
+    let store = Arc::new(QuotaStore::default());
+    store.activate_fresh("alice", 1);
+    let path = dir.path().join("blocked");
+    std::fs::create_dir(&path).unwrap();
+    let owner = QuotaStateOwner::new(path.clone(), store.clone());
+    store.current_handle("alice").unwrap().charge(500);
+    request(&store, &config(Some(UserQuotaPeriod::Daily)), START);
+    assert!(owner.process_schedules(START, false).await.is_err());
+    assert_eq!(store.used("alice"), 500);
+    assert_eq!(
+        store.current_handle("alice").unwrap().schedule_checkpoint(),
+        QuotaScheduleCheckpoint::default()
+    );
+    assert_eq!(
+        store
+            .schedule_status("alice", Some(UserQuotaPeriod::Daily), START)
+            .quota_schedule_state,
+        "error"
+    );
+    std::fs::remove_dir(path).unwrap();
+    owner.process_schedules(START, false).await.unwrap();
+    assert_eq!(store.used("alice"), 500);
+    owner.process_schedules(START + DAY, false).await.unwrap();
+    assert_eq!(store.used("alice"), 0);
+}
+
+#[tokio::test]
+async fn quota_period_old_reservations_cannot_refund_the_new_period() {
+    let (_dir, store, owner) = fixture();
+    request(&store, &config(Some(UserQuotaPeriod::Daily)), START);
+    owner.process_schedules(START, false).await.unwrap();
+    let target = store.current_handle("alice").unwrap();
+    let reservation = target.try_reserve(700, 1024).unwrap();
+    owner.process_schedules(START + DAY, false).await.unwrap();
+    target.charge(200);
+    drop(reservation);
+    assert_eq!(target.used(), 200);
+}
+
+#[tokio::test]
+async fn quota_period_restart_catches_up_once_and_legacy_migration_preserves_usage() {
+    let (dir, store, owner) = fixture();
+    let cfg = config(Some(UserQuotaPeriod::Monthly));
+    std::fs::write(owner.path(), r#"{"last_reset_epoch_secs":0,"users":{"alice":{"used_bytes":712,"last_reset_epoch_secs":0}}}"#).unwrap();
+    owner.load_for_config(&cfg).await.unwrap();
+    request(&store, &cfg, START + 200);
+    owner.process_schedules(START + 200, true).await.unwrap();
+    assert_eq!(store.used("alice"), 712);
+    let restarted = Arc::new(QuotaStore::default());
+    let second = QuotaStateOwner::new(dir.path().join("quota.json"), restarted.clone());
+    second.load_for_config(&cfg).await.unwrap();
+    restarted.activate_fresh("alice", 1);
+    request(&restarted, &cfg, START + 70 * DAY);
+    second
+        .process_schedules(START + 70 * DAY, true)
+        .await
+        .unwrap();
+    assert_eq!(restarted.used("alice"), 0);
+    restarted.current_handle("alice").unwrap().charge(55);
+    second
+        .process_schedules(START + 70 * DAY, false)
+        .await
+        .unwrap();
+    assert_eq!(restarted.used("alice"), 55);
+}
+
+#[tokio::test]
+async fn quota_period_strict_startup_rejects_corruption_but_legacy_policy_does_not() {
+    let (_dir, _store, owner) = fixture();
+    std::fs::write(owner.path(), b"{broken").unwrap();
+    assert!(
+        owner
+            .load_for_config(&config(Some(UserQuotaPeriod::Daily)))
+            .await
+            .is_err()
+    );
+    assert!(owner.load_for_config(&config(None)).await.is_ok());
+    std::fs::write(owner.path(), r#"{"last_reset_epoch_secs":0,"users":{"alice":{"used_bytes":7,"last_reset_epoch_secs":0,"quota_period":"daily","period_cursor_epoch_secs":18446744073709551615}}}"#).unwrap();
+    assert!(
+        owner
+            .load_for_config(&config(Some(UserQuotaPeriod::Daily)))
+            .await
+            .is_err()
+    );
+}
+
+#[tokio::test]
+async fn quota_period_fresh_account_rejects_stale_snapshot_and_does_not_inherit_cursor() {
+    let (_dir, store, owner) = fixture();
+    let old = config(Some(UserQuotaPeriod::Daily));
+    request(&store, &old, START);
+    owner.process_schedules(START, false).await.unwrap();
+    store.retire_through("alice", 2);
+    store.activate_fresh("alice", 3);
+    let credential = credential_id_from_hex(SECRET).unwrap();
+    store.fence_fresh_schedule(
+        "alice",
+        &old,
+        credential,
+        Some(UserQuotaPeriod::Monthly),
+        Some(1024),
+    );
+    request(&store, &old, START);
+    assert_eq!(
+        store.current_handle("alice").unwrap().schedule_checkpoint(),
+        QuotaScheduleCheckpoint::default()
+    );
+    assert_eq!(store.schedule_deadline(), None);
+    request(&store, &config(Some(UserQuotaPeriod::Monthly)), START + 1);
+    owner.process_schedules(START + 1, false).await.unwrap();
+    assert_eq!(store.schedule_deadline(), Some(START + 31 * DAY));
+}
+
+#[tokio::test]
+async fn quota_period_manual_reset_advances_cursor_without_changing_cadence() {
+    let (_dir, store, owner) = fixture();
+    let now = now_epoch_secs();
+    let cfg = config(Some(UserQuotaPeriod::Daily));
+    request(&store, &cfg, now - 3 * DAY);
+    owner.process_schedules(now - 3 * DAY, false).await.unwrap();
+    store.current_handle("alice").unwrap().charge(250);
+    owner
+        .reset_user(&BTreeSet::from(["alice".into()]), "alice")
+        .await
+        .unwrap();
+    let next = store.schedule_deadline().unwrap();
+    assert_eq!(next, UserQuotaPeriod::Daily.next(now_epoch_secs()).unwrap());
+    store.current_handle("alice").unwrap().charge(80);
+    owner
+        .process_schedules(now_epoch_secs(), false)
+        .await
+        .unwrap();
+    assert_eq!(store.used("alice"), 80);
+}
+
+#[test]
+fn quota_period_gate_is_shared_with_the_admission_authority() {
+    let (_dir, store, owner) = fixture();
+    let admission = UserAdmissionAuthority::new_with_quota_store(store.clone());
+    assert!(Arc::ptr_eq(
+        &owner.mutation,
+        &admission.quota_mutation_gate()
+    ));
+    assert!(Arc::ptr_eq(&owner.mutation, &store.mutation_gate()));
+}
+
+#[tokio::test]
+async fn quota_period_disabled_expired_account_keeps_calendar_without_becoming_enabled() {
+    let (_dir, store, owner) = fixture();
+    let authority = UserAdmissionAuthority::new_with_quota_store(store.clone());
+    let mut cfg = (*config(Some(UserQuotaPeriod::Daily))).clone();
+    cfg.access.user_enabled.insert("alice".into(), false);
+    cfg.access.user_expirations.insert(
+        "alice".into(),
+        chrono::DateTime::from_timestamp(0, 0).unwrap(),
+    );
+    let cfg = Arc::new(cfg);
+    authority.activate_config_source(1, None, &cfg.access.users, &cfg.access.user_enabled);
+    let guard = owner.lock().await;
+    authority.publish_quota_config(1, &cfg, true);
+    drop(guard);
+    owner
+        .process_schedules(now_epoch_secs(), false)
+        .await
+        .unwrap();
+    store.current_handle("alice").unwrap().charge(700);
+    let boundary = store.schedule_deadline().unwrap();
+    owner.process_schedules(boundary, false).await.unwrap();
+    assert_eq!(store.used("alice"), 0);
+    assert!(!authority.is_user_enabled("alice"));
+    assert!(cfg.access.user_expirations["alice"].timestamp() < now_epoch_secs() as i64);
+}
+
+mod concurrency;
+
+mod resource_model;

src/quota_state/period_tests/concurrency.rs

@@ -0,0 +1,232 @@
+use super::*;
+use std::sync::atomic::Ordering;
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn quota_period_metadata_commit_preserves_traffic_and_finishes_after_waiter_abort() {
+    let (_dir, store, owner) = fixture();
+    let cfg = config(Some(UserQuotaPeriod::Daily));
+    request(&store, &cfg, START);
+    store.current_handle("alice").unwrap().charge(100);
+    let (hook, started) = test_io::Hook::install(owner.path(), true);
+    let operation_owner = owner.clone();
+    let waiter = tokio::spawn(async move { operation_owner.process_schedules(START, false).await });
+    started.await.unwrap();
+    assert!(owner.mutation.try_lock().is_err());
+    store.current_handle("alice").unwrap().charge(77);
+    waiter.abort();
+    assert!(waiter.await.unwrap_err().is_cancelled());
+    assert!(owner.mutation.try_lock().is_err());
+    hook.release();
+    let _guard = tokio::time::timeout(std::time::Duration::from_secs(2), owner.lock())
+        .await
+        .unwrap();
+    assert_eq!(store.used("alice"), 177);
+    assert_eq!(
+        store
+            .schedule_status("alice", Some(UserQuotaPeriod::Daily), START)
+            .quota_schedule_state,
+        "active"
+    );
+    assert_eq!(hook.writes.load(Ordering::Relaxed), 1);
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn quota_period_durable_reset_fences_recreation_until_commit() {
+    let (_dir, store, owner) = fixture();
+    request(&store, &config(Some(UserQuotaPeriod::Daily)), START);
+    owner.process_schedules(START, false).await.unwrap();
+    let old = store.current_handle("alice").unwrap();
+    old.charge(900);
+    let (hook, started) = test_io::Hook::install(owner.path(), true);
+    let reset_owner = owner.clone();
+    let reset =
+        tokio::spawn(async move { reset_owner.process_schedules(START + DAY, false).await });
+    started.await.unwrap();
+    let mutate_store = store.clone();
+    let mutation = tokio::spawn(async move {
+        let _guard = mutate_store.mutation_gate().lock_owned().await;
+        mutate_store.retire_through("alice", 2);
+        mutate_store.activate_fresh("alice", 3);
+        mutate_store.current_handle("alice").unwrap().charge(400);
+    });
+    tokio::task::yield_now().await;
+    assert!(!mutation.is_finished());
+    assert_eq!(old.used(), 900);
+    hook.release();
+    reset.await.unwrap().unwrap();
+    mutation.await.unwrap();
+    assert_eq!(old.used(), 0);
+    assert_eq!(store.used("alice"), 400);
+    assert_eq!(
+        store.current_handle("alice").unwrap().schedule_checkpoint(),
+        QuotaScheduleCheckpoint::default()
+    );
+    owner.process_schedules(START + DAY, false).await.unwrap();
+    assert_eq!(store.used("alice"), 400);
+    let persisted = read_state_file(owner.path()).await.unwrap().unwrap();
+    assert_eq!(persisted.users["alice"].used_bytes, 400);
+    assert_eq!(
+        persisted.users["alice"].schedule,
+        QuotaScheduleCheckpoint::default()
+    );
+}
+
+#[tokio::test]
+async fn quota_period_restart_after_ambiguous_commit_never_replays_completed_boundary() {
+    let (dir, store, owner) = fixture();
+    let cfg = config(Some(UserQuotaPeriod::Daily));
+    request(&store, &cfg, START);
+    owner.process_schedules(START, false).await.unwrap();
+    store.current_handle("alice").unwrap().charge(900);
+    let (hook, _) = test_io::Hook::install(owner.path(), false);
+    hook.fail_after.store(true, Ordering::Relaxed);
+    assert!(owner.process_schedules(START + DAY, false).await.is_err());
+    assert_eq!(store.used("alice"), 900);
+    assert_eq!(
+        store
+            .schedule_status("alice", Some(UserQuotaPeriod::Daily), START + DAY)
+            .quota_schedule_state,
+        "error"
+    );
+    drop(hook);
+    let restarted = Arc::new(QuotaStore::default());
+    let restart = QuotaStateOwner::new(dir.path().join("quota.json"), restarted.clone());
+    restart.load_for_config(&cfg).await.unwrap();
+    restarted.activate_fresh("alice", 1);
+    request(&restarted, &cfg, START + DAY);
+    restarted.current_handle("alice").unwrap().charge(77);
+    restart.process_schedules(START + DAY, false).await.unwrap();
+    assert_eq!(restarted.used("alice"), 77);
+}
+
+#[tokio::test]
+async fn quota_period_unchanged_reload_is_io_free_and_pending_reversal_is_not_a_reset() {
+    let (_dir, store, owner) = fixture();
+    let cfg = config(Some(UserQuotaPeriod::Daily));
+    request(&store, &cfg, START);
+    owner.process_schedules(START, false).await.unwrap();
+    let (hook, _) = test_io::Hook::install(owner.path(), false);
+    for _ in 0..1000 {
+        request(&store, &cfg, START + 500);
+        owner.process_schedules(START + 500, false).await.unwrap();
+    }
+    request(&store, &config(Some(UserQuotaPeriod::Weekly)), START + 500);
+    request(&store, &cfg, START + 500);
+    owner.process_schedules(START + 500, false).await.unwrap();
+    assert_eq!(hook.writes.load(Ordering::Relaxed), 0);
+    assert_eq!(store.schedule_deadline(), Some(START + DAY));
+}
+
+#[tokio::test]
+async fn quota_period_manual_recovery_clears_error_without_postponing_calendar() {
+    let (_dir, store, owner) = fixture();
+    let now = now_epoch_secs();
+    request(&store, &config(Some(UserQuotaPeriod::Daily)), now - 3 * DAY);
+    owner.process_schedules(now - 3 * DAY, false).await.unwrap();
+    let (hook, _) = test_io::Hook::install(owner.path(), false);
+    hook.fail_after.store(true, Ordering::Relaxed);
+    assert!(owner.process_schedules(now, false).await.is_err());
+    hook.fail_after.store(false, Ordering::Relaxed);
+    owner
+        .reset_user(&BTreeSet::from(["alice".into()]), "alice")
+        .await
+        .unwrap();
+    assert_eq!(
+        store
+            .schedule_status("alice", Some(UserQuotaPeriod::Daily), now)
+            .quota_schedule_state,
+        "active"
+    );
+    assert!(!store.schedule_ready(now));
+}
+
+#[tokio::test]
+async fn quota_period_stale_candidate_cannot_attach_retired_policy_to_fresh_account() {
+    let (_dir, store, owner) = fixture();
+    let old = config(Some(UserQuotaPeriod::Daily));
+    let stale_candidate = Arc::new((*old).clone());
+    let credential = credential_id_from_hex(SECRET).unwrap();
+    store.fence_fresh_schedule(
+        "alice",
+        &old,
+        credential,
+        Some(UserQuotaPeriod::Monthly),
+        Some(1024),
+    );
+    store.request_schedule("alice", 1, credential, &stale_candidate, START, false);
+    assert_eq!(store.schedule_deadline(), None);
+    assert_eq!(
+        store
+            .schedule_status("alice", Some(UserQuotaPeriod::Monthly), START)
+            .quota_schedule_state,
+        "pending"
+    );
+    request(&store, &config(Some(UserQuotaPeriod::Monthly)), START);
+    owner.process_schedules(START, false).await.unwrap();
+    assert_eq!(store.schedule_deadline(), Some(START + 31 * DAY));
+}
+
+#[tokio::test]
+async fn quota_period_worker_drops_obsolete_retry_and_shuts_down_cooperatively() {
+    let (_dir, store, owner) = fixture();
+    let now = now_epoch_secs();
+    request(&store, &config(Some(UserQuotaPeriod::Daily)), now - 3 * DAY);
+    owner.process_schedules(now - 3 * DAY, false).await.unwrap();
+    let (hook, started) = test_io::Hook::install(owner.path(), false);
+    hook.fail_after.store(true, Ordering::Relaxed);
+    let control = crate::maestro::control_plane::ProcessControlPlane::new();
+    owner.spawn_scheduler(&control);
+    started.await.unwrap();
+    let guard = owner.lock().await;
+    assert_eq!(
+        store
+            .schedule_status("alice", Some(UserQuotaPeriod::Daily), now)
+            .quota_schedule_state,
+        "error"
+    );
+    hook.fail_after.store(false, Ordering::Relaxed);
+    owner
+        .reset_user_locked(&BTreeSet::from(["alice".into()]), "alice", now, guard)
+        .await
+        .unwrap();
+    tokio::time::pause();
+    let before = owner.scheduler_iterations.load(Ordering::Relaxed);
+    tokio::time::advance(std::time::Duration::from_secs(65)).await;
+    for _ in 0..100 {
+        tokio::task::yield_now().await;
+    }
+    let iterations = owner.scheduler_iterations.load(Ordering::Relaxed) - before;
+    assert!(iterations < 20, "idle scheduler spun {iterations} times");
+    assert!(!store.schedule_ready(now));
+    assert!(control.shutdown(std::time::Duration::from_secs(2)).await);
+    tokio::time::resume();
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn quota_period_manual_commit_crossing_midnight_handles_new_boundary_once() {
+    let (_dir, store, owner) = fixture();
+    request(&store, &config(Some(UserQuotaPeriod::Daily)), START);
+    owner.process_schedules(START, false).await.unwrap();
+    let (hook, started) = test_io::Hook::install(owner.path(), true);
+    let manual_owner = owner.clone();
+    let manual = tokio::spawn(async move {
+        let guard = manual_owner.lock().await;
+        manual_owner
+            .reset_user_locked(
+                &BTreeSet::from(["alice".into()]),
+                "alice",
+                START + DAY - 1,
+                guard,
+            )
+            .await
+    });
+    started.await.unwrap();
+    hook.release();
+    manual.await.unwrap().unwrap();
+    store.current_handle("alice").unwrap().charge(13);
+    owner.process_schedules(START + DAY, false).await.unwrap();
+    assert_eq!(store.used("alice"), 0);
+    store.current_handle("alice").unwrap().charge(19);
+    owner.process_schedules(START + DAY, false).await.unwrap();
+    assert_eq!(store.used("alice"), 19);
+}

src/quota_state/period_tests/resource_model.rs

@@ -0,0 +1,140 @@
+use super::*;
+use std::sync::atomic::Ordering;
+
+#[tokio::test]
+async fn quota_period_retained_cursor_survives_legacy_inventory_growth() {
+    let (_dir, store, owner) = fixture();
+    request(&store, &config(Some(UserQuotaPeriod::Daily)), START);
+    owner.process_schedules(START, false).await.unwrap();
+    let mut users = (0..QUOTA_STATE_MAX_USERS)
+        .map(|i| format!("a{i:05}"))
+        .collect::<BTreeSet<_>>();
+    for user in &users {
+        store.current_or_legacy_handle(user).charge(1);
+    }
+    users.insert("alice".into());
+    request(&store, &config(None), START + 1);
+    owner.process_schedules(START + 1, false).await.unwrap();
+    owner.save(&users).await.unwrap();
+    let state = read_state_file(owner.path()).await.unwrap().unwrap();
+    assert_eq!(state.users.len(), QUOTA_STATE_MAX_USERS);
+    assert_eq!(state.users["alice"].schedule.quota_period, None);
+    assert_eq!(
+        state.users["alice"].schedule.period_cursor_epoch_secs,
+        START + 1
+    );
+    assert!(!state.users.contains_key("a65535"));
+}
+
+#[tokio::test]
+async fn quota_period_oversized_checkpoint_cannot_publish_reset() {
+    let (_dir, store, owner) = fixture();
+    request(&store, &config(Some(UserQuotaPeriod::Daily)), START);
+    owner.process_schedules(START, false).await.unwrap();
+    let before = std::fs::read(owner.path()).unwrap();
+    let huge_name = "x".repeat(QUOTA_STATE_MAX_BYTES as usize);
+    store.current_or_legacy_handle(&huge_name).charge(1);
+    store.current_handle("alice").unwrap().charge(77);
+    assert!(owner.process_schedules(START + DAY, false).await.is_err());
+    assert_eq!(store.used("alice"), 77);
+    assert_eq!(std::fs::read(owner.path()).unwrap(), before);
+    assert_eq!(
+        store
+            .schedule_status("alice", Some(UserQuotaPeriod::Daily), START + DAY)
+            .quota_schedule_state,
+        "error"
+    );
+}
+
+#[tokio::test]
+async fn quota_period_many_due_users_share_one_durable_batch() {
+    let (_dir, store, owner) = fixture();
+    let mut cfg = (*config(Some(UserQuotaPeriod::Daily))).clone();
+    for i in 0..128 {
+        let user = format!("user{i}");
+        store.activate_fresh(&user, 1);
+        store.current_handle(&user).unwrap().charge(123);
+        cfg.access.users.insert(user.clone(), SECRET.into());
+        cfg.access.user_data_quota.insert(user.clone(), 1024);
+        cfg.access
+            .user_quota_periods
+            .insert(user, UserQuotaPeriod::Daily);
+    }
+    let cfg = Arc::new(cfg);
+    for user in cfg.access.users.keys() {
+        store.request_schedule(
+            user,
+            1,
+            credential_id_from_hex(SECRET).unwrap(),
+            &cfg,
+            START,
+            true,
+        );
+    }
+    let (hook, _) = super::super::test_io::Hook::install(owner.path(), false);
+    owner.process_schedules(START, false).await.unwrap();
+    assert_eq!(hook.writes.load(Ordering::Relaxed), 1);
+    owner.process_schedules(START + DAY, false).await.unwrap();
+    assert_eq!(hook.writes.load(Ordering::Relaxed), 2);
+    for user in cfg.access.users.keys() {
+        assert_eq!(store.used(user), 0);
+    }
+    assert_eq!(store.schedule_deadline(), Some(START + 2 * DAY));
+}
+
+#[test]
+fn quota_period_daily_model_preserves_monotonic_cursor_under_clock_pressure() {
+    let (_dir, store, _owner) = fixture();
+    let cfg = config(Some(UserQuotaPeriod::Daily));
+    request(&store, &cfg, START + 123);
+    for change in store.schedule_changes(START + 123).unwrap() {
+        store.publish_schedule(&change, START + 123);
+    }
+    let mut cursor = START + 123;
+    let mut used = 0;
+    let mut seed = 19u64;
+    for step in 0..10_000 {
+        seed = seed.wrapping_mul(6364136223846793005).wrapping_add(1);
+        let now = START + (seed % 90) * DAY + seed % DAY;
+        let amount = seed % 100;
+        store.current_handle("alice").unwrap().charge(amount);
+        used += amount;
+        let boundary = now / DAY * DAY;
+        if boundary > cursor {
+            cursor = boundary;
+            used = 0;
+        }
+        for change in store.schedule_changes(now).unwrap() {
+            store.publish_schedule(&change, now);
+        }
+        if step % 100 == 0 {
+            store.advance_preserving_usage("alice", step + 2);
+        }
+        assert_eq!(store.used("alice"), used);
+        assert_eq!(
+            store
+                .current_handle("alice")
+                .unwrap()
+                .schedule_checkpoint()
+                .period_cursor_epoch_secs,
+            cursor
+        );
+        assert_eq!(store.schedule_deadline(), Some((cursor / DAY + 1) * DAY));
+    }
+}
+
+#[tokio::test]
+async fn quota_period_invalid_clock_reports_error_without_publishing_metadata() {
+    let (_dir, store, owner) = fixture();
+    request(&store, &config(Some(UserQuotaPeriod::Daily)), u64::MAX);
+    store.current_handle("alice").unwrap().charge(5);
+    assert!(owner.process_schedules(u64::MAX, false).await.is_err());
+    assert_eq!(store.used("alice"), 5);
+    assert_eq!(
+        store
+            .schedule_status("alice", Some(UserQuotaPeriod::Daily), u64::MAX)
+            .quota_schedule_state,
+        "error"
+    );
+    assert!(!owner.path().exists());
+}

src/quota_state/periods.rs

@@ -0,0 +1,174 @@
+use std::sync::Arc;
+use std::time::Duration;
+
+use tracing::{info, warn};
+
+use super::{QuotaStateOwner, now_epoch_secs, wait_for_blocking_io, write_state_file_blocking};
+use crate::config::ProxyConfig;
+use crate::maestro::control_plane::ProcessControlPlane;
+use crate::stats::{QuotaScheduleStatus, UserQuotaSnapshot};
+
+impl QuotaStateOwner {
+    /// Exposes the same control-plane gate used by account lifecycle transitions.
+    pub(crate) async fn lock(&self) -> tokio::sync::OwnedMutexGuard<()> {
+        Arc::clone(&self.mutation).lock_owned().await
+    }
+
+    /// Loads legacy files tolerantly; scheduled policies require trustworthy state.
+    pub(crate) async fn load_for_config(&self, config: &ProxyConfig) -> std::io::Result<()> {
+        self.load_inner(
+            &config.access.users.keys().cloned().collect(),
+            !config.access.user_quota_periods.is_empty(),
+        )
+        .await
+    }
+
+    /// Fences first calendar attachment for an API-created account.
+    pub(crate) fn fence_fresh_schedule(
+        &self,
+        user: &str,
+        source: &Arc<ProxyConfig>,
+        credential: crate::proxy::user_admission::UserCredentialId,
+        period: Option<crate::config::UserQuotaPeriod>,
+        quota: Option<u64>,
+    ) {
+        self.store
+            .fence_fresh_schedule(user, source, credential, period, quota);
+    }
+
+    /// Retains first-attachment fencing through API edits and secret rotation.
+    pub(crate) fn refresh_fresh_schedule(&self, user: &str, config: &ProxyConfig) {
+        self.store.refresh_fresh_schedule(user, config);
+    }
+
+    /// Captures API status under the caller's mutation guard.
+    pub(crate) fn schedule_status(
+        &self,
+        user: &str,
+        desired: Option<crate::config::UserQuotaPeriod>,
+    ) -> QuotaScheduleStatus {
+        self.store.schedule_status(user, desired, now_epoch_secs())
+    }
+
+    /// Checkpoints a bounded batch before publishing any reset or calendar metadata.
+    pub(crate) async fn process_schedules(
+        &self,
+        now: u64,
+        force_save: bool,
+    ) -> std::io::Result<()> {
+        let guard = self.lock().await;
+        self.process_schedules_locked(now, force_save, guard).await
+    }
+
+    async fn process_schedules_locked(
+        &self,
+        now: u64,
+        force_save: bool,
+        guard: tokio::sync::OwnedMutexGuard<()>,
+    ) -> std::io::Result<()> {
+        if !force_save && !self.store.schedule_ready(now) {
+            return Ok(());
+        }
+        let changes = self.store.schedule_changes(now)?;
+        let configured_users = self.store.configured_quota_users();
+        let mut state = self.state_for_users(&configured_users, None);
+        for change in &changes {
+            let quota = if change.reset {
+                UserQuotaSnapshot {
+                    used_bytes: 0,
+                    last_reset_epoch_secs: now,
+                }
+            } else {
+                change.target.snapshot()
+            };
+            let mut entry = super::quota_user_state(quota);
+            entry.schedule = change.checkpoint;
+            state.users.insert(change.user.clone(), entry);
+        }
+        state.last_reset_epoch_secs = state
+            .users
+            .values()
+            .map(|entry| entry.last_reset_epoch_secs)
+            .max()
+            .unwrap_or(0);
+        let path = self.path.clone();
+        let store = Arc::clone(&self.store);
+        let task = tokio::task::spawn_blocking(move || {
+            let _guard = guard;
+            if let Err(error) = write_state_file_blocking(&path, &state) {
+                store.schedule_failed(&changes);
+                return Err(error);
+            }
+            let reset_count = changes.iter().filter(|change| change.reset).count();
+            for change in &changes {
+                store.publish_schedule(change, now);
+            }
+            store.schedule_saved();
+            if reset_count > 0 {
+                info!(users = reset_count, "Completed scheduled quota reset batch");
+            }
+            Ok(())
+        });
+        wait_for_blocking_io(task).await
+    }
+
+    /// Owns one worker across runtime generations; accepted commits finish on shutdown.
+    pub(crate) fn spawn_scheduler(self: &Arc<Self>, control: &ProcessControlPlane) {
+        let owner = Arc::clone(self);
+        let _ = control.spawn_cooperative(move |cancel| async move {
+            let mut retry = 1u64;
+            let mut retry_at = None;
+            let mut last_warning: Option<tokio::time::Instant> = None;
+            loop {
+                #[cfg(test)]
+                owner.scheduler_iterations.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
+                if cancel.is_cancelled() { break; }
+                let now = now_epoch_secs();
+                let can_retry = retry_at.is_none_or(|at| tokio::time::Instant::now() >= at);
+                if can_retry && owner.store.schedule_ready(now) {
+                    // Cancellation can discard waiting work, but never an accepted commit.
+                    let guard = tokio::select! {
+                        biased;
+                        _ = cancel.cancelled() => break,
+                        guard = owner.lock() => guard,
+                    };
+                    if let Err(error) = owner.process_schedules_locked(now_epoch_secs(), false, guard).await {
+                        let instant = tokio::time::Instant::now();
+                        if last_warning.is_none_or(|at| instant.duration_since(at) >= Duration::from_secs(60)) {
+                            warn!(error = %error, "Quota schedule checkpoint failed; usage remains unchanged");
+                            last_warning = Some(instant);
+                        }
+                        retry_at = Some(instant + Duration::from_secs(retry));
+                        retry = (retry * 2).min(60);
+                    } else {
+                        if retry_at.is_some() { info!("Quota schedule persistence recovered"); }
+                        retry_at = None;
+                        retry = 1;
+                        last_warning = None;
+                    }
+                    continue;
+                }
+                // A manual reset or policy removal may resolve the failed work.
+                // An expired retry must never turn an idle worker into sleep(0) polling.
+                if can_retry && !owner.store.schedule_ready(now) {
+                    retry_at = None;
+                    retry = 1;
+                    last_warning = None;
+                }
+                let delay = retry_at.map(|at| at.saturating_duration_since(tokio::time::Instant::now()))
+                    .or_else(|| owner.store.schedule_deadline().map(|at| Duration::from_secs(at.saturating_sub(now).min(1))));
+                tokio::select! {
+                    biased;
+                    _ = cancel.cancelled() => break,
+                    _ = owner.store.schedule_notified() => {},
+                    _ = async {
+                        match delay {
+                            Some(duration) => tokio::time::sleep(duration).await,
+                            None => std::future::pending::<()>().await,
+                        }
+                    } => {},
+                }
+            }
+        });
+    }
+}

src/quota_state/test_io.rs

@@ -0,0 +1,97 @@
+//! Path-scoped deterministic I/O barriers; absent from production builds.
+use std::collections::HashMap;
+use std::path::{Path, PathBuf};
+use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
+use std::sync::{Arc, Condvar, Mutex, OnceLock};
+
+fn hooks() -> &'static Mutex<HashMap<PathBuf, Arc<Hook>>> {
+    static HOOKS: OnceLock<Mutex<HashMap<PathBuf, Arc<Hook>>>> = OnceLock::new();
+    HOOKS.get_or_init(Mutex::default)
+}
+
+pub(super) struct Hook {
+    path: PathBuf,
+    started: Mutex<Option<tokio::sync::oneshot::Sender<()>>>,
+    released: Mutex<bool>,
+    wake: Condvar,
+    pub(super) writes: AtomicUsize,
+    pub(super) fail_after: AtomicBool,
+}
+
+pub(super) struct HookGuard(Arc<Hook>);
+
+impl std::ops::Deref for HookGuard {
+    type Target = Hook;
+    fn deref(&self) -> &Hook {
+        &self.0
+    }
+}
+
+impl Drop for HookGuard {
+    fn drop(&mut self) {
+        self.0.remove();
+    }
+}
+
+impl Hook {
+    pub(super) fn install(
+        path: &Path,
+        block: bool,
+    ) -> (HookGuard, tokio::sync::oneshot::Receiver<()>) {
+        let (tx, rx) = tokio::sync::oneshot::channel();
+        let hook = Arc::new(Self {
+            path: path.into(),
+            started: Mutex::new(Some(tx)),
+            released: Mutex::new(!block),
+            wake: Condvar::new(),
+            writes: AtomicUsize::new(0),
+            fail_after: AtomicBool::new(false),
+        });
+        assert!(
+            hooks()
+                .lock()
+                .unwrap()
+                .insert(path.into(), hook.clone())
+                .is_none()
+        );
+        (HookGuard(hook), rx)
+    }
+
+    pub(super) fn release(&self) {
+        *self.released.lock().unwrap() = true;
+        self.wake.notify_all();
+    }
+
+    pub(super) fn remove(&self) {
+        self.release();
+        hooks().lock().unwrap().remove(&self.path);
+    }
+}
+
+pub(super) fn before(path: &Path) {
+    let hook = hooks().lock().unwrap().get(path).cloned();
+    if let Some(hook) = hook {
+        hook.writes.fetch_add(1, Ordering::Relaxed);
+        if let Some(tx) = hook.started.lock().unwrap().take() {
+            let _ = tx.send(());
+        }
+        let mut released = hook.released.lock().unwrap();
+        while !*released {
+            released = hook.wake.wait(released).unwrap();
+        }
+    }
+}
+
+pub(super) fn after(path: &Path) -> std::io::Result<()> {
+    if hooks()
+        .lock()
+        .unwrap()
+        .get(path)
+        .is_some_and(|hook| hook.fail_after.load(Ordering::Relaxed))
+    {
+        return Err(std::io::Error::other(
+            "injected failure after durable replacement",
+        ));
+    }
+    Ok(())
+}