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}