Crate archiver 

The archiver binary: writes LabJack scans from NATS JetStream to Parquet files.

For every LabJack read, the streamer publishes one FlatBuffer Scan per channel to the local NATS JetStream. The archiver reads the same dashboard configuration as the streamer (from KV bucket avenabox by default), attaches one durable pull consumer per enabled channel, rebuilds each sample’s timestamp from the scan header, and writes the samples to Parquet files under <PARQUET_DIR>/asset<NNN>/<YYYY-MM-DD>/ch<NN>/part-<NNNN>.parquet. Each file carries the channel’s calibration as key-value metadata. The exporter later serves these files as CSV over NATS.

At startup the archiver can mirror the configuration entry from a central NATS server into local KV, and it keeps mirroring updates in a background task. It then watches the local KV key and starts, stops or restarts channel writers as the configuration changes. Ctrl-C closes every open file and acks its messages before the process exits.

Configuration

• NATS_SERVERS - Comma-separated local NATS server URLs. Default: nats://127.0.0.1:4222.
• NATS_CREDS_FILE - Credentials file for the local NATS connection. Default: apt.creds.
• JS_DOMAIN - Optional JetStream domain for the local connection.
• PARQUET_DIR - Root directory for Parquet output. Default: parquet.
• CFG_BUCKET - Local KV bucket holding the dashboard configuration. Default: avenabox. Also the fallback for CENTRAL_CFG_BUCKET.
• CFG_KEY - Local KV key holding the dashboard configuration. Default: labjackd.config.macbook. Also the fallback for CENTRAL_CFG_KEY.
• CENTRAL_NATS_SERVERS - Central NATS server URLs to mirror configuration from. Central mirroring is off when neither this nor CFG_NATS_SERVERS is set.
• CFG_NATS_SERVERS - Fallback for CENTRAL_NATS_SERVERS.
• CENTRAL_CFG_BUCKET - Central KV bucket to mirror from. Default: CFG_BUCKET, then avenabox.
• CENTRAL_CFG_KEY - Central KV key to mirror from. Default: CFG_KEY, then unknown-site.macbook.unknown-source.config.
• CENTRAL_JS_DOMAIN - Optional JetStream domain on the central server. Falls back to CFG_JS_DOMAIN.
• CENTRAL_NATS_CREDS_FILE - Credentials file for the central server. Default: the value used for NATS_CREDS_FILE.
• ARCHIVER_STATE_DIR - Directory for the replay-guard checkpoints. Default: <PARQUET_DIR>/.archiver-state.
• ARCHIVER_REPLAY_GUARD - off (or 0, false, no) turns the replay guard off.

Design

• Aligned source-time rotation. A file covers one window of rotate_secs that starts at a multiple of rotate_secs since the Unix epoch (see rotation_window), measured in sample time rather than wall-clock time. Files line up across channels, and replaying a backlog of several hours still produces one file per window. A new file also starts at a UTC date change or when sample time goes backwards.
• One row group per file. A row group holds up to ROWS_PER_ROW_GROUP rows, which is more than a normal window, so a file is usually a single row group. Files are written with zstd compression and delta-encoded timestamps (see archive_format::writer_properties).
• Acks after close and fsync. A JetStream message is acked only after every sample in it is inside a file that has been closed, fsynced and renamed from .parquet.inprogress to .parquet (see ParquetLogger::close). If the process or machine dies first, the messages stay unacked and JetStream redelivers them once the consumer’s ack_wait expires, so they are written again into a new file.
• Early close at the pending-ack cap. When a file holds MAX_PENDING_ACKS_PER_FILE unacked messages it is closed early, which keeps the consumer below its max_ack_pending limit (CONSUMER_MAX_ACK_PENDING) so delivery never stalls.
• Idle close. A file that has received no data for IDLE_CLOSE_AFTER is closed, for example after the streamer stops, so its messages are acked promptly.
• Quarantine of unfinished files. At startup every leftover .parquet.inprogress file (no footer, from a crash) is renamed aside and kept for diagnosis (see quarantine_incomplete_files). Its samples come back through redelivery.
• Replay guard. JetStream can lose a durable consumer’s progress (its state file is replaced without an fsync, so a power cut can leave it empty) and then deliver the whole stream again under deliver policy all. Each channel saves the consumer’s ack floor to its own fsynced checkpoint and acks messages at or below it without writing them (see ReplayState and replay_guard).
• Filters are never edited in place. When a channel’s subject changes, an existing consumer with the old filter is left alone and a consumer named after the new subject is used (see filtered_consumer_name).

Modules

archive_format 🔒

On-disk format shared by the archiver and the recompress tool.

calibration 🔒

Calibration formulas stored with archived channel data.

nats_config 🔒

Shared NATS server-list and JetStream context helpers.

replay_guard 🔒

Protection against JetStream replaying messages the archiver already wrote.

sample_data_generated 🔒

subjects 🔒

NATS subject builders for LabJack live data and archive exports.

Structs

CentralKvSyncConfig 🔒

Connection and key details for mirroring dashboard config from central NATS.

ChannelLogger 🔒

Runtime state for one active channel consumer and writer task.

CloseOutcome 🔒

Files closed while handling one payload.

NestedConfig 🔒

Raw top-level dashboard configuration loaded from NATS KV.

ParquetLogger 🔒

Buffered writer for one channel’s Parquet file.

ReplayState 🔒

Replay protection for one channel writer.

SampleConfig 🔒

Normalized archiver configuration used to manage channel loggers.

SensorConfig 🔒

Raw sensor settings section from the dashboard configuration.

Constants

CHECKPOINT_INTERVAL 🔒

How often each writer task saves its replay-guard checkpoint (60 s).

CONSUMER_MAX_ACK_PENDING 🔒

max_ack_pending set on each durable consumer, in messages.

IDLE_CHECK_INTERVAL 🔒

How often each writer task checks for an idle file (15 s).

IDLE_CLOSE_AFTER 🔒

Time without new data after which the open file is closed (60 s).

MAX_PENDING_ACKS_PER_FILE 🔒

Unacked messages that force the open file to close early.

MAX_ROW_GROUPS_PER_FILE 🔒

Maximum row groups in one file before it is rotated.

ROWS_PER_ROW_GROUP 🔒

Rows buffered before they are written as one row group (1,048,576).

SKIP_LOG_EVERY 🔒

Skipped messages between two progress lines while the replay guard is skipping.

Functions

ack_messages 🔒

Acks messages whose samples are all inside closed, synced Parquet files.

archiver_consumer_config 🔒

Consumer settings that let acks wait until a file is closed.

archiver_consumer_name 🔒

Builds the durable JetStream consumer name for one archived channel.

central_kv_sync_config_from_env 🔒

Builds optional central KV mirroring settings from environment variables.

close_and_settle 🔒

Closes the open file (if any) and settles the messages it holds.

connect_nats_with_creds 🔒

Connects to NATS with a credentials file and explicit server list.

ensure_kv_bucket 🔒

Opens or creates the KV bucket that contains dashboard configuration.

env_nonempty 🔒

Returns a trimmed environment variable value when it is set and non-empty.

filtered_consumer_name 🔒

Durable consumer name used when the base consumer filters on another subject.

main 🔒

Starts the archiver service and reacts to configuration updates.

mirror_central_kv_once 🔒

Copies the configured central KV entry into local KV during startup.

next_file_index 🔒

Scans a channel/day directory to find the next available Parquet file index.

parse_calibrations 🔒

Parses dashboard calibration map keys into numeric LabJack channels.

process_scan_payload 🔒

Decodes one FlatBuffer scan payload and writes its samples to Parquet.

quarantine_incomplete_files 🔒

Renames unfinished writer files under root so they are kept but never read.

replay_state_dir 🔒

Pure part of replay_state_dir_from_env.

replay_state_dir_from_env 🔒

Reads the replay-guard settings.

representable_samples 🔒

Counts the leading samples of a scan whose timestamps fit in an i64.

rotation_window 🔒

Returns the index of the aligned rotation window holding a timestamp.

run_central_kv_sync 🔒

Continuously mirrors central KV updates into the local configuration bucket.

sample_config_from_nested 🔒

Converts the nested dashboard configuration into archiver runtime config.

sample_timestamp_ns 🔒

Computes the Unix timestamp for a sample index within a FlatBuffer scan.

sanitize_consumer_token 🔒

Converts source identity text into a durable consumer-name token.

spawn_channel_logger 🔒

Starts the durable pull consumer and writer task for one channel.

sync_directory_after_publish 🔒

Fsyncs the directory of a file that has just been renamed into place.

timestamp_ns_to_utc_date 🔒

Converts a Unix nanosecond timestamp to its UTC calendar date.

Type Aliases

DynError 🔒

Error type used by the archiver for fallible async setup and IO paths.