Crate rusteron_archive

Expand description

§rusteron-archive

rusteron-archive is a module within the rusteron project that provides functionality for interacting with Aeron’s archive system in a Rust environment. This module builds on rusteron-client, adding support for recording, managing, and replaying archived streams.

§Sponsored by GSR

Rusteron is proudly sponsored and maintained by GSR, a global leader in algorithmic trading and market making in digital assets.

It powers mission-critical infrastructure in GSR’s real-time trading stack and is now developed under the official GSR GitHub organization as part of our commitment to open-source excellence and community collaboration.

We welcome contributions, feedback, and discussions. If you’re interested in integrating or contributing, please open an issue or reach out directly.

§Overview

The rusteron-archive module enables Rust developers to leverage Aeron’s archive functionality, including recording and replaying messages with minimal friction.

For MacOS users, the easiest way to get started is by using the static library with precompiled C dependencies. This avoids the need for cmake or Java:

rusteron-archive = { version = "0.2", features = ["static", "precompile"] }

If you prefer a rustls-only downloader dependency:

rusteron-archive = { version = "0.2", features = ["static", "precompile-rustls"] }

§Installation

Add rusteron-archive to your Cargo.toml depending on your setup:

# Dynamic linking (default)
rusteron-archive = "0.2"

# Static linking
rusteron-archive = { version = "0.2", features = ["static"] }

# Static linking with precompiled C libraries (best for Mac users, no Java/cmake needed)
rusteron-archive = { version = "0.2", features = ["static", "precompile"] }

# Static linking with precompiled C libraries using rustls downloader
rusteron-archive = { version = "0.2", features = ["static", "precompile-rustls"] }

When using the default dynamic configuration, you must ensure Aeron C libraries are available at runtime. The static option embeds them automatically into the binary.

§Development

Build tasks use just. Run just to list commands, or cargo install just if needed.

§Features

Stream Recording – Record Aeron streams for replay or archival.
Replay Handling – Replay previously recorded messages.
Persistent Subscriptions – Replay recorded history, then seamlessly join the live stream (Aeron Archive 1.51.0). See below.
Publication/Subscription – Publish to and subscribe from Aeron channels.
Callbacks – Receive events such as new publications, subscriptions, and errors.
Automatic Resource Management (via new() only) – Constructors automatically call *_init and clean up with *_close or *_destroy when dropped.
String Handling – new() and setter methods accept &CStr; getter methods return &str.

§General Patterns

§Cloneable Wrappers

All wrapper types in rusteron-archive implement Clone and share the same underlying Aeron C resource. For shallow copies of raw structs, use .clone_struct().

§Mutable and Immutable APIs

Most methods use &self, allowing mutation without full ownership transfer.

§Resource Management Caveats

Automatic cleanup applies only to new() constructors. Other methods (e.g. set_aeron()) require manual lifetime and validity tracking to prevent resource misuse.

§Handlers and errors

Retained-callback setters take the callback by value (a closure or trait impl), keep it alive inside the registering resource, and return the Handler for optional state access. For synchronous polling, pass a stack closure:

// retained (e.g. an error handler on the archive context)
archive_context.set_error_handler(Some(|code: i32, msg: &str| eprintln!("archive error {code}: {msg}")))?;

// synchronous poll — note the fragment-limit argument
subscription.poll_fn(|buf: &[u8], header: AeronHeader| println!("{} bytes", buf.len()), 10)?;

Handlers::NONE fits any optional callback slot.

For comprehensive details on how handler registration, callbacks, error checking, and idle strategies work in the rusteron ecosystem (which are fully applicable here as well), please refer to the corresponding sections in the rusteron-client documentation:

Archive control operations (begin_replay, start_recording, …) return Result<_, AeronArchiveError> — a typed code (AeronArchiveErrorCode) plus the archive’s message. Constructors, async-connect, and context setters return AeronCError; From<AeronArchiveError> for AeronCError keeps ? working across both.

§Documentation & Guides

For detailed guides and code snippets on Aeron features in Rust, see:

§Safety Considerations

Aeron Lifetime – The AeronArchive depends on an external Aeron instance. Ensure Aeron outlives all references to the archive.
Unsafe Bindings – The module interfaces directly with Aeron’s C API. Improper resource handling can cause undefined behavior.
Automatic Handler Cleanup – Handlers are reference-counted; registered callbacks live as long as the resource that registered them and are freed automatically.
Thread Safety – Use care when accessing Aeron objects across threads. Synchronize access appropriately.

§Typical Workflow

Initialize client and archive contexts.
Start Recording a specific channel and stream.
Publish Messages to the stream.
Stop Recording once complete.
Locate the Recording using archive queries.
Replay Setup: Configure replay target/channel.
Subscribe and Receive replayed messages.

§Persistent Subscriptions

A persistent subscription replays a recording from a start position, then seamlessly merges into the live stream — so a consumer catches up on history without missing new messages and without a gap at the handover. Introduced in Aeron Archive 1.51.0.

What it is: Aeron — Persistent Subscriptions (replay-to-live)
How it works: Aeron Wiki — Persistent Subscriptions
Background on publications/subscriptions: Aeron docs

Rusteron exposes it via persistent_subscription_builder() and the PersistentSubscriptionListener trait — a 1:1 wrapper over the Aeron C API (aeron_archive_persistent_subscription_*), mirroring Aeron’s PersistentSubscription.Context field-for-field.

use rusteron_archive::*;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;

// `archive` is a connected AeronArchive; record + publish history first, then resolve recording_id.
let live_channel = "aeron:ipc";
let stream_id = 1001;

struct MyListener { live_joined: Arc<AtomicUsize> }
impl PersistentSubscriptionListener for MyListener {
    fn on_live_joined(&self) { self.live_joined.fetch_add(1, Ordering::SeqCst); }
    fn on_live_left(&self)  { /* fell back to replay */ }
    fn on_error(&self, code: i32, msg: &str) { eprintln!("ps error {code}: {msg}"); }
}
let live_joined = Arc::new(AtomicUsize::new(0));

let ps = persistent_subscription_builder()?
    .aeron(&aeron)?
    .archive_context(&archive_context)?
    .live_channel(live_channel)?        // the live stream to join
    .live_stream_id(stream_id)?
    .replay_channel("aeron:udp?endpoint=localhost:0")?  // scratch channel for the replay
    .replay_stream_id(stream_id + 1)?
    .start_from_beginning()?            // replay from the start (or .start_from_live())
    .recording_id(recording_id)?        // which recording to replay
    .listener(MyListener { live_joined: live_joined.clone() })?
    .build()?;

// Drive it: replay runs, then it joins live. `ps.poll_fn()` drives the archive
// client internally, so no `archive.poll_for_recording_signals()` is needed. Check
// `has_failed()` each iteration (terminal failure) and stop once `is_live()`.
while !ps.is_live() {
    if ps.has_failed() {
        panic!("persistent subscription failed: {:?}", ps.get_failure_reason());
    }
    let _ = publication.offer_with_reserved_value(b"live", Handlers::NONE);
    ps.poll_fn(|buf, _hdr| { /* an assembled replayed or live message */ }, 100)?;
}
ps.close()?;

Polling & errors. ps.poll_fn() drives the PS state machine and the archive async client, so you do not call archive.poll_for_recording_signals() separately. Loop on ps.is_live(), checking ps.has_failed() each iteration (reason via get_failure_reason()). The listener’s on_error covers non-terminal errors; on_live_left/on_live_joined may fire repeatedly as it falls back and rejoins.

Fragment assembly (already done for you). Unlike AeronSubscription, the persistent subscription reassembles fragments internally — the C aeron_archive_persistent_subscription_poll routes each image through aeron_image_fragment_assembler_handler, so your handler receives whole messages directly. Just poll:

loop {
    // handler receives whole messages; no assembler needed
    ps.poll_fn(|buf, _hdr| { /* handle reassembled message */ }, 100)?;
}

If you prefer the shared assembler API (e.g. to reuse a collector across subscription types), AeronFragmentClosureAssembler works too — it polls the PS internally, so it advances the state machine and delivers messages in one call. Do not also call ps.poll_fn(…) separately: that consumes the messages before the assembler sees them.

let mut assembler = AeronFragmentClosureAssembler::new()?;
let mut ctx = Collector::default();
loop {
    assembler.poll(&ps, &mut ctx, Collector::on_msg, 100)?;  // polls the PS internally
    if ctx.done { break; }
}

For a fully runnable version, see the example and integration tests:

examples/persistent_subscription.rs — standalone demo (run with cargo run --release --features "static precompile" --example persistent_subscription)
examples/archive_error_handling.rs — error handlers on both contexts, recording signals, typed control-session errors via archive.poll_for_error() / AeronArchiveError::parse (the archive’s errorCode=N recovered from the message text), and detecting/reconnecting after the archive goes down
examples/persistent_subscription_failover.rs — failure modes: live stream dies → automatic fallback to replay (on_live_left), then rejoins live when it returns
examples/replay_merge.rs — late-joiner catch-up: replay recorded history, then merge seamlessly onto the live MDC stream (AeronArchiveReplayMerge)
examples/recording_throughput.rs — recording throughput measurement (publish rate vs archiver catch-up) and list_recordings descriptor enumeration
examples/recording_replication.rs — archive-to-archive replication (archive.replicate): a destination archive pulls a finished recording from a source archive and the copy is verified (port of RecordingReplicator)
persistent_subscription_tests::test_persistent_subscription_listener_live_joined (callback wiring)
persistent_subscription_integration::test_end_to_end_persistent_subscription (record → replay → live)

§Benchmarks

For latency and throughput benchmarks, refer to BENCHMARKS.md.

§Contributing

Contributions are more than welcome! Please:

Submit bug reports, ideas, or improvements via GitHub Issues
Propose changes via pull requests
Read our CONTRIBUTING.md

We’re especially looking for help with:

API design reviews
Safety and idiomatic improvements
Dockerized and deployment examples

§License

Licensed under either MIT License or Apache License 2.0 at your option.

§Acknowledgments

Special thanks to:

@mimran1980, a core low-latency developer at GSR and the original creator of Rusteron - your work made this possible!
@bspeice for the original libaeron-sys
The Aeron community for open protocol excellence

§Features

static: When enabled, this feature statically links the Aeron C code. By default, the library uses dynamic linking to the Aeron C libraries.
backtrace: When enabled will log a backtrace for each AeronCError
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
log-c-bindings: When enabled will log every C binding call with arguments and return values. Useful for debugging FFI interactions
precompile: When enabled will use precompiled C code instead of requiring cmake and java to be installed

Modules§

bindings
testing

Macros§

cformat: format! for C strings: builds the formatted String and converts it to a CString in one visibly-named step.
skip_unless_java: Place at the top of any test that needs the Java archive. Skips the test (early-returns Ok(())) when java is not on PATH. Requires the test to return a Result<(), _>.

Structs§

Aeron
AeronAgentCloseFuncLogger
AeronAgentDoWorkFuncLogger
AeronAgentRunner
AeronAgentStartFuncLogger
AeronArchive
AeronArchiveAsyncConnect
AeronArchiveContext
AeronArchiveControlResponsePoller
AeronArchiveCredentialsChallengeSupplierFuncLogger
AeronArchiveCredentialsEncodedCredentialsSupplierFuncLogger
AeronArchiveCredentialsFreeFuncLogger
AeronArchiveDelegatingInvokerFuncLogger
AeronArchiveEncodedCredentials
AeronArchiveError: A typed archive control-session error: the AeronArchiveErrorCode plus the full message from the archive.
AeronArchivePersistentSubscription
AeronArchivePersistentSubscriptionContext
AeronArchivePersistentSubscriptionListener: Listener for events from a persistent subscription.
AeronArchiveProxy
AeronArchiveRecordingDescriptor: Struct containing the details of a recording
AeronArchiveRecordingDescriptorConsumerFuncLogger
AeronArchiveRecordingDescriptorPoller
AeronArchiveRecordingSignal: Struct containing the details of a recording signal.
AeronArchiveRecordingSignalConsumerFuncLogger
AeronArchiveRecordingSubscriptionDescriptor: Struct containing the details of a recording subscription
AeronArchiveRecordingSubscriptionDescriptorConsumerFuncLogger
AeronArchiveRecordingSubscriptionDescriptorPoller
AeronArchiveReplayMerge
AeronArchiveReplayParams: Struct containing the available replay parameters.
AeronArchiveReplayParamsBuilder: Fluent builder for AeronArchiveReplayParams, starting from aeron’s defaults (every field AERON_NULL_VALUE: replay from the recording start to its end, with the context-default file IO length and no bounding counter).
AeronArchiveReplicationParams: Struct containing the available replication parameters.
AeronArchiveReplicationParamsBuilder: Fluent builder for AeronArchiveReplicationParams, starting from aeron’s defaults: replicate into a new recording at the destination, no live merge, the context’s default replication channel, and no credentials.
AeronArchiveReplicationParamsOwned: AeronArchiveReplicationParams plus the string storage its C struct points into.
AeronAsyncAddCounter
AeronAsyncAddExclusivePublication
AeronAsyncAddPublication
AeronAsyncAddSubscription
AeronAsyncDestination
AeronAsyncDestinationById
AeronAsyncGetNextAvailableSessionId
AeronAvailableCounterLogger
AeronAvailableCounterPair
AeronAvailableImageLogger
AeronBlockHandlerLogger
AeronBufferClaim: Structure used to hold information for a try_claim function call.
AeronCError: Aeron C API error: code + optional message.
AeronClaim: Zero-copy claim on a publication’s term buffer with a RAII commit-or-abort lifecycle.
AeronClientRegisteringResource
AeronCloseClientLogger
AeronCloseClientPair
AeronCnc
AeronCncConstants
AeronCncMetadata
AeronContext
AeronControlledFragmentAssembler
AeronControlledFragmentClosureAssembler
AeronControlledFragmentHandlerLogger
AeronCounter
AeronCounterConstants: Configuration for a counter that does not change during its lifetime.
AeronCounterMetadataDescriptor
AeronCounterValueDescriptor
AeronCountersReader
AeronCountersReaderBuffers
AeronCountersReaderForeachCounterFuncLogger
AeronDataHeader
AeronDataHeaderAsLongs
AeronError
AeronErrorHandlerLogger
AeronErrorLogReaderFuncLogger
AeronErrorLogger: Production error handler that routes Aeron async errors through the Rust log facade at error level with a concise format — the recommended default for AeronContext::set_error_handler.
AeronExclusivePublication
AeronFprintfHandlerLogger
AeronFragmentAssembler
AeronFragmentClosureAssembler
AeronFragmentHandlerLogger
AeronFrameHeader
AeronHeader
AeronHeaderValues
AeronHeaderValuesFrame
AeronIdleStrategy
AeronIdleStrategyFuncLogger
AeronImage
AeronImageConstants: Configuration for an image that does not change during it’s lifetime.
AeronImageControlledFragmentAssembler
AeronImageFragmentAssembler
AeronIovec
AeronIpcChannelParams
AeronLogBuffer
AeronLogbufferMetadata
AeronLossReporter
AeronLossReporterEntry
AeronLossReporterReadEntryFuncLogger
AeronMappedBuffer
AeronMappedFile
AeronMappedRawLog
AeronNakHeader
AeronNewPublicationLogger
AeronNewSubscriptionLogger
AeronNotificationLogger
AeronOptionHeader
AeronPerThreadError
AeronPublication
AeronPublicationConstants: Configuration for a publication that does not change during it’s lifetime.
AeronPublicationErrorFrameHandlerLogger
AeronPublicationErrorValues
AeronReservedValueSupplierLogger
AeronResolutionHeader
AeronResolutionHeaderIpv4
AeronResolutionHeaderIpv6
AeronResponseSetupHeader
AeronRttmHeader
AeronSetupHeader
AeronStatusMessageHeader
AeronStatusMessageOptionalHeader
AeronStatusTracker: Status transition tracker that emits only when the observed AeronStatus differs from the previously recorded state.
AeronStrToPtrHashMap
AeronStrToPtrHashMapForEachFuncLogger
AeronStrToPtrHashMapKey
AeronSubscription
AeronSubscriptionConstants
AeronUdpChannelParams
AeronUnavailableCounterLogger
AeronUnavailableCounterPair
AeronUnavailableImageLogger
AeronUri
AeronUriParam
AeronUriParams
AeronUriParseCallbackLogger
AeronUriStringBuilder
ChannelUri: Represents the Aeron URI parser and handler.
FnMutControlledMessageHandler
FnMutMessageHandler
Handler: Handler
Handlers: Utility method for setting empty handlers
ManagedCResource: A custom struct for managing C resources with automatic cleanup.
NoHandler: Type-level “no callback” sentinel.
PersistentSubscriptionBuilder: Builder for configuring and creating a persistent subscription. This provides a fluent interface for setting up a persistent subscription with proper CString handling.
RecordingDescriptor: Recording descriptor for owned recording data
RecordingPos

Enums§

AeronArchiveErrorCode: Archive control-response error codes (io.aeron.archive.client.ArchiveException).
AeronErrorType
AeronIdleStrategyKind: Aeron’s named idle strategies, as accepted by the context / media-driver set_*_idle_strategy options (the C aeron_idle_strategy_load symbol table).
AeronOfferError: Typed error for offer / try_claim on a publication.
AeronStatus: High-level connection state of a publication or subscription.
AeronSystemCounterType
CResource
ControlMode: Enum for control modes.
Media: Enum for media types.

Constants§

AERON_DIR_PROP_NAME
AERON_IPC_MEDIA
AERON_UDP_MEDIA
DRIVER_TIMEOUT_MS_DEFAULT
MAX_OFFER_PARTS: Max buffer parts accepted by AeronPublication::offer_parts / AeronExclusivePublication::offer_parts — the aeron_iovec_t array is built on the stack, so it has a fixed capacity. Use the raw offerv with your own iovec array for larger gathers.
PERSISTENT_SUBSCRIPTION_FROM_LIVE: Sentinel for PersistentSubscriptionBuilder::start_position: skip replay and join the live stream immediately. Maps to Aeron’s AERON_ARCHIVE_PERSISTENT_SUBSCRIPTION_FROM_LIVE.
PERSISTENT_SUBSCRIPTION_FROM_START: Sentinel for PersistentSubscriptionBuilder::start_position: replay from the beginning of the recording. Maps to Aeron’s AERON_ARCHIVE_PERSISTENT_SUBSCRIPTION_FROM_START.
PUBLICATION_ADMIN_ACTION
PUBLICATION_BACK_PRESSURED
PUBLICATION_CLOSED
PUBLICATION_ERROR
PUBLICATION_MAX_POSITION_EXCEEDED
PUBLICATION_NOT_CONNECTED: Result codes returned by AeronPublication::offer / try_claim (Aeron aeronc.h). A positive value is the resulting log position; the negatives classify the failure.
SOURCE_LOCATION_LOCAL
SOURCE_LOCATION_REMOTE
SPY_PREFIX
TAG_PREFIX

Statics§

AERON_IPC_STREAM

Traits§

AeronAgentCloseFuncCallback: (note you must copy any arguments that you use afterwards even those with static lifetimes)
AeronAgentDoWorkFuncCallback: (note you must copy any arguments that you use afterwards even those with static lifetimes)
AeronAgentStartFuncCallback: (note you must copy any arguments that you use afterwards even those with static lifetimes)
AeronArchiveCredentialsChallengeSupplierFuncCallback: Callback to return encoded credentials given a specific encoded challenge.
AeronArchiveCredentialsEncodedCredentialsSupplierFuncCallback: Callback to return encoded credentials.
AeronArchiveCredentialsFreeFuncCallback: Callback to return encoded credentials so they may be reused or freed.
AeronArchiveDelegatingInvokerFuncCallback: Callback to allow execution of a delegating invoker to be run.
AeronArchiveRecordingDescriptorConsumerFuncCallback: Callback to return recording descriptors.
AeronArchiveRecordingSignalConsumerFuncCallback: Callback to return recording signals.
AeronArchiveRecordingSubscriptionDescriptorConsumerFuncCallback: Callback to return recording subscription descriptors.
AeronAvailableCounterCallback: Function called by aeron_client_t to deliver notifications that a counter has been added to the driver.
AeronAvailableImageCallback: Function called by aeron_client_t to deliver notifications that an aeron_image_t was added.
AeronBlockHandlerCallback: Callback for handling a block of messages being read from a log.
AeronCloseClientCallback: Function called by aeron_client_t to deliver notifications that the client is closing.
AeronControlledFragmentHandlerCallback: Callback for handling fragments of data being read from a log.
AeronCountersReaderForeachCounterFuncCallback: Function called by aeron_counters_reader_foreach_counter for each counter in the aeron_counters_reader_t.
AeronErrorHandlerCallback: The error handler to be called when an error occurs.
AeronErrorLogReaderFuncCallback: (note you must copy any arguments that you use afterwards even those with static lifetimes)
AeronFprintfHandlerCallback: (note you must copy any arguments that you use afterwards even those with static lifetimes)
AeronFragmentHandlerCallback: Callback for handling fragments of data being read from a log.
AeronIdleStrategyFuncCallback: (note you must copy any arguments that you use afterwards even those with static lifetimes)
AeronLossReporterReadEntryFuncCallback: (note you must copy any arguments that you use afterwards even those with static lifetimes)
AeronNewPublicationCallback: Function called by aeron_client_t to deliver notification that the media driver has added an aeron_publication_t or aeron_exclusive_publication_t successfully.
AeronNewSubscriptionCallback: Function called by aeron_client_t to deliver notification that the media driver has added an aeron_subscription_t successfully.
AeronNotificationCallback: Generalised notification callback.
AeronPublicationErrorFrameHandlerCallback: The error frame handler to be called when the driver notifies the client about an error frame being received. The data passed to this callback will only be valid for the lifetime of the callback. The user should use aeron_publication_error_values_copy if they require the data to live longer than that.
AeronReservedValueSupplierCallback: Function called when filling in the reserved value field of a message.
AeronStrToPtrHashMapForEachFuncCallback: (note you must copy any arguments that you use afterwards even those with static lifetimes)
AeronUnavailableCounterCallback: Function called by aeron_client_t to deliver notifications that a counter has been removed from the driver.
AeronUnavailableImageCallback: Function called by aeron_client_t to deliver notifications that an aeron_image_t has been removed from use and should not be used any longer.
AeronUriParseCallbackCallback: (note you must copy any arguments that you use afterwards even those with static lifetimes)
ControlledFragmentAssemblable: A poll target whose raw fragments can be reassembled by AeronControlledFragmentAssembler.
FragmentAssemblable: A poll target whose raw fragments can be reassembled by AeronFragmentAssembler. Implemented for AeronSubscription here and for archive types in rusteron-archive.
IntoCString
PersistentSubscriptionListener: Trait for persistent subscription event listeners. This provides a safe Rust alternative to using raw C function pointers.

Functions§

find_unused_udp_port
is_udp_port_available
persistent_subscription_builder: Returns a builder for configuring a persistent subscription context
validate_endpoint_for_aeron_udp: Validate a UDP channel URI endpoint (host:port) for use with Aeron.

Type Aliases§

CleanupBox
RcOrArc: Reference-counting smart pointer: Rc by default, Arc under the multi-threaded feature. Swap is transparent — RcOrArc::new, .clone(), strong_count all work on both.
RefCellOrMutex
SourceLocation