diff --git a/src/api/users/period_tests.rs b/src/api/users/period_tests.rs index b34d1f1..bb6b09d 100644 --- a/src/api/users/period_tests.rs +++ b/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, diff --git a/src/api/users/period_tests/freshness.rs b/src/api/users/period_tests/freshness.rs new file mode 100644 index 0000000..1d7e659 --- /dev/null +++ b/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; +} diff --git a/src/maestro/reload_supervisor_admission_tests.rs b/src/maestro/reload_supervisor_admission_tests.rs index 43bfe19..ab518f1 100644 --- a/src/maestro/reload_supervisor_admission_tests.rs +++ b/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"; diff --git a/src/maestro/reload_supervisor_quota_freshness_tests.rs b/src/maestro/reload_supervisor_quota_freshness_tests.rs new file mode 100644 index 0000000..3a44387 --- /dev/null +++ b/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; + } +} diff --git a/src/proxy/user_admission.rs b/src/proxy/user_admission.rs index c913fd9..bcb73c7 100644 --- a/src/proxy/user_admission.rs +++ b/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() } diff --git a/src/quota_state/period_tests.rs b/src/quota_state/period_tests.rs index df52987..cef7eda 100644 --- a/src/quota_state/period_tests.rs +++ b/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; diff --git a/src/quota_state/period_tests/freshness.rs b/src/quota_state/period_tests/freshness.rs new file mode 100644 index 0000000..0295faf --- /dev/null +++ b/src/quota_state/period_tests/freshness.rs @@ -0,0 +1,281 @@ +use super::*; +use std::time::{Duration, Instant}; + +fn source(period: Option, at: Instant) -> Arc { + 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(); +} diff --git a/src/quota_state/periods.rs b/src/quota_state/periods.rs index de9ee6b..afcc25e 100644 --- a/src/quota_state/periods.rs +++ b/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); } diff --git a/src/stats/quota_store/schedule.rs b/src/stats/quota_store/schedule.rs index d86e9f0..95b414b 100644 --- a/src/stats/quota_store/schedule.rs +++ b/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, /// 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, @@ -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, + /// Latest accepted source read or API commit; survives policy attachment. + pub(super) policy_fence_at: Option, + /// 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;