harmos_sidecar/kit/
host.rs

1/* ***
2 *
3 * Harmos
4 *
5 * SPDX-License-Identifier: Apache-2.0
6 *
7 * ***
8 */
9
10/* === Modules === */
11
12/* === Re-Exports === */
13
14/* === Imports === */
15
16// Standard Library Imports
17use std::collections::{BTreeMap, HashMap};
18use std::sync::{Arc, Mutex};
19// Third Party Imports
20use tokio::sync::{Semaphore, SemaphorePermit, mpsc, oneshot};
21// Local Imports
22use crate::Frame;
23use harmos_domain::Message;
24use harmos_domain::operations::instance::Id;
25use harmos_sdk::operations::{CHILD_METHOD, OperationDeclaration, WireChildAction, WireChildControl};
26
27/* === Constants === */
28
29const PENDING_CALLS: usize = 256;
30// One Batched Call Claims No More Host Preparation Than One Maximal Row.
31const PUBLICATION_BATCH_BYTES: usize = 1024 * 1024;
32
33/* === Macros === */
34
35/* === Type Aliases === */
36
37/* === Structs === */
38
39struct PendingCalls {
40    state: Mutex<PendingState>,
41    capacity: Semaphore,
42}
43
44struct PendingState {
45    closed: bool,
46    next_id: u64,
47    waiters: HashMap<u64, oneshot::Sender<Result<Vec<u8>, CallError>>>,
48}
49
50struct PendingCall<'a> {
51    pending: &'a PendingCalls,
52    id: u64,
53    _permit: SemaphorePermit<'a>,
54    cancel: Option<mpsc::Permit<'a, Outgoing>>,
55}
56
57#[derive(Clone)]
58pub struct Supervisor {
59    outgoing: mpsc::Sender<Outgoing>,
60    sequences: Arc<Mutex<BTreeMap<&'static str, u64>>>,
61    pending: Arc<PendingCalls>,
62    operation_context: Option<[u8; 32]>,
63    generation_context: Option<[u8; 32]>,
64    call: Arc<harmos_domain::operations::CallContext>,
65    clock: fn() -> Option<i64>,
66}
67
68/* === Enums === */
69
70pub(crate) enum Outgoing {
71    Frame(Frame),
72    Started(u64),
73    Finished(u64),
74    Exiting(String),
75    Abandon,
76}
77
78#[derive(Clone, Debug, PartialEq, Eq, thiserror::Error)]
79pub enum CallError {
80    // The host's refusal carries the failure unchanged; this layer names only
81    // the operation the host attributed it to.
82    #[error("{}", refused_line(*.code, .failure))]
83    Refused {
84        code: i64,
85
86        #[source]
87        failure: harmos_domain::Failure,
88    },
89
90    #[error("the host connection is unavailable")]
91    Unavailable,
92
93    #[error("the host returned a malformed reply for `{method}`: {message}")]
94    Malformed { method: &'static str, message: String },
95
96    #[error("the request for `{method}` requires {encoded} bytes; the frame limit is {limit}")]
97    Oversized { method: &'static str, encoded: usize, limit: usize },
98
99    #[error(transparent)]
100    Publication(#[from] PublicationError),
101
102    #[error("`{method}` needs process-local resources a connection cannot carry")]
103    LocalResources { method: &'static str },
104}
105
106#[derive(Clone, Copy, Debug, PartialEq, Eq, thiserror::Error)]
107pub enum PublicationError {
108    #[error("publication sequence exhausted for stream `{stream}`")]
109    SequenceExhausted { stream: &'static str },
110
111    #[error("the observer did not declare stream `{stream}` in streams(...)")]
112    Undeclared { stream: &'static str },
113
114    #[error("the publication requires {encoded} bytes; the frame limit is {limit}")]
115    Oversized { encoded: usize, limit: usize },
116}
117
118/* === Traits === */
119
120/* === Implementations === */
121
122// A handler that returns a host refusal forwards its failure unchanged rather
123// than flattening the call error's own line into a new cause.
124impl From<CallError> for harmos_domain::Refusal {
125    fn from(error: CallError) -> Self {
126        match error {
127            CallError::Refused { code, failure } => Self::failed(code, failure),
128            error => Self::of(error),
129        }
130    }
131}
132
133impl harmos_sdk::plugin::Context for Supervisor {
134    fn with_generation(&self, token: [u8; 32]) -> Self {
135        let mut host = self.clone();
136        host.generation_context = Some(token);
137        host
138    }
139}
140
141impl PendingCalls {
142    fn new() -> Self {
143        Self {
144            state: Mutex::new(PendingState { closed: false, next_id: 1, waiters: HashMap::new() }),
145            capacity: Semaphore::new(PENDING_CALLS),
146        }
147    }
148
149    fn register(&self, waiter: oneshot::Sender<Result<Vec<u8>, CallError>>) -> Result<u64, CallError> {
150        let mut state = self.state.lock().expect("the pending-call lock is healthy");
151        if state.closed {
152            return Err(CallError::Unavailable);
153        }
154
155        let id = state.next_id;
156        state.next_id = state.next_id.checked_add(1).ok_or(CallError::Unavailable)?;
157        state.waiters.insert(id, waiter);
158
159        Ok(id)
160    }
161
162    fn answer(&self, id: u64, result: Result<Vec<u8>, CallError>) {
163        let waiter = self.state.lock().expect("the pending-call lock is healthy").waiters.remove(&id);
164        if let Some(waiter) = waiter {
165            let _ = waiter.send(result);
166        }
167    }
168
169    fn close(&self) {
170        let waiters = {
171            let mut state = self.state.lock().expect("the pending-call lock is healthy");
172            state.closed = true;
173            std::mem::take(&mut state.waiters)
174        };
175        self.capacity.close();
176        for (_, waiter) in waiters {
177            let _ = waiter.send(Err(CallError::Unavailable));
178        }
179    }
180}
181
182impl Drop for PendingCall<'_> {
183    fn drop(&mut self) {
184        let unanswered = self.pending.state.lock().expect("the pending-call lock is healthy").waiters.remove(&self.id).is_some();
185        if unanswered && let Some(cancel) = self.cancel.take() {
186            cancel.send(Outgoing::Frame(Frame::Cancel { id: self.id }));
187        }
188    }
189}
190
191impl Supervisor {
192    pub(crate) fn new(outgoing: mpsc::Sender<Outgoing>, clock: fn() -> Option<i64>) -> Self {
193        Self {
194            outgoing,
195            sequences: Arc::new(Mutex::new(BTreeMap::new())),
196            pending: Arc::new(PendingCalls::new()),
197            operation_context: None,
198            generation_context: None,
199            call: Arc::default(),
200            clock,
201        }
202    }
203
204    pub(crate) fn for_operation(&self, context: [u8; 32]) -> Self {
205        let mut host = self.clone();
206        host.operation_context = Some(context);
207        host
208    }
209
210    pub(crate) fn for_attempt(
211        &self, context: [u8; 32], values: Vec<harmos_domain::operations::CallValue>,
212    ) -> Result<Self, harmos_domain::operations::ContextError> {
213        let mut host = self.for_operation(context);
214        host.call = Arc::new(harmos_domain::operations::CallContext::from_portable(values)?);
215        Ok(host)
216    }
217
218    pub fn context(&self) -> &harmos_domain::operations::CallContext {
219        &self.call
220    }
221
222    pub fn with_value<T: harmos_domain::operations::ContextValue>(
223        &self, value: T,
224    ) -> Result<Self, harmos_domain::operations::ContextError> {
225        let mut child = self.clone();
226        child.call = Arc::new(self.call.with_value(value)?);
227        Ok(child)
228    }
229
230    // One child call on this attempt or generation; a submitted child answers
231    // its instance id and stays owned by the attempt.
232    pub(super) async fn child_call<O>(&self, request: &O, submit: bool) -> Result<Vec<u8>, CallError>
233    where
234        O: OperationDeclaration,
235    {
236        if self.operation_context.is_none() && self.generation_context.is_none() {
237            return Err(CallError::Unavailable);
238        }
239        self.exchange_with(|id, call| {
240            let params = self.child::<O>(request, submit)?;
241            let frame = Self::request_frame(id, CHILD_METHOD, params)?;
242            call.send(Outgoing::Frame(frame));
243            Ok::<(), CallError>(())
244        })
245        .await
246    }
247
248    // Drive a child this attempt submitted; the host resolves it only within
249    // the same attempt context.
250    pub(super) async fn child_control(&self, instance: Id, action: WireChildAction) -> Result<Vec<u8>, CallError> {
251        let params = self.control_params(instance, action)?;
252        self.exchange_with(|id, call| {
253            let frame = Self::request_frame(id, CHILD_METHOD, params)?;
254            call.send(Outgoing::Frame(frame));
255            Ok::<(), CallError>(())
256        })
257        .await
258    }
259
260    // Send a control without awaiting its answer, for handles that are gone.
261    pub(super) fn child_control_detached(&self, instance: Id, action: WireChildAction) {
262        let Ok(params) = self.control_params(instance, action) else {
263            return;
264        };
265        let (waiter, _) = oneshot::channel();
266        let Ok(id) = self.pending.register(waiter) else {
267            return;
268        };
269        let Ok(frame) = Self::request_frame(id, CHILD_METHOD, params) else {
270            return;
271        };
272        let frame = match self.outgoing.try_send(Outgoing::Frame(frame)) {
273            Ok(()) => return,
274            Err(mpsc::error::TrySendError::Closed(_)) => return,
275            Err(mpsc::error::TrySendError::Full(frame)) => frame,
276        };
277        // A Full Queue Still Delivers The Release Once Capacity Returns
278        if let Ok(runtime) = tokio::runtime::Handle::try_current() {
279            let outgoing = self.outgoing.clone();
280            runtime.spawn(async move {
281                let _ = outgoing.send(frame).await;
282            });
283        }
284    }
285
286    fn control_params(&self, instance: Id, action: WireChildAction) -> Result<Vec<u8>, CallError> {
287        let context = self.operation_context.ok_or(CallError::Unavailable)?;
288        let control = WireChildControl { context, instance: instance.value(), action };
289        control.encode().map_err(|error| CallError::Malformed { method: CHILD_METHOD, message: error.to_string() })
290    }
291
292    fn child<O: OperationDeclaration>(&self, request: &O, submit: bool) -> Result<Vec<u8>, CallError> {
293        if self.operation_context.is_none() && self.generation_context.is_none() {
294            return Err(CallError::Unavailable);
295        }
296        let context = self.operation_context.unwrap_or([0; 32]);
297        let encoded = Frame::request_encoded_len::<O>(request);
298        if encoded > Frame::LIMIT {
299            return Err(CallError::Oversized { method: O::ID, encoded, limit: Frame::LIMIT });
300        }
301        let payload = harmos_domain::Message::encode_to_vec(request);
302        let command = harmos_sdk::operations::WireChildCall {
303            context,
304            generation_context: self.generation_context,
305            values: self.call.portable(),
306            operation: O::ID.to_owned(),
307            payload,
308            submit,
309        };
310        command.encode().map_err(|error| CallError::Malformed { method: O::ID, message: error.to_string() })
311    }
312
313    fn request_frame(id: u64, method: &'static str, params: Vec<u8>) -> Result<Frame, CallError> {
314        let mut frame = Frame::Call { id: u64::MAX, method: method.to_owned(), params };
315        let encoded = frame.encoded_len();
316        if encoded > Frame::LIMIT {
317            return Err(CallError::Oversized { method, encoded, limit: Frame::LIMIT });
318        }
319        let Frame::Call { id: original, .. } = &mut frame else { unreachable!() };
320        *original = id;
321        Ok(frame)
322    }
323
324    async fn exchange_with<'a, E: From<CallError> + Send>(
325        &'a self, prepare: impl FnOnce(u64, mpsc::Permit<'a, Outgoing>) -> Result<(), E> + Send,
326    ) -> Result<Vec<u8>, E> {
327        let permit = self.pending.capacity.acquire().await.map_err(|_| CallError::Unavailable)?;
328        // Reserve both frames atomically so concurrent callers cannot hold all
329        // queue capacity while waiting for a second slot. The cancel permit
330        // stays with the call and makes cancellation reliable even under load.
331        let mut frames = self.outgoing.reserve_many(2).await.map_err(|_| CallError::Unavailable)?;
332        let call = frames.next().expect("two reserved frame slots");
333        let cancel = frames.next().expect("two reserved frame slots");
334        let (waiter, response) = oneshot::channel();
335        let id = self.pending.register(waiter)?;
336        let mut pending = PendingCall { pending: &self.pending, id, _permit: permit, cancel: Some(cancel) };
337
338        if let Err(source) = prepare(id, call) {
339            // No Call Was Sent, So Release The Reserved Cancel Without Sending It.
340            pending.cancel.take();
341            return Err(source);
342        }
343        response.await.map_err(|_| CallError::Unavailable)?.map_err(E::from)
344    }
345
346    pub(crate) fn receive_answer(&self, id: u64, result: Vec<u8>) {
347        self.pending.answer(id, Ok(result));
348    }
349
350    pub(crate) fn receive_refusal(&self, id: u64, refusal: harmos_domain::Refusal) {
351        let code = refusal.code();
352        let failure = refusal.into_failure();
353        self.pending.answer(id, Err(CallError::Refused { code, failure }));
354    }
355
356    pub(crate) fn disconnected(&self) {
357        self.pending.close();
358    }
359
360    // The kit clock that stamps rows when they are produced.
361    pub(super) fn clock(&self) -> fn() -> Option<i64> {
362        self.clock
363    }
364
365    pub(super) fn now(&self) -> Option<i64> {
366        (self.clock)()
367    }
368
369    // Publish rows in order through as few `host.publish` calls as fit the
370    // frame and batch limits; each call awaits its answer before the next.
371    // `stamps[i]` is the production time of `rows[i]`.
372    pub(super) async fn publication<S>(
373        &self, rows: &[S::Row], stamps: &[Option<i64>], context: Option<&str>,
374    ) -> Result<(), super::PublishError>
375    where
376        S: crate::StreamSpec,
377        S::Row: Message,
378    {
379        debug_assert_eq!(rows.len(), stamps.len());
380        if rows.is_empty() {
381            return Ok(());
382        }
383        if self.operation_context.is_none() && self.generation_context.is_none() {
384            return Err(CallError::Unavailable.into());
385        }
386        let mut rest = (rows, stamps);
387        while !rest.0.is_empty() {
388            let mut sent = 0;
389            let bytes = self
390                .exchange_with(|id, call| {
391                    sent = self.with_publication::<S, super::PublishError>(rest.0, rest.1, context, |request| {
392                        let params = self.child::<harmos_domain::operations::streams::Publish>(&request, false)?;
393                        let frame = Self::request_frame(id, CHILD_METHOD, params)?;
394                        call.send(Outgoing::Frame(frame));
395                        Ok(())
396                    })?;
397                    Ok::<(), super::PublishError>(())
398                })
399                .await?;
400            <() as Message>::decode(bytes.as_slice()).map_err(|source| CallError::Malformed {
401                method: harmos_domain::operations::streams::Publish::ID,
402                message: source.to_string(),
403            })?;
404            rest = (&rest.0[sent..], &rest.1[sent..]);
405        }
406        Ok(())
407    }
408
409    // Take the longest leading run of rows that fits one call, reserve its
410    // contiguous sequences and send it under the sequence lock, so frames
411    // leave in sequence order. A row too large on its own still spends its
412    // sequence, exactly as a single-row publication does. The head carries
413    // its stamp; each following row carries its delta from its predecessor,
414    // and a stamp no delta can express starts the next call. A call carries
415    // at most one window of rows, so the host never evicts its own rows.
416    fn with_publication<S: crate::StreamSpec, E: From<PublicationError>>(
417        &self, rows: &[S::Row], stamps: &[Option<i64>], context: Option<&str>,
418        send: impl FnOnce(harmos_domain::operations::streams::Publish) -> Result<(), E>,
419    ) -> Result<usize, E>
420    where
421        S::Row: Message,
422    {
423        let fits = |body: usize| Frame::request_payload_encoded_len(harmos_domain::operations::streams::Publish::ID, body) <= Frame::LIMIT;
424        let mut sequences = self.sequences.lock().expect("the publication sequence lock is healthy");
425        let next = sequences.entry(S::ID).or_insert(0);
426        let first = next.checked_add(1).ok_or(PublicationError::SequenceExhausted { stream: S::ID })?;
427        *next = first;
428        let published_unix_micros = stamps[0];
429        let head = crate::wire::publication_encoded_len(S::ID, S::VERSION, first, rows[0].encoded_len(), published_unix_micros, context);
430        let encoded = |count: usize, following: usize| head.saturating_add(crate::wire::varint_len(count as u64)).saturating_add(following);
431        if !fits(encoded(0, 0)) {
432            return Err(PublicationError::Oversized { encoded: encoded(0, 0), limit: Frame::LIMIT }.into());
433        }
434        let mut following = 0_usize;
435        let mut deltas = Vec::with_capacity(rows.len() - 1);
436        for (row, pair) in rows[1..].iter().zip(stamps.windows(2)) {
437            if deltas.len() + 1 >= S::WINDOW {
438                break;
439            }
440            let delta = match (pair[0], pair[1]) {
441                (_, None) => None,
442                (Some(previous), Some(stamp)) => match stamp.checked_sub(previous) {
443                    Some(delta) => Some(delta),
444                    None => break,
445                },
446                (None, Some(_)) => break,
447            };
448            let grown = following.saturating_add(crate::wire::following_encoded_len(row.encoded_len(), delta));
449            let total = encoded(deltas.len() + 1, grown);
450            if total > PUBLICATION_BATCH_BYTES || !fits(total) || *next == u64::MAX {
451                break;
452            }
453            following = grown;
454            deltas.push(delta);
455            *next += 1;
456        }
457        let taken = deltas.len() + 1;
458        send(harmos_domain::operations::streams::Publish {
459            stream: S::ID.to_owned(),
460            schema: S::VERSION,
461            sequence: Some(first),
462            published_unix_micros,
463            context: context.map(str::to_owned),
464            payload: rows[0].encode_to_vec(),
465            following: rows[1..taken]
466                .iter()
467                .zip(deltas)
468                .map(|(row, published_delta_micros)| harmos_domain::operations::streams::FollowingRow {
469                    payload: row.encode_to_vec(),
470                    published_delta_micros,
471                })
472                .collect(),
473        })?;
474        Ok(taken)
475    }
476
477    pub fn exiting(&self, reason: impl Into<String>) -> bool {
478        self.outgoing.try_send(Outgoing::Exiting(reason.into())).is_ok()
479    }
480
481    pub fn abandon(&self) -> bool {
482        self.outgoing.try_send(Outgoing::Abandon).is_ok()
483    }
484}
485
486/* === Public Functions === */
487
488/* === Private Functions === */
489
490fn refused_line(code: i64, failure: &harmos_domain::Failure) -> String {
491    match failure.context().operation() {
492        Some(operation) => format!("operation `{operation}` failed"),
493        None => format!("the host refused the call ({code})"),
494    }
495}
496
497pub(super) fn published_at() -> Option<i64> {
498    match std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH) {
499        Ok(elapsed) => i64::try_from(elapsed.as_micros()).ok(),
500        Err(error) => i64::try_from(error.duration().as_micros()).ok().and_then(i64::checked_neg),
501    }
502}
503
504/* === Tests === */
harmos_sidecar/kit/host.rs

harmos_sidecar/kit/
host.rs
