Move reconnect logic into Live (session-style client), add reconnect callback

[tbeason]

Background

PR #16 added exponential backoff with jitter + max_reconnect_attempts to the reconnect loop, but the logic still lives inside _run_unified_session at the streaming layer. That means:

• Users of stream_to_file / stream_multi_to_files get reconnect.
• Users iterating directly over a Live client (for rec in client) or driving subscribe_callback get no reconnect at all — Live is single-shot today and close(c) is the only end-state.

This issue tracks the architectural follow-up: move reconnect into the Live lineage so it's available to any consumer, and borrow useful pieces from how the official Python client structures the same problem.

Why it's currently above Live and not inside

The reconnect logic is coupled to two pieces of state:

1. Stored subscriptions, so we can re-issue them on a fresh connection.
2. Per-instrument timestamps (last_ts_recv_by_id / last_ts_event_by_id in SessionStats), so the new subscribe's start = picks up at the lowest seen timestamp — bridging the gap rather than producing one.

Live doesn't own either today. The streaming layer does (because it's the layer that consumes records and decides what to do with them).

How Python solves the same coupling

databento-python (see databento/live/session.py) puts reconnect inside the LiveSession (which sits inside the Live client). The session naturally owns:

• self._subscriptions: list[tuple[SubscriptionRequest, ...]]
• self._protocol._last_ts_event (via the connected protocol object)

API surface is deliberately minimal:

ReconnectPolicy.NONE | ReconnectPolicy.RECONNECT       # one enum kwarg on Live
Live.add_reconnect_callback(callback)                  # observe each reconnect

Their reconnect task is roughly:

async def _reconnect(self):
    while True:
        try:
            await self._protocol.disconnected
        except Exception:
            gap_start = pd.Timestamp(self._protocol._last_ts_event)
            self._transport, self._protocol = await self._connect_task(...)
            for sub in stored_subscriptions:
                self._protocol.subscribe(..., start=None)   # NB: start=None
            if should_restart:
                self._protocol.start()
                metadata = await new_metadata
                gap_end = pd.Timestamp(metadata.start)
                dispatch_reconnect_callbacks(gap_start, gap_end)
        else:
            return

Key choices they made:

	Python	Our PR #16
Backoff	None — immediate	Full-jitter exponential (1s base, 60s cap)
Max attempts	Unlimited	10 (configurable, nothing = unlimited)
Replay strategy	start = None (resume at gateway-current; produces a gap)	start = min(last_per-instrument_ts) (replays from oldest seen; produces some duplicates at the boundary)
Gap handling	(gap_start, gap_end) callback for user to gap-fill via historical API	None — implicit via record stream
Layer	Inside LiveSession (client lineage)	Above Live (streaming functions)

Python is more conservative on user-facing config but more useful on the gap-reporting side. We're the opposite.

Proposed direction

Structural change

Restructure Live to be a session-style client that owns:

• subscriptions::Vector{SubscriptionRequest} (stored across reconnects)
• A ReplayState that owns the per-instrument timestamps that today live in SessionStats
• An optional reconnect task

Iteration (for rec in client) and subscribe_callback callers get reconnect for free.

Keep PR #16's behaviors

Don't regress to Python's "unlimited immediate retry." Carry over:

• Full-jitter exponential backoff (1s base, 60s cap)
• max_reconnect_attempts with default 10, nothing for unlimited
• Per-instrument timestamp replay (smarter than start = None)
• 24h replay-window cap
• BBO-family uses ts_recv; others use ts_event
• ErrorMsg-is-terminal stays terminal

Borrow from Python

• Reconnect callback — add_reconnect_callback(client, (gap_start, gap_end) -> Nothing). Useful regardless of layer; we don't have it today. Users who want to gap-fill from historical can hook in here.
• ReconnectPolicy enum as the public-facing knob (:none / :reconnect), with the lower-level kwargs (max_reconnect_attempts, base, cap) as advanced opts that default to sensible values. Cleaner than two booleans.

API sketch

client = Live(api_key;
    dataset = \"OPRA.PILLAR\",
    reconnect_policy        = :reconnect,        # :none (default) | :reconnect
    max_reconnect_attempts  = 10,                # nothing = unlimited
    # backoff is module-internal unless we hear a use case
)
add_reconnect_callback(client) do gap_start, gap_end
    # user-defined gap-fill via historical API, alerting, metric emission, etc.
end
connect!(client)
subscribe!(client; schema = Schema.TRADES, symbols = [\"AAPL\"])
start!(client)
for rec in client            # iteration silently bridges reconnects
    handle(rec)
end

stream_to_file / stream_multi_to_files become thin callers of this — they delete their bespoke reconnect loop entirely, keep their replay-state hook, and add the existing max_reconnect_attempts kwarg as a passthrough.

Scope / non-scope

In scope:

• Refactor Live into a session-style client with internal reconnect.
• Add ReconnectPolicy enum + add_reconnect_callback.
• Migrate _run_unified_session reconnect loop into Live, deleting the streaming-layer version.
• Update tests; the existing test_reconnect_backoff.jl integration test should pass against the new layer with minimal changes (probably just a different entry point).

Out of scope (separate issues if we want them):

• Changing replay semantics from per-instrument-min-ts to start = None + callback. (Possibly worth offering both, but not in this PR.)
• Exposing backoff base/cap as user kwargs.
• Heartbeat-monitor refactor (Python has a separate task; we have inline staleness logic).

Risks

• This is a non-trivial refactor of Live's state machine: connected → reconnecting → closed instead of connected → closed. Concurrency around the reconnect task and user iteration needs care.
• API change to Live kwargs is a breaking change in the Live constructor. Pre-1.0 is acceptable, but we should bundle it with the version bump and document clearly.
• If we add the callback, we need to lock around list mutation; Python uses threading.RLock.

References

• Python reconnect implementation: https://github.com/databento/databento-python/blob/main/databento/live/session.py (search for _reconnect)
• ReconnectPolicy: https://github.com/databento/databento-python/blob/main/databento/common/enums.py (search for ReconnectPolicy)
• Live.add_reconnect_callback: https://github.com/databento/databento-python/blob/main/databento/live/client.py
• PR that added the streaming-layer version: Reconnect: exponential backoff with jitter + max_reconnect_attempts #16

🤖 Generated with Claude Code