1use std::collections::BTreeMap;
18use std::fmt;
19use std::iter;
20use std::pin::Pin;
21use std::sync::Arc;
22use std::task::Context;
23use std::task::Poll;
24use std::task::ready;
25use std::vec;
26
27use either::Either;
28use futures::Stream;
29use futures::StreamExt as _;
30use futures::future::BoxFuture;
31use futures::future::try_join;
32use futures::stream::BoxStream;
33use itertools::EitherOrBoth;
34use itertools::Itertools as _;
35use pollster::FutureExt as _;
36
37use crate::backend::BackendResult;
38use crate::backend::CopyId;
39use crate::backend::MergedTreeVal;
40use crate::backend::MergedTreeValue;
41use crate::backend::MergedTreeValueExt as _;
42use crate::backend::TreeId;
43use crate::backend::TreeValue;
44use crate::conflict_labels::ConflictLabels;
45use crate::copies::CopiesTreeDiffEntry;
46use crate::copies::CopiesTreeDiffStream;
47use crate::copies::CopyHistoryDiffStream;
48use crate::copies::CopyHistoryTreeDiffEntry;
49use crate::copies::CopyRecords;
50use crate::matchers::EverythingMatcher;
51use crate::matchers::Matcher;
52use crate::merge::Diff;
53use crate::merge::Merge;
54use crate::merge::MergeBuilder;
55use crate::repo_path::RepoPath;
56use crate::repo_path::RepoPathBuf;
57use crate::repo_path::RepoPathComponent;
58use crate::store::Store;
59use crate::tree::ToTreeMergeExt as _;
60use crate::tree::Tree;
61use crate::tree::TreeMergeExt as _;
62use crate::tree_merge::merge_trees;
63
64#[derive(Clone)]
67pub struct MergedTree {
68 store: Arc<Store>,
69 tree_ids: Merge<TreeId>,
70 labels: ConflictLabels,
71}
72
73impl fmt::Debug for MergedTree {
74 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
75 f.debug_struct("MergedTree")
76 .field("tree_ids", &self.tree_ids)
77 .field("labels", &self.labels)
78 .finish_non_exhaustive()
79 }
80}
81
82impl MergedTree {
83 pub fn resolved(store: Arc<Store>, tree_id: TreeId) -> Self {
85 Self {
86 store,
87 tree_ids: Merge::resolved(tree_id),
88 labels: ConflictLabels::unlabeled(),
89 }
90 }
91
92 pub fn new(store: Arc<Store>, tree_ids: Merge<TreeId>, labels: ConflictLabels) -> Self {
94 if let Some(num_sides) = labels.num_sides() {
95 assert_eq!(tree_ids.num_sides(), num_sides);
96 }
97 Self {
98 store,
99 tree_ids,
100 labels,
101 }
102 }
103
104 pub fn store(&self) -> &Arc<Store> {
106 &self.store
107 }
108
109 pub fn tree_ids(&self) -> &Merge<TreeId> {
112 &self.tree_ids
113 }
114
115 pub fn into_tree_ids(self) -> Merge<TreeId> {
118 self.tree_ids
119 }
120
121 pub fn labels(&self) -> &ConflictLabels {
123 &self.labels
124 }
125
126 pub fn tree_ids_and_labels(&self) -> (&Merge<TreeId>, &ConflictLabels) {
130 (&self.tree_ids, &self.labels)
131 }
132
133 pub fn into_tree_ids_and_labels(self) -> (Merge<TreeId>, ConflictLabels) {
135 (self.tree_ids, self.labels)
136 }
137
138 pub async fn trees(&self) -> BackendResult<Merge<Tree>> {
140 self.tree_ids
141 .try_map_async(|id| self.store.get_tree(RepoPathBuf::root(), id))
142 .await
143 }
144
145 pub fn labels_by_term<'a>(&'a self, label: &'a str) -> Merge<&'a str> {
149 if self.tree_ids.is_resolved() {
150 assert!(!self.labels.has_labels());
151 Merge::resolved(label)
152 } else if self.labels.has_labels() {
153 let labels = self.labels.as_merge();
157 assert_eq!(labels.num_sides(), self.tree_ids.num_sides());
158 labels.map(|label| label.as_str())
159 } else {
160 Merge::repeated("", self.tree_ids.num_sides())
165 }
166 }
167
168 pub async fn resolve(self) -> BackendResult<Self> {
171 let merged = merge_trees(&self.store, self.tree_ids).await?;
172 let (simplified_labels, simplified) = if merged.is_resolved() {
177 (ConflictLabels::unlabeled(), merged)
178 } else {
179 self.labels.simplify_with(&merged)
180 };
181 if cfg!(debug_assertions) {
185 let re_merged = merge_trees(&self.store, simplified.clone()).await.unwrap();
186 debug_assert_eq!(re_merged, simplified);
187 }
188 Ok(Self {
189 store: self.store,
190 tree_ids: simplified,
191 labels: simplified_labels,
192 })
193 }
194
195 pub fn conflicts(
200 &self,
201 ) -> impl Iterator<Item = (RepoPathBuf, BackendResult<MergedTreeValue>)> + use<> {
202 self.conflicts_matching(&EverythingMatcher)
203 }
204
205 pub fn conflicts_matching<'matcher>(
207 &self,
208 matcher: &'matcher dyn Matcher,
209 ) -> impl Iterator<Item = (RepoPathBuf, BackendResult<MergedTreeValue>)> + use<'matcher> {
210 ConflictIterator::new(self, matcher)
211 }
212
213 pub fn has_conflict(&self) -> bool {
215 !self.tree_ids.is_resolved()
216 }
217
218 pub async fn path_value(&self, path: &RepoPath) -> BackendResult<MergedTreeValue> {
222 match path.split() {
223 Some((dir, basename)) => {
224 let trees = self.trees().await?;
225 match trees.sub_tree_recursive(dir).await? {
226 None => Ok(Merge::absent()),
227 Some(tree) => Ok(tree.value(basename).cloned()),
228 }
229 }
230 None => Ok(self.to_merged_tree_value()),
231 }
232 }
233
234 pub async fn copy_value(&self, id: &CopyId) -> BackendResult<Option<TreeValue>> {
237 let copy = self.store().backend().read_copy(id).await?;
238 let merged_val = self.path_value(©.current_path).await?;
239 match merged_val.into_resolved() {
240 Ok(Some(val)) if val.copy_id() == Some(id) => Ok(Some(val)),
241 _ => Ok(None),
242 }
243 }
244
245 fn to_merged_tree_value(&self) -> MergedTreeValue {
246 self.tree_ids
247 .map(|tree_id| Some(TreeValue::Tree(tree_id.clone())))
248 }
249
250 pub fn entries(&self) -> TreeEntriesIterator<'static> {
260 self.entries_matching(&EverythingMatcher)
261 }
262
263 pub fn entries_matching<'matcher>(
265 &self,
266 matcher: &'matcher dyn Matcher,
267 ) -> TreeEntriesIterator<'matcher> {
268 TreeEntriesIterator::new(self, matcher)
269 }
270
271 fn diff_stream_internal<'matcher>(
273 &self,
274 other: &Self,
275 matcher: &'matcher dyn Matcher,
276 ) -> TreeDiffStream<'matcher> {
277 let concurrency = self.store().concurrency();
278 if concurrency <= 1 {
279 futures::stream::iter(TreeDiffIterator::new(self, other, matcher)).boxed()
280 } else {
281 TreeDiffStreamImpl::new(self, other, matcher, concurrency).boxed()
282 }
283 }
284
285 pub fn diff_stream<'matcher>(
287 &self,
288 other: &Self,
289 matcher: &'matcher dyn Matcher,
290 ) -> TreeDiffStream<'matcher> {
291 stream_without_trees(self.diff_stream_internal(other, matcher))
292 }
293
294 pub fn diff_stream_with_trees<'matcher>(
296 &self,
297 other: &Self,
298 matcher: &'matcher dyn Matcher,
299 ) -> TreeDiffStream<'matcher> {
300 self.diff_stream_internal(other, matcher)
301 }
302
303 pub fn diff_stream_for_file_system<'matcher>(
306 &self,
307 other: &Self,
308 matcher: &'matcher dyn Matcher,
309 ) -> TreeDiffStream<'matcher> {
310 DiffStreamForFileSystem::new(self.diff_stream_internal(other, matcher)).boxed()
311 }
312
313 pub fn diff_stream_with_copies<'a>(
315 &self,
316 other: &Self,
317 matcher: &'a dyn Matcher,
318 copy_records: &'a CopyRecords,
319 ) -> BoxStream<'a, CopiesTreeDiffEntry> {
320 let stream = self.diff_stream(other, matcher);
321 CopiesTreeDiffStream::new(stream, self.clone(), other.clone(), copy_records).boxed()
322 }
323
324 pub fn diff_stream_with_copy_history<'a>(
326 &'a self,
327 other: &'a Self,
328 matcher: &'a dyn Matcher,
329 ) -> BoxStream<'a, CopyHistoryTreeDiffEntry> {
330 let stream = self.diff_stream(other, matcher);
331 CopyHistoryDiffStream::new(stream, self, other).boxed()
332 }
333
334 pub async fn merge(merge: Merge<(Self, String)>) -> BackendResult<Self> {
340 Self::merge_no_resolve(merge).resolve().await
341 }
342
343 pub fn merge_no_resolve(merge: Merge<(Self, String)>) -> Self {
346 debug_assert!(
347 merge
348 .iter()
349 .map(|(tree, _)| Arc::as_ptr(tree.store()))
350 .all_equal()
351 );
352 let store = merge.first().0.store().clone();
353 let flattened_labels = ConflictLabels::from_merge(
354 merge
355 .map(|(tree, label)| tree.labels_by_term(label))
356 .flatten()
357 .map(|&label| label.to_owned()),
358 );
359 let flattened_tree_ids: Merge<TreeId> = merge
360 .into_map(|(tree, _label)| tree.into_tree_ids())
361 .flatten();
362
363 let (labels, tree_ids) = flattened_labels.simplify_with(&flattened_tree_ids);
364 Self::new(store, tree_ids, labels)
365 }
366}
367
368#[derive(Debug)]
370pub struct TreeDiffEntry {
371 pub path: RepoPathBuf,
373 pub values: BackendResult<Diff<MergedTreeValue>>,
375}
376
377pub type TreeDiffStream<'matcher> = BoxStream<'matcher, TreeDiffEntry>;
381
382fn all_tree_entries(
383 trees: &Merge<Tree>,
384) -> impl Iterator<Item = (&RepoPathComponent, MergedTreeVal<'_>)> {
385 if let Some(tree) = trees.as_resolved() {
386 let iter = tree
387 .entries_non_recursive()
388 .map(|entry| (entry.name(), Merge::normal(entry.value())));
389 Either::Left(iter)
390 } else {
391 let same_change = trees.first().store().merge_options().same_change;
392 let iter = all_merged_tree_entries(trees).map(move |(name, values)| {
393 let values = match values.resolve_trivial(same_change) {
395 Some(resolved) => Merge::resolved(*resolved),
396 None => values,
397 };
398 (name, values)
399 });
400 Either::Right(iter)
401 }
402}
403
404pub fn all_merged_tree_entries(
408 trees: &Merge<Tree>,
409) -> impl Iterator<Item = (&RepoPathComponent, MergedTreeVal<'_>)> {
410 let mut entries_iters = trees
411 .iter()
412 .map(|tree| tree.entries_non_recursive().peekable())
413 .collect_vec();
414 iter::from_fn(move || {
415 let next_name = entries_iters
416 .iter_mut()
417 .filter_map(|iter| iter.peek())
418 .map(|entry| entry.name())
419 .min()?;
420 let values: MergeBuilder<_> = entries_iters
421 .iter_mut()
422 .map(|iter| {
423 let entry = iter.next_if(|entry| entry.name() == next_name)?;
424 Some(entry.value())
425 })
426 .collect();
427 Some((next_name, values.build()))
428 })
429}
430
431fn merged_tree_entry_diff<'a>(
432 trees1: &'a Merge<Tree>,
433 trees2: &'a Merge<Tree>,
434) -> impl Iterator<Item = (&'a RepoPathComponent, Diff<MergedTreeVal<'a>>)> {
435 itertools::merge_join_by(
436 all_tree_entries(trees1),
437 all_tree_entries(trees2),
438 |(name1, _), (name2, _)| name1.cmp(name2),
439 )
440 .map(|entry| match entry {
441 EitherOrBoth::Both((name, value1), (_, value2)) => (name, Diff::new(value1, value2)),
442 EitherOrBoth::Left((name, value1)) => (name, Diff::new(value1, Merge::absent())),
443 EitherOrBoth::Right((name, value2)) => (name, Diff::new(Merge::absent(), value2)),
444 })
445 .filter(|(_, diff)| diff.is_changed())
446}
447
448pub struct TreeEntriesIterator<'matcher> {
450 store: Arc<Store>,
451 stack: Vec<TreeEntriesDirItem>,
452 matcher: &'matcher dyn Matcher,
453}
454
455struct TreeEntriesDirItem {
456 entries: Vec<(RepoPathBuf, MergedTreeValue)>,
457}
458
459impl TreeEntriesDirItem {
460 fn new(trees: &Merge<Tree>, matcher: &dyn Matcher) -> Self {
461 let mut entries = vec![];
462 let dir = trees.first().dir();
463 for (name, value) in all_tree_entries(trees) {
464 let path = dir.join(name);
465 if value.is_tree() {
466 if matcher.visit(&path).is_nothing() {
468 continue;
469 }
470 } else if !matcher.matches(&path) {
471 continue;
472 }
473 entries.push((path, value.cloned()));
474 }
475 entries.reverse();
476 Self { entries }
477 }
478}
479
480impl<'matcher> TreeEntriesIterator<'matcher> {
481 fn new(trees: &MergedTree, matcher: &'matcher dyn Matcher) -> Self {
482 Self {
483 store: trees.store.clone(),
484 stack: vec![TreeEntriesDirItem {
485 entries: vec![(RepoPathBuf::root(), trees.to_merged_tree_value())],
486 }],
487 matcher,
488 }
489 }
490}
491
492impl Iterator for TreeEntriesIterator<'_> {
493 type Item = (RepoPathBuf, BackendResult<MergedTreeValue>);
494
495 fn next(&mut self) -> Option<Self::Item> {
496 while let Some(top) = self.stack.last_mut() {
497 if let Some((path, value)) = top.entries.pop() {
498 let maybe_trees = match value.to_tree_merge(&self.store, &path).block_on() {
499 Ok(maybe_trees) => maybe_trees,
500 Err(err) => return Some((path, Err(err))),
501 };
502 if let Some(trees) = maybe_trees {
503 self.stack
504 .push(TreeEntriesDirItem::new(&trees, self.matcher));
505 } else {
506 return Some((path, Ok(value)));
507 }
508 } else {
509 self.stack.pop();
510 }
511 }
512 None
513 }
514}
515
516struct ConflictsDirItem {
519 entries: Vec<(RepoPathBuf, MergedTreeValue)>,
520}
521
522impl ConflictsDirItem {
523 fn new(trees: &Merge<Tree>, matcher: &dyn Matcher) -> Self {
524 if trees.is_resolved() {
525 return Self { entries: vec![] };
526 }
527
528 let dir = trees.first().dir();
529 let mut entries = vec![];
530 for (basename, value) in all_tree_entries(trees) {
531 if value.is_resolved() {
532 continue;
533 }
534 let path = dir.join(basename);
535 if value.is_tree() {
536 if matcher.visit(&path).is_nothing() {
537 continue;
538 }
539 } else if !matcher.matches(&path) {
540 continue;
541 }
542 entries.push((path, value.cloned()));
543 }
544 entries.reverse();
545 Self { entries }
546 }
547}
548
549struct ConflictIterator<'matcher> {
550 store: Arc<Store>,
551 stack: Vec<ConflictsDirItem>,
552 matcher: &'matcher dyn Matcher,
553}
554
555impl<'matcher> ConflictIterator<'matcher> {
556 fn new(tree: &MergedTree, matcher: &'matcher dyn Matcher) -> Self {
557 Self {
558 store: tree.store().clone(),
559 stack: vec![ConflictsDirItem {
560 entries: vec![(RepoPathBuf::root(), tree.to_merged_tree_value())],
561 }],
562 matcher,
563 }
564 }
565}
566
567impl Iterator for ConflictIterator<'_> {
568 type Item = (RepoPathBuf, BackendResult<MergedTreeValue>);
569
570 fn next(&mut self) -> Option<Self::Item> {
571 while let Some(top) = self.stack.last_mut() {
572 if let Some((path, tree_values)) = top.entries.pop() {
573 match tree_values.to_tree_merge(&self.store, &path).block_on() {
574 Ok(Some(trees)) => {
575 self.stack.push(ConflictsDirItem::new(&trees, self.matcher));
577 }
578 Ok(None) => {
579 return Some((path, Ok(tree_values)));
583 }
584 Err(err) => {
585 return Some((path, Err(err)));
586 }
587 }
588 } else {
589 self.stack.pop();
590 }
591 }
592 None
593 }
594}
595
596pub struct TreeDiffIterator<'matcher> {
598 store: Arc<Store>,
599 stack: Vec<TreeDiffDir>,
600 matcher: &'matcher dyn Matcher,
601}
602
603struct TreeDiffDir {
604 entries: Vec<(RepoPathBuf, Diff<MergedTreeValue>)>,
605}
606
607impl<'matcher> TreeDiffIterator<'matcher> {
608 pub fn new(tree1: &MergedTree, tree2: &MergedTree, matcher: &'matcher dyn Matcher) -> Self {
610 assert!(Arc::ptr_eq(tree1.store(), tree2.store()));
611 let root_dir = RepoPath::root();
612 let mut stack = Vec::new();
613 let root_diff = Diff::new(tree1.to_merged_tree_value(), tree2.to_merged_tree_value());
614 if root_diff.is_changed() && !matcher.visit(root_dir).is_nothing() {
615 stack.push(TreeDiffDir {
616 entries: vec![(root_dir.to_owned(), root_diff)],
617 });
618 }
619 Self {
620 store: tree1.store().clone(),
621 stack,
622 matcher,
623 }
624 }
625
626 fn trees(
628 store: &Arc<Store>,
629 dir: &RepoPath,
630 values: &MergedTreeValue,
631 ) -> BackendResult<Merge<Tree>> {
632 if let Some(trees) = values.to_tree_merge(store, dir).block_on()? {
633 Ok(trees)
634 } else {
635 Ok(Merge::resolved(Tree::empty(store.clone(), dir.to_owned())))
636 }
637 }
638}
639
640impl TreeDiffDir {
641 fn from_trees(
642 dir: &RepoPath,
643 trees1: &Merge<Tree>,
644 trees2: &Merge<Tree>,
645 matcher: &dyn Matcher,
646 ) -> Self {
647 let mut entries = vec![];
648 for (name, diff) in merged_tree_entry_diff(trees1, trees2) {
649 let path = dir.join(name);
650 let tree_before = diff.before.is_tree();
651 let tree_after = diff.after.is_tree();
652 let tree_matches = (tree_before || tree_after) && !matcher.visit(&path).is_nothing();
655 let file_matches = (!tree_before || !tree_after) && matcher.matches(&path);
656
657 let before = if (tree_before && tree_matches) || (!tree_before && file_matches) {
659 diff.before
660 } else {
661 Merge::absent()
662 };
663 let after = if (tree_after && tree_matches) || (!tree_after && file_matches) {
664 diff.after
665 } else {
666 Merge::absent()
667 };
668 if before.is_absent() && after.is_absent() {
669 continue;
670 }
671 entries.push((path, Diff::new(before.cloned(), after.cloned())));
672 }
673 entries.reverse();
674 Self { entries }
675 }
676}
677
678impl Iterator for TreeDiffIterator<'_> {
679 type Item = TreeDiffEntry;
680
681 fn next(&mut self) -> Option<Self::Item> {
682 while let Some(top) = self.stack.last_mut() {
683 let Some((path, diff)) = top.entries.pop() else {
684 self.stack.pop().unwrap();
685 continue;
686 };
687
688 if diff.before.is_tree() || diff.after.is_tree() {
689 let (before_tree, after_tree) = match (
690 Self::trees(&self.store, &path, &diff.before),
691 Self::trees(&self.store, &path, &diff.after),
692 ) {
693 (Ok(before_tree), Ok(after_tree)) => (before_tree, after_tree),
694 (Err(before_err), _) => {
695 return Some(TreeDiffEntry {
696 path,
697 values: Err(before_err),
698 });
699 }
700 (_, Err(after_err)) => {
701 return Some(TreeDiffEntry {
702 path,
703 values: Err(after_err),
704 });
705 }
706 };
707 let subdir =
708 TreeDiffDir::from_trees(&path, &before_tree, &after_tree, self.matcher);
709 self.stack.push(subdir);
710 }
711 if diff.before.is_file_like()
712 || diff.after.is_file_like()
713 || self.matcher.matches(&path)
714 {
715 return Some(TreeDiffEntry {
716 path,
717 values: Ok(diff),
718 });
719 }
720 }
721 None
722 }
723}
724
725pub struct TreeDiffStreamImpl<'matcher> {
727 store: Arc<Store>,
728 matcher: &'matcher dyn Matcher,
729 items: BTreeMap<RepoPathBuf, BackendResult<Diff<MergedTreeValue>>>,
734 #[expect(clippy::type_complexity)]
736 pending_trees:
737 BTreeMap<RepoPathBuf, BoxFuture<'matcher, BackendResult<(Merge<Tree>, Merge<Tree>)>>>,
738 max_concurrent_reads: usize,
743 max_queued_items: usize,
749}
750
751impl<'matcher> TreeDiffStreamImpl<'matcher> {
752 pub fn new(
755 tree1: &MergedTree,
756 tree2: &MergedTree,
757 matcher: &'matcher dyn Matcher,
758 max_concurrent_reads: usize,
759 ) -> Self {
760 assert!(Arc::ptr_eq(tree1.store(), tree2.store()));
761 let store = tree1.store().clone();
762 let mut stream = Self {
763 store: store.clone(),
764 matcher,
765 items: BTreeMap::new(),
766 pending_trees: BTreeMap::new(),
767 max_concurrent_reads,
768 max_queued_items: 10000,
769 };
770 let dir = RepoPathBuf::root();
771 let merged_tree1 = tree1.to_merged_tree_value();
772 let merged_tree2 = tree2.to_merged_tree_value();
773 let root_diff = Diff::new(merged_tree1.clone(), merged_tree2.clone());
774 if root_diff.is_changed() && matcher.matches(&dir) {
775 stream.items.insert(dir.clone(), Ok(root_diff));
776 }
777 let root_tree_fut = Box::pin(try_join(
778 Self::trees(store.clone(), dir.clone(), merged_tree1),
779 Self::trees(store, dir.clone(), merged_tree2),
780 ));
781 stream.pending_trees.insert(dir, root_tree_fut);
782 stream
783 }
784
785 async fn single_tree(
786 store: &Arc<Store>,
787 dir: RepoPathBuf,
788 value: Option<&TreeValue>,
789 ) -> BackendResult<Tree> {
790 match value {
791 Some(TreeValue::Tree(tree_id)) => store.get_tree(dir, tree_id).await,
792 _ => Ok(Tree::empty(store.clone(), dir.clone())),
793 }
794 }
795
796 async fn trees(
798 store: Arc<Store>,
799 dir: RepoPathBuf,
800 values: MergedTreeValue,
801 ) -> BackendResult<Merge<Tree>> {
802 if values.is_tree() {
803 values
804 .try_map_async(|value| Self::single_tree(&store, dir.clone(), value.as_ref()))
805 .await
806 } else {
807 Ok(Merge::resolved(Tree::empty(store, dir)))
808 }
809 }
810
811 fn add_dir_diff_items(&mut self, dir: &RepoPath, trees1: &Merge<Tree>, trees2: &Merge<Tree>) {
812 for (basename, diff) in merged_tree_entry_diff(trees1, trees2) {
813 let path = dir.join(basename);
814 let tree_before = diff.before.is_tree();
815 let tree_after = diff.after.is_tree();
816 let tree_matches =
819 (tree_before || tree_after) && !self.matcher.visit(&path).is_nothing();
820 let file_matches = (!tree_before || !tree_after) && self.matcher.matches(&path);
821
822 let before = if (tree_before && tree_matches) || (!tree_before && file_matches) {
824 diff.before
825 } else {
826 Merge::absent()
827 };
828 let after = if (tree_after && tree_matches) || (!tree_after && file_matches) {
829 diff.after
830 } else {
831 Merge::absent()
832 };
833 if before.is_absent() && after.is_absent() {
834 continue;
835 }
836
837 if tree_matches {
839 let before_tree_future =
840 Self::trees(self.store.clone(), path.clone(), before.cloned());
841 let after_tree_future =
842 Self::trees(self.store.clone(), path.clone(), after.cloned());
843 let both_trees_future = try_join(before_tree_future, after_tree_future);
844 self.pending_trees
845 .insert(path.clone(), Box::pin(both_trees_future));
846 }
847
848 if file_matches || self.matcher.matches(&path) {
849 self.items
850 .insert(path, Ok(Diff::new(before.cloned(), after.cloned())));
851 }
852 }
853 }
854
855 fn poll_tree_futures(&mut self, cx: &mut Context<'_>) {
856 loop {
857 let mut tree_diffs = vec![];
858 let mut some_pending = false;
859 let mut all_pending = true;
860 for (dir, future) in self
861 .pending_trees
862 .iter_mut()
863 .take(self.max_concurrent_reads)
864 {
865 if let Poll::Ready(tree_diff) = future.as_mut().poll(cx) {
866 all_pending = false;
867 tree_diffs.push((dir.clone(), tree_diff));
868 } else {
869 some_pending = true;
870 }
871 }
872
873 for (dir, tree_diff) in tree_diffs {
874 drop(self.pending_trees.remove_entry(&dir).unwrap());
875 match tree_diff {
876 Ok((trees1, trees2)) => {
877 self.add_dir_diff_items(&dir, &trees1, &trees2);
878 }
879 Err(err) => {
880 self.items.insert(dir, Err(err));
881 }
882 }
883 }
884
885 if all_pending || (some_pending && self.items.len() >= self.max_queued_items) {
889 return;
890 }
891 }
892 }
893}
894
895impl Stream for TreeDiffStreamImpl<'_> {
896 type Item = TreeDiffEntry;
897
898 fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
899 self.poll_tree_futures(cx);
901
902 if let Some((path, _)) = self.items.first_key_value() {
904 if let Some((dir, _)) = self.pending_trees.first_key_value()
907 && dir < path
908 {
909 return Poll::Pending;
910 }
911
912 let (path, values) = self.items.pop_first().unwrap();
913 Poll::Ready(Some(TreeDiffEntry { path, values }))
914 } else if self.pending_trees.is_empty() {
915 Poll::Ready(None)
916 } else {
917 Poll::Pending
918 }
919 }
920}
921
922fn stream_without_trees(stream: TreeDiffStream) -> TreeDiffStream {
923 stream
924 .filter_map(|mut entry| async move {
925 let skip_tree = |merge: MergedTreeValue| {
926 if merge.is_tree() {
927 Merge::absent()
928 } else {
929 merge
930 }
931 };
932 entry.values = entry.values.map(|diff| diff.map(skip_tree));
933
934 let any_present = entry.values.as_ref().map_or(true, |diff| {
936 diff.before.is_present() || diff.after.is_present()
937 });
938 any_present.then_some(entry)
939 })
940 .boxed()
941}
942
943struct DiffStreamForFileSystem<'a> {
946 inner: TreeDiffStream<'a>,
947 next_item: Option<TreeDiffEntry>,
948 held_file: Option<TreeDiffEntry>,
949}
950
951impl<'a> DiffStreamForFileSystem<'a> {
952 fn new(inner: TreeDiffStream<'a>) -> Self {
953 Self {
954 inner,
955 next_item: None,
956 held_file: None,
957 }
958 }
959}
960
961impl Stream for DiffStreamForFileSystem<'_> {
962 type Item = TreeDiffEntry;
963
964 fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
965 while let Some(next) = match self.next_item.take() {
966 Some(next) => Some(next),
967 None => ready!(self.inner.as_mut().poll_next(cx)),
968 } {
969 if let Ok(diff) = &next.values
972 && !diff.before.is_file_like()
973 && !diff.after.is_file_like()
974 {
975 continue;
976 }
977
978 if let Some(held_entry) = self
982 .held_file
983 .take_if(|held_entry| !next.path.starts_with(&held_entry.path))
984 {
985 self.next_item = Some(next);
986 return Poll::Ready(Some(held_entry));
987 }
988
989 match next.values {
990 Ok(diff) if diff.before.is_tree() => {
991 assert!(diff.after.is_present());
992 assert!(self.held_file.is_none());
993 self.held_file = Some(TreeDiffEntry {
994 path: next.path,
995 values: Ok(Diff::new(Merge::absent(), diff.after)),
996 });
997 }
998 Ok(diff) if diff.after.is_tree() => {
999 assert!(diff.before.is_present());
1000 return Poll::Ready(Some(TreeDiffEntry {
1001 path: next.path,
1002 values: Ok(Diff::new(diff.before, Merge::absent())),
1003 }));
1004 }
1005 _ => {
1006 return Poll::Ready(Some(next));
1007 }
1008 }
1009 }
1010 Poll::Ready(self.held_file.take())
1011 }
1012}