frame_conv/handle/request.rs
1//! The request-response pattern (`frame:conv-request@v1`, F-3a R2): typed
2//! request with a caller-supplied deadline, closed outcome set, and defined
3//! fates for late and duplicate replies.
4
5use std::time::{Duration, Instant};
6
7use serde::Serialize;
8use serde::de::DeserializeOwned;
9
10use crate::envelope::Envelope;
11use crate::error::{CallError, PublishError};
12use crate::id::{CorrelationId, MessageKind, PatternId};
13use crate::outcome::{InboundRequest, IncomingRequest, PublishReceipt, RequestOutcome};
14use crate::store::ResumeStore;
15
16use super::pump::PumpStep;
17use super::state::ConversationHandle;
18
19impl<S: ResumeStore> ConversationHandle<S> {
20 /// Sends a typed request and waits for exactly one outcome from the
21 /// closed set: the correlated schema-validated reply, the caller's
22 /// deadline elapsing, or a responder failure (F-3a R2).
23 ///
24 /// The deadline is caller-supplied per call — no default exists — and it
25 /// is the ONLY thing that ends the request: elapsed IO quanta while
26 /// waiting are benign re-arms and never change the outcome (constraint
27 /// 3; the receive-cancel discipline). Correlation is by typed
28 /// correlation identity, never delivery order (constraint 4). A reply
29 /// arriving after the deadline surfaces as a typed late-reply anomaly on
30 /// [`drain_anomalies`](Self::drain_anomalies); a second reply to the
31 /// same correlation surfaces as a duplicate-reply anomaly; the first
32 /// reply's delivery is unaffected (R6).
33 ///
34 /// # Errors
35 ///
36 /// Returns typed publish failures (schema-invalid refused before the
37 /// wire; substrate refusals; transport loss), a typed
38 /// schema-invalid-reply error, or a typed connection fate. A deadline
39 /// elapse is NOT an error — it is the [`RequestOutcome::DeadlineElapsed`]
40 /// outcome.
41 pub fn request<Q: Serialize, R: DeserializeOwned>(
42 &mut self,
43 body: &Q,
44 deadline: Duration,
45 ) -> Result<RequestOutcome<R>, CallError> {
46 let correlation = CorrelationId::mint();
47 let envelope = Envelope::from_typed(
48 PatternId::RequestResponse,
49 correlation,
50 MessageKind::Request,
51 body,
52 )
53 .map_err(|refusal| PublishError::SchemaInvalid {
54 detail: refusal.detail,
55 })?;
56 let bytes = envelope
57 .encode()
58 .map_err(|refusal| PublishError::Protocol {
59 detail: refusal.detail,
60 })?;
61 let receipt = self.admit(bytes)?;
62 self.exchanges.open(correlation);
63 let started = Instant::now();
64 loop {
65 if let Some(held) = self.replies.remove(&correlation) {
66 match held.envelope.decode_payload::<R>() {
67 Ok(reply) => {
68 return Ok(RequestOutcome::Replied {
69 reply,
70 responder: held.responder,
71 seq: held.seq,
72 });
73 }
74 Err(refusal) => {
75 return Err(CallError::ReplySchemaInvalid {
76 correlation,
77 detail: refusal.detail,
78 });
79 }
80 }
81 }
82 if let Some(note) = self
83 .deaths
84 .iter()
85 .find(|note| note.seq > receipt.seq)
86 .cloned()
87 {
88 // A peer died after this request was admitted. Pin: after a
89 // responder failure, a reply to this exchange classifies as
90 // late. (No live producer exists at liminal 0.3.0 — ASK-4.)
91 self.exchanges.close_elapsed(correlation);
92 return Ok(RequestOutcome::ResponderFailed {
93 peer: note.peer,
94 failure: note.failure,
95 seq: note.seq,
96 });
97 }
98 if started.elapsed() >= deadline {
99 self.exchanges.close_elapsed(correlation);
100 return Ok(RequestOutcome::DeadlineElapsed { deadline });
101 }
102 if let Err(detail) = self.pump_step() {
103 return Err(CallError::ConnectionLost { detail });
104 }
105 }
106 }
107
108 /// Waits up to `wait` for the next inbound request. `Ok(None)` is a
109 /// benign quiet wait — never an error, never a protocol outcome (the
110 /// assertion-8 pin at this crate's surface). A schema-invalid request
111 /// surfaces typed as [`InboundRequest::SchemaInvalid`], never dropped.
112 ///
113 /// The effective wait granularity is the substrate's IO quantum
114 /// (hardcoded 5 s at published liminal 0.3.0 — upstream ASK-2): a wait
115 /// smaller than one quantum may still block for one quantum.
116 ///
117 /// # Errors
118 ///
119 /// Returns a typed connection fate.
120 pub fn next_request<Q: DeserializeOwned>(
121 &mut self,
122 wait: Duration,
123 ) -> Result<Option<InboundRequest<Q>>, CallError> {
124 let started = Instant::now();
125 loop {
126 if let Some(held) = self.requests_inbox.pop_front() {
127 let correlation = held.envelope.correlation;
128 return Ok(Some(match held.envelope.decode_payload::<Q>() {
129 Ok(body) => InboundRequest::Valid(IncomingRequest {
130 body,
131 correlation,
132 requester: held.requester,
133 seq: held.seq,
134 }),
135 Err(refusal) => InboundRequest::SchemaInvalid {
136 correlation,
137 requester: held.requester,
138 seq: held.seq,
139 detail: refusal.detail,
140 },
141 }));
142 }
143 if started.elapsed() >= wait {
144 return Ok(None);
145 }
146 match self.pump_step() {
147 Ok(PumpStep::Classified | PumpStep::Quiet) => {}
148 Err(detail) => return Err(CallError::ConnectionLost { detail }),
149 }
150 }
151 }
152
153 /// Publishes the typed reply to a received request, correlated to the
154 /// request's exchange. Schema-invalid replies are refused typed BEFORE
155 /// the wire (F-3a R1).
156 ///
157 /// # Errors
158 ///
159 /// Returns typed publish failures.
160 pub fn reply<R: Serialize>(
161 &mut self,
162 correlation: CorrelationId,
163 body: &R,
164 ) -> Result<PublishReceipt, PublishError> {
165 let envelope = Envelope::from_typed(
166 PatternId::RequestResponse,
167 correlation,
168 MessageKind::Reply,
169 body,
170 )
171 .map_err(|refusal| PublishError::SchemaInvalid {
172 detail: refusal.detail,
173 })?;
174 let bytes = envelope
175 .encode()
176 .map_err(|refusal| PublishError::Protocol {
177 detail: refusal.detail,
178 })?;
179 self.admit(bytes)
180 }
181}