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