Module Algostream_data_ingestion.Connection_supervisor

Per-connection state machine: connect attempts, exponential backoff with jitter, circuit breaker, liveness tracking, and rate-limited Risk_alert emission.

The mutable state is owned by a single Lwt fiber, so every function here except mirror is NOT thread-safe across Domains. Observers on another Domain — a monitoring dashboard, say — must use mirror, which reads an immutable record published on each transition.

type state =
  1. | Connecting
  2. | Connected
  3. | Reconnecting of {
    1. attempt : int;
    2. next_at_ns : int64;
    }
  4. | Open_circuit of {
    1. until_ns : int64;
    }
type t
val current_state : t -> state
val exchange : t -> string
val note_attempt : t -> unit
val note_connected : t -> unit
val note_failure : t -> reason:string -> ?symbol:string -> unit -> unit

Records a failure and transitions to Reconnecting (or Open_circuit when the breaker trips). Optionally fires a Risk_alert with deduplication keyed on (code, symbol).

val note_message : t -> unit
val time_since_last_message_ns : t -> int64

Time since the last successful frame read, in nanoseconds. Int64.max_int before the first message.

val is_ready_to_attempt : t -> bool

Whether a connect attempt may be made now — false during open-circuit. Pure, so it is safe to call from anywhere, including an observer.

val poll_ready_to_attempt : t -> bool

As is_ready_to_attempt, but advances the state machine: an Open_circuit whose timer has expired is closed and returns to Connecting. This is the reconnect loop's entry point; use is_ready_to_attempt to observe without side effects.

val next_attempt_delay_ns : t -> int64

How long to sleep before the next attempt; 0 if ready now.

val read_timed_out : t -> bool

Returns true if there has been no inbound traffic for read_timeout_ms. Used to force a reconnect when the exchange goes silent.

val consecutive_failures : t -> int

Number of consecutive failures since the last successful connection.

val critical_drops : t -> int64

Counts of dropped Critical-band alerts since the supervisor started — exposed for stats.

val note_critical_drop : t -> unit

Record a Critical-band publish failure that occurred outside this module. Connector_runtime calls it when a Data_gap is dropped — exactly the loss critical_drops exists to surface.

type mirror = {
  1. exchange : string;
  2. state : state;
  3. consecutive_failures : int;
  4. critical_drops : int64;
  5. time_since_last_message_ns : int64;
}

Cross-Domain-safe view of the connection.

state, consecutive_failures and critical_drops come from an immutable record republished on every transition; time_since_last_message_ns is computed at call time and is Int64.max_int before the first message.

val mirror : t -> mirror