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