Skip to main content

loonfs_core/metadata/
view_session.rs

1//! [`MetadataViewSession`]: one read operation's memo over a
2//! [`MetadataView`](super::view::MetadataView).
3//!
4//! A session caches the primitive lookups a multi-step read repeats — path
5//! walks, listing pages, grep candidate walks — and layers the directory
6//! page stream that merges manifest candidates with the WAL tail. Every
7//! visibility decision still routes through the canonical
8//! [`super::visibility`] rule bodies; the session only memoizes their
9//! primitive inputs.
10
11use 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    /// Opens a fresh session over this view; the view is `Copy`, so the
27    /// session owns its copy.
28    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
82/// One loaded-view session with memoized visibility reads.
83///
84/// This is part of the read surface consumed by `loonfs-grep`; construction
85/// stays on the loaded view so callers cannot assemble incoherent sessions.
86pub 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                // One group per name key: the stream already deduplicated
160                // and ordered candidates, and the preload seeded the caches
161                // the canonical visibility rules read through.
162                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    /// Pulls up to `group_limit` complete name-key groups from the merged
182    /// manifest+tail candidate stream, in row-key order. A group carries
183    /// every bind row for its name, so the group's newest visible row IS the
184    /// latest bound child for that name.
185    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                    // A full batch closes only at a name boundary, so the
258                    // finished group is guaranteed complete.
259                    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    /// Seeds the session caches for one wave of name groups: the latest bind
274    /// per name from rows the stream already carried, unbind facts from one
275    /// range scan, and the child-keyed lookups batched concurrently. The
276    /// canonical visibility rules then decide over cache hits; anything the
277    /// preload cannot answer (cross-directory binding chases) falls back to
278    /// the ordinary per-key scans.
279    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                // A child bound in this directory within the preloaded name
374                // range gets its unbind fact from the range scan; bindings
375                // elsewhere fall back to the ordinary per-binding scan.
376                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    /// Resolves `absolute_path` with the canonical visibility rules after a
396    /// pipelined preload of the walk's storage lookups.
397    ///
398    /// For each path component, the preload issues the independent probes
399    /// (inode, tombstone, child-keyed binding, the previous binding's
400    /// unbind fact) as one concurrent wave alongside the next component's
401    /// bound-child scan — the only lookup the walk truly serializes on —
402    /// and seeds the session's caches with exactly what those lookups would
403    /// have fetched. The canonical rules then decide over cache hits, so a
404    /// cold resolution costs one round-trip wave per path component instead
405    /// of five-plus sequential lookups each.
406    /// This is part of the read surface consumed by `loonfs-grep`.
407    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    /// The preload behind [`Self::resolve_visible_path`]: batches storage
418    /// probes and seeds primitive caches, decides nothing. Seeded values are
419    /// byte-identical to what the corresponding cache-miss paths compute, so
420    /// visibility decisions are unchanged; anything the preload skips (a
421    /// renamed child's foreign binding, an unbound name) falls back to the
422    /// ordinary per-key scans. Stops probing once a component has no bound
423    /// child or its binding is revoked — the canonical walk reports those.
424    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(&current_inode_id);
443            let want_tombstone = !self.active_tombstone_cache.contains_key(&current_inode_id);
444            let want_parent_binding = !self
445                .latest_parent_binding_cache
446                .contains_key(&current_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(&current_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                    // The walk dead-ends at the previous component; probing
554                    // deeper would warm caches for nothing.
555                    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    /// Returns the inode when it is visible in this session's snapshot.
606    ///
607    /// This is part of the read surface consumed by `loonfs-grep`.
608    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    /// Returns the latest revision for an inode already proven visible at
635    /// this session's seq. Unlike [`MetadataView::latest_revision_head`],
636    /// this skips re-deriving the inode's visibility: the re-check costs a
637    /// full child-binds scan and tombstone probe per entry on a wide listing,
638    /// and it cannot disagree with the enumeration that admitted the entry
639    /// at the same seq.
640    ///
641    /// This is part of the read surface consumed by `loonfs-grep`.
642    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    /// Returns the child's active parent binding at the visible sequence.
669    ///
670    /// This is part of the read surface consumed by `loonfs-grep`.
671    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    /// Latest binding whose child is `child_inode_id` at the visible seq,
692    /// regardless of whether it has since been unbound. The
693    /// [`MetadataVisibilityReads`] primitive backing the session's cached
694    /// [`Self::current_parent_binding_for_child`].
695    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    /// One page of the namespace's recoverable deletions, oldest deletion
737    /// first. Uncached: one call serves one trash listing page request.
738    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
809/// The session as a [`MetadataVisibilityReads`] source. Its primitives route
810/// through the per-session caches, and it overrides exactly the composite
811/// rules it memoizes (`visible_inode`, `current_parent_binding_for_child`,
812/// `covering_subtree_tombstone`) so a cache hit short-circuits the walk;
813/// every override's miss path, and the un-overridden composites, still decide
814/// through the canonical [`super::visibility`] rule bodies.
815impl<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
873/// Cursor state for the merged manifest+tail direntry-bind stream, grouped
874/// by name key across calls.
875struct 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
897/// Every bind row the stream carried for one name key, in row-key order.
898struct 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}