graham/prop
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat(prop-grid-rs): MessagePack ProfilesFile writer

Browse source 
Rust-side writer for the f00 ProfilesFile. Replaces the Elixir ETF
format (:erlang.term_to_binary) with MessagePack + gzip so Rust can
produce files without porting the Erlang serializer.

Wire layout: top-level map with 'v' (format version), 'valid_time'
(ISO string), and 'cells' (array of {lat, lon, profile}). Profile
values are an rmpv::Value so the writer stays agnostic about what
fields the f00 pipeline decides to include — native-duct metrics,
NEXRAD reflectivity, commercial-link degradation, and the sounding
profile list all serialize through the same path.

All keys are strings — MessagePack has no atom type. Elixir's reader
(follow-up commit) converts known keys to atoms via a whitelist.

Atomic writes via rename(2) mirror ScoresFile/ProfilesFile pattern:
write to tmp sibling, rename into place on the shared NFS mount.
4 unit tests cover round-trip, atomic rename, path layout, and
lat/lon snap.
This commit is contained in:
Graham McIntire 2026-04-19 17:59:40 -05:00
parent d145dad3c7
commit 1307b1d8ac
No known key found for this signature in database
GPG key ID: F4ABF488E6029E59
4 changed files with 269 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

43
rust/prop_grid_rs/Cargo.lock generated
View file

@ -1355,6 +1355,7 @@ dependencies = [
"byteorder",
"bytes",
"chrono",
"flate2",
"futures",
"num_cpus",
"png",
@ -1362,6 +1363,8 @@ dependencies = [
"prometheus",
"rayon",
"reqwest",
"rmp-serde",
"rmpv",
"serde",
"serde_json",
"sqlx",
@ -1618,6 +1621,36 @@ dependencies = [
"windows-sys 0.52.0",
]
[[package]]
name = "rmp"
version = "0.8.15"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "4ba8be72d372b2c9b35542551678538b562e7cf86c3315773cae48dfbfe7790c"
dependencies = [
"num-traits",
]
[[package]]
name = "rmp-serde"
version = "1.3.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "72f81bee8c8ef9b577d1681a70ebbc962c232461e397b22c208c43c04b67a155"
dependencies = [
"rmp",
"serde",
]
[[package]]
name = "rmpv"
version = "1.3.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7a4e1d4b9b938a26d2996af33229f0ca0956c652c1375067f0b45291c1df8417"
dependencies = [
"rmp",
"serde",
"serde_bytes",
]
[[package]]
name = "rsa"
version = "0.9.10"
@ -1726,6 +1759,16 @@ dependencies = [
"serde_derive",
]
[[package]]
name = "serde_bytes"
version = "0.11.19"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a5d440709e79d88e51ac01c4b72fc6cb7314017bb7da9eeff678aa94c10e3ea8"
dependencies = [
"serde",
"serde_core",
]
[[package]]
name = "serde_core"
version = "1.0.228"

3
rust/prop_grid_rs/Cargo.toml
View file

@ -33,6 +33,9 @@ rayon = "1.10"
axum = { version = "0.7", default-features = false, features = ["http1", "tokio"] }
prometheus = { version = "0.13", default-features = false }
png = "0.17"
rmp-serde = "1.3"
rmpv = { version = "1.3", features = ["with-serde"] }
flate2 = "1"
[dev-dependencies]
tokio = { version = "1", features = ["full", "test-util"] }

1
rust/prop_grid_rs/src/lib.rs
View file

@ -29,6 +29,7 @@ pub mod metrics;
pub mod native_duct;
pub mod nexrad;
pub mod pipeline;
pub mod profiles_file;
pub mod region;
pub mod scores_file;
pub mod scorer;

222
rust/prop_grid_rs/src/profiles_file.rs Normal file
View file

@ -0,0 +1,222 @@
//! On-disk profile store in MessagePack + gzip.
//!
//! Rust-side writer for the f00 ProfilesFile. Replaces the Elixir ETF
//! format (`:erlang.term_to_binary/2`, `.etf.gz`) so Rust can produce
//! the files without porting the Erlang serializer. Elixir's
//! `ProfilesFile` module learns to read `.mp.gz` alongside the legacy
//! `.etf.gz` during the cutover.
//!
//! Wire layout (top-level msgpack map):
//!
//! ```text
//! {
//! "v": 1,
//! "valid_time": "2026-04-19T14:00:00Z",
//! "cells": [
//! {
//! "lat": 32.9, "lon": -97.0,
//! "profile": {
//! "native_min_gradient": -150.5,
//! "best_duct_freq_ghz": 24.0,
//! "duct_count": 1,
//! "ducts": [
//! {"base_m": 100, "top_m": 200, "thickness_m": 100,
//! "m_deficit": 12.3, "min_freq_ghz": 15.2}
//! ],
//! "surface_temp_c": 22.5,
//! "surface_dewpoint_c": 18.1,
//! "surface_pressure_mb": 1013.2,
//! "hpbl_m": 320.0,
//! "pwat_mm": 25.3,
//! "profile": [
//! {"pres_mb":1000,"hght_m":100,"tmpc":20.0,"dwpc":18.0},
//! ...
//! ]
//! }
//! }
//! ]
//! }
//! ```
//!
//! All keys are strings — MessagePack has no atom type, and Elixir's
//! reader converts known keys back to atoms via a whitelist.
use std::fs::File;
use std::io::{BufWriter, Write};
use std::path::{Path, PathBuf};
use chrono::{DateTime, Utc};
use flate2::Compression;
use flate2::write::GzEncoder;
use serde::Serialize;
/// Bump whenever the on-disk layout changes in a way that isn't a
/// transparent superset. Elixir reader gates on this.
pub const FORMAT_VERSION: u32 = 1;
/// One grid cell to persist. `profile` is an arbitrary map of
/// string-keyed values — sized at write time. Kept as a
/// `rmp_serde::Value` so the writer stays agnostic about what the
/// f00 pipeline decides to include.
#[derive(Debug, Serialize)]
pub struct CellEntry {
pub lat: f64,
pub lon: f64,
pub profile: rmpv::Value,
}
#[derive(Debug, Serialize)]
struct FileBody<'a> {
v: u32,
valid_time: String,
cells: &'a [CellEntry],
}
#[derive(Debug, thiserror::Error)]
pub enum WriteError {
#[error("io: {0}")]
Io(#[from] std::io::Error),
#[error("encode: {0}")]
Encode(#[from] rmp_serde::encode::Error),
}
/// Absolute path the writer lands at for `valid_time`.
/// Sibling of the Elixir `.etf.gz` file; the Elixir reader prefers
/// `.mp.gz` when both exist.
pub fn path_for(scores_dir: &Path, valid_time: DateTime<Utc>) -> PathBuf {
let iso = valid_time.format("%Y-%m-%dT%H:%M:%SZ").to_string();
scores_dir.join("profiles").join(format!("{iso}.mp.gz"))
}
/// Write `cells` to `<scores_dir>/profiles/<iso>.mp.gz` atomically.
/// Pattern mirrors `scores_file::write_atomic`: write to a tmp sibling,
/// rename into place. The NFS rename is atomic, so a concurrent reader
/// sees either the old file, the new file, or nothing — never a torn
/// write.
pub fn write_atomic(
scores_dir: &Path,
valid_time: DateTime<Utc>,
cells: &[CellEntry],
) -> Result<PathBuf, WriteError> {
let path = path_for(scores_dir, valid_time);
if let Some(parent) = path.parent() {
std::fs::create_dir_all(parent)?;
}
let nanos = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_nanos())
.unwrap_or(0);
let pid = std::process::id();
let mut tmp = path.clone().into_os_string();
tmp.push(format!(".tmp.{nanos}.{pid}"));
let tmp = PathBuf::from(tmp);
{
let file = File::create(&tmp)?;
let mut gz = GzEncoder::new(BufWriter::new(file), Compression::default());
let body = FileBody {
v: FORMAT_VERSION,
valid_time: valid_time.format("%Y-%m-%dT%H:%M:%SZ").to_string(),
cells,
};
let buf = rmp_serde::to_vec_named(&body)?;
gz.write_all(&buf)?;
gz.finish()?;
}
std::fs::rename(&tmp, &path)?;
Ok(path)
}
/// Round a lat/lon to 3 decimal places — matches Elixir's
/// `Float.round(_, 3)` snap before keying into the profiles map.
pub fn snap_coords(lat: f64, lon: f64) -> (f64, f64) {
((lat * 1000.0).round() / 1000.0, (lon * 1000.0).round() / 1000.0)
}
/// Convenience helper: build a `rmpv::Value` map from an
/// iterator of `(name, value)` pairs.
pub fn value_map(entries: impl IntoIterator<Item = (&'static str, rmpv::Value)>) -> rmpv::Value {
let pairs: Vec<(rmpv::Value, rmpv::Value)> = entries
.into_iter()
.map(|(k, v)| (rmpv::Value::String(k.into()), v))
.collect();
rmpv::Value::Map(pairs)
}
#[cfg(test)]
mod tests {
use super::*;
use chrono::TimeZone;
use flate2::read::GzDecoder;
use std::io::Read;
#[test]
fn writes_and_reads_round_trip() {
let dir = tempfile::tempdir().unwrap();
let vt = Utc.with_ymd_and_hms(2026, 4, 19, 14, 0, 0).unwrap();
let cells = vec![CellEntry {
lat: 32.9,
lon: -97.0,
profile: value_map([
("surface_temp_c", rmpv::Value::F64(22.5)),
("surface_pressure_mb", rmpv::Value::F64(1013.2)),
("duct_count", rmpv::Value::Integer(1.into())),
]),
}];
let path = write_atomic(dir.path(), vt, &cells).unwrap();
assert!(path.exists());
assert!(path.to_string_lossy().ends_with(".mp.gz"));
// Decode and confirm the top-level shape.
let raw = std::fs::read(&path).unwrap();
let mut gz = GzDecoder::new(&raw[..]);
let mut buf = Vec::new();
gz.read_to_end(&mut buf).unwrap();
let decoded: rmpv::Value = rmp_serde::from_slice(&buf).unwrap();
let map = decoded.as_map().unwrap();
assert!(map.iter().any(|(k, v)| k.as_str() == Some("v") && v.as_u64() == Some(1)));
assert!(map
.iter()
.any(|(k, v)| k.as_str() == Some("valid_time") && v.as_str() == Some("2026-04-19T14:00:00Z")));
let cells_val = &map.iter().find(|(k, _)| k.as_str() == Some("cells")).unwrap().1;
let cells_arr = cells_val.as_array().unwrap();
assert_eq!(cells_arr.len(), 1);
}
#[test]
fn atomic_rename_replaces_existing() {
let dir = tempfile::tempdir().unwrap();
let vt = Utc.with_ymd_and_hms(2026, 4, 19, 14, 0, 0).unwrap();
let first = vec![CellEntry {
lat: 32.0, lon: -97.0,
profile: value_map([("v", rmpv::Value::Integer(1.into()))]),
}];
let second = vec![CellEntry {
lat: 32.0, lon: -97.0,
profile: value_map([("v", rmpv::Value::Integer(2.into()))]),
}];
let p1 = write_atomic(dir.path(), vt, &first).unwrap();
let p2 = write_atomic(dir.path(), vt, &second).unwrap();
assert_eq!(p1, p2);
// Only one file in the profiles dir — the tmp was renamed away.
let entries: Vec<_> = std::fs::read_dir(dir.path().join("profiles"))
.unwrap()
.map(|e| e.unwrap().file_name().into_string().unwrap())
.collect();
assert_eq!(entries.len(), 1);
}
#[test]
fn path_for_layout() {
let vt = Utc.with_ymd_and_hms(2026, 4, 19, 14, 0, 0).unwrap();
let p = path_for(std::path::Path::new("/data/scores"), vt);
assert_eq!(p.to_str().unwrap(), "/data/scores/profiles/2026-04-19T14:00:00Z.mp.gz");
}
#[test]
fn snap_coords_rounds_to_three_decimals() {
assert_eq!(snap_coords(32.8998, -97.04031), (32.900, -97.040));
}
}