use std::{
collections::HashMap,
path::Path,
sync::{Arc, Mutex},
};
use kcode_k1_groups_projection::{ApplyOutcome, GroupAction, Projection};
use kcode_k1_peering::K1Peering;
use kcode_k1_transaction::SubsystemId;
use kcode_k1_txn_ordering::{K1TxnOrdering, Subsystem as K1Subsystem};
pub use kcode_k1_groups_projection::{
Group, GroupId, GroupMemberships, GroupRevision, GroupRole, GroupUser, ModelId, TxId, UserId,
};
const SUBSYSTEM_NAME: &str = "k1-groups-subsystem";
const CREATE_LENGTH: usize = 30;
const SET_USER_ROLE_LENGTH: usize = 55;
const SET_MODEL_MEMBERSHIP_LENGTH: usize = 75;
type OperationId = [u8; 16];
#[derive(Clone, Debug, Eq, PartialEq)]
enum Action {
Create {
owner: UserId,
},
SetUserRole {
group: GroupId,
actor: UserId,
user: UserId,
role: Option<GroupRole>,
},
SetModelMembership {
group: GroupId,
actor: UserId,
model: ModelId,
present: bool,
},
}
impl Action {
fn projection_action(&self) -> GroupAction {
match self {
Self::Create { owner } => GroupAction::Create { owner: *owner },
Self::SetUserRole {
group,
actor,
user,
role,
} => GroupAction::SetUserRole {
group: *group,
actor: *actor,
user: *user,
role: *role,
},
Self::SetModelMembership {
group,
actor,
model,
present,
} => GroupAction::SetModelMembership {
group: *group,
actor: *actor,
model: *model,
present: *present,
},
}
}
fn payload(&self, operation_id: OperationId) -> Vec<u8> {
let capacity = match self {
Self::Create { .. } => CREATE_LENGTH,
Self::SetUserRole { .. } => SET_USER_ROLE_LENGTH,
Self::SetModelMembership { .. } => SET_MODEL_MEMBERSHIP_LENGTH,
};
let mut payload = Vec::with_capacity(capacity);
payload.extend_from_slice(&[1, self.kind()]);
payload.extend_from_slice(&operation_id);
match self {
Self::Create { owner } => payload.extend_from_slice(owner.as_tx_id().as_bytes()),
Self::SetUserRole {
group,
actor,
user,
role,
} => {
payload.extend_from_slice(group.txid().as_bytes());
payload.extend_from_slice(actor.as_tx_id().as_bytes());
payload.extend_from_slice(user.as_tx_id().as_bytes());
payload.push(role_byte(*role));
}
Self::SetModelMembership {
group,
actor,
model,
present,
} => {
payload.extend_from_slice(group.txid().as_bytes());
payload.extend_from_slice(actor.as_tx_id().as_bytes());
payload.extend_from_slice(model.as_bytes());
payload.push(u8::from(*present));
}
}
payload
}
fn kind(&self) -> u8 {
match self {
Self::Create { .. } => 1,
Self::SetUserRole { .. } => 2,
Self::SetModelMembership { .. } => 3,
}
}
}
#[derive(Clone)]
enum Resolution {
Applied(GroupRevision),
Unchanged(GroupRevision),
Rejected(String),
}
impl Resolution {
fn kind(&self) -> OutcomeKind {
match self {
Self::Applied(_) => OutcomeKind::Applied,
Self::Unchanged(_) => OutcomeKind::Unchanged,
Self::Rejected(_) => OutcomeKind::Rejected,
}
}
fn result(self) -> Result<GroupRevision, String> {
match self {
Self::Applied(revision) | Self::Unchanged(revision) => Ok(revision),
Self::Rejected(error) => Err(error),
}
}
}
impl From<ApplyOutcome> for Resolution {
fn from(outcome: ApplyOutcome) -> Self {
match outcome {
ApplyOutcome::Applied(revision) => Self::Applied(revision),
ApplyOutcome::Unchanged(revision) => Self::Unchanged(revision),
ApplyOutcome::Rejected(error) => Self::Rejected(error),
}
}
}
#[derive(Clone)]
struct CallbackRecord {
txid: TxId,
resolution: Resolution,
}
enum CallbackEvidence {
None,
One(CallbackRecord),
Ambiguous,
}
struct PendingOperation {
action: Action,
evidence: CallbackEvidence,
}
struct FacadeState {
available: bool,
pending: HashMap<OperationId, PendingOperation>,
}
struct SharedState {
inner: Mutex<FacadeState>,
}
impl SharedState {
fn new() -> Self {
Self {
inner: Mutex::new(FacadeState {
available: true,
pending: HashMap::new(),
}),
}
}
fn ensure_available(&self) -> Result<(), String> {
let state = self
.inner
.lock()
.map_err(|_| "k1 groups state lock failed".to_string())?;
if state.available {
Ok(())
} else {
Err("k1 groups instance unavailable".to_string())
}
}
fn reserve(&self, action: Action) -> 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
.inner
.lock()
.map_err(|_| "k1 groups state lock failed".to_string())?;
if !state.available {
return Err("k1 groups instance unavailable".to_string());
}
if state.pending.contains_key(&operation_id) {
continue;
}
state
.pending
.try_reserve(1)
.map_err(|_| "pending operation allocation failed".to_string())?;
state.pending.insert(
operation_id,
PendingOperation {
action,
evidence: CallbackEvidence::None,
},
);
return Ok(operation_id);
}
}
fn record(
&self,
operation_id: OperationId,
action: &Action,
txid: TxId,
resolution: Resolution,
) -> Result<(), String> {
let mut state = self
.inner
.lock()
.map_err(|_| "k1 groups state lock failed".to_string())?;
if !state.available {
return Err("k1 groups instance unavailable".to_string());
}
let contradiction = match state.pending.get_mut(&operation_id) {
None => return Ok(()),
Some(pending) if &pending.action != action => {
Some("operation ID collided with a different action")
}
Some(pending) => match pending.evidence {
CallbackEvidence::None => {
pending.evidence = CallbackEvidence::One(CallbackRecord { txid, resolution });
None
}
CallbackEvidence::One(_) | CallbackEvidence::Ambiguous => {
Some("duplicate callback evidence for one operation ID")
}
},
};
if let Some(error) = contradiction {
make_unavailable(&mut state);
Err(error.to_string())
} else {
Ok(())
}
}
fn finish(
&self,
operation_id: OperationId,
submission: Result<TxId, String>,
) -> Result<GroupRevision, String> {
let evidence = {
let mut state = self
.inner
.lock()
.map_err(|_| "k1 groups state lock failed".to_string())?;
let pending = state.pending.remove(&operation_id);
if !state.available {
CallbackEvidence::Ambiguous
} else {
pending
.map(|pending| pending.evidence)
.unwrap_or(CallbackEvidence::Ambiguous)
}
};
let summary = evidence.summary();
match reconcile_decision(&submission, summary) {
ReconcileDecision::UseCallback(expected_kind) => match evidence {
CallbackEvidence::One(record) if record.resolution.kind() == expected_kind => {
record.resolution.result()
}
_ => self.fail_resolution("callback reconciliation state contradiction"),
},
ReconcileDecision::UseSubmissionError => match submission {
Err(error) => Err(error),
Ok(_) => self.fail_resolution("submission reconciliation state contradiction"),
},
ReconcileDecision::Fault(error) => self.fail_resolution(error),
}
}
fn fail_resolution<T>(&self, error: &str) -> Result<T, String> {
let _ = self.make_unavailable();
Err(error.to_string())
}
fn make_unavailable(&self) -> Result<(), String> {
let mut state = self
.inner
.lock()
.map_err(|_| "k1 groups state lock failed".to_string())?;
make_unavailable(&mut state);
Ok(())
}
}
fn make_unavailable(state: &mut FacadeState) {
state.available = false;
for pending in state.pending.values_mut() {
pending.evidence = CallbackEvidence::Ambiguous;
}
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
enum OutcomeKind {
Applied,
Unchanged,
Rejected,
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
enum EvidenceSummary {
None,
One { txid: TxId, kind: OutcomeKind },
Ambiguous,
}
impl CallbackEvidence {
fn summary(&self) -> EvidenceSummary {
match self {
Self::None => EvidenceSummary::None,
Self::One(record) => EvidenceSummary::One {
txid: record.txid,
kind: record.resolution.kind(),
},
Self::Ambiguous => EvidenceSummary::Ambiguous,
}
}
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
enum ReconcileDecision {
UseCallback(OutcomeKind),
UseSubmissionError,
Fault(&'static str),
}
fn reconcile_decision(
submission: &Result<TxId, String>,
evidence: EvidenceSummary,
) -> ReconcileDecision {
match evidence {
EvidenceSummary::Ambiguous => {
ReconcileDecision::Fault("ambiguous callback evidence made instance unavailable")
}
EvidenceSummary::None => {
if submission.is_ok() {
ReconcileDecision::Fault("successful submission had no synchronous callback")
} else {
ReconcileDecision::UseSubmissionError
}
}
EvidenceSummary::One { txid, kind } => match submission {
Ok(returned) if *returned == txid => ReconcileDecision::UseCallback(kind),
Ok(_) => ReconcileDecision::Fault(
"submission transaction ID did not match callback transaction ID",
),
Err(_) => ReconcileDecision::UseCallback(kind),
},
}
}
struct GroupsSubsystem {
projection: Arc<Projection>,
shared: Arc<SharedState>,
}
impl K1Subsystem for GroupsSubsystem {
fn submit_txn(&self, id: TxId, payload: &[u8]) -> Result<(), String> {
let (operation_id, action) = match parse_payload(payload) {
Ok(parsed) => parsed,
Err(error) => {
let _ = self.shared.make_unavailable();
return Err(error);
}
};
self.shared.ensure_available()?;
let outcome = match self.projection.apply(id, action.projection_action()) {
Ok(outcome) => outcome,
Err(error) => {
let _ = self.shared.make_unavailable();
return Err(error);
}
};
self.shared
.record(operation_id, &action, id, outcome.into())
}
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 K1Groups {
projection: Arc<Projection>,
_ordering: Arc<K1TxnOrdering>,
peering: Arc<K1Peering>,
shared: Arc<SharedState>,
subsystem: SubsystemId,
}
impl K1Groups {
pub fn open(
root: &Path,
ordering: Arc<K1TxnOrdering>,
peering: Arc<K1Peering>,
) -> Result<Self, String> {
let (projection, checkpoint) = 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(GroupsSubsystem {
projection: projection.clone(),
shared: shared.clone(),
});
if let Err(error) = ordering.register_subsystem(subsystem, checkpoint, callback) {
let _ = shared.make_unavailable();
return Err(error);
}
Ok(Self {
projection,
_ordering: ordering,
peering,
shared,
subsystem,
})
}
pub fn create(&self, owner: UserId) -> Result<GroupRevision, String> {
self.mutate(Action::Create { owner })
}
pub fn set_user_role(
&self,
actor: UserId,
group: GroupId,
user: UserId,
role: Option<GroupRole>,
) -> Result<GroupRevision, String> {
self.mutate(Action::SetUserRole {
group,
actor,
user,
role,
})
}
pub fn set_model_membership(
&self,
actor: UserId,
group: GroupId,
model: ModelId,
present: bool,
) -> Result<GroupRevision, String> {
self.mutate(Action::SetModelMembership {
group,
actor,
model,
present,
})
}
pub fn get(&self, group: GroupId) -> Result<Option<Group>, String> {
self.query(|projection| projection.get(group))
}
pub fn groups_for_user(&self, user: UserId) -> Result<Vec<GroupId>, String> {
self.query(|projection| projection.groups_for_user(user))
}
pub fn groups_for_model(&self, model: ModelId) -> Result<Vec<GroupId>, String> {
self.query(|projection| projection.groups_for_model(model))
}
pub fn memberships(&self, user: UserId, model: ModelId) -> Result<GroupMemberships, String> {
self.query(|projection| projection.memberships(user, model))
}
fn mutate(&self, action: Action) -> Result<GroupRevision, String> {
let operation_id = self.shared.reserve(action.clone())?;
let payload = action.payload(operation_id);
let submission = self.peering.submit_txn(self.subsystem, &payload);
self.shared.finish(operation_id, submission)
}
fn query<T>(&self, query: impl FnOnce(&Projection) -> Result<T, String>) -> Result<T, String> {
self.shared.ensure_available()?;
let result = match query(&self.projection) {
Ok(result) => result,
Err(error) => {
let _ = self.shared.make_unavailable();
return Err(error);
}
};
self.shared.ensure_available()?;
Ok(result)
}
}
fn parse_payload(payload: &[u8]) -> Result<(OperationId, Action), String> {
if payload.len() < 2 {
return Err("malformed groups payload header".to_string());
}
if payload[0] != 1 {
return Err("unknown groups payload version".to_string());
}
match payload[1] {
1 => parse_create(payload),
2 => parse_set_user_role(payload),
3 => parse_set_model_membership(payload),
_ => Err("unknown groups payload kind".to_string()),
}
}
fn parse_create(payload: &[u8]) -> Result<(OperationId, Action), String> {
require_length(payload, CREATE_LENGTH)?;
Ok((
operation_id(payload),
Action::Create {
owner: UserId::from_tx_id(txid(payload, 18)),
},
))
}
fn parse_set_user_role(payload: &[u8]) -> Result<(OperationId, Action), String> {
require_length(payload, SET_USER_ROLE_LENGTH)?;
let role = match payload[54] {
0 => None,
1 => Some(GroupRole::User),
2 => Some(GroupRole::Admin),
3 => Some(GroupRole::Owner),
_ => return Err("invalid group role discriminant".to_string()),
};
Ok((
operation_id(payload),
Action::SetUserRole {
group: GroupId::new(txid(payload, 18)),
actor: UserId::from_tx_id(txid(payload, 30)),
user: UserId::from_tx_id(txid(payload, 42)),
role,
},
))
}
fn parse_set_model_membership(payload: &[u8]) -> Result<(OperationId, Action), String> {
require_length(payload, SET_MODEL_MEMBERSHIP_LENGTH)?;
let present = match payload[74] {
0 => false,
1 => true,
_ => return Err("invalid model presence discriminant".to_string()),
};
let mut model = [0_u8; 32];
model.copy_from_slice(&payload[42..74]);
Ok((
operation_id(payload),
Action::SetModelMembership {
group: GroupId::new(txid(payload, 18)),
actor: UserId::from_tx_id(txid(payload, 30)),
model: ModelId::from_bytes(model),
present,
},
))
}
fn require_length(payload: &[u8], expected: usize) -> Result<(), String> {
if payload.len() == expected {
Ok(())
} else {
Err(format!(
"wrong groups payload length: expected {expected}, received {}",
payload.len()
))
}
}
fn operation_id(payload: &[u8]) -> OperationId {
let mut operation_id = [0_u8; 16];
operation_id.copy_from_slice(&payload[2..18]);
operation_id
}
fn txid(payload: &[u8], start: usize) -> TxId {
let mut bytes = [0_u8; 12];
bytes.copy_from_slice(&payload[start..start + 12]);
TxId::from_bytes(bytes)
}
fn role_byte(role: Option<GroupRole>) -> u8 {
match role {
None => 0,
Some(GroupRole::User) => 1,
Some(GroupRole::Admin) => 2,
Some(GroupRole::Owner) => 3,
}
}
#[cfg(test)]
mod tests;