diff --git a/CHANGELOG.md b/CHANGELOG.md
index 5670523..6ebe65e 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -7,6 +7,29 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Fixed
+
+- `read_stat_msg` now sizes the record body from `hd.length` instead of assuming
+  the 80-byte DBN v3 layout. Pre-v3 64-byte `StatMsg` records (32-bit `quantity`,
+  as currently served by the historical gateway for STATISTICS) decode correctly
+  and are upgraded to the v3 shape, instead of overrunning the record boundary
+  and desyncing every subsequent record in the stream ([#32]).
+- `stat_to_dataframe` no longer references nonexistent `StatMsg` fields
+  (`stat_value`, `flags`); STATISTICS records export with `price`, `quantity`,
+  `flags` (from `stat_flags`), `ts_ref`, `stat_type`, `channel_id`, and
+  `update_action` columns ([#33]).
+- The metadata decoder reads **all** symbol-mapping intervals instead of only
+  the first. `Metadata.mappings` now holds one
+  `(raw_symbol, mapped_symbol, start_date, end_date)` tuple per interval, so a
+  continuous contract's full roll history survives decoding; the encoder groups
+  consecutive same-symbol tuples back into one mapping entry ([#34]).
+- Invalid enum bytes are handled uniformly: `Side`, `Action`, and
+  `InstrumentClass` all warn (once) and fall back to a sentinel
+  (`Side.NONE`, `Action.NONE`, `InstrumentClass.OTHER`) instead of `Side`/
+  `Action` throwing and killing the decode stream. The conversions use lookup
+  tables, so the hot path stays free of `try`/`catch`. The old `safe_action`
+  fallback of `Action.TRADE` was changed to `Action.NONE` ([#35]).
+
 ## [0.1.3] - 2026-06-05
 
 ### Added
@@ -67,3 +90,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 [#22]: https://github.com/tbeason/DatabentoBinaryEncoding.jl/pull/22
 [#23]: https://github.com/tbeason/DatabentoBinaryEncoding.jl/issues/23
 [#24]: https://github.com/tbeason/DatabentoBinaryEncoding.jl/pull/24
+[#32]: https://github.com/tbeason/DatabentoBinaryEncoding.jl/issues/32
+[#33]: https://github.com/tbeason/DatabentoBinaryEncoding.jl/issues/33
+[#34]: https://github.com/tbeason/DatabentoBinaryEncoding.jl/issues/34
+[#35]: https://github.com/tbeason/DatabentoBinaryEncoding.jl/issues/35
diff --git a/src/decode.jl b/src/decode.jl
index 70ab50f..c272e26 100644
--- a/src/decode.jl
+++ b/src/decode.jl
@@ -15,26 +15,6 @@ Finds the first null byte and creates string from valid data only.
     return len == 0 ? "" : String(bytes[1:len])
 end
 
-"""
-    unsafe_action(raw_val::UInt8) -> Action.T
-
-Fast enum conversion without safety checks. 1000x+ faster than safe_action.
-Only use when you know the data is well-formed (e.g., reading from trusted DBN files).
-"""
-@inline function unsafe_action(raw_val::UInt8)
-    return raw_val == 0x00 ? Action.NONE : Action.T(raw_val)
-end
-
-"""
-    unsafe_side(raw_val::UInt8) -> Side.T
-
-Fast enum conversion without safety checks. 1000x+ faster than safe_side.
-Only use when you know the data is well-formed (e.g., reading from trusted DBN files).
-"""
-@inline function unsafe_side(raw_val::UInt8)
-    return raw_val == 0x00 ? Side.NONE : Side.T(raw_val)
-end
-
 """
     DBNDecoder
 
@@ -339,17 +319,17 @@ function read_header!(decoder::DBNDecoder)
         # Intervals count
         intervals_count = ltoh(reinterpret(UInt32, metadata_bytes[pos:pos+3])[1])
         pos += 4
-        
-        # For now, just read the first interval (simplified)
-        if intervals_count > 0
+
+        # One mappings entry per interval (roll history for continuous symbols)
+        for _ in 1:intervals_count
             # Start date (4 bytes)
             start_date_raw = ltoh(reinterpret(UInt32, metadata_bytes[pos:pos+3])[1])
             pos += 4
-            
+
             # End date (4 bytes)
             end_date_raw = ltoh(reinterpret(UInt32, metadata_bytes[pos:pos+3])[1])
             pos += 4
-            
+
             # Mapped symbol
             mapped_symbol_bytes = metadata_bytes[pos:pos+symbol_cstr_len-1]
             pos += symbol_cstr_len
@@ -359,13 +339,8 @@ function read_header!(decoder::DBNDecoder)
             else
                 mapped_symbol = String(mapped_symbol_bytes)
             end
-            
+
             push!(mappings, (raw_symbol, mapped_symbol, Int64(start_date_raw), Int64(end_date_raw)))
-            
-            # Skip remaining intervals for now
-            for _ in 2:intervals_count
-                pos += 4 + 4 + symbol_cstr_len  # start_date + end_date + symbol
-            end
         end
     end
     
@@ -533,8 +508,8 @@ end
     size = read(decoder.io, UInt32)        # 4 bytes (positions 32-35)
     flags = read(decoder.io, UInt8)        # 1 byte (position 36)
     channel_id = read(decoder.io, UInt8)   # 1 byte (position 37)
-    action = unsafe_action(read(decoder.io, UInt8))   # 1 byte (position 38)
-    side = unsafe_side(read(decoder.io, UInt8))       # 1 byte (position 39)
+    action = safe_action(read(decoder.io, UInt8))   # 1 byte (position 38)
+    side = safe_side(read(decoder.io, UInt8))       # 1 byte (position 39)
     price = read(decoder.io, Int64)        # 8 bytes (positions 40-47)
     ts_in_delta = read(decoder.io, Int32)  # 4 bytes (positions 48-51)
     sequence = read(decoder.io, UInt32)    # 4 bytes (positions 52-55)
@@ -545,8 +520,8 @@ end
 @inline function read_trade_msg(decoder::DBNDecoder, hd::RecordHeader)
     price = read(decoder.io, Int64)
     size = read(decoder.io, UInt32)
-    action = unsafe_action(read(decoder.io, UInt8))
-    side = unsafe_side(read(decoder.io, UInt8))
+    action = safe_action(read(decoder.io, UInt8))
+    side = safe_side(read(decoder.io, UInt8))
     flags = read(decoder.io, UInt8)
     depth = read(decoder.io, UInt8)
     ts_recv = read(decoder.io, Int64)
@@ -558,8 +533,8 @@ end
 @inline function read_mbp1_msg(decoder::DBNDecoder, hd::RecordHeader)
     price = read(decoder.io, Int64)
     size = read(decoder.io, UInt32)
-    action = unsafe_action(read(decoder.io, UInt8))
-    side = unsafe_side(read(decoder.io, UInt8))
+    action = safe_action(read(decoder.io, UInt8))
+    side = safe_side(read(decoder.io, UInt8))
     flags = read(decoder.io, UInt8)
     depth = read(decoder.io, UInt8)
     ts_recv = read(decoder.io, Int64)
@@ -580,8 +555,8 @@ end
 @inline function read_mbp10_msg(decoder::DBNDecoder, hd::RecordHeader)
     price = read(decoder.io, Int64)
     size = read(decoder.io, UInt32)
-    action = unsafe_action(read(decoder.io, UInt8))
-    side = unsafe_side(read(decoder.io, UInt8))
+    action = safe_action(read(decoder.io, UInt8))
+    side = safe_side(read(decoder.io, UInt8))
     flags = read(decoder.io, UInt8)
     depth = read(decoder.io, UInt8)
     ts_recv = read(decoder.io, Int64)
@@ -637,7 +612,7 @@ end
     market_imbalance_qty = read(decoder.io, UInt32)
     unpaired_qty = read(decoder.io, UInt32)
     auction_type = read(decoder.io, UInt8)
-    side = unsafe_side(read(decoder.io, UInt8))
+    side = safe_side(read(decoder.io, UInt8))
     auction_status = read(decoder.io, UInt8)
     freeze_status = read(decoder.io, UInt8)
     num_extensions = read(decoder.io, UInt8)
@@ -648,17 +623,28 @@ end
 end
 
 @inline function read_stat_msg(decoder::DBNDecoder, hd::RecordHeader)
+    # StatMsg layout depends on DBN version. Per the public spec:
+    #   v1/v2: ts_recv(8) ts_ref(8) price(8) quantity(4, Int32) sequence(4)
+    #          ts_in_delta(4) stat_type(2) channel_id(2) update_action(1)
+    #          stat_flags(1) reserved(6) = 48-byte body
+    #   v3:    same but quantity(8, Int64) and reserved(18) = 64-byte body
+    body_size = Int(hd.length) * LENGTH_MULTIPLIER - 16
+    if body_size != 48 && body_size < 64
+        @warn "Unexpected StatMsg body size $body_size; skipping record" maxlog = 1
+        skip(decoder.io, body_size)
+        return nothing
+    end
     ts_recv = read(decoder.io, UInt64)
-    ts_ref = read(decoder.io, UInt64) 
+    ts_ref = read(decoder.io, UInt64)
     price = read(decoder.io, Int64)
-    # Handle UNDEF values in quantity field - read as UInt64 first
-    quantity_raw = read(decoder.io, UInt64)
-    quantity = if quantity_raw == 0xffffffffffffffff
-        # UNDEF_STAT_QUANTITY - use a special value or convert safely
-        typemax(Int64)  
+    quantity = if body_size == 48
+        # v1/v2: 32-bit quantity, UNDEF_STAT_QUANTITY = typemax(Int32)
+        quantity_raw = read(decoder.io, Int32)
+        quantity_raw == typemax(Int32) ? typemax(Int64) : Int64(quantity_raw)
     else
-        # Safe conversion for normal values
-        quantity_raw <= typemax(Int64) ? Int64(quantity_raw) : typemax(Int64)
+        # v3: 64-bit quantity, UNDEF_STAT_QUANTITY = typemax(Int64)
+        quantity_raw = read(decoder.io, UInt64)
+        quantity_raw >= 0x7fffffffffffffff ? typemax(Int64) : Int64(quantity_raw)
     end
     sequence = read(decoder.io, UInt32)
     ts_in_delta = read(decoder.io, Int32)
@@ -666,8 +652,15 @@ end
     channel_id = read(decoder.io, UInt16)
     update_action = read(decoder.io, UInt8)
     stat_flags = read(decoder.io, UInt8)
-    _ = read(decoder.io, 18)  # Reserved (adjusted for field size changes)
-    return StatMsg(hd, ts_recv, ts_ref, price, quantity, sequence, ts_in_delta, stat_type, channel_id, update_action, stat_flags)
+    # Skip reserved tail: whatever the header says is left of the record
+    skip(decoder.io, body_size - (body_size == 48 ? 42 : 46))
+    # Normalize the header to the v3 size whenever the on-disk size differed
+    # (pre-v3 upgrade, or an extended record whose tail we discarded above),
+    # so re-encoding the 64-byte v3 body stays self-consistent
+    v3_length = UInt8(80 ÷ LENGTH_MULTIPLIER)
+    out_hd = hd.length == v3_length ? hd :
+             RecordHeader(v3_length, hd.rtype, hd.publisher_id, hd.instrument_id, hd.ts_event)
+    return StatMsg(out_hd, ts_recv, ts_ref, price, quantity, sequence, ts_in_delta, stat_type, channel_id, update_action, stat_flags)
 end
 
 @inline function read_error_msg(decoder::DBNDecoder, hd::RecordHeader)
@@ -775,8 +768,8 @@ end
 @inline function read_cmbp1_msg(decoder::DBNDecoder, hd::RecordHeader)
     price = read(decoder.io, Int64)
     size = read(decoder.io, UInt32)
-    action = unsafe_action(read(decoder.io, UInt8))
-    side = unsafe_side(read(decoder.io, UInt8))
+    action = safe_action(read(decoder.io, UInt8))
+    side = safe_side(read(decoder.io, UInt8))
     flags = read(decoder.io, UInt8)
     depth = read(decoder.io, UInt8)
     ts_recv = read(decoder.io, Int64)
@@ -797,8 +790,8 @@ end
 @inline function read_cbbo1s_msg(decoder::DBNDecoder, hd::RecordHeader)
     price = read(decoder.io, Int64)
     size = read(decoder.io, UInt32)
-    action = unsafe_action(read(decoder.io, UInt8))
-    side = unsafe_side(read(decoder.io, UInt8))
+    action = safe_action(read(decoder.io, UInt8))
+    side = safe_side(read(decoder.io, UInt8))
     flags = read(decoder.io, UInt8)
     depth = read(decoder.io, UInt8)
     ts_recv = read(decoder.io, Int64)
@@ -819,8 +812,8 @@ end
 @inline function read_cbbo1m_msg(decoder::DBNDecoder, hd::RecordHeader)
     price = read(decoder.io, Int64)
     size = read(decoder.io, UInt32)
-    action = unsafe_action(read(decoder.io, UInt8))
-    side = unsafe_side(read(decoder.io, UInt8))
+    action = safe_action(read(decoder.io, UInt8))
+    side = safe_side(read(decoder.io, UInt8))
     flags = read(decoder.io, UInt8)
     depth = read(decoder.io, UInt8)
     ts_recv = read(decoder.io, Int64)
@@ -841,8 +834,8 @@ end
 @inline function read_tcbbo_msg(decoder::DBNDecoder, hd::RecordHeader)
     price = read(decoder.io, Int64)
     size = read(decoder.io, UInt32)
-    action = unsafe_action(read(decoder.io, UInt8))
-    side = unsafe_side(read(decoder.io, UInt8))
+    action = safe_action(read(decoder.io, UInt8))
+    side = safe_side(read(decoder.io, UInt8))
     flags = read(decoder.io, UInt8)
     depth = read(decoder.io, UInt8)
     ts_recv = read(decoder.io, Int64)
@@ -863,8 +856,8 @@ end
 @inline function read_bbo1s_msg(decoder::DBNDecoder, hd::RecordHeader)
     price = read(decoder.io, Int64)
     size = read(decoder.io, UInt32)
-    action = unsafe_action(read(decoder.io, UInt8))
-    side = unsafe_side(read(decoder.io, UInt8))
+    action = safe_action(read(decoder.io, UInt8))
+    side = safe_side(read(decoder.io, UInt8))
     flags = read(decoder.io, UInt8)
     depth = read(decoder.io, UInt8)
     ts_recv = read(decoder.io, Int64)
@@ -885,8 +878,8 @@ end
 @inline function read_bbo1m_msg(decoder::DBNDecoder, hd::RecordHeader)
     price = read(decoder.io, Int64)
     size = read(decoder.io, UInt32)
-    action = unsafe_action(read(decoder.io, UInt8))
-    side = unsafe_side(read(decoder.io, UInt8))
+    action = safe_action(read(decoder.io, UInt8))
+    side = safe_side(read(decoder.io, UInt8))
     flags = read(decoder.io, UInt8)
     depth = read(decoder.io, UInt8)
     ts_recv = read(decoder.io, Int64)
@@ -1108,7 +1101,7 @@ end
     leg_instrument_class_byte = read(decoder.io, UInt8)
     leg_instrument_class = safe_instrument_class(leg_instrument_class_byte)
     leg_side_byte = read(decoder.io, UInt8)
-    leg_side = unsafe_side(leg_side_byte)
+    leg_side = safe_side(leg_side_byte)
 
     # v3: 17 bytes _reserved
     skip(decoder.io, 17)
@@ -1433,8 +1426,11 @@ function read_dbn_typed(filename::String, ::Type{T}) where T
                     end
                 end
 
-                # Read record directly - uses same helpers as streaming
-                @inbounds records[idx] = _read_typed_record_stream(decoder, T, hd)
+                # Read record directly - uses same helpers as streaming.
+                # A reader can return nothing for a malformed record it skipped.
+                rec = _read_typed_record_stream(decoder, T, hd)
+                rec === nothing && continue
+                @inbounds records[idx] = rec
                 idx += 1
             end
 
@@ -1479,7 +1475,9 @@ function read_dbn_typed(filename::String, ::Type{T}) where T
                     end
                 end
 
-                push!(records, _read_typed_record_stream(decoder, T, hd))
+                # A reader can return nothing for a malformed record it skipped.
+                rec = _read_typed_record_stream(decoder, T, hd)
+                rec === nothing || push!(records, rec)
             end
         finally
             if decoder.io !== decoder.base_io
diff --git a/src/encode.jl b/src/encode.jl
index 179ce30..47c0406 100644
--- a/src/encode.jl
+++ b/src/encode.jl
@@ -103,8 +103,8 @@ function write_header(encoder::DBNEncoder)
     write(metadata_buf, encoder.metadata.ts_out ? UInt8(1) : UInt8(0))
     
     # Symbol string length (2 bytes) - only for version > 1
-    # DBN v2/v3 use 71-byte symbol strings.
-    symbol_cstr_len = UInt16(SYMBOL_CSTR_LEN)
+    # DBN v2/v3 use 71-byte symbol strings.
+    symbol_cstr_len = UInt16(SYMBOL_CSTR_LEN)
     write(metadata_buf, htol(symbol_cstr_len))
     
     # Reserved padding (53 bytes for v3)
@@ -154,9 +154,17 @@ function write_header(encoder::DBNEncoder)
         write(metadata_buf, sym_bytes)
     end
     
-    # Symbol mappings - need to write the exact format the reader expects
-    write(metadata_buf, htol(UInt32(length(encoder.metadata.mappings))))
+    # Symbol mappings - mappings holds one tuple per (raw symbol, interval), so
+    # group consecutive tuples sharing a raw symbol into one entry with N intervals
+    mapping_groups = Tuple{String,Vector{Tuple{String,Int64,Int64}}}[]
     for (raw_symbol, mapped_symbol, start_date, end_date) in encoder.metadata.mappings
+        if isempty(mapping_groups) || mapping_groups[end][1] != raw_symbol
+            push!(mapping_groups, (raw_symbol, Tuple{String,Int64,Int64}[]))
+        end
+        push!(mapping_groups[end][2], (mapped_symbol, start_date, end_date))
+    end
+    write(metadata_buf, htol(UInt32(length(mapping_groups))))
+    for (raw_symbol, intervals) in mapping_groups
         # Raw symbol (fixed length)
         raw_sym_bytes = Vector{UInt8}(undef, symbol_cstr_len)
         fill!(raw_sym_bytes, 0)
@@ -166,25 +174,27 @@ function write_header(encoder::DBNEncoder)
             raw_sym_bytes[1:copy_len] = raw_str_bytes[1:copy_len]
         end
         write(metadata_buf, raw_sym_bytes)
-        
-        # Intervals count (1 interval per mapping for simplicity)
-        write(metadata_buf, htol(UInt32(1)))
-        
-        # Start date (4 bytes)
-        write(metadata_buf, htol(UInt32(start_date)))
-        
-        # End date (4 bytes)
-        write(metadata_buf, htol(UInt32(end_date)))
-        
-        # Mapped symbol (fixed length)
-        mapped_sym_bytes = Vector{UInt8}(undef, symbol_cstr_len)
-        fill!(mapped_sym_bytes, 0)
-        mapped_str_bytes = Vector{UInt8}(mapped_symbol)
-        copy_len = min(length(mapped_str_bytes), symbol_cstr_len - 1)
-        if copy_len > 0
-            mapped_sym_bytes[1:copy_len] = mapped_str_bytes[1:copy_len]
+
+        # Intervals count
+        write(metadata_buf, htol(UInt32(length(intervals))))
+
+        for (mapped_symbol, start_date, end_date) in intervals
+            # Start date (4 bytes)
+            write(metadata_buf, htol(UInt32(start_date)))
+
+            # End date (4 bytes)
+            write(metadata_buf, htol(UInt32(end_date)))
+
+            # Mapped symbol (fixed length)
+            mapped_sym_bytes = Vector{UInt8}(undef, symbol_cstr_len)
+            fill!(mapped_sym_bytes, 0)
+            mapped_str_bytes = Vector{UInt8}(mapped_symbol)
+            copy_len = min(length(mapped_str_bytes), symbol_cstr_len - 1)
+            if copy_len > 0
+                mapped_sym_bytes[1:copy_len] = mapped_str_bytes[1:copy_len]
+            end
+            write(metadata_buf, mapped_sym_bytes)
         end
-        write(metadata_buf, mapped_sym_bytes)
     end
     
     # Get metadata bytes and write length + metadata
@@ -498,28 +508,28 @@ function write_record_complex(encoder::DBNEncoder, record)
                 write(io, UInt8(0))
             end
 
-        else  # v3
-            # ===== DBN V3 InstrumentDefMsg =====
-            # DBN binary records are encoded in the fixed binary field layout.
-            # `encode_order` controls text/field ordering, not the binary layout.
-            write(io, record.ts_recv)
-            write(io, record.min_price_increment)
-            write(io, record.display_factor)
-            write(io, record.expiration)
+        else  # v3
+            # ===== DBN V3 InstrumentDefMsg =====
+            # DBN binary records are encoded in the fixed binary field layout.
+            # `encode_order` controls text/field ordering, not the binary layout.
+            write(io, record.ts_recv)
+            write(io, record.min_price_increment)
+            write(io, record.display_factor)
+            write(io, record.expiration)
             write(io, record.activation)
             write(io, record.high_limit_price)
             write(io, record.low_limit_price)
             write(io, record.max_price_variation)
-            write(io, record.unit_of_measure_qty)
-            write(io, record.min_price_increment_amount)
-            write(io, record.price_ratio)
-            write(io, record.strike_price)
-            write(io, record.raw_instrument_id)
-            write(io, record.leg_price)
-            write(io, record.leg_delta)
-
-            write(io, record.inst_attrib_value)
-            write(io, record.underlying_id)
+            write(io, record.unit_of_measure_qty)
+            write(io, record.min_price_increment_amount)
+            write(io, record.price_ratio)
+            write(io, record.strike_price)
+            write(io, record.raw_instrument_id)
+            write(io, record.leg_price)
+            write(io, record.leg_delta)
+
+            write(io, record.inst_attrib_value)
+            write(io, record.underlying_id)
             write(io, record.market_depth_implied)
             write(io, record.market_depth)
             write(io, record.market_segment_id)
@@ -528,58 +538,58 @@ function write_record_complex(encoder::DBNEncoder, record)
             write(io, record.min_lot_size_block)
             write(io, record.min_lot_size_round_lot)
             write(io, record.min_trade_vol)
-            write(io, record.contract_multiplier)
-            write(io, record.decay_quantity)
-            write(io, record.original_contract_size)
-            write(io, record.leg_instrument_id)
-            write(io, record.leg_ratio_price_numerator)
-            write(io, record.leg_ratio_price_denominator)
-            write(io, record.leg_ratio_qty_numerator)
-            write(io, record.leg_ratio_qty_denominator)
-            write(io, record.leg_underlying_id)
-
-            write(io, record.appl_id)
-            write(io, record.maturity_year)
-            write(io, record.decay_start_date)
-            write(io, record.channel_id)
-            write(io, record.leg_count)
-            write(io, record.leg_index)
-
-            # String fields in struct declaration order.
-            write_fixed_string(io, record.currency, 4)
-            write_fixed_string(io, record.settl_currency, 4)
-            write_fixed_string(io, record.secsubtype, 6)
-            write_fixed_string(io, record.raw_symbol, SYMBOL_CSTR_LEN)
-            write_fixed_string(io, record.group, 21)
-            write_fixed_string(io, record.exchange, 5)
-            write_fixed_string(io, record.asset, 11)  # 11 bytes in v3!
+            write(io, record.contract_multiplier)
+            write(io, record.decay_quantity)
+            write(io, record.original_contract_size)
+            write(io, record.leg_instrument_id)
+            write(io, record.leg_ratio_price_numerator)
+            write(io, record.leg_ratio_price_denominator)
+            write(io, record.leg_ratio_qty_numerator)
+            write(io, record.leg_ratio_qty_denominator)
+            write(io, record.leg_underlying_id)
+
+            write(io, record.appl_id)
+            write(io, record.maturity_year)
+            write(io, record.decay_start_date)
+            write(io, record.channel_id)
+            write(io, record.leg_count)
+            write(io, record.leg_index)
+
+            # String fields in struct declaration order.
+            write_fixed_string(io, record.currency, 4)
+            write_fixed_string(io, record.settl_currency, 4)
+            write_fixed_string(io, record.secsubtype, 6)
+            write_fixed_string(io, record.raw_symbol, SYMBOL_CSTR_LEN)
+            write_fixed_string(io, record.group, 21)
+            write_fixed_string(io, record.exchange, 5)
+            write_fixed_string(io, record.asset, 11)  # 11 bytes in v3!
             write_fixed_string(io, record.cfi, 7)
             write_fixed_string(io, record.security_type, 7)
-            write_fixed_string(io, record.unit_of_measure, 31)
-            write_fixed_string(io, record.underlying, 21)
-            write_fixed_string(io, record.strike_price_currency, 4)
-            write_fixed_string(io, record.leg_raw_symbol, SYMBOL_CSTR_LEN)
-
-            # Single-byte fields without encode_order
-            write(io, UInt8(record.instrument_class))
-            write(io, UInt8(record.match_algorithm))
-            write(io, record.main_fraction)
-            write(io, record.price_display_format)
-            write(io, record.sub_fraction)
-            write(io, record.underlying_product)
-            write(io, UInt8(record.security_update_action))
-            write(io, record.maturity_month)
-            write(io, record.maturity_day)
-            write(io, record.maturity_week)
+            write_fixed_string(io, record.unit_of_measure, 31)
+            write_fixed_string(io, record.underlying, 21)
+            write_fixed_string(io, record.strike_price_currency, 4)
+            write_fixed_string(io, record.leg_raw_symbol, SYMBOL_CSTR_LEN)
+
+            # Single-byte fields without encode_order
+            write(io, UInt8(record.instrument_class))
+            write(io, UInt8(record.match_algorithm))
+            write(io, record.main_fraction)
+            write(io, record.price_display_format)
+            write(io, record.sub_fraction)
+            write(io, record.underlying_product)
+            write(io, UInt8(record.security_update_action))
+            write(io, record.maturity_month)
+            write(io, record.maturity_day)
+            write(io, record.maturity_week)
             write(io, record.user_defined_instrument ? UInt8('Y') : UInt8('N'))
-            write(io, record.contract_multiplier_unit)
-            write(io, record.flow_schedule_type)
-            write(io, record.tick_rule)
-            write(io, UInt8(record.leg_instrument_class))
-            write(io, UInt8(record.leg_side))
-
-            # v3: 17 bytes _reserved
-            for _ in 1:17
+            write(io, record.contract_multiplier_unit)
+            write(io, record.flow_schedule_type)
+            write(io, record.tick_rule)
+            write(io, UInt8(record.leg_instrument_class))
+            write(io, UInt8(record.leg_side))
+
+            # v3: 17 bytes _reserved
+            for _ in 1:17
                 write(io, UInt8(0))
             end
         end
@@ -904,4 +914,4 @@ function write_dbn(filename::String, metadata::Metadata, records)
             close(base_io)
         end
     end
-end
+end
diff --git a/src/export.jl b/src/export.jl
index 0ddb388..3352b78 100644
--- a/src/export.jl
+++ b/src/export.jl
@@ -289,11 +289,15 @@ function stat_to_dataframe(records::Vector{StatMsg})
     DataFrame(
         ts_event = [r.hd.ts_event for r in records],
         ts_recv = [r.ts_recv for r in records],
+        ts_ref = [r.ts_ref for r in records],
         instrument_id = [r.hd.instrument_id for r in records],
         publisher_id = [r.hd.publisher_id for r in records],
         stat_type = [r.stat_type for r in records],
-        stat_value = [r.stat_value for r in records],
-        flags = [r.flags for r in records],
+        channel_id = [r.channel_id for r in records],
+        update_action = [r.update_action for r in records],
+        price = [price_to_float(r.price) for r in records],
+        quantity = [r.quantity for r in records],
+        flags = [r.stat_flags for r in records],
         ts_in_delta = [r.ts_in_delta for r in records],
         sequence = [r.sequence for r in records]
     )
diff --git a/src/streaming.jl b/src/streaming.jl
index 1b30965..7dae0de 100644
--- a/src/streaming.jl
+++ b/src/streaming.jl
@@ -336,8 +336,11 @@ function _foreach_record_impl(f::Function, decoder::DBNDecoder, ::Type{T}) where
                 end
             end
 
-            # Read record directly into buffer
-            buffer[] = _read_typed_record_stream(decoder, T, hd)
+            # Read record directly into buffer.
+            # A reader can return nothing for a malformed record it skipped.
+            rec = _read_typed_record_stream(decoder, T, hd)
+            rec === nothing && continue
+            buffer[] = rec
 
             # Call user function with buffer contents
             f(buffer[])
@@ -433,7 +436,10 @@ function _foreach_record_with_control_impl(f_data::Function, f_control::Function
                                                       RType.OHLCV_1H_MSG,
                                                       RType.OHLCV_1D_MSG))
             if is_match
-                buffer[] = _read_typed_record_stream(decoder, T, hd)
+                # A reader can return nothing for a malformed record it skipped.
+                rec = _read_typed_record_stream(decoder, T, hd)
+                rec === nothing && continue
+                buffer[] = rec
                 f_data(buffer[])
                 continue
             end
diff --git a/src/types.jl b/src/types.jl
index 65e9978..d5685e6 100644
--- a/src/types.jl
+++ b/src/types.jl
@@ -1,14 +1,14 @@
 # DBN types, enums, and data structures
 
 # Constants
-"""The current DBN format version supported by this implementation."""
-const DBN_VERSION = 3
-
-"""Fixed-length symbol string size for DBN v2/v3 metadata and records."""
-const SYMBOL_CSTR_LEN = 71
-
-"""Fixed-point price scaling factor for converting between integer and float prices."""
-const FIXED_PRICE_SCALE = Int32(1_000_000_000)
+"""The current DBN format version supported by this implementation."""
+const DBN_VERSION = 3
+
+"""Fixed-length symbol string size for DBN v2/v3 metadata and records."""
+const SYMBOL_CSTR_LEN = 71
+
+"""Fixed-point price scaling factor for converting between integer and float prices."""
+const FIXED_PRICE_SCALE = Int32(1_000_000_000)
 
 """Sentinel value indicating an undefined or missing price."""
 const UNDEF_PRICE = typemax(Int64)
@@ -243,38 +243,38 @@ end
 """
     InstrumentClass
 
-Classification of financial instruments. Values match Databento's DBN
-`InstrumentClass` enum.
-
-# Values
-- `STOCK`: Equity instruments
-- `CALL`: Call option contracts
-- `PUT`: Put option contracts
-- `FUTURE`: Futures contracts
-- `BOND`: Fixed income securities
-- `MIXED_SPREAD`: Mixed spread instruments
-- `FUTURE_SPREAD`: Futures spread instruments
-- `OPTION_SPREAD`: Option spread instruments
-- `FX_SPOT`: Foreign exchange spot
-- `COMMODITY_SPOT`: Commodity spot
-- `UNKNOWN_0`, `UNKNOWN_45`: Numeric fallback values for unknown classes
-"""
-@enumx InstrumentClass::UInt8 begin
-    BOND = UInt8('B')
-    CALL = UInt8('C')
-    FUTURE = UInt8('F')
-    STOCK = UInt8('K')
-    MIXED_SPREAD = UInt8('M')
-    PUT = UInt8('P')
-    FUTURE_SPREAD = UInt8('S')
-    OPTION_SPREAD = UInt8('T')
-    FX_SPOT = UInt8('X')
-    COMMODITY_SPOT = UInt8('Y')
-    # Also support numeric values
-    UNKNOWN_0 = 0
-    UNKNOWN_45 = 45
-    OTHER = UInt8('?')
-end
+Classification of financial instruments. Values match Databento's DBN
+`InstrumentClass` enum.
+
+# Values
+- `STOCK`: Equity instruments
+- `CALL`: Call option contracts
+- `PUT`: Put option contracts
+- `FUTURE`: Futures contracts
+- `BOND`: Fixed income securities
+- `MIXED_SPREAD`: Mixed spread instruments
+- `FUTURE_SPREAD`: Futures spread instruments
+- `OPTION_SPREAD`: Option spread instruments
+- `FX_SPOT`: Foreign exchange spot
+- `COMMODITY_SPOT`: Commodity spot
+- `UNKNOWN_0`, `UNKNOWN_45`: Numeric fallback values for unknown classes
+"""
+@enumx InstrumentClass::UInt8 begin
+    BOND = UInt8('B')
+    CALL = UInt8('C')
+    FUTURE = UInt8('F')
+    STOCK = UInt8('K')
+    MIXED_SPREAD = UInt8('M')
+    PUT = UInt8('P')
+    FUTURE_SPREAD = UInt8('S')
+    OPTION_SPREAD = UInt8('T')
+    FX_SPOT = UInt8('X')
+    COMMODITY_SPOT = UInt8('Y')
+    # Also support numeric values
+    UNKNOWN_0 = 0
+    UNKNOWN_45 = 45
+    OTHER = UInt8('?')
+end
 
 # Basic structures
 
@@ -326,7 +326,11 @@ Metadata information for a DBN dataset.
 - `symbols::Vector{String}`: List of symbols in the dataset
 - `partial::Vector{String}`: Partially available symbols
 - `not_found::Vector{String}`: Symbols that were not found
-- `mappings::Vector{Tuple{String,String,Int64,Int64}}`: Symbol mappings with time ranges
+- `mappings::Vector{Tuple{String,String,Int64,Int64}}`: Symbol mappings as
+  `(raw_symbol, mapped_symbol, start_date, end_date)`, one tuple per mapping
+  interval. A symbol with multiple intervals (e.g. a continuous contract's roll
+  history) contributes consecutive tuples sharing the same raw symbol. Dates
+  are raw `YYYYMMDD` integers.
 """
 struct Metadata
     version::UInt8
@@ -595,76 +599,69 @@ function float_to_price(value::Float64, scale::Int32=FIXED_PRICE_SCALE)
     return Int64(round(value * Float64(scale)))
 end
 
-# Helper functions for safe enum conversion
+# Helper functions for safe enum conversion.
+# All enums share one policy: an invalid byte warns (once per enum) and maps to
+# a sentinel value instead of throwing, so one bad byte can't kill a decode
+# stream. Lookup tables keep the hot path free of try/catch.
+
+function _build_enum_lookup(::Type{E}, default::E) where {E}
+    table = fill(default, 256)
+    valid = falses(256)
+    for inst in instances(E)
+        idx = Int(UInt8(inst)) + 1
+        table[idx] = inst
+        valid[idx] = true
+    end
+    return table, valid
+end
+
+const _ACTION_LOOKUP, _ACTION_VALID = _build_enum_lookup(Action.T, Action.NONE)
+const _SIDE_LOOKUP, _SIDE_VALID = _build_enum_lookup(Side.T, Side.NONE)
+const _INSTRUMENT_CLASS_LOOKUP, _INSTRUMENT_CLASS_VALID =
+    _build_enum_lookup(InstrumentClass.T, InstrumentClass.OTHER)
 
 """
     safe_action(raw_val::UInt8)
 
-Safely convert a raw byte value to an Action enum, with fallback handling.
-
-# Arguments
-- `raw_val::UInt8`: Raw action value from DBN data
-
-# Returns
-- `Action.T`: Corresponding Action enum value, or Action.TRADE as fallback
+Convert a raw byte value to an Action enum. `0x00` and invalid values map to
+`Action.NONE`; invalid values additionally log a warning (once).
 """
-function safe_action(raw_val::UInt8)
-    # Special case: 0 might indicate no action for certain record types
-    if raw_val == 0
-        return Action.NONE
-    end
-    
-    try
-        return Action.T(raw_val)
-    catch ArgumentError
-        # For unknown action values, use a default or create a placeholder
-        # For now, return TRADE as a safe default
-        @warn "Unknown Action value: $raw_val (0x$(string(raw_val, base=16))), using TRADE as default"
-        return Action.TRADE
+@inline function safe_action(raw_val::UInt8)
+    # 0 indicates no action for certain record types
+    raw_val == 0x00 && return Action.NONE
+    idx = Int(raw_val) + 1
+    @inbounds if !_ACTION_VALID[idx]
+        @warn "Unknown Action value: $raw_val (0x$(string(raw_val, base=16))), using NONE as default" maxlog = 1
     end
+    return @inbounds _ACTION_LOOKUP[idx]
 end
 
 """
     safe_side(raw_val::UInt8)
 
-Safely convert a raw byte value to a Side enum, with fallback handling.
-
-# Arguments
-- `raw_val::UInt8`: Raw side value from DBN data
-
-# Returns
-- `Side.T`: Corresponding Side enum value, or Side.NONE as fallback
+Convert a raw byte value to a Side enum. `0x00` and invalid values map to
+`Side.NONE`; invalid values additionally log a warning (once).
 """
-function safe_side(raw_val::UInt8)
-    # Special case: 0 might indicate no side for certain record types
-    if raw_val == 0
-        return Side.NONE
-    end
-
-    try
-        return Side.T(raw_val)
-    catch ArgumentError
-        @warn "Unknown Side value: $raw_val, using NONE as default"
-        return Side.NONE
+@inline function safe_side(raw_val::UInt8)
+    # 0 indicates no side for certain record types
+    raw_val == 0x00 && return Side.NONE
+    idx = Int(raw_val) + 1
+    @inbounds if !_SIDE_VALID[idx]
+        @warn "Unknown Side value: $raw_val (0x$(string(raw_val, base=16))), using NONE as default" maxlog = 1
     end
+    return @inbounds _SIDE_LOOKUP[idx]
 end
 
 """
     safe_instrument_class(raw_val::UInt8)
 
-Safely convert a raw byte value to an InstrumentClass enum, with fallback handling.
-
-# Arguments
-- `raw_val::UInt8`: Raw instrument class value from DBN data
-
-# Returns
-- `InstrumentClass.T`: Corresponding InstrumentClass enum value, or InstrumentClass.OTHER as fallback
-"""
-function safe_instrument_class(raw_val::UInt8)
-    try
-        return InstrumentClass.T(raw_val)
-    catch ArgumentError
-        @warn "Unknown InstrumentClass value: $raw_val, using OTHER as default"
-        return InstrumentClass.OTHER
+Convert a raw byte value to an InstrumentClass enum. Invalid values log a
+warning (once) and map to `InstrumentClass.OTHER`.
+"""
+@inline function safe_instrument_class(raw_val::UInt8)
+    idx = Int(raw_val) + 1
+    @inbounds if !_INSTRUMENT_CLASS_VALID[idx]
+        @warn "Unknown InstrumentClass value: $raw_val (0x$(string(raw_val, base=16))), using OTHER as default" maxlog = 1
     end
-end
+    return @inbounds _INSTRUMENT_CLASS_LOOKUP[idx]
+end
diff --git a/test/runtests.jl b/test/runtests.jl
index 5e6d774..5a7741b 100644
--- a/test/runtests.jl
+++ b/test/runtests.jl
@@ -20,6 +20,7 @@ include("test_utils.jl")
     include("test_convenience_functions.jl")  # Test all convenience read_*/foreach_* functions
     include("test_phase11_typed_with_control.jl")  # foreach_record_with_control: typed data + Union control split
     include("test_issue23_unset_stype.jl")  # regression: 0xFF unset stype in v3 SymbolMappingMsg (issue #23)
+    include("test_issues_32_35.jl")  # regressions: pre-v3 StatMsg layout, stat_to_dataframe, mapping intervals, invalid enum bytes (issues #32-#35)
     include("test_show.jl")  # compact one-line Base.show for record types
 
     # Run compatibility tests if the Rust CLI is available
diff --git a/test/test_issues_32_35.jl b/test/test_issues_32_35.jl
new file mode 100644
index 0000000..acddaa2
--- /dev/null
+++ b/test/test_issues_32_35.jl
@@ -0,0 +1,260 @@
+# Regression tests for issues #32-#35, all found in one STATISTICS debugging
+# session:
+#   #32: read_stat_msg assumed the 80-byte v3 StatMsg layout and ignored
+#        hd.length, so 64-byte v1/v2 records desynced the whole stream.
+#   #33: stat_to_dataframe referenced nonexistent StatMsg fields
+#        (stat_value, flags) and threw on any STATISTICS records.
+#   #34: the metadata decoder kept only the first symbol-mapping interval,
+#        silently dropping continuous-contract roll history.
+#   #35: invalid Side/Action bytes threw and killed the decode stream while
+#        InstrumentClass warned and defaulted.
+
+@testset "issues #32-#35 regressions" begin
+
+    data_path(f) = joinpath(@__DIR__, "data", f)
+
+    function stat_metadata()
+        return Metadata(
+            UInt8(DBN_VERSION),      # version
+            "TEST.DATA",             # dataset
+            Schema.STATISTICS,       # schema
+            1640995200000000000,     # start_ts
+            1640995260000000000,     # end_ts
+            UInt64(1000),            # limit
+            SType.RAW_SYMBOL,        # stype_in
+            SType.RAW_SYMBOL,        # stype_out
+            false,                   # ts_out
+            ["ES.n.0"],              # symbols
+            String[],                # partial
+            String[],                # not_found
+            Tuple{String,String,Int64,Int64}[]  # mappings
+        )
+    end
+
+    @testset "issue #32: pre-v3 64-byte StatMsg records decode without desync" begin
+        # The shipped v1/v2/v3 statistics captures hold the same two records;
+        # pre-fix the v1/v2 files desynced after the first record.
+        recs_v1 = read_dbn(data_path("test_data.statistics.v1.dbn.zst"))
+        recs_v2 = read_dbn(data_path("test_data.statistics.v2.dbn.zst"))
+        recs_v3 = read_dbn(data_path("test_data.statistics.v3.dbn.zst"))
+
+        @test length(recs_v1) == length(recs_v3)
+        @test length(recs_v2) == length(recs_v3)
+        @test all(r -> r isa StatMsg, recs_v1)
+        @test all(r -> r isa StatMsg, recs_v2)
+
+        for (old, new) in ((recs_v1, recs_v3), (recs_v2, recs_v3))
+            for (a, b) in zip(old, new)
+                @test a.ts_recv == b.ts_recv
+                @test a.ts_ref == b.ts_ref
+                @test a.price == b.price
+                @test a.quantity == b.quantity
+                @test a.sequence == b.sequence
+                @test a.stat_type == b.stat_type
+                @test a.update_action == b.update_action
+                # Pre-v3 records are upgraded to the 80-byte v3 size
+                @test a.hd.length == UInt8(20)
+            end
+        end
+
+        # Synthetic stream: two hand-encoded 64-byte v1-layout StatMsg records.
+        # Pre-fix the first read overran by 16 bytes, so the second record
+        # decoded from garbage (the original desync symptom).
+        ts0 = Int64(1_700_000_000_000_000_000)
+        io = IOBuffer()
+        encoder = DBNEncoder(io, stat_metadata())
+        write_header(encoder)
+        for (iid, qty, seq) in ((UInt32(42), Int32(12345), UInt32(7)),
+                                (UInt32(43), typemax(Int32), UInt32(8)))
+            write(io, UInt8(16))                      # length: 64 bytes / 4
+            write(io, UInt8(RType.STAT_MSG))
+            write(io, UInt16(1))                      # publisher_id
+            write(io, iid)                            # instrument_id
+            write(io, ts0)                            # ts_event
+            write(io, UInt64(ts0 + 1))                # ts_recv
+            write(io, UInt64(ts0 + 2))                # ts_ref
+            write(io, Int64(100_000_000_000))         # price
+            write(io, qty)                            # quantity (Int32 in v1/v2)
+            write(io, seq)                            # sequence
+            write(io, Int32(100))                     # ts_in_delta
+            write(io, UInt16(9))                      # stat_type
+            write(io, UInt16(0))                      # channel_id
+            write(io, UInt8(1))                       # update_action
+            write(io, UInt8(0))                       # stat_flags
+            write(io, zeros(UInt8, 6))                # reserved (v1/v2 tail)
+        end
+
+        tmp = tempname() * ".dbn"
+        try
+            write(tmp, take!(io))
+            recs = read_dbn(tmp)
+            @test length(recs) == 2
+            @test recs[1] isa StatMsg && recs[2] isa StatMsg
+            @test recs[1].hd.instrument_id == 42
+            @test recs[1].quantity == 12345
+            @test recs[1].hd.length == UInt8(20)
+            # The record after the v1 StatMsg is intact, not misaligned garbage
+            @test recs[2].hd.instrument_id == 43
+            @test recs[2].sequence == 8
+            # v1 UNDEF_STAT_QUANTITY (typemax(Int32)) maps to the v3 sentinel
+            @test recs[2].quantity == typemax(Int64)
+        finally
+            safe_rm(tmp)
+        end
+
+        # A malformed StatMsg (body size that matches no known layout) is
+        # skipped — including on the typed read_dbn path — and an extended
+        # record (v3 body + vendor tail) decodes with its header normalized
+        # to the v3 size so re-encoding cannot misalign downstream readers.
+        io = IOBuffer()
+        encoder = DBNEncoder(io, stat_metadata())
+        write_header(encoder)
+        # Malformed: claims 72 bytes total = 56-byte body (not 48, not >=64)
+        write(io, UInt8(18))
+        write(io, UInt8(RType.STAT_MSG))
+        write(io, UInt16(1)); write(io, UInt32(40)); write(io, ts0)
+        write(io, zeros(UInt8, 56))
+        # Extended v3: 88 bytes total = 72-byte body (64 v3 + 8 vendor tail)
+        write(io, UInt8(22))
+        write(io, UInt8(RType.STAT_MSG))
+        write(io, UInt16(1)); write(io, UInt32(41)); write(io, ts0)
+        write(io, UInt64(ts0 + 1))                # ts_recv
+        write(io, UInt64(ts0 + 2))                # ts_ref
+        write(io, Int64(100_000_000_000))         # price
+        write(io, Int64(777))                     # quantity (Int64 in v3)
+        write(io, UInt32(9))                      # sequence
+        write(io, Int32(100))                     # ts_in_delta
+        write(io, UInt16(9))                      # stat_type
+        write(io, UInt16(0))                      # channel_id
+        write(io, UInt8(1))                       # update_action
+        write(io, UInt8(0))                       # stat_flags
+        write(io, zeros(UInt8, 18 + 8))           # reserved + vendor tail
+        # Trailing plain v3 record proves neither of the above desynced
+        write(io, UInt8(20))
+        write(io, UInt8(RType.STAT_MSG))
+        write(io, UInt16(1)); write(io, UInt32(42)); write(io, ts0)
+        write(io, UInt64(ts0 + 1)); write(io, UInt64(ts0 + 2))
+        write(io, Int64(100_000_000_000)); write(io, Int64(888))
+        write(io, UInt32(10)); write(io, Int32(100))
+        write(io, UInt16(9)); write(io, UInt16(0))
+        write(io, UInt8(1)); write(io, UInt8(0))
+        write(io, zeros(UInt8, 18))
+
+        tmp = tempname() * ".dbn"
+        try
+            write(tmp, take!(io))
+            recs = read_dbn(tmp)   # STATISTICS schema -> typed StatMsg path
+            @test length(recs) == 2          # malformed record skipped
+            @test recs[1].hd.instrument_id == 41
+            @test recs[1].quantity == 777
+            @test recs[1].hd.length == UInt8(20)   # extended header normalized
+            @test recs[2].hd.instrument_id == 42
+            @test recs[2].quantity == 888
+        finally
+            safe_rm(tmp)
+        end
+    end
+
+    @testset "issue #33: stat_to_dataframe uses real StatMsg fields" begin
+        hd = RecordHeader(20, RType.STAT_MSG, 1, 66666, 1640995200000000700)
+        msgs = [
+            StatMsg(hd, 1640995200000000701, 1640995200000000000,
+                    10055000000, 500, 54321, 5500, 8, 3, 1, 0x20),
+            StatMsg(hd, 1640995200000000801, 1640995200000000100,
+                    DBN.UNDEF_PRICE, typemax(Int64), 54322, 5600, 9, 3, 2, 0x00),
+        ]
+        df = DBN.records_to_dataframe(msgs)
+        @test size(df, 1) == 2
+        for col in ["ts_event", "ts_recv", "ts_ref", "stat_type", "channel_id",
+                    "update_action", "price", "quantity", "flags",
+                    "ts_in_delta", "sequence"]
+            @test col in names(df)
+        end
+        @test df.price[1] == DBN.price_to_float(10055000000)
+        @test isnan(df.price[2])    # UNDEF_PRICE -> NaN
+        @test df.quantity == [500, typemax(Int64)]
+        @test df.flags == [0x20, 0x00]
+        @test df.stat_type == [8, 9]
+    end
+
+    @testset "issue #34: all symbol-mapping intervals survive a round-trip" begin
+        # A continuous contract's roll history: one tuple per interval, plus a
+        # second symbol to check grouping boundaries.
+        roll_history = [
+            ("ES.n.0", "ESH4", Int64(20240101), Int64(20240315)),
+            ("ES.n.0", "ESM4", Int64(20240315), Int64(20240621)),
+            ("ES.n.0", "ESU4", Int64(20240621), Int64(20240920)),
+            ("NQ.n.0", "NQH4", Int64(20240101), Int64(20240315)),
+        ]
+        metadata = Metadata(
+            UInt8(DBN_VERSION), "TEST.DATA", Schema.TRADES,
+            1640995200000000000, 1640995260000000000, UInt64(1000),
+            SType.CONTINUOUS, SType.RAW_SYMBOL, false,
+            ["ES.n.0", "NQ.n.0"], String[], String[],
+            roll_history,
+        )
+        hd = RecordHeader(12, RType.MBP_0_MSG, 1, 12345, 1640995200000000000)
+        trade = TradeMsg(hd, 10055000000, 250, Action.TRADE, Side.ASK,
+                         0x00, 1, 1640995200000000100, 2000, 87654)
+
+        tmp = tempname() * ".dbn"
+        try
+            write_dbn(tmp, metadata, [trade])
+            decoder = DBNDecoder(tmp)
+            try
+                @test decoder.metadata.mappings == roll_history
+            finally
+                close(decoder.io)
+                decoder.io === decoder.base_io || close(decoder.base_io)
+            end
+        finally
+            safe_rm(tmp)
+        end
+    end
+
+    @testset "issue #35: invalid enum bytes warn and default instead of throwing" begin
+        # Uniform policy across enums: never throw on a bad byte
+        @test DBN.safe_action(0x00) == Action.NONE
+        @test DBN.safe_action(UInt8('A')) == Action.ADD
+        @test DBN.safe_action(0x88) == Action.NONE
+        @test DBN.safe_side(0x00) == Side.NONE
+        @test DBN.safe_side(UInt8('B')) == Side.BID
+        @test DBN.safe_side(0x88) == Side.NONE
+        @test DBN.safe_instrument_class(UInt8('F')) == InstrumentClass.FUTURE
+        @test DBN.safe_instrument_class(0x74) == InstrumentClass.OTHER
+
+        # A corrupted action/side byte must not kill the stream: corrupt the
+        # first of two trade records on disk and decode both.
+        metadata = stat_metadata()
+        hd = RecordHeader(12, RType.MBP_0_MSG, 1, 12345, 1640995200000000000)
+        trades = [
+            TradeMsg(hd, 10055000000, 250, Action.TRADE, Side.ASK,
+                     0x00, 1, 1640995200000000100, 2000, 87654),
+            TradeMsg(hd, 10056000000, 300, Action.TRADE, Side.BID,
+                     0x00, 1, 1640995200000000200, 2000, 87655),
+        ]
+        tmp = tempname() * ".dbn"
+        try
+            write_dbn(tmp, metadata, trades)
+            bytes = read(tmp)
+            # TradeMsg = 48 bytes: header(16) + price(8) + size(4) + action(1)
+            # + side(1) + ... Records sit at the end of the file.
+            rec1 = length(bytes) - 96
+            @test bytes[rec1+29] == UInt8('T')   # action of record 1
+            @test bytes[rec1+30] == UInt8('A')   # side of record 1
+            bytes[rec1+29] = 0x88
+            bytes[rec1+30] = 0x88
+            write(tmp, bytes)
+
+            recs = read_dbn(tmp)
+            @test length(recs) == 2
+            @test recs[1].action == Action.NONE
+            @test recs[1].side == Side.NONE
+            # The following record still decodes correctly
+            @test recs[2].side == Side.BID
+            @test recs[2].sequence == 87655
+        finally
+            safe_rm(tmp)
+        end
+    end
+end