macula_rust/station_link/
admission.rs

1//! Admission of the CALLs a provider node receives, as
2//! macula_request_admission judges them: a request runs once, whichever of
3//! the node's links it arrives on. Its deadline must lie between the
4//! provider's clock minus 5 minutes and plus 10 minutes, and (caller,
5//! request_id) must be new; the entry is kept until the deadline plus 5
6//! minutes. A copy with the same request hash gets the stored reply, or
7//! request_copy while the first still runs; one with another hash is refused.
8//! The entries are bounded, and a full bound refuses rather than evicts, in
9//! this order: each caller holds at most `caller_quota` entries, each share
10//! (one link's place: the station it dialed) at most `share`, and the
11//! admission at most `cap`; stored replies take at most `reply_bytes` per
12//! caller and `reply_bytes_total` in all. A reply past either is not kept, and
13//! a copy of its request is refused reply_not_kept.
14
15use std::collections::HashMap;
16use std::hash::Hash;
17use std::sync::{Arc, Mutex, MutexGuard};
18
19use crate::frame::VerifiedRequest;
20
21use super::LinkError;
22
23const DEADLINE_PAST_TOLERANCE_MS: i64 = 5 * 60_000;
24const DEADLINE_AHEAD_MAX_MS: i64 = 10 * 60_000;
25const KEPT_PAST_DEADLINE_MS: i64 = 5 * 60_000;
26
27/// An admission's bounds. The last four bound the streaming sessions a node
28/// serves, as macula_stream_sessions does: sessions at once per caller and in
29/// all, and the bytes their inboxes hold per caller and in all.
30#[derive(Debug, Clone, Copy, PartialEq, Eq)]
31pub struct AdmissionLimits {
32    pub caller_quota: usize,
33    pub share: usize,
34    pub cap: usize,
35    pub reply_bytes: usize,
36    pub reply_bytes_total: usize,
37    pub sessions_per_caller: usize,
38    pub sessions: usize,
39    pub inbox_bytes_per_caller: usize,
40    pub inbox_bytes: usize,
41}
42
43impl Default for AdmissionLimits {
44    /// macula's defaults, with `cap` one share's worth: the bound of an
45    /// admission a single link holds. A pool sets `cap` to `share` times the
46    /// most links it holds.
47    fn default() -> Self {
48        AdmissionLimits {
49            caller_quota: 256,
50            share: 1024,
51            cap: 1024,
52            reply_bytes: 256 * 1024,
53            reply_bytes_total: 16 * 1024 * 1024,
54            sessions_per_caller: 16,
55            sessions: 1000,
56            inbox_bytes_per_caller: 16 * 1024 * 1024,
57            inbox_bytes: 256 * 1024 * 1024,
58        }
59    }
60}
61
62impl AdmissionLimits {
63    /// Whether macula would start with these limits: every bound positive,
64    /// and each per-caller bound within its total.
65    pub fn validate(&self) -> Result<(), LinkError> {
66        let l = self;
67        let valid = l.caller_quota > 0
68            && l.share > 0
69            && l.cap > 0
70            && l.reply_bytes > 0
71            && l.reply_bytes_total > 0
72            && l.caller_quota <= l.share
73            && l.reply_bytes <= l.reply_bytes_total
74            && l.sessions_per_caller > 0
75            && l.sessions >= l.sessions_per_caller
76            && l.inbox_bytes_per_caller > 0
77            && l.inbox_bytes >= l.inbox_bytes_per_caller;
78        if valid {
79            Ok(())
80        } else {
81            Err(LinkError::InvalidConfig(
82                "admission limits must be positive, each per-caller bound within its total".into(),
83            ))
84        }
85    }
86}
87
88/// The admission's judgement of a request: refused with a code, a copy with
89/// its stored reply (`None` while the first still runs), or new.
90#[derive(Debug, Clone, PartialEq, Eq)]
91pub(super) enum Verdict {
92    Refused(&'static str),
93    Copy(Option<Vec<u8>>),
94    New,
95}
96
97struct Entry {
98    hash: [u8; 48],
99    expires_at: i64,
100    share: String,
101    answered: bool,
102    /// `None` when answered but not kept.
103    reply: Option<Vec<u8>>,
104}
105
106#[derive(Default)]
107struct Held {
108    entries: HashMap<([u8; 32], [u8; 16]), Entry>,
109    callers: HashMap<[u8; 32], usize>,
110    shares: HashMap<String, usize>,
111    reply_bytes: HashMap<[u8; 32], usize>,
112    reply_total: usize,
113    sessions: HashMap<[u8; 32], usize>,
114    sessions_total: usize,
115    inbox: HashMap<[u8; 32], usize>,
116    inbox_total: usize,
117}
118
119/// One provider node's request admission, shared by all its links.
120pub struct Admission {
121    limits: AdmissionLimits,
122    held: Mutex<Held>,
123}
124
125impl Admission {
126    /// An empty admission with `limits`, which must validate: a link given
127    /// limits that do not is refused at dial.
128    pub fn new(limits: AdmissionLimits) -> Admission {
129        Admission {
130            limits,
131            held: Mutex::new(Held::default()),
132        }
133    }
134
135    /// The admission's bounds.
136    pub fn limits(&self) -> AdmissionLimits {
137        self.limits
138    }
139
140    fn lock(&self) -> MutexGuard<'_, Held> {
141        self.held.lock().unwrap_or_else(|p| p.into_inner())
142    }
143
144    /// Judges `request` arriving on `share` at `now_ms`, sweeping the entries
145    /// that expired first.
146    pub(super) fn admit(&self, request: &VerifiedRequest, share: &str, now_ms: i64) -> Verdict {
147        let deadline = request.deadline as i64;
148        if deadline < now_ms - DEADLINE_PAST_TOLERANCE_MS {
149            return Verdict::Refused("expired");
150        }
151        if deadline > now_ms + DEADLINE_AHEAD_MAX_MS {
152            return Verdict::Refused("not_yet_valid");
153        }
154        let mut held = self.lock();
155        held.sweep(now_ms);
156        let key = (request.caller, request.request_id);
157        if let Some(entry) = held.entries.get(&key) {
158            if entry.hash != request.request_hash {
159                return Verdict::Refused("request_id_reused");
160            }
161            if entry.answered && entry.reply.is_none() {
162                return Verdict::Refused("reply_not_kept");
163            }
164            return Verdict::Copy(entry.reply.clone());
165        }
166        if held.callers.get(&request.caller).copied().unwrap_or(0) >= self.limits.caller_quota {
167            return Verdict::Refused("caller_quota");
168        }
169        if held.shares.get(share).copied().unwrap_or(0) >= self.limits.share {
170            return Verdict::Refused("share_full");
171        }
172        if held.entries.len() >= self.limits.cap {
173            return Verdict::Refused("admission_full");
174        }
175        held.entries.insert(
176            key,
177            Entry {
178                hash: request.request_hash,
179                expires_at: deadline + KEPT_PAST_DEADLINE_MS,
180                share: share.to_string(),
181                answered: false,
182                reply: None,
183            },
184        );
185        *held.callers.entry(request.caller).or_default() += 1;
186        *held.shares.entry(share.to_string()).or_default() += 1;
187        Verdict::New
188    }
189
190    /// Keeps the encoded reply of an admitted request for its copies, when
191    /// the byte bounds leave room for it.
192    pub(super) fn store(&self, request: &VerifiedRequest, reply: Vec<u8>) {
193        let mut held = self.lock();
194        let held = &mut *held;
195        let Some(entry) = held.entries.get_mut(&(request.caller, request.request_id)) else {
196            return;
197        };
198        if entry.answered || entry.hash != request.request_hash {
199            return;
200        }
201        entry.answered = true;
202        let caller_bytes = held.reply_bytes.get(&request.caller).copied().unwrap_or(0);
203        if caller_bytes + reply.len() > self.limits.reply_bytes
204            || held.reply_total + reply.len() > self.limits.reply_bytes_total
205        {
206            return;
207        }
208        *held.reply_bytes.entry(request.caller).or_default() += reply.len();
209        held.reply_total += reply.len();
210        entry.reply = Some(reply);
211    }
212
213    /// Takes a streaming session's place for `caller`, or `None` when the
214    /// per-caller or node bound is full. Dropping the place gives it back.
215    pub(super) fn open_session(self: &Arc<Self>, caller: [u8; 32]) -> Option<SessionPlace> {
216        let mut held = self.lock();
217        if held.sessions.get(&caller).copied().unwrap_or(0) >= self.limits.sessions_per_caller
218            || held.sessions_total >= self.limits.sessions
219        {
220            return None;
221        }
222        *held.sessions.entry(caller).or_default() += 1;
223        held.sessions_total += 1;
224        Some(SessionPlace {
225            admission: self.clone(),
226            caller,
227        })
228    }
229
230    /// Counts `n` more bytes that `caller`'s served streams hold unread,
231    /// refusing a charge past either bound.
232    pub(super) fn charge_inbox(&self, caller: [u8; 32], n: usize) -> bool {
233        let mut held = self.lock();
234        if held.inbox.get(&caller).copied().unwrap_or(0) + n > self.limits.inbox_bytes_per_caller
235            || held.inbox_total + n > self.limits.inbox_bytes
236        {
237            return false;
238        }
239        *held.inbox.entry(caller).or_default() += n;
240        held.inbox_total += n;
241        true
242    }
243
244    /// Gives back `n` bytes `caller`'s served streams no longer hold.
245    pub(super) fn release_inbox(&self, caller: [u8; 32], n: usize) {
246        let mut held = self.lock();
247        decrement(&mut held.inbox, caller, n);
248        held.inbox_total = held.inbox_total.saturating_sub(n);
249    }
250}
251
252/// A streaming session's place in the admission, given back when dropped.
253pub(super) struct SessionPlace {
254    admission: Arc<Admission>,
255    caller: [u8; 32],
256}
257
258impl Drop for SessionPlace {
259    fn drop(&mut self) {
260        let mut held = self.admission.lock();
261        decrement(&mut held.sessions, self.caller, 1);
262        held.sessions_total = held.sessions_total.saturating_sub(1);
263    }
264}
265
266impl Held {
267    /// Removes the entries whose deadline plus 5 minutes passed before
268    /// `now_ms`.
269    fn sweep(&mut self, now_ms: i64) {
270        let expired: Vec<_> = self
271            .entries
272            .iter()
273            .filter(|(_, e)| e.expires_at < now_ms)
274            .map(|(k, _)| *k)
275            .collect();
276        for key in expired {
277            let Some(entry) = self.entries.remove(&key) else {
278                continue;
279            };
280            decrement(&mut self.callers, key.0, 1);
281            decrement(&mut self.shares, entry.share, 1);
282            if let Some(reply) = entry.reply {
283                decrement(&mut self.reply_bytes, key.0, reply.len());
284                self.reply_total = self.reply_total.saturating_sub(reply.len());
285            }
286        }
287    }
288}
289
290/// Takes `n` from `m[k]`, dropping the key at zero.
291fn decrement<K: Eq + Hash>(m: &mut HashMap<K, usize>, k: K, n: usize) {
292    if let Some(v) = m.get_mut(&k) {
293        *v = v.saturating_sub(n);
294        if *v == 0 {
295            m.remove(&k);
296        }
297    }
298}
macula_rust/station_link/admission.rs

macula_rust/station_link/
admission.rs
