Struct EmbeddedBroker 

pub struct EmbeddedBroker {
    pub recovered_torn_lines: u64,
    /* private fields */
}

File-backed broker instance.

Fields

recovered_torn_lines: u64

Torn trailing lines dropped during recovery (public field asserted by the contract tests; not currently read by any caller or exported as a metric).

Implementations

impl EmbeddedBroker

pub fn provision( data_dir: &Path, manifest: &BridgeManifest, ) -> Result<Self, BusError>

Provision a fresh bridge in data_dir from manifest alone (R12). Fails if the directory already contains a bridge.

pub fn open(data_dir: &Path) -> Result<Self, BusError>

Open an existing bridge, recovering offsets (and truncating at most one torn trailing line per partition) from the segment files.

pub fn manifest(&self) -> &BridgeManifest

The manifest this bridge was provisioned from.

pub fn data_dir(&self) -> &Path

Data directory.

pub fn fetch_read_only( data_dir: &Path, topic: &str, partition: u32, offset: u64, max: usize, ) -> Result<Vec<StoredEvent>, BusError>

Read-only event fetch for tooling running BESIDE a live daemon (avctl event-tail). Self::open is NOT safe for that use: its recovery “repairs” a torn segment tail via set_len and atomically rewrites the idempotency sidecar — racing a daemon that is mid-append (between write_all of the record bytes and the trailing newline), a second process’s repair truncates bytes the daemon goes on to ack as durable: a silently lost or corrupted acked audit event. There is no cross-process lock, so the only safe concurrent mode is one that mutates nothing.

This path only reads: the manifest (capped, to validate the topic/partition and locate the segment), then a bounded streaming scan of the one segment file. An unparseable line is skipped (same policy as EventBus::fetch); an unterminated trailing line — an in-flight append or crash-torn tail — is ignored, never truncated; a record above the per-line bound stops the scan at that point (fail-stop, mutation-free).

impl EmbeddedBroker

pub fn enforce_retention(&self, now_ms: u64) -> Result<u64, BusError>

Enforce per-topic hot retention at time now_ms: when retention.cold_uri is set, each expired record is first exported to the cold tier as its own write-once object (via the authenticated ColdArchive for scheme:// URIs, or write_cold_event_once for local directory paths) before being removed from the hot segment via atomic rewrite; with cold_uri unset, expired records are dropped from the hot segment without export. Returns the number of records expired.

Trait Implementations

impl EventBus for EmbeddedBroker

fn find_event_by_uid( &self, topic: &str, key: &str, event_uid: &str, ) -> Result<Option<PublishAck>, BusError>

O(1) event-UID lookup via the partition’s in-memory seen_event_uids map. The default trait impl at bus.rs::find_event_by_uid linearly scans the segment from offset 0 with fetch, which for EmbeddedBroker re-opens the segment file each call and streams from the head with a BufReader — so each fetch page for offset > 0 also walks the whole prefix. resolve_lifecycle_ack in the harness calls this on EVERY emit_bridge_event (twice per session close — RECEIPT + SESSION_CLOSE), which meant close latency grew O(total events per partition²) amortised. The Kafka and NATS bus implementations both override with O(1) lookups; only the embedded broker was still on the trait default.

fn set_control_key(&self, _key: [u8; 32]) -> Result<(), BusError>

Configure the signer-derived key used for authenticated local controls.

fn publish( &self, topic: &str, key: &str, value: &Value, ) -> Result<PublishAck, BusError>

Publish value onto topic, partitioned by key. Returns the ack.

fn publish_idempotent( &self, topic: &str, key: &str, value: &Value, event_uid: &str, ) -> Result<PublishAck, BusError>

Publish a stable event UID. Backends with native or local deduplication override this method; the default preserves compatibility while the UID remains embedded in the event payload for downstream deduplication.

fn fetch( &self, topic: &str, partition: u32, offset: u64, max: usize, ) -> Result<Vec<StoredEvent>, BusError>

Read up to max events from topic/partition starting at offset (ordered replay).

fn partitions(&self, topic: &str) -> Result<u32, BusError>

Number of partitions configured for topic.

fn topics(&self) -> Vec<String>

Topics currently provisioned.

fn maintenance(&self, now_ms: u64) -> Result<u64, BusError>

Run backend maintenance such as hot-retention expiry. Managed brokers may return zero when retention is enforced natively.