av_bridge/
bus.rs

1//! The `EventBus` trait — the Bridge's backend portability boundary.
2
3use serde::{Deserialize, Serialize};
4
5/// Bus errors.
6#[derive(Debug, thiserror::Error)]
7#[non_exhaustive]
8pub enum BusError {
9    /// Topic does not exist (provision first — publishing into the void is a
10    /// silent-error class, so it's an error, not an auto-create).
11    #[error("unknown topic {0:?} (provision it via the manifest first)")]
12    UnknownTopic(String),
13    /// I/O failure.
14    #[error("bus io: {0}")]
15    Io(#[from] std::io::Error),
16    /// Serialization failure.
17    #[error("bus serde: {0}")]
18    Serde(#[from] serde_json::Error),
19    /// Backend-specific failure (network brokers).
20    #[error("bus backend: {0}")]
21    Backend(String),
22}
23
24/// Acknowledgment for a published event.
25#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
26pub struct PublishAck {
27    /// Topic the event landed on.
28    pub topic: String,
29    /// Partition index (derived from the partition key).
30    pub partition: u32,
31    /// Backend replay cursor: partition-local offset for the embedded
32    /// broker and Kafka; JetStream stream sequence for NATS.
33    pub offset: u64,
34}
35
36/// An event as stored/replayed.
37#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
38pub struct StoredEvent {
39    /// Partition index.
40    pub partition: u32,
41    /// Backend replay cursor (see [`PublishAck::offset`]).
42    pub offset: u64,
43    /// Partition key (`ai_agent.instance_uid`).
44    pub key: String,
45    /// The event payload.
46    pub value: serde_json::Value,
47    /// Broker-assigned append timestamp (epoch ms).
48    pub stored_at: u64,
49}
50
51/// Publish/consume abstraction. Synchronous by design: the harness calls it
52/// from worker threads (never the hot path), and network backends manage
53/// their own I/O runtime internally.
54pub trait EventBus: Send + Sync {
55    /// Configure the signer-derived key used for authenticated local controls.
56    fn set_control_key(&self, _key: [u8; 32]) -> Result<(), BusError> {
57        Ok(())
58    }
59
60    /// Publish `value` onto `topic`, partitioned by `key`. Returns the ack.
61    fn publish(&self, topic: &str, key: &str, value: &serde_json::Value) -> Result<PublishAck, BusError>;
62
63    /// Publish a stable event UID. Backends with native or local deduplication
64    /// override this method; the default preserves compatibility while the UID
65    /// remains embedded in the event payload for downstream deduplication.
66    fn publish_idempotent(
67        &self,
68        topic: &str,
69        key: &str,
70        value: &serde_json::Value,
71        _event_uid: &str,
72    ) -> Result<PublishAck, BusError> {
73        self.publish(topic, key, value)
74    }
75
76    /// Locate an already committed event by stable UID during crash recovery.
77    fn find_event_by_uid(
78        &self,
79        topic: &str,
80        key: &str,
81        event_uid: &str,
82    ) -> Result<Option<PublishAck>, BusError> {
83        let partition = partition_for(key, self.partitions(topic)?);
84        let mut offset = 0u64;
85        loop {
86            let events = self.fetch(topic, partition, offset, 1_024)?;
87            if events.is_empty() {
88                return Ok(None);
89            }
90            for event in &events {
91                if event
92                    .value
93                    .get("metadata")
94                    .and_then(|metadata| metadata.get("uid"))
95                    .and_then(serde_json::Value::as_str)
96                    == Some(event_uid)
97                {
98                    return Ok(Some(PublishAck {
99                        topic: topic.to_owned(),
100                        partition,
101                        offset: event.offset,
102                    }));
103                }
104            }
105            let next = events
106                .last()
107                .and_then(|event| event.offset.checked_add(1))
108                .ok_or_else(|| BusError::Backend("event lookup offset overflow".to_owned()))?;
109            if next <= offset {
110                return Err(BusError::Backend(
111                    "event lookup made no offset progress".to_owned(),
112                ));
113            }
114            offset = next;
115        }
116    }
117
118    /// Read up to `max` events from `topic`/`partition` starting at `offset`
119    /// (ordered replay).
120    fn fetch(
121        &self,
122        topic: &str,
123        partition: u32,
124        offset: u64,
125        max: usize,
126    ) -> Result<Vec<StoredEvent>, BusError>;
127
128    /// Number of partitions configured for `topic`.
129    fn partitions(&self, topic: &str) -> Result<u32, BusError>;
130
131    /// Topics currently provisioned.
132    fn topics(&self) -> Vec<String>;
133
134    /// Run backend maintenance such as hot-retention expiry. Managed brokers
135    /// may return zero when retention is enforced natively.
136    fn maintenance(&self, _now_ms: u64) -> Result<u64, BusError> {
137        Ok(0)
138    }
139}
140
141/// Stable partition assignment: FNV-1a of the key modulo partition count.
142/// Deterministic across processes and platforms so replay tooling and the
143/// broker always agree (a hash mismatch would silently split an agent's
144/// ordered stream).
145pub fn partition_for(key: &str, partitions: u32) -> u32 {
146    let h = av_core::hash::fnv1a(key.as_bytes());
147    #[allow(clippy::cast_possible_truncation)]
148    ((h % u64::from(partitions.max(1))) as u32)
149}
150
151#[cfg(any(feature = "nats", feature = "kafka", feature = "cold-store"))]
152type ConnectorTask = Box<dyn FnOnce(tokio::runtime::Handle) + Send + 'static>;
153
154#[cfg(any(feature = "nats", feature = "kafka", feature = "cold-store"))]
155enum ConnectorCommand {
156    Run(ConnectorTask),
157    Shutdown,
158}
159
160/// Persistent runtime owner for synchronous network-bus adapters.
161#[cfg(any(feature = "nats", feature = "kafka", feature = "cold-store"))]
162pub(crate) struct ConnectorExecutor {
163    sender: std::sync::mpsc::SyncSender<ConnectorCommand>,
164    thread: parking_lot::Mutex<Option<std::thread::JoinHandle<()>>>,
165}
166
167#[cfg(any(feature = "nats", feature = "kafka", feature = "cold-store"))]
168impl ConnectorExecutor {
169    pub(crate) fn new(name: &str) -> Result<Self, BusError> {
170        let (sender, receiver) = std::sync::mpsc::sync_channel::<ConnectorCommand>(1_024);
171        let (ready_sender, ready_receiver) = std::sync::mpsc::sync_channel(1);
172        let thread = std::thread::Builder::new()
173            .name(name.to_owned())
174            .spawn(move || {
175                let runtime = tokio::runtime::Builder::new_multi_thread()
176                    .worker_threads(1)
177                    .enable_all()
178                    .build()
179                    .map_err(|error| error.to_string());
180                match runtime {
181                    Ok(runtime) => {
182                        let _ = ready_sender.send(Ok(()));
183                        while let Ok(command) = receiver.recv() {
184                            match command {
185                                ConnectorCommand::Run(task) => task(runtime.handle().clone()),
186                                ConnectorCommand::Shutdown => break,
187                            }
188                        }
189                    }
190                    Err(error) => {
191                        let _ = ready_sender.send(Err(error));
192                    }
193                }
194            })
195            .map_err(BusError::Io)?;
196        ready_receiver
197            .recv()
198            .map_err(|_| BusError::Backend("connector runtime failed to start".to_owned()))?
199            .map_err(BusError::Backend)?;
200        Ok(Self {
201            sender,
202            thread: parking_lot::Mutex::new(Some(thread)),
203        })
204    }
205
206    pub(crate) fn run<F, Fut, T>(&self, operation: F) -> Result<T, BusError>
207    where
208        F: FnOnce() -> Fut + Send + 'static,
209        Fut: std::future::Future<Output = T> + Send + 'static,
210        T: Send + 'static,
211    {
212        let (result_sender, result_receiver) = std::sync::mpsc::sync_channel(1);
213        self.sender
214            .send(ConnectorCommand::Run(Box::new(move |runtime| {
215                runtime.spawn(async move {
216                    let result = tokio::time::timeout(std::time::Duration::from_secs(10), operation()).await;
217                    let _ = result_sender.send(result);
218                });
219            })))
220            .map_err(|_| BusError::Backend("connector runtime is closed".to_owned()))?;
221        result_receiver
222            .recv()
223            .map_err(|_| BusError::Backend("connector operation was interrupted".to_owned()))
224            .and_then(|result| {
225                result.map_err(|_| BusError::Backend("connector operation timed out".to_owned()))
226            })
227    }
228}
229
230#[cfg(any(feature = "nats", feature = "kafka", feature = "cold-store"))]
231impl Drop for ConnectorExecutor {
232    fn drop(&mut self) {
233        let _ = self.sender.send(ConnectorCommand::Shutdown);
234        if let Some(thread) = self.thread.get_mut().take() {
235            let _ = thread.join();
236        }
237    }
238}
239
240#[cfg(test)]
241mod tests {
242    #![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
243
244    use super::*;
245    #[cfg(any(feature = "nats", feature = "kafka"))]
246    use std::sync::Arc;
247
248    #[test]
249    fn partition_assignment_is_stable() {
250        // Pinned values: changing the hash silently re-partitions every
251        // deployment's replay order — this test makes that loud.
252        assert_eq!(partition_for("agent-inst-1", 8), partition_for("agent-inst-1", 8));
253        let spread: std::collections::HashSet<u32> =
254            (0..100).map(|i| partition_for(&format!("inst-{i}"), 8)).collect();
255        assert!(spread.len() >= 6, "poor partition spread: {spread:?}");
256    }
257
258    #[test]
259    fn zero_partitions_clamped() {
260        assert_eq!(partition_for("x", 0), 0);
261    }
262
263    #[cfg(any(feature = "nats", feature = "kafka"))]
264    #[test]
265    fn persistent_connector_runtime_is_safe_inside_tokio() {
266        let outer = tokio::runtime::Runtime::new().unwrap();
267        outer.block_on(async {
268            let connector = ConnectorExecutor::new("test-connector").unwrap();
269            assert_eq!(connector.run(|| async { 41u64 }).unwrap(), 41);
270            assert_eq!(connector.run(|| async { 42u64 }).unwrap(), 42);
271            drop(connector);
272        });
273    }
274
275    #[cfg(any(feature = "nats", feature = "kafka"))]
276    #[test]
277    fn connector_operations_can_overlap() {
278        let connector = Arc::new(ConnectorExecutor::new("parallel-connector").unwrap());
279        let barrier = Arc::new(tokio::sync::Barrier::new(2));
280        std::thread::scope(|scope| {
281            let first_connector = Arc::clone(&connector);
282            let first_barrier = Arc::clone(&barrier);
283            let first = scope.spawn(move || {
284                first_connector
285                    .run(move || async move {
286                        first_barrier.wait().await;
287                        1u64
288                    })
289                    .unwrap()
290            });
291            let second_connector = Arc::clone(&connector);
292            let second_barrier = Arc::clone(&barrier);
293            let second = scope.spawn(move || {
294                second_connector
295                    .run(move || async move {
296                        second_barrier.wait().await;
297                        2u64
298                    })
299                    .unwrap()
300            });
301            assert_eq!(first.join().unwrap() + second.join().unwrap(), 3);
302        });
303    }
304}
av_bridge/bus.rs

av_bridge/
bus.rs
