1use super::{
2 fs::{self, Fs},
3 ignore::IgnoreStack,
4};
5use ::ignore::gitignore::{Gitignore, GitignoreBuilder};
6use anyhow::{anyhow, Result};
7use buffer::{Buffer, History, LanguageRegistry, Operation, Rope};
8use client::{proto, Client, PeerId, TypedEnvelope};
9use clock::ReplicaId;
10use futures::{Stream, StreamExt};
11use fuzzy::CharBag;
12use gpui::{
13 executor, AppContext, AsyncAppContext, Entity, ModelContext, ModelHandle, MutableAppContext,
14 Task, UpgradeModelHandle, WeakModelHandle,
15};
16use lazy_static::lazy_static;
17use lsp::LanguageServer;
18use parking_lot::Mutex;
19use postage::{
20 prelude::{Sink as _, Stream as _},
21 watch,
22};
23use serde::Deserialize;
24use smol::channel::{self, Sender};
25use std::{
26 any::Any,
27 cmp::{self, Ordering},
28 collections::HashMap,
29 convert::{TryFrom, TryInto},
30 ffi::{OsStr, OsString},
31 fmt,
32 future::Future,
33 ops::Deref,
34 path::{Path, PathBuf},
35 sync::{
36 atomic::{AtomicUsize, Ordering::SeqCst},
37 Arc,
38 },
39 time::{Duration, SystemTime},
40};
41use sum_tree::Bias;
42use sum_tree::{Edit, SeekTarget, SumTree};
43use util::TryFutureExt;
44
45lazy_static! {
46 static ref GITIGNORE: &'static OsStr = OsStr::new(".gitignore");
47}
48
49#[derive(Clone, Debug)]
50enum ScanState {
51 Idle,
52 Scanning,
53 Err(Arc<anyhow::Error>),
54}
55
56pub enum Worktree {
57 Local(LocalWorktree),
58 Remote(RemoteWorktree),
59}
60
61pub enum Event {
62 Closed,
63}
64
65impl Entity for Worktree {
66 type Event = Event;
67
68 fn release(&mut self, cx: &mut MutableAppContext) {
69 match self {
70 Self::Local(tree) => {
71 if let Some(worktree_id) = *tree.remote_id.borrow() {
72 let rpc = tree.rpc.clone();
73 cx.spawn(|_| async move {
74 if let Err(err) = rpc.send(proto::CloseWorktree { worktree_id }).await {
75 log::error!("error closing worktree: {}", err);
76 }
77 })
78 .detach();
79 }
80 }
81 Self::Remote(tree) => {
82 let rpc = tree.client.clone();
83 let worktree_id = tree.remote_id;
84 cx.spawn(|_| async move {
85 if let Err(err) = rpc.send(proto::LeaveWorktree { worktree_id }).await {
86 log::error!("error closing worktree: {}", err);
87 }
88 })
89 .detach();
90 }
91 }
92 }
93}
94
95impl Worktree {
96 pub async fn open_local(
97 rpc: Arc<Client>,
98 path: impl Into<Arc<Path>>,
99 fs: Arc<dyn Fs>,
100 languages: Arc<LanguageRegistry>,
101 cx: &mut AsyncAppContext,
102 ) -> Result<ModelHandle<Self>> {
103 let (tree, scan_states_tx) =
104 LocalWorktree::new(rpc, path, fs.clone(), languages, cx).await?;
105 tree.update(cx, |tree, cx| {
106 let tree = tree.as_local_mut().unwrap();
107 let abs_path = tree.snapshot.abs_path.clone();
108 let background_snapshot = tree.background_snapshot.clone();
109 let background = cx.background().clone();
110 tree._background_scanner_task = Some(cx.background().spawn(async move {
111 let events = fs.watch(&abs_path, Duration::from_millis(100)).await;
112 let scanner =
113 BackgroundScanner::new(background_snapshot, scan_states_tx, fs, background);
114 scanner.run(events).await;
115 }));
116 });
117 Ok(tree)
118 }
119
120 pub async fn open_remote(
121 rpc: Arc<Client>,
122 id: u64,
123 languages: Arc<LanguageRegistry>,
124 cx: &mut AsyncAppContext,
125 ) -> Result<ModelHandle<Self>> {
126 let response = rpc.request(proto::JoinWorktree { worktree_id: id }).await?;
127 Worktree::remote(response, rpc, languages, cx).await
128 }
129
130 async fn remote(
131 join_response: proto::JoinWorktreeResponse,
132 rpc: Arc<Client>,
133 languages: Arc<LanguageRegistry>,
134 cx: &mut AsyncAppContext,
135 ) -> Result<ModelHandle<Self>> {
136 let worktree = join_response
137 .worktree
138 .ok_or_else(|| anyhow!("empty worktree"))?;
139
140 let remote_id = worktree.id;
141 let replica_id = join_response.replica_id as ReplicaId;
142 let peers = join_response.peers;
143 let root_char_bag: CharBag = worktree
144 .root_name
145 .chars()
146 .map(|c| c.to_ascii_lowercase())
147 .collect();
148 let root_name = worktree.root_name.clone();
149 let (entries_by_path, entries_by_id) = cx
150 .background()
151 .spawn(async move {
152 let mut entries_by_path_edits = Vec::new();
153 let mut entries_by_id_edits = Vec::new();
154 for entry in worktree.entries {
155 match Entry::try_from((&root_char_bag, entry)) {
156 Ok(entry) => {
157 entries_by_id_edits.push(Edit::Insert(PathEntry {
158 id: entry.id,
159 path: entry.path.clone(),
160 is_ignored: entry.is_ignored,
161 scan_id: 0,
162 }));
163 entries_by_path_edits.push(Edit::Insert(entry));
164 }
165 Err(err) => log::warn!("error for remote worktree entry {:?}", err),
166 }
167 }
168
169 let mut entries_by_path = SumTree::new();
170 let mut entries_by_id = SumTree::new();
171 entries_by_path.edit(entries_by_path_edits, &());
172 entries_by_id.edit(entries_by_id_edits, &());
173 (entries_by_path, entries_by_id)
174 })
175 .await;
176
177 let worktree = cx.update(|cx| {
178 cx.add_model(|cx: &mut ModelContext<Worktree>| {
179 let snapshot = Snapshot {
180 id: cx.model_id(),
181 scan_id: 0,
182 abs_path: Path::new("").into(),
183 root_name,
184 root_char_bag,
185 ignores: Default::default(),
186 entries_by_path,
187 entries_by_id,
188 removed_entry_ids: Default::default(),
189 next_entry_id: Default::default(),
190 };
191
192 let (updates_tx, mut updates_rx) = postage::mpsc::channel(64);
193 let (mut snapshot_tx, snapshot_rx) = watch::channel_with(snapshot.clone());
194
195 cx.background()
196 .spawn(async move {
197 while let Some(update) = updates_rx.recv().await {
198 let mut snapshot = snapshot_tx.borrow().clone();
199 if let Err(error) = snapshot.apply_update(update) {
200 log::error!("error applying worktree update: {}", error);
201 }
202 *snapshot_tx.borrow_mut() = snapshot;
203 }
204 })
205 .detach();
206
207 {
208 let mut snapshot_rx = snapshot_rx.clone();
209 cx.spawn_weak(|this, mut cx| async move {
210 while let Some(_) = snapshot_rx.recv().await {
211 if let Some(this) = cx.read(|cx| this.upgrade(cx)) {
212 this.update(&mut cx, |this, cx| this.poll_snapshot(cx));
213 } else {
214 break;
215 }
216 }
217 })
218 .detach();
219 }
220
221 let _subscriptions = vec![
222 rpc.subscribe_to_entity(remote_id, cx, Self::handle_add_peer),
223 rpc.subscribe_to_entity(remote_id, cx, Self::handle_remove_peer),
224 rpc.subscribe_to_entity(remote_id, cx, Self::handle_update),
225 rpc.subscribe_to_entity(remote_id, cx, Self::handle_update_buffer),
226 rpc.subscribe_to_entity(remote_id, cx, Self::handle_buffer_saved),
227 rpc.subscribe_to_entity(remote_id, cx, Self::handle_unshare),
228 ];
229
230 Worktree::Remote(RemoteWorktree {
231 remote_id,
232 replica_id,
233 snapshot,
234 snapshot_rx,
235 updates_tx,
236 client: rpc.clone(),
237 open_buffers: Default::default(),
238 peers: peers
239 .into_iter()
240 .map(|p| (PeerId(p.peer_id), p.replica_id as ReplicaId))
241 .collect(),
242 queued_operations: Default::default(),
243 languages,
244 _subscriptions,
245 })
246 })
247 });
248
249 Ok(worktree)
250 }
251
252 pub fn as_local(&self) -> Option<&LocalWorktree> {
253 if let Worktree::Local(worktree) = self {
254 Some(worktree)
255 } else {
256 None
257 }
258 }
259
260 pub fn as_local_mut(&mut self) -> Option<&mut LocalWorktree> {
261 if let Worktree::Local(worktree) = self {
262 Some(worktree)
263 } else {
264 None
265 }
266 }
267
268 pub fn as_remote_mut(&mut self) -> Option<&mut RemoteWorktree> {
269 if let Worktree::Remote(worktree) = self {
270 Some(worktree)
271 } else {
272 None
273 }
274 }
275
276 pub fn snapshot(&self) -> Snapshot {
277 match self {
278 Worktree::Local(worktree) => worktree.snapshot(),
279 Worktree::Remote(worktree) => worktree.snapshot(),
280 }
281 }
282
283 pub fn replica_id(&self) -> ReplicaId {
284 match self {
285 Worktree::Local(_) => 0,
286 Worktree::Remote(worktree) => worktree.replica_id,
287 }
288 }
289
290 pub fn languages(&self) -> &Arc<LanguageRegistry> {
291 match self {
292 Worktree::Local(worktree) => &worktree.languages,
293 Worktree::Remote(worktree) => &worktree.languages,
294 }
295 }
296
297 pub fn handle_add_peer(
298 &mut self,
299 envelope: TypedEnvelope<proto::AddPeer>,
300 _: Arc<Client>,
301 cx: &mut ModelContext<Self>,
302 ) -> Result<()> {
303 match self {
304 Worktree::Local(worktree) => worktree.add_peer(envelope, cx),
305 Worktree::Remote(worktree) => worktree.add_peer(envelope, cx),
306 }
307 }
308
309 pub fn handle_remove_peer(
310 &mut self,
311 envelope: TypedEnvelope<proto::RemovePeer>,
312 _: Arc<Client>,
313 cx: &mut ModelContext<Self>,
314 ) -> Result<()> {
315 match self {
316 Worktree::Local(worktree) => worktree.remove_peer(envelope, cx),
317 Worktree::Remote(worktree) => worktree.remove_peer(envelope, cx),
318 }
319 }
320
321 pub fn handle_update(
322 &mut self,
323 envelope: TypedEnvelope<proto::UpdateWorktree>,
324 _: Arc<Client>,
325 cx: &mut ModelContext<Self>,
326 ) -> anyhow::Result<()> {
327 self.as_remote_mut()
328 .unwrap()
329 .update_from_remote(envelope, cx)
330 }
331
332 pub fn handle_open_buffer(
333 &mut self,
334 envelope: TypedEnvelope<proto::OpenBuffer>,
335 rpc: Arc<Client>,
336 cx: &mut ModelContext<Self>,
337 ) -> anyhow::Result<()> {
338 let receipt = envelope.receipt();
339
340 let response = self
341 .as_local_mut()
342 .unwrap()
343 .open_remote_buffer(envelope, cx);
344
345 cx.background()
346 .spawn(
347 async move {
348 rpc.respond(receipt, response.await?).await?;
349 Ok(())
350 }
351 .log_err(),
352 )
353 .detach();
354
355 Ok(())
356 }
357
358 pub fn handle_close_buffer(
359 &mut self,
360 envelope: TypedEnvelope<proto::CloseBuffer>,
361 _: Arc<Client>,
362 cx: &mut ModelContext<Self>,
363 ) -> anyhow::Result<()> {
364 self.as_local_mut()
365 .unwrap()
366 .close_remote_buffer(envelope, cx)
367 }
368
369 pub fn peers(&self) -> &HashMap<PeerId, ReplicaId> {
370 match self {
371 Worktree::Local(worktree) => &worktree.peers,
372 Worktree::Remote(worktree) => &worktree.peers,
373 }
374 }
375
376 pub fn open_buffer(
377 &mut self,
378 path: impl AsRef<Path>,
379 cx: &mut ModelContext<Self>,
380 ) -> Task<Result<ModelHandle<Buffer>>> {
381 match self {
382 Worktree::Local(worktree) => worktree.open_buffer(path.as_ref(), cx),
383 Worktree::Remote(worktree) => worktree.open_buffer(path.as_ref(), cx),
384 }
385 }
386
387 #[cfg(feature = "test-support")]
388 pub fn has_open_buffer(&self, path: impl AsRef<Path>, cx: &AppContext) -> bool {
389 let mut open_buffers: Box<dyn Iterator<Item = _>> = match self {
390 Worktree::Local(worktree) => Box::new(worktree.open_buffers.values()),
391 Worktree::Remote(worktree) => {
392 Box::new(worktree.open_buffers.values().filter_map(|buf| {
393 if let RemoteBuffer::Loaded(buf) = buf {
394 Some(buf)
395 } else {
396 None
397 }
398 }))
399 }
400 };
401
402 let path = path.as_ref();
403 open_buffers
404 .find(|buffer| {
405 if let Some(file) = buffer.upgrade(cx).and_then(|buffer| buffer.read(cx).file()) {
406 file.path().as_ref() == path
407 } else {
408 false
409 }
410 })
411 .is_some()
412 }
413
414 pub fn handle_update_buffer(
415 &mut self,
416 envelope: TypedEnvelope<proto::UpdateBuffer>,
417 _: Arc<Client>,
418 cx: &mut ModelContext<Self>,
419 ) -> Result<()> {
420 let payload = envelope.payload.clone();
421 let buffer_id = payload.buffer_id as usize;
422 let ops = payload
423 .operations
424 .into_iter()
425 .map(|op| op.try_into())
426 .collect::<anyhow::Result<Vec<_>>>()?;
427
428 match self {
429 Worktree::Local(worktree) => {
430 let buffer = worktree
431 .open_buffers
432 .get(&buffer_id)
433 .and_then(|buf| buf.upgrade(cx))
434 .ok_or_else(|| {
435 anyhow!("invalid buffer {} in update buffer message", buffer_id)
436 })?;
437 buffer.update(cx, |buffer, cx| buffer.apply_ops(ops, cx))?;
438 }
439 Worktree::Remote(worktree) => match worktree.open_buffers.get_mut(&buffer_id) {
440 Some(RemoteBuffer::Operations(pending_ops)) => pending_ops.extend(ops),
441 Some(RemoteBuffer::Loaded(buffer)) => {
442 if let Some(buffer) = buffer.upgrade(cx) {
443 buffer.update(cx, |buffer, cx| buffer.apply_ops(ops, cx))?;
444 } else {
445 worktree
446 .open_buffers
447 .insert(buffer_id, RemoteBuffer::Operations(ops));
448 }
449 }
450 None => {
451 worktree
452 .open_buffers
453 .insert(buffer_id, RemoteBuffer::Operations(ops));
454 }
455 },
456 }
457
458 Ok(())
459 }
460
461 pub fn handle_save_buffer(
462 &mut self,
463 envelope: TypedEnvelope<proto::SaveBuffer>,
464 rpc: Arc<Client>,
465 cx: &mut ModelContext<Self>,
466 ) -> Result<()> {
467 let sender_id = envelope.original_sender_id()?;
468 let buffer = self
469 .as_local()
470 .unwrap()
471 .shared_buffers
472 .get(&sender_id)
473 .and_then(|shared_buffers| shared_buffers.get(&envelope.payload.buffer_id).cloned())
474 .ok_or_else(|| anyhow!("unknown buffer id {}", envelope.payload.buffer_id))?;
475
476 let receipt = envelope.receipt();
477 let worktree_id = envelope.payload.worktree_id;
478 let buffer_id = envelope.payload.buffer_id;
479 let save = cx.spawn(|_, mut cx| async move {
480 buffer.update(&mut cx, |buffer, cx| buffer.save(cx))?.await
481 });
482
483 cx.background()
484 .spawn(
485 async move {
486 let (version, mtime) = save.await?;
487
488 rpc.respond(
489 receipt,
490 proto::BufferSaved {
491 worktree_id,
492 buffer_id,
493 version: (&version).into(),
494 mtime: Some(mtime.into()),
495 },
496 )
497 .await?;
498
499 Ok(())
500 }
501 .log_err(),
502 )
503 .detach();
504
505 Ok(())
506 }
507
508 pub fn handle_buffer_saved(
509 &mut self,
510 envelope: TypedEnvelope<proto::BufferSaved>,
511 _: Arc<Client>,
512 cx: &mut ModelContext<Self>,
513 ) -> Result<()> {
514 let payload = envelope.payload.clone();
515 let worktree = self.as_remote_mut().unwrap();
516 if let Some(buffer) = worktree
517 .open_buffers
518 .get(&(payload.buffer_id as usize))
519 .and_then(|buf| buf.upgrade(cx))
520 {
521 buffer.update(cx, |buffer, cx| {
522 let version = payload.version.try_into()?;
523 let mtime = payload
524 .mtime
525 .ok_or_else(|| anyhow!("missing mtime"))?
526 .into();
527 buffer.did_save(version, mtime, None, cx);
528 Result::<_, anyhow::Error>::Ok(())
529 })?;
530 }
531 Ok(())
532 }
533
534 pub fn handle_unshare(
535 &mut self,
536 _: TypedEnvelope<proto::UnshareWorktree>,
537 _: Arc<Client>,
538 cx: &mut ModelContext<Self>,
539 ) -> Result<()> {
540 cx.emit(Event::Closed);
541 Ok(())
542 }
543
544 fn poll_snapshot(&mut self, cx: &mut ModelContext<Self>) {
545 match self {
546 Self::Local(worktree) => {
547 let is_fake_fs = worktree.fs.is_fake();
548 worktree.snapshot = worktree.background_snapshot.lock().clone();
549 if worktree.is_scanning() {
550 if worktree.poll_task.is_none() {
551 worktree.poll_task = Some(cx.spawn(|this, mut cx| async move {
552 if is_fake_fs {
553 smol::future::yield_now().await;
554 } else {
555 smol::Timer::after(Duration::from_millis(100)).await;
556 }
557 this.update(&mut cx, |this, cx| {
558 this.as_local_mut().unwrap().poll_task = None;
559 this.poll_snapshot(cx);
560 })
561 }));
562 }
563 } else {
564 worktree.poll_task.take();
565 self.update_open_buffers(cx);
566 }
567 }
568 Self::Remote(worktree) => {
569 worktree.snapshot = worktree.snapshot_rx.borrow().clone();
570 self.update_open_buffers(cx);
571 }
572 };
573
574 cx.notify();
575 }
576
577 fn update_open_buffers(&mut self, cx: &mut ModelContext<Self>) {
578 let open_buffers: Box<dyn Iterator<Item = _>> = match &self {
579 Self::Local(worktree) => Box::new(worktree.open_buffers.iter()),
580 Self::Remote(worktree) => {
581 Box::new(worktree.open_buffers.iter().filter_map(|(id, buf)| {
582 if let RemoteBuffer::Loaded(buf) = buf {
583 Some((id, buf))
584 } else {
585 None
586 }
587 }))
588 }
589 };
590
591 let mut buffers_to_delete = Vec::new();
592 for (buffer_id, buffer) in open_buffers {
593 if let Some(buffer) = buffer.upgrade(cx) {
594 buffer.update(cx, |buffer, cx| {
595 let buffer_is_clean = !buffer.is_dirty();
596
597 if let Some(file) = buffer.file_mut() {
598 let mut file_changed = false;
599
600 if let Some(entry) = file
601 .entry_id()
602 .and_then(|entry_id| self.entry_for_id(entry_id))
603 {
604 if entry.path != *file.path() {
605 file.set_path(entry.path.clone());
606 file_changed = true;
607 }
608
609 if entry.mtime != file.mtime() {
610 file.set_mtime(entry.mtime);
611 file_changed = true;
612 if let Some(worktree) = self.as_local() {
613 if buffer_is_clean {
614 let abs_path = worktree.absolutize(file.path().as_ref());
615 refresh_buffer(abs_path, &worktree.fs, cx);
616 }
617 }
618 }
619 } else if let Some(entry) = self.entry_for_path(file.path().as_ref()) {
620 file.set_entry_id(Some(entry.id));
621 file.set_mtime(entry.mtime);
622 if let Some(worktree) = self.as_local() {
623 if buffer_is_clean {
624 let abs_path = worktree.absolutize(file.path().as_ref());
625 refresh_buffer(abs_path, &worktree.fs, cx);
626 }
627 }
628 file_changed = true;
629 } else if !file.is_deleted() {
630 if buffer_is_clean {
631 cx.emit(buffer::Event::Dirtied);
632 }
633 file.set_entry_id(None);
634 file_changed = true;
635 }
636
637 if file_changed {
638 cx.emit(buffer::Event::FileHandleChanged);
639 }
640 }
641 });
642 } else {
643 buffers_to_delete.push(*buffer_id);
644 }
645 }
646
647 for buffer_id in buffers_to_delete {
648 match self {
649 Self::Local(worktree) => {
650 worktree.open_buffers.remove(&buffer_id);
651 }
652 Self::Remote(worktree) => {
653 worktree.open_buffers.remove(&buffer_id);
654 }
655 }
656 }
657 }
658}
659
660impl Deref for Worktree {
661 type Target = Snapshot;
662
663 fn deref(&self) -> &Self::Target {
664 match self {
665 Worktree::Local(worktree) => &worktree.snapshot,
666 Worktree::Remote(worktree) => &worktree.snapshot,
667 }
668 }
669}
670
671pub struct LocalWorktree {
672 snapshot: Snapshot,
673 config: WorktreeConfig,
674 background_snapshot: Arc<Mutex<Snapshot>>,
675 last_scan_state_rx: watch::Receiver<ScanState>,
676 _background_scanner_task: Option<Task<()>>,
677 _maintain_remote_id_task: Task<Option<()>>,
678 poll_task: Option<Task<()>>,
679 remote_id: watch::Receiver<Option<u64>>,
680 share: Option<ShareState>,
681 open_buffers: HashMap<usize, WeakModelHandle<Buffer>>,
682 shared_buffers: HashMap<PeerId, HashMap<u64, ModelHandle<Buffer>>>,
683 peers: HashMap<PeerId, ReplicaId>,
684 languages: Arc<LanguageRegistry>,
685 queued_operations: Vec<(u64, Operation)>,
686 rpc: Arc<Client>,
687 fs: Arc<dyn Fs>,
688 language_server: Arc<LanguageServer>,
689}
690
691#[derive(Default, Deserialize)]
692struct WorktreeConfig {
693 collaborators: Vec<String>,
694}
695
696impl LocalWorktree {
697 async fn new(
698 rpc: Arc<Client>,
699 path: impl Into<Arc<Path>>,
700 fs: Arc<dyn Fs>,
701 languages: Arc<LanguageRegistry>,
702 cx: &mut AsyncAppContext,
703 ) -> Result<(ModelHandle<Worktree>, Sender<ScanState>)> {
704 let abs_path = path.into();
705 let path: Arc<Path> = Arc::from(Path::new(""));
706 let next_entry_id = AtomicUsize::new(0);
707
708 // After determining whether the root entry is a file or a directory, populate the
709 // snapshot's "root name", which will be used for the purpose of fuzzy matching.
710 let root_name = abs_path
711 .file_name()
712 .map_or(String::new(), |f| f.to_string_lossy().to_string());
713 let root_char_bag = root_name.chars().map(|c| c.to_ascii_lowercase()).collect();
714 let metadata = fs.metadata(&abs_path).await?;
715
716 let mut config = WorktreeConfig::default();
717 if let Ok(zed_toml) = fs.load(&abs_path.join(".zed.toml")).await {
718 if let Ok(parsed) = toml::from_str(&zed_toml) {
719 config = parsed;
720 }
721 }
722
723 let (scan_states_tx, scan_states_rx) = smol::channel::unbounded();
724 let (mut last_scan_state_tx, last_scan_state_rx) = watch::channel_with(ScanState::Scanning);
725 let tree = cx.add_model(move |cx: &mut ModelContext<Worktree>| {
726 let language_server = LanguageServer::rust(&abs_path, cx).unwrap();
727 let mut snapshot = Snapshot {
728 id: cx.model_id(),
729 scan_id: 0,
730 abs_path,
731 root_name: root_name.clone(),
732 root_char_bag,
733 ignores: Default::default(),
734 entries_by_path: Default::default(),
735 entries_by_id: Default::default(),
736 removed_entry_ids: Default::default(),
737 next_entry_id: Arc::new(next_entry_id),
738 };
739 if let Some(metadata) = metadata {
740 snapshot.insert_entry(
741 Entry::new(
742 path.into(),
743 &metadata,
744 &snapshot.next_entry_id,
745 snapshot.root_char_bag,
746 ),
747 fs.as_ref(),
748 );
749 }
750
751 let (mut remote_id_tx, remote_id_rx) = watch::channel();
752 let _maintain_remote_id_task = cx.spawn_weak({
753 let rpc = rpc.clone();
754 move |this, cx| {
755 async move {
756 let mut status = rpc.status();
757 while let Some(status) = status.recv().await {
758 if let Some(this) = this.upgrade(&cx) {
759 let remote_id = if let client::Status::Connected { .. } = status {
760 let collaborator_logins = this.read_with(&cx, |this, _| {
761 this.as_local().unwrap().config.collaborators.clone()
762 });
763 let response = rpc
764 .request(proto::OpenWorktree {
765 root_name: root_name.clone(),
766 collaborator_logins,
767 })
768 .await?;
769
770 Some(response.worktree_id)
771 } else {
772 None
773 };
774 if remote_id_tx.send(remote_id).await.is_err() {
775 break;
776 }
777 }
778 }
779 Ok(())
780 }
781 .log_err()
782 }
783 });
784
785 let tree = Self {
786 snapshot: snapshot.clone(),
787 config,
788 remote_id: remote_id_rx,
789 background_snapshot: Arc::new(Mutex::new(snapshot)),
790 last_scan_state_rx,
791 _background_scanner_task: None,
792 _maintain_remote_id_task,
793 share: None,
794 poll_task: None,
795 open_buffers: Default::default(),
796 shared_buffers: Default::default(),
797 queued_operations: Default::default(),
798 peers: Default::default(),
799 languages,
800 rpc,
801 fs,
802 language_server,
803 };
804
805 cx.spawn_weak(|this, mut cx| async move {
806 while let Ok(scan_state) = scan_states_rx.recv().await {
807 if let Some(handle) = cx.read(|cx| this.upgrade(cx)) {
808 let to_send = handle.update(&mut cx, |this, cx| {
809 last_scan_state_tx.blocking_send(scan_state).ok();
810 this.poll_snapshot(cx);
811 let tree = this.as_local_mut().unwrap();
812 if !tree.is_scanning() {
813 if let Some(share) = tree.share.as_ref() {
814 return Some((tree.snapshot(), share.snapshots_tx.clone()));
815 }
816 }
817 None
818 });
819
820 if let Some((snapshot, snapshots_to_send_tx)) = to_send {
821 if let Err(err) = snapshots_to_send_tx.send(snapshot).await {
822 log::error!("error submitting snapshot to send {}", err);
823 }
824 }
825 } else {
826 break;
827 }
828 }
829 })
830 .detach();
831
832 Worktree::Local(tree)
833 });
834
835 Ok((tree, scan_states_tx))
836 }
837
838 pub fn open_buffer(
839 &mut self,
840 path: &Path,
841 cx: &mut ModelContext<Worktree>,
842 ) -> Task<Result<ModelHandle<Buffer>>> {
843 let handle = cx.handle();
844
845 // If there is already a buffer for the given path, then return it.
846 let mut existing_buffer = None;
847 self.open_buffers.retain(|_buffer_id, buffer| {
848 if let Some(buffer) = buffer.upgrade(cx.as_ref()) {
849 if let Some(file) = buffer.read(cx.as_ref()).file() {
850 if file.worktree_id() == handle.id() && file.path().as_ref() == path {
851 existing_buffer = Some(buffer);
852 }
853 }
854 true
855 } else {
856 false
857 }
858 });
859
860 let path = Arc::from(path);
861 cx.spawn(|this, mut cx| async move {
862 if let Some(existing_buffer) = existing_buffer {
863 Ok(existing_buffer)
864 } else {
865 let (file, contents) = this
866 .update(&mut cx, |this, cx| this.as_local().unwrap().load(&path, cx))
867 .await?;
868 let language = this.read_with(&cx, |this, cx| {
869 use buffer::File;
870
871 this.languages()
872 .select_language(file.full_path(cx))
873 .cloned()
874 });
875 let buffer = cx.add_model(|cx| {
876 Buffer::from_history(
877 0,
878 History::new(contents.into()),
879 Some(Box::new(file)),
880 language,
881 cx,
882 )
883 });
884 this.update(&mut cx, |this, _| {
885 let this = this
886 .as_local_mut()
887 .ok_or_else(|| anyhow!("must be a local worktree"))?;
888 this.open_buffers.insert(buffer.id(), buffer.downgrade());
889 Ok(buffer)
890 })
891 }
892 })
893 }
894
895 pub fn open_remote_buffer(
896 &mut self,
897 envelope: TypedEnvelope<proto::OpenBuffer>,
898 cx: &mut ModelContext<Worktree>,
899 ) -> Task<Result<proto::OpenBufferResponse>> {
900 let peer_id = envelope.original_sender_id();
901 let path = Path::new(&envelope.payload.path);
902
903 let buffer = self.open_buffer(path, cx);
904
905 cx.spawn(|this, mut cx| async move {
906 let buffer = buffer.await?;
907 this.update(&mut cx, |this, cx| {
908 this.as_local_mut()
909 .unwrap()
910 .shared_buffers
911 .entry(peer_id?)
912 .or_default()
913 .insert(buffer.id() as u64, buffer.clone());
914
915 Ok(proto::OpenBufferResponse {
916 buffer: Some(buffer.update(cx.as_mut(), |buffer, cx| buffer.to_proto(cx))),
917 })
918 })
919 })
920 }
921
922 pub fn close_remote_buffer(
923 &mut self,
924 envelope: TypedEnvelope<proto::CloseBuffer>,
925 cx: &mut ModelContext<Worktree>,
926 ) -> Result<()> {
927 if let Some(shared_buffers) = self.shared_buffers.get_mut(&envelope.original_sender_id()?) {
928 shared_buffers.remove(&envelope.payload.buffer_id);
929 cx.notify();
930 }
931
932 Ok(())
933 }
934
935 pub fn add_peer(
936 &mut self,
937 envelope: TypedEnvelope<proto::AddPeer>,
938 cx: &mut ModelContext<Worktree>,
939 ) -> Result<()> {
940 let peer = envelope
941 .payload
942 .peer
943 .as_ref()
944 .ok_or_else(|| anyhow!("empty peer"))?;
945 self.peers
946 .insert(PeerId(peer.peer_id), peer.replica_id as ReplicaId);
947 cx.notify();
948
949 Ok(())
950 }
951
952 pub fn remove_peer(
953 &mut self,
954 envelope: TypedEnvelope<proto::RemovePeer>,
955 cx: &mut ModelContext<Worktree>,
956 ) -> Result<()> {
957 let peer_id = PeerId(envelope.payload.peer_id);
958 let replica_id = self
959 .peers
960 .remove(&peer_id)
961 .ok_or_else(|| anyhow!("unknown peer {:?}", peer_id))?;
962 self.shared_buffers.remove(&peer_id);
963 for (_, buffer) in &self.open_buffers {
964 if let Some(buffer) = buffer.upgrade(cx) {
965 buffer.update(cx, |buffer, cx| buffer.remove_peer(replica_id, cx));
966 }
967 }
968 cx.notify();
969
970 Ok(())
971 }
972
973 pub fn scan_complete(&self) -> impl Future<Output = ()> {
974 let mut scan_state_rx = self.last_scan_state_rx.clone();
975 async move {
976 let mut scan_state = Some(scan_state_rx.borrow().clone());
977 while let Some(ScanState::Scanning) = scan_state {
978 scan_state = scan_state_rx.recv().await;
979 }
980 }
981 }
982
983 pub fn remote_id(&self) -> Option<u64> {
984 *self.remote_id.borrow()
985 }
986
987 pub fn next_remote_id(&self) -> impl Future<Output = Option<u64>> {
988 let mut remote_id = self.remote_id.clone();
989 async move {
990 while let Some(remote_id) = remote_id.recv().await {
991 if remote_id.is_some() {
992 return remote_id;
993 }
994 }
995 None
996 }
997 }
998
999 fn is_scanning(&self) -> bool {
1000 if let ScanState::Scanning = *self.last_scan_state_rx.borrow() {
1001 true
1002 } else {
1003 false
1004 }
1005 }
1006
1007 pub fn snapshot(&self) -> Snapshot {
1008 self.snapshot.clone()
1009 }
1010
1011 pub fn abs_path(&self) -> &Path {
1012 self.snapshot.abs_path.as_ref()
1013 }
1014
1015 pub fn contains_abs_path(&self, path: &Path) -> bool {
1016 path.starts_with(&self.snapshot.abs_path)
1017 }
1018
1019 fn absolutize(&self, path: &Path) -> PathBuf {
1020 if path.file_name().is_some() {
1021 self.snapshot.abs_path.join(path)
1022 } else {
1023 self.snapshot.abs_path.to_path_buf()
1024 }
1025 }
1026
1027 fn load(&self, path: &Path, cx: &mut ModelContext<Worktree>) -> Task<Result<(File, String)>> {
1028 let handle = cx.handle();
1029 let path = Arc::from(path);
1030 let abs_path = self.absolutize(&path);
1031 let background_snapshot = self.background_snapshot.clone();
1032 let fs = self.fs.clone();
1033 cx.spawn(|this, mut cx| async move {
1034 let text = fs.load(&abs_path).await?;
1035 // Eagerly populate the snapshot with an updated entry for the loaded file
1036 let entry = refresh_entry(fs.as_ref(), &background_snapshot, path, &abs_path).await?;
1037 this.update(&mut cx, |this, cx| this.poll_snapshot(cx));
1038 Ok((File::new(entry.id, handle, entry.path, entry.mtime), text))
1039 })
1040 }
1041
1042 pub fn save_buffer_as(
1043 &self,
1044 buffer: ModelHandle<Buffer>,
1045 path: impl Into<Arc<Path>>,
1046 text: Rope,
1047 cx: &mut ModelContext<Worktree>,
1048 ) -> Task<Result<File>> {
1049 let save = self.save(path, text, cx);
1050 cx.spawn(|this, mut cx| async move {
1051 let entry = save.await?;
1052 this.update(&mut cx, |this, cx| {
1053 this.as_local_mut()
1054 .unwrap()
1055 .open_buffers
1056 .insert(buffer.id(), buffer.downgrade());
1057 Ok(File::new(entry.id, cx.handle(), entry.path, entry.mtime))
1058 })
1059 })
1060 }
1061
1062 fn save(
1063 &self,
1064 path: impl Into<Arc<Path>>,
1065 text: Rope,
1066 cx: &mut ModelContext<Worktree>,
1067 ) -> Task<Result<Entry>> {
1068 let path = path.into();
1069 let abs_path = self.absolutize(&path);
1070 let background_snapshot = self.background_snapshot.clone();
1071 let fs = self.fs.clone();
1072 let save = cx.background().spawn(async move {
1073 fs.save(&abs_path, &text).await?;
1074 refresh_entry(fs.as_ref(), &background_snapshot, path.clone(), &abs_path).await
1075 });
1076
1077 cx.spawn(|this, mut cx| async move {
1078 let entry = save.await?;
1079 this.update(&mut cx, |this, cx| this.poll_snapshot(cx));
1080 Ok(entry)
1081 })
1082 }
1083
1084 pub fn share(&mut self, cx: &mut ModelContext<Worktree>) -> Task<anyhow::Result<u64>> {
1085 let snapshot = self.snapshot();
1086 let share_request = self.share_request(cx);
1087 let rpc = self.rpc.clone();
1088 cx.spawn(|this, mut cx| async move {
1089 let share_request = if let Some(request) = share_request.await {
1090 request
1091 } else {
1092 return Err(anyhow!("failed to open worktree on the server"));
1093 };
1094
1095 let remote_id = share_request.worktree.as_ref().unwrap().id;
1096 let share_response = rpc.request(share_request).await?;
1097
1098 log::info!("sharing worktree {:?}", share_response);
1099 let (snapshots_to_send_tx, snapshots_to_send_rx) =
1100 smol::channel::unbounded::<Snapshot>();
1101
1102 cx.background()
1103 .spawn({
1104 let rpc = rpc.clone();
1105 async move {
1106 let mut prev_snapshot = snapshot;
1107 while let Ok(snapshot) = snapshots_to_send_rx.recv().await {
1108 let message = snapshot.build_update(&prev_snapshot, remote_id, false);
1109 match rpc.send(message).await {
1110 Ok(()) => prev_snapshot = snapshot,
1111 Err(err) => log::error!("error sending snapshot diff {}", err),
1112 }
1113 }
1114 }
1115 })
1116 .detach();
1117
1118 this.update(&mut cx, |worktree, cx| {
1119 let _subscriptions = vec![
1120 rpc.subscribe_to_entity(remote_id, cx, Worktree::handle_add_peer),
1121 rpc.subscribe_to_entity(remote_id, cx, Worktree::handle_remove_peer),
1122 rpc.subscribe_to_entity(remote_id, cx, Worktree::handle_open_buffer),
1123 rpc.subscribe_to_entity(remote_id, cx, Worktree::handle_close_buffer),
1124 rpc.subscribe_to_entity(remote_id, cx, Worktree::handle_update_buffer),
1125 rpc.subscribe_to_entity(remote_id, cx, Worktree::handle_save_buffer),
1126 ];
1127
1128 let worktree = worktree.as_local_mut().unwrap();
1129 worktree.share = Some(ShareState {
1130 snapshots_tx: snapshots_to_send_tx,
1131 _subscriptions,
1132 });
1133 });
1134
1135 Ok(remote_id)
1136 })
1137 }
1138
1139 pub fn unshare(&mut self, cx: &mut ModelContext<Worktree>) {
1140 self.share.take();
1141 let rpc = self.rpc.clone();
1142 let remote_id = self.remote_id();
1143 cx.foreground()
1144 .spawn(
1145 async move {
1146 if let Some(worktree_id) = remote_id {
1147 rpc.send(proto::UnshareWorktree { worktree_id }).await?;
1148 }
1149 Ok(())
1150 }
1151 .log_err(),
1152 )
1153 .detach()
1154 }
1155
1156 fn share_request(&self, cx: &mut ModelContext<Worktree>) -> Task<Option<proto::ShareWorktree>> {
1157 let remote_id = self.next_remote_id();
1158 let snapshot = self.snapshot();
1159 let root_name = self.root_name.clone();
1160 cx.background().spawn(async move {
1161 remote_id.await.map(|id| {
1162 let entries = snapshot
1163 .entries_by_path
1164 .cursor::<()>()
1165 .filter(|e| !e.is_ignored)
1166 .map(Into::into)
1167 .collect();
1168 proto::ShareWorktree {
1169 worktree: Some(proto::Worktree {
1170 id,
1171 root_name,
1172 entries,
1173 }),
1174 }
1175 })
1176 })
1177 }
1178}
1179
1180fn build_gitignore(abs_path: &Path, fs: &dyn Fs) -> Result<Gitignore> {
1181 let contents = smol::block_on(fs.load(&abs_path))?;
1182 let parent = abs_path.parent().unwrap_or(Path::new("/"));
1183 let mut builder = GitignoreBuilder::new(parent);
1184 for line in contents.lines() {
1185 builder.add_line(Some(abs_path.into()), line)?;
1186 }
1187 Ok(builder.build()?)
1188}
1189
1190pub fn refresh_buffer(abs_path: PathBuf, fs: &Arc<dyn Fs>, cx: &mut ModelContext<Buffer>) {
1191 let fs = fs.clone();
1192 cx.spawn(|buffer, mut cx| async move {
1193 let new_text = fs.load(&abs_path).await;
1194 match new_text {
1195 Err(error) => log::error!("error refreshing buffer after file changed: {}", error),
1196 Ok(new_text) => {
1197 buffer
1198 .update(&mut cx, |buffer, cx| {
1199 buffer.set_text_from_disk(new_text.into(), cx)
1200 })
1201 .await;
1202 }
1203 }
1204 })
1205 .detach()
1206}
1207
1208impl Deref for LocalWorktree {
1209 type Target = Snapshot;
1210
1211 fn deref(&self) -> &Self::Target {
1212 &self.snapshot
1213 }
1214}
1215
1216impl fmt::Debug for LocalWorktree {
1217 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1218 self.snapshot.fmt(f)
1219 }
1220}
1221
1222struct ShareState {
1223 snapshots_tx: Sender<Snapshot>,
1224 _subscriptions: Vec<client::Subscription>,
1225}
1226
1227pub struct RemoteWorktree {
1228 remote_id: u64,
1229 snapshot: Snapshot,
1230 snapshot_rx: watch::Receiver<Snapshot>,
1231 client: Arc<Client>,
1232 updates_tx: postage::mpsc::Sender<proto::UpdateWorktree>,
1233 replica_id: ReplicaId,
1234 open_buffers: HashMap<usize, RemoteBuffer>,
1235 peers: HashMap<PeerId, ReplicaId>,
1236 languages: Arc<LanguageRegistry>,
1237 queued_operations: Vec<(u64, Operation)>,
1238 _subscriptions: Vec<client::Subscription>,
1239}
1240
1241impl RemoteWorktree {
1242 pub fn open_buffer(
1243 &mut self,
1244 path: &Path,
1245 cx: &mut ModelContext<Worktree>,
1246 ) -> Task<Result<ModelHandle<Buffer>>> {
1247 let mut existing_buffer = None;
1248 self.open_buffers.retain(|_buffer_id, buffer| {
1249 if let Some(buffer) = buffer.upgrade(cx.as_ref()) {
1250 if let Some(file) = buffer.read(cx.as_ref()).file() {
1251 if file.worktree_id() == cx.model_id() && file.path().as_ref() == path {
1252 existing_buffer = Some(buffer);
1253 }
1254 }
1255 true
1256 } else {
1257 false
1258 }
1259 });
1260
1261 let rpc = self.client.clone();
1262 let replica_id = self.replica_id;
1263 let remote_worktree_id = self.remote_id;
1264 let path = path.to_string_lossy().to_string();
1265 cx.spawn_weak(|this, mut cx| async move {
1266 if let Some(existing_buffer) = existing_buffer {
1267 Ok(existing_buffer)
1268 } else {
1269 let entry = this
1270 .upgrade(&cx)
1271 .ok_or_else(|| anyhow!("worktree was closed"))?
1272 .read_with(&cx, |tree, _| tree.entry_for_path(&path).cloned())
1273 .ok_or_else(|| anyhow!("file does not exist"))?;
1274 let response = rpc
1275 .request(proto::OpenBuffer {
1276 worktree_id: remote_worktree_id as u64,
1277 path,
1278 })
1279 .await?;
1280
1281 let this = this
1282 .upgrade(&cx)
1283 .ok_or_else(|| anyhow!("worktree was closed"))?;
1284 let file = File::new(entry.id, this.clone(), entry.path, entry.mtime);
1285 let language = this.read_with(&cx, |this, cx| {
1286 use buffer::File;
1287
1288 this.languages()
1289 .select_language(file.full_path(cx))
1290 .cloned()
1291 });
1292 let remote_buffer = response.buffer.ok_or_else(|| anyhow!("empty buffer"))?;
1293 let buffer_id = remote_buffer.id as usize;
1294 let buffer = cx.add_model(|cx| {
1295 Buffer::from_proto(
1296 replica_id,
1297 remote_buffer,
1298 Some(Box::new(file)),
1299 language,
1300 cx,
1301 )
1302 .unwrap()
1303 });
1304 this.update(&mut cx, |this, cx| {
1305 let this = this.as_remote_mut().unwrap();
1306 if let Some(RemoteBuffer::Operations(pending_ops)) = this
1307 .open_buffers
1308 .insert(buffer_id, RemoteBuffer::Loaded(buffer.downgrade()))
1309 {
1310 buffer.update(cx, |buf, cx| buf.apply_ops(pending_ops, cx))?;
1311 }
1312 Result::<_, anyhow::Error>::Ok(())
1313 })?;
1314 Ok(buffer)
1315 }
1316 })
1317 }
1318
1319 pub fn remote_id(&self) -> u64 {
1320 self.remote_id
1321 }
1322
1323 pub fn close_all_buffers(&mut self, cx: &mut MutableAppContext) {
1324 for (_, buffer) in self.open_buffers.drain() {
1325 if let RemoteBuffer::Loaded(buffer) = buffer {
1326 if let Some(buffer) = buffer.upgrade(cx) {
1327 buffer.update(cx, |buffer, cx| buffer.close(cx))
1328 }
1329 }
1330 }
1331 }
1332
1333 fn snapshot(&self) -> Snapshot {
1334 self.snapshot.clone()
1335 }
1336
1337 fn update_from_remote(
1338 &mut self,
1339 envelope: TypedEnvelope<proto::UpdateWorktree>,
1340 cx: &mut ModelContext<Worktree>,
1341 ) -> Result<()> {
1342 let mut tx = self.updates_tx.clone();
1343 let payload = envelope.payload.clone();
1344 cx.background()
1345 .spawn(async move {
1346 tx.send(payload).await.expect("receiver runs to completion");
1347 })
1348 .detach();
1349
1350 Ok(())
1351 }
1352
1353 pub fn add_peer(
1354 &mut self,
1355 envelope: TypedEnvelope<proto::AddPeer>,
1356 cx: &mut ModelContext<Worktree>,
1357 ) -> Result<()> {
1358 let peer = envelope
1359 .payload
1360 .peer
1361 .as_ref()
1362 .ok_or_else(|| anyhow!("empty peer"))?;
1363 self.peers
1364 .insert(PeerId(peer.peer_id), peer.replica_id as ReplicaId);
1365 cx.notify();
1366 Ok(())
1367 }
1368
1369 pub fn remove_peer(
1370 &mut self,
1371 envelope: TypedEnvelope<proto::RemovePeer>,
1372 cx: &mut ModelContext<Worktree>,
1373 ) -> Result<()> {
1374 let peer_id = PeerId(envelope.payload.peer_id);
1375 let replica_id = self
1376 .peers
1377 .remove(&peer_id)
1378 .ok_or_else(|| anyhow!("unknown peer {:?}", peer_id))?;
1379 for (_, buffer) in &self.open_buffers {
1380 if let Some(buffer) = buffer.upgrade(cx) {
1381 buffer.update(cx, |buffer, cx| buffer.remove_peer(replica_id, cx));
1382 }
1383 }
1384 cx.notify();
1385 Ok(())
1386 }
1387}
1388
1389enum RemoteBuffer {
1390 Operations(Vec<Operation>),
1391 Loaded(WeakModelHandle<Buffer>),
1392}
1393
1394impl RemoteBuffer {
1395 fn upgrade(&self, cx: &impl UpgradeModelHandle) -> Option<ModelHandle<Buffer>> {
1396 match self {
1397 Self::Operations(_) => None,
1398 Self::Loaded(buffer) => buffer.upgrade(cx),
1399 }
1400 }
1401}
1402
1403#[derive(Clone)]
1404pub struct Snapshot {
1405 id: usize,
1406 scan_id: usize,
1407 abs_path: Arc<Path>,
1408 root_name: String,
1409 root_char_bag: CharBag,
1410 ignores: HashMap<Arc<Path>, (Arc<Gitignore>, usize)>,
1411 entries_by_path: SumTree<Entry>,
1412 entries_by_id: SumTree<PathEntry>,
1413 removed_entry_ids: HashMap<u64, usize>,
1414 next_entry_id: Arc<AtomicUsize>,
1415}
1416
1417impl Snapshot {
1418 pub fn id(&self) -> usize {
1419 self.id
1420 }
1421
1422 pub fn build_update(
1423 &self,
1424 other: &Self,
1425 worktree_id: u64,
1426 include_ignored: bool,
1427 ) -> proto::UpdateWorktree {
1428 let mut updated_entries = Vec::new();
1429 let mut removed_entries = Vec::new();
1430 let mut self_entries = self
1431 .entries_by_id
1432 .cursor::<()>()
1433 .filter(|e| include_ignored || !e.is_ignored)
1434 .peekable();
1435 let mut other_entries = other
1436 .entries_by_id
1437 .cursor::<()>()
1438 .filter(|e| include_ignored || !e.is_ignored)
1439 .peekable();
1440 loop {
1441 match (self_entries.peek(), other_entries.peek()) {
1442 (Some(self_entry), Some(other_entry)) => {
1443 match Ord::cmp(&self_entry.id, &other_entry.id) {
1444 Ordering::Less => {
1445 let entry = self.entry_for_id(self_entry.id).unwrap().into();
1446 updated_entries.push(entry);
1447 self_entries.next();
1448 }
1449 Ordering::Equal => {
1450 if self_entry.scan_id != other_entry.scan_id {
1451 let entry = self.entry_for_id(self_entry.id).unwrap().into();
1452 updated_entries.push(entry);
1453 }
1454
1455 self_entries.next();
1456 other_entries.next();
1457 }
1458 Ordering::Greater => {
1459 removed_entries.push(other_entry.id as u64);
1460 other_entries.next();
1461 }
1462 }
1463 }
1464 (Some(self_entry), None) => {
1465 let entry = self.entry_for_id(self_entry.id).unwrap().into();
1466 updated_entries.push(entry);
1467 self_entries.next();
1468 }
1469 (None, Some(other_entry)) => {
1470 removed_entries.push(other_entry.id as u64);
1471 other_entries.next();
1472 }
1473 (None, None) => break,
1474 }
1475 }
1476
1477 proto::UpdateWorktree {
1478 updated_entries,
1479 removed_entries,
1480 worktree_id,
1481 }
1482 }
1483
1484 fn apply_update(&mut self, update: proto::UpdateWorktree) -> Result<()> {
1485 self.scan_id += 1;
1486 let scan_id = self.scan_id;
1487
1488 let mut entries_by_path_edits = Vec::new();
1489 let mut entries_by_id_edits = Vec::new();
1490 for entry_id in update.removed_entries {
1491 let entry_id = entry_id as usize;
1492 let entry = self
1493 .entry_for_id(entry_id)
1494 .ok_or_else(|| anyhow!("unknown entry"))?;
1495 entries_by_path_edits.push(Edit::Remove(PathKey(entry.path.clone())));
1496 entries_by_id_edits.push(Edit::Remove(entry.id));
1497 }
1498
1499 for entry in update.updated_entries {
1500 let entry = Entry::try_from((&self.root_char_bag, entry))?;
1501 if let Some(PathEntry { path, .. }) = self.entries_by_id.get(&entry.id, &()) {
1502 entries_by_path_edits.push(Edit::Remove(PathKey(path.clone())));
1503 }
1504 entries_by_id_edits.push(Edit::Insert(PathEntry {
1505 id: entry.id,
1506 path: entry.path.clone(),
1507 is_ignored: entry.is_ignored,
1508 scan_id,
1509 }));
1510 entries_by_path_edits.push(Edit::Insert(entry));
1511 }
1512
1513 self.entries_by_path.edit(entries_by_path_edits, &());
1514 self.entries_by_id.edit(entries_by_id_edits, &());
1515
1516 Ok(())
1517 }
1518
1519 pub fn file_count(&self) -> usize {
1520 self.entries_by_path.summary().file_count
1521 }
1522
1523 pub fn visible_file_count(&self) -> usize {
1524 self.entries_by_path.summary().visible_file_count
1525 }
1526
1527 fn traverse_from_offset(
1528 &self,
1529 include_dirs: bool,
1530 include_ignored: bool,
1531 start_offset: usize,
1532 ) -> Traversal {
1533 let mut cursor = self.entries_by_path.cursor();
1534 cursor.seek(
1535 &TraversalTarget::Count {
1536 count: start_offset,
1537 include_dirs,
1538 include_ignored,
1539 },
1540 Bias::Right,
1541 &(),
1542 );
1543 Traversal {
1544 cursor,
1545 include_dirs,
1546 include_ignored,
1547 }
1548 }
1549
1550 fn traverse_from_path(
1551 &self,
1552 include_dirs: bool,
1553 include_ignored: bool,
1554 path: &Path,
1555 ) -> Traversal {
1556 let mut cursor = self.entries_by_path.cursor();
1557 cursor.seek(&TraversalTarget::Path(path), Bias::Left, &());
1558 Traversal {
1559 cursor,
1560 include_dirs,
1561 include_ignored,
1562 }
1563 }
1564
1565 pub fn files(&self, include_ignored: bool, start: usize) -> Traversal {
1566 self.traverse_from_offset(false, include_ignored, start)
1567 }
1568
1569 pub fn entries(&self, include_ignored: bool) -> Traversal {
1570 self.traverse_from_offset(true, include_ignored, 0)
1571 }
1572
1573 pub fn paths(&self) -> impl Iterator<Item = &Arc<Path>> {
1574 let empty_path = Path::new("");
1575 self.entries_by_path
1576 .cursor::<()>()
1577 .filter(move |entry| entry.path.as_ref() != empty_path)
1578 .map(|entry| &entry.path)
1579 }
1580
1581 fn child_entries<'a>(&'a self, parent_path: &'a Path) -> ChildEntriesIter<'a> {
1582 let mut cursor = self.entries_by_path.cursor();
1583 cursor.seek(&TraversalTarget::Path(parent_path), Bias::Right, &());
1584 let traversal = Traversal {
1585 cursor,
1586 include_dirs: true,
1587 include_ignored: true,
1588 };
1589 ChildEntriesIter {
1590 traversal,
1591 parent_path,
1592 }
1593 }
1594
1595 pub fn root_entry(&self) -> Option<&Entry> {
1596 self.entry_for_path("")
1597 }
1598
1599 pub fn root_name(&self) -> &str {
1600 &self.root_name
1601 }
1602
1603 pub fn entry_for_path(&self, path: impl AsRef<Path>) -> Option<&Entry> {
1604 let path = path.as_ref();
1605 self.traverse_from_path(true, true, path)
1606 .entry()
1607 .and_then(|entry| {
1608 if entry.path.as_ref() == path {
1609 Some(entry)
1610 } else {
1611 None
1612 }
1613 })
1614 }
1615
1616 pub fn entry_for_id(&self, id: usize) -> Option<&Entry> {
1617 let entry = self.entries_by_id.get(&id, &())?;
1618 self.entry_for_path(&entry.path)
1619 }
1620
1621 pub fn inode_for_path(&self, path: impl AsRef<Path>) -> Option<u64> {
1622 self.entry_for_path(path.as_ref()).map(|e| e.inode)
1623 }
1624
1625 fn insert_entry(&mut self, mut entry: Entry, fs: &dyn Fs) -> Entry {
1626 if !entry.is_dir() && entry.path.file_name() == Some(&GITIGNORE) {
1627 let abs_path = self.abs_path.join(&entry.path);
1628 match build_gitignore(&abs_path, fs) {
1629 Ok(ignore) => {
1630 let ignore_dir_path = entry.path.parent().unwrap();
1631 self.ignores
1632 .insert(ignore_dir_path.into(), (Arc::new(ignore), self.scan_id));
1633 }
1634 Err(error) => {
1635 log::error!(
1636 "error loading .gitignore file {:?} - {:?}",
1637 &entry.path,
1638 error
1639 );
1640 }
1641 }
1642 }
1643
1644 self.reuse_entry_id(&mut entry);
1645 self.entries_by_path.insert_or_replace(entry.clone(), &());
1646 self.entries_by_id.insert_or_replace(
1647 PathEntry {
1648 id: entry.id,
1649 path: entry.path.clone(),
1650 is_ignored: entry.is_ignored,
1651 scan_id: self.scan_id,
1652 },
1653 &(),
1654 );
1655 entry
1656 }
1657
1658 fn populate_dir(
1659 &mut self,
1660 parent_path: Arc<Path>,
1661 entries: impl IntoIterator<Item = Entry>,
1662 ignore: Option<Arc<Gitignore>>,
1663 ) {
1664 let mut parent_entry = self
1665 .entries_by_path
1666 .get(&PathKey(parent_path.clone()), &())
1667 .unwrap()
1668 .clone();
1669 if let Some(ignore) = ignore {
1670 self.ignores.insert(parent_path, (ignore, self.scan_id));
1671 }
1672 if matches!(parent_entry.kind, EntryKind::PendingDir) {
1673 parent_entry.kind = EntryKind::Dir;
1674 } else {
1675 unreachable!();
1676 }
1677
1678 let mut entries_by_path_edits = vec![Edit::Insert(parent_entry)];
1679 let mut entries_by_id_edits = Vec::new();
1680
1681 for mut entry in entries {
1682 self.reuse_entry_id(&mut entry);
1683 entries_by_id_edits.push(Edit::Insert(PathEntry {
1684 id: entry.id,
1685 path: entry.path.clone(),
1686 is_ignored: entry.is_ignored,
1687 scan_id: self.scan_id,
1688 }));
1689 entries_by_path_edits.push(Edit::Insert(entry));
1690 }
1691
1692 self.entries_by_path.edit(entries_by_path_edits, &());
1693 self.entries_by_id.edit(entries_by_id_edits, &());
1694 }
1695
1696 fn reuse_entry_id(&mut self, entry: &mut Entry) {
1697 if let Some(removed_entry_id) = self.removed_entry_ids.remove(&entry.inode) {
1698 entry.id = removed_entry_id;
1699 } else if let Some(existing_entry) = self.entry_for_path(&entry.path) {
1700 entry.id = existing_entry.id;
1701 }
1702 }
1703
1704 fn remove_path(&mut self, path: &Path) {
1705 let mut new_entries;
1706 let removed_entries;
1707 {
1708 let mut cursor = self.entries_by_path.cursor::<TraversalProgress>();
1709 new_entries = cursor.slice(&TraversalTarget::Path(path), Bias::Left, &());
1710 removed_entries = cursor.slice(&TraversalTarget::PathSuccessor(path), Bias::Left, &());
1711 new_entries.push_tree(cursor.suffix(&()), &());
1712 }
1713 self.entries_by_path = new_entries;
1714
1715 let mut entries_by_id_edits = Vec::new();
1716 for entry in removed_entries.cursor::<()>() {
1717 let removed_entry_id = self
1718 .removed_entry_ids
1719 .entry(entry.inode)
1720 .or_insert(entry.id);
1721 *removed_entry_id = cmp::max(*removed_entry_id, entry.id);
1722 entries_by_id_edits.push(Edit::Remove(entry.id));
1723 }
1724 self.entries_by_id.edit(entries_by_id_edits, &());
1725
1726 if path.file_name() == Some(&GITIGNORE) {
1727 if let Some((_, scan_id)) = self.ignores.get_mut(path.parent().unwrap()) {
1728 *scan_id = self.scan_id;
1729 }
1730 }
1731 }
1732
1733 fn ignore_stack_for_path(&self, path: &Path, is_dir: bool) -> Arc<IgnoreStack> {
1734 let mut new_ignores = Vec::new();
1735 for ancestor in path.ancestors().skip(1) {
1736 if let Some((ignore, _)) = self.ignores.get(ancestor) {
1737 new_ignores.push((ancestor, Some(ignore.clone())));
1738 } else {
1739 new_ignores.push((ancestor, None));
1740 }
1741 }
1742
1743 let mut ignore_stack = IgnoreStack::none();
1744 for (parent_path, ignore) in new_ignores.into_iter().rev() {
1745 if ignore_stack.is_path_ignored(&parent_path, true) {
1746 ignore_stack = IgnoreStack::all();
1747 break;
1748 } else if let Some(ignore) = ignore {
1749 ignore_stack = ignore_stack.append(Arc::from(parent_path), ignore);
1750 }
1751 }
1752
1753 if ignore_stack.is_path_ignored(path, is_dir) {
1754 ignore_stack = IgnoreStack::all();
1755 }
1756
1757 ignore_stack
1758 }
1759}
1760
1761impl fmt::Debug for Snapshot {
1762 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1763 for entry in self.entries_by_path.cursor::<()>() {
1764 for _ in entry.path.ancestors().skip(1) {
1765 write!(f, " ")?;
1766 }
1767 writeln!(f, "{:?} (inode: {})", entry.path, entry.inode)?;
1768 }
1769 Ok(())
1770 }
1771}
1772
1773#[derive(Clone, PartialEq)]
1774pub struct File {
1775 entry_id: Option<usize>,
1776 worktree: ModelHandle<Worktree>,
1777 pub path: Arc<Path>,
1778 pub mtime: SystemTime,
1779}
1780
1781impl File {
1782 pub fn new(
1783 entry_id: usize,
1784 worktree: ModelHandle<Worktree>,
1785 path: Arc<Path>,
1786 mtime: SystemTime,
1787 ) -> Self {
1788 Self {
1789 entry_id: Some(entry_id),
1790 worktree,
1791 path,
1792 mtime,
1793 }
1794 }
1795}
1796
1797impl buffer::File for File {
1798 fn worktree_id(&self) -> usize {
1799 self.worktree.id()
1800 }
1801
1802 fn entry_id(&self) -> Option<usize> {
1803 self.entry_id
1804 }
1805
1806 fn set_entry_id(&mut self, entry_id: Option<usize>) {
1807 self.entry_id = entry_id;
1808 }
1809
1810 fn mtime(&self) -> SystemTime {
1811 self.mtime
1812 }
1813
1814 fn set_mtime(&mut self, mtime: SystemTime) {
1815 self.mtime = mtime;
1816 }
1817
1818 fn path(&self) -> &Arc<Path> {
1819 &self.path
1820 }
1821
1822 fn set_path(&mut self, path: Arc<Path>) {
1823 self.path = path;
1824 }
1825
1826 fn full_path(&self, cx: &AppContext) -> PathBuf {
1827 let worktree = self.worktree.read(cx);
1828 let mut full_path = PathBuf::new();
1829 full_path.push(worktree.root_name());
1830 full_path.push(&self.path);
1831 full_path
1832 }
1833
1834 /// Returns the last component of this handle's absolute path. If this handle refers to the root
1835 /// of its worktree, then this method will return the name of the worktree itself.
1836 fn file_name<'a>(&'a self, cx: &'a AppContext) -> Option<OsString> {
1837 self.path
1838 .file_name()
1839 .or_else(|| Some(OsStr::new(self.worktree.read(cx).root_name())))
1840 .map(Into::into)
1841 }
1842
1843 fn is_deleted(&self) -> bool {
1844 self.entry_id.is_none()
1845 }
1846
1847 fn save(
1848 &self,
1849 buffer_id: u64,
1850 text: Rope,
1851 version: clock::Global,
1852 cx: &mut MutableAppContext,
1853 ) -> Task<Result<(clock::Global, SystemTime)>> {
1854 self.worktree.update(cx, |worktree, cx| match worktree {
1855 Worktree::Local(worktree) => {
1856 let rpc = worktree.rpc.clone();
1857 let worktree_id = *worktree.remote_id.borrow();
1858 let save = worktree.save(self.path.clone(), text, cx);
1859 cx.background().spawn(async move {
1860 let entry = save.await?;
1861 if let Some(worktree_id) = worktree_id {
1862 rpc.send(proto::BufferSaved {
1863 worktree_id,
1864 buffer_id,
1865 version: (&version).into(),
1866 mtime: Some(entry.mtime.into()),
1867 })
1868 .await?;
1869 }
1870 Ok((version, entry.mtime))
1871 })
1872 }
1873 Worktree::Remote(worktree) => {
1874 let rpc = worktree.client.clone();
1875 let worktree_id = worktree.remote_id;
1876 cx.foreground().spawn(async move {
1877 let response = rpc
1878 .request(proto::SaveBuffer {
1879 worktree_id,
1880 buffer_id,
1881 })
1882 .await?;
1883 let version = response.version.try_into()?;
1884 let mtime = response
1885 .mtime
1886 .ok_or_else(|| anyhow!("missing mtime"))?
1887 .into();
1888 Ok((version, mtime))
1889 })
1890 }
1891 })
1892 }
1893
1894 fn buffer_updated(&self, buffer_id: u64, operation: Operation, cx: &mut MutableAppContext) {
1895 self.worktree.update(cx, |worktree, cx| {
1896 if let Some((rpc, remote_id)) = match worktree {
1897 Worktree::Local(worktree) => worktree
1898 .remote_id
1899 .borrow()
1900 .map(|id| (worktree.rpc.clone(), id)),
1901 Worktree::Remote(worktree) => Some((worktree.client.clone(), worktree.remote_id)),
1902 } {
1903 cx.spawn(|worktree, mut cx| async move {
1904 if let Err(error) = rpc
1905 .request(proto::UpdateBuffer {
1906 worktree_id: remote_id,
1907 buffer_id,
1908 operations: vec![(&operation).into()],
1909 })
1910 .await
1911 {
1912 worktree.update(&mut cx, |worktree, _| {
1913 log::error!("error sending buffer operation: {}", error);
1914 match worktree {
1915 Worktree::Local(t) => &mut t.queued_operations,
1916 Worktree::Remote(t) => &mut t.queued_operations,
1917 }
1918 .push((buffer_id, operation));
1919 });
1920 }
1921 })
1922 .detach();
1923 }
1924 });
1925 }
1926
1927 fn buffer_removed(&self, buffer_id: u64, cx: &mut MutableAppContext) {
1928 self.worktree.update(cx, |worktree, cx| {
1929 if let Worktree::Remote(worktree) = worktree {
1930 let worktree_id = worktree.remote_id;
1931 let rpc = worktree.client.clone();
1932 cx.background()
1933 .spawn(async move {
1934 if let Err(error) = rpc
1935 .send(proto::CloseBuffer {
1936 worktree_id,
1937 buffer_id,
1938 })
1939 .await
1940 {
1941 log::error!("error closing remote buffer: {}", error);
1942 }
1943 })
1944 .detach();
1945 }
1946 });
1947 }
1948
1949 fn boxed_clone(&self) -> Box<dyn buffer::File> {
1950 Box::new(self.clone())
1951 }
1952
1953 fn as_any(&self) -> &dyn Any {
1954 self
1955 }
1956}
1957
1958#[derive(Clone, Debug)]
1959pub struct Entry {
1960 pub id: usize,
1961 pub kind: EntryKind,
1962 pub path: Arc<Path>,
1963 pub inode: u64,
1964 pub mtime: SystemTime,
1965 pub is_symlink: bool,
1966 pub is_ignored: bool,
1967}
1968
1969#[derive(Clone, Debug)]
1970pub enum EntryKind {
1971 PendingDir,
1972 Dir,
1973 File(CharBag),
1974}
1975
1976impl Entry {
1977 fn new(
1978 path: Arc<Path>,
1979 metadata: &fs::Metadata,
1980 next_entry_id: &AtomicUsize,
1981 root_char_bag: CharBag,
1982 ) -> Self {
1983 Self {
1984 id: next_entry_id.fetch_add(1, SeqCst),
1985 kind: if metadata.is_dir {
1986 EntryKind::PendingDir
1987 } else {
1988 EntryKind::File(char_bag_for_path(root_char_bag, &path))
1989 },
1990 path,
1991 inode: metadata.inode,
1992 mtime: metadata.mtime,
1993 is_symlink: metadata.is_symlink,
1994 is_ignored: false,
1995 }
1996 }
1997
1998 pub fn is_dir(&self) -> bool {
1999 matches!(self.kind, EntryKind::Dir | EntryKind::PendingDir)
2000 }
2001
2002 pub fn is_file(&self) -> bool {
2003 matches!(self.kind, EntryKind::File(_))
2004 }
2005}
2006
2007impl sum_tree::Item for Entry {
2008 type Summary = EntrySummary;
2009
2010 fn summary(&self) -> Self::Summary {
2011 let visible_count = if self.is_ignored { 0 } else { 1 };
2012 let file_count;
2013 let visible_file_count;
2014 if self.is_file() {
2015 file_count = 1;
2016 visible_file_count = visible_count;
2017 } else {
2018 file_count = 0;
2019 visible_file_count = 0;
2020 }
2021
2022 EntrySummary {
2023 max_path: self.path.clone(),
2024 count: 1,
2025 visible_count,
2026 file_count,
2027 visible_file_count,
2028 }
2029 }
2030}
2031
2032impl sum_tree::KeyedItem for Entry {
2033 type Key = PathKey;
2034
2035 fn key(&self) -> Self::Key {
2036 PathKey(self.path.clone())
2037 }
2038}
2039
2040#[derive(Clone, Debug)]
2041pub struct EntrySummary {
2042 max_path: Arc<Path>,
2043 count: usize,
2044 visible_count: usize,
2045 file_count: usize,
2046 visible_file_count: usize,
2047}
2048
2049impl Default for EntrySummary {
2050 fn default() -> Self {
2051 Self {
2052 max_path: Arc::from(Path::new("")),
2053 count: 0,
2054 visible_count: 0,
2055 file_count: 0,
2056 visible_file_count: 0,
2057 }
2058 }
2059}
2060
2061impl sum_tree::Summary for EntrySummary {
2062 type Context = ();
2063
2064 fn add_summary(&mut self, rhs: &Self, _: &()) {
2065 self.max_path = rhs.max_path.clone();
2066 self.visible_count += rhs.visible_count;
2067 self.file_count += rhs.file_count;
2068 self.visible_file_count += rhs.visible_file_count;
2069 }
2070}
2071
2072#[derive(Clone, Debug)]
2073struct PathEntry {
2074 id: usize,
2075 path: Arc<Path>,
2076 is_ignored: bool,
2077 scan_id: usize,
2078}
2079
2080impl sum_tree::Item for PathEntry {
2081 type Summary = PathEntrySummary;
2082
2083 fn summary(&self) -> Self::Summary {
2084 PathEntrySummary { max_id: self.id }
2085 }
2086}
2087
2088impl sum_tree::KeyedItem for PathEntry {
2089 type Key = usize;
2090
2091 fn key(&self) -> Self::Key {
2092 self.id
2093 }
2094}
2095
2096#[derive(Clone, Debug, Default)]
2097struct PathEntrySummary {
2098 max_id: usize,
2099}
2100
2101impl sum_tree::Summary for PathEntrySummary {
2102 type Context = ();
2103
2104 fn add_summary(&mut self, summary: &Self, _: &Self::Context) {
2105 self.max_id = summary.max_id;
2106 }
2107}
2108
2109impl<'a> sum_tree::Dimension<'a, PathEntrySummary> for usize {
2110 fn add_summary(&mut self, summary: &'a PathEntrySummary, _: &()) {
2111 *self = summary.max_id;
2112 }
2113}
2114
2115#[derive(Clone, Debug, Eq, PartialEq, Ord, PartialOrd)]
2116pub struct PathKey(Arc<Path>);
2117
2118impl Default for PathKey {
2119 fn default() -> Self {
2120 Self(Path::new("").into())
2121 }
2122}
2123
2124impl<'a> sum_tree::Dimension<'a, EntrySummary> for PathKey {
2125 fn add_summary(&mut self, summary: &'a EntrySummary, _: &()) {
2126 self.0 = summary.max_path.clone();
2127 }
2128}
2129
2130struct BackgroundScanner {
2131 fs: Arc<dyn Fs>,
2132 snapshot: Arc<Mutex<Snapshot>>,
2133 notify: Sender<ScanState>,
2134 executor: Arc<executor::Background>,
2135}
2136
2137impl BackgroundScanner {
2138 fn new(
2139 snapshot: Arc<Mutex<Snapshot>>,
2140 notify: Sender<ScanState>,
2141 fs: Arc<dyn Fs>,
2142 executor: Arc<executor::Background>,
2143 ) -> Self {
2144 Self {
2145 fs,
2146 snapshot,
2147 notify,
2148 executor,
2149 }
2150 }
2151
2152 fn abs_path(&self) -> Arc<Path> {
2153 self.snapshot.lock().abs_path.clone()
2154 }
2155
2156 fn snapshot(&self) -> Snapshot {
2157 self.snapshot.lock().clone()
2158 }
2159
2160 async fn run(mut self, events_rx: impl Stream<Item = Vec<fsevent::Event>>) {
2161 if self.notify.send(ScanState::Scanning).await.is_err() {
2162 return;
2163 }
2164
2165 if let Err(err) = self.scan_dirs().await {
2166 if self
2167 .notify
2168 .send(ScanState::Err(Arc::new(err)))
2169 .await
2170 .is_err()
2171 {
2172 return;
2173 }
2174 }
2175
2176 if self.notify.send(ScanState::Idle).await.is_err() {
2177 return;
2178 }
2179
2180 futures::pin_mut!(events_rx);
2181 while let Some(events) = events_rx.next().await {
2182 if self.notify.send(ScanState::Scanning).await.is_err() {
2183 break;
2184 }
2185
2186 if !self.process_events(events).await {
2187 break;
2188 }
2189
2190 if self.notify.send(ScanState::Idle).await.is_err() {
2191 break;
2192 }
2193 }
2194 }
2195
2196 async fn scan_dirs(&mut self) -> Result<()> {
2197 let root_char_bag;
2198 let next_entry_id;
2199 let is_dir;
2200 {
2201 let snapshot = self.snapshot.lock();
2202 root_char_bag = snapshot.root_char_bag;
2203 next_entry_id = snapshot.next_entry_id.clone();
2204 is_dir = snapshot.root_entry().map_or(false, |e| e.is_dir())
2205 };
2206
2207 if is_dir {
2208 let path: Arc<Path> = Arc::from(Path::new(""));
2209 let abs_path = self.abs_path();
2210 let (tx, rx) = channel::unbounded();
2211 tx.send(ScanJob {
2212 abs_path: abs_path.to_path_buf(),
2213 path,
2214 ignore_stack: IgnoreStack::none(),
2215 scan_queue: tx.clone(),
2216 })
2217 .await
2218 .unwrap();
2219 drop(tx);
2220
2221 self.executor
2222 .scoped(|scope| {
2223 for _ in 0..self.executor.num_cpus() {
2224 scope.spawn(async {
2225 while let Ok(job) = rx.recv().await {
2226 if let Err(err) = self
2227 .scan_dir(root_char_bag, next_entry_id.clone(), &job)
2228 .await
2229 {
2230 log::error!("error scanning {:?}: {}", job.abs_path, err);
2231 }
2232 }
2233 });
2234 }
2235 })
2236 .await;
2237 }
2238
2239 Ok(())
2240 }
2241
2242 async fn scan_dir(
2243 &self,
2244 root_char_bag: CharBag,
2245 next_entry_id: Arc<AtomicUsize>,
2246 job: &ScanJob,
2247 ) -> Result<()> {
2248 let mut new_entries: Vec<Entry> = Vec::new();
2249 let mut new_jobs: Vec<ScanJob> = Vec::new();
2250 let mut ignore_stack = job.ignore_stack.clone();
2251 let mut new_ignore = None;
2252
2253 let mut child_paths = self.fs.read_dir(&job.abs_path).await?;
2254 while let Some(child_abs_path) = child_paths.next().await {
2255 let child_abs_path = match child_abs_path {
2256 Ok(child_abs_path) => child_abs_path,
2257 Err(error) => {
2258 log::error!("error processing entry {:?}", error);
2259 continue;
2260 }
2261 };
2262 let child_name = child_abs_path.file_name().unwrap();
2263 let child_path: Arc<Path> = job.path.join(child_name).into();
2264 let child_metadata = match self.fs.metadata(&child_abs_path).await? {
2265 Some(metadata) => metadata,
2266 None => continue,
2267 };
2268
2269 // If we find a .gitignore, add it to the stack of ignores used to determine which paths are ignored
2270 if child_name == *GITIGNORE {
2271 match build_gitignore(&child_abs_path, self.fs.as_ref()) {
2272 Ok(ignore) => {
2273 let ignore = Arc::new(ignore);
2274 ignore_stack = ignore_stack.append(job.path.clone(), ignore.clone());
2275 new_ignore = Some(ignore);
2276 }
2277 Err(error) => {
2278 log::error!(
2279 "error loading .gitignore file {:?} - {:?}",
2280 child_name,
2281 error
2282 );
2283 }
2284 }
2285
2286 // Update ignore status of any child entries we've already processed to reflect the
2287 // ignore file in the current directory. Because `.gitignore` starts with a `.`,
2288 // there should rarely be too numerous. Update the ignore stack associated with any
2289 // new jobs as well.
2290 let mut new_jobs = new_jobs.iter_mut();
2291 for entry in &mut new_entries {
2292 entry.is_ignored = ignore_stack.is_path_ignored(&entry.path, entry.is_dir());
2293 if entry.is_dir() {
2294 new_jobs.next().unwrap().ignore_stack = if entry.is_ignored {
2295 IgnoreStack::all()
2296 } else {
2297 ignore_stack.clone()
2298 };
2299 }
2300 }
2301 }
2302
2303 let mut child_entry = Entry::new(
2304 child_path.clone(),
2305 &child_metadata,
2306 &next_entry_id,
2307 root_char_bag,
2308 );
2309
2310 if child_metadata.is_dir {
2311 let is_ignored = ignore_stack.is_path_ignored(&child_path, true);
2312 child_entry.is_ignored = is_ignored;
2313 new_entries.push(child_entry);
2314 new_jobs.push(ScanJob {
2315 abs_path: child_abs_path,
2316 path: child_path,
2317 ignore_stack: if is_ignored {
2318 IgnoreStack::all()
2319 } else {
2320 ignore_stack.clone()
2321 },
2322 scan_queue: job.scan_queue.clone(),
2323 });
2324 } else {
2325 child_entry.is_ignored = ignore_stack.is_path_ignored(&child_path, false);
2326 new_entries.push(child_entry);
2327 };
2328 }
2329
2330 self.snapshot
2331 .lock()
2332 .populate_dir(job.path.clone(), new_entries, new_ignore);
2333 for new_job in new_jobs {
2334 job.scan_queue.send(new_job).await.unwrap();
2335 }
2336
2337 Ok(())
2338 }
2339
2340 async fn process_events(&mut self, mut events: Vec<fsevent::Event>) -> bool {
2341 let mut snapshot = self.snapshot();
2342 snapshot.scan_id += 1;
2343
2344 let root_abs_path = if let Ok(abs_path) = self.fs.canonicalize(&snapshot.abs_path).await {
2345 abs_path
2346 } else {
2347 return false;
2348 };
2349 let root_char_bag = snapshot.root_char_bag;
2350 let next_entry_id = snapshot.next_entry_id.clone();
2351
2352 events.sort_unstable_by(|a, b| a.path.cmp(&b.path));
2353 events.dedup_by(|a, b| a.path.starts_with(&b.path));
2354
2355 for event in &events {
2356 match event.path.strip_prefix(&root_abs_path) {
2357 Ok(path) => snapshot.remove_path(&path),
2358 Err(_) => {
2359 log::error!(
2360 "unexpected event {:?} for root path {:?}",
2361 event.path,
2362 root_abs_path
2363 );
2364 continue;
2365 }
2366 }
2367 }
2368
2369 let (scan_queue_tx, scan_queue_rx) = channel::unbounded();
2370 for event in events {
2371 let path: Arc<Path> = match event.path.strip_prefix(&root_abs_path) {
2372 Ok(path) => Arc::from(path.to_path_buf()),
2373 Err(_) => {
2374 log::error!(
2375 "unexpected event {:?} for root path {:?}",
2376 event.path,
2377 root_abs_path
2378 );
2379 continue;
2380 }
2381 };
2382
2383 match self.fs.metadata(&event.path).await {
2384 Ok(Some(metadata)) => {
2385 let ignore_stack = snapshot.ignore_stack_for_path(&path, metadata.is_dir);
2386 let mut fs_entry = Entry::new(
2387 path.clone(),
2388 &metadata,
2389 snapshot.next_entry_id.as_ref(),
2390 snapshot.root_char_bag,
2391 );
2392 fs_entry.is_ignored = ignore_stack.is_all();
2393 snapshot.insert_entry(fs_entry, self.fs.as_ref());
2394 if metadata.is_dir {
2395 scan_queue_tx
2396 .send(ScanJob {
2397 abs_path: event.path,
2398 path,
2399 ignore_stack,
2400 scan_queue: scan_queue_tx.clone(),
2401 })
2402 .await
2403 .unwrap();
2404 }
2405 }
2406 Ok(None) => {}
2407 Err(err) => {
2408 // TODO - create a special 'error' entry in the entries tree to mark this
2409 log::error!("error reading file on event {:?}", err);
2410 }
2411 }
2412 }
2413
2414 *self.snapshot.lock() = snapshot;
2415
2416 // Scan any directories that were created as part of this event batch.
2417 drop(scan_queue_tx);
2418 self.executor
2419 .scoped(|scope| {
2420 for _ in 0..self.executor.num_cpus() {
2421 scope.spawn(async {
2422 while let Ok(job) = scan_queue_rx.recv().await {
2423 if let Err(err) = self
2424 .scan_dir(root_char_bag, next_entry_id.clone(), &job)
2425 .await
2426 {
2427 log::error!("error scanning {:?}: {}", job.abs_path, err);
2428 }
2429 }
2430 });
2431 }
2432 })
2433 .await;
2434
2435 // Attempt to detect renames only over a single batch of file-system events.
2436 self.snapshot.lock().removed_entry_ids.clear();
2437
2438 self.update_ignore_statuses().await;
2439 true
2440 }
2441
2442 async fn update_ignore_statuses(&self) {
2443 let mut snapshot = self.snapshot();
2444
2445 let mut ignores_to_update = Vec::new();
2446 let mut ignores_to_delete = Vec::new();
2447 for (parent_path, (_, scan_id)) in &snapshot.ignores {
2448 if *scan_id == snapshot.scan_id && snapshot.entry_for_path(parent_path).is_some() {
2449 ignores_to_update.push(parent_path.clone());
2450 }
2451
2452 let ignore_path = parent_path.join(&*GITIGNORE);
2453 if snapshot.entry_for_path(ignore_path).is_none() {
2454 ignores_to_delete.push(parent_path.clone());
2455 }
2456 }
2457
2458 for parent_path in ignores_to_delete {
2459 snapshot.ignores.remove(&parent_path);
2460 self.snapshot.lock().ignores.remove(&parent_path);
2461 }
2462
2463 let (ignore_queue_tx, ignore_queue_rx) = channel::unbounded();
2464 ignores_to_update.sort_unstable();
2465 let mut ignores_to_update = ignores_to_update.into_iter().peekable();
2466 while let Some(parent_path) = ignores_to_update.next() {
2467 while ignores_to_update
2468 .peek()
2469 .map_or(false, |p| p.starts_with(&parent_path))
2470 {
2471 ignores_to_update.next().unwrap();
2472 }
2473
2474 let ignore_stack = snapshot.ignore_stack_for_path(&parent_path, true);
2475 ignore_queue_tx
2476 .send(UpdateIgnoreStatusJob {
2477 path: parent_path,
2478 ignore_stack,
2479 ignore_queue: ignore_queue_tx.clone(),
2480 })
2481 .await
2482 .unwrap();
2483 }
2484 drop(ignore_queue_tx);
2485
2486 self.executor
2487 .scoped(|scope| {
2488 for _ in 0..self.executor.num_cpus() {
2489 scope.spawn(async {
2490 while let Ok(job) = ignore_queue_rx.recv().await {
2491 self.update_ignore_status(job, &snapshot).await;
2492 }
2493 });
2494 }
2495 })
2496 .await;
2497 }
2498
2499 async fn update_ignore_status(&self, job: UpdateIgnoreStatusJob, snapshot: &Snapshot) {
2500 let mut ignore_stack = job.ignore_stack;
2501 if let Some((ignore, _)) = snapshot.ignores.get(&job.path) {
2502 ignore_stack = ignore_stack.append(job.path.clone(), ignore.clone());
2503 }
2504
2505 let mut entries_by_id_edits = Vec::new();
2506 let mut entries_by_path_edits = Vec::new();
2507 for mut entry in snapshot.child_entries(&job.path).cloned() {
2508 let was_ignored = entry.is_ignored;
2509 entry.is_ignored = ignore_stack.is_path_ignored(&entry.path, entry.is_dir());
2510 if entry.is_dir() {
2511 let child_ignore_stack = if entry.is_ignored {
2512 IgnoreStack::all()
2513 } else {
2514 ignore_stack.clone()
2515 };
2516 job.ignore_queue
2517 .send(UpdateIgnoreStatusJob {
2518 path: entry.path.clone(),
2519 ignore_stack: child_ignore_stack,
2520 ignore_queue: job.ignore_queue.clone(),
2521 })
2522 .await
2523 .unwrap();
2524 }
2525
2526 if entry.is_ignored != was_ignored {
2527 let mut path_entry = snapshot.entries_by_id.get(&entry.id, &()).unwrap().clone();
2528 path_entry.scan_id = snapshot.scan_id;
2529 path_entry.is_ignored = entry.is_ignored;
2530 entries_by_id_edits.push(Edit::Insert(path_entry));
2531 entries_by_path_edits.push(Edit::Insert(entry));
2532 }
2533 }
2534
2535 let mut snapshot = self.snapshot.lock();
2536 snapshot.entries_by_path.edit(entries_by_path_edits, &());
2537 snapshot.entries_by_id.edit(entries_by_id_edits, &());
2538 }
2539}
2540
2541async fn refresh_entry(
2542 fs: &dyn Fs,
2543 snapshot: &Mutex<Snapshot>,
2544 path: Arc<Path>,
2545 abs_path: &Path,
2546) -> Result<Entry> {
2547 let root_char_bag;
2548 let next_entry_id;
2549 {
2550 let snapshot = snapshot.lock();
2551 root_char_bag = snapshot.root_char_bag;
2552 next_entry_id = snapshot.next_entry_id.clone();
2553 }
2554 let entry = Entry::new(
2555 path,
2556 &fs.metadata(abs_path)
2557 .await?
2558 .ok_or_else(|| anyhow!("could not read saved file metadata"))?,
2559 &next_entry_id,
2560 root_char_bag,
2561 );
2562 Ok(snapshot.lock().insert_entry(entry, fs))
2563}
2564
2565fn char_bag_for_path(root_char_bag: CharBag, path: &Path) -> CharBag {
2566 let mut result = root_char_bag;
2567 result.extend(
2568 path.to_string_lossy()
2569 .chars()
2570 .map(|c| c.to_ascii_lowercase()),
2571 );
2572 result
2573}
2574
2575struct ScanJob {
2576 abs_path: PathBuf,
2577 path: Arc<Path>,
2578 ignore_stack: Arc<IgnoreStack>,
2579 scan_queue: Sender<ScanJob>,
2580}
2581
2582struct UpdateIgnoreStatusJob {
2583 path: Arc<Path>,
2584 ignore_stack: Arc<IgnoreStack>,
2585 ignore_queue: Sender<UpdateIgnoreStatusJob>,
2586}
2587
2588pub trait WorktreeHandle {
2589 #[cfg(test)]
2590 fn flush_fs_events<'a>(
2591 &self,
2592 cx: &'a gpui::TestAppContext,
2593 ) -> futures::future::LocalBoxFuture<'a, ()>;
2594}
2595
2596impl WorktreeHandle for ModelHandle<Worktree> {
2597 // When the worktree's FS event stream sometimes delivers "redundant" events for FS changes that
2598 // occurred before the worktree was constructed. These events can cause the worktree to perfrom
2599 // extra directory scans, and emit extra scan-state notifications.
2600 //
2601 // This function mutates the worktree's directory and waits for those mutations to be picked up,
2602 // to ensure that all redundant FS events have already been processed.
2603 #[cfg(test)]
2604 fn flush_fs_events<'a>(
2605 &self,
2606 cx: &'a gpui::TestAppContext,
2607 ) -> futures::future::LocalBoxFuture<'a, ()> {
2608 use smol::future::FutureExt;
2609
2610 let filename = "fs-event-sentinel";
2611 let root_path = cx.read(|cx| self.read(cx).abs_path.clone());
2612 let tree = self.clone();
2613 async move {
2614 std::fs::write(root_path.join(filename), "").unwrap();
2615 tree.condition(&cx, |tree, _| tree.entry_for_path(filename).is_some())
2616 .await;
2617
2618 std::fs::remove_file(root_path.join(filename)).unwrap();
2619 tree.condition(&cx, |tree, _| tree.entry_for_path(filename).is_none())
2620 .await;
2621
2622 cx.read(|cx| tree.read(cx).as_local().unwrap().scan_complete())
2623 .await;
2624 }
2625 .boxed_local()
2626 }
2627}
2628
2629#[derive(Clone, Debug)]
2630struct TraversalProgress<'a> {
2631 max_path: &'a Path,
2632 count: usize,
2633 visible_count: usize,
2634 file_count: usize,
2635 visible_file_count: usize,
2636}
2637
2638impl<'a> TraversalProgress<'a> {
2639 fn count(&self, include_dirs: bool, include_ignored: bool) -> usize {
2640 match (include_ignored, include_dirs) {
2641 (true, true) => self.count,
2642 (true, false) => self.file_count,
2643 (false, true) => self.visible_count,
2644 (false, false) => self.visible_file_count,
2645 }
2646 }
2647}
2648
2649impl<'a> sum_tree::Dimension<'a, EntrySummary> for TraversalProgress<'a> {
2650 fn add_summary(&mut self, summary: &'a EntrySummary, _: &()) {
2651 self.max_path = summary.max_path.as_ref();
2652 self.count += summary.count;
2653 self.visible_count += summary.visible_count;
2654 self.file_count += summary.file_count;
2655 self.visible_file_count += summary.visible_file_count;
2656 }
2657}
2658
2659impl<'a> Default for TraversalProgress<'a> {
2660 fn default() -> Self {
2661 Self {
2662 max_path: Path::new(""),
2663 count: 0,
2664 visible_count: 0,
2665 file_count: 0,
2666 visible_file_count: 0,
2667 }
2668 }
2669}
2670
2671pub struct Traversal<'a> {
2672 cursor: sum_tree::Cursor<'a, Entry, TraversalProgress<'a>>,
2673 include_ignored: bool,
2674 include_dirs: bool,
2675}
2676
2677impl<'a> Traversal<'a> {
2678 pub fn advance(&mut self) -> bool {
2679 self.advance_to_offset(self.offset() + 1)
2680 }
2681
2682 pub fn advance_to_offset(&mut self, offset: usize) -> bool {
2683 self.cursor.seek_forward(
2684 &TraversalTarget::Count {
2685 count: offset,
2686 include_dirs: self.include_dirs,
2687 include_ignored: self.include_ignored,
2688 },
2689 Bias::Right,
2690 &(),
2691 )
2692 }
2693
2694 pub fn advance_to_sibling(&mut self) -> bool {
2695 while let Some(entry) = self.cursor.item() {
2696 self.cursor.seek_forward(
2697 &TraversalTarget::PathSuccessor(&entry.path),
2698 Bias::Left,
2699 &(),
2700 );
2701 if let Some(entry) = self.cursor.item() {
2702 if (self.include_dirs || !entry.is_dir())
2703 && (self.include_ignored || !entry.is_ignored)
2704 {
2705 return true;
2706 }
2707 }
2708 }
2709 false
2710 }
2711
2712 pub fn entry(&self) -> Option<&'a Entry> {
2713 self.cursor.item()
2714 }
2715
2716 pub fn offset(&self) -> usize {
2717 self.cursor
2718 .start()
2719 .count(self.include_dirs, self.include_ignored)
2720 }
2721}
2722
2723impl<'a> Iterator for Traversal<'a> {
2724 type Item = &'a Entry;
2725
2726 fn next(&mut self) -> Option<Self::Item> {
2727 if let Some(item) = self.entry() {
2728 self.advance();
2729 Some(item)
2730 } else {
2731 None
2732 }
2733 }
2734}
2735
2736#[derive(Debug)]
2737enum TraversalTarget<'a> {
2738 Path(&'a Path),
2739 PathSuccessor(&'a Path),
2740 Count {
2741 count: usize,
2742 include_ignored: bool,
2743 include_dirs: bool,
2744 },
2745}
2746
2747impl<'a, 'b> SeekTarget<'a, EntrySummary, TraversalProgress<'a>> for TraversalTarget<'b> {
2748 fn cmp(&self, cursor_location: &TraversalProgress<'a>, _: &()) -> Ordering {
2749 match self {
2750 TraversalTarget::Path(path) => path.cmp(&cursor_location.max_path),
2751 TraversalTarget::PathSuccessor(path) => {
2752 if !cursor_location.max_path.starts_with(path) {
2753 Ordering::Equal
2754 } else {
2755 Ordering::Greater
2756 }
2757 }
2758 TraversalTarget::Count {
2759 count,
2760 include_dirs,
2761 include_ignored,
2762 } => Ord::cmp(
2763 count,
2764 &cursor_location.count(*include_dirs, *include_ignored),
2765 ),
2766 }
2767 }
2768}
2769
2770struct ChildEntriesIter<'a> {
2771 parent_path: &'a Path,
2772 traversal: Traversal<'a>,
2773}
2774
2775impl<'a> Iterator for ChildEntriesIter<'a> {
2776 type Item = &'a Entry;
2777
2778 fn next(&mut self) -> Option<Self::Item> {
2779 if let Some(item) = self.traversal.entry() {
2780 if item.path.starts_with(&self.parent_path) {
2781 self.traversal.advance_to_sibling();
2782 return Some(item);
2783 }
2784 }
2785 None
2786 }
2787}
2788
2789impl<'a> From<&'a Entry> for proto::Entry {
2790 fn from(entry: &'a Entry) -> Self {
2791 Self {
2792 id: entry.id as u64,
2793 is_dir: entry.is_dir(),
2794 path: entry.path.to_string_lossy().to_string(),
2795 inode: entry.inode,
2796 mtime: Some(entry.mtime.into()),
2797 is_symlink: entry.is_symlink,
2798 is_ignored: entry.is_ignored,
2799 }
2800 }
2801}
2802
2803impl<'a> TryFrom<(&'a CharBag, proto::Entry)> for Entry {
2804 type Error = anyhow::Error;
2805
2806 fn try_from((root_char_bag, entry): (&'a CharBag, proto::Entry)) -> Result<Self> {
2807 if let Some(mtime) = entry.mtime {
2808 let kind = if entry.is_dir {
2809 EntryKind::Dir
2810 } else {
2811 let mut char_bag = root_char_bag.clone();
2812 char_bag.extend(entry.path.chars().map(|c| c.to_ascii_lowercase()));
2813 EntryKind::File(char_bag)
2814 };
2815 let path: Arc<Path> = Arc::from(Path::new(&entry.path));
2816 Ok(Entry {
2817 id: entry.id as usize,
2818 kind,
2819 path: path.clone(),
2820 inode: entry.inode,
2821 mtime: mtime.into(),
2822 is_symlink: entry.is_symlink,
2823 is_ignored: entry.is_ignored,
2824 })
2825 } else {
2826 Err(anyhow!(
2827 "missing mtime in remote worktree entry {:?}",
2828 entry.path
2829 ))
2830 }
2831 }
2832}
2833
2834#[cfg(test)]
2835mod tests {
2836 use super::*;
2837 use crate::fs::FakeFs;
2838 use anyhow::Result;
2839 use client::test::FakeServer;
2840 use fs::RealFs;
2841 use rand::prelude::*;
2842 use serde_json::json;
2843 use std::{cell::RefCell, rc::Rc};
2844 use std::{
2845 env,
2846 fmt::Write,
2847 time::{SystemTime, UNIX_EPOCH},
2848 };
2849 use util::test::temp_tree;
2850
2851 #[gpui::test]
2852 async fn test_traversal(cx: gpui::TestAppContext) {
2853 let fs = FakeFs::new();
2854 fs.insert_tree(
2855 "/root",
2856 json!({
2857 ".gitignore": "a/b\n",
2858 "a": {
2859 "b": "",
2860 "c": "",
2861 }
2862 }),
2863 )
2864 .await;
2865
2866 let tree = Worktree::open_local(
2867 Client::new(),
2868 Arc::from(Path::new("/root")),
2869 Arc::new(fs),
2870 Default::default(),
2871 &mut cx.to_async(),
2872 )
2873 .await
2874 .unwrap();
2875 cx.read(|cx| tree.read(cx).as_local().unwrap().scan_complete())
2876 .await;
2877
2878 tree.read_with(&cx, |tree, _| {
2879 assert_eq!(
2880 tree.entries(false)
2881 .map(|entry| entry.path.as_ref())
2882 .collect::<Vec<_>>(),
2883 vec![
2884 Path::new(""),
2885 Path::new(".gitignore"),
2886 Path::new("a"),
2887 Path::new("a/c"),
2888 ]
2889 );
2890 })
2891 }
2892
2893 #[gpui::test]
2894 async fn test_save_file(mut cx: gpui::TestAppContext) {
2895 let dir = temp_tree(json!({
2896 "file1": "the old contents",
2897 }));
2898 let tree = Worktree::open_local(
2899 Client::new(),
2900 dir.path(),
2901 Arc::new(RealFs),
2902 Default::default(),
2903 &mut cx.to_async(),
2904 )
2905 .await
2906 .unwrap();
2907 let buffer = tree
2908 .update(&mut cx, |tree, cx| tree.open_buffer("file1", cx))
2909 .await
2910 .unwrap();
2911 let save = buffer.update(&mut cx, |buffer, cx| {
2912 buffer.edit(Some(0..0), "a line of text.\n".repeat(10 * 1024), cx);
2913 buffer.save(cx).unwrap()
2914 });
2915 save.await.unwrap();
2916
2917 let new_text = std::fs::read_to_string(dir.path().join("file1")).unwrap();
2918 assert_eq!(new_text, buffer.read_with(&cx, |buffer, _| buffer.text()));
2919 }
2920
2921 #[gpui::test]
2922 async fn test_save_in_single_file_worktree(mut cx: gpui::TestAppContext) {
2923 let dir = temp_tree(json!({
2924 "file1": "the old contents",
2925 }));
2926 let file_path = dir.path().join("file1");
2927
2928 let tree = Worktree::open_local(
2929 Client::new(),
2930 file_path.clone(),
2931 Arc::new(RealFs),
2932 Default::default(),
2933 &mut cx.to_async(),
2934 )
2935 .await
2936 .unwrap();
2937 cx.read(|cx| tree.read(cx).as_local().unwrap().scan_complete())
2938 .await;
2939 cx.read(|cx| assert_eq!(tree.read(cx).file_count(), 1));
2940
2941 let buffer = tree
2942 .update(&mut cx, |tree, cx| tree.open_buffer("", cx))
2943 .await
2944 .unwrap();
2945 let save = buffer.update(&mut cx, |buffer, cx| {
2946 buffer.edit(Some(0..0), "a line of text.\n".repeat(10 * 1024), cx);
2947 buffer.save(cx).unwrap()
2948 });
2949 save.await.unwrap();
2950
2951 let new_text = std::fs::read_to_string(file_path).unwrap();
2952 assert_eq!(new_text, buffer.read_with(&cx, |buffer, _| buffer.text()));
2953 }
2954
2955 #[gpui::test]
2956 async fn test_rescan_and_remote_updates(mut cx: gpui::TestAppContext) {
2957 let dir = temp_tree(json!({
2958 "a": {
2959 "file1": "",
2960 "file2": "",
2961 "file3": "",
2962 },
2963 "b": {
2964 "c": {
2965 "file4": "",
2966 "file5": "",
2967 }
2968 }
2969 }));
2970
2971 let user_id = 5;
2972 let mut client = Client::new();
2973 let server = FakeServer::for_client(user_id, &mut client, &cx).await;
2974 let tree = Worktree::open_local(
2975 client,
2976 dir.path(),
2977 Arc::new(RealFs),
2978 Default::default(),
2979 &mut cx.to_async(),
2980 )
2981 .await
2982 .unwrap();
2983
2984 let buffer_for_path = |path: &'static str, cx: &mut gpui::TestAppContext| {
2985 let buffer = tree.update(cx, |tree, cx| tree.open_buffer(path, cx));
2986 async move { buffer.await.unwrap() }
2987 };
2988 let id_for_path = |path: &'static str, cx: &gpui::TestAppContext| {
2989 tree.read_with(cx, |tree, _| {
2990 tree.entry_for_path(path)
2991 .expect(&format!("no entry for path {}", path))
2992 .id
2993 })
2994 };
2995
2996 let buffer2 = buffer_for_path("a/file2", &mut cx).await;
2997 let buffer3 = buffer_for_path("a/file3", &mut cx).await;
2998 let buffer4 = buffer_for_path("b/c/file4", &mut cx).await;
2999 let buffer5 = buffer_for_path("b/c/file5", &mut cx).await;
3000
3001 let file2_id = id_for_path("a/file2", &cx);
3002 let file3_id = id_for_path("a/file3", &cx);
3003 let file4_id = id_for_path("b/c/file4", &cx);
3004
3005 // Wait for the initial scan.
3006 cx.read(|cx| tree.read(cx).as_local().unwrap().scan_complete())
3007 .await;
3008
3009 // Create a remote copy of this worktree.
3010 let initial_snapshot = tree.read_with(&cx, |tree, _| tree.snapshot());
3011 let worktree_id = 1;
3012 let share_request = tree.update(&mut cx, |tree, cx| {
3013 tree.as_local().unwrap().share_request(cx)
3014 });
3015 let open_worktree = server.receive::<proto::OpenWorktree>().await.unwrap();
3016 server
3017 .respond(
3018 open_worktree.receipt(),
3019 proto::OpenWorktreeResponse { worktree_id: 1 },
3020 )
3021 .await;
3022
3023 let remote = Worktree::remote(
3024 proto::JoinWorktreeResponse {
3025 worktree: share_request.await.unwrap().worktree,
3026 replica_id: 1,
3027 peers: Vec::new(),
3028 },
3029 Client::new(),
3030 Default::default(),
3031 &mut cx.to_async(),
3032 )
3033 .await
3034 .unwrap();
3035
3036 cx.read(|cx| {
3037 assert!(!buffer2.read(cx).is_dirty());
3038 assert!(!buffer3.read(cx).is_dirty());
3039 assert!(!buffer4.read(cx).is_dirty());
3040 assert!(!buffer5.read(cx).is_dirty());
3041 });
3042
3043 // Rename and delete files and directories.
3044 tree.flush_fs_events(&cx).await;
3045 std::fs::rename(dir.path().join("a/file3"), dir.path().join("b/c/file3")).unwrap();
3046 std::fs::remove_file(dir.path().join("b/c/file5")).unwrap();
3047 std::fs::rename(dir.path().join("b/c"), dir.path().join("d")).unwrap();
3048 std::fs::rename(dir.path().join("a/file2"), dir.path().join("a/file2.new")).unwrap();
3049 tree.flush_fs_events(&cx).await;
3050
3051 let expected_paths = vec![
3052 "a",
3053 "a/file1",
3054 "a/file2.new",
3055 "b",
3056 "d",
3057 "d/file3",
3058 "d/file4",
3059 ];
3060
3061 cx.read(|app| {
3062 assert_eq!(
3063 tree.read(app)
3064 .paths()
3065 .map(|p| p.to_str().unwrap())
3066 .collect::<Vec<_>>(),
3067 expected_paths
3068 );
3069
3070 assert_eq!(id_for_path("a/file2.new", &cx), file2_id);
3071 assert_eq!(id_for_path("d/file3", &cx), file3_id);
3072 assert_eq!(id_for_path("d/file4", &cx), file4_id);
3073
3074 assert_eq!(
3075 buffer2.read(app).file().unwrap().path().as_ref(),
3076 Path::new("a/file2.new")
3077 );
3078 assert_eq!(
3079 buffer3.read(app).file().unwrap().path().as_ref(),
3080 Path::new("d/file3")
3081 );
3082 assert_eq!(
3083 buffer4.read(app).file().unwrap().path().as_ref(),
3084 Path::new("d/file4")
3085 );
3086 assert_eq!(
3087 buffer5.read(app).file().unwrap().path().as_ref(),
3088 Path::new("b/c/file5")
3089 );
3090
3091 assert!(!buffer2.read(app).file().unwrap().is_deleted());
3092 assert!(!buffer3.read(app).file().unwrap().is_deleted());
3093 assert!(!buffer4.read(app).file().unwrap().is_deleted());
3094 assert!(buffer5.read(app).file().unwrap().is_deleted());
3095 });
3096
3097 // Update the remote worktree. Check that it becomes consistent with the
3098 // local worktree.
3099 remote.update(&mut cx, |remote, cx| {
3100 let update_message =
3101 tree.read(cx)
3102 .snapshot()
3103 .build_update(&initial_snapshot, worktree_id, true);
3104 remote
3105 .as_remote_mut()
3106 .unwrap()
3107 .snapshot
3108 .apply_update(update_message)
3109 .unwrap();
3110
3111 assert_eq!(
3112 remote
3113 .paths()
3114 .map(|p| p.to_str().unwrap())
3115 .collect::<Vec<_>>(),
3116 expected_paths
3117 );
3118 });
3119 }
3120
3121 #[gpui::test]
3122 async fn test_rescan_with_gitignore(cx: gpui::TestAppContext) {
3123 let dir = temp_tree(json!({
3124 ".git": {},
3125 ".gitignore": "ignored-dir\n",
3126 "tracked-dir": {
3127 "tracked-file1": "tracked contents",
3128 },
3129 "ignored-dir": {
3130 "ignored-file1": "ignored contents",
3131 }
3132 }));
3133
3134 let tree = Worktree::open_local(
3135 Client::new(),
3136 dir.path(),
3137 Arc::new(RealFs),
3138 Default::default(),
3139 &mut cx.to_async(),
3140 )
3141 .await
3142 .unwrap();
3143 cx.read(|cx| tree.read(cx).as_local().unwrap().scan_complete())
3144 .await;
3145 tree.flush_fs_events(&cx).await;
3146 cx.read(|cx| {
3147 let tree = tree.read(cx);
3148 let tracked = tree.entry_for_path("tracked-dir/tracked-file1").unwrap();
3149 let ignored = tree.entry_for_path("ignored-dir/ignored-file1").unwrap();
3150 assert_eq!(tracked.is_ignored, false);
3151 assert_eq!(ignored.is_ignored, true);
3152 });
3153
3154 std::fs::write(dir.path().join("tracked-dir/tracked-file2"), "").unwrap();
3155 std::fs::write(dir.path().join("ignored-dir/ignored-file2"), "").unwrap();
3156 tree.flush_fs_events(&cx).await;
3157 cx.read(|cx| {
3158 let tree = tree.read(cx);
3159 let dot_git = tree.entry_for_path(".git").unwrap();
3160 let tracked = tree.entry_for_path("tracked-dir/tracked-file2").unwrap();
3161 let ignored = tree.entry_for_path("ignored-dir/ignored-file2").unwrap();
3162 assert_eq!(tracked.is_ignored, false);
3163 assert_eq!(ignored.is_ignored, true);
3164 assert_eq!(dot_git.is_ignored, true);
3165 });
3166 }
3167
3168 #[gpui::test]
3169 async fn test_open_and_share_worktree(mut cx: gpui::TestAppContext) {
3170 let user_id = 100;
3171 let mut client = Client::new();
3172 let server = FakeServer::for_client(user_id, &mut client, &cx).await;
3173
3174 let fs = Arc::new(FakeFs::new());
3175 fs.insert_tree(
3176 "/path",
3177 json!({
3178 "to": {
3179 "the-dir": {
3180 ".zed.toml": r#"collaborators = ["friend-1", "friend-2"]"#,
3181 "a.txt": "a-contents",
3182 },
3183 },
3184 }),
3185 )
3186 .await;
3187
3188 let worktree = Worktree::open_local(
3189 client.clone(),
3190 "/path/to/the-dir".as_ref(),
3191 fs,
3192 Default::default(),
3193 &mut cx.to_async(),
3194 )
3195 .await
3196 .unwrap();
3197
3198 {
3199 let cx = cx.to_async();
3200 client.authenticate_and_connect(&cx).await.unwrap();
3201 }
3202
3203 let open_worktree = server.receive::<proto::OpenWorktree>().await.unwrap();
3204 assert_eq!(
3205 open_worktree.payload,
3206 proto::OpenWorktree {
3207 root_name: "the-dir".to_string(),
3208 collaborator_logins: vec!["friend-1".to_string(), "friend-2".to_string()],
3209 }
3210 );
3211
3212 server
3213 .respond(
3214 open_worktree.receipt(),
3215 proto::OpenWorktreeResponse { worktree_id: 5 },
3216 )
3217 .await;
3218 let remote_id = worktree
3219 .update(&mut cx, |tree, _| tree.as_local().unwrap().next_remote_id())
3220 .await;
3221 assert_eq!(remote_id, Some(5));
3222
3223 cx.update(move |_| drop(worktree));
3224 server.receive::<proto::CloseWorktree>().await.unwrap();
3225 }
3226
3227 #[gpui::test]
3228 async fn test_buffer_is_dirty(mut cx: gpui::TestAppContext) {
3229 use std::fs;
3230
3231 let dir = temp_tree(json!({
3232 "file1": "abc",
3233 "file2": "def",
3234 "file3": "ghi",
3235 }));
3236 let tree = Worktree::open_local(
3237 Client::new(),
3238 dir.path(),
3239 Arc::new(RealFs),
3240 Default::default(),
3241 &mut cx.to_async(),
3242 )
3243 .await
3244 .unwrap();
3245 tree.flush_fs_events(&cx).await;
3246 cx.read(|cx| tree.read(cx).as_local().unwrap().scan_complete())
3247 .await;
3248
3249 let buffer1 = tree
3250 .update(&mut cx, |tree, cx| tree.open_buffer("file1", cx))
3251 .await
3252 .unwrap();
3253 let events = Rc::new(RefCell::new(Vec::new()));
3254
3255 // initially, the buffer isn't dirty.
3256 buffer1.update(&mut cx, |buffer, cx| {
3257 cx.subscribe(&buffer1, {
3258 let events = events.clone();
3259 move |_, _, event, _| events.borrow_mut().push(event.clone())
3260 })
3261 .detach();
3262
3263 assert!(!buffer.is_dirty());
3264 assert!(events.borrow().is_empty());
3265
3266 buffer.edit(vec![1..2], "", cx);
3267 });
3268
3269 // after the first edit, the buffer is dirty, and emits a dirtied event.
3270 buffer1.update(&mut cx, |buffer, cx| {
3271 assert!(buffer.text() == "ac");
3272 assert!(buffer.is_dirty());
3273 assert_eq!(
3274 *events.borrow(),
3275 &[buffer::Event::Edited, buffer::Event::Dirtied]
3276 );
3277 events.borrow_mut().clear();
3278 buffer.did_save(buffer.version(), buffer.file().unwrap().mtime(), None, cx);
3279 });
3280
3281 // after saving, the buffer is not dirty, and emits a saved event.
3282 buffer1.update(&mut cx, |buffer, cx| {
3283 assert!(!buffer.is_dirty());
3284 assert_eq!(*events.borrow(), &[buffer::Event::Saved]);
3285 events.borrow_mut().clear();
3286
3287 buffer.edit(vec![1..1], "B", cx);
3288 buffer.edit(vec![2..2], "D", cx);
3289 });
3290
3291 // after editing again, the buffer is dirty, and emits another dirty event.
3292 buffer1.update(&mut cx, |buffer, cx| {
3293 assert!(buffer.text() == "aBDc");
3294 assert!(buffer.is_dirty());
3295 assert_eq!(
3296 *events.borrow(),
3297 &[
3298 buffer::Event::Edited,
3299 buffer::Event::Dirtied,
3300 buffer::Event::Edited
3301 ],
3302 );
3303 events.borrow_mut().clear();
3304
3305 // TODO - currently, after restoring the buffer to its
3306 // previously-saved state, the is still considered dirty.
3307 buffer.edit(vec![1..3], "", cx);
3308 assert!(buffer.text() == "ac");
3309 assert!(buffer.is_dirty());
3310 });
3311
3312 assert_eq!(*events.borrow(), &[buffer::Event::Edited]);
3313
3314 // When a file is deleted, the buffer is considered dirty.
3315 let events = Rc::new(RefCell::new(Vec::new()));
3316 let buffer2 = tree
3317 .update(&mut cx, |tree, cx| tree.open_buffer("file2", cx))
3318 .await
3319 .unwrap();
3320 buffer2.update(&mut cx, |_, cx| {
3321 cx.subscribe(&buffer2, {
3322 let events = events.clone();
3323 move |_, _, event, _| events.borrow_mut().push(event.clone())
3324 })
3325 .detach();
3326 });
3327
3328 fs::remove_file(dir.path().join("file2")).unwrap();
3329 buffer2.condition(&cx, |b, _| b.is_dirty()).await;
3330 assert_eq!(
3331 *events.borrow(),
3332 &[buffer::Event::Dirtied, buffer::Event::FileHandleChanged]
3333 );
3334
3335 // When a file is already dirty when deleted, we don't emit a Dirtied event.
3336 let events = Rc::new(RefCell::new(Vec::new()));
3337 let buffer3 = tree
3338 .update(&mut cx, |tree, cx| tree.open_buffer("file3", cx))
3339 .await
3340 .unwrap();
3341 buffer3.update(&mut cx, |_, cx| {
3342 cx.subscribe(&buffer3, {
3343 let events = events.clone();
3344 move |_, _, event, _| events.borrow_mut().push(event.clone())
3345 })
3346 .detach();
3347 });
3348
3349 tree.flush_fs_events(&cx).await;
3350 buffer3.update(&mut cx, |buffer, cx| {
3351 buffer.edit(Some(0..0), "x", cx);
3352 });
3353 events.borrow_mut().clear();
3354 fs::remove_file(dir.path().join("file3")).unwrap();
3355 buffer3
3356 .condition(&cx, |_, _| !events.borrow().is_empty())
3357 .await;
3358 assert_eq!(*events.borrow(), &[buffer::Event::FileHandleChanged]);
3359 cx.read(|cx| assert!(buffer3.read(cx).is_dirty()));
3360 }
3361
3362 #[gpui::test]
3363 async fn test_buffer_file_changes_on_disk(mut cx: gpui::TestAppContext) {
3364 use buffer::{Point, Selection, SelectionGoal, ToPoint};
3365 use std::fs;
3366
3367 let initial_contents = "aaa\nbbbbb\nc\n";
3368 let dir = temp_tree(json!({ "the-file": initial_contents }));
3369 let tree = Worktree::open_local(
3370 Client::new(),
3371 dir.path(),
3372 Arc::new(RealFs),
3373 Default::default(),
3374 &mut cx.to_async(),
3375 )
3376 .await
3377 .unwrap();
3378 cx.read(|cx| tree.read(cx).as_local().unwrap().scan_complete())
3379 .await;
3380
3381 let abs_path = dir.path().join("the-file");
3382 let buffer = tree
3383 .update(&mut cx, |tree, cx| {
3384 tree.open_buffer(Path::new("the-file"), cx)
3385 })
3386 .await
3387 .unwrap();
3388
3389 // Add a cursor at the start of each row.
3390 let selection_set_id = buffer.update(&mut cx, |buffer, cx| {
3391 assert!(!buffer.is_dirty());
3392 buffer.add_selection_set(
3393 (0..3)
3394 .map(|row| {
3395 let anchor = buffer.anchor_at(Point::new(row, 0), Bias::Right);
3396 Selection {
3397 id: row as usize,
3398 start: anchor.clone(),
3399 end: anchor,
3400 reversed: false,
3401 goal: SelectionGoal::None,
3402 }
3403 })
3404 .collect::<Vec<_>>(),
3405 cx,
3406 )
3407 });
3408
3409 // Change the file on disk, adding two new lines of text, and removing
3410 // one line.
3411 buffer.read_with(&cx, |buffer, _| {
3412 assert!(!buffer.is_dirty());
3413 assert!(!buffer.has_conflict());
3414 });
3415 let new_contents = "AAAA\naaa\nBB\nbbbbb\n";
3416 fs::write(&abs_path, new_contents).unwrap();
3417
3418 // Because the buffer was not modified, it is reloaded from disk. Its
3419 // contents are edited according to the diff between the old and new
3420 // file contents.
3421 buffer
3422 .condition(&cx, |buffer, _| buffer.text() != initial_contents)
3423 .await;
3424
3425 buffer.update(&mut cx, |buffer, _| {
3426 assert_eq!(buffer.text(), new_contents);
3427 assert!(!buffer.is_dirty());
3428 assert!(!buffer.has_conflict());
3429
3430 let set = buffer.selection_set(selection_set_id).unwrap();
3431 let cursor_positions = set
3432 .selections
3433 .iter()
3434 .map(|selection| {
3435 assert_eq!(selection.start, selection.end);
3436 selection.start.to_point(&*buffer)
3437 })
3438 .collect::<Vec<_>>();
3439 assert_eq!(
3440 cursor_positions,
3441 &[Point::new(1, 0), Point::new(3, 0), Point::new(4, 0),]
3442 );
3443 });
3444
3445 // Modify the buffer
3446 buffer.update(&mut cx, |buffer, cx| {
3447 buffer.edit(vec![0..0], " ", cx);
3448 assert!(buffer.is_dirty());
3449 });
3450
3451 // Change the file on disk again, adding blank lines to the beginning.
3452 fs::write(&abs_path, "\n\n\nAAAA\naaa\nBB\nbbbbb\n").unwrap();
3453
3454 // Becaues the buffer is modified, it doesn't reload from disk, but is
3455 // marked as having a conflict.
3456 buffer
3457 .condition(&cx, |buffer, _| buffer.has_conflict())
3458 .await;
3459 }
3460
3461 #[gpui::test(iterations = 100)]
3462 fn test_random(mut rng: StdRng) {
3463 let operations = env::var("OPERATIONS")
3464 .map(|o| o.parse().unwrap())
3465 .unwrap_or(40);
3466 let initial_entries = env::var("INITIAL_ENTRIES")
3467 .map(|o| o.parse().unwrap())
3468 .unwrap_or(20);
3469
3470 let root_dir = tempdir::TempDir::new("worktree-test").unwrap();
3471 for _ in 0..initial_entries {
3472 randomly_mutate_tree(root_dir.path(), 1.0, &mut rng).unwrap();
3473 }
3474 log::info!("Generated initial tree");
3475
3476 let (notify_tx, _notify_rx) = smol::channel::unbounded();
3477 let fs = Arc::new(RealFs);
3478 let next_entry_id = Arc::new(AtomicUsize::new(0));
3479 let mut initial_snapshot = Snapshot {
3480 id: 0,
3481 scan_id: 0,
3482 abs_path: root_dir.path().into(),
3483 entries_by_path: Default::default(),
3484 entries_by_id: Default::default(),
3485 removed_entry_ids: Default::default(),
3486 ignores: Default::default(),
3487 root_name: Default::default(),
3488 root_char_bag: Default::default(),
3489 next_entry_id: next_entry_id.clone(),
3490 };
3491 initial_snapshot.insert_entry(
3492 Entry::new(
3493 Path::new("").into(),
3494 &smol::block_on(fs.metadata(root_dir.path()))
3495 .unwrap()
3496 .unwrap(),
3497 &next_entry_id,
3498 Default::default(),
3499 ),
3500 fs.as_ref(),
3501 );
3502 let mut scanner = BackgroundScanner::new(
3503 Arc::new(Mutex::new(initial_snapshot.clone())),
3504 notify_tx,
3505 fs.clone(),
3506 Arc::new(gpui::executor::Background::new()),
3507 );
3508 smol::block_on(scanner.scan_dirs()).unwrap();
3509 scanner.snapshot().check_invariants();
3510
3511 let mut events = Vec::new();
3512 let mut snapshots = Vec::new();
3513 let mut mutations_len = operations;
3514 while mutations_len > 1 {
3515 if !events.is_empty() && rng.gen_bool(0.4) {
3516 let len = rng.gen_range(0..=events.len());
3517 let to_deliver = events.drain(0..len).collect::<Vec<_>>();
3518 log::info!("Delivering events: {:#?}", to_deliver);
3519 smol::block_on(scanner.process_events(to_deliver));
3520 scanner.snapshot().check_invariants();
3521 } else {
3522 events.extend(randomly_mutate_tree(root_dir.path(), 0.6, &mut rng).unwrap());
3523 mutations_len -= 1;
3524 }
3525
3526 if rng.gen_bool(0.2) {
3527 snapshots.push(scanner.snapshot());
3528 }
3529 }
3530 log::info!("Quiescing: {:#?}", events);
3531 smol::block_on(scanner.process_events(events));
3532 scanner.snapshot().check_invariants();
3533
3534 let (notify_tx, _notify_rx) = smol::channel::unbounded();
3535 let mut new_scanner = BackgroundScanner::new(
3536 Arc::new(Mutex::new(initial_snapshot)),
3537 notify_tx,
3538 scanner.fs.clone(),
3539 scanner.executor.clone(),
3540 );
3541 smol::block_on(new_scanner.scan_dirs()).unwrap();
3542 assert_eq!(
3543 scanner.snapshot().to_vec(true),
3544 new_scanner.snapshot().to_vec(true)
3545 );
3546
3547 for mut prev_snapshot in snapshots {
3548 let include_ignored = rng.gen::<bool>();
3549 if !include_ignored {
3550 let mut entries_by_path_edits = Vec::new();
3551 let mut entries_by_id_edits = Vec::new();
3552 for entry in prev_snapshot
3553 .entries_by_id
3554 .cursor::<()>()
3555 .filter(|e| e.is_ignored)
3556 {
3557 entries_by_path_edits.push(Edit::Remove(PathKey(entry.path.clone())));
3558 entries_by_id_edits.push(Edit::Remove(entry.id));
3559 }
3560
3561 prev_snapshot
3562 .entries_by_path
3563 .edit(entries_by_path_edits, &());
3564 prev_snapshot.entries_by_id.edit(entries_by_id_edits, &());
3565 }
3566
3567 let update = scanner
3568 .snapshot()
3569 .build_update(&prev_snapshot, 0, include_ignored);
3570 prev_snapshot.apply_update(update).unwrap();
3571 assert_eq!(
3572 prev_snapshot.to_vec(true),
3573 scanner.snapshot().to_vec(include_ignored)
3574 );
3575 }
3576 }
3577
3578 fn randomly_mutate_tree(
3579 root_path: &Path,
3580 insertion_probability: f64,
3581 rng: &mut impl Rng,
3582 ) -> Result<Vec<fsevent::Event>> {
3583 let root_path = root_path.canonicalize().unwrap();
3584 let (dirs, files) = read_dir_recursive(root_path.clone());
3585
3586 let mut events = Vec::new();
3587 let mut record_event = |path: PathBuf| {
3588 events.push(fsevent::Event {
3589 event_id: SystemTime::now()
3590 .duration_since(UNIX_EPOCH)
3591 .unwrap()
3592 .as_secs(),
3593 flags: fsevent::StreamFlags::empty(),
3594 path,
3595 });
3596 };
3597
3598 if (files.is_empty() && dirs.len() == 1) || rng.gen_bool(insertion_probability) {
3599 let path = dirs.choose(rng).unwrap();
3600 let new_path = path.join(gen_name(rng));
3601
3602 if rng.gen() {
3603 log::info!("Creating dir {:?}", new_path.strip_prefix(root_path)?);
3604 std::fs::create_dir(&new_path)?;
3605 } else {
3606 log::info!("Creating file {:?}", new_path.strip_prefix(root_path)?);
3607 std::fs::write(&new_path, "")?;
3608 }
3609 record_event(new_path);
3610 } else if rng.gen_bool(0.05) {
3611 let ignore_dir_path = dirs.choose(rng).unwrap();
3612 let ignore_path = ignore_dir_path.join(&*GITIGNORE);
3613
3614 let (subdirs, subfiles) = read_dir_recursive(ignore_dir_path.clone());
3615 let files_to_ignore = {
3616 let len = rng.gen_range(0..=subfiles.len());
3617 subfiles.choose_multiple(rng, len)
3618 };
3619 let dirs_to_ignore = {
3620 let len = rng.gen_range(0..subdirs.len());
3621 subdirs.choose_multiple(rng, len)
3622 };
3623
3624 let mut ignore_contents = String::new();
3625 for path_to_ignore in files_to_ignore.chain(dirs_to_ignore) {
3626 write!(
3627 ignore_contents,
3628 "{}\n",
3629 path_to_ignore
3630 .strip_prefix(&ignore_dir_path)?
3631 .to_str()
3632 .unwrap()
3633 )
3634 .unwrap();
3635 }
3636 log::info!(
3637 "Creating {:?} with contents:\n{}",
3638 ignore_path.strip_prefix(&root_path)?,
3639 ignore_contents
3640 );
3641 std::fs::write(&ignore_path, ignore_contents).unwrap();
3642 record_event(ignore_path);
3643 } else {
3644 let old_path = {
3645 let file_path = files.choose(rng);
3646 let dir_path = dirs[1..].choose(rng);
3647 file_path.into_iter().chain(dir_path).choose(rng).unwrap()
3648 };
3649
3650 let is_rename = rng.gen();
3651 if is_rename {
3652 let new_path_parent = dirs
3653 .iter()
3654 .filter(|d| !d.starts_with(old_path))
3655 .choose(rng)
3656 .unwrap();
3657
3658 let overwrite_existing_dir =
3659 !old_path.starts_with(&new_path_parent) && rng.gen_bool(0.3);
3660 let new_path = if overwrite_existing_dir {
3661 std::fs::remove_dir_all(&new_path_parent).ok();
3662 new_path_parent.to_path_buf()
3663 } else {
3664 new_path_parent.join(gen_name(rng))
3665 };
3666
3667 log::info!(
3668 "Renaming {:?} to {}{:?}",
3669 old_path.strip_prefix(&root_path)?,
3670 if overwrite_existing_dir {
3671 "overwrite "
3672 } else {
3673 ""
3674 },
3675 new_path.strip_prefix(&root_path)?
3676 );
3677 std::fs::rename(&old_path, &new_path)?;
3678 record_event(old_path.clone());
3679 record_event(new_path);
3680 } else if old_path.is_dir() {
3681 let (dirs, files) = read_dir_recursive(old_path.clone());
3682
3683 log::info!("Deleting dir {:?}", old_path.strip_prefix(&root_path)?);
3684 std::fs::remove_dir_all(&old_path).unwrap();
3685 for file in files {
3686 record_event(file);
3687 }
3688 for dir in dirs {
3689 record_event(dir);
3690 }
3691 } else {
3692 log::info!("Deleting file {:?}", old_path.strip_prefix(&root_path)?);
3693 std::fs::remove_file(old_path).unwrap();
3694 record_event(old_path.clone());
3695 }
3696 }
3697
3698 Ok(events)
3699 }
3700
3701 fn read_dir_recursive(path: PathBuf) -> (Vec<PathBuf>, Vec<PathBuf>) {
3702 let child_entries = std::fs::read_dir(&path).unwrap();
3703 let mut dirs = vec![path];
3704 let mut files = Vec::new();
3705 for child_entry in child_entries {
3706 let child_path = child_entry.unwrap().path();
3707 if child_path.is_dir() {
3708 let (child_dirs, child_files) = read_dir_recursive(child_path);
3709 dirs.extend(child_dirs);
3710 files.extend(child_files);
3711 } else {
3712 files.push(child_path);
3713 }
3714 }
3715 (dirs, files)
3716 }
3717
3718 fn gen_name(rng: &mut impl Rng) -> String {
3719 (0..6)
3720 .map(|_| rng.sample(rand::distributions::Alphanumeric))
3721 .map(char::from)
3722 .collect()
3723 }
3724
3725 impl Snapshot {
3726 fn check_invariants(&self) {
3727 let mut files = self.files(true, 0);
3728 let mut visible_files = self.files(false, 0);
3729 for entry in self.entries_by_path.cursor::<()>() {
3730 if entry.is_file() {
3731 assert_eq!(files.next().unwrap().inode, entry.inode);
3732 if !entry.is_ignored {
3733 assert_eq!(visible_files.next().unwrap().inode, entry.inode);
3734 }
3735 }
3736 }
3737 assert!(files.next().is_none());
3738 assert!(visible_files.next().is_none());
3739
3740 let mut bfs_paths = Vec::new();
3741 let mut stack = vec![Path::new("")];
3742 while let Some(path) = stack.pop() {
3743 bfs_paths.push(path);
3744 let ix = stack.len();
3745 for child_entry in self.child_entries(path) {
3746 stack.insert(ix, &child_entry.path);
3747 }
3748 }
3749
3750 let dfs_paths = self
3751 .entries_by_path
3752 .cursor::<()>()
3753 .map(|e| e.path.as_ref())
3754 .collect::<Vec<_>>();
3755 assert_eq!(bfs_paths, dfs_paths);
3756
3757 for (ignore_parent_path, _) in &self.ignores {
3758 assert!(self.entry_for_path(ignore_parent_path).is_some());
3759 assert!(self
3760 .entry_for_path(ignore_parent_path.join(&*GITIGNORE))
3761 .is_some());
3762 }
3763 }
3764
3765 fn to_vec(&self, include_ignored: bool) -> Vec<(&Path, u64, bool)> {
3766 let mut paths = Vec::new();
3767 for entry in self.entries_by_path.cursor::<()>() {
3768 if include_ignored || !entry.is_ignored {
3769 paths.push((entry.path.as_ref(), entry.inode, entry.is_ignored));
3770 }
3771 }
3772 paths.sort_by(|a, b| a.0.cmp(&b.0));
3773 paths
3774 }
3775 }
3776}