Table of Contents

Class DbnFsm

Namespace
DatabentoDotNet.Dbn
Assembly
DatabentoDotNet.Dbn.dll

The incremental DBN decoder: a sans-I/O state machine that turns a byte stream arriving in arbitrary-sized pieces into metadata and records, without ever touching a Stream or a socket itself.

public sealed class DbnFsm
Inheritance
DbnFsm
Inherited Members

Remarks

The port of upstream's DbnFsm (decode/dbn/fsm.rs). The caller owns the I/O: it writes bytes into the span Space() returns, tells the machine how many with Fill(int), and pulls records out with TryNextRecord(out RecordRef). A TCP socket delivers a DBN stream in pieces that have nothing to do with record boundaries — a single byte is a perfectly ordinary read — which is the entire reason this is a state machine and not a loop over a reader.

Three states, not upstream's four. Upstream has Prelude, Metadata { length }, Record, and a fourth, Consume { read, compat, compat_fill, expand_compat }, whose own doc comment says it exists to "get around mutability requirements" (fsm.rs:62). It models nothing about DBN: it defers the buffer index advance to the next call so that Rust's borrow checker will let the just-decoded record still be read through last_record(). C# has no such restriction, so at each of the three points where upstream constructs State::Consume this port performs the advance inline and hands the record back from the same call. Nothing replaces the state.

Why consuming immediately is safe. Consume(int) and Fill(int) move indices only — they never copy or clear a byte. So a span captured over the record before the advance still points at the same untouched bytes after it.

A record is only valid until the next call on this machine. Read it, or copy what you need out of it, before calling anything else here. Two things end its life, and they are not the same thing:

  • Space() may Shift() the read buffer's unconsumed tail down to offset 0, moving the bytes a read-buffer-backed record points at.
  • An upgraded record lives in the compat buffer, and the next upgraded record overwrites it. The compat buffer is a single-record scratchpad: it is reset to offset 0 as each upgraded record is handed out, so the next one is written straight over the previous one's bytes. No shift is involved and no memory is freed — the bytes simply become a different record. On a v1 or v2 definition or statistics stream, holding two records across two TryNextRecord(out RecordRef) calls therefore gives two references to the same, latest record.

This is the same one-record-at-a-time contract upstream enforces through Rust's borrow checker, which will not let a RecordRef from last_record() outlive the next &mut self call. C# cannot enforce it, so it is stated here instead; RecordRef being a ref struct narrows the blast radius by making a record impossible to box, store in a field, or capture in a closure, but it does not stop a caller holding two of them in the same method. Callers that genuinely need several records at once must copy the bytes out, which is what the decoder tests do.

No async here, on purpose. TryNextRecord(out RecordRef) is synchronous, and RecordRef is a ref struct and so cannot be preserved across an await (CS4007). The asynchronous I/O layer belongs above this type: it awaits a read into SpaceMemory() and calls Fill(int). Draining records may sit in the very same async method — the compiler only objects to a record that survives an await, which is exactly the lifetime rule this type already imposes. See the remarks on SpaceMemory() for the whole loop.

Constructors

DbnFsm(VersionUpgradePolicy, bool, byte?, bool, int)

Creates a state machine.

public DbnFsm(VersionUpgradePolicy upgradePolicy = VersionUpgradePolicy.UpgradeToV3, bool skipMetadata = false, byte? inputDbnVersion = null, bool tsOut = false, int bufferSize = 65536)

Parameters

upgradePolicy VersionUpgradePolicy

How to present records from an older DBN version. The default matches upstream's and converts v1 and v2 records to v3 as they are decoded.

skipMetadata bool

true to start directly in the record state, for a DBN fragment: a bare run of records with no magic prelude and no metadata block.

inputDbnVersion byte?

The DBN version of the input, when it is known ahead of time. Only meaningful together with skipMetadata — otherwise the metadata block states the version and overwrites this. When it is null and records need upgrading, the version is inferred from each record's size, exactly as upstream does.

tsOut bool

Whether every record carries an appended 8-byte ts_out send timestamp. Only meaningful together with skipMetadata; otherwise the metadata says.

bufferSize int

The read buffer's size in bytes. Never smaller than MaxRecordLength — AlignedBuffer enforces that floor — because a buffer that cannot hold the largest possible record could never present one contiguously.

Exceptions

ArgumentOutOfRangeException

upgradePolicy is not a defined value, inputDbnVersion is outside the versions this codec decodes, or bufferSize is negative.

DbnDecodeException

inputDbnVersion and upgradePolicy are incompatible — asking to upgrade a v3 stream to v2, which is a downgrade.

Fields

DefaultBufferSize

The default read-buffer size: 64 KiB, matching upstream's DEFAULT_BUF_SIZE.

public const int DefaultBufferSize = 65536

Field Value

int

Properties

BufferedByteCount

Bytes sitting in the read buffer that no record and no metadata block has claimed yet.

public int BufferedByteCount { get; }

Property Value

int

Remarks

This is what tells a truncated stream from one that simply ended. Both look identical through TryNextRecord(out RecordRef), which answers false either way — a stream ending between records is not an error, and neither the machine nor DbnDecoder treats it as one. But a stream that ends part-way through a record leaves those bytes here, unclaimed and unclaimable, and a reader that has hit end-of-stream with a non-zero count is holding the front of a record whose tail never arrived.

Read it only after TryNextRecord(out RecordRef) has answered false: before that, a non-zero count is the ordinary state of a buffer with records still in it, and means nothing at all. TimeseriesClient is the caller this was added for (#38), where a download cut short has to be an exception rather than a short read.

InputDbnVersion

The DBN version of the input as currently known: from the prelude, from the constructor, or inferred from a record's size. null when still unknown.

public byte? InputDbnVersion { get; }

Property Value

byte?

IsDecodingRecords

true once the machine has left the metadata phase and is decoding records — after the metadata block has been decoded, or immediately when the machine was told to skip it.

public bool IsDecodingRecords { get; }

Property Value

bool

Remarks

Port of upstream's has_decoded_metadata() (fsm.rs:276-278), renamed because the upstream name states something this property does not check. A fragment decoder (skipMetadata: true) reports true from its very first instant, with Metadata null and no metadata block anywhere on the stream — nothing was decoded, and nothing ever will be. What both upstream and every call site actually mean by it is "the machine is past the metadata phase", which is what the name now says.

To ask whether metadata is available, test Metadata for null. That is the only question this property was never answering.

Metadata

The stream's decoded metadata, or null until it has been decoded — which it never is for a fragment.

public Metadata? Metadata { get; }

Property Value

Metadata

Remarks

Already presented according to the upgrade policy, so a v1 stream decoded under UpgradeToV3 reports version 3 here. The input version, which is what drives record upgrades, is InputDbnVersion.

TsOut

Whether every record on this stream carries an appended 8-byte ts_out.

public bool TsOut { get; }

Property Value

bool

UpgradePolicy

The upgrade policy this machine applies to records from older DBN versions.

public VersionUpgradePolicy UpgradePolicy { get; }

Property Value

VersionUpgradePolicy

Methods

Fill(int)

Records that nbytes bytes were written into the span the last Space() call returned.

public void Fill(int nbytes)

Parameters

nbytes int

How many bytes were written. Capped at the space actually available.

Exceptions

ArgumentOutOfRangeException

nbytes is negative.

Process(out int, out RecordRef)

Advances the state machine by one step: decodes the metadata block, or one record, or reports that more bytes are needed.

public ProcessStatus Process(out int bytesNeeded, out RecordRef record)

Parameters

bytesNeeded int

When the result is NeedMoreData, how many more bytes are known to be required before this step can complete; zero otherwise. A hint for sizing the next read, not a contract — supplying fewer just means another NeedMoreData, and supplying more is fine.

record RecordRef

When the result is Record, the decoded record; otherwise default. Valid only until the next call on this machine — an upgraded record is overwritten by the next upgraded record, without any shift; see the remarks on DbnFsm.

Returns

ProcessStatus

What this step produced.

Remarks

The record comes back from the same call that decoded it rather than from a separate "last record" accessor. Upstream needs the accessor because its State::Consume defers the buffer advance across the call boundary; with the advance done inline there is no second call to hang the result off, and no window in which a caller could ask for "the last record" after the machine has moved on.

Exceptions

DbnDecodeException

The buffered bytes are not valid DBN.

Reset()

Returns the machine to its starting state — the prelude, or the first record when it was built to skip metadata — and discards every buffered byte and everything learned about the stream.

public void Reset()

Remarks

Buffer capacity is kept, so a reset costs no allocation. Unlike upstream's reset(), which always returns to the prelude state, this returns to whichever state the machine started in: resetting a fragment decoder into a prelude it will never see would leave it permanently stuck.

Space()

The writable tail of the read buffer. Write bytes here, then call Fill(int) with how many.

public Span<byte> Space()

Returns

Span<byte>

A span of at least MaxRecordLength bytes whenever the buffer's capacity allows, reclaiming the consumed prefix first if it has to.

Remarks

This is the one call that can move buffered bytes, because it may shift the unconsumed tail down to offset 0 to make room. Any RecordRef obtained before it is stale afterwards. Reclaiming here rather than after every record is deliberate: Consume(int) only moves an index, so the memory move is paid once per refill instead of once per record.

SpaceMemory()

The writable tail of the read buffer as a Memory<T> — the same bytes Space() returns, in the form asynchronous I/O requires. Write bytes here, then call Fill(int) with how many.

public Memory<byte> SpaceMemory()

Returns

Memory<byte>

A Memory<T> of at least MaxRecordLength bytes whenever the buffer's capacity allows, reclaiming the consumed prefix first if it has to.

Remarks

This is the async read seam (#15). The canonical loop over a Stream — a socket, or a decompressor wrapped around one — is: 

while (true)
{
    var read = await stream.ReadAsync(fsm.SpaceMemory(), ct);
    if (read == 0) { break; }                        // stream ended; not an error
    fsm.Fill(read);

    while (fsm.TryNextRecord(out var record))        // legal here: no record outlives an await
    {
        Process(record);
    }
}

Draining sits in the same method, and the compiler enforces the one rule that matters. RecordRef is a ref struct; a local of one is perfectly legal inside an async method, and only surviving an await is rejected — as CS4007, "cannot be preserved across 'await' or 'yield' boundary". That is precisely the lifetime rule this type already imposes by hand, now checked at compile time.

What remains impossible is returning one: no async method can return a ref struct, so there is no Task<RecordRef> and upstream's single-call LiveClient::next_record() has no .NET equivalent on the zero-copy path. Its fill_buf() / try_next_record() pair (live/client.rs:386-436) does, and that pair is the shape the M2 live client takes.

Everything Space() warns about applies here too: this call can shift buffered bytes, so any RecordRef obtained before it is stale afterwards, and the returned Memory<T> must be taken fresh for each read rather than cached across iterations.

It is also where the buffer grows for an oversized metadata block (#31), which is what makes this guarantee hold: while the machine still needs metadata bytes, this never hands back an empty span. That is not a nicety. The loop above reads a zero-byte span as end-of-stream, so a buffer that filled with an incomplete metadata block and could not grow would turn a perfectly good stream into "truncated metadata" — a wrong answer rather than an exception. Growth lives at this seam, and not beside the NeedMoreData that asked for the bytes, precisely so it cannot depend on the caller having called Process(out int, out RecordRef) in between.

TryNextRecord(out RecordRef)

Decodes the next record, if the buffered bytes hold a complete one.

public bool TryNextRecord(out RecordRef record)

Parameters

record RecordRef

Receives the decoded record. Valid only until the next call on this machine — read it, or copy what you need out of it, before calling anything else here. An upgraded record in particular is overwritten by the next upgraded record, without any shift; see the remarks on DbnFsm.

Returns

bool

true when a record was decoded. false means "not enough bytes yet" — write more into Space(), Fill(int), and call again. A stream that has ended simply keeps returning false; that is not an error.

Remarks

Metadata decoded along the way is not returned here — it lands in Metadata and the machine carries straight on to the first record.

Exceptions

DbnDecodeException

The buffered bytes are not valid DBN.

Edit this page