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_types::{
AccessCheck, AccessId, AccessRevision, Authorizations, GroupId, ModelId, OwnerSubject,
RequestPrincipal, Target, TxId, UserId, ViewerSubject,
};
pub use kcode_k1_transaction::SubsystemId;
use kcode_k1_groups::K1Groups;
const SUBSYSTEM_NAME: &str = "k1-access-subsystem";
type OperationId = [u8; 16];
struct Pending {
action: AccessAction,
result: Option<(TxId, ApplyOutcome)>,
}
struct FacadeState {
available: bool,
pending: HashMap<OperationId, Pending>,
}
struct SharedState {
inner: Mutex<FacadeState>,
}
impl SharedState {
fn new() -> Self {
Self {
inner: Mutex::new(FacadeState {
available: true,
pending: HashMap::new(),
}),
}
}
fn lock(&self) -> Result<MutexGuard<'_, FacadeState>, String> {
self.inner
.lock()
.map_err(|_| "k1 access state lock failed".to_owned())
}
fn ensure_available(&self) -> Result<(), String> {
if self.lock()?.available {
Ok(())
} else {
Err("k1 access instance 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 instance 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 instance 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 instance 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 instance 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 FacadeState) {
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 K1Access {
projection: Arc<Projection>,
_ordering: Arc<K1TxnOrdering>,
peering: Arc<K1Peering>,
groups: Arc<K1Groups>,
shared: Arc<SharedState>,
subsystem: SubsystemId,
}
impl K1Access {
pub fn open(
root: &Path,
ordering: Arc<K1TxnOrdering>,
peering: Arc<K1Peering>,
groups: Arc<K1Groups>,
) -> 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,
groups,
shared,
subsystem,
})
}
pub fn create(
&self,
target: Target,
authorizations: Authorizations,
) -> Result<AccessRevision, String> {
self.mutate(AccessAction::Create {
target,
authorizations,
})
}
pub fn set_authorizations(
&self,
principal: RequestPrincipal,
access_id: AccessId,
authorizations: Authorizations,
) -> Result<AccessRevision, String> {
self.shared.ensure_available()?;
let memberships = self.groups.memberships(principal.user(), principal.model());
self.shared.ensure_available()?;
let memberships = memberships?;
let witness = self.projection_call(self.projection.owner_witness(
access_id,
principal.user(),
memberships.user_groups(),
))?;
let witness = witness.ok_or_else(|| "principal is not an access owner".to_owned())?;
self.mutate(AccessAction::Replace {
access_id,
actor: principal.user(),
groups_revision: memberships.revision(),
witness,
authorizations,
})
}
pub fn check(
&self,
principal: RequestPrincipal,
access_id: AccessId,
expected_subsystem: SubsystemId,
) -> Result<AccessCheck, String> {
self.shared.ensure_available()?;
let memberships = self.groups.memberships(principal.user(), principal.model());
self.shared.ensure_available()?;
let memberships = memberships?;
self.projection_call(self.projection.check(
principal,
access_id,
expected_subsystem,
memberships.user_groups(),
memberships.model_groups(),
memberships.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![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 exact_callback_controls_submission_resolution() {
let shared = SharedState::new();
let operation_id = [1; 16];
let action = action(1);
let revision = AccessRevision::new(AccessId::new(tx(2)), tx(3));
pending(&shared, operation_id, action.clone());
shared
.record(
operation_id,
&action,
tx(4),
ApplyOutcome::Applied(revision),
)
.unwrap();
assert_eq!(shared.finish(operation_id, Ok(tx(4))), Ok(revision));
let operation_id = [2; 16];
pending(&shared, operation_id, action.clone());
shared
.record(
operation_id,
&action,
tx(5),
ApplyOutcome::Unchanged(revision),
)
.unwrap();
assert_eq!(
shared.finish(operation_id, Err("committed".to_owned())),
Ok(revision)
);
let operation_id = [3; 16];
pending(&shared, operation_id, action.clone());
shared
.record(
operation_id,
&action,
tx(6),
ApplyOutcome::Rejected("exact rejection".to_owned()),
)
.unwrap();
assert_eq!(
shared.finish(operation_id, 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());
}
}