Crate exporter 

Serves archived Parquet sample files as streamed CSV exports.

The streamer publishes LabJack samples to the local NATS JetStream, and the archiver writes them to Parquet files under <PARQUET_DIR>/assetNNN/YYYY-MM-DD/chNN/*.parquet. This binary reads those files back for a requested asset, channel list and time range, applies the calibration stored in each file’s metadata, and sends the rows as CSV with the columns timestamp,channel,raw_value,calibrated_value,calibration_id. A request that names noise filters for a channel (ExportRequest::filters) gets a sixth column, filtered_value, computed by the zero-phase pipeline in filters over the range plus margins; the archive itself is never changed.

It runs in one of two modes that share the request format (ExportRequest) and the scanning logic:

• Worker mode (the default and the normal edge-box setup) subscribes to the core NATS subject <NATS_SUBJECT>.<site>.<box>.<source_id>.export.request and answers each request on its reply subject. Every reply message carries the header Avena-Export-Frame naming the frame: meta, then one or more chunk frames with raw CSV bytes (about 512 KiB each, the last one smaller), then summary and complete. Failures produce an error frame. See process_nats_request and NatsCsvStreamer for the exact payloads, the ack-subject backpressure and the cancel message a client can send there.
• Direct mode serves the same export over a WebSocket at ws://<EXPORTER_ADDR>/export. The client sends one JSON request as a text message and receives the same frames as JSON text messages, with CSV chunks (about 128 KiB) as binary messages. See process_socket.

Configuration

• EXPORTER_MODE - worker, or direct (alias local). Trimmed and case-insensitive; any other value is a startup error. Default: worker.
• EXPORTER_ADDR - TCP listen address for the WebSocket server in direct mode. Default: 0.0.0.0:9001.
• PARQUET_DIR - Root of the Parquet archive. A missing directory only logs a warning at startup. Default: parquet.
• NATS_SERVERS - Worker mode. Comma-separated NATS server URLs. Default: nats://127.0.0.1:4222.
• NATS_CREDS_FILE - Worker mode. NATS credentials file. Default: apt.creds.
• NATS_SUBJECT - Worker mode. Root token of the request subject. Default: avenars.
• SITE_ID - Worker mode. Site token of the request subject. Default: unknown-site.
• EXPORT_BOX_ID - Worker mode. Box token of the request subject; falls back to BOX_ID. One of the two is required in worker mode.
• BOX_ID - Worker mode. Used when EXPORT_BOX_ID is unset.
• SOURCE_ID - Worker mode. Source token of the request subject. When it is unset or empty the token falls back to LABJACK_NAME, then asset<ASSET_NUMBER>, then unknown-source, the same order the streamer and the webapp use (see subjects::archive_export_request_subject).
• LABJACK_NAME - Worker mode. The source’s LabJack name, used only when SOURCE_ID is not set.
• ASSET_NUMBER - Worker mode. The source’s asset number, used only when neither SOURCE_ID nor LABJACK_NAME is set.
• SOURCE_TYPE - Worker mode. Read and passed to the subject builder, which currently ignores it. Default: labjack.

Design

• Worker mode exists so the webapp needs only its existing WebSocket connection to central NATS. The browser publishes a request with a reply inbox, the NATS leaf routes it to the worker on the box that holds the Parquet files, and the CSV comes back the same way. Direct mode requires the client to reach the box itself. Core NATS has no flow control, so the worker waits for client acks on ack_subject every 8 chunks instead of flooding a slow browser.
• Each Parquet row group stores min/max statistics for the timestamp column. read_matching_rows skips row groups whose range lies outside the request without decoding them, and reads the two columns with typed readers rather than building a row object per sample.
• The archive can hold more than one copy of a sample: an archiver fed the same JetStream messages again writes them into new part files, sometimes with other window boundaries or another calibration. Each channel-day’s files are grouped by overlapping time spans (overlapping_file_groups), found from the footer statistics. A file that overlaps nothing is read in one pass, as before. Overlapping files are read together in five-minute slices so memory stays bounded by one slice rather than a day. Either way the rows are sorted by time and each exact (timestamp, raw value) pair is sent once (read_merged_rows); a clean file in time order comes out exactly as stored. Rows with the same timestamp but different values are all kept.
• Each export runs as a three-stage pipeline (spawn_csv): a reader thread lists and decodes the files (and runs the filters) one merge slice ahead, a CSV thread formats the rows into chunks, and the request’s async task sends the chunks and handles acks. Decoding, formatting and sending overlap, and no file reading or formatting blocks the async runtime; the bytes and their order are the same as reading, formatting and sending one after another.
• Formatting a row is the main CPU cost of an export, so CsvRowWriter avoids most of it while writing the same bytes as writeln!: rows are sent in time order, so Rfc3339Formatter builds the date and time up to the second with chrono once per second and writes only the fractional digits per row (identical text to chrono’s to_rfc3339), and ValueTextCache reuses the raw,calibrated text of raw values seen before, which with a 16-bit converter is nearly every row.

Modules

calibration 🔒

Calibration formulas stored with archived channel data.

filters 🔒

Noise filters applied to exported data (never to the archive).

nats_config 🔒

Shared NATS server-list and JetStream context helpers.

subjects 🔒

NATS subject builders for LabJack live data and archive exports.

Structs

AppState 🔒

State shared by the Axum WebSocket handler and the NATS worker tasks.

CsvBuilder 🔒

Formats an export’s rows into CSV chunks on its CSV thread; see spawn_csv.

CsvJob 🔒

What an export reads, handed to its CSV thread (spawn_csv).

CsvRowWriter 🔒

Writes export CSV rows into a chunk buffer.

CsvStreamer 🔒

CSV streamer for direct WebSocket exports.

ErrorFrame 🔒

Direct-mode error frame sent when an export request cannot be served.

ExportCancelled 🔒

Error returned when the client cancels a worker-mode export.

ExportFilters 🔒

Filters asked for by an export request, per channel.

ExportRequest 🔒

Export request accepted by both WebSocket and NATS worker modes.

FileSpan 🔒

Timestamp span of one archived part file, read from its footer.

FilterInputRow 🔒

One row of a filtered export.

FilteredChannelRows 🔒

Reads one channel over the export range plus lead-in and settling margins and runs it through the zero-phase filter pipeline, one merge slice at a time.

MatchedFile 🔒

Rows of one Parquet file that fall inside an export range.

MergedRows 🔒

Rows of one or more files, sorted by time with exact duplicates removed.

MetaFrame 🔒

Direct-mode metadata frame sent before CSV chunks.

NatsCsvStreamer 🔒

CSV streamer for NATS worker exports.

Rfc3339Formatter 🔒

Formats timestamps exactly like chrono’s to_rfc3339, faster.

SummaryFrame 🔒

Direct-mode summary frame sent after all CSV chunks are delivered.

ValueSlot 🔒

One cached raw,calibrated text of ValueTextCache.

ValueTextCache 🔒

Remembers the raw_value,calibrated_value text of recently written raw values.

WebSocketSink 🔒

WebSocket implementation of ExportSink.

Enums

ExportFormat 🔒

Requested export format, written in JSON as "csv" or "parquet".

ExporterMode 🔒

Exporter runtime mode.

FilteredColumn 🔒

Value column of a CSV row after calibration_id, if the export has one.

ReadItem 🔒

Rows of an export in CSV order, from read_rows to CsvBuilder.

Constants

CHUNK_SLACK 🔒

Bytes reserved in a chunk buffer beyond its chunk size, so the row that crosses the size normally fits without the buffer being reallocated and copied.

CSV_CHUNKS_AHEAD 🔒

Finished chunks the CSV thread may hold ready while the previous ones are sent.

CSV_HEADER 🔒

CSV header of an export without filters.

CSV_HEADER_FILTERED 🔒

CSV header of an export with filters.

DEFAULT_EXPORTER_ADDR 🔒

WebSocket listen address used in direct mode when EXPORTER_ADDR is unset.

DEFAULT_EXPORTER_MODE 🔒

Runtime mode used when EXPORTER_MODE is unset.

EXPORT_FRAME_CANCEL 🔒

Frame name a client sends on its ack_subject to stop an export.

EXPORT_FRAME_CHUNK 🔒

Frame name for a CSV payload chunk (raw bytes, not JSON).

EXPORT_FRAME_COMPLETE 🔒

Frame name marking the end of a successful export.

EXPORT_FRAME_ERROR 🔒

Frame name for request validation or processing errors (message).

EXPORT_FRAME_HEADER 🔒

NATS header whose value names the frame type of each worker-mode reply message.

EXPORT_FRAME_META 🔒

Frame name for the metadata frame (fileName, contentType) sent before any CSV.

EXPORT_FRAME_SUMMARY 🔒

Frame name for the final byte count and missing-channel list.

FILTER_LEAD_IN_S 🔒

Seconds read before the export range so the templates have learnt the interference by the first exported row (their memory).

MERGE_SLICE_NS 🔒

Length of the time slices in which overlapping files are merged (300 s).

VALUE_CACHE_BYPASS 🔒

Rows a ValueTextCache formats directly after a window with fewer than one hit in four, before it tries the table again.

VALUE_CACHE_SLOTS 🔒

Number of slots in ValueTextCache; a power of two.

VALUE_CACHE_TEXT 🔒

Longest raw,calibrated text a ValueTextCache slot holds, bytes.

VALUE_CACHE_WINDOW 🔒

Lookups over which ValueTextCache measures its hit rate.

Traits

ExportSink 🔒

Transport abstraction for framed export responses.

Functions

calibration_preference 🔒

Returns how strongly a copy’s calibration is preferred when copies collide.

connect_nats_from_env 🔒

Connects to NATS using the standard credentials and server environment.

date_range 🔒

Builds an inclusive list of UTC dates covered by an export request.

datetime_to_unix_ns 🔒

Converts an instant to Unix nanoseconds, clamping outside the i64 range.

default_format 🔒

Default export format used when a request omits format.

file_time_span 🔒

Finds the earliest and latest timestamp in one archived file.

handle_ws 🔒

Upgrades an HTTP request into a WebSocket export session.

is_cancel_message 🔒

Tells whether a message on the ack subject is a cancel request.

main 🔒

Starts the exporter in direct WebSocket mode or NATS worker mode.

merge_slices 🔒

Splits the export range covered by one file group into merge slices.

overlapping_file_groups 🔒

Lists one channel-day folder and groups its files by overlapping time spans.

parse_range 🔒

Parses an RFC 3339 start/end range into UTC instants.

process_nats_request 🔒

Validates and serves one NATS export request payload.

process_socket 🔒

Handles one direct-mode WebSocket export session.

publish_nats_json 🔒

Publishes a JSON export control frame on a NATS reply subject.

push_fraction 🔒

Appends . and value as exactly width decimal digits, zero-padded, like write!(out, ".{value:0width$}") for a value below 10^width.

read_calibration_from_metadata 🔒

Reads calibration metadata written by the archiver from a Parquet file.

read_column 🔒

Reads every value of one required column in a row group.

read_export_request 🔒

Reads the initial JSON export request from a WebSocket.

read_matching_rows 🔒

Reads the rows of one archived file whose timestamps fall in [start_ns, end_ns].

read_merged_rows 🔒

Reads one slice of a file group, merged into time order without duplicates.

read_rows 🔒

Reads the rows of every channel of an export, in order, and hands them over.

row_group_may_match 🔒

Returns whether a row group’s timestamp statistics overlap [start_ns, end_ns].

run_worker 🔒

Runs exporter worker mode by subscribing to one NATS request subject.

same_formula 🔒

Tells whether two formulas give the same result for every input: same kind and coefficients with the same bits.

sanitize_token 🔒

Normalizes namespace text into a subject-safe token.

serve_export_request 🔒

Validates a request and streams its CSV response through an export sink.

sort_and_drop_duplicates 🔒

Sorts rows by timestamp and removes exact duplicates.

spawn_csv 🔒

Starts building an export’s CSV on a blocking thread.

worker_box_id_from_env 🔒

Reads the box identifier required to construct the worker request subject.

write_values 🔒

Appends {raw},{calibrated} to out, calibrated being spec applied to raw.

Type Aliases

CalibrationTable 🔒

Calibrations and CSV calibration ids of a filtered channel, each once, in the order first seen. Shared with the CSV thread, which reads a batch’s rows with it.