Skip to main content

hyperfile_cleaner/
compact.rs

1//! Offline compaction: rewriting the live blocks of mostly-dead objects so the
2//! objects can be reclaimed, and so a read of one file range costs fewer
3//! requests.
4//!
5//! ## What `prune` cannot do
6//!
7//! [`Cleaner::prune`](crate::Cleaner::prune) reclaims an object nothing points
8//! into. An object that is 5% live is not that object: those few blocks hold the
9//! file's only copy of their contents, so the other 95% cannot be given back
10//! while they stay there. Rewriting them into a new checkpoint is what turns the
11//! old object into one `prune` can take.
12//!
13//! The same rewrite is what makes a read cheaper. A request breaks where the
14//! object changes, so blocks of one file range scattered across many objects
15//! cost one request each; written out together they cost one. Both goals use the
16//! same mechanism and differ only in what they select.
17//!
18//! ## How the rewrite happens
19//!
20//! Not by a format this crate writes. Nothing here builds a segment, assigns a
21//! pointer or publishes an inode — that is `hyperfile`'s write path, and a second
22//! implementation of it would drift from the first. What this does instead is ask
23//! `hyperfile` to rewrite blocks it already knows how to rewrite:
24//!
25//! - [`Hyper::fs_relocate`] marks a run of blocks to be rewritten at the next flush
26//!   with their current bytes. It reads them with the read path's own coalescing,
27//!   so a contiguous run costs a few large requests rather than one per block; it
28//!   skips holes, explicit zero blocks and anything already dirty, so a sparse
29//!   file stays sparse; and it touches neither `mtime`, `ctime` nor the read cache.
30//! - One `fs_flush` at the end of the pass writes the dirty set as one checkpoint.
31//!   The dirty set is keyed by block index, so the blocks land in file order and
32//!   packed — which is both halves of what compaction is for.
33//!
34//! ## A compaction is not a change of content
35//!
36//! The pass has to publish a new checkpoint — blocks moved, so pointers, the map
37//! and the inode changed — and so `last_cno` advances. A consumer that reads a new
38//! `last_cno` as "the content changed" would take every compaction for a change.
39//! `hyperfile` therefore marks a publish whose every dirty block came from
40//! relocation with `InodeRaw::FLAG_COMPACTED`, and
41//! [`Hyper::fs_content_cno`] stays where it was across it. That is the number a
42//! consumer should compare.
43//!
44//! The mark is only set when the publish's number directly follows the previous
45//! one, so it holds for exactly one checkpoint. Two passes back to back would
46//! therefore move `content_cno` on the second, and [`S3Cleaner::compact`] refuses
47//! to start on a checkpoint that is already marked unless
48//! [`CompactOptions::allow_consecutive`] says otherwise.
49//!
50//! ## Why offline
51//!
52//! A compactor is a writer, and it writes *old* bytes — the content it read
53//! earlier. `hyperfile`'s default conflict policy, `RetryLastWriterWins`,
54//! refreshes from storage on a conflict and re-publishes the dirty set, which for
55//! a compactor means republishing a stale copy of a block another writer has
56//! since changed. So compaction runs with
57//! [`FlushConflictPolicy::FailFast`](hyperfile::config::FlushConflictPolicy::FailFast)
58//! and the caller must be the container's only writer. A concurrent publish is
59//! then detected — the inode carries an `If-Match` on its ETag — and the pass
60//! fails rather than losing the other writer's data.
61//!
62//! A log is the other half of that contract. This refuses to run when the
63//! container's prefix holds keys that are neither the inode nor segment objects,
64//! because a WAL is the usual reason for them and a compactor that does not read
65//! the log would publish over entries recovery has not replayed yet.
66
67use std::io::{Error, ErrorKind, Result};
68use std::collections::{BTreeMap, BTreeSet};
69use log::{debug, info, warn};
70use hyperfile::{BlockIndex, Cno, SegmentId};
71use hyperfile::config::{FlushConflictPolicy, HyperFileConfig, HyperFileConfigBuilder,
72    HyperFileRuntimeConfig};
73use hyperfile::file::flags::{FileFlags, HyperFileFlags};
74use hyperfile::file::hyper::Hyper;
75use hyperfile::meta_format::BlockPtrFormat;
76use hyperfile::staging::config::StagingConfig;
77use crate::s3::{ContainerView, S3Cleaner};
78
79/// How many blocks to ask `block_placement` about at a time. One entry per block
80/// comes back, so this bounds the answer rather than the work: the sweep covers
81/// the whole file either way.
82const PLACEMENT_CHUNK_BLOCKS: usize = 64 * 1024;
83
84/// How much of one stored object is still pointed into.
85#[derive(Clone, Debug, PartialEq, Eq)]
86pub struct ObjectUsage {
87    /// The object, `<cno>` or `<cno>.<part>`.
88    pub segid: SegmentId,
89    /// Its size on storage.
90    pub bytes: u64,
91    /// Live data blocks in it.
92    pub live_data_blocks: usize,
93    /// Live metadata nodes in it.
94    pub live_meta_blocks: usize,
95    /// What those occupy, by the container's own block sizes.
96    pub live_bytes: u64,
97}
98
99impl ObjectUsage {
100    /// Nothing points into it: `prune` can take it, compaction has no work here.
101    pub fn is_dead(&self) -> bool {
102        self.live_data_blocks == 0 && self.live_meta_blocks == 0
103    }
104
105    /// Live bytes over stored bytes. Zero for a dead object, and the cost side
106    /// of the trade: rewriting this much frees [`Self::bytes`].
107    pub fn live_ratio(&self) -> f64 {
108        if self.bytes == 0 {
109            return 0.0;
110        }
111        self.live_bytes as f64 / self.bytes as f64
112    }
113
114    /// Whether a live metadata node is in it.
115    ///
116    /// Such an object cannot be freed by rewriting data blocks: a node is
117    /// relocated when the flush rewrites it, which happens because something
118    /// under it was dirtied, and which node covers which keys is not visible from
119    /// outside the map. Compaction skips these rather than rewriting their data
120    /// blocks for nothing.
121    pub fn holds_metadata(&self) -> bool {
122        self.live_meta_blocks > 0
123    }
124}
125
126/// What a container's storage is made of, and what could be given back.
127#[derive(Clone, Debug)]
128pub struct Report {
129    /// The checkpoint the walk was taken from.
130    pub high_cno: Cno,
131    /// Whether the newest published checkpoint is itself a compaction
132    /// (`InodeRaw::FLAG_COMPACTED`). `compact` will not run on top of one unless
133    /// asked to, since a second pass in a row moves `content_cno`.
134    pub latest_compacted: bool,
135    pub data_block_size: usize,
136    pub meta_block_size: usize,
137    /// Objects whose checkpoint falls in the queried `[low, high]`, ascending.
138    pub objects: Vec<ObjectUsage>,
139    /// Objects outside that range, which no operation here will touch.
140    pub out_of_range: usize,
141    /// Keys under the container's prefix that are neither the inode nor a segment
142    /// object. A WAL root nested under the container is the usual reason, and
143    /// compaction refuses to run while any are present.
144    pub foreign_keys: usize,
145    pub foreign_sample: Option<String>,
146}
147
148impl Report {
149    /// Bytes stored by the objects in range.
150    pub fn bytes(&self) -> u64 {
151        self.objects.iter().map(|o| o.bytes).sum()
152    }
153
154    /// Bytes still pointed into.
155    pub fn bytes_live(&self) -> u64 {
156        self.objects.iter().map(|o| o.live_bytes).sum()
157    }
158
159    /// Objects nothing points into, and what they hold. This is `prune`'s work,
160    /// and it needs no rewrite.
161    pub fn dead(&self) -> (usize, u64) {
162        let dead = self.objects.iter().filter(|o| o.is_dead());
163        (dead.clone().count(), dead.map(|o| o.bytes).sum())
164    }
165
166    /// What compaction at `max_live_ratio` would cost and free: `(objects,
167    /// bytes_to_rewrite, bytes_freed)`.
168    ///
169    /// An estimate from the same numbers [`S3Cleaner::compact`] selects by, so a
170    /// dry run and this agree. The bytes freed are the objects' own, which
171    /// `prune` collects afterwards; the rewrite itself publishes a new checkpoint
172    /// holding the live blocks.
173    pub fn compactable(&self, max_live_ratio: f64) -> (usize, u64, u64) {
174        let picked = self.objects.iter()
175            .filter(|o| is_candidate(o, max_live_ratio));
176        (
177            picked.clone().count(),
178            picked.clone().map(|o| o.live_bytes).sum(),
179            picked.map(|o| o.bytes).sum(),
180        )
181    }
182}
183
184/// Which objects compaction may rewrite, and how much it may do at once.
185#[derive(Clone, Debug)]
186pub struct CompactOptions {
187    /// Rewrite an object only if at most this much of it is still live. At `1.0`
188    /// every live object is a candidate, which rewrites the whole container.
189    pub max_live_ratio: f64,
190    /// The most live bytes one pass rewrites. A pass is one checkpoint, so this is
191    /// also what it holds dirty in memory until the flush — `fs_relocate` does not
192    /// write anything out on its own. Objects are taken whole and cheapest first
193    /// until the next would pass this; what is left waits for the next pass.
194    ///
195    /// On a container that is not in the parted format a checkpoint is one object,
196    /// so a pass is further capped at what one object can hold.
197    pub max_pass_bytes: u64,
198    /// How many bytes of blocks to hand `fs_relocate` at once. It holds what it
199    /// reads in a buffer of its own before the dirty cache takes it, so this bounds
200    /// that second copy.
201    pub relocate_chunk_bytes: usize,
202    /// Select and report, write nothing. No file is opened.
203    pub dry_run: bool,
204    /// Run even though the newest checkpoint is already a compaction. The flag that
205    /// keeps `content_cno` still holds for one checkpoint only, so a second pass in
206    /// a row is seen by consumers as a change of content. Set this when that is
207    /// acceptable, for example to finish a container larger than one pass.
208    pub allow_consecutive: bool,
209    /// Run even though the prefix holds keys this crate does not recognize. Only
210    /// set this having established that they are not a WAL.
211    pub allow_foreign_keys: bool,
212}
213
214impl Default for CompactOptions {
215    fn default() -> Self {
216        Self {
217            max_live_ratio: 0.5,
218            max_pass_bytes: 512 * 1024 * 1024,
219            relocate_chunk_bytes: 64 * 1024 * 1024,
220            dry_run: false,
221            allow_consecutive: false,
222            allow_foreign_keys: false,
223        }
224    }
225}
226
227/// What a compaction pass did.
228#[derive(Clone, Debug, Default)]
229pub struct CompactOutcome {
230    /// The objects whose live blocks were rewritten, ascending.
231    pub selected: Vec<SegmentId>,
232    /// Of those, the ones a walk taken afterwards still finds referenced. Empty
233    /// on a pass that did what it set out to; anything here is an object that was
234    /// not freed, and is reported rather than assumed away.
235    pub still_referenced: Vec<SegmentId>,
236    /// Blocks `fs_relocate` marked: what was done, not what was planned. A hole,
237    /// an explicit zero block or a block already dirty is not counted.
238    pub blocks_rewritten: usize,
239    pub bytes_rewritten: u64,
240    /// The selected objects' stored bytes — what `prune` can now reclaim.
241    pub bytes_freed: u64,
242    /// Checkpoints published: one for a pass that rewrote anything, none
243    /// otherwise. That is how far `last_cno` advanced.
244    pub checkpoints: usize,
245    /// `fs_content_cno` before and after the pass. Equal when consumers will not
246    /// see the pass as a change of content, which is the ordinary case. They
247    /// differ when the publish could not be marked: its number did not follow the
248    /// previous one — an earlier flush failed and gave its number up — or the
249    /// newest checkpoint was already a compaction and the pass ran anyway.
250    pub content_cno_before: Cno,
251    pub content_cno_after: Cno,
252    /// Whether the candidates outlasted the pass budget. Another pass would find
253    /// more to do, at the cost of `content_cno` moving once (see
254    /// [`CompactOptions::allow_consecutive`]).
255    pub more_remaining: bool,
256    pub dry_run: bool,
257}
258
259impl CompactOutcome {
260    /// Whether a consumer comparing `content_cno` will see this pass as a change.
261    pub fn content_moved(&self) -> bool {
262        self.content_cno_before != self.content_cno_after
263    }
264}
265
266impl S3Cleaner {
267    /// What the container's storage is made of: per-object liveness for every
268    /// object whose checkpoint falls in `[low, high]`.
269    ///
270    /// Read-only, and the same walk `collect` does plus one `ListObjectsV2` for
271    /// sizes. `high == 0` means the latest published checkpoint.
272    pub async fn report(&self, low: Cno, high: Cno) -> Result<Report> {
273        let view = self.view(low, high).await?;
274        Ok(build_report(&view))
275    }
276
277    /// Rewrite the live blocks of mostly-dead objects so `prune` can reclaim
278    /// them.
279    ///
280    /// The caller must be the container's only writer for the duration; see the
281    /// [module docs](self) for why that is not a nicety. Nothing is deleted here
282    /// — a compaction makes objects reclaimable and `prune` reclaims them, which
283    /// keeps the destructive step in one place.
284    pub async fn compact(
285        &self,
286        low: Cno,
287        high: Cno,
288        opts: &CompactOptions,
289    ) -> Result<CompactOutcome> {
290        let view = self.view(low, high).await?;
291        if view.listing.foreign > 0 && !opts.allow_foreign_keys {
292            return Err(Error::new(
293                ErrorKind::InvalidInput,
294                format!(
295                    "{} key(s) under the container's prefix are neither the inode nor \
296                     segment objects (for example {:?}); a WAL is the usual reason, and \
297                     compacting past a log this pass does not read would publish over \
298                     entries recovery has not replayed. Set allow_foreign_keys once you \
299                     have established otherwise",
300                    view.listing.foreign,
301                    view.listing.foreign_sample.as_deref().unwrap_or("?"),
302                ),
303            ));
304        }
305
306        if view.latest_compacted && !opts.allow_consecutive && !opts.dry_run {
307            return Err(Error::new(
308                ErrorKind::InvalidInput,
309                format!(
310                    "checkpoint {} is already a compaction, and the flag that keeps \
311                     content_cno still holds for one checkpoint only: a second pass in a \
312                     row would be seen by consumers as a change of content. Write to the \
313                     file first, or set allow_consecutive",
314                    view.high_cno,
315                ),
316            ));
317        }
318
319        let report = build_report(&view);
320        let pass_cap = pass_cap(opts.max_pass_bytes, &view.fmt);
321        let (selected, bytes_planned, bytes_freed, more_remaining) =
322            plan_pass(&report.objects, opts.max_live_ratio, pass_cap);
323        debug!("compact: {} object(s) selected of {} in range, {} live bytes (cap {})",
324            selected.len(), report.objects.len(), bytes_planned, pass_cap);
325
326        let mut outcome = CompactOutcome {
327            selected: selected.iter().copied().collect(),
328            bytes_freed,
329            more_remaining,
330            dry_run: opts.dry_run,
331            ..Default::default()
332        };
333        if opts.dry_run {
334            // The plan's own numbers, which is all a dry run can report: nothing
335            // is opened, so nothing can be counted as done.
336            outcome.blocks_rewritten =
337                (bytes_planned / view.data_block_size.max(1) as u64) as usize;
338            outcome.bytes_rewritten = bytes_planned;
339            return Ok(outcome);
340        }
341
342        let config = compaction_config(&self.root_uri(), bytes_planned, &view);
343        let mut file = Hyper::open(
344            self.client().clone(), config, HyperFileFlags::from_flags(FileFlags::rdwr()),
345        ).await?;
346        let result = self.relocate_and_publish(&mut file, &view, &selected, opts,
347            &mut outcome).await;
348        let _ = file.fs_release().await;
349        result?;
350        if outcome.checkpoints == 0 {
351            return Ok(outcome);
352        }
353
354        // Say what was actually freed rather than what should have been. A
355        // selected object that is still referenced means a block was not
356        // relocated, which is worth reporting where it can be seen.
357        let after = self.view(low, 0).await?;
358        let referenced = after.referenced_objects();
359        outcome.still_referenced = outcome.selected.iter()
360            .filter(|segid| referenced.contains(segid))
361            .copied()
362            .collect();
363        if !outcome.still_referenced.is_empty() {
364            warn!("compact: {} of {} selected object(s) are still referenced: {:?}",
365                outcome.still_referenced.len(), outcome.selected.len(),
366                outcome.still_referenced);
367        }
368        if outcome.content_moved() {
369            warn!("compact: content_cno moved {} -> {}; consumers will see this pass \
370                as a change of content", outcome.content_cno_before,
371                outcome.content_cno_after);
372        }
373        info!("compact: rewrote {} block(s) ({} bytes) in {} checkpoint(s), freeing {} bytes \
374            across {} object(s)", outcome.blocks_rewritten, outcome.bytes_rewritten,
375            outcome.checkpoints, outcome.bytes_freed, outcome.selected.len());
376        Ok(outcome)
377    }
378
379    /// Mark every block of the selected objects for relocation, then publish once.
380    async fn relocate_and_publish(
381        &self,
382        file: &mut Hyper<'static>,
383        view: &ContainerView,
384        selected: &BTreeSet<SegmentId>,
385        opts: &CompactOptions,
386        outcome: &mut CompactOutcome,
387    ) -> Result<()> {
388        outcome.content_cno_before = file.fs_content_cno()?;
389        outcome.content_cno_after = outcome.content_cno_before;
390        if selected.is_empty() {
391            return Ok(());
392        }
393        let bs = view.data_block_size;
394        let blocks_in_file = (file.fs_getattr()?.st_size as u64).div_ceil(bs as u64);
395        let chunk_blocks = (opts.relocate_chunk_bytes / bs.max(1)).max(1);
396
397        // Which blocks are in the selected objects. `block_placement` answers
398        // exactly that, per index, from the live map — so it needs no agreement
399        // with a second reading of the container's layout. It is swept in chunks
400        // because the answer is one entry per block and a large file has a lot of
401        // them. Runs go to `fs_relocate` in index order, which is also the order
402        // the flush writes them in.
403        let mut pending: Vec<BlockIndex> = Vec::new();
404        let mut start: BlockIndex = 0;
405        while start < blocks_in_file {
406            let n = PLACEMENT_CHUNK_BLOCKS.min((blocks_in_file - start) as usize);
407            let places = file.fs_block_placement(start, n).await?;
408            for (i, place) in places.iter().enumerate() {
409                if let Some((segid, _)) = place
410                    && selected.contains(segid)
411                {
412                    pending.push(start + i as BlockIndex);
413                    if pending.len() >= chunk_blocks {
414                        outcome.blocks_rewritten += relocate_runs(file, &pending).await?;
415                        pending.clear();
416                    }
417                }
418            }
419            start += n as BlockIndex;
420        }
421        if !pending.is_empty() {
422            outcome.blocks_rewritten += relocate_runs(file, &pending).await?;
423        }
424        outcome.bytes_rewritten = (outcome.blocks_rewritten * bs) as u64;
425        if outcome.blocks_rewritten == 0 {
426            return Ok(());
427        }
428
429        let cno = file.fs_flush().await?;
430        outcome.checkpoints = 1;
431        outcome.content_cno_after = file.fs_content_cno()?;
432        debug!("compact: published checkpoint {} with {} relocated block(s); content_cno \
433            {} -> {}", cno, outcome.blocks_rewritten, outcome.content_cno_before,
434            outcome.content_cno_after);
435        Ok(())
436    }
437}
438
439/// Hand `indices` to `fs_relocate` as runs of consecutive blocks.
440async fn relocate_runs(file: &mut Hyper<'static>, indices: &[BlockIndex]) -> Result<usize> {
441    let runs = into_runs(indices);
442    let marked = file.fs_relocate(&runs).await?;
443    debug!("compact: relocate {} block(s) in {} run(s) marked {}",
444        indices.len(), runs.len(), marked);
445    Ok(marked)
446}
447
448/// What one pass may rewrite: the caller's budget, and on a container that is not
449/// parted, what one object can hold. Under any other format a checkpoint is one
450/// object, and `hyperfile` refuses a flush larger than that. Room is left for the
451/// summary and the index nodes the same object carries.
452fn pass_cap(max_pass_bytes: u64, fmt: &BlockPtrFormat) -> u64 {
453    if fmt.is_parted() {
454        return max_pass_bytes;
455    }
456    let one_object = fmt.max_object_bytes() as u64;
457    max_pass_bytes.min(one_object / 4 * 3)
458}
459
460/// Candidates, cheapest first, taken whole while they fit `cap`.
461///
462/// Returns `(selected, live_bytes, stored_bytes, more_remaining)`. An object is
463/// never split across passes: one that does not fit ends the selection, so the
464/// cheapest work is done first and nothing is half-moved.
465fn plan_pass(
466    objects: &[ObjectUsage],
467    max_live_ratio: f64,
468    cap: u64,
469) -> (BTreeSet<SegmentId>, u64, u64, bool) {
470    let mut selected = BTreeSet::new();
471    let mut live = 0u64;
472    let mut stored = 0u64;
473    let candidates = select_candidates(objects, max_live_ratio);
474    let total = candidates.len();
475    for usage in candidates {
476        if live.saturating_add(usage.live_bytes) > cap {
477            break;
478        }
479        live += usage.live_bytes;
480        stored += usage.bytes;
481        selected.insert(usage.segid);
482    }
483    let more = selected.len() < total;
484    (selected, live, stored, more)
485}
486
487/// Per-object liveness, from one view.
488///
489/// A location's size depends on what it holds, and a pointer does not say: a
490/// metadata node is `meta_block_size` and a data block is `data_block_size`. The
491/// metadata subset of the walk is what separates them.
492pub(crate) fn build_report(view: &ContainerView) -> Report {
493    let mut live_data: BTreeMap<SegmentId, usize> = BTreeMap::new();
494    let mut live_meta: BTreeMap<SegmentId, usize> = BTreeMap::new();
495    for blkptr in &view.live.all {
496        let segid = view.object_of(blkptr);
497        if view.live.meta.contains(blkptr) {
498            *live_meta.entry(segid).or_default() += 1;
499        } else {
500            *live_data.entry(segid).or_default() += 1;
501        }
502    }
503
504    let mut objects = Vec::new();
505    let mut out_of_range = 0;
506    for (segid, bytes) in &view.listing.objects {
507        let cno = segid.as_cno();
508        if cno < view.low || cno > view.high_cno {
509            out_of_range += 1;
510            continue;
511        }
512        let live_data_blocks = live_data.get(segid).copied().unwrap_or(0);
513        let live_meta_blocks = live_meta.get(segid).copied().unwrap_or(0);
514        objects.push(ObjectUsage {
515            segid: *segid,
516            bytes: *bytes,
517            live_data_blocks,
518            live_meta_blocks,
519            live_bytes: (live_data_blocks * view.data_block_size
520                + live_meta_blocks * view.meta_block_size) as u64,
521        });
522    }
523
524    Report {
525        high_cno: view.high_cno,
526        latest_compacted: view.latest_compacted,
527        data_block_size: view.data_block_size,
528        meta_block_size: view.meta_block_size,
529        objects,
530        out_of_range,
531        foreign_keys: view.listing.foreign,
532        foreign_sample: view.listing.foreign_sample.clone(),
533    }
534}
535
536/// Whether rewriting this object's live data would free it.
537///
538/// Three exclusions, each for its own reason: a dead object is `prune`'s work and
539/// rewriting nothing frees nothing; an object holding a live metadata node cannot
540/// be freed by rewriting data blocks, so its data would be rewritten for no
541/// return; and one above the ratio costs more to rewrite than the policy accepts.
542fn is_candidate(usage: &ObjectUsage, max_live_ratio: f64) -> bool {
543    !usage.is_dead()
544        && !usage.holds_metadata()
545        && usage.live_ratio() <= max_live_ratio
546}
547
548/// Candidates, cheapest first.
549///
550/// Ordered by live ratio ascending, which is the same as bytes freed per byte
551/// rewritten descending, with the larger object first among equals so a budget
552/// spent early buys more.
553fn select_candidates(
554    objects: &[ObjectUsage],
555    max_live_ratio: f64,
556) -> Vec<&ObjectUsage> {
557    let mut picked: Vec<&ObjectUsage> = objects.iter()
558        .filter(|o| is_candidate(o, max_live_ratio))
559        .collect();
560    picked.sort_by(|a, b| {
561        a.live_ratio().total_cmp(&b.live_ratio())
562            .then(b.bytes.cmp(&a.bytes))
563            .then(a.segid.cmp(&b.segid))
564    });
565    picked
566}
567
568/// Consecutive indices, as `(first, count)`.
569///
570/// What `read_ahead` wants: it takes a byte range and plans it the way a read of
571/// the same range would, so one call per run of adjacent blocks turns into a few
572/// large requests. Asking per block instead would be one request per block, which
573/// for scattered blocks it will be anyway.
574fn into_runs(indices: &[BlockIndex]) -> Vec<(BlockIndex, usize)> {
575    let mut runs: Vec<(BlockIndex, usize)> = Vec::new();
576    for idx in indices {
577        match runs.last_mut() {
578            Some((start, count)) if *start + *count as BlockIndex == *idx => *count += 1,
579            _ => runs.push((*idx, 1)),
580        }
581    }
582    runs
583}
584
585/// The configuration a compacting handle needs, and why each field is not the
586/// default.
587fn compaction_config(root_uri: &str, pass_bytes: u64, view: &ContainerView) -> HyperFileConfig {
588    let staging = StagingConfig::new_s3_uri(root_uri, None);
589    // Nothing may publish before the pass is complete. `fs_relocate` does not
590    // check the thresholds itself, but the flush interval is checked on the
591    // flush path, and a pass that crossed a threshold would be cut in two —
592    // which moves `last_cno` twice and leaves `content_cno` marked for only the
593    // second. Both thresholds and the segment buffer go above the pass.
594    let room = (pass_bytes as usize)
595        .saturating_add(view.data_block_size)
596        .saturating_mul(2)
597        .max(16 * 1024 * 1024);
598    let runtime = HyperFileRuntimeConfig {
599        data_cache_dirty_max_bytes_threshold: room,
600        data_cache_dirty_max_blocks_threshold: room / view.data_block_size.max(1),
601        segment_buffer_size: room,
602        data_cache_dirty_max_flush_interval: 3_600_000,
603        // A compactor writes bytes it read earlier. Refreshing and re-publishing
604        // on a conflict, which is what the default does, would republish a stale
605        // copy over another writer's newer one.
606        flush_conflict_policy: FlushConflictPolicy::FailFast,
607        ..HyperFileRuntimeConfig::default()
608    };
609    HyperFileConfigBuilder::new()
610        .with_staging_config(&staging)
611        .with_runtime_config(&runtime)
612        .build()
613}
614
615#[cfg(test)]
616mod tests {
617    use super::*;
618
619    fn usage(segid: SegmentId, bytes: u64, data: usize, meta: usize) -> ObjectUsage {
620        ObjectUsage {
621            segid,
622            bytes,
623            live_data_blocks: data,
624            live_meta_blocks: meta,
625            live_bytes: (data * 4096 + meta * 4096) as u64,
626        }
627    }
628
629    #[test]
630    fn a_dead_object_is_pruned_not_compacted() {
631        let o = usage(SegmentId::new(1), 1 << 20, 0, 0);
632        assert!(o.is_dead());
633        assert!(!is_candidate(&o, 1.0), "nothing to rewrite frees nothing");
634    }
635
636    /// An object a live metadata node sits in cannot be freed by rewriting data
637    /// blocks: which keys that node covers is not visible from out here, so
638    /// nothing this crate can do will relocate it.
639    #[test]
640    fn an_object_holding_metadata_is_not_a_candidate() {
641        let o = usage(SegmentId::with_part(3, 0), 1 << 20, 1, 1);
642        assert!(o.holds_metadata());
643        assert!(!is_candidate(&o, 1.0));
644    }
645
646    #[test]
647    fn the_ratio_is_a_ceiling() {
648        let low = usage(SegmentId::new(1), 4096 * 100, 5, 0);   // 5% live
649        let high = usage(SegmentId::new(2), 4096 * 100, 90, 0); // 90% live
650        assert!(is_candidate(&low, 0.5));
651        assert!(!is_candidate(&high, 0.5));
652        assert!(is_candidate(&high, 0.9));
653    }
654
655    /// Cheapest first, and the bigger object first among equals: the budget is
656    /// finite and the point is how much it buys.
657    #[test]
658    fn candidates_come_cheapest_first() {
659        let objects = vec![
660            usage(SegmentId::new(1), 4096 * 100, 40, 0), // 40%
661            usage(SegmentId::new(2), 4096 * 100, 5, 0),  // 5%
662            usage(SegmentId::new(3), 4096 * 200, 10, 0), // 5%, larger
663        ];
664        let picked = select_candidates(&objects, 0.5);
665        let order: Vec<SegmentId> = picked.iter().map(|o| o.segid).collect();
666        assert_eq!(order, vec![SegmentId::new(3), SegmentId::new(2), SegmentId::new(1)]);
667    }
668
669    #[test]
670    fn runs_are_maximal() {
671        assert_eq!(into_runs(&[]), vec![]);
672        assert_eq!(into_runs(&[7]), vec![(7, 1)]);
673        assert_eq!(into_runs(&[0, 1, 2, 5, 6, 9]), vec![(0, 3), (5, 2), (9, 1)]);
674    }
675
676    /// A pass takes objects whole and cheapest first, and stops at the first one
677    /// that does not fit rather than skipping to a smaller one: the cheapest work
678    /// is done first, and the pass says there is more.
679    #[test]
680    fn a_pass_takes_whole_objects_until_the_budget() {
681        let objects = vec![
682            usage(SegmentId::new(1), 4096 * 100, 10, 0), // 10%, 40 KiB live
683            usage(SegmentId::new(2), 4096 * 100, 20, 0), // 20%, 80 KiB live
684            usage(SegmentId::new(3), 4096 * 100, 1, 0),  //  1%,  4 KiB live
685        ];
686        let (picked, live, stored, more) = plan_pass(&objects, 0.5, 4096 * 15);
687        assert_eq!(picked.into_iter().collect::<Vec<_>>(),
688            vec![SegmentId::new(1), SegmentId::new(3)]);
689        assert_eq!(live, 4096 * 11);
690        assert_eq!(stored, 4096 * 200);
691        assert!(more, "object 2 did not fit and is still a candidate");
692
693        let (picked, _, _, more) = plan_pass(&objects, 0.5, u64::MAX);
694        assert_eq!(picked.len(), 3);
695        assert!(!more);
696    }
697
698    /// A parted container splits a large checkpoint into several objects, so only
699    /// the caller's budget applies. Any other format writes a checkpoint as one
700    /// object and refuses a flush past it, so the pass has to fit one, with room
701    /// for what else that object carries.
702    #[test]
703    fn an_unparted_pass_fits_one_object() {
704        let big = u64::MAX;
705        assert_eq!(pass_cap(big, &BlockPtrFormat::PartedSegment), big);
706        let mg = BlockPtrFormat::MicroGroup.max_object_bytes() as u64;
707        assert!(pass_cap(big, &BlockPtrFormat::MicroGroup) < mg);
708        assert!(pass_cap(big, &BlockPtrFormat::MicroGroup) >= mg / 2);
709        assert_eq!(pass_cap(4096, &BlockPtrFormat::MicroGroup), 4096,
710            "a smaller budget is kept as given");
711    }
712
713    /// An object whose live data alone is over the budget is never taken, and
714    /// the pass reports that it left work behind instead of doing nothing quietly.
715    #[test]
716    fn an_object_larger_than_the_budget_is_reported_not_split() {
717        let objects = vec![usage(SegmentId::new(1), 4096 * 1000, 100, 0)];
718        let (picked, live, _, more) = plan_pass(&objects, 0.5, 4096 * 10);
719        assert!(picked.is_empty());
720        assert_eq!(live, 0);
721        assert!(more);
722    }
723
724    /// `compactable` must answer from the same rule the pass selects by,
725    /// otherwise a dry run and the pass disagree about what will happen.
726    #[test]
727    fn the_estimate_agrees_with_the_selection() {
728        let report = Report {
729            high_cno: 9,
730            latest_compacted: false,
731            data_block_size: 4096,
732            meta_block_size: 4096,
733            objects: vec![
734                usage(SegmentId::new(1), 4096 * 100, 5, 0),
735                usage(SegmentId::new(2), 4096 * 100, 80, 0),
736                usage(SegmentId::new(3), 4096 * 100, 0, 0),
737                usage(SegmentId::new(4), 4096 * 100, 2, 3),
738            ],
739            out_of_range: 0,
740            foreign_keys: 0,
741            foreign_sample: None,
742        };
743        let (n, rewrite, freed) = report.compactable(0.5);
744        assert_eq!(n, 1, "only the 5% object qualifies");
745        assert_eq!(rewrite, 5 * 4096);
746        assert_eq!(freed, 4096 * 100);
747        assert_eq!(report.dead(), (1, 4096 * 100));
748        let picked = select_candidates(&report.objects, CompactOptions::default().max_live_ratio);
749        assert_eq!(picked.len(), n);
750    }
751}