dre_core/
release.rs

1//! Held deliveries and `dre release`: a maker/checker step between rendering and sending.
2//!
3//! A destination with `hold: true` (or every destination, with `dre run --hold`) isn't delivered
4//! by the run: the run renders everything, delivers the rest, and records each held delivery in
5//! `held.json` in its run folder, with the files' SHA-256. `dre release <report> --run <id>`
6//! delivers them later from that folder's files, byte for byte what was reviewed: it never
7//! re-renders or re-queries, refuses a file that changed since, takes the Binding's lock, and
8//! never sends a delivery twice. Each release is its own run folder (`release_of`), recording
9//! who released what, and when.
10
11use std::collections::{BTreeSet, HashMap};
12use std::path::{Path, PathBuf};
13
14use serde::{Deserialize, Serialize};
15use serde_json::{Map as JsonMap, Value as Json};
16use sha2::Digest;
17
18use crate::codes::Code;
19use crate::engine::Events;
20use crate::error::Error as Fail;
21use crate::project::Project;
22use crate::run::per_row::{RowDest, deliver_row};
23use crate::run_results::Released;
24use crate::runs::BindingRuns;
25
26/// `held.json` in a run folder.
27pub const HELD: &str = "held.json";
28pub const SCHEMA_VERSION: &str = "dre/held/v1";
29/// In a release's run folder: every entry released so far, this release's included.
30pub const RELEASED: &str = "released.json";
31
32#[derive(Debug, Clone, Serialize, Deserialize)]
33pub struct Held {
34    pub schema_version: String,
35    pub run_id: String,
36    pub entries: Vec<HeldEntry>,
37}
38
39/// One held delivery: the destination entry as rendered (unmasked: it's what will be sent) and
40/// the files it sends.
41#[derive(Debug, Clone, Serialize, Deserialize)]
42pub struct HeldEntry {
43    /// `<output>/<key>/<n>` for a per-row output, `<output>/<n>` otherwise.
44    pub id: String,
45    pub output: Option<String>,
46    #[serde(default, skip_serializing_if = "Option::is_none")]
47    pub key: Option<String>,
48    pub profile: String,
49    #[serde(default, skip_serializing_if = "Option::is_none")]
50    pub path: Option<String>,
51    #[serde(default)]
52    pub options: JsonMap<String, Json>,
53    /// Files attached to a message, with their SHA-256 at the run.
54    #[serde(default, skip_serializing_if = "Vec::is_empty")]
55    pub attach: Vec<HeldFile>,
56    /// The destination's `rate:` at the run, which the release keeps unless `--rate` says otherwise.
57    #[serde(default, skip_serializing_if = "Option::is_none")]
58    pub rate: Option<crate::rate::Rate>,
59    /// A message's title and text.
60    #[serde(default, skip_serializing_if = "Option::is_none")]
61    pub message: Option<(String, String)>,
62    pub files: Vec<HeldFile>,
63}
64
65#[derive(Debug, Clone, Serialize, Deserialize)]
66pub struct HeldFile {
67    pub path: PathBuf,
68    pub sha256: String,
69}
70
71pub fn sha256_file(p: &Path) -> std::io::Result<String> {
72    let bytes = std::fs::read(p)?;
73    Ok(sha2::Sha256::digest(&bytes)
74        .iter()
75        .map(|b| format!("{b:02x}"))
76        .collect())
77}
78
79/// A file and its SHA-256 now, for a held delivery.
80pub fn held_file(path: &Path) -> HeldFile {
81    HeldFile {
82        path: path.to_path_buf(),
83        sha256: sha256_file(path).unwrap_or_default(),
84    }
85}
86
87pub fn write_held(run_dir: &Path, run_id: &str, entries: &[HeldEntry]) -> std::io::Result<()> {
88    let held = Held {
89        schema_version: SCHEMA_VERSION.into(),
90        run_id: run_id.to_string(),
91        entries: entries.to_vec(),
92    };
93    crate::fs::write_atomic(
94        &run_dir.join(HELD),
95        serde_json::to_string_pretty(&held)?.as_bytes(),
96    )
97}
98
99/// `released.json` in a release's folder.
100#[derive(Debug, Clone, Serialize, Deserialize)]
101struct ReleasedFile {
102    release_of: String,
103    released: Vec<String>,
104}
105
106/// What `dre release` was asked to do.
107#[derive(Debug, Clone, Default)]
108pub struct ReleaseOptions {
109    pub report: String,
110    /// The Binding's folder name: the Set's, or `default`.
111    pub binding: String,
112    pub run_id: String,
113    pub only_keys: Option<Vec<String>>,
114    pub exclude_keys: Option<Vec<String>>,
115    /// Who releases (`--by`, default `$USER`).
116    pub by: String,
117    /// `--rate`: every delivery's rate for this release, over the recorded ones.
118    pub rate: Option<crate::rate::Rate>,
119    /// Ctrl-C or a termination: no delivery starts after it.
120    pub cancel: crate::engine::CancelToken,
121}
122
123/// How a release went.
124#[derive(Debug, Clone, Default)]
125pub struct ReleaseOutcome {
126    pub release_id: String,
127    pub released: Vec<Released>,
128    /// Entries skipped because an earlier release delivered them.
129    pub already: usize,
130}
131
132impl ReleaseOutcome {
133    pub fn failed(&self) -> bool {
134        self.released
135            .iter()
136            .any(|r| r.status != "delivered" && r.status != "not_delivered")
137    }
138}
139
140/// Deliver run `opts.run_id`'s held deliveries.
141pub fn release(project: &Project, opts: &ReleaseOptions, events: &Events) -> Result<ReleaseOutcome, Fail> {
142    let store = crate::engine::RunStore::new(&project.target_dir);
143    let runs = BindingRuns::new(&store.run_dir(&opts.report, &opts.binding));
144    let run_dir = runs.run_dir(&opts.run_id);
145    let held_path = run_dir.join(HELD);
146    if !run_dir.is_dir() {
147        return Err(Fail::new(
148            Code::ReleaseFailed,
149            format!(
150                "{} [{}] has no run `{}` in the target path (it may have been pruned: keep it with `keep_runs` for the review window; `dre history` lists the runs)",
151                opts.report, opts.binding, opts.run_id
152            ),
153        ));
154    }
155    let held: Held = match std::fs::read_to_string(&held_path) {
156        Ok(t) => {
157            serde_json::from_str(&t).map_err(|e| format!("{} is malformed: {e}", held_path.display()))?
158        }
159        Err(_) => {
160            return Err(Fail::new(
161                Code::ReleaseFailed,
162                format!(
163                    "run `{}` has no held deliveries (no {HELD} in its folder)",
164                    opts.run_id
165                ),
166            ));
167        }
168    };
169    // Entries an earlier release delivered: the newest release's cumulative list.
170    let mut done: BTreeSet<String> = BTreeSet::new();
171    // Each release keeps every entry released so far (written as each one goes out).
172    for id in runs.list() {
173        if let Ok(t) = std::fs::read_to_string(runs.run_dir(&id).join(RELEASED))
174            && let Ok(r) = serde_json::from_str::<ReleasedFile>(&t)
175            && r.release_of == opts.run_id
176        {
177            done.extend(r.released);
178        }
179    }
180    let started = chrono::Utc::now();
181    let release_id = crate::runs::new_run_id(started);
182    let _guard = runs.lock(&release_id, started).map_err(|e| {
183        let why = match e {
184            crate::runs::LockError::Held(h) => format!("{} holds its lock", h.describe()),
185            crate::runs::LockError::Unreadable(p) => format!("its lock file {} is unreadable", p.display()),
186            crate::runs::LockError::Io(e) => format!("its lock can't be taken: {e}"),
187        };
188        Fail::new(
189            Code::RunInProgress,
190            format!(
191                "{} [{}] is busy: {why}; release when it finishes",
192                opts.report, opts.binding
193            ),
194        )
195    })?;
196    let release_dir = runs.run_dir(&release_id);
197    std::fs::create_dir_all(&release_dir).map_err(|e| e.to_string())?;
198    let mut outcome = ReleaseOutcome {
199        release_id: release_id.clone(),
200        ..ReleaseOutcome::default()
201    };
202    let chosen = |e: &HeldEntry| match &e.key {
203        Some(k) => {
204            opts.only_keys.as_ref().is_none_or(|ks| ks.contains(k))
205                && opts.exclude_keys.as_ref().is_none_or(|ks| !ks.contains(k))
206        }
207        None => opts.only_keys.is_none(),
208    };
209    let log = events.plugin_log();
210    let mut plugins: HashMap<String, dre_protocol::host::PluginProcess> = HashMap::new();
211    let pacer = crate::rate::Pacer::default();
212    let mut progress = RowsLeft::new(
213        held.entries.iter().filter(|e| chosen(e) && !done.contains(&e.id)),
214        opts.rate,
215    );
216    progress.start(events);
217    for e in held.entries.iter().filter(|e| chosen(e)) {
218        if done.contains(&e.id) {
219            outcome.already += 1;
220            continue;
221        }
222        // Addresses and paths in messages are masked, as for a per-row run.
223        let mask = if project.mask_row_values {
224            let mut values: JsonMap<String, Json> = e.options.clone();
225            if let Some(p) = &e.path {
226                values.insert("$path".into(), Json::from(p.clone()));
227            }
228            crate::for_each::RowMask::new(&values, &[])
229        } else {
230            crate::for_each::RowMask::default()
231        };
232        let mut r = Released {
233            id: e.id.clone(),
234            output: e.output.clone(),
235            key: e.key.clone(),
236            profile: e.profile.clone(),
237            status: String::new(),
238            location: None,
239            error: None,
240        };
241        // Byte for byte what was reviewed: a changed or missing file is refused.
242        let changed: Vec<String> = e
243            .files
244            .iter()
245            .chain(&e.attach)
246            .filter(|f| sha256_file(&f.path).ok().as_deref() != Some(f.sha256.as_str()))
247            .map(|f| f.path.display().to_string())
248            .collect();
249        if !changed.is_empty() {
250            r.status = "refused".into();
251            r.error = Some(format!(
252                "refused: {} changed (or is gone) since the run rendered it; re-run the report to render it again",
253                changed.join(", ")
254            ));
255            events.warn(&format!("  {}: {}", e.id, r.error.as_deref().unwrap_or_default()));
256            outcome.released.push(r);
257            progress.entry_done(e, true, events);
258            continue;
259        }
260        let dest = RowDest {
261            idx: 0,
262            profile: e.profile.clone(),
263            path: e.path.clone(),
264            options: e.options.clone(),
265            hold: false,
266            rate: None,
267            attach: e.attach.iter().map(|f| f.path.clone()).collect(),
268            attach_missing: false,
269            // A held zip is already one of the entry's files: sent as it is.
270            zip: None,
271            zip_file: None,
272        };
273        // Paced per destination profile, at the recorded rate or `--rate`.
274        if let Some(rate) = opts.rate.or(e.rate)
275            && !pacer.wait(&e.profile, &rate, &opts.cancel)
276        {
277            break;
278        }
279        if opts.cancel.is_cancelled() {
280            break;
281        }
282        let files: Vec<PathBuf> = e.files.iter().map(|f| f.path.clone()).collect();
283        match deliver_row(
284            project,
285            &log,
286            &dest,
287            &files,
288            e.message.as_ref(),
289            &mut plugins,
290            &project.root,
291        ) {
292            Ok(Some(loc)) => {
293                r.status = "delivered".into();
294                r.location = Some(mask.apply(&loc));
295                done.insert(e.id.clone());
296                // Recorded at once: a release stopped part-way never sends this one again.
297                write_released(&release_dir, &opts.run_id, &done)?;
298            }
299            Ok(None) => r.status = "not_delivered".into(),
300            Err(err) => {
301                let msg = mask.apply(&err.message);
302                r.status = "failed".into();
303                r.error = Some(msg.clone());
304                events.warn(&format!("  {}: {msg}", e.id));
305            }
306        }
307        progress.entry_done(e, r.status == "failed", events);
308        outcome.released.push(r);
309    }
310    progress.finish(events);
311    for (_, p) in plugins {
312        let _ = p.close();
313    }
314    // The release's own record: who, when, what; and every entry released so far.
315    write_released(&release_dir, &opts.run_id, &done)?;
316    let mut record: Json = std::fs::read_to_string(run_dir.join("run_results.json"))
317        .ok()
318        .and_then(|t| serde_json::from_str(&t).ok())
319        .unwrap_or_else(|| serde_json::json!({}));
320    let failed = outcome.failed();
321    if let Json::Object(m) = &mut record {
322        m.insert("run_id".into(), Json::from(release_id.clone()));
323        m.insert(
324            "status".into(),
325            Json::from(if failed { "error" } else { "success" }),
326        );
327        m.insert(
328            "started_at".into(),
329            Json::from(started.to_rfc3339_opts(chrono::SecondsFormat::Secs, true)),
330        );
331        m.insert(
332            "duration_ms".into(),
333            Json::from((chrono::Utc::now() - started).num_milliseconds().max(0)),
334        );
335        m.insert("release_of".into(), Json::from(opts.run_id.clone()));
336        m.insert("released_by".into(), Json::from(opts.by.clone()));
337        m.insert(
338            "released".into(),
339            serde_json::to_value(&outcome.released).unwrap_or_default(),
340        );
341        m.insert("error".into(), Json::Null);
342        m.insert("error_code".into(), Json::Null);
343        m.insert("error_kind".into(), Json::Null);
344        if failed {
345            m.insert(
346                "error".into(),
347                Json::from("one or more held deliveries failed or were refused"),
348            );
349            m.insert("error_code".into(), Json::from(Code::ReleaseFailed.to_string()));
350        }
351        m.remove("retry_of");
352    }
353    crate::fs::write_atomic(
354        &release_dir.join("run_results.json"),
355        serde_json::to_string_pretty(&record)
356            .unwrap_or_default()
357            .as_bytes(),
358    )
359    .map_err(|e| e.to_string())?;
360    Ok(outcome)
361}
362
363/// Every entry released so far, this release's included, in the release's folder.
364fn write_released(release_dir: &Path, run_id: &str, done: &BTreeSet<String>) -> Result<(), Fail> {
365    let file = ReleasedFile {
366        release_of: run_id.to_string(),
367        released: done.iter().cloned().collect(),
368    };
369    crate::fs::write_atomic(
370        &release_dir.join(RELEASED),
371        serde_json::to_string(&file).unwrap_or_default().as_bytes(),
372    )
373    .map_err(|e| Fail::from(e.to_string()))
374}
375
376/// A release's progress in rows, per output: a row is done once each of its held deliveries
377/// this release sends has gone (or failed, or was refused).
378struct RowsLeft {
379    outputs: Vec<crate::run::RowProgress>,
380    /// Per (output label, row): deliveries still to go, and whether one failed.
381    rows: HashMap<(String, String), (usize, bool)>,
382}
383
384impl RowsLeft {
385    fn label(e: &HeldEntry) -> String {
386        match &e.output {
387            Some(o) => format!("output `{o}`"),
388            None => "output".to_string(),
389        }
390    }
391
392    /// A row's key: an entry without one (not a per-row output) is a row of its own.
393    fn row(e: &HeldEntry) -> (String, String) {
394        (Self::label(e), e.key.clone().unwrap_or_else(|| e.id.clone()))
395    }
396
397    fn new<'a>(pending: impl Iterator<Item = &'a HeldEntry>, rate: Option<crate::rate::Rate>) -> RowsLeft {
398        let mut p = RowsLeft {
399            outputs: Vec::new(),
400            rows: HashMap::new(),
401        };
402        for e in pending {
403            let label = Self::label(e);
404            let slot = p.rows.entry(Self::row(e)).or_insert((0, false));
405            slot.0 += 1;
406            let new_row = slot.0 == 1;
407            match p.outputs.iter_mut().find(|o| o.output == label) {
408                Some(o) => {
409                    if new_row {
410                        o.total += 1;
411                    }
412                    o.rate = o.rate.or(rate.or(e.rate));
413                }
414                None => p.outputs.push(crate::run::RowProgress {
415                    output: label,
416                    done: 0,
417                    failed: 0,
418                    total: 1,
419                    rate: rate.or(e.rate),
420                    finished: false,
421                }),
422            }
423        }
424        p
425    }
426
427    fn start(&self, events: &Events) {
428        for o in &self.outputs {
429            events.rows(o.clone());
430        }
431    }
432
433    fn entry_done(&mut self, e: &HeldEntry, failed: bool, events: &Events) {
434        let Some(slot) = self.rows.get_mut(&Self::row(e)) else {
435            return;
436        };
437        slot.0 = slot.0.saturating_sub(1);
438        slot.1 |= failed;
439        if slot.0 > 0 {
440            return;
441        }
442        let row_failed = slot.1;
443        let label = Self::label(e);
444        if let Some(o) = self.outputs.iter_mut().find(|o| o.output == label) {
445            o.done += 1;
446            if row_failed {
447                o.failed += 1;
448            }
449            events.rows(o.clone());
450        }
451    }
452
453    fn finish(&mut self, events: &Events) {
454        for o in &mut self.outputs {
455            o.finished = true;
456            events.rows(o.clone());
457        }
458    }
459}
460
461#[cfg(test)]
462mod tests {
463    use super::*;
464    use crate::engine::RunEvent;
465
466    fn entry(output: &str, key: &str, n: usize) -> HeldEntry {
467        HeldEntry {
468            id: format!("{output}/{key}/{n}"),
469            output: Some(output.into()),
470            key: Some(key.into()),
471            profile: "mail".into(),
472            path: None,
473            options: JsonMap::new(),
474            attach: Vec::new(),
475            rate: Some(crate::rate::Rate::parse("50/second").unwrap()),
476            message: None,
477            files: Vec::new(),
478        }
479    }
480
481    #[test]
482    fn a_row_is_done_once_all_its_deliveries_are() {
483        // Row `a` has two deliveries (email and a copy), row `b` one.
484        let entries = [
485            entry("email", "a", 0),
486            entry("email", "a", 1),
487            entry("email", "b", 0),
488        ];
489        let (tx, rx) = std::sync::mpsc::channel();
490        let events = Events::new(tx);
491        let mut p = RowsLeft::new(entries.iter(), None);
492        p.start(&events);
493        p.entry_done(&entries[0], false, &events);
494        p.entry_done(&entries[1], true, &events);
495        p.entry_done(&entries[2], false, &events);
496        p.finish(&events);
497        drop(events);
498        let seen: Vec<(u64, u64, u64, bool)> = rx
499            .try_iter()
500            .map(|e| match e {
501                RunEvent::Rows(r) => {
502                    assert_eq!(r.output, "output `email`");
503                    assert_eq!(r.rate.map(|r| r.shown()).as_deref(), Some("50/second"));
504                    (r.done, r.failed, r.total, r.finished)
505                }
506                _ => panic!("only progress"),
507            })
508            .collect();
509        // Started; `a` done (one delivery failed, so the row failed); `b` done; finished.
510        assert_eq!(
511            seen,
512            vec![
513                (0, 0, 2, false),
514                (1, 1, 2, false),
515                (2, 1, 2, false),
516                (2, 1, 2, true)
517            ]
518        );
519    }
520}
dre_core/release.rs

dre_core/
release.rs
