frame_conv/handle/
events.rs

1//! The subscription pattern (`frame:conv-subscription@v1`, F-3a R3): typed
2//! events at the proven delivery strength, plus the caller-owned cursor
3//! commitment lever.
4
5use std::time::{Duration, Instant};
6
7use liminal_protocol::wire::{ClientRequest, ParticipantAck, ServerValue};
8use serde::Serialize;
9use serde::de::DeserializeOwned;
10
11use crate::envelope::Envelope;
12use crate::error::{CallError, CursorCommit, PublishError};
13use crate::id::{ConversationSeq, CorrelationId, MessageKind, PatternId};
14use crate::outcome::{PublishReceipt, SubscriptionItem};
15use crate::seam::classify_refusal;
16use crate::store::ResumeStore;
17
18use super::pump::{PumpStep, submit_to_publish};
19use super::state::{ConversationHandle, QueuedItem};
20
21impl<S: ResumeStore> ConversationHandle<S> {
22    /// Publishes a typed subscription event into the conversation.
23    /// Schema-invalid values are refused typed BEFORE the wire (F-3a R1).
24    /// The receipt is the correlated commit answer — delivery evidence for
25    /// the record's admission, NOT evidence any subscriber consumed it
26    /// (finding G1's discipline; the stream carries news, never
27    /// obligations).
28    ///
29    /// # Errors
30    ///
31    /// Returns typed publish failures.
32    pub fn publish_event<E: Serialize>(
33        &mut self,
34        body: &E,
35    ) -> Result<PublishReceipt, PublishError> {
36        let envelope = Envelope::from_typed(
37            PatternId::Subscription,
38            CorrelationId::mint(),
39            MessageKind::Event,
40            body,
41        )
42        .map_err(|refusal| PublishError::SchemaInvalid {
43            detail: refusal.detail,
44        })?;
45        let bytes = envelope
46            .encode()
47            .map_err(|refusal| PublishError::Protocol {
48                detail: refusal.detail,
49            })?;
50        self.admit(bytes)
51    }
52
53    /// Waits up to `wait` for the next subscription item. `Ok(None)` is a
54    /// benign quiet wait — never an error, never a protocol outcome (the
55    /// assertion-8 pin at this crate's surface). Items carry the delivery
56    /// guarantee documented on [`SubscriptionItem`]: ordered, contiguous,
57    /// sender-excluded within a live session; membership-forward at join;
58    /// typed gap surface with resume as the documented resync path.
59    ///
60    /// The effective wait granularity is the substrate's IO quantum
61    /// (hardcoded 5 s at published liminal 0.3.0 — upstream ASK-2): a wait
62    /// smaller than one quantum may still block for one quantum.
63    ///
64    /// # Errors
65    ///
66    /// Returns a typed connection fate.
67    pub fn next_event<E: DeserializeOwned>(
68        &mut self,
69        wait: Duration,
70    ) -> Result<Option<SubscriptionItem<E>>, CallError> {
71        let started = Instant::now();
72        loop {
73            if let Some(item) = self.events_inbox.pop_front() {
74                return Ok(Some(convert_item(item)));
75            }
76            if started.elapsed() >= wait {
77                return Ok(None);
78            }
79            match self.pump_step() {
80                Ok(PumpStep::Classified | PumpStep::Quiet) => {}
81                Err(detail) => return Err(CallError::ConnectionLost { detail }),
82            }
83        }
84    }
85
86    /// Commits the caller-owned consumption cursor through `through`,
87    /// cumulatively. Acking is explicit and caller-owned — this crate never
88    /// acks on the caller's behalf, because the committed cursor is the
89    /// caller's lever on resume replay volume (F-0c §R4).
90    ///
91    /// # Errors
92    ///
93    /// Returns typed publish-path failures; the ack answers themselves
94    /// (committed / no-op / gap / regression) are the typed
95    /// [`CursorCommit`] outcome, not errors.
96    pub fn commit_cursor(
97        &mut self,
98        through: ConversationSeq,
99    ) -> Result<CursorCommit, PublishError> {
100        let answer = self
101            .submit(ClientRequest::ParticipantAck(ParticipantAck {
102                conversation_id: self.conversation.to_wire(),
103                participant_id: self.me.to_wire(),
104                capability_generation: self.generation,
105                through_seq: through.value(),
106            }))
107            .map_err(submit_to_publish)?;
108        match answer {
109            ServerValue::AckCommitted(committed) => Ok(CursorCommit::Committed {
110                through: ConversationSeq::new(committed.current_cursor()),
111            }),
112            ServerValue::AckNoOp(noop) => Ok(CursorCommit::NoOp {
113                current: ConversationSeq::new(noop.current_cursor()),
114            }),
115            ServerValue::AckGap(gap) => Ok(CursorCommit::Gap {
116                requested: ConversationSeq::new(gap.request().through_seq),
117                current: ConversationSeq::new(gap.current_cursor()),
118            }),
119            ServerValue::AckRegression(regression) => Ok(CursorCommit::Regression {
120                requested: ConversationSeq::new(regression.request().through_seq),
121            }),
122            other => {
123                let (class, detail) = classify_refusal(&other);
124                Err(PublishError::Refused { class, detail })
125            }
126        }
127    }
128}
129
130/// Converts a queued item into the typed subscription surface, decoding
131/// event payloads against the caller's typed message (schema-invalid
132/// surfaces typed — F-3a R1).
133fn convert_item<E: DeserializeOwned>(item: QueuedItem) -> SubscriptionItem<E> {
134    match item {
135        QueuedItem::Event {
136            envelope,
137            publisher,
138            seq,
139        } => match envelope.decode_payload::<E>() {
140            Ok(body) => SubscriptionItem::Event {
141                body,
142                publisher,
143                seq,
144            },
145            Err(refusal) => SubscriptionItem::SchemaInvalid {
146                publisher,
147                seq,
148                detail: refusal.detail,
149            },
150        },
151        QueuedItem::Joined { peer, seq } => SubscriptionItem::PeerJoined { peer, seq },
152        QueuedItem::Departed { peer, seq, reason } => {
153            SubscriptionItem::PeerDeparted { peer, seq, reason }
154        }
155        QueuedItem::Failed { peer, seq, failure } => {
156            SubscriptionItem::PeerFailed { peer, seq, failure }
157        }
158        QueuedItem::Compacted { seq } => SubscriptionItem::HistoryCompacted { seq },
159        QueuedItem::Gap { expected, observed } => SubscriptionItem::Gap { expected, observed },
160    }
161}
frame_conv/handle/events.rs

frame_conv/handle/
events.rs
