1use 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#[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 Embed,
43 EmbedOf,
46 KeepVersions,
48 KeepDays,
50 Line,
53}
54
55#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
57pub struct HeldBy {
58 pub source: Source,
59 pub name: String,
61 pub version: Version,
62}
63
64#[derive(Debug, Clone, Default, PartialEq, Eq)]
66pub struct Floor {
67 pub held_by: Vec<HeldBy>,
71 pub branches_read: usize,
73}
74
75impl Floor {
76 pub fn oldest(&self) -> Option<&HeldBy> {
80 self.held_by.first()
81 }
82
83 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
95pub 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
105pub trait PendingDiscards {
107 fn pending(&self, storage: &Storage) -> Result<Vec<(String, Version)>, Failure>;
110}
111
112#[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 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 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 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 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 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 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 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
316fn 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
332const POSITIONS_A_PAGE: usize = 64;
334
335#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
339pub struct Keep {
340 pub versions: Option<NonZeroU64>,
342 pub days: Option<u64>,
344}
345
346#[derive(Debug, Clone, Copy, PartialEq, Eq)]
348pub struct Rules {
349 pub keep: Option<Keep>,
353 pub max_rows: NonZeroUsize,
355 pub now: Timestamp,
357}
358
359#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
361pub struct Span {
362 pub first: Version,
363 pub last: Version,
364}
365
366#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
368pub struct Answer {
369 pub branch: String,
370 pub released: Option<Span>,
371 pub marker: Option<Released>,
372 pub held_by: Vec<HeldBy>,
374 pub more: bool,
376 pub branches_read: usize,
377 pub pass_asked: bool,
379 pub pass_refused: Option<String>,
381 pub due_at: Option<Timestamp>,
385}
386
387const NANOS_A_DAY: u64 = 86_400 * 1_000_000_000;
388
389const 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#[derive(Debug, Clone, Copy, PartialEq, Eq)]
410pub struct Below {
411 pub version: Version,
412 pub written: Timestamp,
413}
414
415fn 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 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#[derive(Debug, Clone, Copy)]
468struct ReadFrom {
469 from: Version,
470 below: Below,
471 limit: NonZeroUsize,
472 live: bool,
473}
474
475type Judged = (Vec<Version>, Vec<HeldBy>, bool);
477
478fn 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
499fn 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
507fn 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
549fn 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 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
609fn 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
624fn 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 head.sequence
655 .saturating_add(1)
656 .saturating_add(u64::try_from(older_after).unwrap_or(0))
657}
658
659#[derive(Debug, Clone, PartialEq, Eq)]
665pub struct UndatedGeneration {
666 pub branch: String,
667 pub generation: u64,
669 pub at: Timestamp,
671}
672
673const UNDATED_PAGE: usize = 256;
675
676#[cfg(test)]
679#[derive(Debug, Clone, Copy, PartialEq, Eq)]
680pub enum Crash {
681 AfterDecision,
683 DuringFiles,
685 AfterFiles,
687 AfterCommit,
689 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#[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 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 self.kept_rows.released(branch);
749 answer
750 }
751
752 #[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 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 #[must_use]
786 pub fn advanced_since_release(&self, branch: &str) -> usize {
787 self.kept_rows.advanced(branch)
788 }
789
790 #[must_use]
793 pub fn sync_counts(&self) -> SyncCounts {
794 self.writer.sync_counts()
795 }
796
797 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 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 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 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 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 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 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 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 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 self.publish_record(&name)?;
1201 }
1202 Ok(kept_due(&record, self.refs.kept(&name), versions, days))
1203 }
1204
1205 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 writes.extend(ended.iter().map(|key| (key.clone(), None)));
1238 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 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 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 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
1340const 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
1355fn 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
1389fn 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
1418fn 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
1435const 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
1447fn 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
1459const 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;