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    for e in held.entries.iter().filter(|e| chosen(e)) {
213        if done.contains(&e.id) {
214            outcome.already += 1;
215            continue;
216        }
217        // Addresses and paths in messages are masked, as for a per-row run.
218        let mask = if project.mask_row_values {
219            let mut values: JsonMap<String, Json> = e.options.clone();
220            if let Some(p) = &e.path {
221                values.insert("$path".into(), Json::from(p.clone()));
222            }
223            crate::for_each::RowMask::new(&values, &[])
224        } else {
225            crate::for_each::RowMask::default()
226        };
227        let mut r = Released {
228            id: e.id.clone(),
229            output: e.output.clone(),
230            key: e.key.clone(),
231            profile: e.profile.clone(),
232            status: String::new(),
233            location: None,
234            error: None,
235        };
236        // Byte for byte what was reviewed: a changed or missing file is refused.
237        let changed: Vec<String> = e
238            .files
239            .iter()
240            .chain(&e.attach)
241            .filter(|f| sha256_file(&f.path).ok().as_deref() != Some(f.sha256.as_str()))
242            .map(|f| f.path.display().to_string())
243            .collect();
244        if !changed.is_empty() {
245            r.status = "refused".into();
246            r.error = Some(format!(
247                "refused: {} changed (or is gone) since the run rendered it; re-run the report to render it again",
248                changed.join(", ")
249            ));
250            events.warn(&format!("  {}: {}", e.id, r.error.as_deref().unwrap_or_default()));
251            outcome.released.push(r);
252            continue;
253        }
254        let dest = RowDest {
255            idx: 0,
256            profile: e.profile.clone(),
257            path: e.path.clone(),
258            options: e.options.clone(),
259            hold: false,
260            rate: None,
261            attach: e.attach.iter().map(|f| f.path.clone()).collect(),
262            attach_missing: false,
263        };
264        // Paced per destination profile, at the recorded rate or `--rate`.
265        if let Some(rate) = opts.rate.or(e.rate)
266            && !pacer.wait(&e.profile, &rate, &opts.cancel)
267        {
268            break;
269        }
270        if opts.cancel.is_cancelled() {
271            break;
272        }
273        let files: Vec<PathBuf> = e.files.iter().map(|f| f.path.clone()).collect();
274        match deliver_row(
275            project,
276            &log,
277            &dest,
278            &files,
279            e.message.as_ref(),
280            &mut plugins,
281            &project.root,
282        ) {
283            Ok(Some(loc)) => {
284                r.status = "delivered".into();
285                r.location = Some(mask.apply(&loc));
286                done.insert(e.id.clone());
287                // Recorded at once: a release stopped part-way never sends this one again.
288                write_released(&release_dir, &opts.run_id, &done)?;
289            }
290            Ok(None) => r.status = "not_delivered".into(),
291            Err(err) => {
292                let msg = mask.apply(&err.message);
293                r.status = "failed".into();
294                r.error = Some(msg.clone());
295                events.warn(&format!("  {}: {msg}", e.id));
296            }
297        }
298        outcome.released.push(r);
299    }
300    for (_, p) in plugins {
301        let _ = p.close();
302    }
303    // The release's own record: who, when, what; and every entry released so far.
304    write_released(&release_dir, &opts.run_id, &done)?;
305    let mut record: Json = std::fs::read_to_string(run_dir.join("run_results.json"))
306        .ok()
307        .and_then(|t| serde_json::from_str(&t).ok())
308        .unwrap_or_else(|| serde_json::json!({}));
309    let failed = outcome.failed();
310    if let Json::Object(m) = &mut record {
311        m.insert("run_id".into(), Json::from(release_id.clone()));
312        m.insert(
313            "status".into(),
314            Json::from(if failed { "error" } else { "success" }),
315        );
316        m.insert(
317            "started_at".into(),
318            Json::from(started.to_rfc3339_opts(chrono::SecondsFormat::Secs, true)),
319        );
320        m.insert(
321            "duration_ms".into(),
322            Json::from((chrono::Utc::now() - started).num_milliseconds().max(0)),
323        );
324        m.insert("release_of".into(), Json::from(opts.run_id.clone()));
325        m.insert("released_by".into(), Json::from(opts.by.clone()));
326        m.insert(
327            "released".into(),
328            serde_json::to_value(&outcome.released).unwrap_or_default(),
329        );
330        m.insert("error".into(), Json::Null);
331        m.insert("error_code".into(), Json::Null);
332        m.insert("error_kind".into(), Json::Null);
333        if failed {
334            m.insert(
335                "error".into(),
336                Json::from("one or more held deliveries failed or were refused"),
337            );
338            m.insert("error_code".into(), Json::from(Code::ReleaseFailed.to_string()));
339        }
340        m.remove("retry_of");
341    }
342    crate::fs::write_atomic(
343        &release_dir.join("run_results.json"),
344        serde_json::to_string_pretty(&record)
345            .unwrap_or_default()
346            .as_bytes(),
347    )
348    .map_err(|e| e.to_string())?;
349    Ok(outcome)
350}
351
352/// Every entry released so far, this release's included, in the release's folder.
353fn write_released(release_dir: &Path, run_id: &str, done: &BTreeSet<String>) -> Result<(), Fail> {
354    let file = ReleasedFile {
355        release_of: run_id.to_string(),
356        released: done.iter().cloned().collect(),
357    };
358    crate::fs::write_atomic(
359        &release_dir.join(RELEASED),
360        serde_json::to_string(&file).unwrap_or_default().as_bytes(),
361    )
362    .map_err(|e| Fail::from(e.to_string()))
363}
dre_core/release.rs

dre_core/
release.rs
