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