henad-explore 0.3.0

Parameter sweeps, searches and result folders for Henad models.
Documentation
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
//! Output directories of a sweep, and the writer that streams each finished run into a directory.
//!
//! A directory holds four files. `runs.csv` has one row per run, `series.csv` the sampled stat rows of every run,
//! `summary.csv` statistics over the replicates of each config, and `manifest.json` the settings, provenance and
//! progress of the sweep. Both CSV tables list runs in order of their ids once a sweep ends. A search adds the
//! tables [`search_tables`] writes.

pub mod details;
pub mod manifest;
pub mod memory;
pub mod read;
pub mod resume;
pub mod runs_csv;
pub mod search_tables;
pub mod series_csv;
pub mod summary_csv;

use std::fmt;
use std::fs::{File, OpenOptions, TryLockError};
use std::io::{self, BufReader, BufWriter, Write};
use std::path::{Path, PathBuf};
use std::sync::Arc;

use henad_core::explore::measure::MeasurePlan;
use henad_core::explore::outcome::RunOutcome;
use henad_core::explore::plan::{Config, Plan};
use henad_core::params::ParamDescriptor;

use crate::exec::RunSink;
use crate::output::manifest::{Manifest, ResultCounts};
use crate::output::read::{ReadError, RunsCsv, SeriesScan, SeriesSegment, merge_series};
use crate::output::runs_csv::RunsWriter;
use crate::output::series_csv::SeriesWriter;
use crate::output::summary_csv::{SummaryError, write_summary};

/// Name of the table with one row per run.
pub const RUNS_FILE: &str = "runs.csv";
/// Name of the table of sampled stat rows.
pub const SERIES_FILE: &str = "series.csv";
/// Name of the table of statistics over the replicates of each config.
pub const SUMMARY_FILE: &str = "summary.csv";
/// Name of the file holding the settings, provenance and progress of a sweep.
pub const MANIFEST_FILE: &str = "manifest.json";
/// Name of a search's table with one row per evaluation.
pub const EVALUATIONS_FILE: &str = "evaluations.csv";
/// Name of a search's table with one row per batch.
pub const BATCHES_FILE: &str = "batches.csv";
/// Name of the table ranking every candidate of a search that scores an objective.
pub const BEST_FILE: &str = "best.csv";
/// Name of the table of cells that a Pattern Space Exploration fills.
pub const ARCHIVE_FILE: &str = "archive.csv";
/// Name of a genetic algorithm's table with one row per finished generation.
pub const GENERATIONS_FILE: &str = "generations.csv";

/// Files a sweep or search writes, any one of which marks a directory as holding results.
const RESULT_FILES: [&str; 9] = [
    RUNS_FILE,
    SERIES_FILE,
    SUMMARY_FILE,
    MANIFEST_FILE,
    EVALUATIONS_FILE,
    BATCHES_FILE,
    BEST_FILE,
    ARCHIVE_FILE,
    GENERATIONS_FILE,
];

/// File a writer locks while it writes to an output directory.
pub const LOCK_FILE: &str = ".lock";

/// Suffix of the file that holds a table's full replacement until it is renamed into place.
const STAGED_SUFFIX: &str = ".staged";

/// File whose presence says both staged tables are complete.
const STAGED_MARKER: &str = "tables.staged";

/// Returns the path of the staged table that replaces `file` in `dir`.
fn staged_path(dir: &Path, file: &str) -> PathBuf {
    dir.join(format!("{file}{STAGED_SUFFIX}"))
}

/// Returns the paths of `runs.csv` and `series.csv` in the directory at `path`, as they stand.
///
/// A staged table is used instead of its original while a replacement waits to be finished.
pub(crate) fn table_paths(path: &Path) -> (PathBuf, PathBuf) {
    let complete = path.join(STAGED_MARKER).exists();
    let current = |file: &str| {
        let staged = staged_path(path, file);
        if complete && staged.exists() {
            staged
        } else {
            path.join(file)
        }
    };
    (current(RUNS_FILE), current(SERIES_FILE))
}

/// A directory that holds, or is about to hold, the results of one sweep, locked against other writers.
///
/// The lock is the operating system's advisory lock on the file [`LOCK_FILE`], released when the value is dropped. The
/// file stays in the directory, and the next writer locks it again. The lock is released when the process that holds it
/// ends, even when the process is killed.
///
/// A writer creates afresh each file it replaces. It appends to `runs.csv` and `series.csv`, and cuts them, only after
/// [`Self::open`] has checked that neither table is a symbolic link.
#[derive(Debug)]
pub struct OutputDir {
    path: PathBuf,
    #[expect(dead_code, reason = "held for its drop, which releases the lock")]
    lock: DirLock,
}

impl OutputDir {
    /// Checks that the directory at `path` holds no results, without creating it.
    ///
    /// A symbolic link under the name of a file counts as that file, even a link that points nowhere.
    ///
    /// # Errors
    ///
    /// Returns [`OutputError::HoldsResults`] when the directory holds a file that a sweep or a search writes,
    /// including a staged table or the marker of staged tables.
    pub fn check_free(path: &Path) -> Result<(), OutputError> {
        let holds_results = |file: String| {
            Err(OutputError::HoldsResults {
                dir: path.to_owned(),
                file,
            })
        };
        if let Some(&file) = RESULT_FILES
            .iter()
            .find(|&&file| std::fs::symlink_metadata(path.join(file)).is_ok())
        {
            return holds_results(file.to_owned());
        }
        // `table_paths` can read a staged table instead of its original, so a staged table counts as results.
        let staged = std::fs::read_dir(path)
            .into_iter()
            .flatten()
            .flatten()
            .map(|entry| entry.file_name().to_string_lossy().into_owned())
            .find(|name| name.ends_with(STAGED_SUFFIX));
        match staged {
            Some(file) => holds_results(file),
            None => Ok(()),
        }
    }

    /// Creates the directory at `path` and its parents, or uses an existing directory that holds no results, and
    /// locks it.
    ///
    /// # Errors
    ///
    /// Returns [`OutputError::HoldsResults`] when the directory holds a file that a sweep or a search writes,
    /// [`OutputError::Locked`] when another writer holds its lock, and [`OutputError::Write`] when it cannot be
    /// created.
    pub fn create(path: &Path) -> Result<Self, OutputError> {
        Self::check_free(path)?;
        std::fs::create_dir_all(path).map_err(write_error_at(path))?;
        let dir = Self {
            path: path.to_owned(),
            lock: DirLock::acquire(path)?,
        };
        // Checked again under the lock. Another writer could have finished between the first check and the lock.
        Self::check_free(path)?;
        Ok(dir)
    }

    /// Returns whether the directory at `path` holds a file that a sweep or a search writes.
    pub fn holds_results(path: &Path) -> bool {
        Self::check_free(path).is_err()
    }

    /// Locks the existing directory at `path`, finishing a table replacement that a process left behind.
    ///
    /// # Errors
    ///
    /// Returns [`OutputError::Locked`] when another writer holds the directory's lock, [`OutputError::Link`] when
    /// `runs.csv` or `series.csv` is a symbolic link, and [`OutputError::Write`] when the replacement cannot be
    /// finished or discarded.
    pub fn open(path: &Path) -> Result<Self, OutputError> {
        let dir = Self {
            path: path.to_owned(),
            lock: DirLock::acquire(path)?,
        };
        dir.finish_staged()?;
        // A resume appends to both tables and cuts them, and both operations would follow a link.
        for file in [RUNS_FILE, SERIES_FILE] {
            let table = path.join(file);
            if std::fs::symlink_metadata(&table).is_ok_and(|metadata| metadata.file_type().is_symlink()) {
                return Err(OutputError::Link { path: table });
            }
        }
        Ok(dir)
    }

    /// Checks that no writer holds the lock of the directory at `path`, without taking it.
    ///
    /// # Errors
    ///
    /// Returns [`OutputError::Locked`] when another writer holds the lock.
    pub fn check_unlocked(path: &Path) -> Result<(), OutputError> {
        // A directory or a lock file that is missing is unlocked. The check never creates the directory or the file.
        let Ok(file) = File::open(path.join(LOCK_FILE)) else {
            return Ok(());
        };
        match file.try_lock_shared() {
            Err(TryLockError::WouldBlock) => Err(OutputError::Locked { dir: path.to_owned() }),
            Ok(()) | Err(TryLockError::Error(_)) => Ok(()),
        }
    }

    /// Path of the directory.
    pub fn path(&self) -> &Path {
        &self.path
    }

    /// Creates `runs.csv` and `series.csv`, and returns a writer for the runs of `plan`.
    ///
    /// `params` are the model's parameters, and `measure` fixes the stat and reducer columns.
    ///
    /// # Errors
    ///
    /// Returns [`OutputError::Write`] when a file exists already or cannot be created, or its header cannot be written.
    pub fn open_writer(
        &self,
        plan: &Arc<Plan>,
        params: &[ParamDescriptor],
        measure: &MeasurePlan,
    ) -> Result<OutputWriter<BufWriter<File>>, OutputError> {
        let runs_path = self.path.join(RUNS_FILE);
        let runs = File::create_new(&runs_path).map_err(write_error_at(&runs_path))?;
        let runs = RunsWriter::new(BufWriter::new(runs), params, plan.actions(), measure.reducers().names())
            .map_err(write_error_at(&runs_path))?;
        let series_path = self.path.join(SERIES_FILE);
        let series = File::create_new(&series_path).map_err(write_error_at(&series_path))?;
        let series =
            SeriesWriter::new(BufWriter::new(series), measure.columns()).map_err(write_error_at(&series_path))?;
        Ok(OutputWriter::new(Arc::clone(plan), runs, series))
    }

    /// Opens `runs.csv` and `series.csv`, whose headers are already written, and returns a writer that adds the runs
    /// of `plan` to them.
    ///
    /// # Errors
    ///
    /// Returns [`OutputError::Write`] when a file cannot be opened.
    pub fn append_writer(
        &self,
        plan: &Arc<Plan>,
        params: &[ParamDescriptor],
    ) -> Result<OutputWriter<BufWriter<File>>, OutputError> {
        let append = |file: &str| {
            let path = self.path.join(file);
            OpenOptions::new()
                .append(true)
                .open(&path)
                .map(BufWriter::new)
                .map_err(write_error_at(&path))
        };
        let runs = RunsWriter::appending(append(RUNS_FILE)?, params);
        let series = SeriesWriter::appending(append(SERIES_FILE)?);
        Ok(OutputWriter::new(Arc::clone(plan), runs, series))
    }

    /// Replaces `runs.csv` and `series.csv` with the text that `write_runs` and `write_series` write.
    ///
    /// Each table is written in full beside the original with a `.staged` suffix. A marker file then records that both
    /// tables are complete, both tables are renamed into place, and the marker is removed. A process that ends between
    /// the marker and its removal leaves the replacement for [`Self::open`] to finish.
    ///
    /// # Errors
    ///
    /// Returns [`OutputError::Write`] when a write, a rename or a removal fails. The originals stay in place until
    /// both new tables are complete.
    pub fn replace_tables(
        &self,
        write_runs: impl FnOnce(&mut dyn Write) -> io::Result<()>,
        write_series: impl FnOnce(&mut dyn Write) -> io::Result<()>,
    ) -> Result<(), OutputError> {
        self.write_staged(RUNS_FILE, write_runs)?;
        self.write_staged(SERIES_FILE, write_series)?;
        let marker = self.path.join(STAGED_MARKER);
        create_fresh(&marker)
            .and_then(|file| file.sync_all())
            .map_err(write_error_at(&marker))?;
        self.finish_staged()
    }

    /// Calls `write` to write the staged table that replaces `file`, and flushes the table to the disk.
    fn write_staged(
        &self,
        file: &str,
        write: impl FnOnce(&mut dyn Write) -> io::Result<()>,
    ) -> Result<(), OutputError> {
        let path = staged_path(&self.path, file);
        let staged = create_fresh(&path).map_err(write_error_at(&path))?;
        let mut writer = BufWriter::new(staged);
        write(&mut writer).map_err(write_error_at(&path))?;
        let staged = writer
            .into_inner()
            .map_err(|error| write_error_at(&path)(error.into_error()))?;
        staged.sync_all().map_err(write_error_at(&path))
    }

    /// Moves both staged tables into place when the marker says they are complete, and removes staged tables left
    /// incomplete otherwise.
    fn finish_staged(&self) -> Result<(), OutputError> {
        let marker = self.path.join(STAGED_MARKER);
        let complete = marker.exists();
        for file in [RUNS_FILE, SERIES_FILE] {
            let staged = staged_path(&self.path, file);
            // A link that points nowhere is a staged table to remove as well.
            if std::fs::symlink_metadata(&staged).is_err() {
                continue;
            }
            if complete {
                let installed = self.path.join(file);
                std::fs::rename(&staged, &installed).map_err(write_error_at(&installed))?;
            } else {
                std::fs::remove_file(&staged).map_err(write_error_at(&staged))?;
            }
        }
        if complete {
            std::fs::remove_file(&marker).map_err(write_error_at(&marker))?;
        }
        Ok(())
    }

    /// Rewrites `runs.csv` and `series.csv` with their runs in order of their ids, leaving tables already in order
    /// alone.
    ///
    /// # Errors
    ///
    /// Returns [`OutputError::Table`] when a table cannot be read back, and [`OutputError::Write`] when the
    /// replacement fails.
    pub fn order_tables(&self) -> Result<(), OutputError> {
        let runs_path = self.path.join(RUNS_FILE);
        let series_path = self.path.join(SERIES_FILE);
        let runs = RunsCsv::read(&runs_path).map_err(OutputError::Table)?;
        let series = SeriesScan::read(&series_path, |_| true).map_err(OutputError::Table)?;
        let in_order = runs.records.is_sorted_by_key(|record| record.run_id) && series.segments.len() <= 1;
        if in_order {
            return Ok(());
        }
        let (Some(runs_header), Some(series_header)) = (runs.header.as_ref(), series.header.as_ref()) else {
            return Ok(());
        };
        let mut records: Vec<_> = runs.records.iter().collect();
        records.sort_by_key(|record| record.run_id);
        let segments: Vec<SeriesSegment> = series
            .segments
            .iter()
            .map(|range| SeriesSegment {
                path: series_path.clone(),
                range: range.clone(),
                input_index: 0,
            })
            .collect();
        let runs_header_line = runs_csv::header_line(runs_header);
        self.replace_tables(
            |dest| {
                dest.write_all(runs_header_line.as_bytes())?;
                records
                    .iter()
                    .try_for_each(|record| dest.write_all(record.text.as_bytes()))
            },
            |dest| merge_series(dest, series_header, &segments, |_, run_id| Some(run_id)),
        )
    }

    /// Creates the table `file` afresh, removing any existing file of that name, and returns a buffered writer over it.
    ///
    /// # Errors
    ///
    /// Returns [`OutputError::Write`] when the file cannot be removed or created.
    pub fn create_table(&self, file: &str) -> Result<BufWriter<File>, OutputError> {
        let path = self.path.join(file);
        create_fresh(&path).map(BufWriter::new).map_err(write_error_at(&path))
    }

    /// Writes `manifest` to `manifest.json`. An earlier manifest is replaced by a single rename, so a reader sees
    /// either the old or the new manifest whole.
    ///
    /// # Errors
    ///
    /// Returns [`OutputError::Manifest`] when the manifest cannot be serialized, and [`OutputError::Write`] when a
    /// write or the rename fails.
    pub fn write_manifest(&self, manifest: &Manifest) -> Result<(), OutputError> {
        let text = manifest_text(manifest)?;
        let partial = self.path.join(format!("{MANIFEST_FILE}.partial"));
        // The file closes at the end of the block, before the rename.
        {
            let mut file = create_fresh(&partial).map_err(write_error_at(&partial))?;
            file.write_all(text.as_bytes()).map_err(write_error_at(&partial))?;
            file.sync_all().map_err(write_error_at(&partial))?;
        }
        let manifest_path = self.path.join(MANIFEST_FILE);
        std::fs::rename(&partial, &manifest_path).map_err(write_error_at(&manifest_path))
    }

    /// Rebuilds `summary.csv` from `runs.csv`.
    ///
    /// # Errors
    ///
    /// Returns [`OutputError::Read`] when `runs.csv` cannot be read, [`OutputError::Write`] when `summary.csv`
    /// cannot be written, and [`OutputError::Summary`] when `runs.csv` cannot be summarized.
    pub fn write_summary(&self) -> Result<(), OutputError> {
        let runs_path = self.path.join(RUNS_FILE);
        let read_error = |source| OutputError::Read {
            path: runs_path.clone(),
            source,
        };
        let runs = File::open(&runs_path).map_err(read_error)?;
        let summary_path = self.path.join(SUMMARY_FILE);
        let summary = create_fresh(&summary_path).map_err(write_error_at(&summary_path))?;
        match write_summary(BufReader::new(runs), BufWriter::new(summary)) {
            Ok(_) => Ok(()),
            Err(SummaryError::Read(source)) => Err(read_error(source)),
            Err(SummaryError::Io(source)) => Err(write_error_at(&summary_path)(source)),
            Err(error) => Err(OutputError::Summary(error)),
        }
    }
}

/// Returns `manifest` as the text of `manifest.json`, pretty-printed JSON ending in a line feed.
///
/// # Errors
///
/// Returns [`OutputError::Manifest`] when the manifest cannot be serialized.
pub(crate) fn manifest_text(manifest: &Manifest) -> Result<String, OutputError> {
    let mut text = serde_json::to_string_pretty(manifest).map_err(OutputError::Manifest)?;
    text.push('\n');
    Ok(text)
}

/// Returns a function that turns an I/O error on `path` into an [`OutputError::Write`].
fn write_error_at(path: &Path) -> impl FnOnce(io::Error) -> OutputError + use<> {
    let path = path.to_owned();
    move |source| OutputError::Write { path, source }
}

/// Creates the file at `path` afresh. A file or a symbolic link already there is removed first, never followed.
///
/// # Errors
///
/// Returns the error of the removal or the creation.
fn create_fresh(path: &Path) -> io::Result<File> {
    match std::fs::remove_file(path) {
        Err(error) if error.kind() != io::ErrorKind::NotFound => return Err(error),
        _ => {}
    }
    File::create_new(path)
}

/// Advisory lock a writer holds on [`LOCK_FILE`] of an output directory, released when dropped.
///
/// Note that the file stays in the directory. If it were removed, a writer that had it open could lock the removed file
/// while another writer locks a new lock file.
#[derive(Debug)]
struct DirLock {
    /// File the lock is held on.
    file: File,
}

impl DirLock {
    /// Locks the directory at `dir`, creating its lock file when missing.
    ///
    /// Note that a filesystem without file locks, such as some network filesystems, leaves the directory unguarded.
    ///
    /// # Errors
    ///
    /// Returns [`OutputError::Locked`] when another writer holds the lock, and [`OutputError::Write`] when the lock
    /// file cannot be opened.
    fn acquire(dir: &Path) -> Result<Self, OutputError> {
        let path = dir.join(LOCK_FILE);
        if std::fs::symlink_metadata(&path).is_ok_and(|metadata| metadata.file_type().is_symlink()) {
            std::fs::remove_file(&path).map_err(write_error_at(&path))?;
        }
        let file = OpenOptions::new()
            .read(true)
            .write(true)
            .create(true)
            .truncate(false)
            .open(&path)
            .map_err(write_error_at(&path))?;
        match file.try_lock() {
            Err(TryLockError::WouldBlock) => Err(OutputError::Locked { dir: dir.to_owned() }),
            Ok(()) | Err(TryLockError::Error(_)) => Ok(Self { file }),
        }
    }
}

impl Drop for DirLock {
    /// Releases the lock before the file closes.
    fn drop(&mut self) {
        drop(self.file.unlock());
    }
}

/// Sink that streams each committed run to `runs.csv` and `series.csv`, and counts the rows by status.
///
/// A run's series is written and flushed before its row in `runs.csv`, so every run in `runs.csv` has its whole
/// series written.
#[derive(Debug)]
pub struct OutputWriter<W: Write> {
    plan: Arc<Plan>,
    runs: RunsWriter<W>,
    series: SeriesWriter<W>,
    counts: ResultCounts,
}

impl<W: Write> OutputWriter<W> {
    /// Returns a writer for the runs of `plan` over writers whose headers are written.
    pub fn new(plan: Arc<Plan>, runs: RunsWriter<W>, series: SeriesWriter<W>) -> Self {
        Self {
            plan,
            runs,
            series,
            counts: ResultCounts::default(),
        }
    }

    /// Writes the series of `outcome`, then its row, and flushes after each write.
    ///
    /// # Errors
    ///
    /// Returns the error of a write or a flush.
    ///
    /// # Panics
    ///
    /// Panics when the run's config is not in the plan.
    pub fn write_run(&mut self, outcome: &RunOutcome) -> io::Result<()> {
        let Self {
            plan,
            runs,
            series,
            counts,
        } = self;
        let config = plan
            .config(outcome.run.config_id)
            .expect("a committed run's config is in its plan");
        write_both(runs, series, counts, outcome, config)
    }

    /// Writes the series of `outcome`, a run of `config`, then its row, and flushes after each write.
    ///
    /// A search writes its runs this way. Its configs are in no plan.
    ///
    /// # Errors
    ///
    /// Returns the error of a write or a flush.
    pub fn write_config_run(&mut self, outcome: &RunOutcome, config: &Config) -> io::Result<()> {
        write_both(&mut self.runs, &mut self.series, &mut self.counts, outcome, config)
    }

    /// Counts of the rows written so far.
    pub fn counts(&self) -> ResultCounts {
        self.counts
    }

    /// Flushes both writers, and returns them as the writers of `runs.csv` and `series.csv`.
    ///
    /// # Errors
    ///
    /// Returns the error of a flush.
    pub fn finish(self) -> io::Result<(W, W)> {
        let series = self.series.into_inner()?;
        let runs = self.runs.into_inner()?;
        Ok((runs, series))
    }
}

/// Writes the series of `outcome`, a run of `config`, to `series`, then its row to `runs`, flushing after each write,
/// and counts the row in `counts`.
fn write_both<W: Write>(
    runs: &mut RunsWriter<W>,
    series: &mut SeriesWriter<W>,
    counts: &mut ResultCounts,
    outcome: &RunOutcome,
    config: &Config,
) -> io::Result<()> {
    series.write_run(outcome.run.run_id, &outcome.series)?;
    series.flush()?;
    runs.write_run(outcome, config)?;
    runs.flush()?;
    counts.count(outcome.status);
    Ok(())
}

impl<W: Write> RunSink for OutputWriter<W> {
    fn commit(&mut self, outcome: RunOutcome) -> io::Result<()> {
        self.write_run(&outcome)
    }
}

/// Results that cannot be written.
#[derive(Debug)]
pub enum OutputError {
    /// Directory `dir` holds `file` from an earlier sweep or search.
    HoldsResults {
        /// Path of the output directory.
        dir: PathBuf,
        /// Name of the file found, such as `runs.csv` or a staged table.
        file: String,
    },
    /// Another sweep, search or merge holds the lock of directory `dir` and writes to it.
    Locked {
        /// Path of the output directory.
        dir: PathBuf,
    },
    /// The table at `path` is a symbolic link. A writer never follows a symbolic link.
    Link {
        /// Path of `runs.csv` or `series.csv` in the output directory.
        path: PathBuf,
    },
    /// Reading `path` failed.
    Read {
        /// Path of `runs.csv`, or its file name alone for runs held in memory.
        path: PathBuf,
        /// Error from the reader, of kind `InvalidData` for text that is not UTF-8.
        source: io::Error,
    },
    /// Creating or writing `path` failed.
    Write {
        /// Path of the file or directory, or the file name alone for results held in memory.
        path: PathBuf,
        /// Error from the operating system, or from the writer of results held in memory.
        source: io::Error,
    },
    /// A table that cannot be read back.
    Table(ReadError),
    /// `runs.csv` cannot be summarized.
    Summary(SummaryError),
    /// Serializing the manifest failed.
    Manifest(serde_json::Error),
}

impl fmt::Display for OutputError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::HoldsResults { dir, file } => {
                write!(f, "'{}' already holds results ({file})", dir.display())
            }
            Self::Locked { dir } => write!(
                f,
                "another sweep, search or merge is writing to '{}'. Wait for it to end, or stop it",
                dir.display()
            ),
            Self::Link { path } => write!(
                f,
                "'{}' is a symbolic link. A sweep writes to no table through a link",
                path.display()
            ),
            Self::Read { path, .. } => write!(f, "cannot read '{}'", path.display()),
            Self::Write { path, .. } => write!(f, "cannot create or write '{}'", path.display()),
            Self::Table(_) => f.write_str("cannot read a table back"),
            Self::Summary(_) => f.write_str("cannot summarize the runs"),
            Self::Manifest(_) => f.write_str("cannot serialize the manifest"),
        }
    }
}

impl std::error::Error for OutputError {
    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
        match self {
            Self::HoldsResults { .. } | Self::Locked { .. } | Self::Link { .. } => None,
            Self::Read { source, .. } | Self::Write { source, .. } => Some(source),
            Self::Table(error) => Some(error),
            Self::Summary(error) => Some(error),
            Self::Manifest(error) => Some(error),
        }
    }
}

#[cfg(test)]
mod tests {
    use std::fs;

    use super::{
        LOCK_FILE, MANIFEST_FILE, OutputDir, OutputError, RUNS_FILE, SERIES_FILE, STAGED_MARKER, staged_path,
        table_paths,
    };
    use crate::tests::support::ScratchDir;

    #[test]
    fn a_replacement_cut_short_is_finished_once_both_tables_are_staged() {
        let scratch = ScratchDir::new("staged-tables");
        drop(OutputDir::create(scratch.path()).expect("a scratch directory"));
        let path = scratch.path();
        for file in [RUNS_FILE, SERIES_FILE] {
            fs::write(path.join(file), "old\n").expect("a table writes");
            fs::write(staged_path(path, file), "new\n").expect("a staged table writes");
        }
        assert_eq!(table_paths(path).0, path.join(RUNS_FILE), "no marker, no replacement");
        drop(OutputDir::open(path).expect("the directory opens"));
        for file in [RUNS_FILE, SERIES_FILE] {
            assert_eq!(fs::read_to_string(path.join(file)).expect("a table reads"), "old\n");
            assert!(
                !staged_path(path, file).exists(),
                "a staged table without the marker is dropped"
            );
        }

        // The process ended after moving runs.csv into place and before series.csv.
        fs::write(staged_path(path, SERIES_FILE), "new\n").expect("a staged table writes");
        fs::write(path.join(RUNS_FILE), "new\n").expect("a table writes");
        fs::write(path.join(STAGED_MARKER), "").expect("the marker writes");
        assert_eq!(
            table_paths(path),
            (path.join(RUNS_FILE), staged_path(path, SERIES_FILE))
        );
        drop(OutputDir::open(path).expect("the directory opens"));
        for file in [RUNS_FILE, SERIES_FILE] {
            assert_eq!(fs::read_to_string(path.join(file)).expect("a table reads"), "new\n");
        }
        assert!(!path.join(STAGED_MARKER).exists());
    }

    #[test]
    fn a_second_writer_is_refused_while_the_first_holds_the_lock() {
        let scratch = ScratchDir::new("locked-directory");
        let path = scratch.path();
        let first = OutputDir::create(path).expect("a scratch directory");
        assert!(path.join(LOCK_FILE).exists(), "the writer holds the lock file");
        for second in [OutputDir::open(path), OutputDir::create(path)] {
            assert!(matches!(second, Err(OutputError::Locked { .. })), "{second:?}");
        }
        assert!(matches!(
            OutputDir::check_unlocked(path),
            Err(OutputError::Locked { .. })
        ));
        drop(first);
        assert!(path.join(LOCK_FILE).exists(), "the writer leaves its lock file");
        assert!(OutputDir::check_unlocked(path).is_ok());
        let next = OutputDir::open(path).expect("the next writer locks the file again");
        assert!(matches!(OutputDir::open(path), Err(OutputError::Locked { .. })));
        drop(next);

        // A killed process leaves its lock file behind, and the lock is released when the process ends.
        fs::write(path.join(LOCK_FILE), "").expect("a stale lock file writes");
        assert!(OutputDir::check_unlocked(path).is_ok());
        drop(OutputDir::open(path).expect("a stale lock file is taken over"));
    }

    #[cfg(unix)]
    #[test]
    fn a_writer_follows_no_link_in_its_directory() {
        use std::os::unix::fs::symlink;

        use henad_core::explore::spec::SweepSpec;

        use super::SUMMARY_FILE;
        use crate::exec::Concurrency;
        use crate::tests::support::{entry, sweep};

        let scratch = ScratchDir::new("planted-links");
        fs::create_dir_all(scratch.path()).expect("a scratch directory");
        let victim = scratch.path().join("victim.txt");
        fs::write(&victim, "kept\n").expect("a scratch file");
        let path = scratch.path().join("out");
        fs::create_dir(&path).expect("a scratch directory");

        symlink(&victim, path.join(format!("{MANIFEST_FILE}.partial"))).expect("a link");
        symlink(&victim, path.join(LOCK_FILE)).expect("a link");
        let mut spec = SweepSpec::new("game_of_life");
        spec.fixed = vec![
            ("grid_width".to_owned(), "8".to_owned()),
            ("grid_height".to_owned(), "8".to_owned()),
        ];
        spec.run.steps = 2;
        sweep(&entry("game_of_life", None), None, &spec, &path, Concurrency::Auto);
        assert_eq!(fs::read_to_string(&victim).expect("the victim reads"), "kept\n");
        let manifest = fs::symlink_metadata(path.join(MANIFEST_FILE)).expect("the manifest is written");
        assert!(manifest.file_type().is_file());

        let dangling = scratch.path().join("dangling");
        fs::create_dir(&dangling).expect("a scratch directory");
        symlink(scratch.path().join("nowhere.csv"), dangling.join(SUMMARY_FILE)).expect("a link");
        let refused = OutputDir::create(&dangling);
        assert!(
            matches!(&refused, Err(OutputError::HoldsResults { file, .. }) if file == SUMMARY_FILE),
            "{refused:?}"
        );

        let linked = scratch.path().join("linked");
        fs::create_dir(&linked).expect("a scratch directory");
        symlink(&victim, linked.join(RUNS_FILE)).expect("a link");
        let refused = OutputDir::open(&linked);
        assert!(matches!(refused, Err(OutputError::Link { .. })), "{refused:?}");
        assert_eq!(fs::read_to_string(&victim).expect("the victim reads"), "kept\n");
    }

    #[test]
    fn a_directory_holding_results_is_refused() {
        let scratch = ScratchDir::new("holding-results");
        let nested = scratch.path().join("a").join("b");
        let dir = OutputDir::create(&nested).expect("a missing directory is created");
        assert!(dir.path().is_dir());
        drop(dir);
        assert!(OutputDir::create(&nested).is_ok(), "an empty directory is taken");

        fs::write(nested.join("notes.txt"), "kept").expect("a scratch file");
        assert!(
            OutputDir::create(&nested).is_ok(),
            "a file a sweep does not write is left alone"
        );

        fs::write(nested.join(MANIFEST_FILE), "{}").expect("a scratch file");
        let error = OutputDir::create(&nested).expect_err("a manifest marks results");
        assert!(
            matches!(&error, OutputError::HoldsResults { file, .. } if file == MANIFEST_FILE),
            "{error:?}"
        );
        fs::remove_file(nested.join(MANIFEST_FILE)).expect("the manifest is removed");
        fs::write(nested.join(RUNS_FILE), "").expect("a scratch file");
        assert!(OutputDir::check_free(&nested).is_err(), "so does an empty runs.csv");
        fs::remove_file(nested.join(RUNS_FILE)).expect("the table is removed");

        for staged in [staged_path(&nested, SERIES_FILE), nested.join(STAGED_MARKER)] {
            fs::write(&staged, "").expect("a scratch file");
            let error = OutputDir::create(&nested).expect_err("a staged file marks results");
            let name = staged.file_name().map(|name| name.to_string_lossy().into_owned());
            assert!(
                matches!(&error, OutputError::HoldsResults { file, .. } if Some(file) == name.as_ref()),
                "{error:?}"
            );
            fs::remove_file(&staged).expect("the staged file is removed");
        }
        assert!(OutputDir::check_free(&nested).is_ok());
    }
}