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 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={}",
        existing.source_frame.size(),
        existing.emergency_packet.size(),
        existing.emergency_slots
    );
    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");
}

examples/request_root_contract.rs (line 142)

fn main() {
    // Storage contract exercise only. R0/A/R must reserve the exact layouts before
    // this constructor is enabled in a real admitted or rejection path.
    let mut publisher = RequestRootPublisher::create(
        ContextLabel::checked("saddle").unwrap(),
        ContextFact::NotEstablished,
    )
    .unwrap();
    let root = publisher.reference();
    let early = root.view(
        RequestLocalFacts::new(RequestViewPhase::Reading).with_task(ContextFact::Present(17)),
    );
    let read_error = std::io::Error::from(std::io::ErrorKind::UnexpectedEof);
    let early_failure = RootDiagnosticScope::new(&early, None).source_error(
        &read_error,
        DiagnosticStage::RequestDecode,
        RootRequestEvent::Ingress,
        RootOutcomeFacts::default(),
    );
    let event = EventContext::new(
        RequestIdentity::new("request-a").unwrap(),
        RouteIdentity::new("example.live").unwrap(),
        1,
    )
    .unwrap();
    assert!(
        publisher
            .publish(
                request_identity_group(
                    &call("0", 1, "entry", "app", "example.live"),
                    &event,
                    ContextFact::Present(ContextLabel::checked("local").unwrap())
                )
                .unwrap()
            )
            .is_ok()
    );
    let active = early
        .refresh(&root)
        .unwrap()
        .with_phase(RequestViewPhase::Handler);
    // Formal Boundary style: real child Call and event originate from its actual
    // ProfuseContract request. Obs validates rather than inventing a child RPC.
    let child_event = EventContext::new(
        RequestIdentity::new("request-a").unwrap(),
        RouteIdentity::new("receipt.invoke").unwrap(),
        1,
    )
    .unwrap();
    let child = request_child_view(
        &active,
        &call("0.1", 2, "local", "profusecontract", "invoke"),
        &child_event,
    )
    .unwrap();
    assert!(active.same_request(&child));
    let (_, issuer) = DbPhysicalDispositionIssuer::issue().into_startup_and_request_issuer();
    let (request, execution) = issuer.issue_request().unwrap();
    // Real D passes &sqlx::Error here before driver_ref/business mapping. This
    // standalone example uses the standard Error contract, with no DB dependency.
    let raw_error = std::io::Error::new(
        std::io::ErrorKind::PermissionDenied,
        "original driver description",
    );
    let db_failure = database_source(&active, &request, &execution, None, &raw_error);
    drop(raw_error);
    let supervised = RootSupervisionReturn::<()>::failed(
        db_failure.map_classification(DiagnosticCode::new("handler.failed").unwrap()),
    );
    let failure = supervised.consume().err().unwrap();
    // Public projections contain no source body or root. Real Runtime must also
    // finish the original physical pair, then drop all views before settlement.
    let public = failure
        .finish(&active, None, RootOutcomeFacts::default())
        .ok()
        .unwrap();
    let early_public = early_failure
        .finish(&active, None, RootOutcomeFacts::default())
        .ok()
        .unwrap();
    let observer = Observer::with_writer(ObserverConfig::default(), std::io::sink()).unwrap();
    let completed = RootDiagnosticScope::new(&active, None)
        .start_stage(&observer, RootRequestEvent::Response)
        .finish_nonfailure(RootOutcomeFacts {
            axes: DiagnosticOutcomeAxes {
                operation: OperationOutcome::Succeeded,
                ..Default::default()
            },
            ..Default::default()
        })
        .ok()
        .unwrap();
    assert_eq!(completed, DiagnosticSubmission::OutputUnavailable);
    assert_eq!(
        RootDiagnosticScope::new(&active, None).ordinary(
            &observer,
            RootRequestEvent::Handler,
            RootOutcomeFacts::default()
        ),
        DiagnosticSubmission::Enqueued
    );
    drop((publisher, root, early, active, child));
    assert_eq!(
        public.source_submission(),
        DiagnosticSubmission::OutputUnavailable
    );
    assert_eq!(
        early_public.terminal_submission(),
        DiagnosticSubmission::OutputUnavailable
    );
    for (name, (payload, shared)) in ["root", "view", "child"]
        .into_iter()
        .zip(saddle_core::request_context::request_context_layouts())
    {
        println!(
            "layout {name}: payload={} align={} arc_allocation={} arc_align={}",
            payload.size(),
            payload.align(),
            shared.size(),
            shared.align()
        );
    }
    println!(
        "layout receipt={} public={} root_ref={} view_ref={} identity_input={} bounded_detail={} borrowed_stage={}",
        size_of::<RootRequestFailure>(),
        size_of::<PublicRequestFailure>(),
        size_of::<RequestRootRef>(),
        size_of::<RequestExecutionView>(),
        size_of::<RequestIdentityGroup>(),
        size_of::<BoundedDiagnostic>(),
        size_of::<RootActiveStage<'static>>()
    );
    println!("C consumer contract GO; production reservation/lifecycle NOT_RUN");
    let layout = saddle_observability::root_diagnostic::request_logging_layouts();
    let (stream, cycle, payload) = saddle_observability::root_diagnostic::original_capture_layout();
    println!(
        "layout original_stream={} align={} cycle_detector={} align={} payload_limit={} total_content_limit=none",
        stream.size(),
        stream.align(),
        cycle.size(),
        cycle.align(),
        payload
    );
    println!(
        "layout source_frame={} source_record={} emergency_packet={} emergency_slots={} ordinary_command={} ordinary_payload_max={}",
        layout.source_frame.size(),
        layout.source_record.size(),
        layout.emergency_packet.size(),
        layout.emergency_slots,
        layout.ordinary_command.size(),
        layout.ordinary_encoded_bytes_max
    );
}

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<'_>