Skip to main content

haematite_cli/serve/storage/
release.rs

1//! STOR-003 HR1: what a branch's versions are relied on by, which a release
2//! never passes (graph-and-query-surface.md, "What a version is relied on
3//! by"): the head; a branch forked from it, while that branch exists; an open
4//! merge's three moments; a discard asked for and not finished; and the
5//! embedding job's position in each declared field (V4).
6//!
7//! The cost is one tree read for each live branch (its fork origin), one
8//! paged read of the open merge rows, and the pending discards; never in
9//! proportion to the branch's history. The branches read are counted so the
10//! answer can show it.
11
12use std::collections::BTreeMap;
13use std::num::{NonZeroU64, NonZeroUsize};
14
15use haematite::SyncCounts;
16use haematite::branch::history::Row;
17use haematite::branch::{CommitDurability, CommitRequest, KeptHead, Timestamp};
18use haematite::db::ReadOnlyDatabase;
19use haematite::tree::Hash;
20use serde::Serialize;
21
22use super::branches::provenance;
23use super::dead;
24use super::embedded;
25use super::history::{self, Decision, Released};
26use super::{
27    BRANCH_PREFIX, BranchRefRecord, Captured, Failure, SERVICE, Storage, branch_name, identifier,
28    open_branch, version,
29};
30use crate::serve::protocol::Version;
31
32/// What relies on a version.
33#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
34#[serde(rename_all = "snake_case")]
35pub enum Source {
36    Head,
37    Fork,
38    Merge,
39    Discard,
40    /// The embedding job's finished `through` or its open walk's target,
41    /// named `collection.field`: its next page reads both.
42    Embed,
43    /// Another branch's companion position naming this branch, as a fork's
44    /// inherits its parent's (F2): named `branch:collection.field`.
45    EmbedOf,
46    /// The branch's own setting to keep its newest versions (STOR-003 HR3).
47    KeepVersions,
48    /// The branch's own setting to keep versions superseded recently.
49    KeepDays,
50    /// The branch is a line (STOR-050 R2): it keeps every version it
51    /// commits until a discard names it.
52    Line,
53}
54
55/// One thing relying on a version of the branch: what, by name, and which.
56#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
57pub struct HeldBy {
58    pub source: Source,
59    /// The branch for a head or a fork, the merge id, or the discarded branch.
60    pub name: String,
61    pub version: Version,
62}
63
64/// Everything relying on a branch's versions, oldest first.
65#[derive(Debug, Clone, Default, PartialEq, Eq)]
66pub struct Floor {
67    /// Ordered by (generation, sequence), oldest first. The head is among
68    /// them when asked of [`Storage::relied_on`]; asked of
69    /// [`Storage::relied_on_by_others`] it is not, and the list may be empty.
70    pub held_by: Vec<HeldBy>,
71    /// Live branches whose fork origin was read.
72    pub branches_read: usize,
73}
74
75impl Floor {
76    /// The oldest version something relies on: a release takes only what is
77    /// below it. `None` when nothing relies on any version: every version may
78    /// go.
79    pub fn oldest(&self) -> Option<&HeldBy> {
80        self.held_by.first()
81    }
82
83    /// The oldest version relied on in `generation` (DR-6): a version holds
84    /// the rows of its own generation only, so a hold on an older generation
85    /// never holds a newer one back.
86    pub fn oldest_in(&self, generation: u64) -> Option<Version> {
87        self.held_by
88            .iter()
89            .map(|held| held.version)
90            .filter(|version| version.generation == generation)
91            .min_by_key(|version| order(*version))
92    }
93}
94
95/// The pending discards as the store lists them from their intents
96/// (STOR-003 H1b2-B): what the release driver reads its floor with.
97pub struct IntentListing;
98
99impl PendingDiscards for IntentListing {
100    fn pending(&self, storage: &Storage) -> Result<Vec<(String, Version)>, Failure> {
101        storage.pending_discards()
102    }
103}
104
105/// The discards asked for and not finished, as (branch, version) each.
106pub trait PendingDiscards {
107    /// # Errors
108    /// Whatever stops the pending discards from being read.
109    fn pending(&self, storage: &Storage) -> Result<Vec<(String, Version)>, Failure>;
110}
111
112/// The pending discards until the intent listing in `discards.rs` can be
113/// called from here (STOR-003 H1b2): refused by name, so no release ever runs
114/// without them. Nothing outside the tests asks for a floor yet (HR3 will),
115/// so only they read it.
116#[cfg(test)]
117pub struct DiscardsUnread;
118
119#[cfg(test)]
120impl PendingDiscards for DiscardsUnread {
121    fn pending(&self, _storage: &Storage) -> Result<Vec<(String, Version)>, Failure> {
122        Err(Failure::new(
123            "release_discards_unread",
124            "the pending discards cannot be read yet, so nothing is released",
125        ))
126    }
127}
128
129impl Storage {
130    /// Everything relying on `branch`'s versions now, its head among them
131    /// (STOR-003 HR1).
132    ///
133    /// # Errors
134    /// `branch_missing`; whatever reading a fork origin, a merge row or the
135    /// pending discards answers.
136    pub fn relied_on(
137        &self,
138        branch: &str,
139        discards: &dyn PendingDiscards,
140    ) -> Result<Floor, Failure> {
141        identifier(branch)?;
142        let head = self
143            .refs
144            .get(&branch_name(branch))
145            .ok_or_else(|| Failure::new("branch_missing", branch))?;
146        let head = HeldBy {
147            source: Source::Head,
148            name: branch.to_owned(),
149            version: version(head),
150        };
151        let mut floor = self.relied_on_by_others(branch, false, discards)?;
152        floor.held_by.push(head);
153        floor
154            .held_by
155            .sort_by_key(|held| (held.version.generation, held.version.sequence));
156        Ok(floor)
157    }
158
159    /// Everything other than `branch`'s own head relying on its versions:
160    /// the branches forked from it, the open merges pinning it, and its
161    /// pending discard unless `skip_own_discard` (STOR-003 `HR1b`). Its ref may
162    /// be hidden or gone, as a discard being finished has it.
163    ///
164    /// An empty floor means nothing relies on any version of `branch`, so
165    /// every version of it may go; it never means that nothing may.
166    ///
167    /// # Errors
168    /// Whatever reading a fork origin, a merge row or the pending discards
169    /// answers.
170    pub(super) fn relied_on_by_others(
171        &self,
172        branch: &str,
173        skip_own_discard: bool,
174        discards: &dyn PendingDiscards,
175    ) -> Result<Floor, Failure> {
176        identifier(branch)?;
177        let mut held_by = Vec::new();
178        let mut branches_read = 0_usize;
179        for record in self.refs.list_prefix(BRANCH_PREFIX) {
180            let Some(fork) = record.name.strip_prefix(BRANCH_PREFIX) else {
181                continue;
182            };
183            if fork == branch {
184                continue;
185            }
186            let at = Captured {
187                branch: fork.to_owned(),
188                version: version(record),
189                shards: record.shards.clone(),
190            };
191            branches_read = branches_read.saturating_add(1);
192            if let Some(origin) = provenance(&self.reader, &at)?
193                && origin.branch == branch
194            {
195                held_by.push(HeldBy {
196                    source: Source::Fork,
197                    name: fork.to_owned(),
198                    version: origin.version,
199                });
200            }
201        }
202        for plan in self.open_merge_plans()? {
203            for pin in [&plan.origin, &plan.source, &plan.target] {
204                if pin.branch == branch {
205                    held_by.push(HeldBy {
206                        source: Source::Merge,
207                        name: plan.id.clone(),
208                        version: pin.version,
209                    });
210                }
211            }
212        }
213        // One pending discard at most names a branch (reconcile refuses two),
214        // so skipping the branch's entries skips exactly its own.
215        for (discarded, at) in discards.pending(self)? {
216            if discarded == branch && !skip_own_discard {
217                held_by.push(HeldBy {
218                    source: Source::Discard,
219                    name: discarded,
220                    version: at,
221                });
222            }
223        }
224        held_by.extend(self.embed_positions(branch)?);
225        held_by.sort_by_key(|held| (held.version.generation, held.version.sequence));
226        Ok(Floor {
227            held_by,
228            branches_read,
229        })
230    }
231}
232
233impl Storage {
234    /// The versions of `branch` the embedding job's positions hold, each on
235    /// the branch it is a version of (F2b): a position's `through` on the
236    /// branch it names, from its own companion (V4) or another's, as a
237    /// fork's inherited one names its parent (F2); an open walk's target on
238    /// the companion's owner. Every companion is read, one
239    /// paged read of its position keys each, so a companion whose branch is
240    /// pending discard holds until that discard finishes; the reads are in
241    /// proportion to the companions and their declared fields.
242    fn embed_positions(&self, branch: &str) -> Result<Vec<HeldBy>, Failure> {
243        let mut held = Vec::new();
244        for record in self.refs.list_prefix(embedded::COMPANION_PREFIX) {
245            let Some(owner) = record.name.strip_prefix(embedded::COMPANION_PREFIX) else {
246                continue;
247            };
248            for ((collection, field), position) in self.companion_positions(record)? {
249                let field = format!("{collection}.{field}");
250                // `through` is a version of the branch the position names.
251                if position.branch == branch
252                    && let Some(through) = position.through
253                {
254                    let (source, name) = if owner == branch {
255                        (Source::Embed, field.clone())
256                    } else {
257                        (Source::EmbedOf, format!("{owner}:{field}"))
258                    };
259                    held.push(HeldBy {
260                        source,
261                        name,
262                        version: through,
263                    });
264                }
265                // An open walk's target is always the owner's own version
266                // (F-2, V2c), whatever branch the position names.
267                if owner == branch
268                    && let Some(open) = position.toward
269                {
270                    held.push(HeldBy {
271                        source: Source::Embed,
272                        name: field,
273                        version: open.version,
274                    });
275                }
276            }
277        }
278        Ok(held)
279    }
280
281    /// Every position in a companion, by (collection, field), in key order:
282    /// one paged read of its position keys. The one reader of that layout,
283    /// for the floor and for a fork's inheriting (F1b).
284    ///
285    /// # Errors
286    /// `embed_position_key_corrupt` for a key that is not UTF-8 or names no
287    /// field; `embed_position_corrupt`; whatever a read answers.
288    pub(super) fn companion_positions(
289        &self,
290        companion: &BranchRefRecord,
291    ) -> Result<BTreeMap<(String, String), embedded::Position>, Failure> {
292        let limit = NonZeroUsize::new(POSITIONS_A_PAGE)
293            .ok_or_else(|| Failure::new("embed_positions_page", "a page must be nonzero"))?;
294        let mut positions = BTreeMap::new();
295        for shard in &companion.shards {
296            let mut from = b"e\0".to_vec();
297            loop {
298                let page = self
299                    .reader
300                    .range_page_at(shard.shard_id, shard.head, &from, b"e\x01", limit)
301                    .map_err(Failure::observer)?;
302                for (key, bytes) in page.entries {
303                    let field = position_field(&companion.name, &key)?;
304                    positions.insert(field, embedded::read_position(&bytes)?);
305                }
306                let Some(next) = page.next_from else {
307                    break;
308                };
309                from = next;
310            }
311        }
312        Ok(positions)
313    }
314}
315
316/// The (collection, field) a position key names, refused by name, never
317/// guessed, when it is not UTF-8 or has no field part. The key's bytes are
318/// never printed.
319fn position_field(companion: &str, key: &[u8]) -> Result<(String, String), Failure> {
320    let corrupt = || {
321        Failure::new(
322            "embed_position_key_corrupt",
323            format!("{companion} holds a position key that names no collection and field"),
324        )
325    };
326    let named = key.strip_prefix(b"e\0".as_slice()).ok_or_else(corrupt)?;
327    let named = std::str::from_utf8(named).map_err(|_error| corrupt())?;
328    let (collection, field) = named.split_once('\0').ok_or_else(corrupt)?;
329    Ok((collection.to_owned(), field.to_owned()))
330}
331
332/// Position keys read a page at a time.
333const POSITIONS_A_PAGE: usize = 64;
334
335/// What a branch keeps, resolved from the store's settings and the branch's
336/// exception (ruled 07:24). Each setting left `None` keeps all (the design's
337/// default), so the default keeps every version.
338#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
339pub struct Keep {
340    /// The newest this many versions, the head included.
341    pub versions: Option<NonZeroU64>,
342    /// Versions superseded fewer than this many days ago.
343    pub days: Option<u64>,
344}
345
346/// How one release judges and bounds what it takes.
347#[derive(Debug, Clone, Copy, PartialEq, Eq)]
348pub struct Rules {
349    /// `None` judges no keep settings at all, as a finishing discard asks
350    /// (ruled 08:05); never the same as `Some(Keep::default())`, which keeps
351    /// every version.
352    pub keep: Option<Keep>,
353    /// `retention.release_max_rows`: the most one release may name.
354    pub max_rows: NonZeroUsize,
355    /// Now, in nanoseconds since the epoch, as row timestamps are.
356    pub now: Timestamp,
357}
358
359/// The oldest and newest version one release took.
360#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
361pub struct Span {
362    pub first: Version,
363    pub last: Version,
364}
365
366/// What `history.release` answers (ruled 07:24, 8).
367#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
368pub struct Answer {
369    pub branch: String,
370    pub released: Option<Span>,
371    pub marker: Option<Released>,
372    /// What stopped the release short of its limit; empty when nothing did.
373    pub held_by: Vec<HeldBy>,
374    /// Rows still qualify beyond the limit.
375    pub more: bool,
376    pub branches_read: usize,
377    /// Whether a reclaim pass was asked for after the release.
378    pub pass_asked: bool,
379    /// Why asking for that pass was refused, when it was.
380    pub pass_refused: Option<String>,
381    /// When the row the keep days hold falls due: when it was superseded,
382    /// plus the days. `None` when no row is held by days with a known
383    /// superseding time. Read from the rows this release read, nothing more.
384    pub due_at: Option<Timestamp>,
385}
386
387const NANOS_A_DAY: u64 = 86_400 * 1_000_000_000;
388
389/// Where a branch's first release starts: below every version it has.
390const LOWEST: Version = Version {
391    generation: 0,
392    sequence: 0,
393};
394
395const fn order(version: Version) -> (u64, u64) {
396    (version.generation, version.sequence)
397}
398
399const fn row_version(row: &Row) -> Version {
400    Version {
401        generation: row.key.generation,
402        sequence: row.key.sequence,
403    }
404}
405
406/// The newest a release may reach: `version` and below never, and when it
407/// was written. A release passes the head; a finishing discard passes the
408/// version after its intent's, so the intent's own row may go.
409#[derive(Debug, Clone, Copy, PartialEq, Eq)]
410pub struct Below {
411    pub version: Version,
412    pub written: Timestamp,
413}
414
415/// What the oldest `cap` of `rows` may go, judged against the floor and the
416/// keep settings: the versions taken, what stopped them, and whether the row
417/// after the cap would have gone too. `rows` runs from the oldest kept row
418/// towards `below`, at most `cap + 1` of them. An empty floor bounds nothing:
419/// only `below` does.
420///
421/// DR-8: until something is taken, rows something relies on are passed, not
422/// stopped at: a fork holding an older generation of a live name holds that
423/// generation only, so the newer rows behind it in the same read are judged.
424/// What was passed counts towards `cap`, and is what held the release when
425/// nothing is taken. Taken rows stay one span: nothing is passed after the
426/// first is taken.
427fn judge(
428    branch: &str,
429    rows: &[Row],
430    cap: usize,
431    floor: &Floor,
432    below: Below,
433    rules: &Rules,
434) -> (Vec<Version>, Vec<HeldBy>, bool) {
435    let mut taken = Vec::new();
436    let mut passed = Vec::new();
437    for (index, row) in rows.iter().enumerate() {
438        let at = row_version(row);
439        if order(at) >= order(below.version) {
440            break;
441        }
442        let held = held_at(branch, rows, index, floor, below, rules);
443        if index == cap {
444            // Rows passed fill the cap without a release: the next would read
445            // and pass the same ones, so no more is asked for.
446            let more = held.is_empty() && !taken.is_empty();
447            let held_by = if taken.is_empty() { passed } else { Vec::new() };
448            return (taken, held_by, more);
449        }
450        if !held.is_empty() {
451            if taken.is_empty() && floor_holds(floor, at) {
452                if passed.is_empty() {
453                    passed = held;
454                }
455                continue;
456            }
457            return (taken, held, false);
458        }
459        taken.push(at);
460    }
461    let held_by = if taken.is_empty() { passed } else { Vec::new() };
462    (taken, held_by, false)
463}
464
465/// Where a release reads its rows: from `from` towards `below`, at most
466/// `limit` + 1 a read, and whether the name is live (DR-13).
467#[derive(Debug, Clone, Copy)]
468struct ReadFrom {
469    from: Version,
470    below: Below,
471    limit: NonZeroUsize,
472    live: bool,
473}
474
475/// What `judge` answers: the rows taken, what held the release, and more.
476type Judged = (Vec<Version>, Vec<HeldBy>, bool);
477
478/// DR-13: the start of the generation after the last of `rows`' first `cap`,
479/// when the read filled its cap with rows `floor` relies on and that
480/// generation is older than `below`'s; `None` otherwise.
481fn past_held(rows: &[Row], cap: usize, floor: &Floor, below: Version) -> Option<Version> {
482    if rows.len() <= cap {
483        return None;
484    }
485    let passed = rows.get(..cap)?;
486    if !passed
487        .iter()
488        .all(|row| floor_holds(floor, row_version(row)))
489    {
490        return None;
491    }
492    let generation = passed.last()?.key.generation;
493    (generation < below.generation).then(|| Version {
494        generation: generation.saturating_add(1),
495        sequence: 0,
496    })
497}
498
499/// Whether `floor` relies on `at` or a version before it in its generation.
500fn floor_holds(floor: &Floor, at: Version) -> bool {
501    let generation = at.generation;
502    floor
503        .oldest_in(generation)
504        .is_some_and(|oldest| order(at) >= order(oldest))
505}
506
507/// The rows a decision left behind names, read no further than its count:
508/// from its first version, at most `count` + 1 rows, those up to its last.
509/// A number found other than its count refuses by name, and the decision
510/// is left as it is: never guessed (HR4-fix2).
511fn decided_rows(
512    reader: &ReadOnlyDatabase,
513    service: &BranchRefRecord,
514    branch: &str,
515    decision: &Decision,
516    below: Below,
517) -> Result<Vec<Version>, Failure> {
518    let read = NonZeroUsize::new(decision.count.saturating_add(1)).unwrap_or(NonZeroUsize::MAX);
519    let rows = history::oldest_rows(
520        reader,
521        service,
522        branch,
523        (decision.first, below.version),
524        read,
525    )?;
526    let taken: Vec<Version> = rows
527        .iter()
528        .map(row_version)
529        .filter(|at| order(*at) <= order(decision.last))
530        .collect();
531    if taken.len() != decision.count {
532        return Err(Failure::new(
533            "release_decision_mismatch",
534            format!(
535                "the release decided for {branch} names {} rows from {}.{} to {}.{}, and {} \
536                 are there; the decision is left",
537                decision.count,
538                decision.first.generation,
539                decision.first.sequence,
540                decision.last.generation,
541                decision.last.sequence,
542                taken.len()
543            ),
544        ));
545    }
546    Ok(taken)
547}
548
549/// What keeps `rows[index]`: empty when it may go.
550fn held_at(
551    branch: &str,
552    rows: &[Row],
553    index: usize,
554    floor: &Floor,
555    below: Below,
556    rules: &Rules,
557) -> Vec<HeldBy> {
558    let Some(row) = rows.get(index) else {
559        return Vec::new();
560    };
561    let at = row_version(row);
562    if let Some(oldest) = floor.oldest_in(at.generation)
563        && order(at) >= order(oldest)
564    {
565        return floor
566            .held_by
567            .iter()
568            .filter(|held| held.version == oldest)
569            .cloned()
570            .collect();
571    }
572    let Some(keep) = rules.keep else {
573        return Vec::new();
574    };
575    let kept = |source| {
576        vec![HeldBy {
577            source,
578            name: branch.to_owned(),
579            version: at,
580        }]
581    };
582    let Some(versions) = keep.versions else {
583        return kept(Source::KeepVersions);
584    };
585    let Some(days) = keep.days else {
586        return kept(Source::KeepDays);
587    };
588    let head = below.version;
589    let newer = newer_than(rows, index, head);
590    if newer < versions.get() {
591        return kept(Source::KeepVersions);
592    }
593    // Superseded when the next version was written: the next row's
594    // timestamp, or the head's for the last row. Unknown is kept.
595    let superseded = match rows.get(index + 1) {
596        Some(next) => Some(next.timestamp),
597        None if order(at) < order(head) && rows.len() <= index + 1 => None,
598        None => Some(below.written),
599    };
600    let old_enough = superseded.is_some_and(|superseded| {
601        rules.now.saturating_sub(superseded) >= days.saturating_mul(NANOS_A_DAY)
602    });
603    if !old_enough {
604        return kept(Source::KeepDays);
605    }
606    Vec::new()
607}
608
609/// When the row the keep days hold falls due: the next row's timestamp, which
610/// is when it was superseded, plus the days. `None` when the days hold none,
611/// or hold the last row read, whose superseding time is unknown.
612fn due_at(rows: &[Row], held_by: &[HeldBy], rules: &Rules) -> Option<Timestamp> {
613    let days = rules.keep?.days?;
614    let held = held_by
615        .iter()
616        .find(|held| held.source == Source::KeepDays)?;
617    let index = rows
618        .iter()
619        .position(|row| row_version(row) == held.version)?;
620    let superseded = rows.get(index + 1)?.timestamp;
621    Some(superseded.saturating_add(days.saturating_mul(NANOS_A_DAY)))
622}
623
624/// How many versions are newer than `rows[index]`, the head among them, as
625/// far as the rows read show: never more than the truth, so a count short
626/// of a keep setting keeps.
627///
628/// A generation's versions run from sequence 0 up by one: a record is
629/// created at seq 0 (haematite `branch/lifecycle.rs`, `create_branch_record`;
630/// `branch/fork_changes.rs` for a fork) and each advance writes
631/// `expected_seq + 1` only over a record at `expected_seq`
632/// (`branch/refstore.rs`, `advance_with_source`). So in the head's
633/// generation exactly `head.sequence + 1` versions are newer than a row of
634/// an older generation, and `head.sequence - at.sequence` than a row of its
635/// own. Rows read after an older row are added only where their generation
636/// is older than the head's: the head generation's are already in
637/// `head.sequence + 1`. Versions of an older generation with no row read
638/// are not counted.
639fn newer_than(rows: &[Row], index: usize, head: Version) -> u64 {
640    let Some(row) = rows.get(index) else {
641        return 0;
642    };
643    let at = row_version(row);
644    if at.generation == head.generation {
645        return head.sequence.saturating_sub(at.sequence);
646    }
647    let older_after = rows
648        .get(index + 1..)
649        .unwrap_or_default()
650        .iter()
651        .filter(|later| later.key.generation < head.generation)
652        .count();
653    // Too few counted keeps; a count that cannot be made is none.
654    head.sequence
655        .saturating_add(1)
656        .saturating_add(u64::try_from(older_after).unwrap_or(0))
657}
658
659/// DR-11: an operator's release of a generation discarded before DR-1.
660///
661/// Such a generation has rows and no marker, so no release finds it; naming
662/// it writes its marker, dated by the request, and the ordinary rules then
663/// take its rows.
664#[derive(Debug, Clone, PartialEq, Eq)]
665pub struct UndatedGeneration {
666    pub branch: String,
667    /// The generation: its record's creation time, as its rows are keyed.
668    pub generation: u64,
669    /// Taken as the discard's time, so `keep.days` counts from it.
670    pub at: Timestamp,
671}
672
673/// The rows of one generation a dating request reads at a time.
674const UNDATED_PAGE: usize = 256;
675
676/// A named point in a release's durable order where a test stops it, as a
677/// crash would (STOR-003 HR3).
678#[cfg(test)]
679#[derive(Debug, Clone, Copy, PartialEq, Eq)]
680pub enum Crash {
681    /// The decision row is committed; nothing else is done.
682    AfterDecision,
683    /// Half the history files are removed.
684    DuringFiles,
685    /// Every history file is removed; the rows are not.
686    AfterFiles,
687    /// The rows, the mark and the decision are committed; no pass is asked.
688    AfterCommit,
689    /// The rows of the kept heads past their days are committed; the record
690    /// still keeps them (DR-7).
691    BeforeTrim,
692}
693
694#[cfg(test)]
695thread_local! {
696    static CRASH: std::cell::Cell<Option<Crash>> = const { std::cell::Cell::new(None) };
697}
698
699/// Stop this thread's next release at `point`.
700#[cfg(test)]
701pub fn crash_at(point: Crash) {
702    CRASH.with(|crash| crash.set(Some(point)));
703}
704
705#[cfg(test)]
706fn crashed(point: Crash) -> Result<(), Failure> {
707    if CRASH.with(|crash| crash.get() == Some(point)) {
708        CRASH.with(|crash| crash.set(None));
709        return Err(Failure::new("release_crashed", format!("{point:?}")));
710    }
711    Ok(())
712}
713
714impl Storage {
715    /// `history.release {branch, limit}` (STOR-003 HR3): takes the oldest
716    /// versions of `branch` that nothing relies on and the keep settings let
717    /// go, at most `limit`, in this durable order: one decision row; then the
718    /// older history files, in one ref operation; then the rows, the mark and
719    /// the decision's removal in one commit. A pass is then asked for. A
720    /// decision a crash left is finished here first, taking only what may
721    /// still go and saying what held the rest.
722    ///
723    /// Every version taken reads whole until that last commit: its row is
724    /// still in the service tree, and every reclaim pass marks from the
725    /// service's rows (STOR-003 R2), whether or not its older file is gone.
726    ///
727    /// No second release runs beside a decision's owner: this takes the
728    /// storage mutably, the service holds it behind one lock, and another
729    /// process is kept out by the writer lock. A call that meets a decision
730    /// left behind finishes that decision as it was written, asking no keep
731    /// setting, hold or limit again (HR4-fix2), and decides nothing new.
732    ///
733    /// # Errors
734    /// `release_limit_exceeded` for a limit above `rules.max_rows`; whatever
735    /// the floor, a read or a commit answers.
736    pub fn release(
737        &mut self,
738        branch: &str,
739        limit: Option<NonZeroUsize>,
740        rules: &Rules,
741        discards: &dyn PendingDiscards,
742    ) -> Result<Answer, Failure> {
743        let answer = self.release_once(branch, limit, rules, discards, None);
744        // Ended, whatever it took and whether or not it failed: the driver's
745        // count for the branch starts again, so neither a branch its settings
746        // keep whole nor one whose release fails is asked again until
747        // another batch of advances (STOR-003 HR4, HR4-fix).
748        self.kept_rows.released(branch);
749        answer
750    }
751
752    /// Every live branch, for the driver to ask once at start (STOR-003
753    /// HR4-fix): the records in memory, no read.
754    #[must_use]
755    pub fn branch_names(&self) -> Vec<String> {
756        self.refs
757            .list_prefix(BRANCH_PREFIX)
758            .filter_map(|record| record.name.strip_prefix(BRANCH_PREFIX))
759            .map(str::to_owned)
760            .collect()
761    }
762
763    /// The branches whose release a crash left decided and unfinished:
764    /// one range read of the decision rows, never the history (STOR-003
765    /// HR4). A decision whose branch has no live record is not named.
766    ///
767    /// # Errors
768    /// Storage failures.
769    pub fn decided_branches(&self) -> Result<Vec<String>, Failure> {
770        let service = self
771            .refs
772            .get(SERVICE)
773            .ok_or_else(|| Failure::new("reference_missing", "missing service namespace"))?;
774        let decided = history::decided(&self.reader, service)?;
775        Ok(self
776            .refs
777            .list_prefix(BRANCH_PREFIX)
778            .filter_map(|record| record.name.strip_prefix(BRANCH_PREFIX))
779            .filter(|name| decided.contains(&Hash::of(name.as_bytes())))
780            .map(str::to_owned)
781            .collect())
782    }
783
784    /// `branch`'s advances since its last release (STOR-003 HR4).
785    #[must_use]
786    pub fn advanced_since_release(&self, branch: &str) -> usize {
787        self.kept_rows.advanced(branch)
788    }
789
790    /// The durable operations the store has made, for the release driver to
791    /// count its own by (STOR-003 HR4).
792    #[must_use]
793    pub fn sync_counts(&self) -> SyncCounts {
794        self.writer.sync_counts()
795    }
796
797    /// [`Self::release`] of `branch`'s discarded generation `generation`
798    /// alone, a name with no record: what a discarded line's release asks
799    /// (STOR-050 R2), so another generation of the name is never taken by
800    /// rules asked for this one.
801    ///
802    /// # Errors
803    /// As [`Self::release`].
804    pub(super) fn release_generation(
805        &mut self,
806        branch: &str,
807        generation: u64,
808        rules: &Rules,
809    ) -> Result<Answer, Failure> {
810        let answer = self.release_once(branch, None, rules, &IntentListing, Some(generation));
811        self.kept_rows.released(branch);
812        answer
813    }
814
815    /// What a release of the live line `branch` answers: nothing taken, the
816    /// line holding every version (STOR-050 R2).
817    fn line_holds(&self, branch: &str) -> Result<Option<Answer>, Failure> {
818        let Some(record) = self.refs.get(&branch_name(branch)) else {
819            return Ok(None);
820        };
821        let head = Captured {
822            branch: branch.to_owned(),
823            version: version(record),
824            shards: record.shards.clone(),
825        };
826        if !super::lineage::is_line(&self.reader, &head)? {
827            return Ok(None);
828        }
829        Ok(Some(Answer {
830            branch: branch.to_owned(),
831            released: None,
832            marker: None,
833            held_by: vec![HeldBy {
834                source: Source::Line,
835                name: branch.to_owned(),
836                version: head.version,
837            }],
838            more: false,
839            branches_read: 0,
840            pass_asked: false,
841            pass_refused: None,
842            due_at: None,
843        }))
844    }
845
846    fn release_once(
847        &mut self,
848        branch: &str,
849        limit: Option<NonZeroUsize>,
850        rules: &Rules,
851        discards: &dyn PendingDiscards,
852        only: Option<u64>,
853    ) -> Result<Answer, Failure> {
854        let limit = match limit {
855            Some(limit) if limit > rules.max_rows => {
856                return Err(Failure::new(
857                    "release_limit_exceeded",
858                    format!(
859                        "limit {limit} is above retention.release_max_rows {}",
860                        rules.max_rows
861                    ),
862                ));
863            }
864            Some(limit) => limit,
865            None => rules.max_rows,
866        };
867        let service = self
868            .refs
869            .get(SERVICE)
870            .cloned()
871            .ok_or_else(|| Failure::new("reference_missing", "missing service namespace"))?;
872        let live = self.refs.get(&branch_name(branch)).is_some();
873        let pending = history::read_decision(&self.reader, &service, branch)?;
874        if pending.is_none()
875            && live
876            && let Some(held) = self.line_holds(branch)?
877        {
878            return Ok(held);
879        }
880        // A decision left behind is the commit point: it is rolled forward
881        // as written, and no keep setting, hold or limit is asked again
882        // (Apollo 14:24, HR4-fix2). Its files may already be gone.
883        let floor = match (&pending, live) {
884            (Some(_decision), _) => None,
885            (None, true) => Some(self.relied_on(branch, discards)?),
886            (None, false) => Some(self.relied_on_by_others(branch, false, discards)?),
887        };
888        let (below, mark) = self.below(
889            &service,
890            branch,
891            (pending.as_ref(), only),
892            floor.as_ref(),
893            limit,
894        )?;
895        let (taken, held_by, more, branches_read, due) = if let Some(decision) = &pending {
896            let taken = decided_rows(&self.reader, &service, branch, decision, below)?;
897            (taken, Vec::new(), false, 0, None)
898        } else {
899            let floor = floor.unwrap_or_default();
900            let from = if live {
901                LOWEST
902            } else {
903                past_mark(mark, below.version)
904            };
905            let read = ReadFrom {
906                from,
907                below,
908                limit,
909                live,
910            };
911            let (rows, (taken, held_by, more)) =
912                self.judge_from(&service, branch, read, &floor, rules)?;
913            let mut due = due_at(&rows, &held_by, rules);
914            if live {
915                let trim_due = self.trim_past_days(branch, rules)?;
916                due = due.into_iter().chain(trim_due).min();
917            }
918            (taken, held_by, more, floor.branches_read, due)
919        };
920        let mut answer = Answer {
921            branch: branch.to_owned(),
922            released: None,
923            marker: mark,
924            held_by,
925            more,
926            branches_read,
927            pass_asked: false,
928            pass_refused: None,
929            due_at: due,
930        };
931        let span = match (taken.first(), taken.last()) {
932            (Some(first), Some(last)) => Some(Span {
933                first: *first,
934                last: *last,
935            }),
936            _ => None,
937        };
938        if pending.is_none() {
939            let Some(span) = span else {
940                return Ok(answer);
941            };
942            let decision = Decision {
943                first: span.first,
944                last: span.last,
945                count: taken.len(),
946                at: rules.now,
947            };
948            let bytes = serde_json::to_vec(&decision).map_err(Failure::storage)?;
949            self.commit_service(&[(history::decision_key(branch), Some(bytes))])?;
950            #[cfg(test)]
951            crashed(Crash::AfterDecision)?;
952        }
953        let ended = dead::ended(&self.reader, &service, branch, &taken)?;
954        self.carry_out(branch, &taken, span, rules.now, &ended, &mut answer)?;
955        Ok(answer)
956    }
957
958    /// Judge `branch`'s oldest rows from `read.from`, at most `limit` + 1 a
959    /// read (DR-13). When the first `limit` rows are all relied on by
960    /// `floor`, `judge` passes them and takes nothing, and the next release
961    /// would read the same ones. The rest of the last passed row's generation
962    /// is relied on as well (`floor_holds` is by order within a generation),
963    /// so the read starts again at the next generation. Rows are keyed by
964    /// generation, then sequence, so that start is a key, not a walk: one
965    /// more read for each generation the floor holds, and at most `limit`.
966    ///
967    /// # Errors
968    /// As [`history::oldest_rows`] and [`Self::after_newest`].
969    fn judge_from(
970        &self,
971        service: &BranchRefRecord,
972        branch: &str,
973        mut read: ReadFrom,
974        floor: &Floor,
975        rules: &Rules,
976    ) -> Result<(Vec<Row>, Judged), Failure> {
977        let cap = read.limit.get();
978        let ask = NonZeroUsize::new(cap.saturating_add(1)).unwrap_or(NonZeroUsize::MAX);
979        let mut skips = 0;
980        loop {
981            let span = (read.from, read.below.version);
982            let mut rows = history::oldest_rows(&self.reader, service, branch, span, ask)?;
983            if rows.len() <= cap {
984                let newest = rows.last();
985                rows.extend(self.after_newest(service, branch, newest, read.below, read.live)?);
986            }
987            let judged = judge(branch, &rows, cap, floor, read.below, rules);
988            match past_held(&rows, cap, floor, read.below.version) {
989                Some(from) if skips < cap => {
990                    read.from = from;
991                    skips += 1;
992                }
993                _ => return Ok((rows, judged)),
994            }
995        }
996    }
997
998    /// A read that reached `below` knows when its `newest` row was superseded:
999    /// a discarded generation's at its discard, a live branch's when the
1000    /// record's next version was installed, the first it keeps after `newest`
1001    /// or else the head; no read
1002    /// (docs/design/HAEM-STALLED-BRANCH-NEWEST-ROW-2026-10-06.md). It is a row
1003    /// at `below`, which a release never takes. A live row of an older
1004    /// generation was superseded at that generation's discard, read from its
1005    /// marker, one entry on each shard (DR-6); with no marker it is unknown,
1006    /// so kept.
1007    ///
1008    /// # Errors
1009    /// As [`dead::marker`].
1010    fn after_newest(
1011        &self,
1012        service: &BranchRefRecord,
1013        branch: &str,
1014        newest: Option<&Row>,
1015        below: Below,
1016        live: bool,
1017    ) -> Result<Option<Row>, Failure> {
1018        let written = match newest {
1019            Some(newest) if live && newest.key.generation != below.version.generation => {
1020                let generation = newest.key.generation;
1021                match dead::marker(&self.reader, service, branch, generation)? {
1022                    Some(dead) => dead.discarded_at,
1023                    None => return Ok(None),
1024                }
1025            }
1026            Some(newest) if live => self
1027                .refs
1028                .kept(&branch_name(branch))
1029                .iter()
1030                .find(|kept| kept.seq > newest.key.sequence)
1031                .map_or(below.written, |kept| kept.timestamp),
1032            _ => below.written,
1033        };
1034        Ok(Some(dead::discard_row(branch, below.version, written)))
1035    }
1036
1037    /// Where `branch`'s release stops, and the mark of that generation. A
1038    /// name with no record releases one discarded generation, the discard
1039    /// counted as the version that superseded its last (DR-6): of its
1040    /// markers, at most `limit` and oldest first, the first with rows left
1041    /// past its mark that `floor` does not hold; the oldest when there is
1042    /// none. A decision left behind keeps the generation it was made for.
1043    /// One or two point reads of a mark for each generation looked at.
1044    /// With `only` and no decision left behind, the generation named is the
1045    /// only one looked at (STOR-050 R2).
1046    fn below(
1047        &self,
1048        service: &BranchRefRecord,
1049        branch: &str,
1050        (pending, only): (Option<&Decision>, Option<u64>),
1051        floor: Option<&Floor>,
1052        limit: NonZeroUsize,
1053    ) -> Result<(Below, Option<Released>), Failure> {
1054        if let Some(record) = self.refs.get(&branch_name(branch)) {
1055            let below = Below {
1056                version: version(record),
1057                written: record.timestamp,
1058            };
1059            let generation = below.version.generation;
1060            let mark = history::read_released(&self.reader, service, branch, generation)?;
1061            return Ok((below, mark));
1062        }
1063        let mut markers = dead::of_name(&self.reader, service, branch, limit)?;
1064        if let (None, Some(generation)) = (pending, only) {
1065            markers.retain(|dead| dead.last.generation == generation);
1066        }
1067        if let (None, Some(floor)) = (pending, floor)
1068            && let Some(unheld) = self.first_unheld(service, branch, &markers, floor)?
1069        {
1070            return Ok(unheld);
1071        }
1072        let dead = pending
1073            .and_then(|decision| {
1074                markers
1075                    .iter()
1076                    .find(|dead| dead.last.generation == decision.last.generation)
1077            })
1078            .or_else(|| markers.first())
1079            .ok_or_else(|| Failure::new("branch_missing", branch))?;
1080        let generation = dead.last.generation;
1081        let mark = history::read_released(&self.reader, service, branch, generation)?;
1082        Ok((ending(dead), mark))
1083    }
1084
1085    /// The first of `markers` with rows left past its generation's mark whose
1086    /// next row is below what `floor` relies on in that generation, or which
1087    /// nothing relies on, with its mark: one or two point reads each. A
1088    /// generation a mark from before DR-6 already covers has none left
1089    /// (DR-6b).
1090    fn first_unheld(
1091        &self,
1092        service: &BranchRefRecord,
1093        branch: &str,
1094        markers: &[dead::Dead],
1095        floor: &Floor,
1096    ) -> Result<Option<(Below, Option<Released>)>, Failure> {
1097        for dead in markers {
1098            let below = ending(dead);
1099            let generation = dead.last.generation;
1100            let mark = history::read_released(&self.reader, service, branch, generation)?;
1101            let from = past_mark(mark, below.version);
1102            if order(from) >= order(below.version) {
1103                continue;
1104            }
1105            let held = floor
1106                .oldest_in(generation)
1107                .is_some_and(|relied| order(relied) <= order(from));
1108            if !held {
1109                return Ok(Some((below, mark)));
1110            }
1111        }
1112        Ok(None)
1113    }
1114
1115    /// DR-11: writes the marker `request` names in one service commit: its
1116    /// last version the generation's newest row, discarded at `request.at`.
1117    /// Refused by name: `branch_live` while the name has a record (its
1118    /// release reads every generation), `generation_marked` when the
1119    /// generation has its marker, `generation_empty` when it has no rows.
1120    /// Reads the generation's rows once, a page at a time: an operator's one
1121    /// request, never a release's wake. Answers the marker's last version.
1122    ///
1123    /// # Errors
1124    /// As above; `history_row_corrupt`, `dead_generation_corrupt`; storage
1125    /// failures.
1126    pub fn date_undated_generation(
1127        &mut self,
1128        request: &UndatedGeneration,
1129    ) -> Result<Version, Failure> {
1130        let branch = request.branch.as_str();
1131        if self.refs.get(&branch_name(branch)).is_some() {
1132            return Err(Failure::new(
1133                "branch_live",
1134                format!("{branch} has a record"),
1135            ));
1136        }
1137        let service = self
1138            .refs
1139            .get(SERVICE)
1140            .cloned()
1141            .ok_or_else(|| Failure::new("reference_missing", "missing service namespace"))?;
1142        let generation = request.generation;
1143        if dead::marker(&self.reader, &service, branch, generation)?.is_some() {
1144            return Err(Failure::new(
1145                "generation_marked",
1146                format!("{branch} generation {generation} has its marker"),
1147            ));
1148        }
1149        let newest = newest_row(&self.reader, &service, branch, generation)?;
1150        let Some(last) = newest else {
1151            return Err(Failure::new(
1152                "generation_empty",
1153                format!("{branch} generation {generation} has no rows"),
1154            ));
1155        };
1156        let dead = dead::Dead {
1157            branch: branch.to_owned(),
1158            last,
1159            discarded_at: request.at,
1160        };
1161        let bytes = serde_json::to_vec(&dead).map_err(Failure::storage)?;
1162        self.commit_service(&[(dead::dead_key(branch, generation), Some(bytes))])?;
1163        Ok(last)
1164    }
1165
1166    /// DR-7 (candidate B, docs/design/HAEM-RELEASE-REMAINING-RULES-2026-10-06.md):
1167    /// the heads `branch`'s record keeps beyond its newest `keep.versions`
1168    /// whose next version was installed `keep.days` ago leave the record, in
1169    /// two durable operations in this order: their rows, in one service
1170    /// commit, as an advance writes them for the heads it drops; then one
1171    /// install of the record keeping the rest (`trim_kept`). A crash between
1172    /// the two leaves the rows and the heads still kept: the next release
1173    /// finds the rows, writes none, and installs the trim. The rows are then
1174    /// released by the same rules as any other, from the next release on.
1175    /// Nothing is written while no head is due, and nothing without both
1176    /// settings. Answers when the next kept head falls due, for the releaser
1177    /// to wake then: no advance wakes a branch that stopped.
1178    fn trim_past_days(
1179        &mut self,
1180        branch: &str,
1181        rules: &Rules,
1182    ) -> Result<Option<Timestamp>, Failure> {
1183        let Some((versions, days)) = rules.keep.and_then(|keep| keep.versions.zip(keep.days))
1184        else {
1185            return Ok(None);
1186        };
1187        let name = branch_name(branch);
1188        let Some(record) = self.refs.get(&name).cloned() else {
1189            return Ok(None);
1190        };
1191        let kept = self.refs.kept(&name);
1192        if let Some(keep_from) = past_days(&record, kept, versions, days, rules.now) {
1193            self.rows_before_trim(branch, &record, keep_from)?;
1194            #[cfg(test)]
1195            crashed(Crash::BeforeTrim)?;
1196            self.refs
1197                .trim_kept(&name, record.created, record.seq, keep_from)
1198                .map_err(|error| Failure::new("release_kept_untrimmed", error.to_string()))?;
1199            // The catalogue's readers see the record keep fewer from here on.
1200            self.publish_record(&name)?;
1201        }
1202        Ok(kept_due(&record, self.refs.kept(&name), versions, days))
1203    }
1204
1205    /// A decided release's last two durable operations, in order: the files
1206    /// of `taken` (absent ones included), then its rows, mark and decision in
1207    /// one commit; then the reclaim pass asked for, a refusal named in
1208    /// `answer`, never a failure of the release.
1209    fn carry_out(
1210        &mut self,
1211        branch: &str,
1212        taken: &[Version],
1213        span: Option<Span>,
1214        now: Timestamp,
1215        ended: &[Vec<u8>],
1216        answer: &mut Answer,
1217    ) -> Result<(), Failure> {
1218        let files: Vec<String> = taken
1219            .iter()
1220            .map(|at| history::file_name(branch, *at))
1221            .collect();
1222        #[cfg(test)]
1223        if CRASH.with(|crash| crash.get() == Some(Crash::DuringFiles)) {
1224            let half = files.len() / 2;
1225            self.remove_files(files.get(..half).unwrap_or_default())?;
1226            crashed(Crash::DuringFiles)?;
1227        }
1228        self.remove_files(&files)?;
1229        #[cfg(test)]
1230        crashed(Crash::AfterFiles)?;
1231        let mut writes: Vec<(Vec<u8>, Option<Vec<u8>>)> = taken
1232            .iter()
1233            .map(|at| (history::row_key(branch, *at), None))
1234            .collect();
1235        writes.push((history::decision_key(branch), None));
1236        // A discarded generation's marker goes with its last row.
1237        writes.extend(ended.iter().map(|key| (key.clone(), None)));
1238        // DR-6: each generation taken is marked at its last version taken,
1239        // in the same commit as its rows; the name's old mark is not written.
1240        for last in last_of_each_generation(taken) {
1241            let released = Released {
1242                version: last,
1243                at: now,
1244            };
1245            let bytes = serde_json::to_vec(&released).map_err(Failure::storage)?;
1246            writes.push((
1247                history::released_key_at(branch, last.generation),
1248                Some(bytes),
1249            ));
1250        }
1251        if let Some(span) = span {
1252            answer.marker = Some(Released {
1253                version: span.last,
1254                at: now,
1255            });
1256        }
1257        self.commit_service(&writes)?;
1258        answer.released = span;
1259        #[cfg(test)]
1260        crashed(Crash::AfterCommit)?;
1261        // The release is durable by now: a pass refused is said by name in
1262        // the answer, never a failure of the release.
1263        match self.writer.request_reclaim() {
1264            Ok(()) => answer.pass_asked = true,
1265            Err(error) => answer.pass_refused = Some(error.to_string()),
1266        }
1267        Ok(())
1268    }
1269
1270    /// Remove the older history files `names` in one ref operation, absent
1271    /// ones included.
1272    fn remove_files(&mut self, names: &[String]) -> Result<(), Failure> {
1273        let removed = self
1274            .refs
1275            .remove_set(names.iter().map(String::as_str))
1276            .map_err(|error| Failure::new("release_files_unremoved", error.to_string()))?;
1277        for record in removed.removed {
1278            self.visible.remove(&record.name)?;
1279        }
1280        Ok(())
1281    }
1282
1283    /// One durable service commit of `writes`: a value puts, `None` deletes.
1284    /// A commit that failed may or may not have installed, so what the kept
1285    /// rows know is checked again, as for every service commit of rows.
1286    fn commit_service(&mut self, writes: &[(Vec<u8>, Option<Vec<u8>>)]) -> Result<(), Failure> {
1287        let committed = self.try_commit_service(writes);
1288        if committed.is_err() {
1289            self.kept_rows.mark_unsure();
1290        }
1291        committed
1292    }
1293
1294    fn try_commit_service(&mut self, writes: &[(Vec<u8>, Option<Vec<u8>>)]) -> Result<(), Failure> {
1295        let at = self
1296            .refs
1297            .get(SERVICE)
1298            .map(version)
1299            .ok_or_else(|| Failure::new("reference_missing", "missing service namespace"))?;
1300        let service =
1301            open_branch(SERVICE, &self.refs, &self.registry).map_err(Failure::branch_commit)?;
1302        for (key, value) in writes {
1303            let shard = self.writer.shard_for(key);
1304            value
1305                .as_ref()
1306                .map_or_else(
1307                    || service.delete(shard, key),
1308                    |value| service.put(shard, key, value),
1309                )
1310                .map_err(Failure::branch)?;
1311        }
1312        self.writer
1313            .commit(
1314                &service,
1315                &self.registry,
1316                CommitRequest {
1317                    durability: CommitDurability::Durable {
1318                        refs: &mut self.refs,
1319                    },
1320                    extra_parents: &[],
1321                    timestamp: haematite::branch::current_timestamp(),
1322                    note: None,
1323                },
1324            )
1325            .map_err(|error| {
1326                let installed = self.refs.get(SERVICE).is_some_and(|now| version(now) != at);
1327                Failure::new(
1328                    if installed {
1329                        "durability_unknown"
1330                    } else {
1331                        "write_refused"
1332                    },
1333                    error.to_string(),
1334                )
1335            })?;
1336        self.publish_record(SERVICE)
1337    }
1338}
1339
1340/// Where a discarded generation's release starts, the release stopping at
1341/// `below` (DR-6): past its own mark, else at its first version. A mark from
1342/// before DR-6 at a later generation covers this one whole, so the answer is
1343/// `below` and the range is empty, never inverted (DR-6b).
1344const fn past_mark(mark: Option<Released>, below: Version) -> Version {
1345    match mark {
1346        Some(mark) if mark.version.generation > below.generation => below,
1347        Some(mark) => after(mark),
1348        None => Version {
1349            generation: below.generation,
1350            sequence: 0,
1351        },
1352    }
1353}
1354
1355/// The newest row of `branch`'s `generation`, read oldest first a page at a
1356/// time (DR-11).
1357fn newest_row(
1358    reader: &ReadOnlyDatabase,
1359    service: &BranchRefRecord,
1360    branch: &str,
1361    generation: u64,
1362) -> Result<Option<Version>, Failure> {
1363    let page = NonZeroUsize::new(UNDATED_PAGE).unwrap_or(NonZeroUsize::MIN);
1364    let end = Version {
1365        generation,
1366        sequence: u64::MAX,
1367    };
1368    let mut from = Version {
1369        generation,
1370        sequence: 0,
1371    };
1372    let mut newest = None;
1373    loop {
1374        let rows = history::oldest_rows(reader, service, branch, (from, end), page)?;
1375        let Some(last) = rows.last().map(row_version) else {
1376            return Ok(newest);
1377        };
1378        newest = Some(last);
1379        if rows.len() < page.get() {
1380            return Ok(newest);
1381        }
1382        from = Version {
1383            generation,
1384            sequence: last.sequence.saturating_add(1),
1385        };
1386    }
1387}
1388
1389/// The first seq `record` keeps on once DR-7's trim at `now` is done: the
1390/// kept heads below it are beyond the newest `versions`, the head counted,
1391/// and were superseded `days` ago or more, by the next kept head's install
1392/// or, for the newest, the head's. `None` when no head is.
1393fn past_days(
1394    record: &BranchRefRecord,
1395    kept: &[KeptHead],
1396    versions: NonZeroU64,
1397    days: u64,
1398    now: Timestamp,
1399) -> Option<u64> {
1400    let age = days.saturating_mul(NANOS_A_DAY);
1401    let superseded = kept
1402        .iter()
1403        .skip(1)
1404        .map(|next| next.timestamp)
1405        .chain(std::iter::once(record.timestamp));
1406    let leaving = kept
1407        .iter()
1408        .zip(superseded)
1409        .take_while(|(entry, superseded)| {
1410            record.seq.saturating_sub(entry.seq) >= versions.get()
1411                && now.saturating_sub(*superseded) >= age
1412        })
1413        .count();
1414    let keep_from = kept.get(leaving).map_or(record.seq, |entry| entry.seq);
1415    (leaving > 0).then_some(keep_from)
1416}
1417
1418/// When the oldest head `record` keeps beyond its newest `versions` falls
1419/// due under `days`: its next version's install, plus the days.
1420fn kept_due(
1421    record: &BranchRefRecord,
1422    kept: &[KeptHead],
1423    versions: NonZeroU64,
1424    days: u64,
1425) -> Option<Timestamp> {
1426    let index = kept
1427        .iter()
1428        .position(|entry| record.seq.saturating_sub(entry.seq) >= versions.get())?;
1429    let superseded = kept
1430        .get(index + 1)
1431        .map_or(record.timestamp, |next| next.timestamp);
1432    Some(superseded.saturating_add(days.saturating_mul(NANOS_A_DAY)))
1433}
1434
1435/// Where a release of `dead`'s generation stops: past its last version,
1436/// superseded when it was discarded.
1437const fn ending(dead: &dead::Dead) -> Below {
1438    Below {
1439        version: Version {
1440            generation: dead.last.generation,
1441            sequence: dead.last.sequence.saturating_add(1),
1442        },
1443        written: dead.discarded_at,
1444    }
1445}
1446
1447/// The last version of each generation in `taken`, which runs oldest first.
1448fn last_of_each_generation(taken: &[Version]) -> Vec<Version> {
1449    let mut last: Vec<Version> = Vec::new();
1450    for at in taken {
1451        match last.last_mut() {
1452            Some(previous) if previous.generation == at.generation => *previous = *at,
1453            _ => last.push(*at),
1454        }
1455    }
1456    last
1457}
1458
1459/// The version just after `released`, where the next release starts.
1460const fn after(released: Released) -> Version {
1461    Version {
1462        generation: released.version.generation,
1463        sequence: released.version.sequence.saturating_add(1),
1464    }
1465}
1466
1467#[cfg(test)]
1468#[path = "release_tests.rs"]
1469mod tests;