boxology_runtime/
transport.rs

1//! Transport binding lifecycle and composition-owned runtime carriers.
2use boxology_contract::{
3    CallContext, CapabilityDescriptor, Detail, ErasedCallError, ErasedTarget, ExposureLevel,
4    SlotValue, call_guarded,
5};
6use std::{future::Future, pin::Pin, sync::Arc};
7use tokio_util::sync::CancellationToken;
8/// The composition-owned task tracker shared with every transport binding.
9pub type TransportTaskTracker = tokio_util::task::TaskTracker;
10/// A payload-safe completion future for every task owned by one transport.
11pub type TransportJoinFuture = Pin<Box<dyn Future<Output = Result<(), Detail>> + Send + 'static>>;
12/// One capability exposed through a transport binding.
13#[derive(Clone)]
14pub struct TransportExposure {
15    descriptor: &'static CapabilityDescriptor,
16    level: ExposureLevel,
17    target: Arc<dyn ErasedTarget>,
18}
19#[allow(dead_code)]
20impl TransportExposure {
21    pub(crate) fn new(
22        descriptor: &'static CapabilityDescriptor,
23        level: ExposureLevel,
24        target: Arc<dyn ErasedTarget>,
25    ) -> Self {
26        Self {
27            descriptor,
28            level,
29            target,
30        }
31    }
32    /// Returns the exact descriptor selected for this exposure.
33    pub fn descriptor(&self) -> &'static CapabilityDescriptor {
34        self.descriptor
35    }
36    /// Returns the boundary level selected for this exposure.
37    pub fn level(&self) -> ExposureLevel {
38        self.level
39    }
40    /// Delegates to the retained target through guarded dispatch without adding policy.
41    pub fn dispatch<'a>(
42        &'a self,
43        context: CallContext,
44        input: SlotValue,
45    ) -> Pin<Box<dyn Future<Output = Result<SlotValue, ErasedCallError>> + Send + 'a>> {
46        call_guarded(self.target.as_ref(), self.descriptor.id(), context, input)
47    }
48}
49/// Per-binding state whose activation gate commits startup, not call cancellation.
50pub struct TransportRuntime<C>
51where
52    C: Send + Sync + 'static,
53{
54    exposures: Arc<[TransportExposure]>,
55    tracker: TransportTaskTracker,
56    config: Arc<C>,
57    activation: CancellationToken,
58}
59impl<C: Send + Sync + 'static> Clone for TransportRuntime<C> {
60    fn clone(&self) -> Self {
61        Self {
62            exposures: self.exposures.clone(),
63            tracker: self.tracker.clone(),
64            config: self.config.clone(),
65            activation: self.activation.clone(),
66        }
67    }
68}
69#[allow(dead_code)]
70impl<C: Send + Sync + 'static> TransportRuntime<C> {
71    pub(crate) fn new(
72        exposures: Arc<[TransportExposure]>,
73        tracker: TransportTaskTracker,
74        config: Arc<C>,
75        activation: CancellationToken,
76    ) -> Self {
77        Self {
78            exposures,
79            tracker,
80            config,
81            activation,
82        }
83    }
84    /// Returns exposures in builder-call order.
85    pub fn exposures(&self) -> &[TransportExposure] {
86        &self.exposures
87    }
88    /// Returns the composition-owned tracker shared by all bindings.
89    pub fn tracker(&self) -> &TransportTaskTracker {
90        &self.tracker
91    }
92    /// Returns this binding's concrete retained configuration.
93    pub fn config(&self) -> &C {
94        &self.config
95    }
96    /// Returns whether startup has committed and traffic may be admitted.
97    pub fn is_active(&self) -> bool {
98        self.activation.is_cancelled()
99    }
100    /// Waits until startup commits and traffic may be admitted.
101    pub async fn wait_until_active(&self) {
102        self.activation.cancelled().await;
103    }
104    pub(crate) fn activate(&self) {
105        self.activation.cancel();
106    }
107}
108/// A configured transport lifecycle participating in composition startup.
109/// Conformance is repeatable and non-authorizing. Preparation only preflights
110/// the complete descriptor set. Startup stays closed until activation and an
111/// error leaves no live intake, task, resource, or handle requiring cleanup.
112pub trait TransportBinding: Send + Sync + 'static {
113    /// Concrete configuration retained for this binding.
114    type Config: Send + Sync + 'static;
115    /// Live handle returned after transactional startup.
116    type Handle: TransportHandle;
117    /// Returns the configuration shared with the binding's runtime carrier.
118    fn config(&self) -> Arc<Self::Config>;
119    /// Checks one requested exposure without authorizing traffic.
120    fn conform(
121        &self,
122        descriptor: &CapabilityDescriptor,
123        level: ExposureLevel,
124    ) -> Result<(), Detail>;
125    /// Transactionally preflights the complete ordered descriptor set.
126    fn prepare(&self, descriptors: &[&'static CapabilityDescriptor]) -> Result<(), Detail>;
127    /// Starts closed intake and returns its synchronous lifecycle handle.
128    fn start(&self, runtime: TransportRuntime<Self::Config>) -> Result<Self::Handle, Detail>;
129}
130/// Synchronous lifecycle controls; dropping a handle has no defined behavior.
131pub trait TransportHandle: Send + Sync + 'static {
132    /// Prevents admission of new transport requests.
133    fn stop_intake(&self);
134    /// Requests cooperative cancellation of transport tasks.
135    fn cancel_tasks(&self);
136    /// Aborts transport tasks that remain live.
137    fn abort_tasks(&self);
138    /// Consumes the handle and joins every transport-owned task.
139    fn join_tasks(self: Box<Self>) -> TransportJoinFuture;
140}
141#[cfg(test)]
142mod tests {
143    use super::{TransportTaskTracker as Tracker, *};
144    use boxology_contract::{
145        BoxId, Caller, CancelToken, CapabilityId, CapabilityName, CapabilityShape, ContractValue,
146        Idempotency, TraceContext, TypeDescriptor,
147    };
148    use std::future::{Future, ready};
149    use std::sync::Mutex;
150    use std::task::{Context, Poll, Waker};
151    fn descriptor() -> &'static CapabilityDescriptor {
152        Box::leak(Box::new(CapabilityDescriptor::new(
153            CapabilityId::new(
154                BoxId::new("transport-test").unwrap(),
155                CapabilityName::new("call").unwrap(),
156            ),
157            TypeDescriptor::bool(),
158            TypeDescriptor::bool(),
159            TypeDescriptor::bool(),
160            CapabilityShape::Unary,
161            ExposureLevel::External,
162            Idempotency::None,
163            None,
164        )))
165    }
166    fn context() -> CallContext {
167        CallContext::new(
168            Caller::Anonymous,
169            None,
170            CancelToken::new(),
171            TraceContext::empty(),
172            None,
173        )
174    }
175    fn poll_once<F: Future + ?Sized>(future: Pin<&mut F>) -> Poll<F::Output> {
176        future.poll(&mut Context::from_waker(Waker::noop()))
177    }
178    #[derive(Default)]
179    struct Target(Mutex<Vec<CapabilityId>>);
180    impl ErasedTarget for Target {
181        fn call<'a>(
182            &'a self,
183            capability: &'a CapabilityId,
184            _context: CallContext,
185            input: SlotValue,
186        ) -> Pin<Box<dyn Future<Output = Result<SlotValue, ErasedCallError>> + Send + 'a>> {
187            self.0.lock().unwrap().push(capability.clone());
188            let result = match input {
189                SlotValue::Null => Ok(SlotValue::Null),
190                SlotValue::Missing => Err(ErasedCallError::Unavailable(Detail::new("test_call"))),
191                SlotValue::Value(_) => panic!("guarded transport panic"),
192            };
193            Box::pin(ready(result))
194        }
195    }
196    struct Config(u8);
197    struct Handle;
198    struct Binding;
199    impl TransportHandle for Handle {
200        fn stop_intake(&self) {}
201        fn cancel_tasks(&self) {}
202        fn abort_tasks(&self) {}
203        fn join_tasks(self: Box<Self>) -> TransportJoinFuture {
204            Box::pin(ready(Ok(())))
205        }
206    }
207    impl TransportBinding for Binding {
208        type Config = Config;
209        type Handle = Handle;
210        fn config(&self) -> Arc<Config> {
211            Arc::new(Config(0))
212        }
213        fn conform(
214            &self,
215            _descriptor: &CapabilityDescriptor,
216            _level: ExposureLevel,
217        ) -> Result<(), Detail> {
218            Ok(())
219        }
220        fn prepare(&self, _: &[&'static CapabilityDescriptor]) -> Result<(), Detail> {
221            Ok(())
222        }
223        fn start(&self, _: TransportRuntime<Config>) -> Result<Handle, Detail> {
224            Ok(Handle)
225        }
226    }
227    #[test]
228    fn transport_carriers_preserve_dispatch_sharing_and_activation_contracts() {
229        fn bounds<T: Send + Sync + 'static>() {}
230        bounds::<TransportExposure>();
231        bounds::<TransportRuntime<Config>>();
232        bounds::<Tracker>();
233        let descriptor = descriptor();
234        let target = Arc::new(Target::default());
235        let exposure = TransportExposure::new(descriptor, ExposureLevel::Internal, target.clone());
236        assert!(std::ptr::eq(exposure.descriptor(), descriptor));
237        assert_eq!(exposure.level(), ExposureLevel::Internal);
238        let mut success = exposure.dispatch(context(), SlotValue::Null);
239        let result = poll_once(success.as_mut());
240        assert_eq!(result, Poll::Ready(Ok(SlotValue::Null)));
241        let mut failure = exposure.dispatch(context(), SlotValue::Missing);
242        let result = poll_once(failure.as_mut());
243        let expected = ErasedCallError::Unavailable(Detail::new("test_call"));
244        assert_eq!(result, Poll::Ready(Err(expected)));
245        let mut panic = exposure.dispatch(context(), SlotValue::Value(ContractValue::bool(true)));
246        let Poll::Ready(Err(ErasedCallError::Internal(detail))) = poll_once(panic.as_mut()) else {
247            panic!("transport panic escaped guarded dispatch")
248        };
249        assert_eq!(detail.code(), "panic");
250        assert_eq!(*target.0.lock().unwrap(), vec![descriptor.id().clone(); 3]);
251        let runtime = TransportRuntime::new(
252            Arc::from([exposure.clone()]),
253            Tracker::new(),
254            Binding.config(),
255            CancellationToken::new(),
256        );
257        let clone = runtime.clone();
258        let same_tracker = Tracker::ptr_eq(&runtime.tracker, &clone.tracker);
259        assert!(same_tracker);
260        assert!(Arc::ptr_eq(&runtime.config, &clone.config));
261        assert_eq!(runtime.config().0, 0);
262        assert!(Arc::ptr_eq(&runtime.exposures, &clone.exposures));
263        let token = clone.tracker().token();
264        assert_eq!(runtime.tracker().len(), 1);
265        drop(token);
266        assert!(runtime.tracker().is_empty());
267        let mut active = Box::pin(runtime.wait_until_active());
268        assert!(!runtime.is_active() && matches!(poll_once(active.as_mut()), Poll::Pending));
269        clone.activate();
270        assert!(runtime.is_active() && matches!(poll_once(active.as_mut()), Poll::Ready(())));
271    }
272}
boxology_runtime/transport.rs

boxology_runtime/
transport.rs
