Retain Policy Freshness after Schedule Activation for Quotas

src/api/users/period_tests.rs

@@ -11,6 +11,9 @@ use std::sync::Arc;
 use std::sync::atomic::{AtomicBool, AtomicU64};
 use tokio::sync::{Mutex, watch};
 
+// Period-only mutations must fence already attached accounts as well as new ones.
+mod freshness;
+
 struct Fixture {
     _dir: tempfile::TempDir,
     shared: ApiShared,

src/api/users/period_tests/freshness.rs

@@ -0,0 +1,104 @@
+use super::*;
+
+#[tokio::test]
+async fn quota_period_b11_patch_preserves_active_policy_against_delayed_source() {
+    for accepted in [false, true] {
+        let fixture = Fixture::new();
+        // The startup account has no API-creation fence.
+        patch_user(
+            "admin",
+            patch(serde_json::json!({"data_quota_bytes":1024,"quota_period":"daily"})),
+            None,
+            &fixture.shared,
+        )
+        .await
+        .unwrap();
+        fixture.publish().await;
+        let stale = ProxyConfig::parse_source(&fixture.shared.config_path).unwrap();
+        let authority = fixture.shared.proxy_shared.user_admission();
+        let epoch = authority.epoch();
+        patch_user(
+            "admin",
+            patch(serde_json::json!({"quota_period":"monthly"})),
+            None,
+            &fixture.shared,
+        )
+        .await
+        .unwrap();
+        assert_eq!(authority.epoch(), epoch);
+        assert!(authority.has_fresh_quota_schedules());
+        fixture.publish().await;
+        assert!(!authority.has_fresh_quota_schedules());
+        let account = fixture.store.current_handle("admin").unwrap();
+        account.charge(700);
+        let checkpoint = account.schedule_checkpoint();
+        let stale = Arc::new(stale.prepare().await.unwrap().config);
+        let gate = fixture.shared.quota_state.lock().await;
+        fixture
+            .shared
+            .proxy_shared
+            .publish_quota_config(1, &stale, accepted);
+        drop(gate);
+        fixture
+            .shared
+            .quota_state
+            .process_schedules(crate::quota_state::now_epoch_secs(), false)
+            .await
+            .unwrap();
+        assert_eq!(account.schedule_checkpoint(), checkpoint);
+        assert_eq!(fixture.store.used("admin"), 700);
+        assert_eq!(
+            fixture
+                .shared
+                .quota_state
+                .schedule_status("admin", Some(UserQuotaPeriod::Monthly))
+                .quota_schedule_state,
+            "active"
+        );
+        assert_eq!(
+            fixture.store.current_handle("admin").unwrap().incarnation(),
+            account.incarnation()
+        );
+        fixture.publish().await;
+        assert!(!authority.has_fresh_quota_schedules());
+        fixture.finish().await;
+    }
+}
+
+#[tokio::test]
+async fn quota_period_b11_noop_patch_keeps_active_status_without_publication() {
+    let fixture = Fixture::new();
+    patch_user(
+        "admin",
+        patch(serde_json::json!({"data_quota_bytes":1024,"quota_period":"monthly"})),
+        None,
+        &fixture.shared,
+    )
+    .await
+    .unwrap();
+    fixture.publish().await;
+    let account = fixture.store.current_handle("admin").unwrap();
+    account.charge(700);
+    let checkpoint = account.schedule_checkpoint();
+    patch_user("admin", patch(serde_json::json!({})), None, &fixture.shared)
+        .await
+        .unwrap();
+    assert_eq!(account.schedule_checkpoint(), checkpoint);
+    assert_eq!(fixture.store.used("admin"), 700);
+    assert!(
+        !fixture
+            .shared
+            .proxy_shared
+            .user_admission()
+            .has_fresh_quota_schedules()
+    );
+    assert_eq!(
+        fixture
+            .shared
+            .quota_state
+            .schedule_status("admin", Some(UserQuotaPeriod::Monthly))
+            .quota_schedule_state,
+        "active"
+    );
+    fixture.finish().await;
+}

src/maestro/reload_supervisor_admission_tests.rs

@@ -7,6 +7,10 @@ use crate::proxy::user_admission::{
 use crate::quota_state::QuotaStateOwner;
 use crate::stats::Stats;
 
+// Quota policy freshness is independent of admission epoch changes.
+#[path = "reload_supervisor_quota_freshness_tests.rs"]
+mod freshness;
+
 const OLD_SECRET: &str = "00112233445566778899aabbccddeeff";
 const NEW_SECRET: &str = "ffeeddccbbaa99887766554433221100";
 

src/maestro/reload_supervisor_quota_freshness_tests.rs

@@ -0,0 +1,64 @@
+use super::*;
+
+#[tokio::test]
+async fn quota_period_b11_prepared_runtime_cannot_restore_superseded_period() {
+    for mode in [ReloadMode::Instant, ReloadMode::Drain] {
+        let mut daily = user_config();
+        daily.quota_source_read_at = Some(std::time::Instant::now());
+        daily
+            .access
+            .user_quota_periods
+            .insert("alice".into(), UserQuotaPeriod::Daily);
+        let (fixture, authority, store) = shared_fixture(daily.clone(), daily.clone(), mode).await;
+        let dir = tempfile::tempdir().unwrap();
+        let quota = QuotaStateOwner::new(dir.path().join("quota.json"), store.clone());
+        let gate = quota.lock().await;
+        authority.publish_quota_config(1, &fixture.old_runtime.config(), true);
+        drop(gate);
+        quota
+            .process_schedules(crate::quota_state::now_epoch_secs(), false)
+            .await
+            .unwrap();
+        let prepared = prepared_runtime(fixture.new_runtime.clone());
+        let mut monthly = daily;
+        monthly
+            .access
+            .user_quota_periods
+            .insert("alice".into(), UserQuotaPeriod::Monthly);
+        let gate = quota.lock().await;
+        quota.refresh_fresh_schedule("alice", &monthly);
+        monthly.quota_source_read_at = Some(std::time::Instant::now());
+        authority.publish_quota_config(1, &Arc::new(monthly), true);
+        assert_eq!(authority.epoch(), prepared.user_admission_epoch);
+        drop(gate);
+        quota
+            .process_schedules(crate::quota_state::now_epoch_secs(), false)
+            .await
+            .unwrap();
+        let account = store.current_handle("alice").unwrap();
+        let checkpoint = account.schedule_checkpoint();
+        account.charge(700);
+        fixture
+            .supervisor
+            .activate_prepared(
+                fixture.command,
+                fixture.old_runtime.clone(),
+                prepared,
+                RevisionGateAction::Warn("config revision changed during preparation".into()),
+            )
+            .await;
+        quota
+            .process_schedules(crate::quota_state::now_epoch_secs(), false)
+            .await
+            .unwrap();
+        assert_eq!(account.schedule_checkpoint(), checkpoint);
+        assert_eq!(store.used("alice"), 700);
+        assert_eq!(fixture.supervisor.active_runtime.load().id, 2);
+        assert_eq!(
+            fixture.control.status(1).await.unwrap().state,
+            ReloadPhase::Succeeded
+        );
+        fixture.new_runtime.stop_sessions().await;
+        fixture.new_runtime.stop_background_tasks().await;
+    }
+}

src/proxy/user_admission.rs

@@ -121,7 +121,7 @@ impl UserAdmissionAuthority {
         })
     }
 
-    /// Requests source-evidence refresh only while a fresh account is fenced.
+    /// Requests source evidence for pending API intent or a rejected quota snapshot.
     pub(crate) fn has_fresh_quota_schedules(&self) -> bool {
         self.quota_store.has_fresh_schedules()
     }

src/quota_state/period_tests.rs

@@ -295,3 +295,6 @@ async fn quota_period_disabled_expired_account_keeps_calendar_without_becoming_e
 mod concurrency;
 
 mod resource_model;
+
+// Account-lifetime freshness and delayed source publication regressions.
+mod freshness;

src/quota_state/period_tests/freshness.rs

@@ -0,0 +1,281 @@
+use super::*;
+use std::time::{Duration, Instant};
+
+fn source(period: Option<UserQuotaPeriod>, at: Instant) -> Arc<ProxyConfig> {
+    let mut cfg = (*config(period)).clone();
+    cfg.quota_source_read_at = Some(at);
+    Arc::new(cfg)
+}
+
+#[tokio::test]
+async fn quota_period_b11_unchanged_api_intent_fences_prepared_conflicting_source() {
+    let (_dir, store, owner) = fixture();
+    let monthly = source(Some(UserQuotaPeriod::Monthly), Instant::now());
+    request(&store, &monthly, START);
+    owner.process_schedules(START, false).await.unwrap();
+    let delayed = source(Some(UserQuotaPeriod::Daily), Instant::now());
+    let gate = owner.lock().await;
+    store.refresh_fresh_schedule("alice", &monthly);
+    assert!(!store.has_fresh_schedules());
+    request(&store, &delayed, START);
+    drop(gate);
+    owner.process_schedules(START, false).await.unwrap();
+    assert_eq!(store.schedule_deadline(), Some(START + 31 * DAY));
+    assert_eq!(
+        store
+            .schedule_status("alice", Some(UserQuotaPeriod::Monthly), START)
+            .quota_schedule_state,
+        "active"
+    );
+}
+
+#[tokio::test]
+async fn quota_period_b11_stale_publication_preserves_monthly_boundary() {
+    for accepted in [false, true] {
+        let (_dir, store, owner) = fixture();
+        let at = Instant::now();
+        let daily = source(Some(UserQuotaPeriod::Daily), at);
+        let monthly = source(Some(UserQuotaPeriod::Monthly), at + Duration::from_secs(1));
+        let now = START + 19 * DAY + 86340;
+        request(&store, &daily, now);
+        owner.process_schedules(now, false).await.unwrap();
+        request(&store, &monthly, now);
+        owner.process_schedules(now, false).await.unwrap();
+        let account = store.current_handle("alice").unwrap();
+        account.charge(700);
+        let checkpoint = account.schedule_checkpoint();
+        store.request_schedule(
+            "alice",
+            1,
+            credential_id_from_hex(SECRET).unwrap(),
+            &daily,
+            now,
+            accepted,
+        );
+        owner.process_schedules(now, false).await.unwrap();
+        assert_eq!(
+            account.schedule_checkpoint(),
+            checkpoint,
+            "accepted={accepted}"
+        );
+        assert_eq!(store.schedule_deadline(), Some(START + 31 * DAY));
+        owner
+            .process_schedules(START + 20 * DAY, false)
+            .await
+            .unwrap();
+        assert_eq!(store.used("alice"), 700);
+        assert_eq!(
+            store
+                .schedule_status("alice", Some(UserQuotaPeriod::Monthly), now)
+                .quota_schedule_state,
+            "active"
+        );
+        // Missing evidence cannot bypass an established source watermark either.
+        request(&store, &config(Some(UserQuotaPeriod::Daily)), now);
+        owner.process_schedules(now, false).await.unwrap();
+        assert_eq!(account.schedule_checkpoint(), checkpoint);
+        owner
+            .process_schedules(START + 31 * DAY, false)
+            .await
+            .unwrap();
+        assert_eq!(store.used("alice"), 0);
+    }
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn quota_period_b11_concurrent_publication_preserves_latest_source() {
+    let (_dir, store, owner) = fixture();
+    let at = Instant::now();
+    request(&store, &source(Some(UserQuotaPeriod::Monthly), at), START);
+    owner.process_schedules(START, false).await.unwrap();
+    store.current_handle("alice").unwrap().charge(700);
+    let start = Arc::new(tokio::sync::Barrier::new(4));
+    let mut tasks = Vec::new();
+    for worker in 0..4 {
+        let store = store.clone();
+        let start = start.clone();
+        tasks.push(tokio::spawn(async move {
+            start.wait().await;
+            for index in (0..128).rev() {
+                let sequence = index * 4 + worker + 1;
+                let period = if sequence % 2 == 0 {
+                    UserQuotaPeriod::Monthly
+                } else {
+                    UserQuotaPeriod::Daily
+                };
+                let cfg = source(Some(period), at + Duration::from_secs(sequence));
+                let gate = store.mutation_gate().lock_owned().await;
+                request(&store, &cfg, START);
+                drop(gate);
+                tokio::task::yield_now().await;
+            }
+        }));
+    }
+    for task in tasks {
+        task.await.unwrap();
+    }
+    owner.process_schedules(START, false).await.unwrap();
+    assert_eq!(store.schedule_deadline(), Some(START + 31 * DAY));
+    assert_eq!(store.used("alice"), 700);
+    assert_eq!(
+        store
+            .schedule_status("alice", Some(UserQuotaPeriod::Monthly), START)
+            .quota_schedule_state,
+        "active"
+    );
+}
+
+#[tokio::test]
+async fn quota_period_b11_unchanged_policy_advances_source_watermark() {
+    let (_dir, store, owner) = fixture();
+    let at = Instant::now();
+    let daily = source(Some(UserQuotaPeriod::Daily), at);
+    let delayed = source(Some(UserQuotaPeriod::Monthly), at + Duration::from_secs(1));
+    let confirmed = source(Some(UserQuotaPeriod::Daily), at + Duration::from_secs(2));
+    request(&store, &daily, START);
+    owner.process_schedules(START, false).await.unwrap();
+    request(&store, &confirmed, START);
+    request(&store, &delayed, START);
+    owner.process_schedules(START, false).await.unwrap();
+    assert_eq!(store.schedule_deadline(), Some(START + DAY));
+    let removed = source(None, at + Duration::from_secs(3));
+    request(&store, &removed, START);
+    owner.process_schedules(START, false).await.unwrap();
+    request(&store, &confirmed, START);
+    owner.process_schedules(START, false).await.unwrap();
+    assert_eq!(store.schedule_deadline(), None);
+    assert_eq!(
+        store
+            .schedule_status("alice", None, START)
+            .quota_schedule_state,
+        "disabled"
+    );
+    // A genuinely newer edit may intentionally restore the previous policy.
+    request(
+        &store,
+        &source(Some(UserQuotaPeriod::Daily), at + Duration::from_secs(4)),
+        START,
+    );
+    owner.process_schedules(START, false).await.unwrap();
+    assert_eq!(store.schedule_deadline(), Some(START + DAY));
+    // Account recreation must not inherit the retired account's source watermark.
+    store.retire_through("alice", 2);
+    store.activate_fresh("alice", 3);
+    request(&store, &delayed, START);
+    owner.process_schedules(START, false).await.unwrap();
+    assert_eq!(store.schedule_deadline(), Some(START + 31 * DAY));
+}
+
+#[tokio::test]
+async fn quota_period_b11_api_fence_survives_matching_attachment_and_failed_write() {
+    let (_dir, store, owner) = fixture();
+    let daily = source(Some(UserQuotaPeriod::Daily), Instant::now());
+    let monthly = source(Some(UserQuotaPeriod::Monthly), Instant::now());
+    request(&store, &daily, START);
+    owner.process_schedules(START, false).await.unwrap();
+    store.current_handle("alice").unwrap().charge(700);
+    let gate = owner.lock().await;
+    store.refresh_fresh_schedule("alice", &monthly);
+    assert!(store.has_fresh_schedules());
+    store.request_schedule(
+        "alice",
+        1,
+        credential_id_from_hex(SECRET).unwrap(),
+        &monthly,
+        START,
+        false,
+    );
+    drop(gate);
+    let (hook, _) = test_io::Hook::install(owner.path(), false);
+    hook.fail_after
+        .store(true, std::sync::atomic::Ordering::Relaxed);
+    assert!(owner.process_schedules(START, false).await.is_err());
+    request(&store, &daily, START);
+    assert_eq!(
+        store
+            .schedule_status("alice", Some(UserQuotaPeriod::Monthly), START)
+            .quota_schedule_state,
+        "error"
+    );
+    hook.fail_after
+        .store(false, std::sync::atomic::Ordering::Relaxed);
+    owner.process_schedules(START, false).await.unwrap();
+    assert_eq!(store.schedule_deadline(), Some(START + 31 * DAY));
+    assert_eq!(store.used("alice"), 700);
+    let gate = owner.lock().await;
+    owner
+        .reset_user_locked(
+            &BTreeSet::from(["alice".into()]),
+            "alice",
+            START + DAY,
+            gate,
+        )
+        .await
+        .unwrap();
+    store.current_handle("alice").unwrap().charge(23);
+    request(&store, &daily, START + DAY);
+    owner
+        .process_schedules(START + 2 * DAY, false)
+        .await
+        .unwrap();
+    assert_eq!(store.used("alice"), 23);
+    assert_eq!(store.schedule_deadline(), Some(START + 31 * DAY));
+}
+
+#[tokio::test]
+async fn quota_period_b11_rejected_source_gets_fresh_evidence_from_unchanged_watcher() {
+    let (dir, store, owner) = fixture();
+    let path = dir.path().join("config.toml");
+    let text = |period: &str| {
+        format!(
+            "[access.users]\nalice = '{SECRET}'\n[access.user_data_quota]\nalice = 1024\n[access.user_quota_periods]\nalice = '{period}'\n"
+        )
+    };
+    std::fs::write(&path, text("daily")).unwrap();
+    let daily = Arc::new(ProxyConfig::load(&path).unwrap());
+    std::fs::write(&path, text("monthly")).unwrap();
+    let monthly = Arc::new(ProxyConfig::load(&path).unwrap());
+    request(&store, &monthly, START);
+    owner.process_schedules(START, false).await.unwrap();
+    request(&store, &daily, START);
+    assert_eq!(
+        store
+            .schedule_status("alice", Some(UserQuotaPeriod::Monthly), START)
+            .quota_schedule_state,
+        "active"
+    );
+    assert!(store.has_fresh_schedules());
+    // The new runtime already has these bytes; only its read evidence is stale.
+    std::fs::write(&path, text("daily")).unwrap();
+    let cancel = tokio_util::sync::CancellationToken::new();
+    let refresh_store = store.clone();
+    let (mut rx, _, task) = crate::config::hot_reload::spawn_config_watcher(
+        path,
+        daily.clone(),
+        None,
+        None,
+        cancel.clone(),
+        None,
+        None,
+        Some(store.mutation_gate()),
+        Some(Arc::new(move || refresh_store.has_fresh_schedules())),
+    );
+    let task = tokio::spawn(task);
+    tokio::time::timeout(Duration::from_secs(5), rx.changed())
+        .await
+        .unwrap()
+        .unwrap();
+    let refreshed = rx.borrow_and_update().clone();
+    assert!(refreshed.quota_source_read_at > monthly.quota_source_read_at);
+    let gate = owner.lock().await;
+    request(&store, &refreshed, START);
+    drop(gate);
+    owner.process_schedules(START, false).await.unwrap();
+    assert!(!store.has_fresh_schedules());
+    assert_eq!(store.schedule_deadline(), Some(START + DAY));
+    cancel.cancel();
+    tokio::time::timeout(Duration::from_secs(5), task)
+        .await
+        .unwrap()
+        .unwrap();
+}

src/quota_state/periods.rs

@@ -36,7 +36,7 @@ impl QuotaStateOwner {
             .fence_fresh_schedule(user, source, credential, period, quota);
     }
 
-    /// Retains first-attachment fencing through API edits and secret rotation.
+    /// Fences committed API edits and secret rotation throughout the account lifetime.
     pub(crate) fn refresh_fresh_schedule(&self, user: &str, config: &ProxyConfig) {
         self.store.refresh_fresh_schedule(user, config);
     }

src/stats/quota_store/schedule.rs

@@ -22,14 +22,14 @@ fn is_zero(value: &u64) -> bool {
     *value == 0
 }
 
-/// First policy attachment fence for an API-created account.
+/// Latest API intent awaiting a matching or demonstrably newer policy source.
 #[derive(Clone)]
 pub(super) struct FreshSchedule {
     /// Snapshot visible at creation; preparation can publish another older snapshot later.
     pub(super) source: Weak<ProxyConfig>,
     /// Monotonic commit time, independent from quota calendar wall time.
     pub(super) created_at: std::time::Instant,
-    /// Latest committed credential until first policy attachment.
+    /// Latest committed credential while a policy source is pending.
     pub(super) credential: UserCredentialId,
     /// Latest committed desired calendar.
     pub(super) period: Option<UserQuotaPeriod>,
@@ -50,8 +50,12 @@ pub(super) struct QuotaSchedule {
     pub(super) pending: bool,
     /// Reports the last unsuccessful attempt without modifying the durable cursor.
     pub(super) failed: bool,
-    /// Rejects pre-creation policies until a matching or newer source arrives.
+    /// Rejects pre-mutation policies until a matching or newer source arrives.
     pub(super) fresh: Option<FreshSchedule>,
+    /// Latest accepted source read or API commit; survives policy attachment.
+    pub(super) policy_fence_at: Option<std::time::Instant>,
+    /// Requests new read evidence after rejection without changing calendar status.
+    pub(super) needs_source_refresh: bool,
 }
 
 /// Bounded replacement index: a username owns at most one deadline and dirty entry.
@@ -99,7 +103,7 @@ impl QuotaStore {
         self.schedule_notify.notified().await;
     }
 
-    /// Detects accounts still awaiting first policy attachment on a cold reload.
+    /// Detects API fences or rejected snapshots needing new source evidence.
     pub(crate) fn has_fresh_schedules(&self) -> bool {
         self.users.iter().any(|entry| {
             entry
@@ -108,7 +112,10 @@ impl QuotaStore {
                 .lock()
                 .current
                 .as_ref()
-                .is_some_and(|account| account.counters.schedule.lock().fresh.is_some())
+                .is_some_and(|account| {
+                    let schedule = account.counters.schedule.lock();
+                    schedule.fresh.is_some() || schedule.needs_source_refresh
+                })
         })
     }
 
@@ -190,25 +197,36 @@ impl QuotaStore {
         });
     }
 
-    /// Updates an unconsumed creation fence after another committed API mutation.
+    /// Fences committed API intent for both existing and newly created accounts.
     pub(crate) fn refresh_fresh_schedule(&self, user: &str, config: &ProxyConfig) {
         let Some(target) = self.current_handle(user) else {
             return;
         };
         let mut schedule = target.counters.schedule.lock();
-        let Some(fresh) = schedule.fresh.as_mut() else {
-            return;
-        };
         if let Some(credential) = config
             .access
             .users
             .get(user)
             .and_then(|secret| crate::proxy::user_admission::credential_id_from_hex(secret))
         {
-            fresh.credential = credential;
-            fresh.period = config.access.user_quota_periods.get(user).copied();
-            fresh.quota = config.access.user_data_quota.get(user).copied();
-            fresh.created_at = std::time::Instant::now();
+            let created_at = std::time::Instant::now();
+            let period = config.access.user_quota_periods.get(user).copied();
+            schedule.policy_fence_at = schedule.policy_fence_at.max(Some(created_at));
+            // An unchanged policy needs a barrier, but no new attachment or file event.
+            if schedule.fresh.is_none() && schedule.requested == period {
+                return;
+            }
+            schedule.fresh = Some(FreshSchedule {
+                source: schedule
+                    .fresh
+                    .as_ref()
+                    .map(|fresh| fresh.source.clone())
+                    .unwrap_or_default(),
+                created_at,
+                credential,
+                period,
+                quota: config.access.user_data_quota.get(user).copied(),
+            });
         }
     }
 
@@ -239,8 +257,22 @@ impl QuotaStore {
             {
                 return;
             }
+            // A matching API intent may arrive with an old admission epoch. Accept it
+            // without discarding the commit barrier that protects later publications.
+            schedule.policy_fence_at = schedule.policy_fence_at.max(Some(fresh.created_at));
             schedule.fresh = None;
+        } else if schedule.policy_fence_at.is_some_and(|fence| {
+            config
+                .quota_source_read_at
+                .is_none_or(|read_at| read_at < fence)
+        }) {
+            // The active runtime may already contain the latest disk bytes. Ask the
+            // watcher for new evidence even when those bytes are otherwise unchanged.
+            schedule.needs_source_refresh = true;
+            return;
         }
+        schedule.policy_fence_at = schedule.policy_fence_at.max(config.quota_source_read_at);
+        schedule.needs_source_refresh = false;
         if schedule.requested != period {
             schedule.requested = period;
             schedule.accepted_at = now;