khive_runtime/events_split/
protocol.rs

1use super::*;
2
3/// Request frame sent from a domain process to the events daemon.
4#[derive(Debug, Serialize, Deserialize)]
5#[serde(tag = "op", rename_all = "snake_case")]
6pub enum EventsRequest {
7    /// Fire-and-forget append lane. The daemon replies with a summary, but the
8    /// forwarder treats a failed reply as a counted drop, never an error to
9    /// the original caller.
10    AppendEvents {
11        protocol_version: u32,
12        namespace: String,
13        events: Vec<Event>,
14    },
15    /// ADR-133 audit-batch lane: real dispositions come back.
16    AppendEventsIdempotent {
17        protocol_version: u32,
18        namespace: String,
19        events: Vec<Event>,
20    },
21    GetEvent {
22        protocol_version: u32,
23        namespace: String,
24        id: Uuid,
25    },
26    QueryEvents {
27        protocol_version: u32,
28        namespace: String,
29        filter: EventFilter,
30        page: PageRequest,
31    },
32    QueryEventPage {
33        protocol_version: u32,
34        namespace: String,
35        query: EventPageQuery,
36    },
37    CountEvents {
38        protocol_version: u32,
39        namespace: String,
40        #[serde(default)]
41        operator_read_budget_secs: Option<u64>,
42        filter: EventFilter,
43    },
44    CountEventsGrouped {
45        protocol_version: u32,
46        namespace: String,
47        #[serde(default)]
48        operator_read_budget_secs: Option<u64>,
49        filter: EventFilter,
50        group_by: EventGroupBy,
51    },
52}
53
54impl EventsRequest {
55    pub(super) fn protocol_version(&self) -> u32 {
56        match self {
57            Self::AppendEvents {
58                protocol_version, ..
59            }
60            | Self::AppendEventsIdempotent {
61                protocol_version, ..
62            }
63            | Self::GetEvent {
64                protocol_version, ..
65            }
66            | Self::QueryEvents {
67                protocol_version, ..
68            }
69            | Self::QueryEventPage {
70                protocol_version, ..
71            }
72            | Self::CountEvents {
73                protocol_version, ..
74            }
75            | Self::CountEventsGrouped {
76                protocol_version, ..
77            } => *protocol_version,
78        }
79    }
80
81    pub(super) fn count_budget_secs(&self) -> Option<Option<u64>> {
82        match self {
83            Self::CountEvents {
84                operator_read_budget_secs,
85                ..
86            }
87            | Self::CountEventsGrouped {
88                operator_read_budget_secs,
89                ..
90            } => Some(*operator_read_budget_secs),
91            _ => None,
92        }
93    }
94
95    pub(super) fn namespace(&self) -> &str {
96        match self {
97            Self::AppendEvents { namespace, .. }
98            | Self::AppendEventsIdempotent { namespace, .. }
99            | Self::GetEvent { namespace, .. }
100            | Self::QueryEvents { namespace, .. }
101            | Self::QueryEventPage { namespace, .. }
102            | Self::CountEvents { namespace, .. }
103            | Self::CountEventsGrouped { namespace, .. } => namespace,
104        }
105    }
106}
107
108/// Response frame from the events daemon.
109#[derive(Debug, Serialize, Deserialize)]
110#[serde(tag = "kind", rename_all = "snake_case")]
111pub enum EventsResponse {
112    Appended {
113        summary: BatchWriteSummary,
114    },
115    Idempotent {
116        result: IdempotentEventBatchResult,
117    },
118    Event {
119        event: Option<Event>,
120    },
121    Pageful {
122        page: Page<Event>,
123    },
124    EventPageWindow {
125        window: EventPageWindow,
126    },
127    Count {
128        count: u64,
129    },
130    GroupedCount {
131        counts: BTreeMap<String, u64>,
132    },
133    /// Typed refusal. `retryable` distinguishes transient daemon-side
134    /// conditions from contract errors (bad frame, version skew).
135    /// `writer_task_failure` carries both request finality and whether the
136    /// remote writer seam terminated. The retryable bit cannot express that
137    /// distinction: an ordinary operation or COMMIT error may be transient or
138    /// permanent while its transaction is independently proven rolled back.
139    /// `None` means the error is not from a writer request. Cross-version
140    /// frames never reach this mapping because `dispatch_events_request`
141    /// refuses a mismatched version before any write executes.
142    Error {
143        message: String,
144        retryable: bool,
145        #[serde(default)]
146        writer_task_failure: Option<WireWriterTaskFailure>,
147        /// Optional native write evidence; older frames retain their legacy mapping.
148        #[serde(default, skip_serializing_if = "Option::is_none")]
149        sqlite_write_failure: Option<khive_storage::error::SqliteWriteFailure>,
150        #[serde(default, skip_serializing_if = "Option::is_none")]
151        read_failure: Option<WireReadFailure>,
152    },
153}
154
155/// Whether a writer-state error belongs to one failed request on a still-live
156/// seam or to a permanently retired writer task.
157#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
158#[serde(tag = "kind", rename_all = "snake_case")]
159pub enum WireWriterTaskFailure {
160    RequestFailed {
161        request_state: WireWriterTaskState,
162    },
163    TaskTerminated {
164        request_state: WireWriterTaskState,
165        #[serde(default, skip_serializing_if = "Option::is_none")]
166        sqlite_full_codes: Option<(i32, i32)>,
167    },
168}
169
170/// Wire mirror of [`khive_storage::WriterTaskRequestState`]. A separate type
171/// because the storage enum is not serializable and the wire shape must stay
172/// under this module's protocol-version control, not the storage crate's.
173#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
174#[serde(rename_all = "snake_case")]
175pub enum WireWriterTaskState {
176    NotStarted,
177    TransactionRolledBack,
178    SideEffectsUnknown,
179}
180
181impl From<khive_storage::WriterTaskRequestState> for WireWriterTaskState {
182    fn from(state: khive_storage::WriterTaskRequestState) -> Self {
183        use khive_storage::WriterTaskRequestState as S;
184        match state {
185            S::NotStarted => Self::NotStarted,
186            S::TransactionRolledBack => Self::TransactionRolledBack,
187            S::SideEffectsUnknown => Self::SideEffectsUnknown,
188        }
189    }
190}
191
192impl From<WireWriterTaskState> for khive_storage::WriterTaskRequestState {
193    fn from(state: WireWriterTaskState) -> Self {
194        use khive_storage::WriterTaskRequestState as S;
195        match state {
196            WireWriterTaskState::NotStarted => S::NotStarted,
197            WireWriterTaskState::TransactionRolledBack => S::TransactionRolledBack,
198            WireWriterTaskState::SideEffectsUnknown => S::SideEffectsUnknown,
199        }
200    }
201}
khive_runtime/events_split/protocol.rs

khive_runtime/events_split/
protocol.rs
