yeti_types/backend/

observer.rs

//! Transaction observer trait — the unified post-commit hook surface
//! that audit, replication, and (future) other write-observation
//! consumers share.
//!
//! ## Why one trait
//!
//! Before this surface, audit and replication would have lived as
//! parallel hook paths — audit firing in `yeti-table::query_exec`
//! above `KvBackend` (fire-and-forget, drops on full channel),
//! replication firing at commit time inside the backend (durable,
//! cannot drop). Two paths means:
//!
//! - Slight differences in what each path sees (audit gets JSON-
//!   cooked records; replication gets binary write ops) creating
//!   "audit and replication disagree about state" failure modes
//! - Doubled instrumentation and observability cost
//! - More places to keep in sync when the write path evolves
//!
//! Replication's data + reliability requirements are a strict
//! superset of audit's, and replication's fire-point (commit time
//! inside the backend) is also valid for audit. Unifying as a single
//! trait means one observation point, multiple subscribers, each
//! declaring its requirements and reacting to commits according to
//! its own contract.
//!
//! ## Contract
//!
//! Observers are registered on the [`BackendManager`](crate::backend::BackendManager) at startup and
//! are invoked **synchronously inside the backend's commit path**,
//! after the data has landed in the storage engine but before
//! `KvBackend::write_batch` returns to the caller.
//!
//! **Failure semantics**: returning `Err` from
//! [`on_commit`](TransactionObserver::on_commit) does NOT roll back
//! the data write — yeti's underlying storage doesn't support
//! two-phase commit across the data write and arbitrary observer
//! side-effects. Observer failures are logged, metered, and surfaced
//! to operators; the write itself succeeds. Observers that need
//! transactional durability with the data write must arrange it
//! themselves (e.g., the replication observer writes its log entry
//! into the same `RocksDB` instance via a separate column family, so
//! "data committed but log missing" is a single-machine atomicity
//! failure mode rather than a cross-system one).
//!
//! Observers that want fire-and-forget semantics (e.g., audit) must
//! return `Ok(())` always and swallow their own internal failures —
//! a returned `Err` should mean "operator should be paged," not
//! "this write was rejected."
//!
//! ## Ordering
//!
//! Observers fire in **registration order**. The replication
//! observer registers first (durability is the foundational
//! contract); audit registers second (telemetry on top of durability).
//! Order matters: if the replication log write fails and audit's
//! observer would have logged the failure, audit needs to see it.

use std::sync::Arc;

use crate::error::Result;
use crate::hlc::HlcTimestamp;
use crate::types::TableId;

use super::WriteOp;

/// Per-observer declaration of which optional fields the backend
/// must populate on a [`TransactionEvent`].
///
/// The backend OR's all registered observers' requirements and
/// produces exactly the fields some observer wants. Fields nobody
/// needs are skipped — for example, the "previous value" read
/// before commit is expensive (an extra `RocksDB` get); if neither
/// audit (with `capture_state: false`) nor replication (today) asks
/// for it, the backend skips that read entirely.
#[derive(Debug, Clone, Copy, Default)]
pub struct ObserverRequirements {
    /// Read the existing value at the key (if any) before committing
    /// the new write. Costs an extra storage `get`. Audit with
    /// `capture_state: true` needs this; replication today does not
    /// (CRDT merge happens at apply time on the receiver using its
    /// local state).
    pub needs_before_value: bool,

    /// Look up the HLC timestamp of the existing record at the key
    /// (if any) and pass it as `prev_hlc` for causal predecessor
    /// tracking. Cheap because the HLC is stored alongside the
    /// record metadata in a known location. Replication needs this
    /// for the "this write observed prev HLC X" causal-chain
    /// invariant; audit does not.
    pub needs_prev_hlc: bool,
}

impl ObserverRequirements {
    /// Combine two requirement sets — the OR of all flags. Used by
    /// the backend to compute the union of all registered observers'
    /// needs.
    #[must_use]
    pub const fn union(self, other: Self) -> Self {
        Self {
            needs_before_value: self.needs_before_value || other.needs_before_value,
            needs_prev_hlc: self.needs_prev_hlc || other.needs_prev_hlc,
        }
    }
}

/// A single committed transaction event — what observers see.
///
/// Borrowed view: the backend produces this on the stack at commit
/// time and passes it by reference. Observers that want to outlive
/// the borrow must copy the fields they care about.
#[derive(Debug)]
pub struct TransactionEvent<'a> {
    /// HLC timestamp assigned to this write at commit time.
    pub hlc: HlcTimestamp,

    /// Node identifier of the writer.
    ///
    /// For locally-originated writes, this is `self.node_id` — the
    /// id of the node that called `write_batch`. For writes applied
    /// by the replication observer in response to a peer's log
    /// entry, this is the **originating peer's** id, not the local
    /// node's. Observers that should only act on local writes
    /// (e.g., the replication observer's outgoing-log writer) filter
    /// on `node_id == self.node_id`; observers that act on all
    /// observed state (e.g., audit) do not filter.
    pub node_id: &'a str,

    /// In-deployment table identity (app / database / table). Tenant
    /// scoping lives at the deployment frame in the multiplexer, not
    /// inside the table id — see [`crate::types::DeploymentHash`].
    pub table: &'a TableId,

    /// The write operation. Put or Delete; CRDT operations are
    /// encoded inside the value via `__op__` markers (per the
    /// Harper convention). The wire shape stays uniform; CRDT-ness
    /// is in the data, not the protocol.
    pub op: &'a WriteOp,

    /// Existing value bytes at the key (if any) prior to this write,
    /// populated only if some registered observer set
    /// [`ObserverRequirements::needs_before_value`].
    pub before: Option<&'a [u8]>,

    /// HLC of the existing record at the key (if any), populated
    /// only if some registered observer set
    /// [`ObserverRequirements::needs_prev_hlc`]. Used by replication
    /// to record causal-predecessor relationships in the log.
    pub prev_hlc: Option<HlcTimestamp>,
}

/// Trait implemented by every consumer of the post-commit observation
/// surface (audit, replication, etc.).
///
/// See module docs for the full contract.
pub trait TransactionObserver: Send + Sync + std::fmt::Debug + 'static {
    /// A short identifier for diagnostics and observability — appears
    /// in error logs and metric labels. Conventionally
    /// `crate-shortname` like `"audit"` or `"replication"`.
    fn name(&self) -> &'static str;

    /// Declare which optional [`TransactionEvent`] fields this
    /// observer needs the backend to populate. Called once at
    /// registration time; cached by the backend.
    fn requirements(&self) -> ObserverRequirements;

    /// Called inside the backend's commit path, after the storage
    /// write succeeded. See module docs for failure semantics.
    ///
    /// # Errors
    ///
    /// Returns the observer-specific [`YetiError`](crate::error::YetiError)
    /// when post-commit processing fails. The backend logs the error
    /// and continues — observer failures do not roll back the write.
    fn on_commit(&self, event: &TransactionEvent<'_>) -> Result<()>;
}

/// Shareable, registerable observer handle. The [`BackendManager`](crate::backend::BackendManager)
/// stores `Vec<ObserverHandle>` and iterates it on every commit.
pub type ObserverHandle = Arc<dyn TransactionObserver>;

// ============================================================================
// ObserverRegistry
// ============================================================================

/// Shared, lock-free registry of [`TransactionObserver`]s.
///
/// One registry instance lives per-deployment process, held by
/// [`BackendManager`](crate::backend::BackendManager) and cloned into every `LoggingBackend` so the
/// commit path can fan out without coordination. The internal
/// snapshot is swapped via [`arc_swap::ArcSwap`]: registration takes
/// a brief write, every commit-path read is wait-free.
///
/// Registration happens during deployment startup before traffic is
/// accepted. The contract assumes near-zero registrations after
/// startup completes; nothing in the type enforces that, but the
/// trait doc's "registration order matters" + "registered first
/// fires first" promises hold as long as registration is
/// startup-serialized.
#[derive(Debug, Default)]
pub struct ObserverRegistry {
    /// Snapshot of the registered observers, swapped atomically on
    /// registration. The empty default is `Arc::new(Vec::new())`.
    observers: arc_swap::ArcSwap<Vec<ObserverHandle>>,
}

impl ObserverRegistry {
    /// Construct an empty registry wrapped in an [`Arc`] for sharing.
    #[must_use]
    pub fn new() -> Arc<Self> {
        Arc::new(Self::default())
    }

    /// Register an observer. Appends to the end of the snapshot so
    /// registration order is preserved (replication first, audit
    /// second per the trait doc).
    pub fn register(&self, observer: ObserverHandle) {
        let cur = self.observers.load_full();
        let mut next = Vec::with_capacity(cur.len() + 1);
        next.extend(cur.iter().cloned());
        next.push(observer);
        self.observers.store(Arc::new(next));
    }

    /// Snapshot of the currently-registered observers. Cheap (`Arc`
    /// clone) — call from the commit path on every write.
    #[must_use]
    pub fn snapshot(&self) -> Arc<Vec<ObserverHandle>> {
        self.observers.load_full()
    }

    /// Computed union of all registered observers' requirements.
    /// Recomputed on each call; cache at the caller if hot.
    #[must_use]
    pub fn requirements(&self) -> ObserverRequirements {
        self.observers
            .load()
            .iter()
            .map(|o| o.requirements())
            .fold(ObserverRequirements::default(), ObserverRequirements::union)
    }

    /// `true` if no observers are registered. Commit paths can use
    /// this to skip building the [`TransactionEvent`] entirely.
    #[must_use]
    pub fn is_empty(&self) -> bool {
        self.observers.load().is_empty()
    }
}

#[cfg(test)]
#[allow(clippy::unwrap_used, clippy::expect_used)]
mod tests {
    use super::*;

    #[derive(Debug)]
    struct NullObserver;
    impl TransactionObserver for NullObserver {
        fn name(&self) -> &'static str {
            "null"
        }
        fn requirements(&self) -> ObserverRequirements {
            ObserverRequirements::default()
        }
        fn on_commit(&self, _event: &TransactionEvent<'_>) -> Result<()> {
            Ok(())
        }
    }

    #[test]
    fn requirements_union_or_combines() {
        let a = ObserverRequirements {
            needs_before_value: true,
            needs_prev_hlc: false,
        };
        let b = ObserverRequirements {
            needs_before_value: false,
            needs_prev_hlc: true,
        };
        let c = a.union(b);
        assert!(c.needs_before_value);
        assert!(c.needs_prev_hlc);
    }

    #[test]
    fn observer_handle_is_arc_send_sync() {
        // Compile-time check: the trait object via ObserverHandle is
        // shareable across threads. Important because BackendManager
        // is shared across the runtime and observers are read from
        // many tasks.
        fn assert_send_sync<T: Send + Sync>() {}
        assert_send_sync::<ObserverHandle>();
        let _: ObserverHandle = Arc::new(NullObserver);
    }

    #[derive(Debug)]
    struct WithReqs(ObserverRequirements);
    impl TransactionObserver for WithReqs {
        fn name(&self) -> &'static str {
            "with-reqs"
        }
        fn requirements(&self) -> ObserverRequirements {
            self.0
        }
        fn on_commit(&self, _event: &TransactionEvent<'_>) -> Result<()> {
            Ok(())
        }
    }

    #[test]
    fn registry_starts_empty() {
        let r = ObserverRegistry::new();
        assert!(r.is_empty());
        assert_eq!(r.snapshot().len(), 0);
        assert!(!r.requirements().needs_before_value);
        assert!(!r.requirements().needs_prev_hlc);
    }

    #[test]
    fn registry_appends_in_registration_order() {
        let r = ObserverRegistry::new();
        r.register(Arc::new(WithReqs(ObserverRequirements::default())));
        r.register(Arc::new(NullObserver));
        let snap = r.snapshot();
        assert_eq!(snap.len(), 2);
        assert_eq!(snap[0].name(), "with-reqs");
        assert_eq!(snap[1].name(), "null");
    }

    #[test]
    fn registry_requirements_is_union() {
        let r = ObserverRegistry::new();
        r.register(Arc::new(WithReqs(ObserverRequirements {
            needs_before_value: true,
            needs_prev_hlc: false,
        })));
        r.register(Arc::new(WithReqs(ObserverRequirements {
            needs_before_value: false,
            needs_prev_hlc: true,
        })));
        let reqs = r.requirements();
        assert!(reqs.needs_before_value);
        assert!(reqs.needs_prev_hlc);
    }

    #[test]
    fn registry_snapshot_is_cheap_arc_clone() {
        // After register the snapshot reflects the new state; older
        // snapshots taken before the register still point at the
        // previous Arc — proving the swap is non-disruptive to readers.
        let r = ObserverRegistry::new();
        let before = r.snapshot();
        r.register(Arc::new(NullObserver));
        let after = r.snapshot();
        assert_eq!(before.len(), 0);
        assert_eq!(after.len(), 1);
    }
}