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<crate::FramePayload, 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    assignments: harmos_sdk::operations::OperationAssignments,
65    call: Arc<harmos_domain::operations::CallContext>,
66    clock: fn() -> Option<i64>,
67}
68
69/* === Enums === */
70
71pub(crate) enum Outgoing {
72    Frame(Frame),
73    Started(u64),
74    Finished(u64),
75    Exiting(String),
76    Abandon,
77}
78
79#[derive(Clone, Debug, PartialEq, Eq, thiserror::Error)]
80pub enum CallError {
81    #[error("`{method}` requires process-local request ownership")]
82    LocalResources { method: &'static str },
83
84    #[error("caller may not invoke operation `{operation}` without a matching host assignment")]
85    Unassigned { operation: &'static str },
86
87    // The host's refusal carries the failure unchanged; this layer names only
88    // the operation the host attributed it to.
89    #[error("{}", refused_line(*.code, .failure))]
90    Refused {
91        code: i64,
92
93        #[source]
94        failure: harmos_domain::Failure,
95    },
96
97    #[error("the host connection is unavailable")]
98    Unavailable,
99
100    #[error("the host returned a malformed reply for `{method}`: {message}")]
101    Malformed { method: &'static str, message: String },
102
103    #[error("the request for `{method}` requires {encoded} bytes; the frame limit is {limit}")]
104    Oversized { method: &'static str, encoded: usize, limit: usize },
105
106    #[error(transparent)]
107    Publication(#[from] PublicationError),
108}
109
110#[derive(Clone, Copy, Debug, PartialEq, Eq, thiserror::Error)]
111pub enum PublicationError {
112    #[error("publication sequence exhausted for stream `{stream}`")]
113    SequenceExhausted { stream: &'static str },
114
115    #[error("the observer did not declare stream `{stream}` in streams(...)")]
116    Undeclared { stream: &'static str },
117
118    #[error("the publication requires {encoded} bytes; the frame limit is {limit}")]
119    Oversized { encoded: usize, limit: usize },
120}
121
122/* === Traits === */
123
124/* === Implementations === */
125
126// A handler that returns a host refusal forwards its failure unchanged rather
127// than flattening the call error's own line into a new cause.
128impl From<CallError> for harmos_domain::Refusal {
129    fn from(error: CallError) -> Self {
130        match error {
131            CallError::Refused { code, failure } => Self::failed(code, failure),
132            error => Self::of(error),
133        }
134    }
135}
136
137impl harmos_sdk::lifecycle::Context for Supervisor {
138    type Configuration = harmos_sdk::__private::WireTable;
139}
140
141impl harmos_sdk::plugin::Context for Supervisor {
142    fn assign_operations(&self, bytes: &[u8]) -> Result<(), harmos_sdk::operations::ServingError> {
143        self.assignments.install(bytes)
144    }
145
146    fn with_generation(&self, token: [u8; 32]) -> Self {
147        let mut host = self.clone();
148        host.generation_context = Some(token);
149        host
150    }
151}
152
153impl harmos_sdk::operations::Caller for Supervisor {
154    fn call(
155        &self, target: harmos_sdk::operations::Target, request: Box<dyn std::any::Any + Send>,
156    ) -> harmos_sdk::operations::CallFuture<'_> {
157        Box::pin(async move {
158            let result = async move {
159                if self.operation_context.is_none() && self.generation_context.is_none() {
160                    return Err(CallError::Unavailable);
161                }
162                let assigned = self.assignments.resolve_target(&target).map_err(|_| CallError::Unassigned { operation: target.id })?;
163                let payload = (target.encode)(request.as_ref()).ok_or(CallError::Unavailable)?;
164                let values = self.call.portable();
165                let encoded = Frame::child_payload_encoded_len(
166                    &assigned.operation,
167                    assigned.generation,
168                    payload.len(),
169                    self.generation_context.is_some(),
170                    &values,
171                );
172                if encoded > Frame::LIMIT {
173                    return Err(CallError::Oversized {
174                        method: target.id,
175                        encoded,
176                        limit: Frame::LIMIT,
177                    });
178                }
179                let command = harmos_sdk::operations::WireChildCall {
180                    context: self.operation_context.unwrap_or([0; 32]),
181                    generation_context: self.generation_context,
182                    values,
183                    operation: assigned.operation.clone(),
184                    fingerprint: assigned.fingerprint,
185                    generation: assigned.generation,
186                    payload,
187                    submit: false,
188                };
189                let params = command.encode().map_err(|error| CallError::Malformed {
190                    method: target.id,
191                    message: error.to_string(),
192                })?;
193                let bytes = self
194                    .exchange_with(|id, call| {
195                        let frame = Self::request_frame(id, CHILD_METHOD, params)?;
196                        call.send(Outgoing::Frame(frame));
197                        Ok::<(), CallError>(())
198                    })
199                    .await?;
200                (target.decode)(bytes.as_slice()).map_err(|error| CallError::Malformed {
201                    method: target.id,
202                    message: error.to_string(),
203                })
204            }
205            .await;
206            result.map_err(|error| harmos_sdk::operations::InvocationError::failed(target.id, error))
207        })
208    }
209}
210
211impl PendingCalls {
212    fn new() -> Self {
213        Self {
214            state: Mutex::new(PendingState {
215                closed: false,
216                next_id: 1,
217                waiters: HashMap::new(),
218            }),
219            capacity: Semaphore::new(PENDING_CALLS),
220        }
221    }
222
223    fn register(&self, waiter: oneshot::Sender<Result<crate::FramePayload, CallError>>) -> Result<u64, CallError> {
224        let mut state = self.state.lock().expect("the pending-call lock is healthy");
225        if state.closed {
226            return Err(CallError::Unavailable);
227        }
228
229        let id = state.next_id;
230        state.next_id = state.next_id.checked_add(1).ok_or(CallError::Unavailable)?;
231        state.waiters.insert(id, waiter);
232
233        Ok(id)
234    }
235
236    fn answer(&self, id: u64, result: Result<crate::FramePayload, CallError>) {
237        let waiter = self.state.lock().expect("the pending-call lock is healthy").waiters.remove(&id);
238        if let Some(waiter) = waiter {
239            let _ = waiter.send(result);
240        }
241    }
242
243    fn close(&self) {
244        let waiters = {
245            let mut state = self.state.lock().expect("the pending-call lock is healthy");
246            state.closed = true;
247            std::mem::take(&mut state.waiters)
248        };
249        self.capacity.close();
250        for (_, waiter) in waiters {
251            let _ = waiter.send(Err(CallError::Unavailable));
252        }
253    }
254}
255
256impl Drop for PendingCall<'_> {
257    fn drop(&mut self) {
258        let unanswered = self.pending.state.lock().expect("the pending-call lock is healthy").waiters.remove(&self.id).is_some();
259        if unanswered && let Some(cancel) = self.cancel.take() {
260            cancel.send(Outgoing::Frame(Frame::Cancel { id: self.id }));
261        }
262    }
263}
264
265impl Supervisor {
266    pub(crate) fn new(outgoing: mpsc::Sender<Outgoing>, clock: fn() -> Option<i64>) -> Self {
267        Self {
268            outgoing,
269            sequences: Arc::new(Mutex::new(BTreeMap::new())),
270            pending: Arc::new(PendingCalls::new()),
271            operation_context: None,
272            generation_context: None,
273            assignments: Default::default(),
274            call: Arc::default(),
275            clock,
276        }
277    }
278
279    pub(crate) fn for_operation(&self, context: [u8; 32]) -> Self {
280        let mut host = self.clone();
281        host.operation_context = Some(context);
282        host
283    }
284
285    pub(crate) fn for_attempt(
286        &self, context: [u8; 32], values: Vec<harmos_domain::operations::CallValue>,
287    ) -> Result<Self, harmos_domain::operations::ContextError> {
288        let mut host = self.for_operation(context);
289        host.call = Arc::new(harmos_domain::operations::CallContext::from_portable(values)?);
290        Ok(host)
291    }
292
293    pub fn context(&self) -> &harmos_domain::operations::CallContext {
294        &self.call
295    }
296
297    pub fn with_value<T: harmos_domain::operations::ContextValue>(
298        &self, value: T,
299    ) -> Result<Self, harmos_domain::operations::ContextError> {
300        let mut child = self.clone();
301        child.call = Arc::new(self.call.with_value(value)?);
302        Ok(child)
303    }
304
305    // One child call on this attempt or generation; a submitted child answers
306    // its instance id and stays owned by the attempt.
307    pub(super) async fn child_call<O>(&self, request: &O::Request, submit: bool) -> Result<crate::FramePayload, CallError>
308    where
309        O: OperationDeclaration,
310    {
311        if self.operation_context.is_none() && self.generation_context.is_none() {
312            return Err(CallError::Unavailable);
313        }
314        self.exchange_with(|id, call| {
315            let params = self.child::<O>(request, submit)?;
316            let frame = Self::request_frame(id, CHILD_METHOD, params)?;
317            call.send(Outgoing::Frame(frame));
318            Ok::<(), CallError>(())
319        })
320        .await
321    }
322
323    // Drive a child this attempt submitted; the host resolves it only within
324    // the same attempt context.
325    pub(super) async fn child_control(&self, instance: Id, action: WireChildAction) -> Result<crate::FramePayload, CallError> {
326        let params = self.control_params(instance, action)?;
327        self.exchange_with(|id, call| {
328            let frame = Self::request_frame(id, CHILD_METHOD, params)?;
329            call.send(Outgoing::Frame(frame));
330            Ok::<(), CallError>(())
331        })
332        .await
333    }
334
335    // Send a control without awaiting its answer, for handles that are gone.
336    pub(super) fn child_control_detached(&self, instance: Id, action: WireChildAction) {
337        let Ok(params) = self.control_params(instance, action) else {
338            return;
339        };
340        let (waiter, _) = oneshot::channel();
341        let Ok(id) = self.pending.register(waiter) else {
342            return;
343        };
344        let Ok(frame) = Self::request_frame(id, CHILD_METHOD, params) else {
345            return;
346        };
347        let frame = match self.outgoing.try_send(Outgoing::Frame(frame)) {
348            Ok(()) => return,
349            Err(mpsc::error::TrySendError::Closed(_)) => return,
350            Err(mpsc::error::TrySendError::Full(frame)) => frame,
351        };
352        // A Full Queue Still Delivers The Release Once Capacity Returns
353        if let Ok(runtime) = tokio::runtime::Handle::try_current() {
354            let outgoing = self.outgoing.clone();
355            runtime.spawn(async move {
356                let _ = outgoing.send(frame).await;
357            });
358        }
359    }
360
361    fn control_params(&self, instance: Id, action: WireChildAction) -> Result<Vec<u8>, CallError> {
362        let context = self.operation_context.ok_or(CallError::Unavailable)?;
363        let control = WireChildControl {
364            context,
365            instance: instance.value(),
366            action,
367        };
368        control.encode().map_err(|error| CallError::Malformed {
369            method: CHILD_METHOD,
370            message: error.to_string(),
371        })
372    }
373
374    fn child<O: OperationDeclaration>(&self, request: &O::Request, submit: bool) -> Result<Vec<u8>, CallError> {
375        if self.operation_context.is_none() && self.generation_context.is_none() {
376            return Err(CallError::Unavailable);
377        }
378        let values = self.call.portable();
379        self.child_with_values::<O>(request, submit, values)
380    }
381
382    fn child_with_values<O: OperationDeclaration>(
383        &self, request: &O::Request, submit: bool, values: Vec<harmos_domain::operations::CallValue>,
384    ) -> Result<Vec<u8>, CallError> {
385        let context = self.operation_context.unwrap_or([0; 32]);
386        let target = self.assignments.resolve::<O>().map_err(|_| CallError::Unassigned { operation: O::ID })?;
387        let payload_bytes = harmos_domain::Message::encoded_size(request);
388        let encoded = Frame::child_payload_encoded_len(
389            &target.operation,
390            target.generation,
391            payload_bytes,
392            self.generation_context.is_some(),
393            &values,
394        );
395        if encoded > Frame::LIMIT {
396            return Err(CallError::Oversized {
397                method: O::ID,
398                encoded,
399                limit: Frame::LIMIT,
400            });
401        }
402        let payload = harmos_domain::Message::encode(request);
403        let command = harmos_sdk::operations::WireChildCall {
404            context,
405            generation_context: self.generation_context,
406            values,
407            operation: target.operation.clone(),
408            fingerprint: target.fingerprint,
409            generation: target.generation,
410            payload,
411            submit,
412        };
413        command.encode().map_err(|error| CallError::Malformed {
414            method: O::ID,
415            message: error.to_string(),
416        })
417    }
418
419    fn request_frame(id: u64, method: &'static str, params: Vec<u8>) -> Result<Frame, CallError> {
420        let mut frame = Frame::Call {
421            id: u64::MAX,
422            method: method.to_owned(),
423            params: params.into(),
424        };
425        let encoded = frame.encoded_len();
426        if encoded > Frame::LIMIT {
427            return Err(CallError::Oversized { method, encoded, limit: Frame::LIMIT });
428        }
429        let Frame::Call { id: original, .. } = &mut frame else { unreachable!() };
430        *original = id;
431        Ok(frame)
432    }
433
434    async fn exchange_with<'a, E: From<CallError> + Send>(
435        &'a self, prepare: impl FnOnce(u64, mpsc::Permit<'a, Outgoing>) -> Result<(), E> + Send,
436    ) -> Result<crate::FramePayload, E> {
437        let permit = self.pending.capacity.acquire().await.map_err(|_| CallError::Unavailable)?;
438        // Reserve both frames atomically so concurrent callers cannot hold all
439        // queue capacity while waiting for a second slot. The cancel permit
440        // stays with the call and makes cancellation reliable even under load.
441        let mut frames = self.outgoing.reserve_many(2).await.map_err(|_| CallError::Unavailable)?;
442        let call = frames.next().expect("two reserved frame slots");
443        let cancel = frames.next().expect("two reserved frame slots");
444        let (waiter, response) = oneshot::channel();
445        let id = self.pending.register(waiter)?;
446        let mut pending = PendingCall {
447            pending: &self.pending,
448            id,
449            _permit: permit,
450            cancel: Some(cancel),
451        };
452
453        if let Err(source) = prepare(id, call) {
454            // No Call Was Sent, So Release The Reserved Cancel Without Sending It.
455            pending.cancel.take();
456            return Err(source);
457        }
458        response.await.map_err(|_| CallError::Unavailable)?.map_err(E::from)
459    }
460
461    pub(crate) fn receive_answer(&self, id: u64, result: crate::FramePayload) {
462        self.pending.answer(id, Ok(result));
463    }
464
465    pub(crate) fn receive_refusal(&self, id: u64, refusal: harmos_domain::Refusal) {
466        let code = refusal.code();
467        let failure = refusal.into_failure();
468        self.pending.answer(id, Err(CallError::Refused { code, failure }));
469    }
470
471    pub(crate) fn disconnected(&self) {
472        self.pending.close();
473    }
474
475    // The kit clock that stamps rows when they are produced.
476    pub(super) fn clock(&self) -> fn() -> Option<i64> {
477        self.clock
478    }
479
480    pub(super) fn now(&self) -> Option<i64> {
481        (self.clock)()
482    }
483
484    // Publish rows in order through as few `host.publish` calls as fit the
485    // frame and batch limits; each call awaits its answer before the next.
486    // `stamps[i]` is the production time of `rows[i]`, each already encoded.
487    pub(super) async fn publication<S: crate::StreamSpec>(
488        &self, rows: &[Vec<u8>], stamps: &[Option<i64>], context: Option<&str>,
489    ) -> Result<(), super::PublishError> {
490        debug_assert_eq!(rows.len(), stamps.len());
491        if rows.is_empty() {
492            return Ok(());
493        }
494        if self.operation_context.is_none() && self.generation_context.is_none() {
495            return Err(CallError::Unavailable.into());
496        }
497        let target = self.assignments.resolve::<harmos_operations::streams::Publish>().map_err(|_| CallError::Unassigned {
498            operation: harmos_operations::streams::Publish::ID,
499        })?;
500        let mut rest = (rows, stamps);
501        while !rest.0.is_empty() {
502            let mut sent = 0;
503            let bytes = self
504                .exchange_with(|id, call| {
505                    let values = self.call.portable();
506                    let envelope = harmos_sdk::operations::WireChildCall::payload_encoded_len(
507                        &target.operation,
508                        target.generation,
509                        0,
510                        self.generation_context.is_some(),
511                        &values,
512                    )
513                    .map_err(|error| CallError::Malformed {
514                        method: harmos_operations::streams::Publish::ID,
515                        message: error.to_string(),
516                    })?;
517                    sent = self.with_publication::<S, super::PublishError>(rest.0, rest.1, context, envelope, |request| {
518                        let params = self.child_with_values::<harmos_operations::streams::Publish>(&request, false, values)?;
519                        let frame = Self::request_frame(id, CHILD_METHOD, params)?;
520                        call.send(Outgoing::Frame(frame));
521                        Ok(())
522                    })?;
523                    Ok::<(), super::PublishError>(())
524                })
525                .await?;
526            <() as Message>::decode(bytes.as_slice()).map_err(|source| CallError::Malformed {
527                method: harmos_operations::streams::Publish::ID,
528                message: source.to_string(),
529            })?;
530            rest = (&rest.0[sent..], &rest.1[sent..]);
531        }
532        Ok(())
533    }
534
535    // Take the longest leading run of rows that fits one call, reserve its
536    // contiguous sequences and send it under the sequence lock, so frames
537    // leave in sequence order. A row too large on its own still spends its
538    // sequence, exactly as a single-row publication does. The head carries
539    // its stamp; each following row carries its delta from its predecessor,
540    // and a stamp no delta can express starts the next call. A call carries
541    // at most one window of rows, so the host never evicts its own rows.
542    pub(super) fn with_publication<S: crate::StreamSpec, E: From<PublicationError>>(
543        &self, rows: &[Vec<u8>], stamps: &[Option<i64>], context: Option<&str>, empty_envelope: usize,
544        send: impl FnOnce(harmos_operations::streams::Publish) -> Result<(), E>,
545    ) -> Result<usize, E> {
546        let fits = |body: usize| Frame::child_encoded_from_empty(empty_envelope, body) <= Frame::LIMIT;
547        let mut sequences = self.sequences.lock().expect("the publication sequence lock is healthy");
548        let next = sequences.entry(S::ID).or_insert(0);
549        let first = next.checked_add(1).ok_or(PublicationError::SequenceExhausted { stream: S::ID })?;
550        *next = first;
551        let published_unix_micros = stamps[0];
552        let head = crate::wire::publication_encoded_len(S::ID, S::VERSION, first, rows[0].len(), published_unix_micros, context);
553        let encoded = |count: usize, following: usize| head.saturating_add(crate::wire::varint_len(count as u64)).saturating_add(following);
554        if !fits(encoded(0, 0)) {
555            return Err(PublicationError::Oversized {
556                encoded: encoded(0, 0),
557                limit: Frame::LIMIT,
558            }
559            .into());
560        }
561        let mut following = 0_usize;
562        let mut deltas = Vec::with_capacity(rows.len() - 1);
563        for (row, pair) in rows[1..].iter().zip(stamps.windows(2)) {
564            if deltas.len() + 1 >= S::WINDOW {
565                break;
566            }
567            let delta = match (pair[0], pair[1]) {
568                (_, None) => None,
569                (Some(previous), Some(stamp)) => match stamp.checked_sub(previous) {
570                    Some(delta) => Some(delta),
571                    None => break,
572                },
573                (None, Some(_)) => break,
574            };
575            let grown = following.saturating_add(crate::wire::following_encoded_len(row.len(), delta));
576            let total = encoded(deltas.len() + 1, grown);
577            if total > PUBLICATION_BATCH_BYTES || !fits(total) || *next == u64::MAX {
578                break;
579            }
580            following = grown;
581            deltas.push(delta);
582            *next += 1;
583        }
584        let taken = deltas.len() + 1;
585        send(harmos_operations::streams::Publish {
586            stream: S::ID.to_owned(),
587            schema: S::VERSION,
588            sequence: Some(first),
589            published_unix_micros,
590            context: context.map(str::to_owned),
591            payload: rows[0].clone(),
592            following: rows[1..taken]
593                .iter()
594                .zip(deltas)
595                .map(|(row, published_delta_micros)| harmos_operations::streams::FollowingRow {
596                    payload: row.clone(),
597                    published_delta_micros,
598                })
599                .collect(),
600        })?;
601        Ok(taken)
602    }
603
604    pub fn exiting(&self, reason: impl Into<String>) -> bool {
605        self.outgoing.try_send(Outgoing::Exiting(reason.into())).is_ok()
606    }
607
608    pub fn abandon(&self) -> bool {
609        self.outgoing.try_send(Outgoing::Abandon).is_ok()
610    }
611}
612
613/* === Public Functions === */
614
615/* === Private Functions === */
616
617fn refused_line(code: i64, failure: &harmos_domain::Failure) -> String {
618    match failure.context().operation() {
619        Some(operation) => format!("operation `{operation}` failed"),
620        None => format!("the host refused the call ({code})"),
621    }
622}
623
624pub(super) fn published_at() -> Option<i64> {
625    match std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH) {
626        Ok(elapsed) => i64::try_from(elapsed.as_micros()).ok(),
627        Err(error) => i64::try_from(error.duration().as_micros()).ok().and_then(i64::checked_neg),
628    }
629}
630
631/* === Tests === */
632
633#[cfg(test)]
634mod tests {
635    use super::*;
636    use harmos_domain::model::storage::files::{ByteRange, FileBatch, FileSource};
637    use harmos_domain::operations::CallValue;
638    use harmos_operations::{storage::ReadFile, streams::Publish};
639    use harmos_sdk::operations::WireChildCall;
640
641    struct Samples;
642
643    impl crate::StreamSpec for Samples {
644        type Row = FileBatch;
645        const ID: &'static str = "context.samples";
646        const VERSION: u32 = 1;
647        const WINDOW: usize = 8;
648    }
649
650    fn host(values: Vec<CallValue>) -> Supervisor {
651        let (outgoing, _) = mpsc::channel(1);
652        let host = Supervisor::new(outgoing, || None).for_attempt([7; 32], values).unwrap();
653        let assignments = [ReadFile::definition(), Publish::definition()]
654            .into_iter()
655            .map(|definition| harmos_sdk::operations::AssignedOperation {
656                operation: format!("host.{}", definition.id),
657                fingerprint: definition.fingerprint.0,
658                generation: u64::MAX,
659            })
660            .collect::<Vec<_>>();
661        host.assignments.install(&harmos_sdk::operations::AssignedOperation::encode(&assignments).unwrap()).unwrap();
662        host
663    }
664
665    #[test]
666    fn child_context_admitted() {
667        let values = vec![CallValue {
668            key: "context".into(),
669            payload: vec![3; 65536 - 7],
670        }];
671        let host = host(values.clone());
672        let request = ReadFile {
673            source: FileSource::Local("data.bin".into()),
674            range: ByteRange { offset: 0, limit: None },
675        };
676        let encoded = host.child::<ReadFile>(&request, false).unwrap();
677        let decoded = WireChildCall::decode(&encoded).unwrap();
678        assert_eq!(decoded.values, values);
679        assert_eq!(decoded.payload, request.encode());
680
681        let request = ReadFile {
682            source: FileSource::Local("a".repeat(Frame::LIMIT - 1024).into()),
683            range: request.range,
684        };
685        assert!(Frame::request_fits::<ReadFile>(&request));
686        assert!(matches!(host.child::<ReadFile>(&request, false), Err(CallError::Oversized { .. })));
687    }
688
689    #[test]
690    fn publication_context_boundary() {
691        let values = vec![CallValue {
692            key: "context".into(),
693            payload: vec![3; 65536 - 7],
694        }];
695        let host = host(values.clone());
696        let empty = WireChildCall::payload_encoded_len(&format!("host.{}", Publish::ID), u64::MAX, 0, false, &values).unwrap();
697        let fits = |bytes: usize| {
698            let row = FileBatch { offset: 0, bytes: vec![0; bytes] };
699            let body = crate::wire::publication_encoded_len("context.samples", 1, 1, row.encoded_size(), None, None) + 1;
700            Frame::child_encoded_from_empty(empty, body) <= Frame::LIMIT
701        };
702        let mut low = 0;
703        let mut high = Frame::LIMIT;
704        while low + 1 < high {
705            let middle = low + (high - low) / 2;
706            if fits(middle) {
707                low = middle;
708            } else {
709                high = middle;
710            }
711        }
712        let rows = [FileBatch { offset: 0, bytes: vec![0; low] }.encode()];
713        let taken = host
714            .with_publication::<Samples, CallError>(&rows, &[None], None, empty, |request| {
715                let envelope = host.child_with_values::<Publish>(&request, true, values.clone())?;
716                let frame = Frame::Call {
717                    id: u64::MAX,
718                    method: CHILD_METHOD.into(),
719                    params: envelope.into(),
720                };
721                assert_eq!(frame.encoded_len(), Frame::LIMIT);
722                Ok(())
723            })
724            .unwrap();
725        assert_eq!(taken, 1);
726        let rows = [FileBatch { offset: 0, bytes: vec![0; low + 1] }.encode()];
727        let result =
728            host.with_publication::<Samples, CallError>(&rows, &[None], None, empty, |_| panic!("an oversized publication must not send"));
729        assert!(matches!(result, Err(CallError::Publication(PublicationError::Oversized { .. }))));
730        assert_eq!(host.sequences.lock().unwrap()["context.samples"], 2);
731    }
732}
harmos_sidecar/kit/host.rs

harmos_sidecar/kit/
host.rs
