prns_runtime/runtime/
event.rs

1//! The app-facing event lane, curated from the engine's `Journaled` stream, split so an app
2//! can silo its two concerns:
3//!
4//!   - [`Message`]: payload arrived *for the app* (delivered singles/links, requests to
5//!     answer, responses, resources). The data plane.
6//!   - [`Diagnostic`]: what the engine did (announces heard, settlements, link lifecycle,
7//!     route churn, failures). Observability, not payload.
8//!
9//! The mapping is total: every `Journaled` lands in exactly one bucket.
10
11use crate::engine::LinkClosedReason;
12use crate::engine::{CommandId, HeldDropCause, LinkEstablished, RouteRemovalCause, Settlement};
13use crate::engine::{InstantMillis, Journaled, PersistenceFlushCause, PersistenceFlushTarget};
14use crate::identity::IdentityHash;
15use crate::interfaces::InterfaceId;
16use crate::routing::delivery::Delivery;
17use crate::routing::links::channel::MessageType;
18use crate::routing::links::request::RequestId;
19use crate::routing::links::resources::{ResourceFailureCause, ResourceHash};
20use crate::routing::links::LinkId;
21use crate::routing::request_handlers::RequestPathHash;
22use crate::units::RttMillis;
23use crate::wire::DestinationHash;
24
25#[derive(Debug)]
26pub enum PrnsEvent<'a> {
27    Message(Message<'a>),
28    Diagnostic(Diagnostic),
29}
30
31/// The data plane: bytes the app owns.
32#[derive(Debug)]
33pub enum Message<'a> {
34    Delivered(Delivery<'a>),
35    Request {
36        destination: DestinationHash,
37        link_id: LinkId,
38        request_id: RequestId,
39        requester: Option<IdentityHash>,
40        path_hash: RequestPathHash,
41        requested_at: InstantMillis,
42        rtt: RttMillis,
43        data: &'a [u8],
44    },
45    Response {
46        link_id: LinkId,
47        request_id: RequestId,
48        data: &'a [u8],
49    },
50    /// One in-order segment of a split response; the request's settlement arrives as a [`Diagnostic::CommandSettled`] when the final segment assembles.
51    ResponseSegment {
52        link_id: LinkId,
53        request_id: RequestId,
54        segment_index: u64,
55        total_segments: u64,
56        data: &'a [u8],
57    },
58    Resource {
59        link_id: LinkId,
60        hash: ResourceHash,
61        /// The transfer's packed metadata, stripped from the stream head, opaque to the engine; `None` when none traveled.
62        metadata: Option<&'a [u8]>,
63        data: &'a [u8],
64    },
65    ResourceNeedsDecompression {
66        link_id: LinkId,
67        hash: ResourceHash,
68        stream: &'a [u8],
69        uncompressed_data_bytes: u64,
70    },
71    ResourceSegment {
72        link_id: LinkId,
73        original_hash: ResourceHash,
74        segment_index: u64,
75        total_segments: u64,
76        /// Rides segment one only, stripped from the stream head like the single-segment delivery.
77        metadata: Option<&'a [u8]>,
78        data: &'a [u8],
79    },
80    ChannelMessage {
81        link_id: LinkId,
82        message_type: MessageType,
83        data: &'a [u8],
84    },
85}
86
87/// The control/observability plane: what the engine did, not what arrived. Fully owned —
88/// no borrow into the inbound frame, so it can outlive the reaction if an app buffers it.
89#[derive(Debug)]
90pub enum Diagnostic {
91    /// An announce just minted a fresh self-ratchet: flush this destination's record to the
92    /// vault now — a secret peers may already encrypt toward must never exist only in memory.
93    SelfRatchetRotated {
94        destination: DestinationHash,
95    },
96    AnnounceHeard {
97        destination: DestinationHash,
98        hops: u8,
99        source_interface: InterfaceId,
100    },
101    /// The recipe's persistence store was seeded into this boot's engine before the first frame moved.
102    PersistenceRestored {
103        routes: u32,
104        destination_identities: u32,
105        tunnels: u32,
106        ratchets: u32,
107        refused: u32,
108        dropped: u32,
109    },
110    /// The persistence worker landed one independently stored part of a save.
111    PersistenceFlushed {
112        cause: PersistenceFlushCause,
113        target: PersistenceFlushTarget,
114    },
115    /// The persistence worker could not land one independently stored part of a save.
116    ///
117    /// Storage-specific error detail is written to the host log; this owned diagnostic
118    /// preserves the stable policy-relevant facts for applications.
119    PersistenceFlushFailed {
120        cause: PersistenceFlushCause,
121        target: PersistenceFlushTarget,
122    },
123    AnnounceHeldDropped {
124        destination: DestinationHash,
125        source_interface: InterfaceId,
126        cause: HeldDropCause,
127    },
128    CommandSettled {
129        id: CommandId,
130        settlement: Settlement,
131    },
132    LinkEstablished(LinkEstablished),
133    PeerIdentified {
134        link_id: LinkId,
135        identity: IdentityHash,
136    },
137    LinkClosed {
138        link_id: LinkId,
139        reason: LinkClosedReason,
140    },
141    /// A packet for this active link arrived on `arrived_on`, not the `attached_interface` the link
142    /// runs over — dropped unprocessed (RNS 1.4.2 `Link.receive`), surfaced as a possible attempt to
143    /// inject into the link from a foreign interface.
144    LinkInterfaceMismatch {
145        link_id: LinkId,
146        attached_interface: InterfaceId,
147        arrived_on: InterfaceId,
148    },
149    ResourceFailed {
150        link_id: LinkId,
151        hash: ResourceHash,
152        cause: ResourceFailureCause,
153    },
154    ResourceAssembled {
155        link_id: LinkId,
156        original_hash: ResourceHash,
157        total_size_bytes: u64,
158    },
159    RouteRemoved {
160        destination: DestinationHash,
161        cause: RouteRemovalCause,
162    },
163}
164
165impl<'a> From<Journaled<'a>> for PrnsEvent<'a> {
166    fn from(journaled: Journaled<'a>) -> Self {
167        match journaled {
168            Journaled::Delivered(delivery) => PrnsEvent::Message(Message::Delivered(delivery)),
169            Journaled::RequestReceived {
170                destination,
171                link_id,
172                request_id,
173                requester,
174                path_hash,
175                requested_at,
176                rtt,
177                data,
178            } => PrnsEvent::Message(Message::Request {
179                destination,
180                link_id,
181                request_id,
182                requester,
183                path_hash,
184                requested_at,
185                rtt,
186                data,
187            }),
188            Journaled::ResponseReceived {
189                link_id,
190                request_id,
191                data,
192                ..
193            } => PrnsEvent::Message(Message::Response {
194                link_id,
195                request_id,
196                data,
197            }),
198            Journaled::ResponseSegmentReceived {
199                link_id,
200                request_id,
201                segment_index,
202                total_segments,
203                data,
204                ..
205            } => PrnsEvent::Message(Message::ResponseSegment {
206                link_id,
207                request_id,
208                segment_index,
209                total_segments,
210                data,
211            }),
212            Journaled::ResourceReceived {
213                link_id,
214                hash,
215                metadata,
216                data,
217            } => PrnsEvent::Message(Message::Resource {
218                link_id,
219                hash,
220                metadata,
221                data,
222            }),
223            Journaled::ResourceNeedsDecompression {
224                link_id,
225                hash,
226                stream,
227                uncompressed_data_bytes,
228            } => PrnsEvent::Message(Message::ResourceNeedsDecompression {
229                link_id,
230                hash,
231                stream,
232                uncompressed_data_bytes,
233            }),
234            Journaled::ResourceSegmentReceived {
235                link_id,
236                original_hash,
237                segment_index,
238                total_segments,
239                metadata,
240                data,
241            } => PrnsEvent::Message(Message::ResourceSegment {
242                link_id,
243                original_hash,
244                segment_index,
245                total_segments,
246                metadata,
247                data,
248            }),
249            Journaled::ChannelMessageReceived {
250                link_id,
251                message_type,
252                data,
253            } => PrnsEvent::Message(Message::ChannelMessage {
254                link_id,
255                message_type,
256                data,
257            }),
258            Journaled::AnnounceHeard { observation, .. } => {
259                PrnsEvent::Diagnostic(Diagnostic::AnnounceHeard {
260                    destination: observation.destination,
261                    hops: observation.hops.0,
262                    source_interface: observation.source_interface,
263                })
264            }
265            Journaled::SelfRatchetRotated { destination } => {
266                PrnsEvent::Diagnostic(Diagnostic::SelfRatchetRotated { destination })
267            }
268            Journaled::AnnounceHeldDropped {
269                destination,
270                source_interface,
271                cause,
272            } => PrnsEvent::Diagnostic(Diagnostic::AnnounceHeldDropped {
273                destination,
274                source_interface,
275                cause,
276            }),
277            Journaled::CommandSettled { id, settlement } => {
278                PrnsEvent::Diagnostic(Diagnostic::CommandSettled { id, settlement })
279            }
280            Journaled::PersistenceFlushed { cause, target } => {
281                PrnsEvent::Diagnostic(Diagnostic::PersistenceFlushed { cause, target })
282            }
283            Journaled::PersistenceFlushFailed { cause, target } => {
284                PrnsEvent::Diagnostic(Diagnostic::PersistenceFlushFailed { cause, target })
285            }
286            Journaled::LinkEstablished(established) => {
287                PrnsEvent::Diagnostic(Diagnostic::LinkEstablished(established))
288            }
289            Journaled::PeerIdentified { link_id, identity } => {
290                PrnsEvent::Diagnostic(Diagnostic::PeerIdentified { link_id, identity })
291            }
292            Journaled::LinkInterfaceMismatch {
293                link_id,
294                attached_interface,
295                arrived_on,
296            } => PrnsEvent::Diagnostic(Diagnostic::LinkInterfaceMismatch {
297                link_id,
298                attached_interface,
299                arrived_on,
300            }),
301            Journaled::LinkClosed { link_id, reason } => {
302                PrnsEvent::Diagnostic(Diagnostic::LinkClosed { link_id, reason })
303            }
304            Journaled::ResourceFailed {
305                link_id,
306                hash,
307                cause,
308            } => PrnsEvent::Diagnostic(Diagnostic::ResourceFailed {
309                link_id,
310                hash,
311                cause,
312            }),
313            Journaled::ResourceAssembled {
314                link_id,
315                original_hash,
316                total_size_bytes,
317            } => PrnsEvent::Diagnostic(Diagnostic::ResourceAssembled {
318                link_id,
319                original_hash,
320                total_size_bytes,
321            }),
322            Journaled::RouteRemoved { destination, cause } => {
323                PrnsEvent::Diagnostic(Diagnostic::RouteRemoved { destination, cause })
324            }
325        }
326    }
327}
prns_runtime/runtime/event.rs

prns_runtime/runtime/
event.rs
