liminal_server/auth_membership/
index.rs

1use super::{
2    BTreeMap, BTreeSet, Decided, MappedChannel, MembershipAction, MembershipRefusal,
3    MembershipSubject, Mode, PageRow, Resource, Subject,
4};
5
6#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord)]
7pub(super) struct DecisionKey {
8    pub(super) subject: MembershipSubject,
9    pub(super) channel: String,
10    pub(super) action: MembershipAction,
11}
12
13/// A retained decision, or the ticket of the ask settling it.
14#[derive(Clone, Debug, PartialEq, Eq)]
15pub(super) enum Indexed {
16    Applying { ticket: u64 },
17    Allowed { grant: String, revision: u64 },
18    Refused { refusal: String },
19    Held { mode: Mode },
20}
21
22impl Indexed {
23    pub(super) fn result(&self) -> Result<Decided, MembershipRefusal> {
24        match self {
25            Self::Applying { .. } => Err(MembershipRefusal::ChangeApplying),
26            Self::Allowed { grant, revision } => Ok(Decided::Allowed {
27                grant: grant.clone(),
28                revision: *revision,
29            }),
30            Self::Refused { refusal } => Err(MembershipRefusal::Refused {
31                refusal: refusal.clone(),
32            }),
33            Self::Held { mode } => Err(MembershipRefusal::Held { mode: *mode }),
34        }
35    }
36}
37
38pub(super) type ResourceKey = (String, String);
39
40pub(super) fn resource_key(resource: &Resource) -> ResourceKey {
41    (resource.kind.clone(), resource.id.clone())
42}
43
44#[derive(Debug)]
45pub(super) struct Entry {
46    pub(super) verdict: Indexed,
47    pub(super) connections: BTreeSet<u64>,
48    pub(super) reached_by: [ResourceKey; 2],
49}
50
51#[derive(Debug, Default)]
52pub(super) struct DecisionIndex {
53    pub(super) generation: Option<u64>,
54    pub(super) applied: u64,
55    pub(super) next_ticket: u64,
56    pub(super) entries: BTreeMap<DecisionKey, Entry>,
57    pub(super) by_resource: BTreeMap<ResourceKey, BTreeSet<DecisionKey>>,
58    pub(super) by_connection: BTreeMap<u64, BTreeSet<DecisionKey>>,
59}
60
61impl DecisionIndex {
62    pub(super) const fn adopt(&mut self, generation: u64) -> Result<(), MembershipRefusal> {
63        match self.generation {
64            None => {
65                self.generation = Some(generation);
66                Ok(())
67            }
68            Some(current) if current == generation => Ok(()),
69            Some(_) => Err(MembershipRefusal::ResumePending),
70        }
71    }
72
73    pub(super) fn ticket(&mut self) -> Result<u64, MembershipRefusal> {
74        let ticket = self.next_ticket;
75        self.next_ticket = ticket
76            .checked_add(1)
77            .ok_or(MembershipRefusal::CapacityExceeded { limit: usize::MAX })?;
78        Ok(ticket)
79    }
80
81    /// The retained verdict for `key`, binding `pid` to it.
82    pub(super) fn known(&mut self, key: &DecisionKey, pid: u64) -> Option<Indexed> {
83        let verdict = self.entries.get_mut(key).map(|entry| {
84            entry.connections.insert(pid);
85            entry.verdict.clone()
86        })?;
87        self.by_connection
88            .entry(pid)
89            .or_default()
90            .insert(key.clone());
91        Some(verdict)
92    }
93
94    /// Reserves `key` for one ask, refusing at `limit` before allocating.
95    pub(super) fn reserve(
96        &mut self,
97        key: &DecisionKey,
98        mapped: &MappedChannel,
99        pid: u64,
100        limit: usize,
101    ) -> Result<u64, MembershipRefusal> {
102        if self.entries.len() >= limit {
103            return Err(MembershipRefusal::CapacityExceeded { limit });
104        }
105        let ticket = self.ticket()?;
106        let reached_by = [
107            resource_key(&mapped.resource),
108            resource_key(&mapped.workspace),
109        ];
110        for reach in &reached_by {
111            self.by_resource
112                .entry(reach.clone())
113                .or_default()
114                .insert(key.clone());
115        }
116        self.by_connection
117            .entry(pid)
118            .or_default()
119            .insert(key.clone());
120        self.entries.insert(
121            key.clone(),
122            Entry {
123                verdict: Indexed::Applying { ticket },
124                connections: BTreeSet::from([pid]),
125                reached_by,
126            },
127        );
128        Ok(ticket)
129    }
130
131    /// Re-reserves an existing binding for a change or resume ask.
132    pub(super) fn reapply(&mut self, key: &DecisionKey) -> Result<Option<u64>, MembershipRefusal> {
133        if !self.entries.contains_key(key) {
134            return Ok(None);
135        }
136        let ticket = self.ticket()?;
137        if let Some(entry) = self.entries.get_mut(key) {
138            entry.verdict = Indexed::Applying { ticket };
139        }
140        Ok(Some(ticket))
141    }
142
143    /// Settles the ask holding `ticket`; `false` when a later change or
144    /// resume overtook it.
145    pub(super) fn settle(&mut self, key: &DecisionKey, ticket: u64, verdict: Indexed) -> bool {
146        match self.entries.get_mut(key) {
147            Some(entry) if entry.verdict == (Indexed::Applying { ticket }) => {
148                entry.verdict = verdict;
149                true
150            }
151            Some(_) | None => false,
152        }
153    }
154
155    /// Forgets the binding still held by `ticket`'s ask.
156    pub(super) fn forget(&mut self, key: &DecisionKey, ticket: u64) {
157        let held = self
158            .entries
159            .get(key)
160            .is_some_and(|entry| entry.verdict == (Indexed::Applying { ticket }));
161        if held {
162            self.remove(key);
163        }
164    }
165
166    pub(super) fn remove(&mut self, key: &DecisionKey) {
167        let Some(entry) = self.entries.remove(key) else {
168            return;
169        };
170        for reach in &entry.reached_by {
171            let emptied = self.by_resource.get_mut(reach).is_some_and(|keys| {
172                keys.remove(key);
173                keys.is_empty()
174            });
175            if emptied {
176                self.by_resource.remove(reach);
177            }
178        }
179        for pid in &entry.connections {
180            let emptied = self.by_connection.get_mut(pid).is_some_and(|keys| {
181                keys.remove(key);
182                keys.is_empty()
183            });
184            if emptied {
185                self.by_connection.remove(pid);
186            }
187        }
188    }
189
190    /// Releases every binding `pid` held; a binding no connection holds is
191    /// removed. Visits only `pid`'s own bindings.
192    pub(super) fn release(&mut self, pid: u64) -> u64 {
193        let Some(keys) = self.by_connection.remove(&pid) else {
194            return 0;
195        };
196        let mut released = 0_u64;
197        for key in keys {
198            let unheld = self.entries.get_mut(&key).is_some_and(|entry| {
199                entry.connections.remove(&pid);
200                entry.connections.is_empty()
201            });
202            if unheld {
203                self.remove(&key);
204                released = released.saturating_add(1);
205            }
206        }
207        released
208    }
209}
210
211/// The question a page continuation is bound to.
212#[derive(Clone, Debug, PartialEq, Eq)]
213pub(super) enum PageQuestion {
214    Resources {
215        workspace: String,
216        subject: MembershipSubject,
217        action: MembershipAction,
218    },
219    Recipients {
220        channel: String,
221        action: MembershipAction,
222    },
223}
224
225/// An opaque page continuation, bound to its exact question, grant log and
226/// generation; it grants nothing (ACCESS-006 R4).
227#[derive(Clone, PartialEq, Eq)]
228pub struct PageCursor {
229    pub(super) token: String,
230    pub(super) question: PageQuestion,
231    pub(super) generation: u64,
232}
233
234impl std::fmt::Debug for PageCursor {
235    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
236        formatter
237            .debug_struct("PageCursor")
238            .field("token", &crate::auth_pass::ByteLength(self.token.len()))
239            .field("generation", &self.generation)
240            .finish_non_exhaustive()
241    }
242}
243
244/// One page of the protected channels a holder may act on in a workspace.
245#[derive(Clone, Debug, PartialEq, Eq)]
246pub struct ChannelPage {
247    /// The liminal channels, in the provider's stable order.
248    pub channels: Vec<String>,
249    /// Rows the provider returned.
250    pub returned: u64,
251    /// Candidates the provider decided and left out.
252    pub skipped: u64,
253    /// Rows held for approval: a membership exists, the act still waits.
254    pub held: u64,
255    /// Rows naming a resource the product mapping does not list.
256    pub unmapped: u64,
257    /// The rows' encoded bytes.
258    pub bytes: u64,
259    /// The revision every row was decided at.
260    pub revision: u64,
261    /// Whether no row remains.
262    pub complete: bool,
263    /// The continuation, absent when complete.
264    pub next: Option<PageCursor>,
265}
266
267/// One page of the subjects who may act on a protected channel.
268#[derive(Clone, Debug, PartialEq, Eq)]
269pub struct RecipientPage {
270    /// The subjects, in the provider's stable order.
271    pub recipients: Vec<Subject>,
272    /// Rows the provider returned.
273    pub returned: u64,
274    /// Candidates the provider decided and left out.
275    pub skipped: u64,
276    /// Rows held for approval.
277    pub held: u64,
278    /// The rows' encoded bytes.
279    pub bytes: u64,
280    /// The revision every row was decided at.
281    pub revision: u64,
282    /// Whether no row remains.
283    pub complete: bool,
284    /// The continuation, absent when complete.
285    pub next: Option<PageCursor>,
286}
287
288/// A validated page's parts.
289pub(super) struct PageParts<'p> {
290    pub(super) rows: &'p [PageRow],
291    pub(super) skipped: u64,
292    pub(super) complete: bool,
293    pub(super) next: Option<String>,
294    pub(super) revision: u64,
295    pub(super) bytes: u64,
296}
liminal_server/auth_membership/index.rs

liminal_server/auth_membership/
index.rs
