use std::collections::BTreeMap;
use std::sync::Arc;
use parking_lot::Mutex;
use super::fetch::FanoutRegistry;
use crate::facade::Pin;
use crate::keys::Key;
use crate::store::Store;
use crate::Error;
#[derive(Clone, Copy, PartialEq, Eq, Hash)]
pub struct PinToken {
bytes: [u8; 32],
}
impl PinToken {
pub fn generate() -> Result<PinToken, Error> {
let mut bytes = [0u8; 32];
getrandom::fill(&mut bytes)
.map_err(|e| Error::Io(format!("pin-token minting: the entropy source failed: {e}")))?;
Ok(PinToken { bytes })
}
pub fn as_bytes(&self) -> &[u8; 32] {
&self.bytes
}
pub fn to_hex(&self) -> String {
hex::encode(self.bytes)
}
pub fn from_hex(token_hex: &str) -> Option<PinToken> {
if token_hex.len() != 64 {
return None;
}
let bytes = hex::decode(token_hex).ok()?;
bytes.try_into().ok().map(|bytes| PinToken { bytes })
}
}
impl std::fmt::Debug for PinToken {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.write_str("PinToken(#redacted)")
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum RenewalOutcome {
Renewed,
UnknownToken,
DigestMismatch,
Expired,
}
impl RenewalOutcome {
pub fn present(&self) -> bool {
matches!(self, RenewalOutcome::Renewed)
}
}
pub trait PinConverter: Send + Sync + 'static {
fn convert(&self, key: &Key);
}
struct PinTicket {
namespace: String,
key: Key,
custody: PinCustody,
}
enum PinCustody {
Fleet { owner: Vec<u8> },
Held(Pin),
}
enum RenewalCustody {
Held,
Fleet { owner: Vec<u8> },
}
struct RenewalTarget {
namespace: String,
key: Key,
custody: RenewalCustody,
}
pub struct TokenRegistry {
tickets: Mutex<BTreeMap<[u8; 32], PinTicket>>,
max_entries: usize,
}
impl Default for TokenRegistry {
fn default() -> Self {
Self::new()
}
}
impl TokenRegistry {
pub const DEFAULT_CAP: usize = 65_536;
pub fn new() -> TokenRegistry {
TokenRegistry::with_cap(Self::DEFAULT_CAP)
}
pub fn with_cap(max_entries: usize) -> TokenRegistry {
TokenRegistry {
tickets: Mutex::new(BTreeMap::new()),
max_entries,
}
}
pub fn len(&self) -> usize {
self.tickets.lock().len()
}
pub fn is_empty(&self) -> bool {
self.tickets.lock().is_empty()
}
fn insert_ticket(
&self,
namespace: String,
key: Key,
custody: PinCustody,
) -> Result<PinToken, Error> {
let mut tickets = self.tickets.lock();
if tickets.len() >= self.max_entries {
return Err(Error::RegistryFull);
}
let token = PinToken::generate()?;
let previous = tickets.insert(
*token.as_bytes(),
PinTicket {
namespace,
key,
custody,
},
);
drop(tickets);
debug_assert!(previous.is_none(), "the CSPRNG minted a colliding token");
Ok(token)
}
pub fn mint_fleet(&self, namespace: &str, key: Key, owner: Vec<u8>) -> Result<PinToken, Error> {
self.insert_ticket(namespace.to_owned(), key, PinCustody::Fleet { owner })
}
pub fn mint_custodial(&self, namespace: &str, key: Key, guard: Pin) -> Result<PinToken, Error> {
self.insert_ticket(namespace.to_owned(), key, PinCustody::Held(guard))
}
pub fn holds(&self, token: &PinToken) -> bool {
self.tickets.lock().contains_key(token.as_bytes())
}
}
#[derive(Debug, Clone)]
pub struct OpsConfig {
pub fanout_max_subscribers: usize,
pub fanout_channel_bound: usize,
pub fetch_max_ranges: usize,
pub fetch_max_total_bytes: u64,
pub token_registry_max_entries: usize,
}
impl Default for OpsConfig {
fn default() -> OpsConfig {
OpsConfig {
fanout_max_subscribers: 64,
fanout_channel_bound: 8,
fetch_max_ranges: 256,
fetch_max_total_bytes: 64 * 1024 * 1024,
token_registry_max_entries: TokenRegistry::DEFAULT_CAP,
}
}
}
pub struct OpsContext {
store: Arc<Store>,
config: OpsConfig,
tokens: TokenRegistry,
converters: Mutex<BTreeMap<String, Arc<dyn PinConverter>>>,
fanout: FanoutRegistry,
}
impl OpsContext {
pub fn new(store: Arc<Store>, config: OpsConfig) -> OpsContext {
let fanout =
FanoutRegistry::new(config.fanout_max_subscribers, config.fanout_channel_bound);
let tokens = TokenRegistry::with_cap(config.token_registry_max_entries);
OpsContext {
store,
config,
tokens,
converters: Mutex::new(BTreeMap::new()),
fanout,
}
}
pub fn config(&self) -> &OpsConfig {
&self.config
}
pub fn store(&self) -> &Arc<Store> {
&self.store
}
pub fn tokens(&self) -> &TokenRegistry {
&self.tokens
}
pub fn fanout(&self) -> &FanoutRegistry {
&self.fanout
}
pub fn fleet_owner(&self) -> Option<Vec<u8>> {
self.store.fleet().map(|fleet| fleet.owner().to_vec())
}
pub fn bind_converter(&self, namespace: &str, converter: Arc<dyn PinConverter>) -> bool {
self.converters
.lock()
.insert(namespace.to_owned(), converter)
.is_some()
}
pub fn mint_fleet(&self, namespace: &str, key: Key, owner: &[u8]) -> Result<PinToken, Error> {
let Some(fleet) = self.store.fleet() else {
return Err(Error::Io(
"fleet-token minting refused: the bound store is not fleet-declared \
(ADR-012 §2) — a non-fleet serving node's put mints the custodial form, \
whose guard the registry holds"
.to_owned(),
));
};
if fleet.owner() != owner {
return Err(Error::Io(
"fleet-token minting refused: the owner is not the bound store's \
instance id (ADR-011's vocabulary: the pin row's identity half is \
the minting instance)"
.to_owned(),
));
}
self.tokens.mint_fleet(namespace, key, owner.to_vec())
}
pub fn mint_custodial(&self, namespace: &str, key: Key, guard: Pin) -> Result<PinToken, Error> {
if self.store.fleet().is_some() {
return Err(Error::Io(
"custodial-token minting refused: the bound store is fleet-declared \
(ADR-012 §2) — the fleet put path's guard drops at response completion, \
the pin row's expiry is the protection, and the mint is the fleet form"
.to_owned(),
));
}
self.tokens.mint_custodial(namespace, key, guard)
}
pub async fn renew(&self, namespace: &str, token: &PinToken, digest: &Key) -> RenewalOutcome {
let Some(target) = self.renewal_target(token) else {
return RenewalOutcome::UnknownToken;
};
if target.namespace != namespace {
return RenewalOutcome::UnknownToken;
}
if &target.key != digest {
return RenewalOutcome::DigestMismatch;
}
match target.custody {
RenewalCustody::Held => RenewalOutcome::Renewed,
RenewalCustody::Fleet { owner } => self.renew_fleet(&target.key, &owner).await,
}
}
#[cfg(feature = "postgres")]
async fn renew_fleet(&self, key: &Key, owner: &[u8]) -> RenewalOutcome {
use crate::facade::PinState;
match self.store.ops_pin_check(&key.key_bytes()).await {
Ok(PinState::Protected) => {}
Ok(PinState::Unprotected) => return RenewalOutcome::Expired,
Err(_) => return RenewalOutcome::Expired,
}
match self.store.ops_pin_renew(&key.key_bytes(), owner).await {
Ok(true) => RenewalOutcome::Renewed,
Ok(false) | Err(_) => RenewalOutcome::Expired,
}
}
#[cfg(not(feature = "postgres"))]
async fn renew_fleet(&self, key: &Key, owner: &[u8]) -> RenewalOutcome {
let _ = (key, owner);
RenewalOutcome::Expired
}
fn renewal_target(&self, token: &PinToken) -> Option<RenewalTarget> {
self.tokens
.tickets
.lock()
.get(token.as_bytes())
.map(|ticket| RenewalTarget {
namespace: ticket.namespace.clone(),
key: ticket.key.clone(),
custody: match &ticket.custody {
PinCustody::Held(_) => RenewalCustody::Held,
PinCustody::Fleet { owner } => RenewalCustody::Fleet {
owner: owner.clone(),
},
},
})
}
pub fn confirm(&self, namespace: &str, token: &PinToken) -> bool {
let ticket = {
let mut tickets = self.tokens.tickets.lock();
match tickets.remove(token.as_bytes()) {
Some(ticket) => ticket,
None => return false,
}
};
if ticket.namespace != namespace {
self.tokens.tickets.lock().insert(*token.as_bytes(), ticket);
return false;
}
let converter = {
let converters = self.converters.lock();
converters.get(&ticket.namespace).cloned()
};
let Some(converter) = converter else {
self.tokens.tickets.lock().insert(*token.as_bytes(), ticket);
return false;
};
match ticket.custody {
PinCustody::Held(pin) => self
.store
.gc()
.convert(pin, || converter.convert(&ticket.key)),
PinCustody::Fleet { .. } => converter.convert(&ticket.key),
}
true
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::ops::tests::{key_of, test_store, RecordingConverter};
#[test]
fn token_mints_unguessable_and_the_hex_wire_form_round_trips() {
let first = PinToken::generate().unwrap();
let second = PinToken::generate().unwrap();
assert_ne!(
first, second,
"CSPRNG minting never collides in the same process"
);
assert_eq!(first.as_bytes().len(), 32);
let token_hex = first.to_hex();
assert_eq!(token_hex.len(), 64);
assert!(token_hex
.chars()
.all(|c| c.is_ascii_digit() || ('a'..='f').contains(&c)));
assert_eq!(PinToken::from_hex(&token_hex), Some(first));
assert_eq!(PinToken::from_hex(&token_hex.to_uppercase()), Some(first));
let wrong_length = format!("{token_hex}ab");
assert_eq!(PinToken::from_hex("0"), None);
assert_eq!(PinToken::from_hex("non-hexadecimal-material!!"), None);
assert_eq!(PinToken::from_hex(&wrong_length), None);
assert_eq!(format!("{first:?}"), "PinToken(#redacted)");
assert!(!format!("{first:?}").contains(&token_hex));
}
#[test]
fn from_hex_rejects_wrong_length_input_before_decoding() {
assert_eq!(
PinToken::from_hex(&"a".repeat(1 << 20)),
None,
"a huge string is rejected at the length guard"
);
assert_eq!(PinToken::from_hex(&"a".repeat(65)), None);
assert_eq!(PinToken::from_hex(&"a".repeat(63)), None);
assert_eq!(PinToken::from_hex(""), None);
}
#[cfg(any(feature = "mem", feature = "sqlite"))]
#[tokio::test]
async fn custodial_custody_holds_the_guard_and_converts_at_the_confirmation() {
let store = test_store().await;
let ctx = OpsContext::new(Arc::clone(&store), OpsConfig::default());
let key = key_of(b"custodial");
let other = key_of(b"other-digest");
let unknown = PinToken::from_hex(&"0".repeat(64)).unwrap();
let pin = store.gc().acquire(&key).await;
let token = ctx.mint_custodial("ns", key.clone(), pin).unwrap();
assert_eq!(
store.gc().pin_count(&key),
1,
"the registry holds the guard"
);
assert!(ctx.tokens().holds(&token));
assert_eq!(ctx.renew("ns", &token, &key).await, RenewalOutcome::Renewed);
assert!(ctx.renew("ns", &token, &key).await.present());
assert_eq!(
ctx.renew("ns", &token, &other).await,
RenewalOutcome::DigestMismatch
);
assert_eq!(
ctx.renew("ns", &unknown, &key).await,
RenewalOutcome::UnknownToken
);
assert_eq!(
store.gc().pin_count(&key),
1,
"the refusals left the custody intact"
);
assert_eq!(
ctx.renew("other-namespace", &token, &key).await,
RenewalOutcome::UnknownToken,
"the namespace parity (ADR-017 §3): a namespace-mismatched renewal is the \
validity-domain refusal — the same 'not this token's domain' answer confirm's \
refusal is"
);
assert!(
ctx.tokens().holds(&token),
"the parity refusal kept the mapping intact"
);
assert!(
ctx.renew("ns", &token, &key).await.present(),
"a subsequent correct-namespace renewal succeeds (the parity check retires nothing)"
);
assert_eq!(
store.gc().pin_count(&key),
1,
"the parity refusals left the custody intact"
);
assert!(
!ctx.confirm("other-namespace", &token),
"the minting namespace mismatches"
);
assert!(
ctx.tokens().holds(&token),
"the refused confirm kept the mapping"
);
assert!(
!ctx.confirm("ns", &token),
"an unbound converter refuses, mapping intact"
);
assert!(ctx.tokens().holds(&token));
let converted = Arc::new(RecordingConverter {
converted: Mutex::new(Vec::new()),
});
assert!(!ctx.bind_converter("ns", Arc::clone(&converted) as Arc<dyn PinConverter>));
assert!(ctx.confirm("ns", &token));
assert_eq!(
converted.captured(),
vec![key.clone()],
"the converter received the mapping's key"
);
assert_eq!(
store.gc().pin_count(&key),
0,
"the ordered-step conversion released the held guard"
);
assert!(
!ctx.tokens().holds(&token),
"conversion-confirmed retires the mapping"
);
assert_eq!(
ctx.renew("ns", &token, &key).await,
RenewalOutcome::UnknownToken,
"a retired token's renewal probe is the validity-domain refusal"
);
assert!(
!ctx.confirm("ns", &token),
"a retired token confirms nowhere twice"
);
}
#[cfg(any(feature = "mem", feature = "sqlite"))]
#[tokio::test]
async fn fleet_custody_holds_no_guard_and_the_renewal_degrades_closed() {
let store = test_store().await;
let ctx = OpsContext::new(Arc::clone(&store), OpsConfig::default());
let key = key_of(b"fleet-form");
let other = key_of(b"other-fleet");
let token = ctx
.tokens()
.mint_fleet("ns", key.clone(), b"instance-owner".to_vec())
.unwrap();
assert_eq!(
store.gc().pin_count(&key),
0,
"the fleet form records the mapping; no guard rides it"
);
assert_eq!(
ctx.renew("ns", &token, &other).await,
RenewalOutcome::DigestMismatch
);
assert_eq!(
ctx.renew("ns", &token, &key).await,
RenewalOutcome::Expired,
"a fleet mapping's renewal reaches the store core's fleet host and, on a \
non-fleet-bound context shape, folds closed"
);
let converted = Arc::new(RecordingConverter {
converted: Mutex::new(Vec::new()),
});
ctx.bind_converter("ns", Arc::clone(&converted) as Arc<dyn PinConverter>);
assert!(ctx.confirm("ns", &token));
assert_eq!(
converted.captured(),
vec![key.clone()],
"the fleet conversion is the converter act itself (no guard to release)"
);
assert!(
!ctx.tokens().holds(&token),
"conversion-confirmed retires the mapping"
);
}
#[cfg(any(feature = "mem", feature = "sqlite"))]
#[tokio::test]
async fn mint_fleet_is_refused_over_a_non_fleet_store() {
let store = test_store().await;
let ctx = OpsContext::new(Arc::clone(&store), OpsConfig::default());
assert!(ctx
.mint_fleet("ns", key_of(b"attempted"), b"owner")
.is_err());
}
#[test]
fn fleet_owner_lookup_follows_the_bound_store_and_the_registry_mints_standalone() {
let tokens = TokenRegistry::new();
let without = PinToken::from_hex(&"7".repeat(64)).unwrap();
assert!(!tokens.holds(&without));
let minted = tokens
.mint_fleet("ns", key_of(b"standalone"), b"instance".to_vec())
.unwrap();
assert!(tokens.holds(&minted));
assert_ne!(minted, without);
}
#[cfg(any(feature = "mem", feature = "sqlite"))]
#[tokio::test]
async fn converter_bindings_are_per_namespace_and_rebinding_replaces() {
let store = test_store().await;
let ctx = OpsContext::new(Arc::clone(&store), OpsConfig::default());
let first = Arc::new(RecordingConverter {
converted: Mutex::new(Vec::new()),
});
let second = Arc::new(RecordingConverter {
converted: Mutex::new(Vec::new()),
});
assert!(!ctx.bind_converter("ns", Arc::clone(&first) as Arc<dyn PinConverter>));
assert!(ctx.bind_converter("ns", Arc::clone(&second) as Arc<dyn PinConverter>));
let pin = store.gc().acquire(&key_of(b"rebind")).await;
let token = ctx.mint_custodial("ns", key_of(b"rebind"), pin).unwrap();
assert!(ctx.confirm("ns", &token));
assert!(
first.captured().is_empty(),
"the replaced binding never converted"
);
assert_eq!(second.captured(), vec![key_of(b"rebind")]);
}
#[cfg(any(feature = "mem", feature = "sqlite"))]
#[tokio::test]
async fn the_non_confirming_loop_hits_the_admission_cap_and_refusal_takes_no_custody() {
let store = test_store().await;
let config = OpsConfig {
token_registry_max_entries: 3,
..OpsConfig::default()
};
let ctx = OpsContext::new(Arc::clone(&store), config);
ctx.bind_converter(
"ns",
Arc::new(RecordingConverter {
converted: Mutex::new(Vec::new()),
}),
);
let mut minted = Vec::new();
for round in 0..3 {
let key = key_of(format!("loop-{round}").as_bytes());
let pin = store.gc().acquire(&key).await;
let token = ctx
.mint_custodial("ns", key.clone(), pin)
.unwrap_or_else(|e| panic!("the pre-cap mint (round {round}): {e}"));
assert_eq!(
store.gc().pin_count(&key),
1,
"each iteration's registry hold (round {round})"
);
minted.push((key, token));
}
assert_eq!(ctx.tokens().len(), 3, "the loop's entries registered");
let refused_key = key_of(b"loop-at-cap");
let pin = store.gc().acquire(&refused_key).await;
let refused = ctx.mint_custodial("ns", refused_key.clone(), pin);
assert_eq!(refused.unwrap_err(), Error::RegistryFull);
assert_eq!(
store.gc().pin_count(&refused_key),
0,
"no custody taken for the refused mint"
);
assert_eq!(
ctx.tokens().len(),
3,
"the refusal inserted nothing (never evicted)"
);
for (key, token) in &minted {
assert!(ctx.tokens().holds(token));
assert!(ctx.renew("ns", token, key).await.present());
}
assert_eq!(
ctx.tokens()
.mint_fleet("ns", key_of(b"fleet-at-cap"), b"owner".to_vec())
.unwrap_err(),
Error::RegistryFull
);
assert_eq!(ctx.tokens().len(), 3);
assert!(ctx.confirm("ns", &minted[0].1));
let resume_key = key_of(b"loop-resumes");
let pin = store.gc().acquire(&resume_key).await;
assert!(
ctx.mint_custodial("ns", resume_key, pin).is_ok(),
"a confirm's retirement frees the headroom"
);
}
}