MCPcopy Create free account
hub / github.com/MaterializeInc/materialize / from

Method from

src/persist-client/src/usage.rs:549–662  ·  view source on GitHub ↗
(x: ShardUsageCumulativeMaybeRacy<'_>)

Source from the content-addressed store, hash-verified

547
548impl From<ShardUsageCumulativeMaybeRacy<'_>> for ShardUsageAudit {
549 fn from(x: ShardUsageCumulativeMaybeRacy<'_>) -> Self {
550 let mut not_leaked_bytes = 0;
551 let mut total_bytes = 0;
552 for (writer_key, bytes) in x.blob_usage.by_writer.iter() {
553 total_bytes += *bytes;
554 let writer_key_is_live = *writer_key >= x.minimum_key;
555 if writer_key_is_live {
556 not_leaked_bytes += *bytes;
557 } else {
558 // This writer is no longer live, so it can never again link
559 // anything into state. As a result, we know that anything it
560 // hasn't linked into state is now leaked and eligible for
561 // reclamation by a (future) leaked blob detector.
562 let writer_referenced =
563 x.referenced_batches_bytes.get(writer_key).map_or(0, |x| *x);
564 // It's possible, due to races, that a writer has more
565 // referenced batches in state than we saw for that writer in
566 // blob. Cap it at the number of bytes we saw in blob, otherwise
567 // we could hit the "blob inputs should be cumulative" panic
568 // below.
569 not_leaked_bytes += std::cmp::min(*bytes, writer_referenced);
570 }
571 }
572 // For now, assume rollups aren't leaked. We could compute which rollups
573 // are leaked by plumbing things more precisely, if that's necessary.
574 total_bytes += x.blob_usage.rollup_bytes;
575 not_leaked_bytes += x.blob_usage.rollup_bytes;
576
577 let leaked_bytes = total_bytes
578 .checked_sub(not_leaked_bytes)
579 .expect("blob inputs should be cumulative");
580 let referenced_batches_bytes = x.referenced_batches_bytes.values().sum::<u64>();
581 let referenced_bytes = referenced_batches_bytes + x.referenced_other_bytes;
582 let mut referenced_not_current_state_bytes = referenced_bytes
583 .checked_sub(x.current_state_bytes)
584 .expect("state inputs should be cumulative");
585 let mut current_state_rollups_bytes = x
586 .current_state_bytes
587 .checked_sub(x.current_state_batches_bytes)
588 .expect("state inputs should be cumulative");
589 let mut current_state_batches_bytes = x.current_state_batches_bytes;
590
591 // If we could transactionally read both blob and consensus, the
592 // cumulative numbers would all line up. We can't, so we have to adjust
593 // them up a bit to account for the race condition. We read blob first,
594 // and then consensus, but the race could go either way: a blob that is
595 // currently in state could be deleted from both in between the reads,
596 // OR a blob could be written and linked into state in between the
597 // reads. We could do a blob-state-blob sandwich, and then use
598 // differences between the two blob reads to reason about what
599 // specifically happens in a race, but this: (a) takes memory
600 // proportional to `O(blobs)` and (b) is overkill. Instead, we adjust by
601 // category.
602 //
603 // In the event of a discrepancy, we ensure that numbers will only get
604 // smaller (by policy, we prefer to under-count for billing).
605 // Concretely:
606 // - If referenced_bytes (which comes from state) is > not_leaked_bytes

Callers

nothing calls this directly

Calls 6

expectMethod · 0.80
saturating_subMethod · 0.80
iterMethod · 0.45
getMethod · 0.45
checked_subMethod · 0.45
valuesMethod · 0.45

Tested by

no test coverage detected