Struct Observer 

pub struct Observer { /* private fields */ }

Process-level structured logger and trace correlator.

Cloning this value is cheap. Emission uses try_send; output and JSON serialization happen on the dedicated writer thread.

Implementations

impl Observer

pub fn record_admission_capacity( &self, output: Option<&EmergencyDiagnosticHandle>, context: AdmissionEventContext<'_>, facts: &AdmissionCapacityFacts, construction: AdmissionConstruction, ) -> AdmissionCapacitySubmission

Emit original four-dimensional facts without synthetic Call/Event/root. Metrics use the existing fixed dimensions even when output is absent.

Examples found in repository

examples/admission_capacity.rs (line 16)

pub fn consume_original(
    original: ProfuseGwAdmissionObservationReceipt,
    observer: &Observer,
    output: Option<&EmergencyDiagnosticHandle>,
    context: AdmissionEventContext<'_>,
    construction: AdmissionConstruction,
) -> AdmissionCapacitySubmission {
    let facts = AdmissionCapacityFacts::from_receipt(&original);
    observer.record_admission_capacity(output, context, &facts, construction)
}

impl Observer

pub fn record_execution_pressure( &self, output: Option<&EmergencyDiagnosticHandle>, facts: ExecutionPressureFacts, ) -> DiagnosticSubmission

impl Observer

pub fn record_standard_rpc_shutdown( &self, output: Option<&EmergencyDiagnosticHandle>, joined: bool, rpc: &RpcStageSnapshot, ) -> DiagnosticSubmission

Original same-lock ledger sample after application driver join and physical pool reclaim. Values are not manufactured from join success.

impl Observer

pub fn record_ingress_fifo( &self, output: Option<&EmergencyDiagnosticHandle>, request_id: Option<ContextIdentity>, queue_sequence: u64, transition: &'static str, reason: &'static str, ) -> DiagnosticSubmission

impl Observer

pub fn record_ingress_fifo_attempt( &self, output: Option<&EmergencyDiagnosticHandle>, request_id: Option<ContextIdentity>, queue_sequence: u64, decision_attempt: u64, input_peak: usize, outcome: &'static str, ) -> DiagnosticSubmission

Attempt identity only. Resource amounts remain on the original same-lock capacity receipt; input_peak is the already recognized input demand.

impl Observer

pub fn record_resource_quiescent( &self, output: Option<&EmergencyDiagnosticHandle>, context: Option<RequestContextSnapshot>, completion_sequence: u64, resource: &ProcessSnapshot, rpc: &RpcStageSnapshot, ) -> DiagnosticSubmission

A post-join sample, not an admission decision or an atomic combined ledger receipt.

impl Observer

pub fn start_external_call( &self, application: impl Into<ApplicationId>, module: impl Into<ModuleId>, service: impl Into<ServiceId>, operation: impl Into<OperationId>, inbound_trace_id: Option<&str>, ) -> (ActiveCall, InboundTrace)

Starts an external request using the legacy compatibility policy.

Invalid supplied identifiers are replaced for compatibility with 0.3.0 callers. Production entry points must use Observer::start_external_call_checked so only a missing identifier can cause generation.

pub fn start_external_call_checked( &self, application: impl Into<ApplicationId>, module: impl Into<ModuleId>, service: impl Into<ServiceId>, operation: impl Into<OperationId>, inbound_trace_id: Option<&str>, ) -> Result<(ActiveCall, InboundTrace), TraceIdError>

Starts a production entry call. A supplied opaque protocol identifier is inherited byte-for-byte after bounded validation; only an absent identifier may cause generation.

pub fn start_external_call_with_rpc( &self, application: impl Into<ApplicationId>, module: impl Into<ModuleId>, service: impl Into<ServiceId>, operation: impl Into<OperationId>, inbound_trace_id: Option<&str>, rpc: RpcCorrelationId, ) -> Result<(ActiveCall, InboundTrace), TraceIdError>

Entry supplied protocol RPC, already validated by its typed carrier. No child RPC is created for local stages or database transactions.

Examples found in repository

examples/outbound_child_diagnostic.rs (lines 15-22)

fn main() {
    let observer = Observer::with_writer(ObserverConfig::default(), std::io::sink()).unwrap();
    let (incoming, _) = observer
        .start_external_call_with_rpc(
            "shop",
            "ingress",
            "orders",
            "query",
            Some("gateway-opaque"),
            RpcCorrelationId::new("0.3").unwrap(),
        )
        .unwrap();
    let ingress_event = EventContext::new(
        RequestIdentity::new("request-7").unwrap(),
        RouteIdentity::new("/incoming").unwrap(),
        1,
    )
    .unwrap();
    let parent = RequestDiagnosticScope::output_unavailable(incoming.context(), &ingress_event)
        .with_task(DiagnosticTaskId::from_runtime_id("42").unwrap())
        .unwrap_or_else(|_| panic!("task"))
        .with_zone(DiagnosticZone::from_validated_ingress("zone-a").unwrap());

    // programming.rs generates the wire child; Boundary uses its trace/rpc plus
    // request_id/function. The adapter below does NOT generate these identities.
    let wire_trace = "gateway-opaque";
    let wire_rpc = "0.3.1";
    let request_id = "request-7";
    let function = "remote.query";
    let (root, _) = observer
        .start_external_call_with_rpc(
            "saddle",
            "zone-a",
            "profusecontract",
            "invoke",
            Some(wire_trace),
            RpcCorrelationId::new(wire_rpc).unwrap(),
        )
        .unwrap();
    let event = EventContext::new(
        RequestIdentity::new(request_id).unwrap(),
        RouteIdentity::new(function).unwrap(),
        1,
    )
    .unwrap();
    let child = parent
        .derive_outbound_child(root.context(), &event)
        .unwrap();
    let failure = child.fail(
        std::io::ErrorKind::ConnectionRefused,
        DiagnosticCategory::UnexpectedError,
        BoundedDiagnosticCause::new(
            DiagnosticStage::RequestDecode,
            DiagnosticCode::new("outbound.connect_failed").unwrap(),
        ),
    );
    let id = failure.source_diagnostic().id();
    let (retained, delivery) =
        failure.finish_boundary_retained(None, &DiagnosticOutcomeAxes::default());
    assert_eq!(retained.source_diagnostic().id(), id);
    assert_eq!(*retained.error(), std::io::ErrorKind::ConnectionRefused);
    assert_eq!(
        delivery.source_submission(),
        DiagnosticSubmission::OutputUnavailable
    );
    assert_eq!(
        delivery.boundary_submission(),
        DiagnosticSubmission::OutputUnavailable
    );
    println!("GO: Boundary-shaped child Call/Event consumed; absent output retains same failure");
}

pub fn start_child_call( &self, parent: &CallContext, kind: CallKind, module: impl Into<ModuleId>, service: impl Into<ServiceId>, operation: impl Into<OperationId>, ) -> ActiveCall

Starts a Service, DB or transaction child call on the parent’s trace.

pub fn record_domain_event(&self, context: &CallContext, event: DomainEvent)

Records an intentional domain event without accepting arbitrary objects.

impl Observer

pub fn record_transaction_terminal( observation: TransactionTerminalObservation, outcome: TransactionOutcome, )

DB owning driver calls once before returning its result/physical receipt. Uses the carrier’s observer/context, with no caller-supplied scope or sink. Queue loss preserves the DB result; a missing log remains NOT_PROVEN.

use saddle_observability::{Observer, TransactionOutcome, TransactionTerminalObservation};
fn replay(observation: TransactionTerminalObservation) {
    Observer::record_transaction_terminal(observation, TransactionOutcome::Committed);
    Observer::record_transaction_terminal(observation, TransactionOutcome::Rejected);
}

pub fn record_lifecycle_timeout( &self, application: &str, stage: LifecycleTimeoutStage, elapsed_ms: u64, )

pub fn start_stage( &self, parent: &CallContext, stage: Stage, event_context: EventContext, ) -> ActiveStage

pub fn record_capacity( &self, context: &CallContext, event_context: &EventContext, value: CapacityObservation, )

pub fn observe_database_disposition(&self, disposition: DatabaseDisposition)

Update the existing fixed-label counter without allocating or emitting a second log. The DB owning terminal path calls this exactly once with its actual physical disposition, independently of operation outcome.

pub fn record_database_disposition( &self, context: &CallContext, event_context: &EventContext, disposition: DatabaseDisposition, )

pub fn record_resource_finalization( &self, context: &CallContext, event_context: &EventContext, disposition: DatabaseDisposition, elapsed_ms: u64, )

pub fn record_outbound( &self, context: &CallContext, event_context: &EventContext, observation: OutboundObservation, )

pub fn record_lifecycle( &self, context: &CallContext, event_context: &EventContext, state: LifecycleState, health: Health, )

pub fn record_logger_health( &self, context: &CallContext, event_context: &EventContext, dropped: u64, failure: Option<OutputStage>, )

impl Observer

pub fn record_diagnostic( &self, diagnostic: &Diagnostic, emergency: &EmergencyDiagnosticHandle, context: Option<(&CallContext, &EventContext)>, ) -> DiagnosticSubmission

Mirrors one safe diagnostic into independent emergency output and the existing logger. The result acknowledges only emergency queue admission.

impl Observer

pub fn with_writer( config: ObserverConfig, writer: impl Write + Send + 'static, ) -> Result<Self, InitError>

Creates an observer with a framework-owned writer.

Application startup should normally use init. This constructor is useful for embedding and deterministic tests.

Examples found in repository

examples/admission_capacity.rs (line 57)

fn main() {
    let process = example_profile();
    let observer = Observer::with_writer(ObserverConfig::default(), std::io::sink()).unwrap();
    let app = ContextLabel::checked("app").unwrap();
    let (permit, original) = match process.verified_profile().try_admit_observed() {
        ProfuseGwLightweightObservedAdmissionOutcome::Ready(permit, original) => (permit, original),
        _ => panic!("example first admission must succeed"),
    };
    let submission = consume_original(
        original,
        &observer,
        None,
        AdmissionEventContext::Unrooted {
            application: &app,
            lifecycle: RequestViewPhase::Admitted,
        },
        AdmissionConstruction::Failed(AdmissionConstructionFailure::Storage),
    );
    assert_eq!(submission.cpu, DiagnosticSubmission::OutputUnavailable);
    assert_eq!(submission.memory, DiagnosticSubmission::OutputUnavailable);
    assert_eq!(submission.database, DiagnosticSubmission::OutputUnavailable);
    assert_eq!(
        submission.profuse_contract,
        DiagnosticSubmission::OutputUnavailable
    );
    // The output did not release or remake admission: original permit remains.
    permit.into_execution().cancel();
    process.finish().unwrap();
    for (name, layout) in ["facts", "borrowed_context", "record", "submission"]
        .into_iter()
        .zip(admission_capacity_layouts())
    {
        println!(
            "AE layout {name}: bytes={} align={}",
            layout.size(),
            layout.align()
        );
    }
    let existing = root_diagnostic::request_logging_layouts();
    println!(
        "AE existing frame={} packet={} slots={} confirmation_owner={}",
        existing.source_frame.size(),
        existing.emergency_packet.size(),
        existing.emergency_slots,
        existing.confirmation_owner.size()
    );
    println!("AE consumer example GO; AF/Service production wiring NOT_RUN");
}

examples/outbound_child_diagnostic.rs (line 13)

fn main() {
    let observer = Observer::with_writer(ObserverConfig::default(), std::io::sink()).unwrap();
    let (incoming, _) = observer
        .start_external_call_with_rpc(
            "shop",
            "ingress",
            "orders",
            "query",
            Some("gateway-opaque"),
            RpcCorrelationId::new("0.3").unwrap(),
        )
        .unwrap();
    let ingress_event = EventContext::new(
        RequestIdentity::new("request-7").unwrap(),
        RouteIdentity::new("/incoming").unwrap(),
        1,
    )
    .unwrap();
    let parent = RequestDiagnosticScope::output_unavailable(incoming.context(), &ingress_event)
        .with_task(DiagnosticTaskId::from_runtime_id("42").unwrap())
        .unwrap_or_else(|_| panic!("task"))
        .with_zone(DiagnosticZone::from_validated_ingress("zone-a").unwrap());

    // programming.rs generates the wire child; Boundary uses its trace/rpc plus
    // request_id/function. The adapter below does NOT generate these identities.
    let wire_trace = "gateway-opaque";
    let wire_rpc = "0.3.1";
    let request_id = "request-7";
    let function = "remote.query";
    let (root, _) = observer
        .start_external_call_with_rpc(
            "saddle",
            "zone-a",
            "profusecontract",
            "invoke",
            Some(wire_trace),
            RpcCorrelationId::new(wire_rpc).unwrap(),
        )
        .unwrap();
    let event = EventContext::new(
        RequestIdentity::new(request_id).unwrap(),
        RouteIdentity::new(function).unwrap(),
        1,
    )
    .unwrap();
    let child = parent
        .derive_outbound_child(root.context(), &event)
        .unwrap();
    let failure = child.fail(
        std::io::ErrorKind::ConnectionRefused,
        DiagnosticCategory::UnexpectedError,
        BoundedDiagnosticCause::new(
            DiagnosticStage::RequestDecode,
            DiagnosticCode::new("outbound.connect_failed").unwrap(),
        ),
    );
    let id = failure.source_diagnostic().id();
    let (retained, delivery) =
        failure.finish_boundary_retained(None, &DiagnosticOutcomeAxes::default());
    assert_eq!(retained.source_diagnostic().id(), id);
    assert_eq!(*retained.error(), std::io::ErrorKind::ConnectionRefused);
    assert_eq!(
        delivery.source_submission(),
        DiagnosticSubmission::OutputUnavailable
    );
    assert_eq!(
        delivery.boundary_submission(),
        DiagnosticSubmission::OutputUnavailable
    );
    println!("GO: Boundary-shaped child Call/Event consumed; absent output retains same failure");
}

pub fn metrics_snapshot(&self) -> MetricsSnapshot

Returns a fixed-size, allocation-free copy of the current process metrics.

pub fn flush(&self) -> FlushFuture

Flushes records accepted before the lifecycle coordinator reaches the writer. The returned future never blocks the async executor thread.

pub fn shutdown(&self) -> FlushFuture

Stops admission, drains accepted records, flushes output and joins the writer without blocking the async executor thread.

pub fn dropped_events(&self) -> u64

Trait Implementations

impl Clone for Observer

fn clone(&self) -> Self

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl ComponentLifecycle for Observer

fn name(&self) -> &'static str

fn start(&self) -> LifecycleFuture<'_>

fn shutdown(&self) -> LifecycleFuture<'_>

fn shutdown_with_primary( &self, _primary: Option<DiagnosticOccurrence>, ) -> Pin<Box<dyn Future<Output = Result<(), SaddleError>> + Send + '_>>

The cleanup owner may correlate its original source with an existing primary before recording. Components without their own source writer retain the ordinary shutdown behavior.

impl RecordedComponentLifecycle for Observer

The formal component boundary keeps the actual FlushError until the controlled source file has acknowledged it. The Core lifecycle impl above remains a legacy compatibility surface for callers migrating to this one.

fn name(&self) -> &'static str

fn start<'a>( &'a self, application: &'a ContextLabel, output: &'a SourceOutput, ) -> RecordedLifecycleFuture<'a>

fn shutdown<'a>( &'a self, application: &'a ContextLabel, output: &'a SourceOutput, ) -> RecordedLifecycleFuture<'a>

fn shutdown_with_primary<'a>( &'a self, application: &'a ContextLabel, output: &'a SourceOutput, primary: Option<DiagnosticOccurrence>, ) -> RecordedLifecycleFuture<'a>