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