1use std::{
2 collections::HashMap,
3 path::Path,
4 sync::{Arc, Mutex, MutexGuard},
5};
6
7use kcode_k1_access_format::{AccessAction, decode, encode};
8use kcode_k1_access_projection::{ApplyOutcome, Projection};
9use kcode_k1_peering::K1Peering;
10use kcode_k1_txn_ordering::{K1TxnOrdering, Subsystem as K1Subsystem};
11
12pub use kcode_k1_access_types::{
13 AccessCheck, AccessId, AccessRevision, Authorizations, GroupId, ModelId, OwnerSubject,
14 RequestPrincipal, Target, TxId, UserId, ViewerSubject,
15};
16pub use kcode_k1_transaction::SubsystemId;
17
18use kcode_k1_groups::K1Groups;
19
20const SUBSYSTEM_NAME: &str = "k1-access-subsystem";
21type OperationId = [u8; 16];
22
23struct Pending {
24 action: AccessAction,
25 result: Option<(TxId, ApplyOutcome)>,
26}
27
28struct FacadeState {
29 available: bool,
30 pending: HashMap<OperationId, Pending>,
31}
32
33struct SharedState {
34 inner: Mutex<FacadeState>,
35}
36
37impl SharedState {
38 fn new() -> Self {
39 Self {
40 inner: Mutex::new(FacadeState {
41 available: true,
42 pending: HashMap::new(),
43 }),
44 }
45 }
46
47 fn lock(&self) -> Result<MutexGuard<'_, FacadeState>, String> {
48 self.inner
49 .lock()
50 .map_err(|_| "k1 access state lock failed".to_owned())
51 }
52
53 fn ensure_available(&self) -> Result<(), String> {
54 if self.lock()?.available {
55 Ok(())
56 } else {
57 Err("k1 access instance unavailable".to_owned())
58 }
59 }
60
61 fn reserve(&self, action: AccessAction) -> Result<OperationId, String> {
62 self.ensure_available()?;
63 loop {
64 let mut operation_id = [0_u8; 16];
65 getrandom::fill(&mut operation_id)
66 .map_err(|error| format!("operation ID randomness failed: {error}"))?;
67 let mut state = self.lock()?;
68 if !state.available {
69 return Err("k1 access instance unavailable".to_owned());
70 }
71 if state.pending.contains_key(&operation_id) {
72 continue;
73 }
74 state
75 .pending
76 .try_reserve(1)
77 .map_err(|_| "pending operation allocation failed".to_owned())?;
78 state.pending.insert(
79 operation_id,
80 Pending {
81 action,
82 result: None,
83 },
84 );
85 return Ok(operation_id);
86 }
87 }
88
89 fn cancel(&self, operation_id: OperationId) -> Result<(), String> {
90 let mut state = self.lock()?;
91 if state.pending.remove(&operation_id).is_none() {
92 invalidate(&mut state);
93 return Err("pending operation disappeared".to_owned());
94 }
95 if state.available {
96 Ok(())
97 } else {
98 Err("k1 access instance unavailable".to_owned())
99 }
100 }
101
102 fn record(
103 &self,
104 operation_id: OperationId,
105 action: &AccessAction,
106 txid: TxId,
107 outcome: ApplyOutcome,
108 ) -> Result<(), String> {
109 let mut state = self.lock()?;
110 if !state.available {
111 return Err("k1 access instance unavailable".to_owned());
112 }
113 let contradiction = match state.pending.get_mut(&operation_id) {
114 None => return Ok(()),
115 Some(pending) if &pending.action != action => {
116 Some("operation ID correlated with a different access action")
117 }
118 Some(pending) if pending.result.is_some() => {
119 Some("duplicate callback evidence for one operation ID")
120 }
121 Some(pending) => {
122 pending.result = Some((txid, outcome));
123 None
124 }
125 };
126 if let Some(error) = contradiction {
127 invalidate(&mut state);
128 Err(error.to_owned())
129 } else {
130 Ok(())
131 }
132 }
133
134 fn finish(
135 &self,
136 operation_id: OperationId,
137 submission: Result<TxId, String>,
138 ) -> Result<AccessRevision, String> {
139 let result = {
140 let mut state = self.lock()?;
141 let Some(pending) = state.pending.remove(&operation_id) else {
142 invalidate(&mut state);
143 return Err("pending operation disappeared".to_owned());
144 };
145 if !state.available {
146 return Err("k1 access instance unavailable".to_owned());
147 }
148 pending.result
149 };
150 match submission {
151 Ok(submitted) => match result {
152 Some((callback, outcome)) if callback == submitted => outcome_result(outcome),
153 Some(_) => {
154 self.fault("submission transaction ID did not match callback transaction ID")
155 }
156 None => self.fault("successful submission had no synchronous callback"),
157 },
158 Err(error) => match result {
159 Some((_, outcome)) => outcome_result(outcome),
160 None => Err(error),
161 },
162 }
163 }
164
165 fn make_unavailable(&self) -> Result<(), String> {
166 let mut state = self.lock()?;
167 invalidate(&mut state);
168 Ok(())
169 }
170
171 fn fault<T>(&self, error: &str) -> Result<T, String> {
172 let _ = self.make_unavailable();
173 Err(error.to_owned())
174 }
175}
176
177fn invalidate(state: &mut FacadeState) {
178 state.available = false;
179 for pending in state.pending.values_mut() {
180 pending.result = None;
181 }
182}
183
184fn outcome_result(outcome: ApplyOutcome) -> Result<AccessRevision, String> {
185 match outcome {
186 ApplyOutcome::Applied(revision) | ApplyOutcome::Unchanged(revision) => Ok(revision),
187 ApplyOutcome::Rejected(reason) => Err(reason),
188 }
189}
190
191struct AccessSubsystem {
192 projection: Arc<Projection>,
193 shared: Arc<SharedState>,
194}
195
196impl K1Subsystem for AccessSubsystem {
197 fn submit_txn(&self, id: TxId, payload: &[u8]) -> Result<(), String> {
198 let (operation_id, action) = match decode(payload) {
199 Ok(decoded) => decoded,
200 Err(error) => {
201 let _ = self.shared.make_unavailable();
202 return Err(error);
203 }
204 };
205 self.shared.ensure_available()?;
206 let outcome = match self.projection.apply(id, action.clone()) {
207 Ok(outcome) => outcome,
208 Err(error) => {
209 let _ = self.shared.make_unavailable();
210 return Err(error);
211 }
212 };
213 self.shared.record(operation_id, &action, id, outcome)
214 }
215
216 fn reorg(&self) -> Result<(), String> {
217 let state_result = self.shared.make_unavailable();
218 match self.projection.clear() {
219 Ok(()) => state_result,
220 Err(error) => Err(error),
221 }
222 }
223}
224
225pub struct K1Access {
226 projection: Arc<Projection>,
227 _ordering: Arc<K1TxnOrdering>,
228 peering: Arc<K1Peering>,
229 groups: Arc<K1Groups>,
230 shared: Arc<SharedState>,
231 subsystem: SubsystemId,
232}
233
234impl K1Access {
235 pub fn open(
236 root: &Path,
237 ordering: Arc<K1TxnOrdering>,
238 peering: Arc<K1Peering>,
239 groups: Arc<K1Groups>,
240 ) -> Result<Self, String> {
241 let (projection, cursor) = Projection::open(root, &ordering)?;
242 let projection = Arc::new(projection);
243 let shared = Arc::new(SharedState::new());
244 let subsystem = SubsystemId::from_str(SUBSYSTEM_NAME)?;
245 let callback = Arc::new(AccessSubsystem {
246 projection: projection.clone(),
247 shared: shared.clone(),
248 });
249 if let Err(error) = ordering.register_subsystem(subsystem, cursor, callback) {
250 let _ = shared.make_unavailable();
251 return Err(error);
252 }
253 Ok(Self {
254 projection,
255 _ordering: ordering,
256 peering,
257 groups,
258 shared,
259 subsystem,
260 })
261 }
262
263 pub fn create(
264 &self,
265 target: Target,
266 authorizations: Authorizations,
267 ) -> Result<AccessRevision, String> {
268 self.mutate(AccessAction::Create {
269 target,
270 authorizations,
271 })
272 }
273
274 pub fn set_authorizations(
275 &self,
276 principal: RequestPrincipal,
277 access_id: AccessId,
278 authorizations: Authorizations,
279 ) -> Result<AccessRevision, String> {
280 self.shared.ensure_available()?;
281 let memberships = self.groups.memberships(principal.user(), principal.model());
282 self.shared.ensure_available()?;
283 let memberships = memberships?;
284 let witness = self.projection_call(self.projection.owner_witness(
285 access_id,
286 principal.user(),
287 memberships.user_groups(),
288 ))?;
289 let witness = witness.ok_or_else(|| "principal is not an access owner".to_owned())?;
290 self.mutate(AccessAction::Replace {
291 access_id,
292 actor: principal.user(),
293 groups_revision: memberships.revision(),
294 witness,
295 authorizations,
296 })
297 }
298
299 pub fn check(
300 &self,
301 principal: RequestPrincipal,
302 access_id: AccessId,
303 expected_subsystem: SubsystemId,
304 ) -> Result<AccessCheck, String> {
305 self.shared.ensure_available()?;
306 let memberships = self.groups.memberships(principal.user(), principal.model());
307 self.shared.ensure_available()?;
308 let memberships = memberships?;
309 self.projection_call(self.projection.check(
310 principal,
311 access_id,
312 expected_subsystem,
313 memberships.user_groups(),
314 memberships.model_groups(),
315 memberships.revision(),
316 ))
317 }
318
319 fn mutate(&self, action: AccessAction) -> Result<AccessRevision, String> {
320 let operation_id = self.shared.reserve(action.clone())?;
321 let payload = match encode(operation_id, &action) {
322 Ok(payload) => payload,
323 Err(error) => {
324 self.shared.cancel(operation_id)?;
325 return Err(error);
326 }
327 };
328 if let Err(error) = self.shared.ensure_available() {
329 let _ = self.shared.cancel(operation_id);
330 return Err(error);
331 }
332 let submission = self.peering.submit_txn(self.subsystem, &payload);
333 self.shared.finish(operation_id, submission)
334 }
335
336 fn projection_call<T>(&self, result: Result<T, String>) -> Result<T, String> {
337 match result {
338 Ok(value) => {
339 self.shared.ensure_available()?;
340 Ok(value)
341 }
342 Err(error) => {
343 let _ = self.shared.make_unavailable();
344 Err(error)
345 }
346 }
347 }
348}
349
350#[cfg(test)]
351mod tests {
352 use super::*;
353
354 fn tx(byte: u8) -> TxId {
355 TxId::from_bytes([byte; 12])
356 }
357
358 fn action(byte: u8) -> AccessAction {
359 let owner = UserId::from_tx_id(tx(byte));
360 AccessAction::Create {
361 target: Target::new(SubsystemId::from_str("test").unwrap(), vec![byte]),
362 authorizations: Authorizations::new(vec![OwnerSubject::User(owner)], Vec::new())
363 .unwrap(),
364 }
365 }
366
367 fn pending(shared: &SharedState, operation_id: OperationId, action: AccessAction) {
368 shared.lock().unwrap().pending.insert(
369 operation_id,
370 Pending {
371 action,
372 result: None,
373 },
374 );
375 }
376
377 #[test]
378 fn exact_callback_controls_submission_resolution() {
379 let shared = SharedState::new();
380 let operation_id = [1; 16];
381 let action = action(1);
382 let revision = AccessRevision::new(AccessId::new(tx(2)), tx(3));
383 pending(&shared, operation_id, action.clone());
384 shared
385 .record(
386 operation_id,
387 &action,
388 tx(4),
389 ApplyOutcome::Applied(revision),
390 )
391 .unwrap();
392 assert_eq!(shared.finish(operation_id, Ok(tx(4))), Ok(revision));
393
394 let operation_id = [2; 16];
395 pending(&shared, operation_id, action.clone());
396 shared
397 .record(
398 operation_id,
399 &action,
400 tx(5),
401 ApplyOutcome::Unchanged(revision),
402 )
403 .unwrap();
404 assert_eq!(
405 shared.finish(operation_id, Err("committed".to_owned())),
406 Ok(revision)
407 );
408
409 let operation_id = [3; 16];
410 pending(&shared, operation_id, action.clone());
411 shared
412 .record(
413 operation_id,
414 &action,
415 tx(6),
416 ApplyOutcome::Rejected("exact rejection".to_owned()),
417 )
418 .unwrap();
419 assert_eq!(
420 shared.finish(operation_id, Err("committed".to_owned())),
421 Err("exact rejection".to_owned())
422 );
423 }
424
425 #[test]
426 fn missing_mismatched_and_duplicate_evidence_fault() {
427 let missing = SharedState::new();
428 pending(&missing, [1; 16], action(1));
429 assert!(missing.finish([1; 16], Ok(tx(1))).is_err());
430 assert!(missing.ensure_available().is_err());
431
432 let mismatched = SharedState::new();
433 pending(&mismatched, [2; 16], action(2));
434 assert!(
435 mismatched
436 .record(
437 [2; 16],
438 &action(3),
439 tx(2),
440 ApplyOutcome::Rejected("remote".to_owned())
441 )
442 .is_err()
443 );
444 assert!(mismatched.ensure_available().is_err());
445
446 let duplicate = SharedState::new();
447 let exact = action(4);
448 let revision = AccessRevision::new(AccessId::new(tx(4)), tx(5));
449 pending(&duplicate, [3; 16], exact.clone());
450 duplicate
451 .record([3; 16], &exact, tx(5), ApplyOutcome::Applied(revision))
452 .unwrap();
453 assert!(
454 duplicate
455 .record([3; 16], &exact, tx(5), ApplyOutcome::Applied(revision))
456 .is_err()
457 );
458 assert!(duplicate.ensure_available().is_err());
459 }
460}