rusteron_archive/
lib.rs

1/**/
2#![allow(non_upper_case_globals)]
3#![allow(non_camel_case_types)]
4#![allow(non_snake_case)]
5#![allow(clippy::all)]
6#![allow(unused_unsafe)]
7#![allow(unused_variables)]
8#![doc = include_str!("../README.md")]
9//! # Features
10//!
11//! - **`static`**: When enabled, this feature statically links the Aeron C code.
12//!   By default, the library uses dynamic linking to the Aeron C libraries.
13//! - **`backtrace`**: When enabled will log a backtrace for each AeronCError
14//! - **`extra-logging`**: When enabled will log when resource is created and destroyed. Useful if you're seeing a segfault due to a resource being closed
15//! - **`log-c-bindings`**: When enabled will log every C binding call with arguments and return values. Useful for debugging FFI interactions
16//! - **`precompile`**: When enabled will use precompiled C code instead of requiring cmake and java to be installed
17
18#[allow(improper_ctypes_definitions)]
19#[allow(unpredictable_function_pointer_comparisons)]
20pub mod bindings {
21    include!(concat!(env!("OUT_DIR"), "/bindings.rs"));
22}
23
24use bindings::*;
25use std::cell::Cell;
26use std::os::raw::c_int;
27#[cfg(feature = "multi-threaded")]
28use std::sync::atomic::Ordering;
29use std::time::{Duration, Instant};
30
31/// Result codes returned by `AeronPublication::offer` / `try_claim` (Aeron `aeronc.h`). A positive
32/// value is the resulting log position; the negatives classify the failure.
33///
34/// **Fatal** (stop offering): [`PUBLICATION_CLOSED`], [`PUBLICATION_MAX_POSITION_EXCEEDED`],
35/// [`PUBLICATION_ERROR`]. **Transient** (retry): [`PUBLICATION_BACK_PRESSURED`],
36/// [`PUBLICATION_NOT_CONNECTED`], [`PUBLICATION_ADMIN_ACTION`].
37pub const PUBLICATION_NOT_CONNECTED: i64 = bindings::AERON_PUBLICATION_NOT_CONNECTED as i64;
38pub const PUBLICATION_BACK_PRESSURED: i64 = bindings::AERON_PUBLICATION_BACK_PRESSURED as i64;
39pub const PUBLICATION_ADMIN_ACTION: i64 = bindings::AERON_PUBLICATION_ADMIN_ACTION as i64;
40pub const PUBLICATION_CLOSED: i64 = bindings::AERON_PUBLICATION_CLOSED as i64;
41pub const PUBLICATION_MAX_POSITION_EXCEEDED: i64 = bindings::AERON_PUBLICATION_MAX_POSITION_EXCEEDED as i64;
42pub const PUBLICATION_ERROR: i64 = bindings::AERON_PUBLICATION_ERROR as i64;
43
44pub mod testing;
45
46#[cfg(test)]
47pub mod persistent_subscription_integration;
48#[cfg(test)]
49pub mod persistent_subscription_tests;
50
51include!(concat!(env!("OUT_DIR"), "/aeron.rs"));
52include!(concat!(env!("OUT_DIR"), "/aeron_custom.rs"));
53
54pub type SourceLocation = bindings::aeron_archive_source_location_t;
55pub const SOURCE_LOCATION_LOCAL: aeron_archive_source_location_en = SourceLocation::AERON_ARCHIVE_SOURCE_LOCATION_LOCAL;
56pub const SOURCE_LOCATION_REMOTE: aeron_archive_source_location_en =
57    SourceLocation::AERON_ARCHIVE_SOURCE_LOCATION_REMOTE;
58
59pub struct RecordingPos;
60impl RecordingPos {
61    pub fn find_counter_id_by_session(counter_reader: &AeronCountersReader, session_id: i32) -> i32 {
62        unsafe { aeron_archive_recording_pos_find_counter_id_by_session_id(counter_reader.get_inner(), session_id) }
63    }
64    pub fn find_counter_id_by_recording_id(counter_reader: &AeronCountersReader, recording_id: i64) -> i32 {
65        unsafe { aeron_archive_recording_pos_find_counter_id_by_recording_id(counter_reader.get_inner(), recording_id) }
66    }
67
68    /// Return the recordingId embedded in the key of the given counter
69    /// if it is indeed a "recording position" counter. Otherwise return -1.
70    pub fn get_recording_id_block(
71        counters_reader: &AeronCountersReader,
72        counter_id: i32,
73        wait: Duration,
74    ) -> Result<i64, AeronCError> {
75        let mut result = Self::get_recording_id(counters_reader, counter_id);
76        let instant = Instant::now();
77
78        while result.is_err() && instant.elapsed() < wait {
79            result = Self::get_recording_id(counters_reader, counter_id);
80            #[cfg(debug_assertions)]
81            std::thread::sleep(Duration::from_millis(10));
82        }
83
84        return result;
85    }
86
87    /// Return the recordingId embedded in the key of the given counter
88    /// if it is indeed a "recording position" counter. Otherwise return -1.
89    pub fn get_recording_id(counters_reader: &AeronCountersReader, counter_id: i32) -> Result<i64, AeronCError> {
90        /// The type id for an Aeron Archive recording position counter.
91        /// In Aeron Java, this is AeronCounters.ARCHIVE_RECORDING_POSITION_TYPE_ID (which is typically 100).
92        pub const RECORDING_POSITION_TYPE_ID: i32 = 100;
93
94        /// from Aeron Java code
95        pub const RECORD_ALLOCATED: i32 = 1;
96
97        /// A constant to mean "no valid recording ID".
98        pub const NULL_RECORDING_ID: i64 = -1;
99
100        if counter_id < 0 {
101            return Err(AeronCError::from_code(NULL_RECORDING_ID as i32));
102        }
103
104        let state = counters_reader.counter_state(counter_id)?;
105        if state != RECORD_ALLOCATED {
106            return Err(AeronCError::from_code(NULL_RECORDING_ID as i32));
107        }
108
109        let type_id = counters_reader.counter_type_id(counter_id)?;
110        if type_id != RECORDING_POSITION_TYPE_ID {
111            return Err(AeronCError::from_code(NULL_RECORDING_ID as i32));
112        }
113
114        // Read the key area. For a RECORDING_POSITION_TYPE_ID counter:
115        //    - offset 0..8 => the i64 recording_id
116        //    - offset 8..12 => the session_id (int)
117        //    etc...
118        // only need the first 8 bytes to get the recordingId.
119        let recording_id = Cell::new(-1);
120        counters_reader.foreach_counter_fn(|value, id, type_id, key, label| {
121            if id == counter_id && type_id == RECORDING_POSITION_TYPE_ID {
122                let mut val = [0u8; 8];
123                val.copy_from_slice(&key[0..8]);
124                let Ok(value) = i64::from_le_bytes(val).try_into();
125                recording_id.set(value);
126            }
127        });
128        let recording_id = recording_id.get();
129        if recording_id < 0 {
130            return Err(AeronCError::from_code(NULL_RECORDING_ID as i32));
131        }
132
133        Ok(recording_id)
134    }
135}
136
137impl AeronArchive {
138    pub fn aeron(&self) -> Aeron {
139        self.get_archive_context().get_aeron()
140    }
141
142    /// Find the latest recording matching a predicate.
143    /// Returns the recording with the highest recording_id that matches the predicate.
144    pub fn find_recording<F>(&self, mut predicate: F) -> Result<Option<RecordingDescriptor>, AeronCError>
145    where
146        F: FnMut(&RecordingDescriptor) -> bool,
147    {
148        // Find the latest matching recording by fetching all recordings in one call.
149        // Uses record_count=i32::MAX to fetch all available recordings.
150        let mut result = None;
151        let mut count = 0;
152        self.list_recordings_fn(&mut count, 0, i32::MAX, |desc| {
153            let descriptor = self.descriptor_to_owned(&desc);
154            if predicate(&descriptor) {
155                if result.is_none()
156                    || result
157                        .as_ref()
158                        .map(|r: &RecordingDescriptor| r.recording_id)
159                        .unwrap_or(0)
160                        < descriptor.recording_id
161                {
162                    result = Some(descriptor);
163                }
164            }
165        })?;
166        Ok(result)
167    }
168
169    /// Find the latest recording for a given stream ID.
170    pub fn find_recording_for_stream(&self, stream_id: i32) -> Result<Option<RecordingDescriptor>, AeronCError> {
171        self.find_recording(|desc| desc.stream_id == stream_id)
172    }
173
174    /// Collect all recordings matching a predicate.
175    pub fn collect_recordings<F>(&self, mut predicate: F) -> Result<Vec<RecordingDescriptor>, AeronCError>
176    where
177        F: FnMut(&RecordingDescriptor) -> bool,
178    {
179        // Collect all matching recordings by fetching all recordings in one call.
180        // Uses record_count=i32::MAX to fetch all available recordings.
181        let mut recordings = Vec::new();
182        let mut count = 0;
183        self.list_recordings_fn(&mut count, 0, i32::MAX, |desc| {
184            let descriptor = self.descriptor_to_owned(&desc);
185            if predicate(&descriptor) {
186                recordings.push(descriptor);
187            }
188        })?;
189        Ok(recordings)
190    }
191
192    /// Convert a callback-scoped recording descriptor to an owned struct.
193    fn descriptor_to_owned(&self, desc: &AeronArchiveRecordingDescriptor) -> RecordingDescriptor {
194        let start_position = desc.start_position();
195        let stop_position = desc.stop_position();
196        RecordingDescriptor {
197            recording_id: desc.recording_id(),
198            start_position,
199            stop_position,
200            start_timestamp: desc.start_timestamp(),
201            stop_timestamp: desc.stop_timestamp(),
202            position: stop_position.saturating_sub(start_position),
203            recording_length: stop_position.saturating_sub(start_position),
204            control_session_id: desc.control_session_id() as i32,
205            correlation_id: desc.correlation_id(),
206            session_id: desc.session_id(),
207            stream_id: desc.stream_id(),
208            channel: desc.stripped_channel().to_string(),
209            source_identity: desc.source_identity().to_string(),
210            original_channel: desc.original_channel().to_string(),
211        }
212    }
213}
214
215impl AeronArchiveAsyncConnect {
216    #[inline]
217    /// recommend using this method instead of standard `new` as it will link the archive to aeron so if a drop occurs archive is dropped before aeron
218    pub fn new_with_aeron(ctx: &AeronArchiveContext, aeron: &Aeron) -> Result<Self, AeronCError> {
219        let resource_async = Self::new(ctx)?;
220        resource_async.inner.add_dependency(aeron.clone());
221        Ok(resource_async)
222    }
223}
224
225macro_rules! impl_archive_position_methods {
226    ($pub_type:ty) => {
227        impl $pub_type {
228            /// Retrieves the current active live archive position using the Aeron counters.
229            /// Returns an error if not found.
230            pub fn get_archive_position(&self) -> Result<i64, AeronCError> {
231                if let Some(aeron) = self.inner.get_dependency::<Aeron>() {
232                    let counter_reader = &aeron.counters_reader();
233                    self.get_archive_position_with(counter_reader)
234                } else {
235                    Err(AeronCError::from_code(-1))
236                }
237            }
238
239            /// Retrieves the current active live archive position using the provided counter reader.
240            /// Returns an error if not found.
241            pub fn get_archive_position_with(&self, counters: &AeronCountersReader) -> Result<i64, AeronCError> {
242                let session_id = self.get_constants()?.session_id();
243                let counter_id = RecordingPos::find_counter_id_by_session(counters, session_id);
244                if counter_id < 0 {
245                    return Err(AeronCError::from_code(counter_id));
246                }
247                let position = counters.get_counter_value(counter_id);
248                if position < 0 {
249                    return Err(AeronCError::from_code(position as i32));
250                }
251                Ok(position)
252            }
253
254            /// Checks if the publication's current position is within a specified inclusive length
255            /// of the archive position.
256            pub fn is_archive_position_with(&self, length_inclusive: usize) -> bool {
257                let archive_position = self.get_archive_position().unwrap_or(-1);
258                if archive_position < 0 {
259                    return false;
260                }
261                self.position() - archive_position <= length_inclusive as i64
262            }
263        }
264    };
265}
266
267impl_archive_position_methods!(AeronPublication);
268impl_archive_position_methods!(AeronExclusivePublication);
269
270/// Recording descriptor for owned recording data
271#[derive(Debug, Clone)]
272pub struct RecordingDescriptor {
273    pub recording_id: i64,
274    pub start_position: i64,
275    pub stop_position: i64,
276    pub start_timestamp: i64,
277    pub stop_timestamp: i64,
278    pub position: i64,
279    pub recording_length: i64,
280    pub control_session_id: i32,
281    pub correlation_id: i64,
282    pub session_id: i32,
283    pub stream_id: i32,
284    pub channel: String,
285    pub source_identity: String,
286    pub original_channel: String,
287}
288
289/// Wrapper for `Box<dyn PersistentSubscriptionListener>` that provides a stable
290/// thin pointer for C FFI callbacks.
291struct ListenerHolder {
292    listener: Box<dyn PersistentSubscriptionListener>,
293}
294
295// Hand-written trampolines: the code generator only wires single-callback args,
296// but this listener has 3 callbacks sharing one clientd.
297unsafe extern "C" fn persistent_subscription_on_live_joined(clientd: *mut std::ffi::c_void) {
298    if !clientd.is_null() {
299        // SAFETY: clientd is the ListenerHolder kept alive as a dependency of the subscription.
300        let holder = &*(clientd as *const ListenerHolder);
301        holder.listener.on_live_joined();
302    }
303}
304
305unsafe extern "C" fn persistent_subscription_on_live_left(clientd: *mut std::ffi::c_void) {
306    if !clientd.is_null() {
307        let holder = &*(clientd as *const ListenerHolder);
308        holder.listener.on_live_left();
309    }
310}
311
312unsafe extern "C" fn persistent_subscription_on_error(
313    clientd: *mut std::ffi::c_void,
314    error_code: c_int,
315    error_message: *const std::os::raw::c_char,
316) {
317    if !clientd.is_null() {
318        let holder = &*(clientd as *const ListenerHolder);
319        let msg = if !error_message.is_null() {
320            std::ffi::CStr::from_ptr(error_message).to_string_lossy()
321        } else {
322            std::borrow::Cow::Borrowed("")
323        };
324        holder.listener.on_error(error_code, &msg);
325    }
326}
327
328/// Safe creation method for AeronArchivePersistentSubscription
329impl AeronArchivePersistentSubscription {
330    /// Create a persistent subscription from a context.
331    ///
332    /// If `listener` is `Some`, it is wired into the context (the C layer copies
333    /// the callback pointers + `clientd`) and kept alive for the subscription's
334    /// lifetime; it is freed once the subscription is closed.
335    ///
336    /// The context is consumed and owned by the subscription from this point on —
337    /// it will be closed when the subscription is closed, so it must not be used
338    /// afterwards.
339    pub fn create(
340        ctx: AeronArchivePersistentSubscriptionContext,
341        listener: Option<Box<dyn PersistentSubscriptionListener>>,
342    ) -> Result<Self, AeronCError> {
343        use std::os::raw::c_void;
344
345        // Box the listener so we can hand C a stable clientd; a Box's heap address
346        // never moves, so the pointer C copies stays valid until the box drops.
347        let holder_box: Option<Box<ListenerHolder>> = match listener {
348            Some(listener) => {
349                let mut hb = Box::new(ListenerHolder { listener });
350                let holder_ptr: *mut ListenerHolder = &mut *hb;
351                let c_listener = match AeronArchivePersistentSubscriptionListener::new(
352                    Some(persistent_subscription_on_live_joined),
353                    Some(persistent_subscription_on_live_left),
354                    Some(persistent_subscription_on_error),
355                    holder_ptr as *mut c_void,
356                ) {
357                    Ok(l) => l,
358                    Err(e) => return Err(e),
359                };
360                if let Err(e) = ctx.set_listener(c_listener.get_inner()) {
361                    return Err(e);
362                }
363                Some(hb)
364            }
365            None => None,
366        };
367
368        let mut raw_ptr: *mut aeron_archive_persistent_subscription_t = std::ptr::null_mut();
369        unsafe {
370            let result = aeron_archive_persistent_subscription_create(&mut raw_ptr, ctx.get_inner());
371            if result < 0 {
372                return Err(AeronCError::from_code(result));
373            }
374        }
375
376        // The C subscription now owns the context. Mark it already-closed so dropping
377        // `ctx` frees the Rust bookkeeping without re-running the C context close.
378        if let Some(inner) = ctx.inner.as_owned() {
379            #[cfg(feature = "multi-threaded")]
380            inner.close_already_called.store(true, Ordering::SeqCst);
381            #[cfg(not(feature = "multi-threaded"))]
382            inner.close_already_called.set(true);
383        }
384
385        // C close, run on drop if close() wasn't called explicitly.
386        let resource = match ManagedCResource::new(
387            move |ctx_field| unsafe {
388                *ctx_field = raw_ptr;
389                0
390            },
391            Some(Box::new(move |ctx_field| unsafe {
392                aeron_archive_persistent_subscription_close(*ctx_field)
393            })),
394            true,
395        ) {
396            Ok(r) => r,
397            Err(e) => {
398                unsafe {
399                    aeron_archive_persistent_subscription_close(raw_ptr);
400                }
401                return Err(e);
402            }
403        };
404
405        // Reclaim via a dependency, not the cleanup closure — the generated
406        // close() bypasses the closure, but a field always drops.
407        if let Some(hb) = holder_box {
408            resource.add_dependency(hb);
409        }
410
411        Ok(Self {
412            inner: CResource::OwnedOnHeap(RcOrArc::new(resource)),
413        })
414    }
415
416    /// Get the failure reason as a tuple of (error_code, error_message).
417    /// Returns None if there is no failure. This is a safe wrapper around the C function.
418    pub fn get_failure_reason(&self) -> Option<(i32, String)> {
419        let mut error_code: i32 = 0;
420        let mut reason_ptr: *const std::os::raw::c_char = std::ptr::null();
421        let has_reason = unsafe {
422            bindings::aeron_archive_persistent_subscription_failure_reason(
423                self.get_inner(),
424                &mut error_code,
425                &mut reason_ptr,
426            )
427        };
428        if has_reason && !reason_ptr.is_null() {
429            let cstr = unsafe { std::ffi::CStr::from_ptr(reason_ptr) };
430            Some((error_code, cstr.to_string_lossy().into_owned()))
431        } else {
432            None
433        }
434    }
435}
436
437/// Sentinel for [`PersistentSubscriptionBuilder::start_position`]: replay from the
438/// beginning of the recording. Maps to Aeron's `AERON_ARCHIVE_PERSISTENT_SUBSCRIPTION_FROM_START`.
439pub const PERSISTENT_SUBSCRIPTION_FROM_START: i64 = -1;
440
441/// Sentinel for [`PersistentSubscriptionBuilder::start_position`]: skip replay and join
442/// the live stream immediately. Maps to Aeron's `AERON_ARCHIVE_PERSISTENT_SUBSCRIPTION_FROM_LIVE`.
443pub const PERSISTENT_SUBSCRIPTION_FROM_LIVE: i64 = -2;
444
445/// Returns a builder for configuring a persistent subscription context
446pub fn persistent_subscription_builder() -> Result<PersistentSubscriptionBuilder, AeronCError> {
447    PersistentSubscriptionBuilder::new()
448}
449
450/// Trait for persistent subscription event listeners.
451/// This provides a safe Rust alternative to using raw C function pointers.
452///
453/// In the poll loop, prefer the state queries `is_live()` / `is_replaying()` /
454/// `has_failed()` for control flow and treat this listener as observational
455/// (logging/metrics). See Aeron's `PersistentSubscriptionListener`.
456pub trait PersistentSubscriptionListener: Send + 'static {
457    /// Called when the persistent subscription transitions to consuming from the
458    /// live stream. Can fire more than once: if the live image is lost the
459    /// subscription falls back to replay and this fires again on rejoin.
460    fn on_live_joined(&self) {}
461
462    /// Called when the persistent subscription stops consuming from the live
463    /// stream (e.g. the live image closed). The subscription automatically falls
464    /// back to replay; no user action required. Can fire repeatedly.
465    fn on_live_left(&self) {}
466
467    /// Called for **both** non-terminal and terminal errors. Non-terminal errors
468    /// (timeouts, transient resource unavailability) are retried automatically.
469    /// A terminal failure flips [`AeronArchivePersistentSubscription::has_failed`]
470    /// to true — check it in the poll loop and read the reason with
471    /// [`AeronArchivePersistentSubscription::get_failure_reason`].
472    fn on_error(&self, error_code: i32, error_message: &str) {}
473}
474
475/// Builder for configuring and creating a persistent subscription.
476/// This provides a fluent interface for setting up a persistent subscription
477/// with proper CString handling.
478pub struct PersistentSubscriptionBuilder {
479    ctx: AeronArchivePersistentSubscriptionContext,
480    listener: Option<Box<dyn PersistentSubscriptionListener>>,
481}
482
483impl PersistentSubscriptionBuilder {
484    /// Create a new builder with a default context.
485    pub fn new() -> Result<Self, AeronCError> {
486        Ok(Self {
487            ctx: AeronArchivePersistentSubscriptionContext::new()?,
488            listener: None,
489        })
490    }
491
492    /// Set the Aeron client to use.
493    pub fn aeron(self, aeron: &Aeron) -> Result<Self, AeronCError> {
494        self.ctx.set_aeron(aeron)?;
495        Ok(self)
496    }
497
498    /// Set the archive context to use.
499    pub fn archive_context(self, ctx: &AeronArchiveContext) -> Result<Self, AeronCError> {
500        self.ctx.set_archive_context(ctx)?;
501        Ok(self)
502    }
503
504    /// Set the live channel (accepts &str, handles CString conversion internally).
505    pub fn live_channel(self, channel: &str) -> Result<Self, AeronCError> {
506        let channel = std::ffi::CString::new(channel).map_err(|_| AeronCError::from_code(-1))?;
507        self.ctx.set_live_channel(&channel)?;
508        Ok(self)
509    }
510
511    /// Set the live stream ID.
512    pub fn live_stream_id(self, id: i32) -> Result<Self, AeronCError> {
513        self.ctx.set_live_stream_id(id)?;
514        Ok(self)
515    }
516
517    /// Set the replay channel (accepts &str, handles CString conversion internally).
518    pub fn replay_channel(self, channel: &str) -> Result<Self, AeronCError> {
519        let channel = std::ffi::CString::new(channel).map_err(|_| AeronCError::from_code(-1))?;
520        self.ctx.set_replay_channel(&channel)?;
521        Ok(self)
522    }
523
524    /// Set the replay stream ID.
525    pub fn replay_stream_id(self, id: i32) -> Result<Self, AeronCError> {
526        self.ctx.set_replay_stream_id(id)?;
527        Ok(self)
528    }
529
530    /// Set the start position.
531    pub fn start_position(self, pos: i64) -> Result<Self, AeronCError> {
532        self.ctx.set_start_position(pos)?;
533        Ok(self)
534    }
535
536    /// Replay from the beginning of the recording (Aeron `FROM_START`).
537    pub fn start_from_beginning(self) -> Result<Self, AeronCError> {
538        self.start_position(PERSISTENT_SUBSCRIPTION_FROM_START)
539    }
540
541    /// Skip replay and join the live stream immediately (Aeron `FROM_LIVE`).
542    pub fn start_from_live(self) -> Result<Self, AeronCError> {
543        self.start_position(PERSISTENT_SUBSCRIPTION_FROM_LIVE)
544    }
545
546    /// Set the recording ID.
547    pub fn recording_id(self, id: i64) -> Result<Self, AeronCError> {
548        self.ctx.set_recording_id(id)?;
549        Ok(self)
550    }
551
552    /// Set the listener for events.
553    pub fn listener<L: PersistentSubscriptionListener>(mut self, listener: L) -> Result<Self, AeronCError> {
554        self.listener = Some(Box::new(listener));
555        Ok(self)
556    }
557
558    /// Pre-allocate the state counter so an external observer can read the PS state-machine
559    /// state. If unset the PS allocates one itself. Maps to Aeron's `Context.stateCounter`.
560    pub fn state_counter(self, counter: &AeronCounter) -> Result<Self, AeronCError> {
561        self.ctx.set_state_counter(counter)?;
562        Ok(self)
563    }
564
565    /// Counter holding the byte gap between replay and live when the live image is added.
566    /// Maps to Aeron's `Context.joinDifferenceCounter`.
567    pub fn join_difference_counter(self, counter: &AeronCounter) -> Result<Self, AeronCError> {
568        self.ctx.set_join_difference_counter(counter)?;
569        Ok(self)
570    }
571
572    /// Counter holding the number of times the PS has dropped off the live stream.
573    /// Maps to Aeron's `Context.liveLeftCounter`.
574    pub fn live_left_counter(self, counter: &AeronCounter) -> Result<Self, AeronCError> {
575        self.ctx.set_live_left_counter(counter)?;
576        Ok(self)
577    }
578
579    /// Counter holding the number of times the PS has switched to the live stream.
580    /// Maps to Aeron's `Context.liveJoinedCounter`.
581    pub fn live_joined_counter(self, counter: &AeronCounter) -> Result<Self, AeronCError> {
582        self.ctx.set_live_joined_counter(counter)?;
583        Ok(self)
584    }
585
586    /// Build the persistent subscription.
587    pub fn build(mut self) -> Result<AeronArchivePersistentSubscription, AeronCError> {
588        AeronArchivePersistentSubscription::create(self.ctx, self.listener.take())
589    }
590}
591
592#[cfg(test)]
593mod tests {
594    use super::*;
595    use log::{error, info};
596
597    use crate::testing::{valgrind_timeout, EmbeddedArchiveMediaDriverProcess};
598
599    #[test]
600    fn archive_error_parse_extracts_error_code() {
601        // exact shape emitted by aeron_archive_client.c
602        let msg = "(-11) generic error, see message\n[aeron_archive_poll_for_response, aeron_archive_client.c:2105] response for correlationId=32, errorCode=5, error: unknown recording id: 424242\n";
603        let err = AeronArchiveError::parse(msg);
604        assert_eq!(err.code, AeronArchiveErrorCode::UnknownRecording);
605        assert!(err.message.contains("unknown recording id"));
606
607        assert_eq!(
608            AeronArchiveError::parse("errorCode=11, error: no space").code,
609            AeronArchiveErrorCode::StorageSpace
610        );
611        assert!(AeronArchiveError::parse("errorCode=11, x").code.is_resource_exhausted());
612        assert_eq!(
613            AeronArchiveError::parse("errorCode=99, ?").code,
614            AeronArchiveErrorCode::Unknown(99)
615        );
616        // no code present -> Generic
617        assert_eq!(
618            AeronArchiveError::parse("subscription to archive is not connected").code,
619            AeronArchiveErrorCode::Generic
620        );
621    }
622
623    #[test]
624    fn archive_error_codes_round_trip() {
625        for code in 0..=16 {
626            let parsed = AeronArchiveErrorCode::from_code(code);
627            assert_ne!(
628                parsed,
629                AeronArchiveErrorCode::Unknown(code),
630                "code {code} must map to a named variant"
631            );
632        }
633    }
634
635    /// Pins the enum to the aeron C header: every `ARCHIVE_ERROR_CODE_*` constant the
636    /// submodule defines must map to a named variant. When `just update-aeron-version`
637    /// pulls a release that adds or renumbers codes, this fails and points at the gap.
638    #[test]
639    fn archive_error_codes_match_the_c_header() {
640        let header = std::fs::read_to_string(concat!(
641            env!("CARGO_MANIFEST_DIR"),
642            "/aeron/aeron-archive/src/main/c/client/aeron_archive.h"
643        ))
644        .expect("aeron submodule header missing");
645        let mut found = 0;
646        for line in header.lines() {
647            let Some(rest) = line.trim().strip_prefix("#define ARCHIVE_ERROR_CODE_") else {
648                continue;
649            };
650            let mut parts = rest.split_whitespace();
651            let name = parts.next().unwrap_or_default();
652            let value: i32 = parts
653                .next()
654                .unwrap_or_default()
655                .trim_matches(|c| c == '(' || c == ')')
656                .parse()
657                .unwrap_or_else(|_| panic!("unparseable value for ARCHIVE_ERROR_CODE_{name}"));
658            let parsed = AeronArchiveErrorCode::from_code(value);
659            assert_ne!(
660                parsed,
661                AeronArchiveErrorCode::Unknown(value),
662                "C header defines ARCHIVE_ERROR_CODE_{name} = {value} but AeronArchiveErrorCode has no variant for it"
663            );
664            found += 1;
665        }
666        assert!(
667            found >= 14,
668            "expected at least 14 ARCHIVE_ERROR_CODE_* defines, found {found}"
669        );
670    }
671    use serial_test::serial;
672    use std::cell::Cell;
673    use std::error;
674    use std::error::Error;
675    use std::os::raw::c_int;
676    use std::str::FromStr;
677    use std::sync::atomic::{AtomicBool, Ordering};
678    use std::sync::Arc;
679    use std::thread::{sleep, JoinHandle};
680    use std::time::{Duration, Instant};
681
682    #[derive(Default, Debug)]
683    struct ErrorCount {
684        error_count: usize,
685    }
686
687    impl AeronErrorHandlerCallback for ErrorCount {
688        fn handle_aeron_error_handler(&mut self, error_code: c_int, msg: &str) {
689            error!("Aeron error {}: {}", error_code, msg);
690            self.error_count += 1;
691        }
692    }
693
694    pub const ARCHIVE_CONTROL_REQUEST: &str = "aeron:udp?endpoint=localhost:8010";
695    pub const ARCHIVE_CONTROL_RESPONSE: &str = "aeron:udp?endpoint=localhost:8011";
696    pub const ARCHIVE_RECORDING_EVENTS: &str = "aeron:udp?control-mode=dynamic|control=localhost:8012";
697
698    #[test]
699    fn test_uri_string_builder() -> Result<(), AeronCError> {
700        let builder = AeronUriStringBuilder::default();
701
702        builder.init_new()?;
703        builder
704            .media(Media::Udp)? // very important to set media else set_initial_position will give an error of -1
705            .mtu_length(1024 * 64)?
706            .set_initial_position(127424949617280, 1182294755, 65536)?;
707        let uri = builder.build(1024)?;
708        assert_eq!(
709            "aeron:udp?term-id=-1168322114|term-length=65536|mtu=65536|init-term-id=1182294755|term-offset=33408",
710            uri
711        );
712
713        builder.init_new()?;
714        let uri = builder
715            .media(Media::Udp)?
716            .control_mode(ControlMode::Dynamic)?
717            .reliable(false)?
718            .ttl(2)?
719            .endpoint("localhost:1235")?
720            .control("localhost:1234")?
721            .build(1024)?;
722        assert_eq!(
723            "aeron:udp?ttl=2|control-mode=dynamic|endpoint=localhost:1235|control=localhost:1234|reliable=false",
724            uri
725        );
726
727        let uri = AeronUriStringBuilder::from_str("aeron:udp?endpoint=localhost:8010")?
728            .ttl(5)?
729            .build(1024)?;
730
731        assert_eq!("aeron:udp?ttl=5|endpoint=localhost:8010", uri);
732
733        let uri = uri.parse::<AeronUriStringBuilder>()?.ttl(6)?.build(1024)?;
734
735        assert_eq!("aeron:udp?ttl=6|endpoint=localhost:8010", uri);
736
737        Ok(())
738    }
739
740    pub const STREAM_ID: i32 = 1033;
741    pub const MESSAGE_PREFIX: &str = "Message-Prefix-";
742    pub const CONTROL_ENDPOINT: &str = "localhost:23265";
743    pub const RECORDING_ENDPOINT: &str = "localhost:23266";
744    pub const LIVE_ENDPOINT: &str = "localhost:23267";
745    pub const REPLAY_ENDPOINT: &str = "localhost:0";
746    // pub const REPLAY_ENDPOINT: &str = "localhost:23268";
747
748    #[test]
749    #[serial]
750    fn test_simple_replay_merge() -> Result<(), AeronCError> {
751        // Skip test under Valgrind due to timeout issues
752        if std::env::var_os("RUSTERON_VALGRIND").is_some() {
753            return Ok(());
754        }
755
756        rusteron_code_gen::test_logger::init(log::LevelFilter::Info);
757
758        EmbeddedArchiveMediaDriverProcess::kill_all_java_processes().expect("failed to kill all java processes");
759
760        assert!(is_udp_port_available(23265));
761        assert!(is_udp_port_available(23266));
762        assert!(is_udp_port_available(23267));
763        assert!(is_udp_port_available(23268));
764        let id = Aeron::nano_clock();
765        let aeron_dir = format!("target/aeron/{}/shm", id);
766        let archive_dir = format!("target/aeron/{}/archive", id);
767
768        info!("starting archive media driver");
769        let media_driver = EmbeddedArchiveMediaDriverProcess::build_and_start(
770            &aeron_dir,
771            &format!("{}/archive", aeron_dir),
772            ARCHIVE_CONTROL_REQUEST,
773            ARCHIVE_CONTROL_RESPONSE,
774            ARCHIVE_RECORDING_EVENTS,
775        )
776        .expect("Failed to start embedded media driver");
777
778        info!("connecting to archive");
779        let (archive, aeron) = media_driver
780            .archive_connect()
781            .expect("Could not connect to archive client");
782
783        let running = Arc::new(AtomicBool::new(true));
784
785        info!("connected to archive, adding publication");
786        assert!(!aeron.is_closed());
787
788        let (session_id, publisher_thread) = reply_merge_publisher(&archive, aeron.clone(), running.clone())?;
789
790        {
791            let context = AeronContext::new()?;
792            context.set_dir(&media_driver.aeron_dir)?;
793            let error_handler = Handler::new(ErrorCount::default());
794            context.set_error_handler(Some(error_handler.clone()))?;
795            context.set_driver_timeout_ms(60_000)?;
796
797            // Wrap fallible code so teardown ordering holds even on error/panic
798            let inner: Result<(), AeronCError> = (|| {
799                let aeron = Aeron::new(&context)?;
800                aeron.start()?;
801                let source_archive_context = archive.get_archive_context();
802                let aeron_archive_context = AeronArchiveContext::new()?;
803                aeron_archive_context.set_aeron(&aeron)?;
804                aeron_archive_context.set_control_request_channel(
805                    &source_archive_context.get_control_request_channel().into_c_string(),
806                )?;
807                aeron_archive_context.set_control_response_channel(
808                    &source_archive_context.get_control_response_channel().into_c_string(),
809                )?;
810                aeron_archive_context.set_recording_events_channel(
811                    &source_archive_context.get_recording_events_channel().into_c_string(),
812                )?;
813                aeron_archive_context.set_message_timeout_ns(60_000_000_000)?;
814                aeron_archive_context.set_error_handler(Some(error_handler.clone()))?;
815                let merge_archive = AeronArchiveAsyncConnect::new_with_aeron(&aeron_archive_context, &aeron)?
816                    .poll_blocking(Duration::from_secs(60))?;
817                replay_merge_subscription(&merge_archive, aeron.clone(), session_id)?;
818                Ok(())
819            })();
820
821            inner?;
822        }
823
824        running.store(false, Ordering::Release);
825        publisher_thread.join().unwrap();
826        drop(media_driver);
827
828        Ok(())
829    }
830
831    fn reply_merge_publisher(
832        archive: &AeronArchive,
833        aeron: Aeron,
834        running: Arc<AtomicBool>,
835    ) -> Result<(i32, JoinHandle<()>), AeronCError> {
836        let publication = aeron.add_publication(
837            // &format!("aeron:udp?control={CONTROL_ENDPOINT}|control-mode=dynamic|term-length=65536|fc=tagged,g:99901/1,t:5s"),
838            &format!("aeron:udp?control={CONTROL_ENDPOINT}|control-mode=dynamic|term-length=65536").into_c_string(),
839            STREAM_ID,
840            Duration::from_secs(5),
841        )?;
842
843        info!(
844            "publication {} [status={:?}]",
845            publication.channel(),
846            publication.channel_status()
847        );
848        assert_eq!(1, publication.channel_status());
849
850        let session_id = publication.session_id();
851        let recording_channel = format!(
852            // "aeron:udp?endpoint={RECORDING_ENDPOINT}|control={CONTROL_ENDPOINT}|session-id={session_id}|gtag=99901"
853            "aeron:udp?endpoint={RECORDING_ENDPOINT}|control={CONTROL_ENDPOINT}|session-id={session_id}"
854        );
855        info!("recording channel {}", recording_channel);
856        archive.start_recording(
857            &recording_channel.into_c_string(),
858            STREAM_ID,
859            SOURCE_LOCATION_REMOTE,
860            true,
861        )?;
862
863        info!("waiting for publisher to be connected");
864        while !publication.is_connected() {
865            thread::sleep(Duration::from_millis(100));
866        }
867        info!("publisher to be connected");
868        let counters_reader = aeron.counters_reader();
869        let mut caught_up_count = 0;
870        let publisher_thread = thread::spawn(move || {
871            let mut message_count = 0;
872
873            while running.load(Ordering::Acquire) {
874                let message = format!("{}{}", MESSAGE_PREFIX, message_count);
875                while publication.offer_raw(message.as_bytes(), Handlers::NONE) <= 0 {
876                    thread::sleep(Duration::from_millis(10));
877                }
878                message_count += 1;
879                if message_count % 10_000 == 0 {
880                    info!(
881                        "Published {} messages [position={}]",
882                        message_count,
883                        publication.position()
884                    );
885                }
886                // slow down publishing so can catch up
887                if message_count > 10_000 {
888                    // ensure archiver is caught up
889                    while !publication.is_archive_position_with(0) {
890                        thread::sleep(Duration::from_micros(300));
891                    }
892                    caught_up_count += 1;
893                }
894            }
895            assert!(caught_up_count > 0);
896            if let Err(err) = publication.close() {
897                info!("publisher close returned error: {err:?}");
898            }
899            info!("Publisher thread terminated");
900        });
901        Ok((session_id, publisher_thread))
902    }
903
904    fn replay_merge_subscription(archive: &AeronArchive, aeron: Aeron, session_id: i32) -> Result<(), AeronCError> {
905        // let replay_channel = format!("aeron:udp?control-mode=manual|session-id={session_id}");
906        let replay_channel = format!("aeron:udp?session-id={session_id}").into_c_string();
907        info!("replay channel {:?}", replay_channel);
908
909        let replay_destination = format!("aeron:udp?endpoint={REPLAY_ENDPOINT}").into_c_string();
910        info!("replay destination {:?}", replay_destination);
911
912        let live_destination = format!("aeron:udp?endpoint={LIVE_ENDPOINT}|control={CONTROL_ENDPOINT}").into_c_string();
913        info!("live destination {:?}", live_destination);
914
915        let counters_reader = aeron.counters_reader();
916        let mut counter_id = -1;
917
918        while counter_id < 0 {
919            counter_id = RecordingPos::find_counter_id_by_session(&counters_reader, session_id);
920        }
921        info!(
922            "counter id {} {:?}",
923            counter_id,
924            counters_reader.get_counter_label(counter_id, 1024)
925        );
926        info!(
927            "counter id {} position={:?}",
928            counter_id,
929            counters_reader.get_counter_value(counter_id)
930        );
931
932        // let recording_id = Cell::new(-1);
933        // let start_position = Cell::new(-1);
934
935        // let mut count = 0;
936        // assert!(
937        //     archive.list_recordings_fn(&mut count, 0, 1000, |descriptor| {
938        //         info!("Recording descriptor: {:?}", descriptor);
939        //         recording_id.set(descriptor.recording_id);
940        //         start_position.set(descriptor.start_position);
941        //         assert_eq!(descriptor.session_id, session_id);
942        //         assert_eq!(descriptor.stream_id, STREAM_ID);
943        //     })? >= 0
944        // );
945        // assert!(count > 0);
946        // assert!(recording_id.get() >= 0);
947
948        // let record_id = RecordingPos::get_recording_id(&aeron.counters_reader(), counter_id)?;
949        // assert_eq!(recording_id.get(), record_id);
950        //
951        // let recording_id = recording_id.get();
952        // let start_position = start_position.get();
953        let start_position = 0;
954        let recording_id =
955            RecordingPos::get_recording_id_block(&aeron.counters_reader(), counter_id, Duration::from_secs(5))?;
956
957        let subscribe_channel = format!("aeron:udp?control-mode=manual|session-id={session_id}").into_c_string();
958        info!("subscribe channel {:?}", subscribe_channel);
959        let subscription = aeron.add_subscription(
960            &subscribe_channel,
961            STREAM_ID,
962            Handlers::NONE,
963            Handlers::NONE,
964            Duration::from_secs(5),
965        )?;
966
967        let replay_merge = AeronArchiveReplayMerge::new(
968            &subscription,
969            &archive,
970            &replay_channel,
971            &replay_destination,
972            &live_destination,
973            recording_id,
974            start_position,
975            Aeron::epoch_clock(),
976            60_000,
977        )?;
978
979        info!(
980            "ReplayMerge initialization: recordingId={}, startPosition={}, subscriptionChannel={:?}, replayChannel={:?}, replayDestination={:?}, liveDestination={:?}",
981            recording_id,
982            start_position,
983            subscribe_channel,
984            &replay_channel,
985            &replay_destination,
986            &live_destination
987        );
988
989        // media_driver
990        //     .run_aeron_stats()
991        //     .expect("Failed to run aeron stats");
992
993        // info!("Waiting for subscription to connect...");
994        // while !subscription.is_connected() {
995        //     thread::sleep(Duration::from_millis(100));
996        // }
997        // info!("Subscription connected");
998
999        info!(
1000            "about to start_replay [maxRecordPosition={:?}]",
1001            archive.get_max_recorded_position(recording_id)
1002        );
1003
1004        let mut reply_count = 0;
1005        while !replay_merge.is_merged() {
1006            assert!(!replay_merge.has_failed());
1007            if replay_merge.poll_fn(
1008                |buffer, _header| {
1009                    reply_count += 1;
1010                    if reply_count % 10_000 == 0 {
1011                        info!(
1012                            "replay-merge [count={}, isMerged={}, isLive={}]",
1013                            reply_count,
1014                            replay_merge.is_merged(),
1015                            replay_merge.is_live_added()
1016                        );
1017                    }
1018                },
1019                100,
1020            )? == 0
1021            {
1022                let err = archive.poll_for_error_response_as_string(4096)?;
1023                if !err.is_empty() {
1024                    panic!("{}", err);
1025                }
1026                if Aeron::errmsg().len() > 0 && "no error" != Aeron::errmsg() {
1027                    panic!("{}", Aeron::errmsg());
1028                }
1029                thread::sleep(Duration::from_millis(100));
1030            }
1031        }
1032        assert!(!replay_merge.has_failed());
1033        assert!(replay_merge.is_live_added());
1034        assert!(reply_count > 10_000, "no replay-merge fragments received");
1035        Ok(())
1036    }
1037
1038    #[test]
1039    fn version_check() {
1040        let major = unsafe { crate::aeron_version_major() };
1041        let minor = unsafe { crate::aeron_version_minor() };
1042        let patch = unsafe { crate::aeron_version_patch() };
1043
1044        let aeron_version = format!("{}.{}.{}", major, minor, patch);
1045
1046        let cargo_version = "1.52.0";
1047        assert_eq!(aeron_version, cargo_version);
1048    }
1049
1050    use std::thread;
1051
1052    fn start_aeron_archive() -> Result<
1053        (
1054            Aeron,
1055            AeronArchiveContext,
1056            EmbeddedArchiveMediaDriverProcess,
1057            Handler<AeronPublicationErrorFrameHandlerLogger>,
1058            Handler<ErrorCount>,
1059        ),
1060        Box<dyn Error>,
1061    > {
1062        let id = Aeron::nano_clock();
1063        let aeron_dir = format!("target/aeron/{}/shm", id);
1064        let archive_dir = format!("target/aeron/{}/archive", id);
1065
1066        let request_port = find_unused_udp_port(8000).expect("Could not find port");
1067        let response_port = find_unused_udp_port(request_port + 1).expect("Could not find port");
1068        let recording_event_port = find_unused_udp_port(response_port + 1).expect("Could not find port");
1069        let request_control_channel = &format!("aeron:udp?endpoint=localhost:{}", request_port);
1070        let response_control_channel = &format!("aeron:udp?endpoint=localhost:{}", response_port);
1071        let recording_events_channel = &format!("aeron:udp?endpoint=localhost:{}", recording_event_port);
1072        assert_ne!(request_control_channel, response_control_channel);
1073
1074        let archive_media_driver = EmbeddedArchiveMediaDriverProcess::build_and_start(
1075            &aeron_dir,
1076            &archive_dir,
1077            request_control_channel,
1078            response_control_channel,
1079            recording_events_channel,
1080        )
1081        .expect("Failed to start Java process");
1082
1083        let aeron_context = AeronContext::new()?;
1084        aeron_context.set_dir(&aeron_dir.into_c_string())?;
1085        aeron_context.set_client_name(c"test")?;
1086        let pub_error_frame_handler = Handler::new(AeronPublicationErrorFrameHandlerLogger);
1087        aeron_context.set_publication_error_frame_handler(Some(pub_error_frame_handler.clone()))?;
1088        let error_handler = Handler::new(ErrorCount::default());
1089        aeron_context.set_error_handler(Some(error_handler.clone()))?;
1090
1091        // Use inner closure so teardown ordering holds on any error path after handlers are created
1092        let inner: Result<(Aeron, AeronArchiveContext), Box<dyn Error>> = (|| {
1093            let aeron = Aeron::new(&aeron_context)?;
1094            aeron.start()?;
1095            let archive_context = AeronArchiveContext::new()?;
1096            archive_context.set_aeron(&aeron)?;
1097            archive_context.set_control_request_channel(&request_control_channel.as_str().into_c_string())?;
1098            archive_context.set_control_response_channel(&response_control_channel.as_str().into_c_string())?;
1099            archive_context.set_recording_events_channel(&recording_events_channel.as_str().into_c_string())?;
1100            archive_context.set_error_handler(Some(error_handler.clone()))?;
1101            Ok((aeron, archive_context))
1102        })();
1103
1104        match inner {
1105            Ok((aeron, archive_context)) => Ok((
1106                aeron,
1107                archive_context,
1108                archive_media_driver,
1109                pub_error_frame_handler,
1110                error_handler,
1111            )),
1112            Err(e) => Err(e),
1113        }
1114    }
1115
1116    /// Deep-graph close: closing a *clone* of the archive client defers the C close —
1117    /// the original stays fully usable (control session intact, error polling clean).
1118    #[test]
1119    #[serial]
1120    pub fn archive_clone_close_defers_and_original_remains_usable() -> Result<(), Box<dyn error::Error>> {
1121        rusteron_code_gen::test_logger::init(log::LevelFilter::Info);
1122        EmbeddedArchiveMediaDriverProcess::kill_all_java_processes().expect("failed to kill all java processes");
1123
1124        let (aeron, archive_context, media_driver, pub_error_frame_handler, error_handler) = start_aeron_archive()?;
1125
1126        let test_result: Result<(), Box<dyn error::Error>> = (|| {
1127            let archive = AeronArchiveAsyncConnect::new_with_aeron(&archive_context.clone(), &aeron)?
1128                .poll_blocking(Duration::from_secs(30))
1129                .expect("failed to connect to aeron archive media driver");
1130
1131            let session_id = archive.control_session_id();
1132            let clone = archive.clone();
1133            assert!(clone.close().is_ok());
1134
1135            // original remains fully usable after the clone's close
1136            assert_eq!(session_id, archive.control_session_id());
1137            assert!(archive.poll_for_error()?.is_none());
1138            let subscription_id = archive.start_recording(AERON_IPC_STREAM, 42, SOURCE_LOCATION_LOCAL, true)?;
1139            assert!(subscription_id >= 0);
1140            archive.stop_recording_subscription(subscription_id)?;
1141
1142            assert!(archive.close().is_ok());
1143            Ok(())
1144        })();
1145
1146        drop(aeron);
1147        drop(archive_context);
1148        drop(media_driver);
1149        drop(pub_error_frame_handler);
1150        drop(error_handler);
1151        test_result
1152    }
1153
1154    #[test]
1155    #[serial]
1156    pub fn test_aeron_archive() -> Result<(), Box<dyn error::Error>> {
1157        rusteron_code_gen::test_logger::init(log::LevelFilter::Info);
1158        EmbeddedArchiveMediaDriverProcess::kill_all_java_processes().expect("failed to kill all java processes");
1159
1160        let (aeron, archive_context, media_driver, pub_error_frame_handler, error_handler) = start_aeron_archive()?;
1161
1162        let test_result: Result<(), Box<dyn error::Error>> = (|| {
1163            assert!(!aeron.is_closed());
1164
1165            info!("connected to aeron");
1166
1167            let archive_connector = AeronArchiveAsyncConnect::new_with_aeron(&archive_context.clone(), &aeron)?;
1168            let archive = archive_connector
1169                .poll_blocking(Duration::from_secs(30))
1170                .expect("failed to connect to aeron archive media driver");
1171
1172            assert!(archive.get_archive_id() > 0);
1173
1174            let channel = AERON_IPC_STREAM;
1175            let stream_id = 10;
1176
1177            let subscription_id = archive.start_recording(channel, stream_id, SOURCE_LOCATION_LOCAL, true)?;
1178
1179            assert!(subscription_id >= 0);
1180            info!("subscription id {}", subscription_id);
1181
1182            let publication = aeron
1183                .async_add_exclusive_publication(channel, stream_id)?
1184                .poll_blocking(Duration::from_secs(5))?;
1185
1186            for i in 0..11 {
1187                while publication.offer_raw("123456".as_bytes(), Handlers::NONE) <= 0 {
1188                    sleep(Duration::from_millis(50));
1189                    let err = archive.poll_for_error_response_as_string(4096)?;
1190                    if !err.is_empty() {
1191                        return Err(std::io::Error::other(err).into());
1192                    }
1193                    archive.idle();
1194                }
1195                info!("sent message {i} [test_aeron_archive]");
1196            }
1197
1198            archive.idle();
1199            let session_id = publication.get_constants()?.session_id;
1200            info!("publication session id {}", session_id);
1201            // since this is single threaded need to make sure it did write to archiver, usually not required in multi-proccess app
1202            let stop_position = publication.position();
1203            info!(
1204                "publication stop position {} [publication={:?}]",
1205                stop_position,
1206                publication.get_constants()
1207            );
1208            let counters_reader = aeron.counters_reader();
1209            info!("counters reader ready {:?}", counters_reader);
1210
1211            let mut counter_id = -1;
1212
1213            let start = Instant::now();
1214            while counter_id <= 0 && start.elapsed() < Duration::from_secs(5) {
1215                counter_id = RecordingPos::find_counter_id_by_session(&counters_reader, session_id);
1216                info!("counter id {}", counter_id);
1217            }
1218
1219            assert!(counter_id >= 0);
1220
1221            info!("counter id {counter_id}, session id {session_id}");
1222            while counters_reader.get_counter_value(counter_id) < stop_position {
1223                info!(
1224                    "current archive publication stop position {}",
1225                    counters_reader.get_counter_value(counter_id)
1226                );
1227                sleep(Duration::from_millis(50));
1228            }
1229            info!(
1230                "found archive publication stop position {}",
1231                counters_reader.get_counter_value(counter_id)
1232            );
1233
1234            archive.stop_recording_channel_and_stream(channel, stream_id)?;
1235            drop(publication);
1236
1237            info!("list recordings");
1238            let found_recording_id = Cell::new(-1);
1239            let start_pos = Cell::new(-1);
1240            let end_pos = Cell::new(-1);
1241            let start = Instant::now();
1242            while start.elapsed() < Duration::from_secs(5) && found_recording_id.get() == -1 {
1243                let mut count = 0;
1244                archive.list_recordings_for_uri_fn(
1245                    &mut count,
1246                    0,
1247                    i32::MAX,
1248                    channel,
1249                    stream_id,
1250                    |d: AeronArchiveRecordingDescriptor| {
1251                        assert_eq!(d.stream_id, stream_id);
1252                        info!("found recording {:#?}", d);
1253                        info!(
1254                            "strippedChannel={}, originalChannel={}",
1255                            d.stripped_channel(),
1256                            d.original_channel()
1257                        );
1258                        if d.stop_position > d.start_position && d.stop_position > 0 {
1259                            found_recording_id.set(d.recording_id);
1260                            start_pos.set(d.start_position);
1261                            end_pos.set(d.stop_position);
1262                        }
1263
1264                        // verify clone_struct works
1265                        let copy = d.clone_struct();
1266                        assert_eq!(copy.deref(), d.deref());
1267                        assert_eq!(copy.recording_id, d.recording_id);
1268                        assert_eq!(copy.control_session_id, d.control_session_id);
1269                        assert_eq!(copy.mtu_length, d.mtu_length);
1270                        assert_eq!(copy.source_identity_length, d.source_identity_length);
1271                    },
1272                )?;
1273                let err = archive.poll_for_error_response_as_string(4096)?;
1274                if !err.is_empty() {
1275                    return Err(std::io::Error::other(err).into());
1276                }
1277            }
1278            assert!(start.elapsed() < Duration::from_secs(5));
1279            info!("start replay");
1280            let params =
1281                AeronArchiveReplayParams::new(0, i32::MAX, start_pos.get(), end_pos.get() - start_pos.get(), 0, 0)?;
1282            info!("replay params {:#?}", params);
1283            let replay_stream_id = 45;
1284            let replay_session_id =
1285                archive.start_replay(found_recording_id.get(), channel, replay_stream_id, &params)?;
1286            let session_id = replay_session_id as i32;
1287
1288            info!("replay session id {}", replay_session_id);
1289            info!("session id {}", session_id);
1290            let channel_replay = format!("{}?session-id={}", channel.to_str().unwrap(), session_id).into_c_string();
1291            info!("archive id: {}", archive.get_archive_id());
1292
1293            info!("add subscription {:?}", channel_replay);
1294            let avail_image_handler = Handler::new(AeronAvailableImageLogger);
1295            let unavail_image_handler = Handler::new(AeronUnavailableImageLogger);
1296            let replay_result: Result<(), Box<dyn error::Error>> = (|| {
1297                let subscription = aeron
1298                    .async_add_subscription(
1299                        &channel_replay,
1300                        replay_stream_id,
1301                        Some(&avail_image_handler),
1302                        Some(&unavail_image_handler),
1303                    )?
1304                    .poll_blocking(Duration::from_secs(10))?;
1305
1306                #[derive(Default)]
1307                struct FragmentHandler {
1308                    count: Cell<usize>,
1309                }
1310
1311                impl AeronFragmentHandlerCallback for FragmentHandler {
1312                    fn handle_aeron_fragment_handler(&mut self, buffer: &[u8], _header: AeronHeader) {
1313                        assert_eq!(buffer, "123456".as_bytes());
1314
1315                        // Update count (using Cell for interior mutability)
1316                        self.count.set(self.count.get() + 1);
1317                    }
1318                }
1319
1320                let poll = Handler::new(FragmentHandler::default());
1321                let poll_result: Result<(), Box<dyn error::Error>> = (|| {
1322                    let wait_timeout = Duration::from_secs(30);
1323                    let start = Instant::now();
1324                    while start.elapsed() < wait_timeout && subscription.poll(Some(&poll), 100)? <= 0 {
1325                        let err = archive.poll_for_error_response_as_string(4096)?;
1326                        if !err.is_empty() {
1327                            return Err(std::io::Error::other(err).into());
1328                        }
1329                    }
1330
1331                    if start.elapsed() >= wait_timeout {
1332                        return Err(std::io::Error::other(format!("messages not received {:?}", poll.count)).into());
1333                    }
1334
1335                    info!("aeron {:?}", aeron);
1336                    info!("ctx {:?}", archive_context);
1337                    if poll.count.get() != 11 {
1338                        return Err(std::io::Error::other(format!(
1339                            "expected 11 replayed messages, got {}",
1340                            poll.count.get()
1341                        ))
1342                        .into());
1343                    }
1344                    Ok(())
1345                })();
1346
1347                drop(subscription);
1348                poll_result
1349            })();
1350
1351            replay_result?;
1352            Ok(())
1353        })();
1354
1355        drop(aeron);
1356        drop(media_driver);
1357        test_result
1358    }
1359
1360    #[test]
1361    #[serial]
1362    fn test_invalid_recording_channel() -> Result<(), Box<dyn Error>> {
1363        let (aeron, archive_context, media_driver, pub_error_frame_handler, error_handler) = start_aeron_archive()?;
1364        let archive_connector = AeronArchiveAsyncConnect::new_with_aeron(&archive_context.clone(), &aeron)?;
1365        let archive = archive_connector
1366            .poll_blocking(Duration::from_secs(30))
1367            .expect("failed to connect to archive");
1368
1369        let invalid_channel = c"invalid:channel";
1370        let result = archive.start_recording(&invalid_channel, STREAM_ID, SOURCE_LOCATION_LOCAL, true);
1371        assert!(
1372            result.is_err(),
1373            "Expected error when starting recording with an invalid channel"
1374        );
1375        drop(media_driver);
1376        Ok(())
1377    }
1378
1379    #[test]
1380    #[serial]
1381    fn test_stop_recording_on_nonexistent_channel() -> Result<(), Box<dyn Error>> {
1382        let (aeron, archive_context, media_driver, pub_error_frame_handler, error_handler) = start_aeron_archive()?;
1383        let archive_connector = AeronArchiveAsyncConnect::new_with_aeron(&archive_context.clone(), &aeron)?;
1384        let archive = archive_connector
1385            .poll_blocking(Duration::from_secs(30))
1386            .expect("failed to connect to archive");
1387
1388        let nonexistent_channel = c"aeron:udp?endpoint=localhost:9999";
1389        let result = archive.stop_recording_channel_and_stream(nonexistent_channel, STREAM_ID);
1390        assert!(
1391            result.is_err(),
1392            "Expected error when stopping recording on a non-existent channel"
1393        );
1394        drop(media_driver);
1395        Ok(())
1396    }
1397
1398    #[test]
1399    #[serial]
1400    fn test_replay_with_invalid_recording_id() -> Result<(), Box<dyn Error>> {
1401        let (aeron, archive_context, media_driver, pub_error_frame_handler, error_handler) = start_aeron_archive()?;
1402        let archive_connector = AeronArchiveAsyncConnect::new_with_aeron(&archive_context.clone(), &aeron)?;
1403        let archive = archive_connector
1404            .poll_blocking(Duration::from_secs(30))
1405            .expect("failed to connect to archive");
1406
1407        let invalid_recording_id = -999;
1408        let params = AeronArchiveReplayParams::new(0, i32::MAX, 0, 100, 0, 0)?;
1409        let result = archive.start_replay(
1410            invalid_recording_id,
1411            c"aeron:udp?endpoint=localhost:8888",
1412            STREAM_ID,
1413            &params,
1414        );
1415        assert!(
1416            result.is_err(),
1417            "Expected error when starting replay with an invalid recording id"
1418        );
1419        drop(media_driver);
1420        Ok(())
1421    }
1422
1423    #[test]
1424    #[serial]
1425    fn test_archive_reconnect_after_close() -> Result<(), Box<dyn std::error::Error>> {
1426        let (aeron, archive_context, media_driver, pub_error_frame_handler, error_handler) = start_aeron_archive()?;
1427        let archive_connector = AeronArchiveAsyncConnect::new_with_aeron(&archive_context.clone(), &aeron)?;
1428        let archive = archive_connector
1429            .poll_blocking(Duration::from_secs(30))
1430            .expect("failed to connect to archive");
1431
1432        archive.close()?;
1433
1434        // Retry reconnection with exponential backoff to handle race condition
1435        // where close() is async and the endpoint may still be CLOSING
1436        let mut retry_delay = Duration::from_millis(100);
1437        let max_retries = 10;
1438        let mut new_archive = None;
1439
1440        for attempt in 0..max_retries {
1441            let archive_connector = AeronArchiveAsyncConnect::new_with_aeron(&archive_context, &aeron)?;
1442            match archive_connector.poll_blocking(valgrind_timeout(5)) {
1443                Ok(archive) => {
1444                    new_archive = Some(archive);
1445                    break;
1446                }
1447                Err(e) if attempt < max_retries - 1 => {
1448                    // Check if error is about CLOSING state - retry with backoff
1449                    let error_msg = e.to_string();
1450                    if error_msg.contains("CLOSING") || error_msg.contains("temporarily unavailable") {
1451                        std::thread::sleep(retry_delay);
1452                        retry_delay = retry_delay.saturating_mul(2);
1453                        continue;
1454                    }
1455                    // Other errors should fail immediately
1456                    return Err(format!("Failed to reconnect to archive: {}", e).into());
1457                }
1458                Err(e) => {
1459                    return Err(format!("Failed to reconnect to archive after {} retries: {}", max_retries, e).into());
1460                }
1461            }
1462        }
1463
1464        let new_archive = new_archive.expect("failed to reconnect to archive after retries");
1465        assert!(
1466            new_archive.get_archive_id() > 0,
1467            "Reconnected archive should have a valid archive id"
1468        );
1469
1470        drop(media_driver);
1471        Ok(())
1472    }
1473
1474    #[test]
1475    #[serial]
1476    fn test_archive_close_defers_with_live_clone() -> Result<(), Box<dyn std::error::Error>> {
1477        let (aeron, archive_context, media_driver, pub_error_frame_handler, error_handler) = start_aeron_archive()?;
1478        let archive_connector = AeronArchiveAsyncConnect::new_with_aeron(&archive_context, &aeron)?;
1479        let archive = archive_connector
1480            .poll_blocking(Duration::from_secs(30))
1481            .expect("failed to connect to archive");
1482        let archive_id = archive.get_archive_id();
1483        assert!(archive_id > 0);
1484
1485        // Ordering 1: close clone first, original stays alive
1486        let clone = archive.clone();
1487        clone.close()?;
1488        assert_eq!(
1489            archive_id,
1490            archive.get_archive_id(),
1491            "clone close defers while original alive"
1492        );
1493
1494        // Ordering 2: close original first, clone stays alive
1495        let clone2 = archive.clone();
1496        archive.close()?;
1497        assert_eq!(
1498            archive_id,
1499            clone2.get_archive_id(),
1500            "original close defers while clone alive"
1501        );
1502
1503        drop(media_driver);
1504        Ok(())
1505    }
1506
1507    #[test]
1508    #[serial]
1509    fn test_archive_close_does_not_close_aeron_client() -> Result<(), Box<dyn std::error::Error>> {
1510        let (aeron, archive_context, media_driver, pub_error_frame_handler, error_handler) = start_aeron_archive()?;
1511        let archive_connector = AeronArchiveAsyncConnect::new_with_aeron(&archive_context, &aeron)?;
1512        let archive = archive_connector
1513            .poll_blocking(Duration::from_secs(30))
1514            .expect("failed to connect to archive");
1515
1516        archive.close()?;
1517
1518        let publication = aeron
1519            .add_publication(AERON_IPC_STREAM, 30, Duration::from_secs(5))
1520            .expect("Aeron client should remain usable after archive close");
1521        assert!(!publication.get_inner().is_null());
1522        publication.close()?;
1523
1524        drop(media_driver);
1525        Ok(())
1526    }
1527
1528    #[test]
1529    #[serial]
1530    fn close_now_after_all_archive_children_dropped_is_safe() -> Result<(), Box<dyn std::error::Error>> {
1531        rusteron_code_gen::test_logger::init(log::LevelFilter::Info);
1532        let (aeron, archive_context, media_driver, _pub_error_frame_handler, error_handler) = start_aeron_archive()?;
1533
1534        // Build a complex object graph: Aeron subscriptions, archive recording
1535        let publication = aeron
1536            .add_publication(AERON_IPC_STREAM, 30, Duration::from_secs(5))
1537            .expect("add publication");
1538
1539        let subscription = aeron
1540            .add_subscription(
1541                AERON_IPC_STREAM,
1542                30,
1543                Handlers::NONE,
1544                Handlers::NONE,
1545                Duration::from_secs(5),
1546            )
1547            .expect("add subscription");
1548
1549        let archive = AeronArchiveAsyncConnect::new_with_aeron(&archive_context, &aeron)?
1550            .poll_blocking(Duration::from_secs(30))
1551            .expect("failed to connect to archive");
1552
1553        let recording_id = archive
1554            .start_recording(AERON_IPC_STREAM, 30, SOURCE_LOCATION_LOCAL, true)
1555            .expect("start recording");
1556
1557        // Publish a few messages so the recording has content
1558        for _ in 0..5 {
1559            let _ = publication.offer(b"test data");
1560        }
1561
1562        // Drop all Aeron and archive children before close_now
1563        drop(subscription);
1564        drop(publication);
1565        // recording_id is just a u64, not a resource handle
1566
1567        // Call close_now on the archive — must not segfault or double-free
1568        unsafe {
1569            assert!(
1570                archive.close_now().is_ok(),
1571                "close_now should succeed after all children dropped"
1572            );
1573        }
1574
1575        // Aeron client should still be usable (archive close doesn't close aeron)
1576        let pub2 = aeron
1577            .add_publication(AERON_IPC_STREAM, 31, Duration::from_secs(5))
1578            .expect("aeron still usable after archive close_now");
1579        drop(pub2);
1580        aeron.close()?;
1581
1582        drop(media_driver);
1583        drop(error_handler);
1584        Ok(())
1585    }
1586}
1587
1588// run `just slow-tests`
1589#[cfg(test)]
1590mod slow_consumer_test;
1591
1592///////////////////////////////////////////////////////////////////////////////
1593// Backtest Tests
1594///////////////////////////////////////////////////////////////////////////////
rusteron_archive/lib.rs

rusteron_archive/
lib.rs
