Live symbol resolution (instrument_id → symbol)

CHANGELOG.md

@@ -7,6 +7,17 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added
+- **Live symbol resolution.** A `Live` client now keeps a running
+  `instrument_id → symbol` map, built by the reader from the gateway's
+  `SymbolMappingMsg` records. `symbol_for(client, instrument_id_or_record;
+  which=:in)` resolves the symbol for a data record (`which=:out` for the output
+  symbol), `symbol_map(client; which=:in)` returns a snapshot, and
+  `add_symbol_mapping_callback(client, cb)` fires `cb(instrument_id, msg)` on each
+  (re)mapping. Works in both typed and untyped mode and survives reconnects; in
+  typed mode the mapping is captured before any `control_channel` overflow drop,
+  so it's reliable even without draining the control channel.
+
 ## [0.3.0] - 2026-06-22
 
 ### Added

docs/src/api/live.md

@@ -38,3 +38,16 @@ live_session
 ```@docs
 add_reconnect_callback
 ```
+
+## Symbol resolution
+
+Live records carry only the numeric `instrument_id`; the reader maintains a
+running `instrument_id → symbol` map from the gateway's `SymbolMappingMsg`
+records (in both typed and untyped mode, without needing to drain
+`control_channel`).
+
+```@docs
+symbol_for(::Live, ::Integer)
+symbol_map(::Live)
+add_symbol_mapping_callback
+```

src/DatabentoAPI.jl

@@ -52,7 +52,9 @@ using URIs
 
 import DatabentoBinaryEncoding as DBN
 using DatabentoBinaryEncoding: Schema, SType, Compression, Encoding, Action, Side, InstrumentClass
-using DatabentoBinaryEncoding: symbol_map, symbol_for, add_symbol_column!
+# `import` (not `using`) so the Live client can add `symbol_map`/`symbol_for`
+# methods (extending the DBN functions) for live, current-state resolution.
+import DatabentoBinaryEncoding: symbol_map, symbol_for, add_symbol_column!
 
 include("errors.jl")
 include("enums.jl")
@@ -76,6 +78,7 @@ include("live/subscribe.jl")
 include("live/heartbeat.jl")
 include("live/streaming.jl")
 include("live/reconnect.jl")
+include("live/symbols.jl")
 
 # Re-exported DatabentoBinaryEncoding enums (single import for users)
 export Schema, SType, Compression, Encoding, Action, Side, InstrumentClass
@@ -121,4 +124,7 @@ export open_dbn_writer, write_record!
 # Live reconnect callback
 export add_reconnect_callback
 
+# Live symbol resolution (symbol_map / symbol_for also re-exported above)
+export add_symbol_mapping_callback
+
 end # module DatabentoAPI

src/live/client.jl

@@ -98,6 +98,18 @@ mutable struct Live
     # gap_end in the reconnect callback.
     last_metadata_start_ns::Union{Nothing,Int64}
 
+    # Running instrument_id → SymbolMappingMsg map, built by the reader as
+    # SymbolMappingMsg records arrive. Captured in the reader *before* any
+    # control-channel drop, so it stays reliable even when control_channel
+    # overflows. Owned by the client, so it survives reconnects (the gateway
+    # re-sends mappings on re-subscribe). `symbols_lock` guards it: the reader
+    # task writes, consumer tasks read. The full record is stored so callers
+    # can resolve either the input or output symbol (and its [start,end] window).
+    symbols_by_id::Dict{UInt32,DBN.SymbolMappingMsg}
+    symbols_lock::ReentrantLock
+    # Callbacks fired on each SymbolMappingMsg: `cb(instrument_id, msg)`.
+    symbol_callbacks::Vector{Any}
+
     # Reconnect policy + retry budget.
     reconnect_policy::ReconnectPolicy.T
     max_reconnect_attempts::Union{Int,Nothing}
@@ -188,6 +200,9 @@ function Live(key::Union{Nothing,AbstractString} = nothing;
         # --- reconnect fields ---
         Dict{UInt32,Int64}(), Dict{UInt32,Int64}(),
         nothing,                                 # last_metadata_start_ns
+        Dict{UInt32,DBN.SymbolMappingMsg}(),     # symbols_by_id
+        ReentrantLock(),                         # symbols_lock
+        Any[],                                   # symbol_callbacks
         rp,
         max_reconnect_attempts === nothing ? nothing : Int(max_reconnect_attempts),
         Int(immediate_reconnect_attempts),

src/live/reader.jl

@@ -295,6 +295,10 @@ function _reader_loop_typed(c::Live)
                         c.terminal_error = "gateway ErrorMsg"
                     end
                 end
+                # Update the symbol map before the drop decision below, so it
+                # stays correct even when control_channel overflows (a parent/
+                # continuous subscription floods SymbolMappingMsg at start).
+                rec === nothing || _record_symbol!(c, rec)
                 if rec !== nothing && ctrl_chan !== nothing && isopen(ctrl_chan)
                     # Non-blocking put. This reader is the SOLE producer to
                     # ctrl_chan (reconnect spawns readers sequentially, never
@@ -442,6 +446,9 @@ function _reader_loop(c::Live)
             # reconnect supervisor sees timestamps from records that haven't
             # been consumed yet.
             _record_replay!(c, rec)
+            # Maintain the running instrument_id → symbol map (no-op unless the
+            # record is a SymbolMappingMsg).
+            _record_symbol!(c, rec)
             # ErrorMsg sets c.terminal_error *before* the put! so the
             # supervisor sees it once the reader exits and refuses to
             # reconnect — gateway-side errors are deterministic and

src/live/symbols.jl

@@ -0,0 +1,115 @@
+# Live symbol resolution.
+#
+# Live data records carry only the opaque numeric `hd.instrument_id`. The gateway
+# announces the human-readable symbol for each id out-of-band via
+# `SymbolMappingMsg` records (a burst at subscription start, then occasional
+# updates — e.g. a continuous/parent symbol rolling to a new contract). This
+# layer keeps a running `instrument_id → SymbolMappingMsg` map on the `Live`
+# client so consumers can resolve a record's symbol without manually tracking
+# those control records.
+#
+# This is the live counterpart to the historical/DBN side, where the mapping is
+# read once from `Metadata.mappings`. The difference: the historical map is
+# interval-keyed (an id is reused across a date range, so `symbol_for` there
+# takes a timestamp), whereas the live map is *current state* — it reflects the
+# latest mapping seen, so the live `symbol_for` needs no timestamp.
+
+# Called by both reader loops on every control record; updates the map (and
+# fires callbacks) only for SymbolMappingMsg. Invoked before the typed reader's
+# control-channel drop logic, so the map is captured even if control_channel is
+# full and the record itself is dropped.
+@inline function _record_symbol!(c::Live, rec)
+    rec isa DBN.SymbolMappingMsg || return nothing
+    iid = rec.hd.instrument_id
+    @lock c.symbols_lock c.symbols_by_id[iid] = rec
+    _fire_symbol_callbacks(c, iid, rec)
+    return nothing
+end
+
+# Snapshot callbacks under the lock, then fire outside it so a slow callback
+# can't block the reader (mirrors _fire_reconnect_callbacks).
+function _fire_symbol_callbacks(c::Live, iid::UInt32, rec::DBN.SymbolMappingMsg)
+    isempty(c.symbol_callbacks) && return nothing
+    cbs = @lock c.symbols_lock copy(c.symbol_callbacks)
+    for cb in cbs
+        try
+            cb(iid, rec)
+        catch e
+            @error "symbol-mapping callback raised" dataset = c.dataset exception = (e, catch_backtrace())
+        end
+    end
+    return nothing
+end
+
+# Pull the requested symbol string out of a SymbolMappingMsg.
+@inline function _symbol_field(msg::DBN.SymbolMappingMsg, which::Symbol)
+    which === :in && return msg.stype_in_symbol
+    which === :out && return msg.stype_out_symbol
+    throw(ArgumentError("`which` must be :in or :out, got $(repr(which))"))
+end
+
+"""
+    symbol_for(client::Live, instrument_id; which=:in) -> Union{String,Nothing}
+    symbol_for(client::Live, record;        which=:in) -> Union{String,Nothing}
+
+Resolve the human-readable symbol the gateway last mapped to `instrument_id`
+(or to `record.hd.instrument_id`), or `nothing` if no `SymbolMappingMsg` has been
+seen for that id yet. The map is maintained automatically by the reader as
+mappings stream in, so this works in both typed and untyped mode and without
+draining `control_channel`.
+
+`which` selects which side of the mapping to return:
+- `:in` (default) — the input symbol you subscribed with (e.g. `"AAPL"` for a
+  raw-symbol subscription, or the parent/continuous symbol such as `"ES.FUT"`).
+- `:out` — the resolved output symbol (`stype_out_symbol`).
+
+See also [`symbol_map`](@ref) for a full snapshot and
+[`add_symbol_mapping_callback`](@ref) for push-style updates.
+"""
+function symbol_for(c::Live, instrument_id::Integer; which::Symbol = :in)
+    msg = @lock c.symbols_lock get(c.symbols_by_id, UInt32(instrument_id), nothing)
+    return msg === nothing ? nothing : _symbol_field(msg, which)
+end
+
+symbol_for(c::Live, rec; which::Symbol = :in) =
+    symbol_for(c, rec.hd.instrument_id; which = which)
+
+"""
+    symbol_map(client::Live; which=:in) -> Dict{UInt32,String}
+
+Return a thread-safe snapshot of the current `instrument_id → symbol` map for a
+live `client`, built from the `SymbolMappingMsg` records seen so far. `which`
+(`:in` default, or `:out`) selects which side of each mapping to use — see
+[`symbol_for`](@ref).
+
+The returned `Dict` is a copy; mutating it does not affect the client, and the
+client may add entries after the snapshot is taken.
+"""
+function symbol_map(c::Live; which::Symbol = :in)
+    out = Dict{UInt32,String}()
+    @lock c.symbols_lock begin
+        sizehint!(out, length(c.symbols_by_id))
+        for (iid, msg) in c.symbols_by_id
+            out[iid] = _symbol_field(msg, which)
+        end
+    end
+    return out
+end
+
+"""
+    add_symbol_mapping_callback(client::Live, cb)
+
+Register `cb(instrument_id::UInt32, msg::SymbolMappingMsg)` to be invoked each
+time the gateway maps (or remaps) an instrument id — useful for reacting to
+continuous/parent symbol rolls or persisting the mapping. Callbacks fire from
+the reader task; they are snapshotted under a lock and run outside it, and an
+exception in a callback is logged and swallowed (it won't disrupt the reader or
+other callbacks). The map itself is updated before callbacks fire, so
+[`symbol_for`](@ref) already reflects the new mapping inside the callback.
+"""
+function add_symbol_mapping_callback(c::Live, cb)
+    cb isa Function || cb isa Base.Callable || throw(ArgumentError(
+        "symbol-mapping callback must be callable, got $(typeof(cb))"))
+    @lock c.symbols_lock push!(c.symbol_callbacks, cb)
+    return nothing
+end

test/runtests.jl

@@ -26,6 +26,7 @@ try
         include("test_reconnect_backoff.jl")
         include("test_live_reconnect_supervisor.jl")
         include("test_typed_mode_errors.jl")
+        include("test_live_symbols.jl")
     end
 catch e
     global offline_failed = true

test/test_live_symbols.jl

@@ -0,0 +1,87 @@
+# Live symbol resolution: the running instrument_id → symbol map maintained by
+# the reader, plus symbol_for / symbol_map / add_symbol_mapping_callback.
+#
+# These exercise the map directly via the reader hook `_record_symbol!` (no TCP
+# needed) — the reader-loop integration is covered by the live reader tests.
+
+@testset "Live symbol resolution" begin
+    _smm(iid, insym, outsym) = DBN.SymbolMappingMsg(
+        DBN.RecordHeader(UInt8(0), DBN.RType.SYMBOL_MAPPING_MSG, UInt16(1), UInt32(iid), Int64(0)),
+        DBN.SType.PARENT, insym, DBN.SType.INSTRUMENT_ID, outsym, Int64(0), Int64(0))
+
+    _trade(iid) = DBN.TradeMsg(
+        DBN.RecordHeader(UInt8(0), DBN.RType.MBP_0_MSG, UInt16(1), UInt32(iid), Int64(0)),
+        Int64(1), UInt32(1), DBN.Action.TRADE, DBN.Side.BID, 0x00, 0x00, Int64(0), Int32(0), UInt32(1))
+
+    newclient() = Live("db-1234567890abcdef12345"; dataset = "GLBX.MDP3")
+
+    @testset "empty map" begin
+        c = newclient()
+        @test symbol_for(c, 42) === nothing
+        @test isempty(symbol_map(c))
+    end
+
+    @testset "record + resolve (in/out)" begin
+        c = newclient()
+        DatabentoAPI._record_symbol!(c, _smm(42, "ES.FUT", "ESH4"))
+        DatabentoAPI._record_symbol!(c, _smm(99, "AAPL", "12345"))
+        # non-mapping records are a no-op
+        DatabentoAPI._record_symbol!(c, _trade(7))
+
+        @test symbol_for(c, 42) == "ES.FUT"            # :in default
+        @test symbol_for(c, 42; which = :out) == "ESH4"
+        @test symbol_for(c, 99) == "AAPL"
+        @test symbol_for(c, 7) === nothing             # data record didn't register
+        @test symbol_for(c, 123) === nothing           # unknown id
+
+        # record overload resolves via hd.instrument_id
+        @test symbol_for(c, _trade(42)) == "ES.FUT"
+
+        @test symbol_map(c) == Dict(UInt32(42) => "ES.FUT", UInt32(99) => "AAPL")
+        @test symbol_map(c; which = :out) == Dict(UInt32(42) => "ESH4", UInt32(99) => "12345")
+
+        @test_throws ArgumentError symbol_for(c, 42; which = :bogus)
+    end
+
+    @testset "remap overwrites (e.g. continuous roll)" begin
+        c = newclient()
+        DatabentoAPI._record_symbol!(c, _smm(42, "ES.c.0", "ESH4"))
+        DatabentoAPI._record_symbol!(c, _smm(42, "ES.c.0", "ESM4"))   # rolled
+        @test symbol_for(c, 42; which = :out) == "ESM4"
+        @test length(symbol_map(c)) == 1
+    end
+
+    @testset "snapshot is a copy" begin
+        c = newclient()
+        DatabentoAPI._record_symbol!(c, _smm(1, "A", "1"))
+        snap = symbol_map(c)
+        DatabentoAPI._record_symbol!(c, _smm(2, "B", "2"))
+        @test snap == Dict(UInt32(1) => "A")           # unaffected by later mapping
+        @test length(symbol_map(c)) == 2
+    end
+
+    @testset "callbacks" begin
+        c = newclient()
+        seen = Tuple{UInt32,String}[]
+        add_symbol_mapping_callback(c, (iid, msg) -> push!(seen, (iid, msg.stype_in_symbol)))
+        # map is already updated when the callback fires
+        add_symbol_mapping_callback(c, (iid, _) -> push!(seen, (iid, something(symbol_for(c, iid), "?"))))
+
+        DatabentoAPI._record_symbol!(c, _smm(42, "ES.FUT", "ESH4"))
+        @test (UInt32(42), "ES.FUT") in seen
+        @test count(==((UInt32(42), "ES.FUT")), seen) == 2   # both callbacks saw it
+
+        @test_throws ArgumentError add_symbol_mapping_callback(c, 123)
+    end
+
+    @testset "a raising callback is swallowed" begin
+        c = newclient()
+        add_symbol_mapping_callback(c, (_, _) -> error("boom"))
+        ok = Ref(false)
+        add_symbol_mapping_callback(c, (_, _) -> ok[] = true)
+        # should not throw, and the second callback still runs
+        @test (DatabentoAPI._record_symbol!(c, _smm(1, "A", "1")); true)
+        @test ok[]
+        @test symbol_for(c, 1) == "A"
+    end
+end