mctx_core/
jsonl.rs

1#[cfg(feature = "metrics")]
2use fs2::FileExt;
3#[cfg(feature = "metrics")]
4use serde_json::{Map, Value, json};
5#[cfg(feature = "metrics")]
6use std::fs;
7#[cfg(feature = "metrics")]
8use std::fs::File;
9#[cfg(feature = "metrics")]
10use std::fs::OpenOptions;
11#[cfg(feature = "metrics")]
12use std::io::{BufRead, BufReader, Seek, SeekFrom, Write};
13#[cfg(feature = "metrics")]
14use std::path::{Path, PathBuf};
15#[cfg(feature = "metrics")]
16use std::time::{SystemTime, UNIX_EPOCH};
17
18/// Canonical Heimdall JSONL schema version used by metrics writers.
19#[cfg(feature = "metrics")]
20pub const HEIMDALL_JSONL_SCHEMA: &str = "heimdall-jsonl-v1";
21
22/// Canonical sender-side network artifact type.
23#[cfg(feature = "metrics")]
24pub const NETWORK_ARTIFACT_TYPE: &str = "mctx-network";
25
26/// Canonical process hardware artifact type.
27#[cfg(feature = "metrics")]
28pub const HARDWARE_ARTIFACT_TYPE: &str = "process-hardware";
29
30/// Common JSONL output configuration for one metrics file.
31#[cfg(feature = "metrics")]
32#[derive(Debug, Clone, PartialEq, Eq)]
33pub struct MetricsJsonlOutputConfig {
34    pub network_path: PathBuf,
35    pub node_id: String,
36    pub flags: Map<String, Value>,
37}
38
39/// Stateful JSONL writer that validates or writes the canonical header once,
40/// then appends compact sample rows without rescanning the whole file.
41#[cfg(feature = "metrics")]
42#[derive(Debug)]
43pub struct MetricsJsonlWriter {
44    file: File,
45}
46
47/// Infers a `node_id` from a metrics file path.
48///
49/// Resolution order:
50/// 1. parent directory name
51/// 2. file stem
52/// 3. `"unknown"`
53#[cfg(feature = "metrics")]
54pub fn infer_node_id_from_path(path: &Path) -> String {
55    path.parent()
56        .and_then(Path::file_name)
57        .and_then(|name| name.to_str())
58        .filter(|name| !name.is_empty())
59        .map(str::to_string)
60        .or_else(|| {
61            path.file_stem()
62                .and_then(|stem| stem.to_str())
63                .filter(|stem| !stem.is_empty())
64                .map(str::to_string)
65        })
66        .unwrap_or_else(|| "unknown".to_string())
67}
68
69/// Converts a system time to a Unix timestamp in fractional seconds.
70#[cfg(feature = "metrics")]
71pub fn unix_timestamp_secs(time: SystemTime) -> f64 {
72    time.duration_since(UNIX_EPOCH)
73        .map(|duration| duration.as_secs_f64())
74        .unwrap_or(0.0)
75}
76
77/// Builds the canonical header object for a Heimdall JSONL file.
78#[cfg(feature = "metrics")]
79pub fn header_json(
80    artifact_type: &'static str,
81    producer: &'static str,
82    node_id: &str,
83    created_at: SystemTime,
84    flags: &Map<String, Value>,
85) -> Value {
86    json!({
87        "schema": HEIMDALL_JSONL_SCHEMA,
88        "artifact_type": artifact_type,
89        "node_id": node_id,
90        "producer": producer,
91        "created_at": unix_timestamp_secs(created_at),
92        "flags": Value::Object(flags.clone()),
93    })
94}
95
96/// Returns the first non-empty line from a valid canonical JSONL file, if any.
97///
98/// Existing non-empty files without a valid single header and valid sample rows
99/// are rejected rather than exposed as headerless input.
100#[cfg(feature = "metrics")]
101pub fn first_non_empty_line(path: &Path) -> Result<Option<String>, std::io::Error> {
102    let file = match File::open(path) {
103        Ok(file) => file,
104        Err(err) if err.kind() == std::io::ErrorKind::NotFound => return Ok(None),
105        Err(err) => return Err(err),
106    };
107
108    validate_existing_header_reader(BufReader::new(file))
109        .map(|header| header.map(|header| header.line))
110}
111
112/// Validates the existing first non-empty JSONL line as a Heimdall header.
113#[cfg(feature = "metrics")]
114pub fn validate_existing_header(path: &Path) -> Result<Option<Value>, std::io::Error> {
115    let file = match File::open(path) {
116        Ok(file) => file,
117        Err(err) if err.kind() == std::io::ErrorKind::NotFound => return Ok(None),
118        Err(err) => return Err(err),
119    };
120
121    validate_existing_header_reader(BufReader::new(file))
122        .map(|header| header.map(|header| header.value))
123}
124
125#[cfg(feature = "metrics")]
126struct ValidatedJsonlHeader {
127    value: Value,
128    line: String,
129}
130
131#[cfg(feature = "metrics")]
132fn validate_existing_header_reader(
133    reader: impl BufRead,
134) -> Result<Option<ValidatedJsonlHeader>, std::io::Error> {
135    let mut non_empty_line_index = 0usize;
136    let mut parsed_header = None;
137
138    for line in reader.lines() {
139        let line = line?;
140        if line.trim().is_empty() {
141            continue;
142        }
143
144        non_empty_line_index += 1;
145        let parsed = serde_json::from_str::<Value>(&line).map_err(|err| {
146            let message = if non_empty_line_index == 1 {
147                format!("existing JSONL header is invalid JSON: {err}")
148            } else {
149                format!(
150                    "existing JSONL sample line {} is invalid JSON: {err}",
151                    non_empty_line_index
152                )
153            };
154            std::io::Error::new(std::io::ErrorKind::InvalidData, message)
155        })?;
156
157        if non_empty_line_index == 1 {
158            validate_header_object(&parsed).map_err(|message| {
159                std::io::Error::new(
160                    std::io::ErrorKind::InvalidData,
161                    format!("existing JSONL header is invalid: {message}"),
162                )
163            })?;
164
165            parsed_header = Some(ValidatedJsonlHeader {
166                value: parsed,
167                line,
168            });
169            continue;
170        }
171
172        if is_header_object(&parsed) {
173            return Err(std::io::Error::new(
174                std::io::ErrorKind::InvalidData,
175                "existing JSONL file contains more than one Heimdall header object",
176            ));
177        }
178
179        validate_sample_row(&parsed).map_err(|err| {
180            std::io::Error::new(
181                std::io::ErrorKind::InvalidData,
182                format!(
183                    "existing JSONL sample line {} is invalid: {}",
184                    non_empty_line_index, err
185                ),
186            )
187        })?;
188    }
189
190    Ok(parsed_header)
191}
192
193/// Ensures a JSONL file has exactly one header at the top before samples.
194#[cfg(feature = "metrics")]
195pub fn ensure_single_header(path: &Path, header: &Value) -> Result<(), std::io::Error> {
196    open_jsonl_append_file(path, header).map(|_| ())
197}
198
199#[cfg(feature = "metrics")]
200fn open_jsonl_append_file(path: &Path, header: &Value) -> Result<File, std::io::Error> {
201    validate_header_object(header).map_err(|message| {
202        std::io::Error::new(
203            std::io::ErrorKind::InvalidInput,
204            format!("requested JSONL header is invalid: {message}"),
205        )
206    })?;
207
208    if let Some(parent) = path.parent()
209        && !parent.as_os_str().is_empty()
210    {
211        fs::create_dir_all(parent)?;
212    }
213
214    let mut file = OpenOptions::new()
215        .create(true)
216        .read(true)
217        .append(true)
218        .open(path)?;
219    file.try_lock_exclusive().map_err(|error| {
220        std::io::Error::new(
221            error.kind(),
222            format!(
223                "failed to acquire exclusive JSONL writer lock for {}: {error}",
224                path.display()
225            ),
226        )
227    })?;
228
229    file.seek(SeekFrom::Start(0))?;
230    let existing =
231        validate_existing_header_reader(BufReader::new(&mut file))?.map(|header| header.value);
232    match existing {
233        Some(existing) => {
234            if !headers_are_compatible(&existing, header) {
235                return Err(std::io::Error::new(
236                    std::io::ErrorKind::InvalidData,
237                    "existing JSONL header does not match the requested schema metadata",
238                ));
239            }
240
241            Ok(file)
242        }
243        None => {
244            write_json_line(&mut file, header)?;
245            Ok(file)
246        }
247    }
248}
249
250#[cfg(feature = "metrics")]
251fn write_json_line(file: &mut File, value: &Value) -> Result<(), std::io::Error> {
252    let mut line = serde_json::to_vec(value).map_err(std::io::Error::other)?;
253    line.push(b'\n');
254    file.write_all(&line)
255}
256
257/// Appends one compact sample row to a JSONL file with a canonical header.
258#[cfg(feature = "metrics")]
259pub fn append_jsonl_sample_row(
260    path: &Path,
261    header: &Value,
262    sample: &Value,
263) -> Result<(), std::io::Error> {
264    validate_sample_row(sample)?;
265    let mut file = open_jsonl_append_file(path, header)?;
266    write_json_line(&mut file, sample)
267}
268
269#[cfg(feature = "metrics")]
270impl MetricsJsonlWriter {
271    /// Opens or creates a single-header JSONL file for repeated sample appends.
272    pub fn open(path: &Path, header: &Value) -> Result<Self, std::io::Error> {
273        Ok(Self {
274            file: open_jsonl_append_file(path, header)?,
275        })
276    }
277
278    /// Appends one compact sample row without rescanning the existing file.
279    pub fn append_sample_row(&mut self, sample: &Value) -> Result<(), std::io::Error> {
280        validate_sample_row(sample)?;
281        write_json_line(&mut self.file, sample)
282    }
283}
284
285#[cfg(feature = "metrics")]
286fn headers_are_compatible(existing: &Value, expected: &Value) -> bool {
287    let comparable_keys = ["schema", "artifact_type", "node_id", "producer", "flags"];
288    comparable_keys
289        .into_iter()
290        .all(|key| existing.get(key) == expected.get(key))
291}
292
293#[cfg(feature = "metrics")]
294fn is_header_object(value: &Value) -> bool {
295    validate_header_object(value).is_ok()
296}
297
298#[cfg(feature = "metrics")]
299fn validate_header_object(value: &Value) -> Result<(), String> {
300    let object = value
301        .as_object()
302        .ok_or_else(|| "header must be a JSON object".to_string())?;
303    if object.get("schema").and_then(Value::as_str) != Some(HEIMDALL_JSONL_SCHEMA) {
304        return Err(format!("schema must be `{HEIMDALL_JSONL_SCHEMA}`"));
305    }
306
307    for field in ["artifact_type", "node_id", "producer"] {
308        if object
309            .get(field)
310            .and_then(Value::as_str)
311            .is_none_or(|value| value.trim().is_empty())
312        {
313            return Err(format!("{field} must be a non-empty string"));
314        }
315    }
316
317    if !object
318        .get("created_at")
319        .and_then(Value::as_f64)
320        .is_some_and(f64::is_finite)
321    {
322        return Err("created_at must be a finite JSON number".to_string());
323    }
324    if !matches!(object.get("flags"), Some(Value::Object(_))) {
325        return Err("flags must be a JSON object".to_string());
326    }
327
328    Ok(())
329}
330
331#[cfg(feature = "metrics")]
332fn validate_sample_row(sample: &Value) -> Result<(), std::io::Error> {
333    let Some(sample_object) = sample.as_object() else {
334        return Err(std::io::Error::new(
335            std::io::ErrorKind::InvalidInput,
336            "JSONL sample rows must be JSON objects",
337        ));
338    };
339
340    for reserved_key in ["schema", "artifact_type", "node_id", "producer", "flags"] {
341        if sample_object.contains_key(reserved_key) {
342            return Err(std::io::Error::new(
343                std::io::ErrorKind::InvalidInput,
344                format!(
345                    "JSONL sample rows must not contain reserved header field `{reserved_key}`"
346                ),
347            ));
348        }
349    }
350
351    for required_number in ["ts", "interval_secs"] {
352        if !sample_object
353            .get(required_number)
354            .and_then(Value::as_f64)
355            .is_some_and(f64::is_finite)
356        {
357            return Err(std::io::Error::new(
358                std::io::ErrorKind::InvalidInput,
359                format!("JSONL sample field `{required_number}` must be a finite number"),
360            ));
361        }
362    }
363
364    if sample_object["interval_secs"].as_f64().unwrap_or(-1.0) < 0.0 {
365        return Err(std::io::Error::new(
366            std::io::ErrorKind::InvalidInput,
367            "JSONL sample field `interval_secs` must not be negative",
368        ));
369    }
370
371    Ok(())
372}
373
374#[cfg(all(test, feature = "metrics"))]
375mod tests {
376    use super::*;
377    use std::fs;
378    use std::time::{Duration, SystemTime};
379
380    #[test]
381    fn node_id_inference_prefers_parent_dir_over_file_stem() {
382        let path = PathBuf::from("/tmp/sender-0001/network-metrics.jsonl");
383        assert_eq!(infer_node_id_from_path(&path), "sender-0001");
384    }
385
386    #[test]
387    fn writes_single_header_then_compact_samples() {
388        let parent_name = format!(
389            "mctx_jsonl_header_{}",
390            SystemTime::now()
391                .duration_since(UNIX_EPOCH)
392                .unwrap_or(Duration::ZERO)
393                .as_nanos()
394        );
395        let parent = std::env::temp_dir().join(&parent_name);
396        fs::create_dir_all(&parent).unwrap();
397        let path = parent.join("network.jsonl");
398
399        let mut flags = Map::new();
400        flags.insert("role".to_string(), Value::String("sender".to_string()));
401        let header = header_json(
402            NETWORK_ARTIFACT_TYPE,
403            "mctx-core/test",
404            &infer_node_id_from_path(&path),
405            SystemTime::UNIX_EPOCH + Duration::from_secs(10),
406            &flags,
407        );
408        let sample1 = json!({"ts": 11.0, "interval_secs": 1.0, "packets_sent_total": 5});
409        let sample2 = json!({"ts": 12.0, "interval_secs": 1.0, "packets_sent_total": 10});
410
411        append_jsonl_sample_row(&path, &header, &sample1).unwrap();
412        append_jsonl_sample_row(&path, &header, &sample2).unwrap();
413
414        let contents = fs::read_to_string(&path).unwrap();
415        let lines = contents
416            .lines()
417            .filter(|line| !line.trim().is_empty())
418            .collect::<Vec<_>>();
419
420        assert_eq!(lines.len(), 3);
421        assert_eq!(
422            first_non_empty_line(&path).unwrap().as_deref(),
423            Some(lines[0])
424        );
425
426        let parsed_header: Value = serde_json::from_str(lines[0]).unwrap();
427        assert_eq!(parsed_header["schema"], HEIMDALL_JSONL_SCHEMA);
428        assert_eq!(parsed_header["artifact_type"], NETWORK_ARTIFACT_TYPE);
429        assert_eq!(parsed_header["node_id"], parent_name);
430        assert!(parsed_header["flags"].is_object());
431
432        for sample_line in &lines[1..] {
433            let sample: Value = serde_json::from_str(sample_line).unwrap();
434            assert!(sample.get("schema").is_none());
435            assert!(sample.get("artifact_type").is_none());
436            assert!(sample.get("node_id").is_none());
437            assert!(sample.get("producer").is_none());
438            assert!(sample.get("flags").is_none());
439        }
440
441        let _ = fs::remove_file(path);
442        let _ = fs::remove_dir(parent);
443    }
444
445    #[test]
446    fn invalid_first_line_header_is_rejected() {
447        let path = std::env::temp_dir().join(format!(
448            "mctx_invalid_header_{}.jsonl",
449            SystemTime::now()
450                .duration_since(UNIX_EPOCH)
451                .unwrap_or(Duration::ZERO)
452                .as_nanos()
453        ));
454        fs::write(&path, "{\"ts\":1.0,\"interval_secs\":1.0}\n").unwrap();
455
456        let err = validate_existing_header(&path).unwrap_err();
457
458        assert_eq!(err.kind(), std::io::ErrorKind::InvalidData);
459        assert_eq!(
460            first_non_empty_line(&path).unwrap_err().kind(),
461            std::io::ErrorKind::InvalidData
462        );
463
464        let _ = fs::remove_file(path);
465    }
466
467    #[test]
468    fn existing_header_without_created_at_is_rejected() {
469        let path = std::env::temp_dir().join(format!(
470            "mctx_missing_created_at_{}.jsonl",
471            SystemTime::now()
472                .duration_since(UNIX_EPOCH)
473                .unwrap_or(Duration::ZERO)
474                .as_nanos()
475        ));
476        fs::write(
477            &path,
478            r#"{"schema":"heimdall-jsonl-v1","artifact_type":"mctx-network","node_id":"sender-a","producer":"mctx-core/test","flags":{}}
479"#,
480        )
481        .unwrap();
482
483        let err = validate_existing_header(&path).unwrap_err();
484
485        assert_eq!(err.kind(), std::io::ErrorKind::InvalidData);
486        let _ = fs::remove_file(path);
487    }
488
489    #[test]
490    fn invalid_requested_header_is_rejected_before_file_creation() {
491        let path = std::env::temp_dir().join(format!(
492            "mctx_invalid_requested_header_{}.jsonl",
493            SystemTime::now()
494                .duration_since(UNIX_EPOCH)
495                .unwrap_or(Duration::ZERO)
496                .as_nanos()
497        ));
498
499        let err = MetricsJsonlWriter::open(&path, &json!({"not": "a header"})).unwrap_err();
500
501        assert_eq!(err.kind(), std::io::ErrorKind::InvalidInput);
502        assert!(!path.exists());
503    }
504
505    #[test]
506    fn mismatched_existing_header_is_rejected() {
507        let path = std::env::temp_dir().join(format!(
508            "mctx_mismatched_header_{}.jsonl",
509            SystemTime::now()
510                .duration_since(UNIX_EPOCH)
511                .unwrap_or(Duration::ZERO)
512                .as_nanos()
513        ));
514        let header1 = json!({
515            "schema": HEIMDALL_JSONL_SCHEMA,
516            "artifact_type": NETWORK_ARTIFACT_TYPE,
517            "node_id": "sender-a",
518            "producer": "mctx-core/test",
519            "created_at": 1.0,
520            "flags": {"role": "sender"},
521        });
522        let header2 = json!({
523            "schema": HEIMDALL_JSONL_SCHEMA,
524            "artifact_type": NETWORK_ARTIFACT_TYPE,
525            "node_id": "sender-b",
526            "producer": "mctx-core/test",
527            "created_at": 2.0,
528            "flags": {"role": "sender"},
529        });
530
531        append_jsonl_sample_row(&path, &header1, &json!({"ts": 1.0, "interval_secs": 1.0}))
532            .unwrap();
533        let err =
534            append_jsonl_sample_row(&path, &header2, &json!({"ts": 2.0, "interval_secs": 1.0}))
535                .unwrap_err();
536
537        assert_eq!(err.kind(), std::io::ErrorKind::InvalidData);
538
539        let _ = fs::remove_file(path);
540    }
541
542    #[test]
543    fn additional_header_line_is_rejected() {
544        let path = std::env::temp_dir().join(format!(
545            "mctx_extra_header_{}.jsonl",
546            SystemTime::now()
547                .duration_since(UNIX_EPOCH)
548                .unwrap_or(Duration::ZERO)
549                .as_nanos()
550        ));
551        let header = json!({
552            "schema": HEIMDALL_JSONL_SCHEMA,
553            "artifact_type": NETWORK_ARTIFACT_TYPE,
554            "node_id": "sender-a",
555            "producer": "mctx-core/test",
556            "created_at": 1.0,
557            "flags": {"role": "sender"},
558        });
559        let contents = format!(
560            "{}\n{}\n{}\n",
561            serde_json::to_string(&header).unwrap(),
562            serde_json::to_string(&json!({"ts": 1.0, "interval_secs": 1.0})).unwrap(),
563            serde_json::to_string(&header).unwrap(),
564        );
565        fs::write(&path, contents).unwrap();
566
567        let err = validate_existing_header(&path).unwrap_err();
568
569        assert_eq!(err.kind(), std::io::ErrorKind::InvalidData);
570
571        let _ = fs::remove_file(path);
572    }
573
574    #[test]
575    fn sample_rows_with_reserved_header_fields_are_rejected() {
576        let path = std::env::temp_dir().join(format!(
577            "mctx_reserved_sample_fields_{}.jsonl",
578            SystemTime::now()
579                .duration_since(UNIX_EPOCH)
580                .unwrap_or(Duration::ZERO)
581                .as_nanos()
582        ));
583        let header = json!({
584            "schema": HEIMDALL_JSONL_SCHEMA,
585            "artifact_type": NETWORK_ARTIFACT_TYPE,
586            "node_id": "sender-a",
587            "producer": "mctx-core/test",
588            "created_at": 1.0,
589            "flags": {"role": "sender"},
590        });
591        let err = append_jsonl_sample_row(
592            &path,
593            &header,
594            &json!({"ts": 1.0, "interval_secs": 1.0, "schema": HEIMDALL_JSONL_SCHEMA}),
595        )
596        .unwrap_err();
597
598        assert_eq!(err.kind(), std::io::ErrorKind::InvalidInput);
599
600        let _ = fs::remove_file(path);
601    }
602
603    #[test]
604    fn existing_sample_rows_with_reserved_header_fields_are_rejected() {
605        let path = std::env::temp_dir().join(format!(
606            "mctx_reserved_existing_sample_fields_{}.jsonl",
607            SystemTime::now()
608                .duration_since(UNIX_EPOCH)
609                .unwrap_or(Duration::ZERO)
610                .as_nanos()
611        ));
612        let header = json!({
613            "schema": HEIMDALL_JSONL_SCHEMA,
614            "artifact_type": NETWORK_ARTIFACT_TYPE,
615            "node_id": "sender-a",
616            "producer": "mctx-core/test",
617            "created_at": 1.0,
618            "flags": {"role": "sender"},
619        });
620        let contents = format!(
621            "{}\n{}\n",
622            serde_json::to_string(&header).unwrap(),
623            serde_json::to_string(
624                &json!({"ts": 1.0, "interval_secs": 1.0, "schema": HEIMDALL_JSONL_SCHEMA})
625            )
626            .unwrap(),
627        );
628        fs::write(&path, contents).unwrap();
629
630        let err = validate_existing_header(&path).unwrap_err();
631
632        assert_eq!(err.kind(), std::io::ErrorKind::InvalidData);
633
634        let _ = fs::remove_file(path);
635    }
636
637    #[test]
638    fn stateful_writer_appends_without_repeating_the_header() {
639        let path = std::env::temp_dir().join(format!(
640            "mctx_stateful_writer_{}.jsonl",
641            SystemTime::now()
642                .duration_since(UNIX_EPOCH)
643                .unwrap_or(Duration::ZERO)
644                .as_nanos()
645        ));
646        let header = json!({
647            "schema": HEIMDALL_JSONL_SCHEMA,
648            "artifact_type": NETWORK_ARTIFACT_TYPE,
649            "node_id": "sender-a",
650            "producer": "mctx-core/test",
651            "created_at": 1.0,
652            "flags": {"role": "sender"},
653        });
654
655        let mut writer = MetricsJsonlWriter::open(&path, &header).unwrap();
656        writer
657            .append_sample_row(&json!({"ts": 1.0, "interval_secs": 1.0}))
658            .unwrap();
659        writer
660            .append_sample_row(&json!({"ts": 2.0, "interval_secs": 1.0}))
661            .unwrap();
662        drop(writer);
663
664        let contents = fs::read_to_string(&path).unwrap();
665        let lines = contents
666            .lines()
667            .filter(|line| !line.trim().is_empty())
668            .collect::<Vec<_>>();
669        assert_eq!(lines.len(), 3);
670        assert!(is_header_object(&serde_json::from_str(lines[0]).unwrap()));
671        assert!(!is_header_object(&serde_json::from_str(lines[1]).unwrap()));
672        assert!(!is_header_object(&serde_json::from_str(lines[2]).unwrap()));
673
674        let _ = fs::remove_file(path);
675    }
676
677    #[test]
678    fn second_stateful_writer_cannot_race_the_active_writer() {
679        let path = std::env::temp_dir().join(format!(
680            "mctx_locked_writer_{}.jsonl",
681            SystemTime::now()
682                .duration_since(UNIX_EPOCH)
683                .unwrap_or(Duration::ZERO)
684                .as_nanos()
685        ));
686        let header = json!({
687            "schema": HEIMDALL_JSONL_SCHEMA,
688            "artifact_type": NETWORK_ARTIFACT_TYPE,
689            "node_id": "sender-a",
690            "producer": "mctx-core/test",
691            "created_at": 1.0,
692            "flags": {"role": "sender"},
693        });
694        let writer = MetricsJsonlWriter::open(&path, &header).unwrap();
695
696        // Lock-contention errors map to different ErrorKinds across platforms.
697        let _err = MetricsJsonlWriter::open(&path, &header).unwrap_err();
698        drop(writer);
699        MetricsJsonlWriter::open(&path, &header).unwrap();
700        let _ = fs::remove_file(path);
701    }
702}
mctx_core/jsonl.rs

mctx_core/
jsonl.rs
