use crate::{AccountView, AdminReceipt, AdminState};
use std::collections::{HashMap, HashSet};
use std::num::NonZeroUsize;
use std::sync::Arc;
use std::sync::Mutex;
use async_trait::async_trait;
use jiff::{SignedDuration, Timestamp};
use tokio::sync::broadcast;
use tollgate_core::{
AccountId, AccountStatus, BudgetSchedule, BudgetView, CapacityClass, CostUnits, FencingToken,
Generation, KeyId, LeaseGrant, LeaseId, Principal, PublishableSnapshot, UsageEvent,
UsageSource,
};
use crate::leases::{LeaseRecord, Leases, Settled};
pub use crate::traits::{AccountConfig, Conservation, StatusChange};
use crate::traits::{
AdminStore, AllocateError, Allocation, BudgetError, CreateAccountError, GrantPolicy,
GrantPolicyError, IngestError, IngestReport, KeyDirectory, KeyError, KeyRecord,
KeySnapshotError, KeySummary, LeaseAllocator, PUSH_CHANNEL_CAPACITY, PublishSnapshotError,
ReclaimBatch, ReclaimedLease, Revocation, RolledAccount, RolloverBatch, SetStatusError,
SnapshotPush, SnapshotResolution, SnapshotSource, StoreError, StoreHealth, UsageSink,
pushes_exceed_capacity,
};
#[derive(Debug, Clone, Copy, Default)]
struct Balance {
allowance: CostUnits,
topup: CostUnits,
}
impl Balance {
fn total(self) -> CostUnits {
self.allowance
.checked_add(self.topup)
.expect("a balance that was funded in halves fits the sum it came from")
}
fn take(&mut self, units: CostUnits) -> Option<Drawn> {
let from_allowance = self.allowance.min(units);
let from_topup = units.checked_sub(from_allowance)?;
self.allowance = self.allowance.checked_sub(from_allowance)?;
self.topup = self.topup.checked_sub(from_topup)?;
Some(Drawn {
from_allowance,
from_topup,
})
}
}
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub(crate) struct Drawn {
pub(crate) from_allowance: CostUnits,
pub(crate) from_topup: CostUnits,
}
#[derive(Debug)]
struct AccountRecord {
balance: Balance,
deposited: CostUnits,
schedule: Option<BudgetSchedule>,
period_start: Timestamp,
expired: CostUnits,
status: AccountStatus,
capacity_class: CapacityClass,
next_fence: u64,
usage_recorded: CostUnits,
overage_recorded: CostUnits,
settlement_loss: CostUnits,
}
impl AccountRecord {
fn budget_view(&self) -> BudgetView {
let funded = self
.deposited
.checked_add(self.overage_recorded)
.expect("an account cannot be funded past what it was funded with");
let consumed = self
.usage_recorded
.checked_add(self.settlement_loss)
.and_then(|spent| spent.checked_add(self.expired))
.expect("consumption cannot exceed the funding it came from");
BudgetView {
balance_at_publish: funded
.checked_sub(consumed)
.expect("conservation keeps consumption within funding"),
period_end: self
.schedule
.map(|schedule| schedule.period.end_after(self.period_start)),
}
}
}
fn allowance_lapsed(record: &AccountRecord, period_start: Timestamp) -> bool {
period_start < record.period_start
}
fn credit_settlement(
record: &mut AccountRecord,
funding: Drawn,
period_start: Timestamp,
unspent: CostUnits,
) {
let to_topup = funding.from_topup.min(unspent);
let to_allowance = unspent
.checked_sub(to_topup)
.expect("the top-up half never exceeds the grant it was drawn from");
record.balance.topup = record
.balance
.topup
.checked_add(to_topup)
.expect("settlement credit overflow");
let bucket = if allowance_lapsed(record, period_start) {
&mut record.expired
} else {
&mut record.balance.allowance
};
*bucket = bucket
.checked_add(to_allowance)
.expect("settlement credit overflow");
}
#[derive(Debug)]
enum SnapshotRecord {
Present(PublishableSnapshot),
Revoked(Generation),
}
impl SnapshotRecord {
fn generation(&self) -> Generation {
match self {
SnapshotRecord::Present(snapshot) => snapshot.generation,
SnapshotRecord::Revoked(generation) => *generation,
}
}
}
#[derive(Default)]
struct Inner {
accounts: HashMap<AccountId, AccountRecord>,
leases: Leases,
snapshots: HashMap<Principal, SnapshotRecord>,
usage: HashMap<tollgate_core::RequestId, UsageEvent>,
credential_activity: HashMap<KeyId, Timestamp>,
keys: HashMap<KeyId, StoredKey>,
unrevoked_keys: std::collections::BTreeSet<KeyId>,
key_principals: HashMap<Principal, KeyId>,
credential_revision: u64,
next_lease_id: u128,
}
#[derive(Debug, Clone)]
struct StoredKey {
record: KeyRecord,
revoked_at: Option<Timestamp>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct StoredRecords {
pub credential_activity: usize,
pub usage_events: usize,
pub leases: usize,
pub active_leases: usize,
}
pub struct MemoryStore {
inner: Mutex<Inner>,
policy: GrantPolicy,
push: broadcast::Sender<SnapshotPush>,
}
impl MemoryStore {
pub fn new(policy: GrantPolicy) -> Result<Arc<Self>, GrantPolicyError> {
policy.validate()?;
let (push, _) = broadcast::channel(PUSH_CHANNEL_CAPACITY);
Ok(Arc::new(MemoryStore {
inner: Mutex::new(Inner::default()),
policy,
push,
}))
}
fn grant_expiry(
&self,
ttl: SignedDuration,
now: Timestamp,
) -> Result<Timestamp, AllocateError> {
if ttl <= SignedDuration::ZERO {
return Err(AllocateError::InvalidTtl);
}
let ttl = ttl.min(self.policy.max_ttl);
now.checked_add(ttl)
.map_err(|e| AllocateError::Storage(StoreError(format!("ttl overflow: {e}"))))
}
fn lock(&self) -> std::sync::MutexGuard<'_, Inner> {
self.inner.lock().expect("memory store lock poisoned")
}
pub fn create_account(&self, config: AccountConfig) {
self.try_create_account(config)
.expect("account already exists");
}
pub fn try_create_account(&self, config: AccountConfig) -> Result<(), CreateAccountError> {
self.create_account_audited(config)
.map(|receipt| receipt.outcome)
}
fn create_account_audited(
&self,
config: AccountConfig,
) -> Result<AdminReceipt<()>, CreateAccountError> {
let mut inner = self.lock();
if inner.accounts.contains_key(&config.account_id) {
return Err(CreateAccountError::AlreadyExists);
}
inner.accounts.insert(
config.account_id,
AccountRecord {
balance: Balance {
allowance: CostUnits::ZERO,
topup: config.initial_balance,
},
deposited: config.initial_balance,
schedule: None,
period_start: Timestamp::UNIX_EPOCH,
expired: CostUnits::ZERO,
status: config.status,
capacity_class: config.capacity_class,
next_fence: 1,
usage_recorded: CostUnits::ZERO,
overage_recorded: CostUnits::ZERO,
settlement_loss: CostUnits::ZERO,
},
);
Ok(AdminReceipt::new(
(),
AdminState::Absent,
AdminState::AccountCreated {
initial_balance: config.initial_balance,
status: config.status,
capacity_class: config.capacity_class,
},
))
}
pub fn deposit(&self, account: AccountId, units: CostUnits) -> Result<(), AllocateError> {
self.deposit_audited(account, units)
.map(|receipt| receipt.outcome)
}
fn deposit_audited(
&self,
account: AccountId,
units: CostUnits,
) -> Result<AdminReceipt<()>, AllocateError> {
let mut inner = self.lock();
let record = inner
.accounts
.get_mut(&account)
.ok_or(AllocateError::UnknownAccount)?;
let topup = record
.balance
.topup
.checked_add(units)
.ok_or_else(|| AllocateError::Storage(StoreError("balance overflow".into())))?;
let deposited = record
.deposited
.checked_add(units)
.ok_or_else(|| AllocateError::Storage(StoreError("deposit overflow".into())))?;
let before = AdminState::Funding {
topup: record.balance.topup,
deposited: record.deposited,
};
record.balance.topup = topup;
record.deposited = deposited;
Ok(AdminReceipt::new(
(),
before,
AdminState::Funding { topup, deposited },
))
}
pub fn publish_snapshot(
&self,
principal: Principal,
snapshot: PublishableSnapshot,
) -> Result<(), PublishSnapshotError> {
let published = {
let mut inner = self.lock();
publish_locked(&mut inner, principal, snapshot)?
};
if let Some(snapshot) = published {
self.push_to_subscribers(SnapshotPush {
principal,
resolution: SnapshotResolution::Present(snapshot),
});
}
Ok(())
}
fn push_to_subscribers(&self, push: SnapshotPush) {
let principal = push.principal;
let subscribers = self.push.send(push).unwrap_or(0);
tracing::debug!(
%principal,
subscribers,
"snapshot pushed to subscribers"
);
}
pub fn remove_snapshot(&self, principal: Principal) {
self.remove_snapshot_audited(principal);
}
fn remove_snapshot_audited(&self, principal: Principal) -> AdminReceipt<()> {
let receipt = remove_locked(&mut self.lock(), principal);
self.announce_removal(principal, &receipt);
receipt
}
fn announce_removal(&self, principal: Principal, receipt: &AdminReceipt<()>) {
if receipt.before != receipt.after
&& let AdminState::Snapshot { generation, .. } = receipt.after
{
self.push_to_subscribers(SnapshotPush {
principal,
resolution: SnapshotResolution::Revoked { generation },
});
}
}
#[must_use]
pub fn conservation(&self, account: AccountId) -> Option<Conservation> {
let inner = self.lock();
let record = inner.accounts.get(&account)?;
let mut active_grants = CostUnits::ZERO;
let mut active_used = CostUnits::ZERO;
for lease in inner.leases.active_of(account) {
active_grants = active_grants
.checked_add(lease.granted)
.expect("grant sum overflow");
active_used = active_used
.checked_add(lease.used)
.expect("used sum overflow");
}
Some(Conservation {
deposited: record.deposited,
overage_recorded: record.overage_recorded,
balance: record.balance.total(),
active_lease_grants: active_grants,
settled_usage: record
.usage_recorded
.checked_sub(active_used)
.expect("active usage never exceeds recorded usage"),
settlement_loss: record.settlement_loss,
expired: record.expired,
})
}
#[must_use]
pub fn usage_recorded(&self, account: AccountId) -> CostUnits {
self.lock()
.accounts
.get(&account)
.map(|a| a.usage_recorded)
.unwrap_or(CostUnits::ZERO)
}
#[must_use]
pub fn settled_event(&self, request_id: tollgate_core::RequestId) -> Option<UsageEvent> {
self.lock().usage.get(&request_id).copied()
}
#[must_use]
pub fn balance(&self, account: AccountId) -> CostUnits {
self.lock()
.accounts
.get(&account)
.map(|a| a.balance.total())
.unwrap_or(CostUnits::ZERO)
}
#[cfg(test)]
fn leases_examined(&self) -> usize {
self.lock().leases.examined()
}
#[must_use]
pub fn stored_records(&self) -> StoredRecords {
let inner = self.lock();
StoredRecords {
credential_activity: inner.credential_activity.len(),
usage_events: inner.usage.len(),
leases: inner.leases.len(),
active_leases: inner.leases.active_len(),
}
}
}
fn spendable_credit(
record: &AccountRecord,
funding: Drawn,
period_start: Timestamp,
unspent: CostUnits,
) -> CostUnits {
if allowance_lapsed(record, period_start) {
funding.from_topup.min(unspent)
} else {
unspent
}
}
struct ReleasePlan {
account_id: AccountId,
funding: Drawn,
period_start: Timestamp,
unspent: CostUnits,
loss: CostUnits,
}
struct GrantPlan {
granted: CostUnits,
fencing_token: FencingToken,
next_fence: u64,
next_lease_id: u128,
}
fn plan_release(
inner: &Inner,
policy: &GrantPolicy,
lease_id: LeaseId,
fencing_token: FencingToken,
unspent: CostUnits,
now: Timestamp,
) -> Result<ReleasePlan, AllocateError> {
let lease = inner
.leases
.get(lease_id)
.ok_or(AllocateError::UnknownLease)?;
if lease.fencing_token != fencing_token {
return Err(AllocateError::Fenced);
}
if !lease.is_active()
|| policy
.reclaim_cutoff(now)
.is_some_and(|cutoff| lease.expires_at <= cutoff)
{
return Err(AllocateError::LeaseNotActive);
}
let spent_plus_unspent = lease
.used
.checked_add(unspent)
.ok_or(AllocateError::InvalidRelease)?;
let loss = lease
.granted
.checked_sub(spent_plus_unspent)
.ok_or(AllocateError::InvalidRelease)?;
Ok(ReleasePlan {
account_id: lease.account_id,
funding: lease.funding,
period_start: lease.period_start,
unspent,
loss,
})
}
fn apply_release(inner: &mut Inner, lease_id: LeaseId, plan: &ReleasePlan) {
assert!(
inner
.leases
.settle(lease_id, Settled::Released, plan.unspent),
"the plan validated this lease as active under this same lock"
);
let record = inner
.accounts
.get_mut(&plan.account_id)
.expect("lease account exists");
credit_settlement(record, plan.funding, plan.period_start, plan.unspent);
record.settlement_loss = record
.settlement_loss
.checked_add(plan.loss)
.expect("loss overflow");
}
fn plan_grant(
inner: &Inner,
policy: &GrantPolicy,
account: AccountId,
requested: CostUnits,
incoming: CostUnits,
needed: CostUnits,
attest_refusal: bool,
) -> Result<GrantPlan, AllocateError> {
let next_lease_id = inner
.next_lease_id
.checked_add(1)
.ok_or_else(|| AllocateError::Storage(StoreError("lease id overflow".into())))?;
let record = inner
.accounts
.get(&account)
.ok_or(AllocateError::UnknownAccount)?;
if record.status != AccountStatus::Active {
return Err(AllocateError::AccountInactive);
}
let balance = record
.balance
.total()
.checked_add(incoming)
.ok_or_else(|| AllocateError::Storage(StoreError("account balance overflow".into())))?;
let granted = policy
.consolidation_grant(requested, balance, incoming, needed)
.ok_or_else(|| {
if requested.is_zero() || !attest_refusal {
return AllocateError::InsufficientBalance;
}
funding_refusal(record.budget_view().shortfall())
})?;
let next_fence = record
.next_fence
.checked_add(1)
.ok_or_else(|| AllocateError::Storage(StoreError("fencing token overflow".into())))?;
Ok(GrantPlan {
granted,
fencing_token: FencingToken(record.next_fence),
next_fence,
next_lease_id,
})
}
fn funding_refusal(evidence: tollgate_core::BalanceShortfall) -> AllocateError {
match evidence.exhaustion() {
Some(exhausted) => AllocateError::BalanceExhausted(exhausted),
None => AllocateError::BalanceInsufficient(evidence),
}
}
fn allocation(inner: &Inner, grant: LeaseGrant) -> Allocation {
let funding = inner
.accounts
.get(&grant.account_id)
.expect("a granted account exists")
.budget_view()
.shortfall();
Allocation {
grant,
funding: Some(funding),
}
}
fn apply_grant(
inner: &mut Inner,
account: AccountId,
plan: GrantPlan,
expires_at: Timestamp,
) -> LeaseGrant {
let record = inner
.accounts
.get_mut(&account)
.expect("the plan validated this account under this same lock");
let drawn = record
.balance
.take(plan.granted)
.expect("grant never exceeds the balance the plan sized it against");
let period_start = record.period_start;
record.next_fence = plan.next_fence;
inner.next_lease_id = plan.next_lease_id;
let lease_id = LeaseId(plan.next_lease_id);
inner.leases.open(
lease_id,
LeaseRecord::opened(account, plan.fencing_token, plan.granted, expires_at)
.funded_by(drawn, period_start),
);
LeaseGrant {
lease_id,
account_id: account,
fencing_token: plan.fencing_token,
units: plan.granted,
expires_at,
}
}
#[async_trait]
impl LeaseAllocator for MemoryStore {
async fn acquire(
&self,
account: AccountId,
requested: CostUnits,
ttl: SignedDuration,
now: Timestamp,
) -> Result<Allocation, AllocateError> {
let expires_at = self.grant_expiry(ttl, now)?;
let mut inner = self.lock();
let plan = plan_grant(
&inner,
&self.policy,
account,
requested,
CostUnits::ZERO,
CostUnits::ZERO,
true,
)?;
let grant = apply_grant(&mut inner, account, plan, expires_at);
Ok(allocation(&inner, grant))
}
async fn release(
&self,
lease_id: LeaseId,
fencing_token: FencingToken,
unspent: CostUnits,
now: Timestamp,
) -> Result<(), AllocateError> {
let mut inner = self.lock();
let plan = plan_release(&inner, &self.policy, lease_id, fencing_token, unspent, now)?;
apply_release(&mut inner, lease_id, &plan);
Ok(())
}
async fn consolidate(
&self,
lease_id: LeaseId,
fencing_token: FencingToken,
unspent: CostUnits,
requested: CostUnits,
needed: CostUnits,
ttl: SignedDuration,
now: Timestamp,
) -> Result<Allocation, AllocateError> {
let expires_at = self.grant_expiry(ttl, now)?;
let mut inner = self.lock();
let release = plan_release(&inner, &self.policy, lease_id, fencing_token, unspent, now)?;
let account = release.account_id;
let record = inner
.accounts
.get(&account)
.ok_or(AllocateError::UnknownAccount)?;
let restored = spendable_credit(record, release.funding, release.period_start, unspent);
let preserves_funding = release.loss.is_zero() && restored == unspent;
let grant = plan_grant(
&inner,
&self.policy,
account,
requested,
restored,
needed,
preserves_funding,
)?;
apply_release(&mut inner, lease_id, &release);
let grant = apply_grant(&mut inner, account, grant, expires_at);
Ok(allocation(&inner, grant))
}
async fn reclaim_expired_batch(
&self,
now: Timestamp,
limit: NonZeroUsize,
) -> Result<ReclaimBatch, StoreError> {
let mut inner = self.lock();
let expired = inner
.leases
.reclaimable(self.policy.reclaim_cutoff(now), limit.get());
let mut reclaimed = Vec::with_capacity(expired.len());
for &lease_id in &expired {
let lease = inner.leases.get(lease_id).expect("just listed");
let forfeited = lease
.granted
.checked_sub(lease.used)
.expect("usage never exceeds grant");
reclaimed.push(ReclaimedLease {
lease_id,
account_id: lease.account_id,
forfeited,
});
}
let batch = ReclaimBatch::try_new(reclaimed, limit)?;
for entry in batch.reclaimed() {
assert!(
inner
.leases
.settle(entry.lease_id, Settled::Expired, CostUnits::ZERO),
"reclaimable only yields active leases"
);
let record = inner
.accounts
.get_mut(&entry.account_id)
.expect("lease account exists");
record.settlement_loss = record
.settlement_loss
.checked_add(entry.forfeited)
.expect("loss overflow");
}
if batch.reclaimed().len() < limit.get() {
let held = StoredRecords {
credential_activity: inner.credential_activity.len(),
usage_events: inner.usage.len(),
leases: inner.leases.len(),
active_leases: inner.leases.active_len(),
};
tracing::debug!(
credential_activity = held.credential_activity,
usage_events = held.usage_events,
leases = held.leases,
active_leases = held.active_leases,
"memory store holdings; usage events and settled leases are never reclaimed"
);
}
Ok(batch)
}
}
fn publish_locked(
inner: &mut Inner,
principal: Principal,
snapshot: PublishableSnapshot,
) -> Result<Option<PublishableSnapshot>, PublishSnapshotError> {
if let Some(key_id) = snapshot.key_id
&& !inner.keys.get(&key_id).is_some_and(|stored| {
stored.record.principal == principal && stored.record.account_id == snapshot.account_id
})
{
return Err(PublishSnapshotError::CredentialMismatch { key_id });
}
if let Some(record) = inner.accounts.get(&snapshot.account_id) {
if record.status != snapshot.status {
return Err(PublishSnapshotError::StatusMismatch {
ledger: record.status,
submitted: snapshot.status,
});
}
if record.capacity_class != snapshot.capacity_class {
return Err(PublishSnapshotError::CapacityClassMismatch {
ledger: record.capacity_class,
submitted: snapshot.capacity_class,
});
}
}
if let Some(existing) = inner.snapshots.get(&principal)
&& existing.generation() >= snapshot.generation
{
return Ok(None);
}
let view = inner
.accounts
.get(&snapshot.account_id)
.map(AccountRecord::budget_view);
let snapshot = snapshot.with_budget(view);
inner
.snapshots
.insert(principal, SnapshotRecord::Present(snapshot.clone()));
Ok(Some(snapshot))
}
fn remove_locked(inner: &mut Inner, principal: Principal) -> AdminReceipt<()> {
let before = snapshot_audit(inner.snapshots.get(&principal));
if let Some(SnapshotRecord::Present(snapshot)) = inner.snapshots.get(&principal) {
let generation = snapshot.generation;
inner
.snapshots
.insert(principal, SnapshotRecord::Revoked(generation));
}
AdminReceipt::new((), before, snapshot_audit(inner.snapshots.get(&principal)))
}
fn account_key(
inner: &Inner,
account: AccountId,
key: KeyId,
) -> Result<&StoredKey, KeySnapshotError> {
inner
.keys
.get(&key)
.filter(|stored| stored.record.account_id == account)
.ok_or(KeySnapshotError::UnknownCredential)
}
#[derive(Debug, Clone, Copy)]
enum Restamp {
Status(AccountStatus),
CapacityClass(CapacityClass),
}
impl Restamp {
fn already_applied(self, snapshot: &tollgate_core::AccountSnapshot) -> bool {
match self {
Restamp::Status(status) => snapshot.status == status,
Restamp::CapacityClass(class) => snapshot.capacity_class == class,
}
}
fn apply(self, snapshot: &PublishableSnapshot, generation: Generation) -> PublishableSnapshot {
match self {
Restamp::Status(status) => snapshot.restamped(status, generation),
Restamp::CapacityClass(class) => snapshot.reclassified(class, generation),
}
}
}
fn plan_republish(
inner: &Inner,
account: AccountId,
restamp: Restamp,
) -> Result<Vec<(Principal, PublishableSnapshot)>, SetStatusError> {
let mut planned = Vec::new();
for (principal, record) in &inner.snapshots {
let SnapshotRecord::Present(snapshot) = record else {
continue;
};
if snapshot.account_id != account || restamp.already_applied(snapshot) {
continue;
}
let generation = snapshot
.generation
.0
.checked_add(1)
.map(Generation)
.ok_or_else(|| {
SetStatusError::Storage(StoreError("snapshot generation overflow".into()))
})?;
planned.push((*principal, restamp.apply(snapshot, generation)));
}
planned.sort_unstable_by_key(|(principal, _)| *principal);
Ok(planned)
}
fn apply_republish(
inner: &mut Inner,
planned: Vec<(Principal, PublishableSnapshot)>,
) -> Vec<(Principal, PublishableSnapshot)> {
let mut pushes = Vec::with_capacity(planned.len());
for (principal, snapshot) in planned {
inner
.snapshots
.insert(principal, SnapshotRecord::Present(snapshot.clone()));
pushes.push((principal, snapshot));
}
pushes
}
#[async_trait]
impl StoreHealth for MemoryStore {
async fn ping(&self) -> Result<(), StoreError> {
Ok(())
}
}
#[async_trait]
impl AdminStore for MemoryStore {
async fn create_account(
&self,
config: AccountConfig,
) -> Result<crate::AdminReceipt<()>, CreateAccountError> {
self.create_account_audited(config)
}
async fn deposit(
&self,
account: AccountId,
units: CostUnits,
) -> Result<crate::AdminReceipt<()>, AllocateError> {
self.deposit_audited(account, units)
}
async fn set_budget_schedule(
&self,
account: AccountId,
schedule: Option<BudgetSchedule>,
) -> Result<crate::AdminReceipt<()>, BudgetError> {
let mut inner = self.lock();
let record = inner
.accounts
.get_mut(&account)
.ok_or(BudgetError::UnknownAccount)?;
let before = AdminState::Budget {
schedule: record.schedule,
};
record.schedule = schedule;
Ok(crate::AdminReceipt::new(
(),
before,
AdminState::Budget { schedule },
))
}
async fn roll_due_periods(
&self,
now: Timestamp,
limit: NonZeroUsize,
) -> Result<RolloverBatch, StoreError> {
let mut inner = self.lock();
let mut rolled = Vec::new();
#[allow(
clippy::disallowed_methods,
reason = "sorted below before the limit is applied, so the batch's membership is a function of the stored state"
)]
let mut due: Vec<(Timestamp, AccountId)> = inner
.accounts
.iter()
.filter_map(|(account_id, record)| {
let schedule = record.schedule?;
(schedule.period.start_of(now) > record.period_start)
.then_some((record.period_start, *account_id))
})
.collect();
due.sort_unstable();
due.truncate(limit.get());
for (_, account_id) in due {
let account_id = &account_id;
let record = inner
.accounts
.get_mut(account_id)
.expect("the due set was taken from this map under the same lock");
let Some(schedule) = record.schedule else {
continue;
};
let boundary = schedule.period.start_of(now);
if boundary <= record.period_start {
continue;
}
let expired = record.balance.allowance;
let deposited = record
.deposited
.checked_add(schedule.allowance)
.ok_or_else(|| StoreError(format!("deposit overflow for account {account_id}")))?;
let total_expired = record
.expired
.checked_add(expired)
.ok_or_else(|| StoreError(format!("expiry overflow for account {account_id}")))?;
record.deposited = deposited;
record.expired = total_expired;
record.balance.allowance = schedule.allowance;
record.period_start = boundary;
rolled.push(RolledAccount {
account_id: *account_id,
deposited: schedule.allowance,
expired,
});
}
RolloverBatch::try_new(rolled, limit)
}
async fn set_account_status(
&self,
account: AccountId,
status: AccountStatus,
) -> Result<crate::AdminReceipt<StatusChange>, SetStatusError> {
let (republished, before) = {
let mut inner = self.lock();
let record = inner
.accounts
.get(&account)
.ok_or(SetStatusError::UnknownAccount)?;
if record.status == AccountStatus::Closed && status != AccountStatus::Closed {
return Err(SetStatusError::AccountClosed);
}
let before = AdminState::Status {
status: record.status,
};
let planned = plan_republish(&inner, account, Restamp::Status(status))?;
inner
.accounts
.get_mut(&account)
.expect("the account was found under this same guard")
.status = status;
(apply_republish(&mut inner, planned), before)
};
if pushes_exceed_capacity(republished.len()) {
tracing::warn!(
%account,
principals = republished.len(),
capacity = PUSH_CHANNEL_CAPACITY,
"status change emitted more pushes than the channel holds; subscribers will resync"
);
}
let planned_pushes = republished;
let republished = planned_pushes.len();
for (principal, snapshot) in planned_pushes {
self.push_to_subscribers(SnapshotPush {
principal,
resolution: SnapshotResolution::Present(snapshot),
});
}
Ok(AdminReceipt::new(
StatusChange {
republished,
unreadable: 0,
},
before,
AdminState::Status { status },
))
}
async fn set_capacity_class(
&self,
account: AccountId,
class: CapacityClass,
) -> Result<crate::AdminReceipt<StatusChange>, SetStatusError> {
let (republished, before) = {
let mut inner = self.lock();
let record = inner
.accounts
.get(&account)
.ok_or(SetStatusError::UnknownAccount)?;
if record.status == AccountStatus::Closed {
return Err(SetStatusError::AccountClosed);
}
let before = AdminState::CapacityClass {
capacity_class: record.capacity_class,
};
let planned = plan_republish(&inner, account, Restamp::CapacityClass(class))?;
inner
.accounts
.get_mut(&account)
.expect("the account was found under this same guard")
.capacity_class = class;
(apply_republish(&mut inner, planned), before)
};
if pushes_exceed_capacity(republished.len()) {
tracing::warn!(
%account,
principals = republished.len(),
capacity = PUSH_CHANNEL_CAPACITY,
"capacity class change emitted more pushes than the channel holds; \
subscribers will resync"
);
}
let planned_pushes = republished;
let republished = planned_pushes.len();
for (principal, snapshot) in planned_pushes {
self.push_to_subscribers(SnapshotPush {
principal,
resolution: SnapshotResolution::Present(snapshot),
});
}
Ok(AdminReceipt::new(
StatusChange {
republished,
unreadable: 0,
},
before,
AdminState::CapacityClass {
capacity_class: class,
},
))
}
async fn publish_snapshot(
&self,
principal: Principal,
snapshot: PublishableSnapshot,
) -> Result<crate::AdminReceipt<()>, PublishSnapshotError> {
let (published, before, after) = {
let mut inner = self.lock();
let before = snapshot_audit(inner.snapshots.get(&principal));
let published = publish_locked(&mut inner, principal, snapshot)?;
(
published,
before,
snapshot_audit(inner.snapshots.get(&principal)),
)
};
if let Some(snapshot) = published {
self.push_to_subscribers(SnapshotPush {
principal,
resolution: SnapshotResolution::Present(snapshot),
});
}
Ok(AdminReceipt::new((), before, after))
}
async fn account_view(&self, account: AccountId) -> Result<Option<AccountView>, StoreError> {
let inner = self.lock();
let Some(record) = inner.accounts.get(&account) else {
return Ok(None);
};
let mut active_grants = CostUnits::ZERO;
let mut active_used = CostUnits::ZERO;
for lease in inner.leases.active_of(account) {
active_grants = active_grants
.checked_add(lease.granted)
.ok_or_else(|| StoreError(format!("grant sum overflow for account {account}")))?;
active_used = active_used
.checked_add(lease.used)
.ok_or_else(|| StoreError(format!("used sum overflow for account {account}")))?;
}
Ok(Some(AccountView {
account_id: account,
status: record.status,
capacity_class: record.capacity_class,
schedule: record.schedule,
period_start: record.period_start,
conservation: Conservation {
deposited: record.deposited,
overage_recorded: record.overage_recorded,
balance: record.balance.total(),
active_lease_grants: active_grants,
settled_usage: record
.usage_recorded
.checked_sub(active_used)
.ok_or_else(|| {
StoreError(format!("active usage exceeds recorded for {account}"))
})?,
settlement_loss: record.settlement_loss,
expired: record.expired,
},
}))
}
async fn remove_snapshot(
&self,
principal: Principal,
) -> Result<crate::AdminReceipt<()>, StoreError> {
Ok(self.remove_snapshot_audited(principal))
}
}
#[async_trait]
impl SnapshotSource for MemoryStore {
async fn snapshot(&self, principal: Principal) -> Result<SnapshotResolution, StoreError> {
Ok(match self.lock().snapshots.get(&principal) {
Some(SnapshotRecord::Present(snapshot)) => {
SnapshotResolution::Present(snapshot.clone())
}
Some(SnapshotRecord::Revoked(generation)) => SnapshotResolution::Revoked {
generation: *generation,
},
None => SnapshotResolution::Unknown,
})
}
fn subscribe(&self) -> broadcast::Receiver<SnapshotPush> {
self.push.subscribe()
}
async fn principals(&self) -> Result<Option<Vec<Principal>>, StoreError> {
#[allow(
clippy::disallowed_methods,
reason = "sorted on the next line before it leaves the function, so the hash order never reaches the caller"
)]
let mut principals: Vec<Principal> = self.lock().snapshots.keys().copied().collect();
principals.sort_unstable_by_key(|principal| principal.0);
Ok(Some(principals))
}
}
#[async_trait]
impl UsageSink for MemoryStore {
async fn ingest(
&self,
events: &[UsageEvent],
_now: Timestamp,
) -> Result<IngestReport, IngestError> {
let mut inner = self.lock();
let mut report = IngestReport::default();
let mut accepted: Vec<UsageEvent> = Vec::new();
let mut lease_used: HashMap<LeaseId, CostUnits> = HashMap::new();
let mut usage_recorded: HashMap<AccountId, CostUnits> = HashMap::new();
let mut overage_recorded: HashMap<AccountId, CostUnits> = HashMap::new();
let mut settlement_loss: HashMap<AccountId, CostUnits> = HashMap::new();
let mut planned: HashSet<tollgate_core::RequestId> = HashSet::new();
for event in events {
if inner.usage.contains_key(&event.request_id) || planned.contains(&event.request_id) {
report.duplicate += 1;
continue;
}
let UsageSource::Leased {
lease_id,
fencing_token,
} = event.source
else {
let Some(record) = inner.accounts.get(&event.account_id) else {
report.rejected += 1;
continue;
};
let recorded = *usage_recorded
.get(&event.account_id)
.unwrap_or(&record.usage_recorded);
let overage = *overage_recorded
.get(&event.account_id)
.unwrap_or(&record.overage_recorded);
let (Some(next_recorded), Some(next_overage)) = (
recorded.checked_add(event.units),
overage.checked_add(event.units),
) else {
return Err(IngestError::Refused(StoreError(format!(
"overage accounting overflow for account {:#034x}: recorded usage {} \
and overage {} cannot absorb {}",
event.account_id.0, recorded, overage, event.units
))));
};
usage_recorded.insert(event.account_id, next_recorded);
overage_recorded.insert(event.account_id, next_overage);
planned.insert(event.request_id);
accepted.push(*event);
report.accepted += 1;
continue;
};
let Some(lease) = inner.leases.get(lease_id) else {
report.rejected += 1;
continue;
};
if lease.fencing_token != fencing_token || lease.account_id != event.account_id {
report.rejected += 1;
continue;
}
let was_settled = !lease.is_active();
let used = *lease_used.get(&lease_id).unwrap_or(&lease.used);
let capacity = used
.checked_add(lease.credited())
.and_then(|committed| lease.granted.checked_sub(committed));
let fits = matches!(capacity, Some(cap) if event.units <= cap);
if !fits {
report.rejected += 1;
continue;
}
let account_id = lease.account_id;
let record = inner
.accounts
.get(&account_id)
.expect("lease account exists");
let recorded = *usage_recorded
.get(&account_id)
.unwrap_or(&record.usage_recorded);
lease_used.insert(
lease_id,
used.checked_add(event.units).expect("fits within grant"),
);
usage_recorded.insert(
account_id,
recorded.checked_add(event.units).ok_or_else(|| {
IngestError::Refused(StoreError(format!(
"usage accounting overflow for account {:#034x}: recorded usage {} \
cannot absorb {}",
account_id.0, recorded, event.units
)))
})?,
);
if was_settled {
let loss = *settlement_loss
.get(&account_id)
.unwrap_or(&record.settlement_loss);
settlement_loss.insert(
account_id,
loss.checked_sub(event.units)
.expect("straggler fits within recorded loss"),
);
}
planned.insert(event.request_id);
accepted.push(*event);
report.accepted += 1;
}
let mut activity = HashMap::<KeyId, Timestamp>::new();
let mut unattributed = 0;
for event in &accepted {
if let Some(key_id) = event.key_id
&& inner
.keys
.get(&key_id)
.is_some_and(|key| key.record.account_id == event.account_id)
{
let at = crate::clock::timestamp_from_micros(event.occurred_at.as_microsecond())
.expect("truncating a valid Timestamp to microseconds stays representable");
activity
.entry(key_id)
.and_modify(|old| *old = (*old).max(at))
.or_insert(at);
} else {
unattributed += 1;
}
}
report.unattributed = Some(unattributed);
for (lease_id, used) in lease_used {
inner
.leases
.get_mut(lease_id)
.expect("planned against a lease that exists")
.used = used;
}
for (account_id, value) in usage_recorded {
inner
.accounts
.get_mut(&account_id)
.expect("planned against an account that exists")
.usage_recorded = value;
}
for (account_id, value) in overage_recorded {
inner
.accounts
.get_mut(&account_id)
.expect("planned against an account that exists")
.overage_recorded = value;
}
for (account_id, value) in settlement_loss {
inner
.accounts
.get_mut(&account_id)
.expect("planned against an account that exists")
.settlement_loss = value;
}
for event in accepted {
inner.usage.insert(event.request_id, event);
}
for (key_id, at) in activity {
inner
.credential_activity
.entry(key_id)
.and_modify(|old| *old = (*old).max(at))
.or_insert(at);
}
Ok(report)
}
}
#[async_trait]
impl KeyDirectory for MemoryStore {
async fn credential_activity(
&self,
keys: &[KeyId],
) -> Result<Vec<crate::CredentialActivity>, StoreError> {
let inner = self.lock();
Ok(keys
.iter()
.map(|&key_id| crate::CredentialActivity {
key_id,
state: if !inner.keys.contains_key(&key_id) {
crate::CredentialActivityState::Unknown
} else if let Some(&last_committed_at) = inner.credential_activity.get(&key_id) {
crate::CredentialActivityState::Committed { last_committed_at }
} else {
crate::CredentialActivityState::Unobserved
},
})
.collect())
}
async fn insert_key(&self, record: KeyRecord) -> Result<(), KeyError> {
let mut inner = self.lock();
insert_key_locked(&mut inner, record).map(|receipt| receipt.outcome)
}
async fn revoke_key(&self, key_id: KeyId, now: Timestamp) -> Result<Revocation, KeyError> {
self.revoke_key_audited(key_id, now)
.await
.map(|receipt| receipt.outcome)
}
async fn publish_key_snapshot(
&self,
account: AccountId,
key: KeyId,
snapshot: PublishableSnapshot,
) -> Result<crate::AdminReceipt<()>, KeySnapshotError> {
let (principal, published, before, after) = {
let mut inner = self.lock();
let stored = account_key(&inner, account, key)?;
if stored.revoked_at.is_some() {
return Err(KeySnapshotError::Retired { key_id: key });
}
let principal = stored.record.principal;
if snapshot.key_id != Some(key) {
return Err(PublishSnapshotError::CredentialMismatch { key_id: key }.into());
}
let before = snapshot_audit(inner.snapshots.get(&principal));
let published = publish_locked(&mut inner, principal, snapshot)?;
let after = snapshot_audit(inner.snapshots.get(&principal));
(principal, published, before, after)
};
if let Some(snapshot) = published {
self.push_to_subscribers(SnapshotPush {
principal,
resolution: SnapshotResolution::Present(snapshot),
});
}
Ok(AdminReceipt::new((), before, after))
}
async fn remove_key_snapshot(
&self,
account: AccountId,
key: KeyId,
) -> Result<crate::AdminReceipt<()>, KeySnapshotError> {
let (principal, receipt) = {
let mut inner = self.lock();
let principal = account_key(&inner, account, key)?.record.principal;
(principal, remove_locked(&mut inner, principal))
};
self.announce_removal(principal, &receipt);
Ok(receipt)
}
async fn revoke_key_audited(
&self,
key_id: KeyId,
now: Timestamp,
) -> Result<crate::AdminReceipt<Revocation>, KeyError> {
let mut inner = self.lock();
let Some(stored) = inner.keys.get(&key_id) else {
return Err(KeyError::UnknownKey);
};
let account_id = stored.record.account_id;
let before = AdminState::Credential {
account_id,
key_id,
revoked: stored.revoked_at.is_some(),
};
if stored.revoked_at.is_some() {
return Ok(crate::AdminReceipt::new(
Revocation::AlreadyRetired,
before,
before,
));
}
let revision = next_credential_revision(inner.credential_revision)?;
inner
.keys
.get_mut(&key_id)
.expect("key exists under the same guard")
.revoked_at = Some(now);
inner.unrevoked_keys.remove(&key_id);
inner.credential_revision = revision;
Ok(crate::AdminReceipt::new(
Revocation::Retired,
before,
AdminState::Credential {
account_id,
key_id,
revoked: true,
},
))
}
async fn active_keys(&self, now: Timestamp) -> Result<Vec<KeyRecord>, StoreError> {
let inner = self.lock();
let active: Vec<KeyRecord> = inner
.unrevoked_keys
.iter()
.map(|id| &inner.keys[id])
.filter(|stored| {
stored
.record
.not_after
.is_none_or(|not_after| now < not_after)
})
.map(|stored| stored.record.clone())
.collect();
Ok(active)
}
async fn account_keys(
&self,
account: AccountId,
after: Option<KeyId>,
limit: NonZeroUsize,
) -> Result<Vec<KeySummary>, StoreError> {
crate::validate_key_page_limit(limit)?;
let inner = self.lock();
#[allow(
clippy::disallowed_methods,
reason = "sorted below before the page is truncated, so the hash order never reaches the caller"
)]
let mut summaries: Vec<KeySummary> = inner
.keys
.values()
.filter(|stored| stored.record.account_id == account)
.filter(|stored| after.is_none_or(|cursor| stored.record.key_id > cursor))
.map(|stored| KeySummary {
key_id: stored.record.key_id,
not_after: stored.record.not_after,
revoked_at: stored.revoked_at,
})
.collect();
summaries.sort_unstable_by_key(|summary| summary.key_id);
summaries.truncate(limit.get());
Ok(summaries)
}
async fn insert_key_within(
&self,
record: KeyRecord,
max_active: NonZeroUsize,
now: Timestamp,
) -> Result<(), KeyError> {
self.insert_key_within_audited(record, max_active, now)
.await
.map(|receipt| receipt.outcome)
}
async fn insert_key_within_audited(
&self,
record: KeyRecord,
max_active: NonZeroUsize,
now: Timestamp,
) -> Result<crate::AdminReceipt<()>, KeyError> {
let mut inner = self.lock();
if !inner.accounts.contains_key(&record.account_id) {
return Err(KeyError::UnknownAccount);
}
if inner.keys.contains_key(&record.key_id)
|| inner.key_principals.contains_key(&record.principal)
{
return Err(KeyError::AlreadyExists);
}
#[allow(
clippy::disallowed_methods,
reason = "counts live credentials; a count does not depend on the order they are counted in"
)]
let live = inner
.keys
.values()
.filter(|stored| stored.record.account_id == record.account_id)
.filter(|stored| {
KeySummary {
key_id: stored.record.key_id,
not_after: stored.record.not_after,
revoked_at: stored.revoked_at,
}
.is_live(now)
})
.count();
if live >= max_active.get() {
return Err(KeyError::ActiveKeyLimit { limit: max_active });
}
insert_key_locked(&mut inner, record)
}
}
fn insert_key_locked(
inner: &mut Inner,
record: KeyRecord,
) -> Result<crate::AdminReceipt<()>, KeyError> {
if !inner.accounts.contains_key(&record.account_id) {
return Err(KeyError::UnknownAccount);
}
if inner.keys.contains_key(&record.key_id) {
return Err(KeyError::AlreadyExists);
}
if inner.key_principals.contains_key(&record.principal) {
return Err(KeyError::AlreadyExists);
}
let revision = next_credential_revision(inner.credential_revision)?;
inner.key_principals.insert(record.principal, record.key_id);
inner.unrevoked_keys.insert(record.key_id);
let after = AdminState::Credential {
account_id: record.account_id,
key_id: record.key_id,
revoked: false,
};
inner.keys.insert(
record.key_id,
StoredKey {
record,
revoked_at: None,
},
);
inner.credential_revision = revision;
Ok(crate::AdminReceipt::new((), AdminState::Absent, after))
}
fn next_credential_revision(revision: u64) -> Result<u64, KeyError> {
revision
.checked_add(1)
.filter(|next| *next <= crate::MAX_KEY_REVISION)
.ok_or_else(|| KeyError::Storage(StoreError("credential revision exhausted".into())))
}
#[async_trait]
impl crate::KeySource for MemoryStore {
async fn active_keys_page(
&self,
now: Timestamp,
after: Option<KeyId>,
limit: NonZeroUsize,
) -> Result<crate::KeyPage, StoreError> {
use std::ops::Bound::{Excluded, Unbounded};
crate::validate_key_page_limit(limit)?;
let inner = self.lock();
let lower = after.map_or(Unbounded, Excluded);
let mut records: Vec<crate::CredentialRecord> = inner
.unrevoked_keys
.range((lower, Unbounded))
.map(|id| &inner.keys[id].record)
.filter(|record| record.not_after.is_none_or(|end| now < end))
.take(limit.get() + 1)
.cloned()
.map(Into::into)
.collect();
let next_after = if records.len() > limit.get() {
records.pop();
records.last().map(|key| key.key_id)
} else {
None
};
crate::KeyPage::try_new(
inner.credential_revision,
now,
after,
limit,
records,
next_after,
)
}
}
fn snapshot_audit(record: Option<&SnapshotRecord>) -> AdminState {
match record {
None => AdminState::Absent,
Some(SnapshotRecord::Present(snapshot)) => AdminState::Snapshot {
generation: snapshot.generation,
revoked: false,
},
Some(SnapshotRecord::Revoked(generation)) => AdminState::Snapshot {
generation: *generation,
revoked: true,
},
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::num::NonZeroUsize;
#[tokio::test]
async fn credential_revision_overflow_preserves_records_and_both_indexes() {
use crate::KeySource;
let store = store_with(100);
let record = |id: u128| {
let mut digest = [0; 32];
digest[..16].copy_from_slice(&id.to_be_bytes());
KeyRecord {
key_id: KeyId(id),
account_id: ACCOUNT,
principal: Principal(id),
digest,
not_after: None,
}
};
store.insert_key(record(1)).await.unwrap();
store.inner.lock().unwrap().credential_revision = crate::MAX_KEY_REVISION - 1;
store.insert_key(record(2)).await.unwrap();
assert!(matches!(
store.insert_key(record(3)).await,
Err(KeyError::Storage(_))
));
assert!(matches!(
store.revoke_key(KeyId(1), t(100)).await,
Err(KeyError::Storage(_))
));
let page = store
.active_keys_page(t(100), None, crate::DEFAULT_KEY_PAGE_LIMIT)
.await
.unwrap();
assert_eq!(page.revision(), crate::MAX_KEY_REVISION);
assert_eq!(
page.records().iter().map(|k| k.key_id).collect::<Vec<_>>(),
vec![KeyId(1), KeyId(2)]
);
store.inner.lock().unwrap().credential_revision = 2;
store.insert_key(record(3)).await.unwrap(); }
const ACCOUNT: AccountId = AccountId(1);
const TTL: SignedDuration = SignedDuration::from_secs(60);
fn t(secs: i64) -> Timestamp {
Timestamp::from_second(secs).unwrap()
}
fn exact_grants() -> GrantPolicy {
GrantPolicy {
shrink_divisor: 1,
min_grant: CostUnits(1),
max_ttl: SignedDuration::from_secs(300),
reclaim_grace: SignedDuration::ZERO,
}
}
fn store_with(balance: u64) -> Arc<MemoryStore> {
let store = MemoryStore::new(exact_grants()).expect("policy is valid");
store.create_account(AccountConfig {
account_id: ACCOUNT,
initial_balance: CostUnits(balance),
status: AccountStatus::Active,
capacity_class: CapacityClass::Assured,
});
store
}
async fn examined_by_a_sweep_past(settled: usize) -> usize {
let store = store_with(1_000_000);
for _ in 0..settled {
let lease = store
.acquire(ACCOUNT, CostUnits(1), TTL, t(0))
.await
.expect("funded")
.grant;
store
.release(lease.lease_id, lease.fencing_token, lease.units, t(1))
.await
.expect("active");
}
let due = store
.acquire(ACCOUNT, CostUnits(1), TTL, t(0))
.await
.expect("funded")
.grant;
let before = store.leases_examined();
let batch = store
.reclaim_expired_batch(t(120), NonZeroUsize::new(64).unwrap())
.await
.expect("sweep");
assert_eq!(batch.len(), 1, "exactly the one expired lease is settled");
assert_eq!(batch.reclaimed()[0].lease_id, due.lease_id);
store.leases_examined() - before
}
#[tokio::test]
async fn sweeping_examines_only_live_leases() {
let few = examined_by_a_sweep_past(10).await;
let many = examined_by_a_sweep_past(1_000).await;
assert_eq!(
few, many,
"1,000 settled leases cost the sweep {many} record reads against \
{few} for 10; the sweep is walking history again"
);
assert!(
many <= 4,
"one live lease should not cost {many} record reads"
);
}
#[tokio::test]
async fn conservation_examines_only_live_leases() {
async fn examined_by_conservation_past(settled: usize) -> usize {
let store = store_with(1_000_000);
for _ in 0..settled {
let lease = store
.acquire(ACCOUNT, CostUnits(1), TTL, t(0))
.await
.expect("funded")
.grant;
store
.release(lease.lease_id, lease.fencing_token, lease.units, t(1))
.await
.expect("active");
}
let _live = store.acquire(ACCOUNT, CostUnits(5), TTL, t(0)).await;
let before = store.leases_examined();
let conservation = store.conservation(ACCOUNT).expect("account exists");
assert!(
conservation.holds(),
"conservation violated: {conservation:?}"
);
assert_eq!(conservation.active_lease_grants, CostUnits(5));
store.leases_examined() - before
}
let few = examined_by_conservation_past(10).await;
let many = examined_by_conservation_past(1_000).await;
assert_eq!(
few, many,
"conservation read {many} records against {few}; it is summing \
over settled leases again"
);
}
}