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
24impl BusError {
25    /// Distinguish permanent config errors
26    /// from transient outages so downstream error mapping can route
27    /// them to the right HTTP status. `UnknownTopic` is a
28    /// misconfiguration — the topic must be provisioned via the
29    /// manifest before publishing, so a retry without operator
30    /// action cannot succeed. `Serde` is a wire-shape violation
31    /// from the sender — also permanent for that specific payload.
32    /// `Io` and `Backend` are transient by default (network blip,
33    /// disk temporarily full, broker restarting); reclassifying
34    /// specific-message shapes here would require pattern-matching
35    /// the underlying error, which is fragile.
36    ///
37    /// Callers should use `is_permanent()` to decide whether to
38    /// return HTTP 400/409 (permanent, no Retry-After) or 503
39    /// (transient, with Retry-After) — see
40    /// `av-harness::routes::finalize_error_response`.
41    #[must_use]
42    pub fn is_permanent(&self) -> bool {
43        matches!(self, Self::UnknownTopic(_) | Self::Serde(_))
44    }
45}
46
47/// Acknowledgment for a published event.
48#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
49pub struct PublishAck {
50    /// Topic the event landed on.
51    pub topic: String,
52    /// Partition index (derived from the partition key).
53    pub partition: u32,
54    /// Backend replay cursor: partition-local offset for the embedded
55    /// broker and Kafka; JetStream stream sequence for NATS.
56    pub offset: u64,
57}
58
59/// An event as stored/replayed.
60#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
61pub struct StoredEvent {
62    /// Partition index.
63    pub partition: u32,
64    /// Backend replay cursor (see [`PublishAck::offset`]).
65    pub offset: u64,
66    /// Partition key (`ai_agent.instance_uid`).
67    pub key: String,
68    /// The event payload.
69    pub value: serde_json::Value,
70    /// Broker-assigned append timestamp (epoch ms).
71    pub stored_at: u64,
72}
73
74/// Publish/consume abstraction. Synchronous by design: the harness calls it
75/// from worker threads (never the hot path), and network backends manage
76/// their own I/O runtime internally.
77pub trait EventBus: Send + Sync {
78    /// Configure the signer-derived key used for authenticated local controls.
79    fn set_control_key(&self, _key: [u8; 32]) -> Result<(), BusError> {
80        Ok(())
81    }
82
83    /// Publish `value` onto `topic`, partitioned by `key`. Returns the ack.
84    fn publish(&self, topic: &str, key: &str, value: &serde_json::Value) -> Result<PublishAck, BusError>;
85
86    /// Publish a stable event UID. Backends with native or local deduplication
87    /// override this method; the default preserves compatibility while the UID
88    /// remains embedded in the event payload for downstream deduplication.
89    fn publish_idempotent(
90        &self,
91        topic: &str,
92        key: &str,
93        value: &serde_json::Value,
94        _event_uid: &str,
95    ) -> Result<PublishAck, BusError> {
96        self.publish(topic, key, value)
97    }
98
99    /// Locate an already committed event by stable UID during crash recovery.
100    fn find_event_by_uid(
101        &self,
102        topic: &str,
103        key: &str,
104        event_uid: &str,
105    ) -> Result<Option<PublishAck>, BusError> {
106        let partition = partition_for(key, self.partitions(topic)?);
107        let mut offset = 0u64;
108        loop {
109            let events = self.fetch(topic, partition, offset, 1_024)?;
110            if events.is_empty() {
111                return Ok(None);
112            }
113            for event in &events {
114                if event
115                    .value
116                    .get("metadata")
117                    .and_then(|metadata| metadata.get("uid"))
118                    .and_then(serde_json::Value::as_str)
119                    == Some(event_uid)
120                {
121                    return Ok(Some(PublishAck {
122                        topic: topic.to_owned(),
123                        partition,
124                        offset: event.offset,
125                    }));
126                }
127            }
128            let next = events
129                .last()
130                .and_then(|event| event.offset.checked_add(1))
131                .ok_or_else(|| BusError::Backend("event lookup offset overflow".to_owned()))?;
132            if next <= offset {
133                return Err(BusError::Backend(
134                    "event lookup made no offset progress".to_owned(),
135                ));
136            }
137            offset = next;
138        }
139    }
140
141    /// Read up to `max` events from `topic`/`partition` starting at `offset`
142    /// (ordered replay).
143    fn fetch(
144        &self,
145        topic: &str,
146        partition: u32,
147        offset: u64,
148        max: usize,
149    ) -> Result<Vec<StoredEvent>, BusError>;
150
151    /// Number of partitions configured for `topic`.
152    fn partitions(&self, topic: &str) -> Result<u32, BusError>;
153
154    /// Topics currently provisioned.
155    fn topics(&self) -> Vec<String>;
156
157    /// Run backend maintenance such as hot-retention expiry. Managed brokers
158    /// may return zero when retention is enforced natively.
159    fn maintenance(&self, _now_ms: u64) -> Result<u64, BusError> {
160        Ok(0)
161    }
162}
163
164/// Stable partition assignment: FNV-1a of the key modulo partition count.
165/// Deterministic across processes and platforms so replay tooling and the
166/// broker always agree (a hash mismatch would silently split an agent's
167/// ordered stream).
168pub fn partition_for(key: &str, partitions: u32) -> u32 {
169    let h = av_core::hash::fnv1a(key.as_bytes());
170    #[allow(clippy::cast_possible_truncation)]
171    ((h % u64::from(partitions.max(1))) as u32)
172}
173
174#[cfg(any(feature = "nats", feature = "kafka", feature = "cold-store"))]
175type ConnectorTask = Box<dyn FnOnce(tokio::runtime::Handle) + Send + 'static>;
176
177#[cfg(any(feature = "nats", feature = "kafka", feature = "cold-store"))]
178enum ConnectorCommand {
179    Run(ConnectorTask),
180    Shutdown,
181}
182
183/// Persistent runtime owner for synchronous network-bus adapters.
184#[cfg(any(feature = "nats", feature = "kafka", feature = "cold-store"))]
185pub(crate) struct ConnectorExecutor {
186    sender: std::sync::mpsc::SyncSender<ConnectorCommand>,
187    thread: parking_lot::Mutex<Option<std::thread::JoinHandle<()>>>,
188}
189
190#[cfg(any(feature = "nats", feature = "kafka", feature = "cold-store"))]
191impl ConnectorExecutor {
192    pub(crate) fn new(name: &str) -> Result<Self, BusError> {
193        let (sender, receiver) = std::sync::mpsc::sync_channel::<ConnectorCommand>(1_024);
194        let (ready_sender, ready_receiver) = std::sync::mpsc::sync_channel(1);
195        let thread = std::thread::Builder::new()
196            .name(name.to_owned())
197            .spawn(move || {
198                let runtime = tokio::runtime::Builder::new_multi_thread()
199                    .worker_threads(1)
200                    .enable_all()
201                    .build()
202                    .map_err(|error| error.to_string());
203                match runtime {
204                    Ok(runtime) => {
205                        let _ = ready_sender.send(Ok(()));
206                        while let Ok(command) = receiver.recv() {
207                            match command {
208                                ConnectorCommand::Run(task) => task(runtime.handle().clone()),
209                                ConnectorCommand::Shutdown => break,
210                            }
211                        }
212                    }
213                    Err(error) => {
214                        let _ = ready_sender.send(Err(error));
215                    }
216                }
217            })
218            .map_err(BusError::Io)?;
219        ready_receiver
220            .recv()
221            .map_err(|_| BusError::Backend("connector runtime failed to start".to_owned()))?
222            .map_err(BusError::Backend)?;
223        Ok(Self {
224            sender,
225            thread: parking_lot::Mutex::new(Some(thread)),
226        })
227    }
228
229    pub(crate) fn run<F, Fut, T>(&self, operation: F) -> Result<T, BusError>
230    where
231        F: FnOnce() -> Fut + Send + 'static,
232        Fut: std::future::Future<Output = T> + Send + 'static,
233        T: Send + 'static,
234    {
235        let (result_sender, result_receiver) = std::sync::mpsc::sync_channel(1);
236        self.sender
237            .send(ConnectorCommand::Run(Box::new(move |runtime| {
238                runtime.spawn(async move {
239                    let result = tokio::time::timeout(std::time::Duration::from_secs(10), operation()).await;
240                    let _ = result_sender.send(result);
241                });
242            })))
243            .map_err(|_| BusError::Backend("connector runtime is closed".to_owned()))?;
244        result_receiver
245            .recv()
246            .map_err(|_| BusError::Backend("connector operation was interrupted".to_owned()))
247            .and_then(|result| {
248                result.map_err(|_| BusError::Backend("connector operation timed out".to_owned()))
249            })
250    }
251}
252
253#[cfg(any(feature = "nats", feature = "kafka", feature = "cold-store"))]
254impl Drop for ConnectorExecutor {
255    fn drop(&mut self) {
256        let _ = self.sender.send(ConnectorCommand::Shutdown);
257        if let Some(thread) = self.thread.get_mut().take() {
258            let _ = thread.join();
259        }
260    }
261}
262
263#[cfg(test)]
264mod tests {
265    #![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
266
267    use super::*;
268    #[cfg(any(feature = "nats", feature = "kafka"))]
269    use std::sync::Arc;
270
271    /// Round-21 mutation finding: every shipped backend overrides
272    /// `find_event_by_uid`, so the trait DEFAULT (the contract any
273    /// future backend inherits) had zero coverage — replacing its whole
274    /// body with `Ok(None)` survived, as did flipping the uid compare
275    /// and the no-progress guard. A future backend inheriting a broken
276    /// default would silently lose publish idempotency (dup events on
277    /// retry). Pin the default against a minimal in-memory bus that
278    /// deliberately does NOT override it.
279    struct DefaultLookupBus {
280        events: Vec<StoredEvent>,
281    }
282
283    impl EventBus for DefaultLookupBus {
284        fn publish(
285            &self,
286            _topic: &str,
287            _key: &str,
288            _value: &serde_json::Value,
289        ) -> Result<PublishAck, BusError> {
290            Err(BusError::Backend("publish unused in this test".to_owned()))
291        }
292        fn fetch(
293            &self,
294            _topic: &str,
295            _partition: u32,
296            offset: u64,
297            max: usize,
298        ) -> Result<Vec<StoredEvent>, BusError> {
299            Ok(self
300                .events
301                .iter()
302                .filter(|event| event.offset >= offset)
303                .take(max)
304                .cloned()
305                .collect())
306        }
307        fn partitions(&self, _topic: &str) -> Result<u32, BusError> {
308            Ok(1)
309        }
310        fn topics(&self) -> Vec<String> {
311            vec!["t".to_owned()]
312        }
313    }
314
315    fn stored(offset: u64, uid: &str) -> StoredEvent {
316        StoredEvent {
317            partition: 0,
318            offset,
319            key: "k".to_owned(),
320            value: serde_json::json!({"metadata": {"uid": uid}, "n": offset}),
321            stored_at: 0,
322        }
323    }
324
325    #[test]
326    fn default_find_event_by_uid_matches_exact_uid_only() {
327        let bus = DefaultLookupBus {
328            events: vec![stored(0, "uid-a"), stored(1, "uid-b"), stored(2, "uid-c")],
329        };
330        let hit = bus.find_event_by_uid("t", "k", "uid-b").unwrap();
331        assert_eq!(
332            hit.map(|ack| ack.offset),
333            Some(1),
334            "default lookup must find the exact uid at its offset"
335        );
336        assert!(
337            bus.find_event_by_uid("t", "k", "uid-missing").unwrap().is_none(),
338            "default lookup must return None for an absent uid, not a false ack"
339        );
340        // Empty bus: terminates with None (no progress-loop hang).
341        let empty = DefaultLookupBus { events: vec![] };
342        assert!(empty.find_event_by_uid("t", "k", "uid-a").unwrap().is_none());
343        // Default maintenance is an explicit no-op count.
344        assert_eq!(
345            bus.maintenance(0).unwrap(),
346            0,
347            "default maintenance must report zero work"
348        );
349    }
350
351    #[test]
352    fn partition_assignment_is_stable() {
353        // Pinned values (FNV-1a 64 mod 8): changing the hash silently
354        // re-partitions every deployment's replay order — this test makes
355        // that loud. A self-comparison would pass under ANY hash; the
356        // literals are the regression anchor.
357        assert_eq!(partition_for("agent-inst-1", 8), 3);
358        assert_eq!(partition_for("agent-inst-2", 8), 6);
359        assert_eq!(partition_for("session-abc", 8), 2);
360        assert_eq!(partition_for("", 8), 5);
361        let spread: std::collections::HashSet<u32> =
362            (0..100).map(|i| partition_for(&format!("inst-{i}"), 8)).collect();
363        assert!(spread.len() >= 6, "poor partition spread: {spread:?}");
364    }
365
366    #[test]
367    fn zero_partitions_clamped() {
368        assert_eq!(partition_for("x", 0), 0);
369    }
370
371    /// Permanent vs transient classification is what
372    /// downstream error mapping (av-harness FinalizeError) uses to
373    /// decide between HTTP 400 (permanent misconfig, no retry) and
374    /// HTTP 503 (transient outage, Retry-After). Pin the two-way
375    /// split.
376    #[test]
377    fn bus_error_is_permanent_split() {
378        assert!(
379            BusError::UnknownTopic("orders".to_owned()).is_permanent(),
380            "unknown topic is a misconfiguration; retry cannot succeed"
381        );
382        assert!(
383            !BusError::Backend("network reset".to_owned()).is_permanent(),
384            "backend outages are transient by default"
385        );
386        assert!(
387            !BusError::Io(std::io::Error::new(std::io::ErrorKind::TimedOut, "slow disk")).is_permanent(),
388            "I/O timeouts are transient"
389        );
390        // Serde errors are permanent for that payload — the sender
391        // must fix the wire shape.
392        let serde_err = serde_json::from_slice::<serde_json::Value>(b"not json").unwrap_err();
393        assert!(
394            BusError::Serde(serde_err).is_permanent(),
395            "serde errors are permanent for that payload"
396        );
397    }
398
399    #[cfg(any(feature = "nats", feature = "kafka"))]
400    #[test]
401    fn persistent_connector_runtime_is_safe_inside_tokio() {
402        let outer = tokio::runtime::Runtime::new().unwrap();
403        outer.block_on(async {
404            let connector = ConnectorExecutor::new("test-connector").unwrap();
405            assert_eq!(connector.run(|| async { 41u64 }).unwrap(), 41);
406            assert_eq!(connector.run(|| async { 42u64 }).unwrap(), 42);
407            drop(connector);
408        });
409    }
410
411    #[cfg(any(feature = "nats", feature = "kafka"))]
412    #[test]
413    fn connector_operations_can_overlap() {
414        let connector = Arc::new(ConnectorExecutor::new("parallel-connector").unwrap());
415        let barrier = Arc::new(tokio::sync::Barrier::new(2));
416        std::thread::scope(|scope| {
417            let first_connector = Arc::clone(&connector);
418            let first_barrier = Arc::clone(&barrier);
419            let first = scope.spawn(move || {
420                first_connector
421                    .run(move || async move {
422                        first_barrier.wait().await;
423                        1u64
424                    })
425                    .unwrap()
426            });
427            let second_connector = Arc::clone(&connector);
428            let second_barrier = Arc::clone(&barrier);
429            let second = scope.spawn(move || {
430                second_connector
431                    .run(move || async move {
432                        second_barrier.wait().await;
433                        2u64
434                    })
435                    .unwrap()
436            });
437            assert_eq!(first.join().unwrap() + second.join().unwrap(), 3);
438        });
439    }
440}
av_bridge/bus.rs

av_bridge/
bus.rs
