use std::{
collections::HashMap,
path::Path,
sync::{Arc, Mutex, MutexGuard},
};
use kcode_k1_access_format::{AccessAction, decode, encode};
use kcode_k1_access_projection::{ApplyOutcome, Projection};
use kcode_k1_peering::K1Peering;
use kcode_k1_txn_ordering::{K1TxnOrdering, Subsystem as K1Subsystem};
pub use kcode_k1_access_projection::{
AccessCheck, AccessId, AccessRevision, Authorizations, GroupId, OwnerWitness, RequestPrincipal,
SubsystemId, Target, TxId, UserId,
};
const SUBSYSTEM_NAME: &str = "k1-access-subsystem";
type OperationId = [u8; 16];
struct Pending {
action: AccessAction,
result: Option<(TxId, ApplyOutcome)>,
}
struct DriverState {
available: bool,
pending: HashMap<OperationId, Pending>,
}
struct SharedState {
inner: Mutex<DriverState>,
}
impl SharedState {
fn new() -> Self {
Self {
inner: Mutex::new(DriverState {
available: true,
pending: HashMap::new(),
}),
}
}
fn lock(&self) -> Result<MutexGuard<'_, DriverState>, String> {
self.inner
.lock()
.map_err(|_| "k1 access driver state lock failed".to_owned())
}
fn ensure_available(&self) -> Result<(), String> {
if self.lock()?.available {
Ok(())
} else {
Err("k1 access driver unavailable".to_owned())
}
}
fn reserve(&self, action: AccessAction) -> Result<OperationId, String> {
self.ensure_available()?;
loop {
let mut operation_id = [0_u8; 16];
getrandom::fill(&mut operation_id)
.map_err(|error| format!("operation ID randomness failed: {error}"))?;
let mut state = self.lock()?;
if !state.available {
return Err("k1 access driver unavailable".to_owned());
}
if state.pending.contains_key(&operation_id) {
continue;
}
state
.pending
.try_reserve(1)
.map_err(|_| "pending operation allocation failed".to_owned())?;
state.pending.insert(
operation_id,
Pending {
action,
result: None,
},
);
return Ok(operation_id);
}
}
fn cancel(&self, operation_id: OperationId) -> Result<(), String> {
let mut state = self.lock()?;
if state.pending.remove(&operation_id).is_none() {
invalidate(&mut state);
return Err("pending operation disappeared".to_owned());
}
if state.available {
Ok(())
} else {
Err("k1 access driver unavailable".to_owned())
}
}
fn record(
&self,
operation_id: OperationId,
action: &AccessAction,
txid: TxId,
outcome: ApplyOutcome,
) -> Result<(), String> {
let mut state = self.lock()?;
if !state.available {
return Err("k1 access driver unavailable".to_owned());
}
let contradiction = match state.pending.get_mut(&operation_id) {
None => return Ok(()),
Some(pending) if &pending.action != action => {
Some("operation ID correlated with a different access action")
}
Some(pending) if pending.result.is_some() => {
Some("duplicate callback evidence for one operation ID")
}
Some(pending) => {
pending.result = Some((txid, outcome));
None
}
};
if let Some(error) = contradiction {
invalidate(&mut state);
Err(error.to_owned())
} else {
Ok(())
}
}
fn finish(
&self,
operation_id: OperationId,
submission: Result<TxId, String>,
) -> Result<AccessRevision, String> {
let result = {
let mut state = self.lock()?;
let Some(pending) = state.pending.remove(&operation_id) else {
invalidate(&mut state);
return Err("pending operation disappeared".to_owned());
};
if !state.available {
return Err("k1 access driver unavailable".to_owned());
}
pending.result
};
match submission {
Ok(submitted) => match result {
Some((callback, outcome)) if callback == submitted => outcome_result(outcome),
Some(_) => {
self.fault("submission transaction ID did not match callback transaction ID")
}
None => self.fault("successful submission had no synchronous callback"),
},
Err(error) => match result {
Some((_, outcome)) => outcome_result(outcome),
None => Err(error),
},
}
}
fn make_unavailable(&self) -> Result<(), String> {
let mut state = self.lock()?;
invalidate(&mut state);
Ok(())
}
fn fault<T>(&self, error: &str) -> Result<T, String> {
let _ = self.make_unavailable();
Err(error.to_owned())
}
}
fn invalidate(state: &mut DriverState) {
state.available = false;
for pending in state.pending.values_mut() {
pending.result = None;
}
}
fn outcome_result(outcome: ApplyOutcome) -> Result<AccessRevision, String> {
match outcome {
ApplyOutcome::Applied(revision) | ApplyOutcome::Unchanged(revision) => Ok(revision),
ApplyOutcome::Rejected(reason) => Err(reason),
}
}
struct AccessSubsystem {
projection: Arc<Projection>,
shared: Arc<SharedState>,
}
impl K1Subsystem for AccessSubsystem {
fn submit_txn(&self, id: TxId, payload: &[u8]) -> Result<(), String> {
let (operation_id, action) = match decode(payload) {
Ok(decoded) => decoded,
Err(error) => {
let _ = self.shared.make_unavailable();
return Err(error);
}
};
self.shared.ensure_available()?;
let outcome = match self.projection.apply(id, action.clone()) {
Ok(outcome) => outcome,
Err(error) => {
let _ = self.shared.make_unavailable();
return Err(error);
}
};
self.shared.record(operation_id, &action, id, outcome)
}
fn reorg(&self) -> Result<(), String> {
let state_result = self.shared.make_unavailable();
match self.projection.clear() {
Ok(()) => state_result,
Err(error) => Err(error),
}
}
}
pub struct K1AccessDriver {
projection: Arc<Projection>,
_ordering: Arc<K1TxnOrdering>,
peering: Arc<K1Peering>,
shared: Arc<SharedState>,
subsystem: SubsystemId,
}
impl K1AccessDriver {
pub fn open(
root: &Path,
ordering: Arc<K1TxnOrdering>,
peering: Arc<K1Peering>,
) -> Result<Self, String> {
let (projection, cursor) = Projection::open(root, &ordering)?;
let projection = Arc::new(projection);
let shared = Arc::new(SharedState::new());
let subsystem = SubsystemId::from_str(SUBSYSTEM_NAME)?;
let callback = Arc::new(AccessSubsystem {
projection: projection.clone(),
shared: shared.clone(),
});
if let Err(error) = ordering.register_subsystem(subsystem, cursor, callback) {
let _ = shared.make_unavailable();
return Err(error);
}
Ok(Self {
projection,
_ordering: ordering,
peering,
shared,
subsystem,
})
}
pub fn create(
&self,
target: Target,
authorizations: Authorizations,
) -> Result<AccessRevision, String> {
self.mutate(AccessAction::Create {
target,
authorizations,
})
}
pub fn replace(
&self,
access_id: AccessId,
actor: UserId,
groups_revision: Option<TxId>,
witness: OwnerWitness,
authorizations: Authorizations,
) -> Result<AccessRevision, String> {
self.mutate(AccessAction::Replace {
access_id,
actor,
groups_revision,
witness,
authorizations,
})
}
pub fn ensure_discovery(&self, access_id: AccessId) -> Result<AccessRevision, String> {
self.mutate(AccessAction::EnsureDiscovery { access_id })
}
pub fn discovery_missing(
&self,
access_id: AccessId,
expected_subsystem: SubsystemId,
) -> Result<bool, String> {
self.projection_call(
self.projection
.discovery_missing(access_id, expected_subsystem),
)
}
pub fn discovered_for_user(
&self,
user: UserId,
expected_subsystem: SubsystemId,
) -> Result<Vec<AccessId>, String> {
self.projection_call(
self.projection
.discovered_for_user(user, expected_subsystem),
)
}
pub fn discovered_for_group(
&self,
group: GroupId,
expected_subsystem: SubsystemId,
) -> Result<Vec<AccessId>, String> {
self.projection_call(
self.projection
.discovered_for_group(group, expected_subsystem),
)
}
pub fn owner_witness(
&self,
access_id: AccessId,
user: UserId,
user_groups: &[GroupId],
) -> Result<Option<OwnerWitness>, String> {
self.projection_call(self.projection.owner_witness(access_id, user, user_groups))
}
pub fn check(
&self,
principal: RequestPrincipal,
access_id: AccessId,
expected_subsystem: SubsystemId,
user_groups: &[GroupId],
model_groups: &[GroupId],
groups_revision: Option<TxId>,
) -> Result<AccessCheck, String> {
self.projection_call(self.projection.check(
principal,
access_id,
expected_subsystem,
user_groups,
model_groups,
groups_revision,
))
}
fn mutate(&self, action: AccessAction) -> Result<AccessRevision, String> {
let operation_id = self.shared.reserve(action.clone())?;
let payload = match encode(operation_id, &action) {
Ok(payload) => payload,
Err(error) => {
self.shared.cancel(operation_id)?;
return Err(error);
}
};
if let Err(error) = self.shared.ensure_available() {
let _ = self.shared.cancel(operation_id);
return Err(error);
}
let submission = self.peering.submit_txn(self.subsystem, &payload);
self.shared.finish(operation_id, submission)
}
fn projection_call<T>(&self, result: Result<T, String>) -> Result<T, String> {
match result {
Ok(value) => {
self.shared.ensure_available()?;
Ok(value)
}
Err(error) => {
let _ = self.shared.make_unavailable();
Err(error)
}
}
}
}
#[cfg(test)]
mod tests {
use super::*;
fn tx(byte: u8) -> TxId {
TxId::from_bytes([byte; 12])
}
fn action(byte: u8) -> AccessAction {
let owner = UserId::from_tx_id(tx(byte));
AccessAction::Create {
target: Target::new(SubsystemId::from_str("test").unwrap(), vec![byte]),
authorizations: Authorizations::new(
vec![kcode_k1_access_projection::OwnerSubject::User(owner)],
Vec::new(),
)
.unwrap(),
}
}
fn pending(shared: &SharedState, operation_id: OperationId, action: AccessAction) {
shared.lock().unwrap().pending.insert(
operation_id,
Pending {
action,
result: None,
},
);
}
#[test]
fn recorded_callback_controls_submission_resolution() {
let shared = SharedState::new();
let revision = AccessRevision::new(AccessId::new(tx(2)), tx(3));
let action = action(1);
pending(&shared, [1; 16], action.clone());
shared
.record([1; 16], &action, tx(4), ApplyOutcome::Applied(revision))
.unwrap();
assert_eq!(shared.finish([1; 16], Ok(tx(4))), Ok(revision));
pending(&shared, [2; 16], action.clone());
shared
.record(
[2; 16],
&action,
tx(5),
ApplyOutcome::Rejected("exact rejection".to_owned()),
)
.unwrap();
assert_eq!(
shared.finish([2; 16], Err("committed".to_owned())),
Err("exact rejection".to_owned())
);
}
#[test]
fn missing_mismatched_and_duplicate_evidence_fault() {
let missing = SharedState::new();
pending(&missing, [1; 16], action(1));
assert!(missing.finish([1; 16], Ok(tx(1))).is_err());
assert!(missing.ensure_available().is_err());
let mismatched = SharedState::new();
pending(&mismatched, [2; 16], action(2));
assert!(
mismatched
.record(
[2; 16],
&action(3),
tx(2),
ApplyOutcome::Rejected("remote".to_owned()),
)
.is_err()
);
assert!(mismatched.ensure_available().is_err());
let duplicate = SharedState::new();
let exact = action(4);
let revision = AccessRevision::new(AccessId::new(tx(4)), tx(5));
pending(&duplicate, [3; 16], exact.clone());
duplicate
.record([3; 16], &exact, tx(5), ApplyOutcome::Applied(revision))
.unwrap();
assert!(
duplicate
.record([3; 16], &exact, tx(5), ApplyOutcome::Applied(revision),)
.is_err()
);
assert!(duplicate.ensure_available().is_err());
}
#[test]
fn ensure_uses_the_exact_action() {
let access_id = AccessId::new(tx(6));
let action = AccessAction::EnsureDiscovery { access_id };
let shared = SharedState::new();
let revision = AccessRevision::new(access_id, tx(5));
pending(&shared, [6; 16], action.clone());
shared
.record([6; 16], &action, tx(7), ApplyOutcome::Unchanged(revision))
.unwrap();
assert_eq!(shared.finish([6; 16], Ok(tx(7))), Ok(revision));
}
}