1use super::durable_cache::{BindingCacheKey, ParentNameCacheKey};
12use super::manifest_index;
13use super::view::DIRECTORY_PAGE_RAW_SCAN_LIMIT;
14use super::visibility::{self, MetadataVisibilityReads};
15use super::{
16 DirentryBindRecord, InodeRecord, MetadataView, RecoverableDeletion, ResolvedVisiblePath,
17 RevisionRecord, SubtreeTombstoneRecord,
18};
19use crate::error::CoreError;
20use loonfs_api::wire::manifest::{MetadataRow, MetadataTableFamily};
21use loonfs_api::{AbsolutePath, ChangeSeq, InodeId, InodeKind, NameKey, ROOT_INODE_ID};
22use loonfs_objectstore::ObjectStore;
23use std::collections::{HashMap, HashSet, VecDeque};
24
25impl<'a, 'store, S: ObjectStore + ?Sized> MetadataView<'a, 'store, S> {
26 pub(crate) fn session(self) -> MetadataViewSession<'a, 'store, S> {
29 MetadataViewSession::new(self)
30 }
31
32 fn tail_direntry_bind_page_candidates(
33 &self,
34 parent_inode_id: InodeId,
35 start_after_name_key: Option<&str>,
36 ) -> Vec<DirentryBindPageCandidate> {
37 let mut candidates = self
38 .row_states()
39 .flat_map(|state| state.direntry_binds())
40 .filter(|direntry| {
41 direntry.parent_inode_id == parent_inode_id
42 && start_after_name_key
43 .map(|last_name_key| direntry.name_key.as_str() > last_name_key)
44 .unwrap_or(true)
45 })
46 .map(|record| DirentryBindPageCandidate {
47 row_key: direntry_bind_row_key(record),
48 record: record.clone(),
49 })
50 .collect::<Vec<_>>();
51 candidates.sort_by(|left, right| left.row_key.cmp(&right.row_key));
52 candidates
53 }
54}
55
56#[derive(Debug, Clone)]
57pub(crate) struct VisibleChildEntry {
58 pub(crate) binding: DirentryBindRecord,
59 pub(crate) inode: InodeRecord,
60}
61
62#[derive(Debug, Clone, Copy, PartialEq, Eq)]
63pub enum LeafRevisionPrefetch {
64 Prefetch,
65 Skip,
66}
67
68#[derive(Debug, Clone, Copy, Default)]
69pub(crate) struct MetadataViewSessionCounters {
70 pub(crate) visible_child_calls: u64,
71 pub(crate) visible_inode_calls: u64,
72 pub(crate) current_parent_binding_calls: u64,
73 pub(crate) covering_tombstone_calls: u64,
74 pub(crate) latest_revision_calls: u64,
75 pub(crate) direntry_child_scan_calls: u64,
76 pub(crate) scan_prefix_calls: u64,
77 pub(crate) scan_range_page_calls: u64,
78 pub(crate) list_preload_unbind_range_scans: u64,
79 pub(crate) list_preload_child_lookups: u64,
80}
81
82pub struct MetadataViewSession<'a, 'store, S: ObjectStore + ?Sized> {
87 base: MetadataView<'a, 'store, S>,
88 inode_at_seq_cache: HashMap<InodeId, Option<InodeRecord>>,
89 visible_inode_cache: HashMap<InodeId, Option<InodeRecord>>,
90 bound_child_cache: HashMap<ParentNameCacheKey, Option<DirentryBindRecord>>,
91 current_parent_binding_cache: HashMap<InodeId, Option<DirentryBindRecord>>,
92 latest_parent_binding_cache: HashMap<InodeId, Option<DirentryBindRecord>>,
93 latest_revision_head_cache: HashMap<InodeId, Option<RevisionRecord>>,
94 active_tombstone_cache: HashMap<InodeId, Option<SubtreeTombstoneRecord>>,
95 covering_tombstone_cache: HashMap<InodeId, Option<SubtreeTombstoneRecord>>,
96 unbind_cache: HashMap<BindingCacheKey, bool>,
97 counters: MetadataViewSessionCounters,
98}
99
100impl<'a, 'store, S: ObjectStore + ?Sized> MetadataViewSession<'a, 'store, S> {
101 fn new(base: MetadataView<'a, 'store, S>) -> Self {
102 Self {
103 base,
104 inode_at_seq_cache: HashMap::new(),
105 visible_inode_cache: HashMap::new(),
106 bound_child_cache: HashMap::new(),
107 current_parent_binding_cache: HashMap::new(),
108 latest_parent_binding_cache: HashMap::new(),
109 latest_revision_head_cache: HashMap::new(),
110 active_tombstone_cache: HashMap::new(),
111 covering_tombstone_cache: HashMap::new(),
112 unbind_cache: HashMap::new(),
113 counters: MetadataViewSessionCounters::default(),
114 }
115 }
116
117 pub(crate) fn counters(&self) -> MetadataViewSessionCounters {
118 self.counters
119 }
120
121 pub(crate) async fn visible_children_page_by_name_key(
122 &mut self,
123 parent_inode_id: InodeId,
124 start_after_name_key: Option<&str>,
125 limit: usize,
126 ) -> Result<Vec<VisibleChildEntry>, CoreError> {
127 if limit == 0 {
128 return Ok(Vec::new());
129 }
130 let Some(parent) = self.visible_inode(parent_inode_id).await? else {
131 return Ok(Vec::new());
132 };
133 if parent.inode_kind != InodeKind::Directory {
134 return Ok(Vec::new());
135 }
136
137 let raw_scan_limit = limit.max(DIRECTORY_PAGE_RAW_SCAN_LIMIT);
138 let mut stream = DirentryBindNameGroupStream::new(
139 self.base
140 .tail_direntry_bind_page_candidates(parent_inode_id, start_after_name_key),
141 self.base.manifest_tables().is_none(),
142 );
143 let mut children = Vec::with_capacity(limit);
144 'pages: while children.len() < limit {
145 let groups = self
146 .next_candidate_name_groups(
147 &mut stream,
148 parent_inode_id,
149 start_after_name_key,
150 raw_scan_limit,
151 )
152 .await?;
153 if groups.is_empty() {
154 break;
155 }
156 self.preload_group_visibility(parent_inode_id, &groups)
157 .await?;
158 for group in &groups {
159 let Some(active) = self.visible_child(parent_inode_id, &group.name_key).await?
163 else {
164 continue;
165 };
166 let Some(inode) = self.visible_inode(active.child_inode_id).await? else {
167 continue;
168 };
169 children.push(VisibleChildEntry {
170 binding: active,
171 inode,
172 });
173 if children.len() == limit {
174 break 'pages;
175 }
176 }
177 }
178 Ok(children)
179 }
180
181 async fn next_candidate_name_groups(
186 &mut self,
187 stream: &mut DirentryBindNameGroupStream,
188 parent_inode_id: InodeId,
189 start_after_name_key: Option<&str>,
190 group_limit: usize,
191 ) -> Result<Vec<DirentryBindNameGroup>, CoreError> {
192 let mut groups: Vec<DirentryBindNameGroup> = Vec::new();
193 loop {
194 if stream.manifest_candidates.is_empty() && !stream.manifest_exhausted {
195 self.counters.scan_range_page_calls =
196 self.counters.scan_range_page_calls.saturating_add(1);
197 let page = if let Some(tables) = self.base.manifest_tables() {
198 manifest_index::direntry_binds_for_parent_name_key_page(
199 tables,
200 parent_inode_id,
201 start_after_name_key,
202 stream.manifest_after_row_key.as_deref(),
203 group_limit.max(DIRECTORY_PAGE_RAW_SCAN_LIMIT),
204 )
205 .await?
206 } else {
207 Vec::new()
208 };
209 if page.is_empty() {
210 stream.manifest_exhausted = true;
211 } else {
212 stream.manifest_after_row_key =
213 page.last().map(|candidate| candidate.row_key.clone());
214 stream
215 .manifest_candidates
216 .extend(page.into_iter().map(|candidate| DirentryBindPageCandidate {
217 row_key: candidate.row_key,
218 record: candidate.record,
219 }));
220 }
221 continue;
222 }
223
224 let candidate = if let Some(pushed_back) = stream.pushed_back.take() {
225 pushed_back
226 } else {
227 let next_manifest = stream.manifest_candidates.front();
228 let next_tail = stream.tail_candidates.get(stream.tail_index);
229 let take_tail = match (next_manifest, next_tail) {
230 (Some(manifest), Some(tail)) => tail.row_key < manifest.row_key,
231 (None, Some(_)) => true,
232 (Some(_), None) => false,
233 (None, None) => {
234 if stream.manifest_exhausted {
235 break;
236 }
237 continue;
238 }
239 };
240 if take_tail {
241 let candidate = stream.tail_candidates[stream.tail_index].clone();
242 stream.tail_index += 1;
243 candidate
244 } else {
245 stream
246 .manifest_candidates
247 .pop_front()
248 .expect("manifest candidate should exist")
249 }
250 };
251
252 match groups.last_mut() {
253 Some(group) if group.name_key == candidate.record.name_key => {
254 group.rows.push(candidate.record);
255 }
256 _ => {
257 if groups.len() == group_limit {
260 stream.pushed_back = Some(candidate);
261 break;
262 }
263 groups.push(DirentryBindNameGroup {
264 name_key: candidate.record.name_key.clone(),
265 rows: vec![candidate.record],
266 });
267 }
268 }
269 }
270 Ok(groups)
271 }
272
273 async fn preload_group_visibility(
280 &mut self,
281 parent_inode_id: InodeId,
282 groups: &[DirentryBindNameGroup],
283 ) -> Result<(), CoreError> {
284 let visible_seq = self.base.visible_seq();
285 let mut latest_binds = Vec::with_capacity(groups.len());
286 for group in groups {
287 let latest = group
288 .rows
289 .iter()
290 .filter(|direntry| direntry.bind_seq <= visible_seq)
291 .max_by_key(|direntry| (direntry.bind_seq, direntry.bind_delta_index))
292 .cloned();
293 self.bound_child_cache.insert(
294 ParentNameCacheKey {
295 parent_inode_id,
296 name_key: group.name_key.clone(),
297 },
298 latest.clone(),
299 );
300 if let Some(latest) = latest {
301 latest_binds.push(latest);
302 }
303 }
304 let (Some(first_group), Some(last_group)) = (groups.first(), groups.last()) else {
305 return Ok(());
306 };
307
308 self.counters.list_preload_unbind_range_scans = self
309 .counters
310 .list_preload_unbind_range_scans
311 .saturating_add(1);
312 let unbinds = self
313 .base
314 .direntry_unbinds_for_parent_name_range(
315 parent_inode_id,
316 &first_group.name_key,
317 &last_group.name_key,
318 )
319 .await?;
320 let unbound_identities: HashSet<BindingCacheKey> = unbinds
321 .iter()
322 .filter(|unbind| unbind.unbind_seq <= visible_seq)
323 .map(BindingCacheKey::from)
324 .collect();
325 for direntry in &latest_binds {
326 let cache_key = BindingCacheKey::from(direntry);
327 let unbound = unbound_identities.contains(&cache_key);
328 self.unbind_cache.insert(cache_key, unbound);
329 }
330
331 let mut pending_child_ids: Vec<InodeId> = latest_binds
332 .iter()
333 .map(|direntry| direntry.child_inode_id)
334 .filter(|child_inode_id| {
335 !(self.inode_at_seq_cache.contains_key(child_inode_id)
336 && self
337 .latest_parent_binding_cache
338 .contains_key(child_inode_id)
339 && self.active_tombstone_cache.contains_key(child_inode_id))
340 })
341 .collect();
342 pending_child_ids.sort_unstable();
343 pending_child_ids.dedup();
344 if pending_child_ids.is_empty() {
345 return Ok(());
346 }
347 self.counters.list_preload_child_lookups = self
348 .counters
349 .list_preload_child_lookups
350 .saturating_add(pending_child_ids.len() as u64);
351
352 let base = &self.base;
353 let lookups = futures::future::try_join_all(pending_child_ids.iter().map(
354 |&child_inode_id| async move {
355 let (inode, bindings, tombstones) = futures::try_join!(
356 base.inode_at_seq(child_inode_id),
357 base.direntry_binds_for_child(child_inode_id),
358 base.tombstones_for_root(child_inode_id),
359 )?;
360 Ok::<_, CoreError>((child_inode_id, inode, bindings, tombstones))
361 },
362 ))
363 .await?;
364
365 for (child_inode_id, inode, bindings, tombstones) in lookups {
366 self.inode_at_seq_cache.insert(child_inode_id, inode);
367 let latest_binding = bindings
368 .iter()
369 .filter(|direntry| direntry.bind_seq <= visible_seq)
370 .max_by_key(|direntry| (direntry.bind_seq, direntry.bind_delta_index))
371 .cloned();
372 if let Some(latest_binding) = &latest_binding {
373 if latest_binding.parent_inode_id == parent_inode_id
377 && latest_binding.name_key.as_str() >= first_group.name_key.as_str()
378 && latest_binding.name_key.as_str() <= last_group.name_key.as_str()
379 {
380 let cache_key = BindingCacheKey::from(latest_binding);
381 let unbound = unbound_identities.contains(&cache_key);
382 self.unbind_cache.entry(cache_key).or_insert(unbound);
383 }
384 }
385 self.latest_parent_binding_cache
386 .insert(child_inode_id, latest_binding);
387 let active_tombstone =
388 super::rows::active_tombstone_from_records(tombstones.iter().cloned(), visible_seq);
389 self.active_tombstone_cache
390 .insert(child_inode_id, active_tombstone);
391 }
392 Ok(())
393 }
394
395 pub async fn resolve_visible_path(
408 &mut self,
409 absolute_path: &AbsolutePath,
410 prefetch_leaf_revision: LeafRevisionPrefetch,
411 ) -> Result<ResolvedVisiblePath, CoreError> {
412 self.preload_path_walk(absolute_path, prefetch_leaf_revision)
413 .await?;
414 visibility::resolve_visible_path(self, absolute_path).await
415 }
416
417 async fn preload_path_walk(
425 &mut self,
426 absolute_path: &AbsolutePath,
427 prefetch_leaf_revision: LeafRevisionPrefetch,
428 ) -> Result<(), CoreError> {
429 let visible_seq = self.base.visible_seq();
430 let component_name_keys: Vec<NameKey> = absolute_path
431 .components()
432 .iter()
433 .map(|component| NameKey::for_display_name(&component.to_display_name()))
434 .collect();
435
436 let mut current_inode_id = ROOT_INODE_ID;
437 let mut pending_binding: Option<DirentryBindRecord> = None;
438 for wave in 0..=component_name_keys.len() {
439 let lookup_name_key = component_name_keys.get(wave);
440 let is_leaf_wave = wave == component_name_keys.len();
441
442 let want_inode = !self.inode_at_seq_cache.contains_key(¤t_inode_id);
443 let want_tombstone = !self.active_tombstone_cache.contains_key(¤t_inode_id);
444 let want_parent_binding = !self
445 .latest_parent_binding_cache
446 .contains_key(¤t_inode_id);
447 let arrived_by_binding = pending_binding.take();
448 let unbind_lookup = arrived_by_binding
449 .as_ref()
450 .filter(|binding| {
451 !self
452 .unbind_cache
453 .contains_key(&BindingCacheKey::from(*binding))
454 })
455 .cloned();
456 let bound_child_lookup = lookup_name_key
457 .filter(|name_key| {
458 !self.bound_child_cache.contains_key(&ParentNameCacheKey {
459 parent_inode_id: current_inode_id,
460 name_key: (*name_key).clone(),
461 })
462 })
463 .cloned();
464 let want_revision = is_leaf_wave
465 && prefetch_leaf_revision == LeafRevisionPrefetch::Prefetch
466 && !self
467 .latest_revision_head_cache
468 .contains_key(¤t_inode_id);
469
470 let base = &self.base;
471 let (inode, tombstones, child_bindings, unbinds, bound_rows, revision) = futures::try_join!(
472 async {
473 match want_inode {
474 true => base.inode_at_seq(current_inode_id).await.map(Some),
475 false => Ok(None),
476 }
477 },
478 async {
479 match want_tombstone {
480 true => base.tombstones_for_root(current_inode_id).await.map(Some),
481 false => Ok(None),
482 }
483 },
484 async {
485 match want_parent_binding {
486 true => base
487 .direntry_binds_for_child(current_inode_id)
488 .await
489 .map(Some),
490 false => Ok(None),
491 }
492 },
493 async {
494 match &unbind_lookup {
495 Some(binding) => base.direntry_unbinds_for_binding(binding).await.map(Some),
496 None => Ok(None),
497 }
498 },
499 async {
500 match &bound_child_lookup {
501 Some(name_key) => base
502 .direntry_binds_for_parent_name(current_inode_id, name_key)
503 .await
504 .map(Some),
505 None => Ok(None),
506 }
507 },
508 async {
509 match want_revision {
510 true => base
511 .latest_revision_record(current_inode_id)
512 .await
513 .map(Some),
514 false => Ok(None),
515 }
516 },
517 )?;
518
519 if let Some(inode) = inode {
520 self.inode_at_seq_cache.insert(current_inode_id, inode);
521 }
522 if let Some(tombstones) = tombstones {
523 let active = super::rows::active_tombstone_from_records(
524 tombstones.iter().cloned(),
525 visible_seq,
526 );
527 self.active_tombstone_cache.insert(current_inode_id, active);
528 }
529 if let Some(bindings) = child_bindings {
530 let latest = bindings
531 .iter()
532 .filter(|direntry| direntry.bind_seq <= visible_seq)
533 .max_by_key(|direntry| (direntry.bind_seq, direntry.bind_delta_index))
534 .cloned();
535 self.latest_parent_binding_cache
536 .insert(current_inode_id, latest);
537 }
538 if let Some(revision) = revision {
539 self.latest_revision_head_cache
540 .insert(current_inode_id, revision);
541 }
542 if let Some(binding) = &arrived_by_binding {
543 let cache_key = BindingCacheKey::from(binding);
544 let unbound = match unbinds {
545 Some(rows) => {
546 let unbound = rows.iter().any(|unbind| unbind.unbind_seq <= visible_seq);
547 self.unbind_cache.insert(cache_key, unbound);
548 unbound
549 }
550 None => self.unbind_cache.get(&cache_key).copied().unwrap_or(false),
551 };
552 if unbound {
553 break;
556 }
557 }
558
559 let Some(name_key) = lookup_name_key else {
560 break;
561 };
562 let bound = match bound_rows {
563 Some(rows) => {
564 let latest = rows
565 .iter()
566 .filter(|direntry| direntry.bind_seq <= visible_seq)
567 .max_by_key(|direntry| (direntry.bind_seq, direntry.bind_delta_index))
568 .cloned();
569 self.bound_child_cache.insert(
570 ParentNameCacheKey {
571 parent_inode_id: current_inode_id,
572 name_key: name_key.clone(),
573 },
574 latest.clone(),
575 );
576 latest
577 }
578 None => self
579 .bound_child_cache
580 .get(&ParentNameCacheKey {
581 parent_inode_id: current_inode_id,
582 name_key: name_key.clone(),
583 })
584 .cloned()
585 .flatten(),
586 };
587 let Some(binding) = bound else {
588 break;
589 };
590 current_inode_id = binding.child_inode_id;
591 pending_binding = Some(binding);
592 }
593 Ok(())
594 }
595
596 pub(crate) async fn visible_child(
597 &mut self,
598 parent_inode_id: InodeId,
599 name_key: &NameKey,
600 ) -> Result<Option<DirentryBindRecord>, CoreError> {
601 self.counters.visible_child_calls = self.counters.visible_child_calls.saturating_add(1);
602 visibility::visible_child(self, parent_inode_id, name_key).await
603 }
604
605 pub async fn visible_inode(
609 &mut self,
610 inode_id: InodeId,
611 ) -> Result<Option<InodeRecord>, CoreError> {
612 self.counters.visible_inode_calls = self.counters.visible_inode_calls.saturating_add(1);
613 if let Some(cached) = self.visible_inode_cache.get(&inode_id).cloned() {
614 return Ok(cached);
615 }
616
617 let visible = visibility::visible_inode(self, inode_id).await?;
618 self.visible_inode_cache.insert(inode_id, visible.clone());
619 Ok(visible)
620 }
621
622 pub(crate) async fn inode_at_seq(
623 &mut self,
624 inode_id: InodeId,
625 ) -> Result<Option<InodeRecord>, CoreError> {
626 if let Some(cached) = self.inode_at_seq_cache.get(&inode_id).cloned() {
627 return Ok(cached);
628 }
629 let inode = self.base.inode_at_seq(inode_id).await?;
630 self.inode_at_seq_cache.insert(inode_id, inode.clone());
631 Ok(inode)
632 }
633
634 pub async fn latest_revision_head_of_visible(
643 &mut self,
644 inode_id: InodeId,
645 ) -> Result<Option<RevisionRecord>, CoreError> {
646 self.counters.latest_revision_calls = self.counters.latest_revision_calls.saturating_add(1);
647 if let Some(cached) = self.latest_revision_head_cache.get(&inode_id).cloned() {
648 return Ok(cached);
649 }
650 let revision = self.base.latest_revision_record(inode_id).await?;
651 self.latest_revision_head_cache
652 .insert(inode_id, revision.clone());
653 Ok(revision)
654 }
655
656 pub(crate) async fn revisions_for_inode_page_desc(
657 &mut self,
658 inode_id: InodeId,
659 start_after: Option<manifest_index::RevisionPagePosition>,
660 limit: usize,
661 ) -> Result<Vec<RevisionRecord>, CoreError> {
662 self.counters.scan_range_page_calls = self.counters.scan_range_page_calls.saturating_add(1);
663 self.base
664 .revisions_for_inode_page_desc(inode_id, start_after, limit)
665 .await
666 }
667
668 pub async fn current_parent_binding_for_child(
672 &mut self,
673 child_inode_id: InodeId,
674 ) -> Result<Option<DirentryBindRecord>, CoreError> {
675 self.counters.current_parent_binding_calls =
676 self.counters.current_parent_binding_calls.saturating_add(1);
677 if let Some(cached) = self
678 .current_parent_binding_cache
679 .get(&child_inode_id)
680 .cloned()
681 {
682 return Ok(cached);
683 }
684
685 let binding = visibility::current_parent_binding_for_child(self, child_inode_id).await?;
686 self.current_parent_binding_cache
687 .insert(child_inode_id, binding.clone());
688 Ok(binding)
689 }
690
691 async fn latest_parent_binding_for_child(
696 &mut self,
697 child_inode_id: InodeId,
698 ) -> Result<Option<DirentryBindRecord>, CoreError> {
699 if let Some(cached) = self
700 .latest_parent_binding_cache
701 .get(&child_inode_id)
702 .cloned()
703 {
704 return Ok(cached);
705 }
706 self.counters.direntry_child_scan_calls =
707 self.counters.direntry_child_scan_calls.saturating_add(1);
708 self.counters.scan_prefix_calls = self.counters.scan_prefix_calls.saturating_add(1);
709 let bindings = self.base.direntry_binds_for_child(child_inode_id).await?;
710 let latest = bindings
711 .iter()
712 .filter(|direntry| direntry.bind_seq <= self.base.visible_seq())
713 .max_by_key(|direntry| (direntry.bind_seq, direntry.bind_delta_index))
714 .cloned();
715 self.latest_parent_binding_cache
716 .insert(child_inode_id, latest.clone());
717 Ok(latest)
718 }
719
720 pub(crate) async fn covering_subtree_tombstone(
721 &mut self,
722 inode_id: InodeId,
723 ) -> Result<Option<SubtreeTombstoneRecord>, CoreError> {
724 self.counters.covering_tombstone_calls =
725 self.counters.covering_tombstone_calls.saturating_add(1);
726 if let Some(cached) = self.covering_tombstone_cache.get(&inode_id).cloned() {
727 return Ok(cached);
728 }
729
730 let tombstone = visibility::covering_subtree_tombstone(self, inode_id).await?;
731 self.covering_tombstone_cache
732 .insert(inode_id, tombstone.clone());
733 Ok(tombstone)
734 }
735
736 pub(crate) async fn active_deletions_page(
739 &mut self,
740 start_after: Option<(ChangeSeq, InodeId)>,
741 limit: usize,
742 ) -> Result<Vec<RecoverableDeletion>, CoreError> {
743 self.counters.scan_range_page_calls = self.counters.scan_range_page_calls.saturating_add(1);
744 self.base.active_deletions_page(start_after, limit).await
745 }
746
747 pub(crate) async fn active_subtree_tombstone(
748 &mut self,
749 root_inode_id: InodeId,
750 ) -> Result<Option<SubtreeTombstoneRecord>, CoreError> {
751 if let Some(cached) = self.active_tombstone_cache.get(&root_inode_id).cloned() {
752 return Ok(cached);
753 }
754 self.counters.scan_prefix_calls = self.counters.scan_prefix_calls.saturating_add(1);
755 let tombstones = self.base.tombstones_for_root(root_inode_id).await?;
756 let tombstone = super::rows::active_tombstone_from_records(
757 tombstones.iter().cloned(),
758 self.base.visible_seq(),
759 );
760 self.active_tombstone_cache
761 .insert(root_inode_id, tombstone.clone());
762 Ok(tombstone)
763 }
764
765 async fn bound_child(
766 &mut self,
767 parent_inode_id: InodeId,
768 name_key: &NameKey,
769 ) -> Result<Option<DirentryBindRecord>, CoreError> {
770 let cache_key = ParentNameCacheKey {
771 parent_inode_id,
772 name_key: name_key.clone(),
773 };
774 if let Some(cached) = self.bound_child_cache.get(&cache_key).cloned() {
775 return Ok(cached);
776 }
777 self.counters.scan_prefix_calls = self.counters.scan_prefix_calls.saturating_add(1);
778 let bindings = self
779 .base
780 .direntry_binds_for_parent_name(parent_inode_id, name_key)
781 .await?;
782 let binding = bindings
783 .iter()
784 .filter(|direntry| direntry.bind_seq <= self.base.visible_seq())
785 .max_by_key(|direntry| (direntry.bind_seq, direntry.bind_delta_index))
786 .cloned();
787 self.bound_child_cache.insert(cache_key, binding.clone());
788 Ok(binding)
789 }
790
791 async fn is_direntry_unbound(
792 &mut self,
793 direntry: &DirentryBindRecord,
794 ) -> Result<bool, CoreError> {
795 let cache_key = BindingCacheKey::from(direntry);
796 if let Some(cached) = self.unbind_cache.get(&cache_key).copied() {
797 return Ok(cached);
798 }
799 self.counters.scan_prefix_calls = self.counters.scan_prefix_calls.saturating_add(1);
800 let unbinds = self.base.direntry_unbinds_for_binding(direntry).await?;
801 let unbound = unbinds
802 .iter()
803 .any(|unbind| unbind.unbind_seq <= self.base.visible_seq());
804 self.unbind_cache.insert(cache_key, unbound);
805 Ok(unbound)
806 }
807}
808
809impl<S: ObjectStore + ?Sized> MetadataVisibilityReads for MetadataViewSession<'_, '_, S> {
816 type Error = CoreError;
817
818 async fn find_inode(&mut self, inode_id: InodeId) -> Result<Option<InodeRecord>, Self::Error> {
819 self.inode_at_seq(inode_id).await
820 }
821
822 async fn find_latest_bound_child(
823 &mut self,
824 parent_inode_id: InodeId,
825 name_key: &NameKey,
826 ) -> Result<Option<DirentryBindRecord>, Self::Error> {
827 self.bound_child(parent_inode_id, name_key).await
828 }
829
830 async fn find_latest_parent_binding_for_child(
831 &mut self,
832 child_inode_id: InodeId,
833 ) -> Result<Option<DirentryBindRecord>, Self::Error> {
834 self.latest_parent_binding_for_child(child_inode_id).await
835 }
836
837 async fn find_active_subtree_tombstone(
838 &mut self,
839 root_inode_id: InodeId,
840 ) -> Result<Option<SubtreeTombstoneRecord>, Self::Error> {
841 self.active_subtree_tombstone(root_inode_id).await
842 }
843
844 async fn is_binding_unbound(
845 &mut self,
846 direntry: &DirentryBindRecord,
847 ) -> Result<bool, Self::Error> {
848 self.is_direntry_unbound(direntry).await
849 }
850
851 async fn current_parent_binding_for_child(
852 &mut self,
853 child_inode_id: InodeId,
854 ) -> Result<Option<DirentryBindRecord>, Self::Error> {
855 MetadataViewSession::current_parent_binding_for_child(self, child_inode_id).await
856 }
857
858 async fn covering_subtree_tombstone(
859 &mut self,
860 inode_id: InodeId,
861 ) -> Result<Option<SubtreeTombstoneRecord>, Self::Error> {
862 MetadataViewSession::covering_subtree_tombstone(self, inode_id).await
863 }
864
865 async fn visible_inode(
866 &mut self,
867 inode_id: InodeId,
868 ) -> Result<Option<InodeRecord>, Self::Error> {
869 MetadataViewSession::visible_inode(self, inode_id).await
870 }
871}
872
873struct DirentryBindNameGroupStream {
876 manifest_after_row_key: Option<String>,
877 manifest_exhausted: bool,
878 manifest_candidates: VecDeque<DirentryBindPageCandidate>,
879 tail_candidates: Vec<DirentryBindPageCandidate>,
880 tail_index: usize,
881 pushed_back: Option<DirentryBindPageCandidate>,
882}
883
884impl DirentryBindNameGroupStream {
885 fn new(tail_candidates: Vec<DirentryBindPageCandidate>, manifest_exhausted: bool) -> Self {
886 Self {
887 manifest_after_row_key: None,
888 manifest_exhausted,
889 manifest_candidates: VecDeque::new(),
890 tail_candidates,
891 tail_index: 0,
892 pushed_back: None,
893 }
894 }
895}
896
897struct DirentryBindNameGroup {
899 name_key: NameKey,
900 rows: Vec<DirentryBindRecord>,
901}
902
903#[derive(Clone)]
904struct DirentryBindPageCandidate {
905 row_key: String,
906 record: DirentryBindRecord,
907}
908
909fn direntry_bind_row_key(record: &DirentryBindRecord) -> String {
910 MetadataRow::DirentryBind {
911 parent_inode_id: record.parent_inode_id,
912 name_key: record.name_key.clone(),
913 display_name: record.display_name.clone(),
914 child_inode_id: record.child_inode_id,
915 bind_seq: record.bind_seq,
916 bind_delta_index: record.bind_delta_index,
917 }
918 .row_key_for_family(MetadataTableFamily::DirentryBinds)
919}