mz_persist_client/internal/
gc.rs

1// Copyright Materialize, Inc. and contributors. All rights reserved.
2//
3// Use of this software is governed by the Business Source License
4// included in the LICENSE file.
5//
6// As of the Change Date specified in that file, in accordance with
7// the Business Source License, use of this software will be governed
8// by the Apache License, Version 2.0.
9
10use std::borrow::Borrow;
11use std::collections::BTreeSet;
12use std::fmt::Debug;
13use std::marker::PhantomData;
14use std::mem;
15use std::sync::Arc;
16use std::time::Instant;
17
18use differential_dataflow::difference::Semigroup;
19use differential_dataflow::lattice::Lattice;
20use futures_util::StreamExt;
21use futures_util::stream::FuturesUnordered;
22use prometheus::Counter;
23use timely::progress::Timestamp;
24use tokio::sync::mpsc::UnboundedSender;
25use tokio::sync::{Semaphore, mpsc, oneshot};
26use tracing::{Instrument, Span, debug, debug_span, warn};
27
28use crate::async_runtime::IsolatedRuntime;
29use crate::batch::PartDeletes;
30use crate::cfg::GC_BLOB_DELETE_CONCURRENCY_LIMIT;
31
32use mz_ore::cast::CastFrom;
33use mz_ore::collections::HashSet;
34use mz_ore::soft_assert_or_log;
35use mz_persist::location::{Blob, SeqNo};
36use mz_persist_types::{Codec, Codec64};
37
38use crate::ShardId;
39use crate::internal::machine::{Machine, retry_external};
40use crate::internal::maintenance::RoutineMaintenance;
41use crate::internal::metrics::{GcStepTimings, RetryMetrics};
42use crate::internal::paths::{BlobKey, PartialBlobKey, PartialRollupKey};
43use crate::internal::state::HollowBlobRef;
44use crate::internal::state_versions::{InspectDiff, StateVersionsIter};
45
46#[derive(Debug, Clone, PartialEq)]
47pub struct GcReq {
48    pub shard_id: ShardId,
49    pub new_seqno_since: SeqNo,
50}
51
52#[derive(Debug)]
53pub struct GarbageCollector<K, V, T, D> {
54    sender: UnboundedSender<(GcReq, oneshot::Sender<RoutineMaintenance>)>,
55    _phantom: PhantomData<fn() -> (K, V, T, D)>,
56}
57
58impl<K, V, T, D> Clone for GarbageCollector<K, V, T, D> {
59    fn clone(&self) -> Self {
60        GarbageCollector {
61            sender: self.sender.clone(),
62            _phantom: PhantomData,
63        }
64    }
65}
66
67/// Cleanup for no longer necessary blobs and consensus versions.
68///
69/// - Every read handle, snapshot, and listener is given a capability on seqno
70///   with a very long lease (allowing for infrequent heartbeats). This is a
71///   guarantee that no blobs referenced by the state at that version will be
72///   deleted (even if they've been compacted in some newer version of the
73///   state). This is called a seqno_since in the code as it has obvious
74///   parallels to how sinces work at the shard/collection level. (Is reusing
75///   "since" here a good idea? We could also call it a "seqno capability" or
76///   something else instead.)
77/// - Every state transition via apply_unbatched_cmd has the opportunity to
78///   determine that the overall seqno_since for the shard has changed. In the
79///   common case in production, this will be in response to a snapshot
80///   finishing or a listener emitting some batch.
81/// - It would be nice if this only ever happened in response to read-y things
82///   (there'd be a nice parallel to how compaction background work is only
83///   spawned by write activity), but if there are no readers, we still very
84///   much want to continue to garbage collect. Notably, if there are no
85///   readers, we naturally only need to hold a capability on the current
86///   version of state. This means that if there are only writers, a write
87///   commands will result in the seqno_since advancing immediately from the
88///   previous version of the state to the new one.
89/// - Like Compacter, GarbageCollector uses a heuristic to ignore some requests
90///   to save work. In this case, the tradeoff is between consensus traffic
91///   (plus a bit of cpu) and keeping blobs around longer than strictly
92///   necessary. This is correct because a process could always die while
93///   executing one of these requests (or be slow and still working on it when
94///   the next request is generated), so we anyway need to handle them being
95///   dropped.
96/// - GarbageCollector works by `Consensus::scan`-ing for every live version of
97///   state (ignoring what the request things the prev_state_seqno was for the
98///   reasons mentioned immediately above). It then walks through them in a
99///   loop, accumulating a BTreeSet of every referenced blob key. When it finds
100///   the version corresponding to the new_seqno_since, it removes every blob in
101///   that version of the state from the BTreeSet and exits the loop. This
102///   results in the BTreeSet containing every blob eligible for deletion. It
103///   deletes those blobs and then truncates the state to the new_seqno_since to
104///   indicate that this work doesn't need to be done again.
105/// - Note that these requests are being processed concurrently, so it's always
106///   possible that some future request has already deleted the blobs and
107///   truncated consensus. It's also possible that this is the future request.
108///   As a result, the only guarantee that we get is that the current version of
109///   head is >= new_seqno_since.
110/// - (Aside: The above also means that if Blob is not linearizable, there is a
111///   possible race where a blob gets deleted before it written and thus is
112///   leaked. We anyway always have the possibility of a write process being
113///   killed between when it writes a blob and links it into state, so this is
114///   fine; it'll be caught and fixed by the same mechanism.)
115impl<K, V, T, D> GarbageCollector<K, V, T, D>
116where
117    K: Debug + Codec,
118    V: Debug + Codec,
119    T: Timestamp + Lattice + Codec64 + Sync,
120    D: Semigroup + Codec64,
121{
122    pub fn new(machine: Machine<K, V, T, D>, isolated_runtime: Arc<IsolatedRuntime>) -> Self {
123        let (gc_req_sender, mut gc_req_recv) =
124            mpsc::unbounded_channel::<(GcReq, oneshot::Sender<RoutineMaintenance>)>();
125
126        // spin off a single task responsible for executing GC requests.
127        // work is enqueued into the task through a channel
128        let _worker_handle = mz_ore::task::spawn(|| "PersistGcWorker", async move {
129            while let Some((req, completer)) = gc_req_recv.recv().await {
130                let mut consolidated_req = req;
131                let mut gc_completed_senders = vec![completer];
132
133                // check if any further gc requests have built up. we'll merge their requests
134                // together and run a single GC pass to satisfy all of them
135                while let Ok((req, completer)) = gc_req_recv.try_recv() {
136                    assert_eq!(req.shard_id, consolidated_req.shard_id);
137                    gc_completed_senders.push(completer);
138                    consolidated_req.new_seqno_since =
139                        std::cmp::max(req.new_seqno_since, consolidated_req.new_seqno_since);
140                }
141
142                let merged_requests = gc_completed_senders.len() - 1;
143                if merged_requests > 0 {
144                    machine
145                        .applier
146                        .metrics
147                        .gc
148                        .merged
149                        .inc_by(u64::cast_from(merged_requests));
150                    debug!(
151                        "Merged {} gc requests together for shard {}",
152                        merged_requests, consolidated_req.shard_id
153                    );
154                }
155
156                let gc_span = debug_span!(parent: None, "gc_and_truncate", shard_id=%consolidated_req.shard_id);
157                gc_span.follows_from(&Span::current());
158
159                let start = Instant::now();
160                machine.applier.metrics.gc.started.inc();
161                let (mut maintenance, _stats) = {
162                    let name = format!("gc_and_truncate ({})", &consolidated_req.shard_id);
163                    let machine = machine.clone();
164                    isolated_runtime
165                        .spawn_named(|| name, async move {
166                            Self::gc_and_truncate(&machine, consolidated_req)
167                                .instrument(gc_span)
168                                .await
169                        })
170                        .await
171                        .expect("gc_and_truncate failed")
172                };
173                machine.applier.metrics.gc.finished.inc();
174                machine.applier.shard_metrics.gc_finished.inc();
175                machine
176                    .applier
177                    .metrics
178                    .gc
179                    .seconds
180                    .inc_by(start.elapsed().as_secs_f64());
181
182                // inform all callers who enqueued GC reqs that their work is complete
183                for sender in gc_completed_senders {
184                    // we can safely ignore errors here, it's possible the caller
185                    // wasn't interested in waiting and dropped their receiver.
186                    // maintenance will be somewhat-arbitrarily assigned to the first oneshot.
187                    let _ = sender.send(mem::take(&mut maintenance));
188                }
189            }
190        });
191
192        GarbageCollector {
193            sender: gc_req_sender,
194            _phantom: PhantomData,
195        }
196    }
197
198    /// Enqueues a [GcReq] to be consumed by the GC background task when available.
199    ///
200    /// Returns a future that indicates when GC has cleaned up to at least [GcReq::new_seqno_since]
201    pub fn gc_and_truncate_background(
202        &self,
203        req: GcReq,
204    ) -> Option<oneshot::Receiver<RoutineMaintenance>> {
205        let (gc_completed_sender, gc_completed_receiver) = oneshot::channel();
206        let new_gc_sender = self.sender.clone();
207        let send = new_gc_sender.send((req, gc_completed_sender));
208
209        if let Err(e) = send {
210            // In the steady state we expect this to always succeed, but during
211            // shutdown it is possible the destination task has already spun down
212            warn!(
213                "gc_and_truncate_background failed to send gc request: {}",
214                e
215            );
216            return None;
217        }
218
219        Some(gc_completed_receiver)
220    }
221
222    pub(crate) async fn gc_and_truncate(
223        machine: &Machine<K, V, T, D>,
224        req: GcReq,
225    ) -> (RoutineMaintenance, GcResults) {
226        let mut step_start = Instant::now();
227        let mut report_step_timing = |counter: &Counter| {
228            let now = Instant::now();
229            counter.inc_by(now.duration_since(step_start).as_secs_f64());
230            step_start = now;
231        };
232        assert_eq!(req.shard_id, machine.shard_id());
233
234        // Double check our GC req: seqno_since will never regress
235        // so we can verify it's not somehow greater than the last-
236        // known seqno_since
237        if req.new_seqno_since > machine.applier.seqno_since() {
238            machine
239                .applier
240                .fetch_and_update_state(Some(req.new_seqno_since))
241                .await;
242            let current_seqno_since = machine.applier.seqno_since();
243            assert!(
244                req.new_seqno_since <= current_seqno_since,
245                "invalid gc req: {:?} vs machine seqno_since {}",
246                req,
247                current_seqno_since
248            );
249        }
250
251        // First, check the latest known state to this process to see
252        // if there's relevant GC work for this seqno_since
253        let gc_rollups =
254            GcRollups::new(machine.applier.rollups_lte_seqno(req.new_seqno_since), &req);
255        let rollups_to_remove_from_state = gc_rollups.rollups_to_remove_from_state();
256        report_step_timing(&machine.applier.metrics.gc.steps.find_removable_rollups);
257
258        let mut gc_results = GcResults::default();
259
260        if rollups_to_remove_from_state.is_empty() {
261            // If there are no rollups to remove from state (either the work has already
262            // been done, or the there aren't enough rollups <= seqno_since to have any
263            // to delete), we can safely exit.
264            machine.applier.metrics.gc.noop.inc();
265            return (RoutineMaintenance::default(), gc_results);
266        }
267
268        debug!(
269            "Finding all rollups <= ({}). Will truncate: {:?}. Will remove rollups from state: {:?}",
270            req.new_seqno_since,
271            gc_rollups.truncate_seqnos().collect::<Vec<_>>(),
272            rollups_to_remove_from_state,
273        );
274
275        let mut states = machine
276            .applier
277            .state_versions
278            .fetch_all_live_states(req.shard_id)
279            .await
280            .expect("state is initialized")
281            .check_ts_codec()
282            .expect("ts codec has not changed");
283        let initial_seqno = states.state().seqno;
284        report_step_timing(&machine.applier.metrics.gc.steps.fetch_seconds);
285
286        machine
287            .applier
288            .shard_metrics
289            .gc_live_diffs
290            .set(u64::cast_from(states.len()));
291
292        debug!(
293            "gc seqno_since: ({}) got {} versions from scan",
294            req.new_seqno_since,
295            states.len()
296        );
297
298        Self::incrementally_delete_and_truncate(
299            &mut states,
300            &gc_rollups,
301            machine,
302            &mut report_step_timing,
303            &mut gc_results,
304        )
305        .await;
306
307        // Now that the blobs are deleted / Consensus is truncated, remove
308        // the rollups from state. Doing this at the end ensures that our
309        // invariant is maintained that the current state contains a rollup
310        // to the earliest state in Consensus, and ensures that if GC crashes
311        // part-way through, we still have a reference to these rollups to
312        // resume their deletion.
313        //
314        // This does mean that if GC crashes part-way through we would
315        // repeat work when it resumes. However the redundant work should
316        // be minimal as Consensus is incrementally truncated, allowing
317        // the next run of GC to skip any work needed for rollups less
318        // than the last truncation.
319        //
320        // In short, while this step is not incremental, it does not need
321        // to be for GC to efficiently resume. And in fact, making it
322        // incremental could be quite expensive (e.g. more CaS operations).
323        let (removed_rollups, maintenance) =
324            machine.remove_rollups(rollups_to_remove_from_state).await;
325        report_step_timing(&machine.applier.metrics.gc.steps.remove_rollups_from_state);
326        debug!("CaS removed rollups from state: {:?}", removed_rollups);
327        gc_results.rollups_removed_from_state = removed_rollups;
328
329        // Everything here and below is not strictly needed for GC to complete,
330        // but it's a good opportunity, while we have all live states in hand,
331        // to run some metrics and assertions.
332
333        // Apply all remaining live states to rollup some metrics, like how many
334        // parts are being held (in Blob) that are not part of the latest state.
335        let mut seqno_held_parts = 0;
336        while let Some(_) = states.next(|diff| match diff {
337            InspectDiff::FromInitial(_) => {}
338            InspectDiff::Diff(diff) => {
339                seqno_held_parts += diff.part_deletes().count();
340            }
341        }) {}
342
343        machine
344            .applier
345            .shard_metrics
346            .gc_seqno_held_parts
347            .set(u64::cast_from(seqno_held_parts));
348
349        // verify that the "current" state (as of `fetch_all_live_states`) contains
350        // a rollup to the earliest state we fetched. this invariant isn't affected
351        // by the GC work we just performed, but it is a property of GC correctness
352        // overall / is a convenient place to run the assertion.
353        let valid_pre_gc_state = states
354            .state()
355            .collections
356            .rollups
357            .contains_key(&initial_seqno);
358
359        // this should never be true in the steady-state, but may be true the
360        // first time GC runs after fixing any correctness bugs related to our
361        // state version invariants. we'll make it an error so we can track
362        // any violations in Sentry, but opt not to panic because the root
363        // cause of the violation cannot be from this GC run (in fact, this
364        // GC run, assuming it's correct, should have fixed the violation!)
365        soft_assert_or_log!(
366            valid_pre_gc_state,
367            "earliest state fetched during GC did not have corresponding rollup: rollups = {:?}, state seqno = {}",
368            states.state().collections.rollups,
369            initial_seqno
370        );
371
372        report_step_timing(
373            &machine
374                .applier
375                .metrics
376                .gc
377                .steps
378                .post_gc_calculations_seconds,
379        );
380
381        (maintenance, gc_results)
382    }
383
384    /// Physically deletes all blobs from Blob and live diffs from Consensus that
385    /// are safe to delete, given the `seqno_since`, ensuring that the earliest
386    /// live diff in Consensus has a rollup of seqno `<= seqno_since`.
387    ///
388    /// Internally, performs deletions for each rollup encountered, ensuring that
389    /// incremental progress is made even if the process is interrupted before
390    /// completing all gc work.
391    async fn incrementally_delete_and_truncate<F>(
392        states: &mut StateVersionsIter<T>,
393        gc_rollups: &GcRollups,
394        machine: &Machine<K, V, T, D>,
395        timer: &mut F,
396        gc_results: &mut GcResults,
397    ) where
398        F: FnMut(&Counter),
399    {
400        assert_eq!(states.state().shard_id, machine.shard_id());
401        let shard_id = states.state().shard_id;
402        let mut batch_parts_to_delete = PartDeletes::default();
403        let mut rollups_to_delete: BTreeSet<PartialRollupKey> = BTreeSet::new();
404
405        for truncate_lt in gc_rollups.truncate_seqnos() {
406            assert!(batch_parts_to_delete.is_empty());
407            assert!(rollups_to_delete.is_empty());
408
409            // our state is already past the truncation point. there's no work to do --
410            // some process already truncated this far
411            if states.state().seqno >= truncate_lt {
412                continue;
413            }
414
415            // By our invariant, `states` should always begin on a rollup.
416            assert!(
417                gc_rollups.contains_seqno(&states.state().seqno),
418                "must start with a present rollup before searching for blobs: rollups = {:#?}, state seqno = {}",
419                gc_rollups,
420                states.state().seqno
421            );
422
423            Self::find_removable_blobs(
424                states,
425                truncate_lt,
426                &machine.applier.metrics.gc.steps,
427                timer,
428                &mut batch_parts_to_delete,
429                &mut rollups_to_delete,
430            );
431
432            // After finding removable blobs, our state should be exactly `truncate_lt`,
433            // to ensure we've seen all blob deletions in the diffs needed to reach
434            // this seqno.
435            //
436            // That we can always reach `truncate_lt` given the live diffs we fetched
437            // earlier is a little subtle:
438            // * Our GC request was generated after `seqno_since` was written.
439            // * If our initial seqno on this loop was < `truncate_lt`, then our read
440            //   to `fetch_all_live_states` must have seen live diffs through at least
441            //   `seqno_since`, because the diffs were not yet truncated.
442            // * `seqno_since` >= `truncate_lt`, therefore we must have enough live
443            //   diffs to reach `truncate_lt`.
444            assert_eq!(states.state().seqno, truncate_lt);
445            // `truncate_lt` _is_ the seqno of a rollup, but let's very explicitly
446            // assert that we're about to truncate everything less than a rollup
447            // to maintain our invariant.
448            assert!(
449                gc_rollups.contains_seqno(&states.state().seqno),
450                "must start with a present rollup after searching for blobs: rollups = {:#?}, state seqno = {}",
451                gc_rollups,
452                states.state().seqno
453            );
454
455            // Extra paranoia: verify that none of the blobs we're about to delete
456            // are in our current state (we should only be truncating blobs from
457            // before this state!)
458            states.state().blobs().for_each(|blob| match blob {
459                HollowBlobRef::Batch(batch) => {
460                    for live_part in &batch.parts {
461                        assert!(!batch_parts_to_delete.contains(live_part));
462                    }
463                }
464                HollowBlobRef::Rollup(live_rollup) => {
465                    assert_eq!(rollups_to_delete.get(&live_rollup.key), None);
466                    // And double check that the rollups we're about to delete are
467                    // earlier than our truncation point:
468                    match BlobKey::parse_ids(&live_rollup.key.complete(&shard_id)) {
469                        Ok((_shard, PartialBlobKey::Rollup(rollup_seqno, _rollup))) => {
470                            assert!(rollup_seqno < truncate_lt);
471                        }
472                        _ => {
473                            panic!("invalid rollup during deletion: {:?}", live_rollup);
474                        }
475                    }
476                }
477            });
478
479            gc_results.truncated_consensus_to.push(truncate_lt);
480            gc_results.batch_parts_deleted_from_blob += batch_parts_to_delete.len();
481            gc_results.rollups_deleted_from_blob += rollups_to_delete.len();
482
483            Self::delete_and_truncate(
484                truncate_lt,
485                &mut batch_parts_to_delete,
486                &mut rollups_to_delete,
487                machine,
488                timer,
489            )
490            .await;
491        }
492    }
493
494    /// Iterates through `states`, accumulating all deleted blobs (both batch parts
495    /// and rollups) until reaching the seqno `truncate_lt`.
496    ///
497    /// * The initial seqno of `states` MUST be less than `truncate_lt`.
498    /// * The seqno of `states` after this fn will be exactly `truncate_lt`.
499    fn find_removable_blobs<F>(
500        states: &mut StateVersionsIter<T>,
501        truncate_lt: SeqNo,
502        metrics: &GcStepTimings,
503        timer: &mut F,
504        batch_parts_to_delete: &mut PartDeletes<T>,
505        rollups_to_delete: &mut BTreeSet<PartialRollupKey>,
506    ) where
507        F: FnMut(&Counter),
508    {
509        assert!(states.state().seqno < truncate_lt);
510        while let Some(state) = states.next(|diff| match diff {
511            InspectDiff::FromInitial(_) => {}
512            InspectDiff::Diff(diff) => {
513                diff.rollup_deletes().for_each(|rollup| {
514                    // we use BTreeSets for fast lookups elsewhere, but we should never
515                    // see repeat rollup insertions within a single GC run, otherwise we
516                    // have a logic error or our diffs are incorrect (!)
517                    assert!(rollups_to_delete.insert(rollup.key.to_owned()));
518                });
519                diff.part_deletes().for_each(|part| {
520                    assert!(batch_parts_to_delete.add(part));
521                });
522            }
523        }) {
524            if state.seqno == truncate_lt {
525                break;
526            }
527        }
528        timer(&metrics.find_deletable_blobs_seconds);
529    }
530
531    /// Deletes `batch_parts` and `rollups` from Blob.
532    /// Truncates Consensus to `truncate_lt`.
533    async fn delete_and_truncate<F>(
534        truncate_lt: SeqNo,
535        batch_parts: &mut PartDeletes<T>,
536        rollups: &mut BTreeSet<PartialRollupKey>,
537        machine: &Machine<K, V, T, D>,
538        timer: &mut F,
539    ) where
540        F: FnMut(&Counter),
541    {
542        let shard_id = machine.shard_id();
543        let concurrency_limit = GC_BLOB_DELETE_CONCURRENCY_LIMIT.get(&machine.applier.cfg);
544        let delete_semaphore = Semaphore::new(concurrency_limit);
545
546        let batch_parts = std::mem::take(batch_parts);
547        batch_parts
548            .delete(
549                machine.applier.state_versions.blob.borrow(),
550                shard_id,
551                concurrency_limit,
552                &*machine.applier.metrics,
553                &machine.applier.metrics.retries.external.batch_delete,
554            )
555            .instrument(debug_span!("batch::delete"))
556            .await;
557        timer(&machine.applier.metrics.gc.steps.delete_batch_part_seconds);
558
559        Self::delete_all(
560            machine.applier.state_versions.blob.borrow(),
561            rollups.iter().map(|k| k.complete(&shard_id)),
562            &machine.applier.metrics.retries.external.rollup_delete,
563            debug_span!("rollup::delete"),
564            &delete_semaphore,
565        )
566        .await;
567        rollups.clear();
568        timer(&machine.applier.metrics.gc.steps.delete_rollup_seconds);
569
570        machine
571            .applier
572            .state_versions
573            .truncate_diffs(&shard_id, truncate_lt)
574            .await;
575        timer(&machine.applier.metrics.gc.steps.truncate_diff_seconds);
576    }
577
578    // There's also a bulk delete API in s3 if the performance of this
579    // becomes an issue. Maybe make Blob::delete take a list of keys?
580    //
581    // https://docs.aws.amazon.com/AmazonS3/latest/API/API_DeleteObjects.html
582    async fn delete_all(
583        blob: &dyn Blob,
584        keys: impl Iterator<Item = BlobKey>,
585        metrics: &RetryMetrics,
586        span: Span,
587        semaphore: &Semaphore,
588    ) {
589        let futures = FuturesUnordered::new();
590        for key in keys {
591            futures.push(
592                retry_external(metrics, move || {
593                    let key = key.clone();
594                    async move {
595                        let _permit = semaphore
596                            .acquire()
597                            .await
598                            .expect("acquiring permit from open semaphore");
599                        blob.delete(&key).await.map(|_| ())
600                    }
601                })
602                .instrument(span.clone()),
603            )
604        }
605
606        futures.collect().await
607    }
608}
609
610#[derive(Debug, Default)]
611pub(crate) struct GcResults {
612    pub(crate) batch_parts_deleted_from_blob: usize,
613    pub(crate) rollups_deleted_from_blob: usize,
614    pub(crate) truncated_consensus_to: Vec<SeqNo>,
615    pub(crate) rollups_removed_from_state: Vec<SeqNo>,
616}
617
618#[derive(Debug)]
619struct GcRollups {
620    rollups_lte_seqno_since: Vec<(SeqNo, PartialRollupKey)>,
621    rollup_seqnos: HashSet<SeqNo>,
622}
623
624impl GcRollups {
625    fn new(rollups_lte_seqno_since: Vec<(SeqNo, PartialRollupKey)>, gc_req: &GcReq) -> Self {
626        assert!(
627            rollups_lte_seqno_since
628                .iter()
629                .all(|(seqno, _rollup)| *seqno <= gc_req.new_seqno_since)
630        );
631        let rollup_seqnos = rollups_lte_seqno_since.iter().map(|(x, _)| *x).collect();
632        Self {
633            rollups_lte_seqno_since,
634            rollup_seqnos,
635        }
636    }
637
638    fn contains_seqno(&self, seqno: &SeqNo) -> bool {
639        self.rollup_seqnos.contains(seqno)
640    }
641
642    /// Returns the seqnos we can safely truncate state to when performing
643    /// incremental GC (all rollups with seqnos <= seqno_since).
644    fn truncate_seqnos(&self) -> impl Iterator<Item = SeqNo> + '_ {
645        self.rollups_lte_seqno_since
646            .iter()
647            .map(|(seqno, _rollup)| *seqno)
648    }
649
650    /// Returns the rollups we can safely remove from state (all rollups
651    /// `<` than the latest rollup `<=` seqno_since).
652    ///
653    /// See the full explanation in [crate::internal::state_versions::StateVersions]
654    /// for how this is derived.
655    fn rollups_to_remove_from_state(&self) -> &[(SeqNo, PartialRollupKey)] {
656        match self.rollups_lte_seqno_since.split_last() {
657            None => &[],
658            Some((_rollup_to_keep, rollups_to_remove_from_state)) => rollups_to_remove_from_state,
659        }
660    }
661}