harmos_sidecar/kit/
mod.rs

1/* ***
2 *
3 * Harmos
4 *
5 * SPDX-License-Identifier: Apache-2.0
6 *
7 * ***
8 */
9
10/* === Modules === */
11
12mod carrier;
13mod catalog;
14mod connection;
15mod host;
16mod identity;
17mod instance;
18mod local;
19mod observer;
20mod observer_host;
21mod plugin;
22mod publisher;
23mod task;
24
25pub mod __private {
26    pub use super::observer::{Declarations as ObserverDeclarations, Inventory as ObserverInventory, Outputs as ObserverOutputs};
27    pub use super::task::spawn as spawn_observer;
28    pub use crate as sidecar;
29    pub use harmos_domain::{qualified, qualified_len, qualify};
30    pub use harmos_sdk::__private::*;
31}
32
33/* === Re-Exports === */
34
35pub use harmos_sdk::{ExecutionContext, Operation, Service};
36
37pub use catalog::{CATALOG_EXPORT_FLAG, CATALOG_EXPORT_LIMIT};
38pub use host::{CallError, PublicationError, Supervisor};
39pub use identity::ActivationId;
40pub use instance::OperationInstance;
41pub use observer::{Activation, Activity, DemandKey, Instance, Observer, ObserverAttachment, ObserverContext, ObserverError, Tick};
42pub use observer_host::{ObserverCaller, ObserverHost, ObserverPublisher, ObserverStreamPublisher};
43pub use plugin::{PluginNative, run_plugin, serve_plugin_channels};
44pub use publisher::{Batch, PublishError, Publisher};
45
46/* === Imports === */
47
48// Standard Library Imports
49use std::collections::BTreeMap;
50use std::future::Future;
51use std::pin::Pin;
52use std::sync::{Arc, Mutex};
53// Third Party Imports
54use tokio::sync::{mpsc, oneshot};
55// Local Imports
56use crate::Frame;
57use carrier::{Failure, Input, Sender};
58use harmos_domain::{Refusal, UNKNOWN, UNREADABLE, reserved};
59use host::Outgoing;
60
61/* === Constants === */
62
63const INCOMING_REQUESTS: usize = 128;
64
65/* === Macros === */
66
67/* === Type Aliases === */
68
69type Prepare = Box<dyn Fn(&str) -> Result<Serving, Refusal> + Send + Sync>;
70
71type Cancellations = Arc<Mutex<BTreeMap<u64, Option<oneshot::Sender<()>>>>>;
72type Table = Arc<Mutex<harmos_sdk::__private::WireTable>>;
73type SharedOpening = Arc<tokio::sync::Mutex<Opening>>;
74type Retirement = Box<dyn Fn() -> Pin<Box<dyn Future<Output = Result<(), Refusal>> + Send>> + Send + Sync>;
75
76/* === Structs === */
77
78struct Incoming {
79    frame: Frame,
80    cancelled: Option<oneshot::Receiver<()>>,
81    consumption: Option<crate::transport::Consumption>,
82}
83
84struct ConnectionGuard(Supervisor);
85
86struct Worker(tokio::task::JoinHandle<Result<(), Refusal>>);
87
88pub(crate) struct Opening {
89    serving: Option<Serving>,
90    prepare: Prepare,
91    catalog: crate::PluginCatalog,
92}
93
94struct Serving {
95    owner: String,
96    operations: Table,
97    retirement: Retirement,
98    quiescence: Retirement,
99}
100
101/* === Enums === */
102
103enum Answered {
104    Nothing,
105    Frame(Frame),
106    Deferred(Pin<Box<dyn Future<Output = Frame> + Send>>),
107}
108
109/* === Traits === */
110
111/* === Implementations === */
112
113impl Drop for ConnectionGuard {
114    fn drop(&mut self) {
115        self.0.disconnected();
116    }
117}
118
119impl Drop for Worker {
120    fn drop(&mut self) {
121        self.0.abort();
122    }
123}
124
125impl Opening {
126    async fn cleanup(&mut self) -> Result<(), Refusal> {
127        let Some(serving) = &self.serving else {
128            return Ok(());
129        };
130        let settlement = serving.operations.lock().map_err(Refusal::of)?.close();
131        settlement.await.map_err(Refusal::of)?;
132        (serving.retirement)().await?;
133        let retired = serving.operations.lock().map_err(Refusal::of)?.release().map_err(Refusal::of)?;
134        drop(retired);
135        Ok(())
136    }
137}
138
139/* === Public Functions === */
140
141/* === Private Functions === */
142
143async fn exchange(
144    opening: SharedOpening, inbound: &mut Input, outgoing: &Sender, completed: oneshot::Sender<()>,
145    mut stopping: tokio::sync::watch::Receiver<bool>, clock: fn() -> Option<i64>,
146) -> Result<(), Refusal> {
147    let (to_host, mut outbound) = mpsc::channel(1024);
148    let host = Supervisor::new(to_host.clone(), clock);
149    let _connection = ConnectionGuard(host.clone());
150    let (requests_tx, requests_rx) = mpsc::channel(INCOMING_REQUESTS);
151    let cancellations: Cancellations = Arc::new(Mutex::new(BTreeMap::new()));
152    let (shutdown_tx, shutdown_rx) = oneshot::channel();
153    let mut worker =
154        Worker(tokio::spawn(dispatch_requests(opening, host.clone(), requests_rx, cancellations.clone(), to_host.clone(), shutdown_rx)));
155    let mut abandoned = false;
156    let mut active_requests = std::collections::BTreeSet::new();
157    let mut exit_after_request = None;
158    let mut abandon_after_request = false;
159    let mut worker_finished = false;
160    let mut outcome = Ok(());
161
162    let serving = async {
163        loop {
164            let sending = {
165                let worker_done = async {
166                    if worker_finished {
167                        std::future::pending::<Result<Result<(), Refusal>, tokio::task::JoinError>>().await
168                    } else {
169                        (&mut worker.0).await
170                    }
171                };
172                tokio::select! {
173                    arrival = inbound.receive() => match arrival {
174                        Ok(Some(carrier::Arrival { frame, cancelled: early_cancelled, consumption })) => match frame {
175                            Frame::Answer { id, result } => {
176                                host.receive_answer(id, result);
177                                if let Some(receipt) = consumption { let _ = receipt.consumed(); }
178                                continue;
179                            }
180                            Frame::Refusal { id, refusal } => {
181                                host.receive_refusal(id, refusal);
182                                if let Some(receipt) = consumption { let _ = receipt.consumed(); }
183                                continue;
184                            }
185                            Frame::Cancel { id } => {
186                                if let Some(cancel) = cancellations.lock().expect("request cancellation lock")
187                                    .get_mut(&id).and_then(Option::take)
188                                {
189                                    let _ = cancel.send(());
190                                }
191                                continue;
192                            }
193                            frame @ (Frame::Call { .. } | Frame::Push { .. }) => {
194                                let cancelled = if let Frame::Call { id, .. } = &frame {
195                                    let mut pending = cancellations.lock().expect("request cancellation lock");
196                                    if pending.contains_key(id) {
197                                        outgoing.fail(Failure::Invalid("duplicate live call id".to_owned()));
198                                        break;
199                                    }
200                                    let (cancel, cancelled) = oneshot::channel();
201                                    if early_cancelled {
202                                        let _ = cancel.send(());
203                                        pending.insert(*id, None);
204                                    } else {
205                                        pending.insert(*id, Some(cancel));
206                                    }
207                                    Some(cancelled)
208                                } else {
209                                    None
210                                };
211                                match requests_tx.try_send(Incoming { frame, cancelled, consumption }) {
212                                    Ok(()) => continue,
213                                    Err(mpsc::error::TrySendError::Full(request)) => {
214                                        let id = match &request.frame { Frame::Call { id, .. } => Some(*id), _ => None };
215                                        drop(request.frame);
216                                        if let Some(receipt) = request.consumption { let _ = receipt.consumed(); }
217                                        if let Some(id) = id {
218                                            cancellations.lock().expect("request cancellation lock").remove(&id);
219                                            let refusal = Refusal::of("the sidecar request queue is full");
220                                            if to_host.try_send(Outgoing::Frame(Frame::Refusal {
221                                                id,
222                                                refusal,
223                                            })).is_err() {
224                                                break;
225                                            }
226                                            continue;
227                                        }
228                                        outgoing.fail(Failure::Capacity);
229                                        break;
230                                    }
231                                    Err(mpsc::error::TrySendError::Closed(_)) => break,
232                                }
233                            }
234                        },
235                        Ok(None) => break,
236                        Err(error) => { outgoing.fail(Failure::Invalid(error.to_string())); break; },
237                    },
238                    asked = outbound.recv() => match asked {
239                        None => break,
240                        Some(asked) => asked,
241                    },
242                    result = worker_done => {
243                        worker_finished = true;
244                        match result {
245                            Err(error) => {
246                                outcome = Err(Refusal::of(error));
247                                outgoing.fail(Failure::Panicked);
248                                break;
249                            }
250                            Ok(result) => {
251                                outcome = result;
252                                match outbound.try_recv() {
253                                    Ok(asked) => asked,
254                                    Err(_) => break,
255                                }
256                            },
257                        }
258                    }
259                }
260            };
261
262            match sending {
263                Outgoing::Frame(frame) => {
264                    let answered = match &frame {
265                        Frame::Answer { id, .. } | Frame::Refusal { id, .. } => Some(*id),
266                        _ => None,
267                    };
268                    if send(outgoing, frame).await.is_err() {
269                        break;
270                    }
271                    if answered.is_some_and(|id| active_requests.remove(&id)) && active_requests.is_empty() {
272                        if abandon_after_request {
273                            abandoned = true;
274                            break;
275                        }
276                        if let Some(reason) = exit_after_request.take() {
277                            let _ = send(outgoing, Frame::exiting(reason)).await;
278                            break;
279                        }
280                    }
281                }
282                Outgoing::Started(id) => {
283                    active_requests.insert(id);
284                }
285                Outgoing::Finished(id) => {
286                    if active_requests.remove(&id) && active_requests.is_empty() {
287                        if abandon_after_request {
288                            abandoned = true;
289                            break;
290                        }
291                        if let Some(reason) = exit_after_request.take() {
292                            let _ = send(outgoing, Frame::exiting(reason)).await;
293                            break;
294                        }
295                    }
296                }
297                Outgoing::Exiting(reason) => {
298                    if !active_requests.is_empty() {
299                        exit_after_request.get_or_insert(reason);
300                    } else {
301                        let _ = send(outgoing, Frame::exiting(reason)).await;
302                        break;
303                    }
304                }
305                Outgoing::Abandon => {
306                    if !active_requests.is_empty() {
307                        abandon_after_request = true;
308                    } else {
309                        abandoned = true;
310                        break;
311                    }
312                }
313            }
314        }
315    };
316    tokio::select! {
317        biased;
318        _ = stopping.wait_for(|closing| *closing) => {}
319        _ = serving => {}
320    }
321
322    host.disconnected();
323    outbound.close();
324    drop(requests_tx);
325    let _ = shutdown_tx.send(abandoned);
326    if !worker_finished {
327        outcome = (&mut worker.0).await.map_err(Refusal::of).and_then(|result| result);
328    }
329    let _ = completed.send(());
330    outcome
331}
332
333async fn dispatch_requests(
334    opening: SharedOpening, host: Supervisor, mut requests: mpsc::Receiver<Incoming>, cancellations: Cancellations,
335    outgoing: mpsc::Sender<Outgoing>, mut shutdown: oneshot::Receiver<bool>,
336) -> Result<(), Refusal> {
337    let mut abandoned = false;
338    let mut replies = tokio::task::JoinSet::new();
339    loop {
340        let request = tokio::select! {
341            biased;
342            ended = &mut shutdown => {
343                abandoned = ended.unwrap_or(false);
344                break;
345            }
346            completed = replies.join_next(), if !replies.is_empty() => {
347                if let Some(Err(error)) = completed {
348                    tracing::warn!(%error, "carrier reply task failed");
349                    break;
350                }
351                continue;
352            }
353            request = requests.recv() => request,
354        };
355        let Some(Incoming {
356            frame: request,
357            mut cancelled,
358            consumption,
359        }) = request
360        else {
361            break;
362        };
363        let id = match &request {
364            Frame::Call { id, .. } => Some(*id),
365            _ => None,
366        };
367
368        if let Frame::Call { id, .. } = &request
369            && outgoing.send(Outgoing::Started(*id)).await.is_err()
370        {
371            break;
372        }
373
374        if replies.len() >= INCOMING_REQUESTS {
375            drop(request);
376            if let Some(receipt) = consumption {
377                let _ = receipt.consumed();
378            }
379            if let Some(id) = id {
380                cancellations.lock().expect("request cancellation lock").remove(&id);
381                let _ = outgoing
382                    .send(Outgoing::Frame(Frame::Refusal {
383                        id,
384                        refusal: Refusal::new(UNKNOWN, "carrier reply capacity exhausted".to_owned()),
385                    }))
386                    .await;
387            }
388            continue;
389        }
390
391        let mut was_cancelled = false;
392        let answered = tokio::select! {
393            biased;
394            ended = &mut shutdown => {
395                abandoned = ended.unwrap_or(false);
396                None
397            }
398            _ = async {
399                match cancelled.as_mut() {
400                    Some(cancelled) => { let _ = cancelled.await; }
401                    None => std::future::pending::<()>().await,
402                }
403            } => {
404                was_cancelled = true;
405                None
406            }
407            answered = async {
408                let mut opening = opening.lock().await;
409                answer(&mut opening, &host, request).await
410            } => Some(answered),
411        };
412        if let Some(receipt) = consumption {
413            let _ = receipt.consumed();
414        }
415        if let Some(id) = id {
416            if !matches!(&answered, Some(Answered::Deferred(_))) {
417                cancellations.lock().expect("request cancellation lock").remove(&id);
418            }
419            if was_cancelled {
420                if outgoing.send(Outgoing::Finished(id)).await.is_err() {
421                    break;
422                }
423                continue;
424            }
425        }
426        let Some(answered) = answered else {
427            break;
428        };
429
430        match answered {
431            Answered::Deferred(reply) => {
432                let outgoing = outgoing.clone();
433                let cancellations = Arc::clone(&cancellations);
434                replies.spawn(async move {
435                    let response = tokio::select! {
436                        biased;
437                        _ = async {
438                            match cancelled.as_mut() {
439                                Some(cancelled) => { let _ = cancelled.await; }
440                                None => std::future::pending::<()>().await,
441                            }
442                        } => id.map(Outgoing::Finished),
443                        frame = reply => Some(Outgoing::Frame(frame)),
444                    };
445                    if let Some(response) = response {
446                        let _ = outgoing.send(response).await;
447                    }
448                    if let Some(id) = id {
449                        cancellations.lock().expect("request cancellation lock").remove(&id);
450                    }
451                });
452            }
453            Answered::Nothing => continue,
454            Answered::Frame(frame) => {
455                if outgoing.send(Outgoing::Frame(frame)).await.is_err() {
456                    break;
457                }
458            }
459        }
460    }
461
462    replies.abort_all();
463    while replies.join_next().await.is_some() {}
464    cancellations.lock().expect("request cancellation lock").clear();
465
466    // Abandon represents process death. Every other ending joins work and
467    // cleans up the state, including a handler cancelled when its connection
468    // disappears; outbound host calls have already been failed by `exchange`.
469    let mut opening = opening.lock().await;
470    if abandoned {
471        // A local double simulates process death but its native threads still
472        // belong to this process. Join them before declaring the peer ended.
473        if let Some(serving) = &opening.serving {
474            (serving.quiescence)().await?;
475        }
476    } else {
477        opening.cleanup().await?;
478    }
479    Ok(())
480}
481
482async fn answer(opening: &mut Opening, host: &Supervisor, frame: Frame) -> Answered {
483    let (id, method, params) = match frame {
484        Frame::Call { id, method, params } => (id, method, params),
485        _ => return Answered::Nothing,
486    };
487    if method == Frame::REGISTER {
488        let owner = match <String as crate::Message>::decode(&params) {
489            Ok(owner) => owner,
490            Err(error) => {
491                return Answered::Frame(Frame::Refusal {
492                    id,
493                    refusal: Refusal::new(UNREADABLE, format!("invalid owner assignment: {error}")),
494                });
495            }
496        };
497        if let Some(serving) = &opening.serving {
498            if serving.owner != owner {
499                return Answered::Frame(Frame::Refusal {
500                    id,
501                    refusal: Refusal::of("carrier owner assignment cannot change"),
502                });
503            }
504        } else {
505            match (opening.prepare)(&owner) {
506                Ok(serving) => opening.serving = Some(serving),
507                Err(error) => return Answered::Frame(Frame::Refusal { id, refusal: error }),
508            }
509        }
510        return match serde_json::to_vec(&opening.catalog) {
511            Ok(result) => Answered::Frame(Frame::Answer { id, result: result.into() }),
512            Err(error) => Answered::Frame(Frame::Refusal {
513                id,
514                refusal: Refusal::new(UNREADABLE, error.to_string()),
515            }),
516        };
517    }
518    if method == crate::operations::WIRE_METHOD {
519        let command = match crate::operations::WireCommand::decode(&params) {
520            Ok(command) => command,
521            Err(error) => {
522                return Answered::Frame(Frame::Refusal {
523                    id,
524                    refusal: Refusal::new(UNREADABLE, error.to_string()),
525                });
526            }
527        };
528        let Some(serving) = &opening.serving else {
529            return Answered::Frame(Frame::Refusal {
530                id,
531                refusal: Refusal::of("carrier owner has not been assigned"),
532            });
533        };
534        let table = &serving.operations;
535        let operation_host = match &command {
536            crate::operations::WireCommand::Initialize { context, .. } => Ok(host.for_operation(*context)),
537            crate::operations::WireCommand::Attempt { context, values, .. }
538            | crate::operations::WireCommand::StreamAttempt { context, values, .. } => host.for_attempt(*context, values.clone()),
539            _ => Ok(host.clone()),
540        };
541        let operation_host = match operation_host {
542            Ok(operation_host) => operation_host,
543            Err(error) => {
544                return Answered::Frame(Frame::Refusal {
545                    id,
546                    refusal: Refusal::new(UNREADABLE, error.to_string()),
547                });
548            }
549        };
550        let phase = match table.lock() {
551            Ok(mut table) => {
552                let call = operation_host.context().clone();
553                let mut dependencies = harmos_sdk::__private::ResourceSet::new();
554                let caller = Arc::new(harmos_sdk::operations::Calls(Arc::new(operation_host.clone())));
555                if let Err(error) = dependencies.insert(caller).and_then(|()| dependencies.insert(Arc::new(operation_host))) {
556                    return Answered::Frame(Frame::Refusal { id, refusal: Refusal::of(error) });
557                }
558                let context = crate::ExecutionContext::new(dependencies, call);
559                table.begin(command, Arc::new(context))
560            }
561            Err(error) => {
562                return Answered::Frame(Frame::Refusal {
563                    id,
564                    refusal: Refusal::new(UNREADABLE, error.to_string()),
565                });
566            }
567        };
568        return Answered::Deferred(Box::pin(async move {
569            let answer = phase.await;
570            // Refuse Before Serialization
571            // Binary Envelopes Add Framing Bytes To The Unchanged Payload.
572            let minimum = match &answer {
573                crate::operations::WireAnswer::Ready { payload } => payload.len(),
574                crate::operations::WireAnswer::Stream { payload, .. } => payload.as_ref().map_or(0, Vec::len),
575                crate::operations::WireAnswer::Failed { failure } => failure.cause().len(),
576            };
577            if minimum >= Frame::LIMIT {
578                return oversized_reply(id);
579            }
580            match answer.encode() {
581                Ok(result) => Frame::Answer { id, result: result.into() },
582                Err(harmos_sdk::operations::WireCodecError::Oversized) => oversized_reply(id),
583                Err(error) => Frame::Refusal {
584                    id,
585                    refusal: Refusal::new(UNREADABLE, error.to_string()),
586                },
587            }
588        }));
589    }
590    if reserved(&method) {
591        return Answered::Frame(Frame::Refusal {
592            id,
593            refusal: Refusal::new(UNKNOWN, format!("this sidecar does not serve the reserved `{method}`")),
594        });
595    }
596
597    Answered::Frame(Frame::Refusal {
598        id,
599        refusal: Refusal::new(
600            UNKNOWN,
601            format!("operation work must use the `{}` phase contract; received `{method}`", crate::operations::WIRE_METHOD),
602        ),
603    })
604}
605
606fn oversized_reply(id: u64) -> Frame {
607    Frame::Refusal {
608        id,
609        refusal: Refusal::new(Frame::OVERSIZED, format!("the route reply exceeds the {}-byte sidecar frame limit", Frame::LIMIT)),
610    }
611}
612
613async fn send(outgoing: &Sender, frame: Frame) -> std::io::Result<()> {
614    let call = match &frame {
615        Frame::Answer { id, .. } | Frame::Refusal { id, .. } => Some(*id),
616        _ => None,
617    };
618    let frame = match call {
619        Some(id) if !frame.fits() => oversized_reply(id),
620        _ => frame,
621    };
622    outgoing.send(frame).await
623}
624
625/* === Tests === */
harmos_sidecar/kit/mod.rs

harmos_sidecar/kit/
mod.rs
