saga/telemetry

Defines the Sinal events a run emits: run start and stop, step start and stop, compensation decisions, and undo outcomes.

Use this module to observe runs started with saga/execution or saga/durable. Each function returns a typed sinal.Event descriptor; attach a handler to it with sinal.observe or sinal.attach. This module owns the events, not handler registration.

Every metadata record carries:

workflow: the workflow’s name;
run: this VM’s id for one run; a durable execution gets a new one on every durable.drive;
execution: the durable execution id given to durable.start_or_reconnect, or None for a local run;
correlation: the sinal/correlation value set with execution.with_correlation or durable.with_correlation. A durable execution that set none carries correlation.from_key of its execution id; a local run that set none gets a fresh correlation.unique() at start, so the field is always a Correlation. A durable execution saves the value of its first drive, so every drive of it reports the same one. The steps of the run read the same value with saga.correlation_of of their saga.EffectKey.

Read metadata fields by label: a later release may add fields.

A run’s coordinator emits each event after the state change it describes, with sinal.emit. Handlers run synchronously in the coordinator process unless the application routes the saga prefix to a sinal/forwarder: a slow inline handler delays the run. Observations never control a run.

import saga/telemetry
import sinal

let attachment =
  sinal.observe(telemetry.run_stopped(), fn(_measurements, metadata) {
    log_outcome(metadata.correlation, metadata.workflow, metadata.outcome)
  })

Types

AttemptKind

</>

The closed set of ways one attempt can end, for [saga, step, stop]. AttemptUnknown is a returned error that saga.unknown_when classified as a possible effect.

pub type AttemptKind {
  AttemptSucceeded
  AttemptFailed
  AttemptUnknown
  AttemptCrashed
  AttemptTimedOut
  AttemptInterrupted
}

Constructors

```
AttemptSucceeded
```
```
AttemptFailed
```
```
AttemptUnknown
```
```
AttemptCrashed
```
```
AttemptTimedOut
```
```
AttemptInterrupted
```

CompensationMetadata

</>

[saga, step, compensate, stop] metadata. retry_delay is the delay that a RetryAfter decision was scheduled with after the execution.with_max_retry_delay cap (carried as whole milliseconds in the event’s map), and retry_delay_capped is True when the cap shortened the requested delay. They are None and False for every other decision.

pub type CompensationMetadata {
  CompensationMetadata(
    workflow: String,
    run: Int,
    execution: option.Option(String),
    correlation: correlation.Correlation,
    step: String,
    attempt: Int,
    decision: DecisionKind,
    retry_delay: option.Option(duration.Duration),
    retry_delay_capped: Bool,
  )
}

Constructors

CompensationMetadata(
  workflow: String,
  run: Int,
  execution: option.Option(String),
  correlation: correlation.Correlation,
  step: String,
  attempt: Int,
  decision: DecisionKind,
  retry_delay: option.Option(duration.Duration),
  retry_delay_capped: Bool,
)

DecisionKind

</>

The closed set of recovery decisions, for [saga, step, compensate, stop].

pub type DecisionKind {
  DecisionRetry
  DecisionContinue
  DecisionAbort
  DecisionHold
  DecisionCrashed
  DecisionTimedOut
}

Constructors

```
DecisionRetry
```
```
DecisionContinue
```
```
DecisionAbort
```
```
DecisionHold
```
```
DecisionCrashed
```
```
DecisionTimedOut
```

OutcomeKind

</>

The closed set of ways a run can end. execution.kind returns it for an execution.Outcome, and [saga, run, stop] carries it.

pub type OutcomeKind {
  OutcomeCompleted
  OutcomeCompletedWithUnknownEffects
  OutcomeFailed
  OutcomeCancelled
  OutcomeUnresolved
}

Constructors

```
OutcomeCompleted
```
```
OutcomeCompletedWithUnknownEffects
```
```
OutcomeFailed
```
```
OutcomeCancelled
```
```
OutcomeUnresolved
```

RunMetadata

</>

[saga, run, start] metadata.

pub type RunMetadata {
  RunMetadata(
    workflow: String,
    run: Int,
    execution: option.Option(String),
    correlation: correlation.Correlation,
  )
}

Constructors

RunMetadata(
  workflow: String,
  run: Int,
  execution: option.Option(String),
  correlation: correlation.Correlation,
)

RunStartMeasurements

</>

[saga, run, start] measurements: native system time in milliseconds.

pub type RunStartMeasurements {
  RunStartMeasurements(system_time: Int)
}

Constructors

```
RunStartMeasurements(system_time: Int)
```

RunStopMeasurements

</>

[saga, run, stop] measurements: total duration and settlement counts.

pub type RunStopMeasurements {
  RunStopMeasurements(
    duration: Int,
    undone: Int,
    undo_failures: Int,
    interrupted: Int,
  )
}

Constructors

RunStopMeasurements(
  duration: Int,
  undone: Int,
  undo_failures: Int,
  interrupted: Int,
)

RunStopMetadata

</>

[saga, run, stop] metadata.

pub type RunStopMetadata {
  RunStopMetadata(
    workflow: String,
    run: Int,
    execution: option.Option(String),
    correlation: correlation.Correlation,
    outcome: OutcomeKind,
  )
}

Constructors

RunStopMetadata(
  workflow: String,
  run: Int,
  execution: option.Option(String),
  correlation: correlation.Correlation,
  outcome: OutcomeKind,
)

StepMetadata

</>

[saga, step, start] metadata: the step address (rendered with saga.address_to_string) and the 1-based attempt number.

pub type StepMetadata {
  StepMetadata(
    workflow: String,
    run: Int,
    execution: option.Option(String),
    correlation: correlation.Correlation,
    step: String,
    attempt: Int,
  )
}

Constructors

StepMetadata(
  workflow: String,
  run: Int,
  execution: option.Option(String),
  correlation: correlation.Correlation,
  step: String,
  attempt: Int,
)

StepStartMeasurements

</>

[saga, step, start] measurements: native system time in milliseconds.

pub type StepStartMeasurements {
  StepStartMeasurements(system_time: Int)
}

Constructors

```
StepStartMeasurements(system_time: Int)
```

StepStopMeasurements

</>

[saga, step, stop], [saga, step, compensate, stop], and [saga, step, undo, stop] share this measurement shape: the duration of the attempt, compensation decision, or undo action, in milliseconds.

pub type StepStopMeasurements {
  StepStopMeasurements(duration: Int)
}

Constructors

```
StepStopMeasurements(duration: Int)
```

StepStopMetadata

</>

[saga, step, stop] metadata.

pub type StepStopMetadata {
  StepStopMetadata(
    workflow: String,
    run: Int,
    execution: option.Option(String),
    correlation: correlation.Correlation,
    step: String,
    attempt: Int,
    result: AttemptKind,
  )
}

Constructors

StepStopMetadata(
  workflow: String,
  run: Int,
  execution: option.Option(String),
  correlation: correlation.Correlation,
  step: String,
  attempt: Int,
  result: AttemptKind,
)

UndoKind

</>

The closed set of ways one undo action can end, for [saga, step, undo, stop].

pub type UndoKind {
  UndoUndone
  UndoFailedKind
  UndoCrashedKind
  UndoTimedOutKind
}

Constructors

```
UndoUndone
```
```
UndoFailedKind
```
```
UndoCrashedKind
```
```
UndoTimedOutKind
```

UndoMetadata

</>

[saga, step, undo, stop] metadata.

pub type UndoMetadata {
  UndoMetadata(
    workflow: String,
    run: Int,
    execution: option.Option(String),
    correlation: correlation.Correlation,
    step: String,
    result: UndoKind,
  )
}

Constructors

UndoMetadata(
  workflow: String,
  run: Int,
  execution: option.Option(String),
  correlation: correlation.Correlation,
  step: String,
  result: UndoKind,
)

Values

compensation_stopped

</>

pub fn compensation_stopped() -> sinal.Event(
  StepStopMeasurements,
  CompensationMetadata,
)

The [saga, step, compensate, stop] event descriptor.

outcome_kind_name

</>

pub fn outcome_kind_name(kind: OutcomeKind) -> String

The stable name of an outcome kind, as [saga, run, stop] encodes it: "completed", "completed_with_unknown_effects", "failed", "cancelled" or "unresolved".

run_started

</>

pub fn run_started() -> sinal.Event(
  RunStartMeasurements,
  RunMetadata,
)

The [saga, run, start] event descriptor.

run_stopped

</>

pub fn run_stopped() -> sinal.Event(
  RunStopMeasurements,
  RunStopMetadata,
)

The [saga, run, stop] event descriptor.

step_started

</>

pub fn step_started() -> sinal.Event(
  StepStartMeasurements,
  StepMetadata,
)

The [saga, step, start] event descriptor.

step_stopped

</>

pub fn step_stopped() -> sinal.Event(
  StepStopMeasurements,
  StepStopMetadata,
)

The [saga, step, stop] event descriptor.

undo_stopped

</>

pub fn undo_stopped() -> sinal.Event(
  StepStopMeasurements,
  UndoMetadata,
)

The [saga, step, undo, stop] event descriptor.