Yehowshua/ZIm
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Merge branch 'main' into user-timeline

Browse source
This commit is contained in:
Antonio Scandurra 2022-07-04 09:23:16 +02:00
commit d3b9eca791
50 changed files with 3076 additions and 1109 deletions
Download patch file Download diff file Expand all files Collapse all files

11
crates/collab/src/db.rs
View file

@ -2282,7 +2282,7 @@ pub mod tests {
Self {
background,
users: Default::default(),
next_user_id: Mutex::new(1),
next_user_id: Mutex::new(0),
projects: Default::default(),
worktree_extensions: Default::default(),
next_project_id: Mutex::new(1),
@ -2346,6 +2346,7 @@ pub mod tests {
}
async fn get_user_by_id(&self, id: UserId) -> Result<Option<User>> {
self.background.simulate_random_delay().await;
Ok(self.get_users_by_ids(vec![id]).await?.into_iter().next())
}
@ -2360,6 +2361,7 @@ pub mod tests {
}
async fn get_user_by_github_login(&self, github_login: &str) -> Result<Option<User>> {
self.background.simulate_random_delay().await;
Ok(self
.users
.lock()
@ -2393,6 +2395,7 @@ pub mod tests {
}
async fn get_invite_code_for_user(&self, _id: UserId) -> Result<Option<(String, u32)>> {
self.background.simulate_random_delay().await;
Ok(None)
}
@ -2430,6 +2433,7 @@ pub mod tests {
}
async fn unregister_project(&self, project_id: ProjectId) -> Result<()> {
self.background.simulate_random_delay().await;
self.projects
.lock()
.get_mut(&project_id)
@ -2543,6 +2547,7 @@ pub mod tests {
requester_id: UserId,
responder_id: UserId,
) -> Result<()> {
self.background.simulate_random_delay().await;
let mut contacts = self.contacts.lock();
for contact in contacts.iter_mut() {
if contact.requester_id == requester_id && contact.responder_id == responder_id {
@ -2572,6 +2577,7 @@ pub mod tests {
}
async fn remove_contact(&self, requester_id: UserId, responder_id: UserId) -> Result<()> {
self.background.simulate_random_delay().await;
self.contacts.lock().retain(|contact| {
!(contact.requester_id == requester_id && contact.responder_id == responder_id)
});
@ -2583,6 +2589,7 @@ pub mod tests {
user_id: UserId,
contact_user_id: UserId,
) -> Result<()> {
self.background.simulate_random_delay().await;
let mut contacts = self.contacts.lock();
for contact in contacts.iter_mut() {
if contact.requester_id == contact_user_id
@ -2609,6 +2616,7 @@ pub mod tests {
requester_id: UserId,
accept: bool,
) -> Result<()> {
self.background.simulate_random_delay().await;
let mut contacts = self.contacts.lock();
for (ix, contact) in contacts.iter_mut().enumerate() {
if contact.requester_id == requester_id && contact.responder_id == responder_id {
@ -2804,6 +2812,7 @@ pub mod tests {
count: usize,
before_id: Option<MessageId>,
) -> Result<Vec<ChannelMessage>> {
self.background.simulate_random_delay().await;
let mut messages = self
.channel_messages
.lock()

39
crates/collab/src/integration_tests.rs
View file

@ -50,7 +50,6 @@ use std::{
time::Duration,
};
use theme::ThemeRegistry;
use tokio::sync::RwLockReadGuard;
use workspace::{Item, SplitDirection, ToggleFollow, Workspace};
#[ctor::ctor]
@ -596,7 +595,7 @@ async fn test_offline_projects(
deterministic.run_until_parked();
assert!(server
.store
.read()
.lock()
.await
.project_metadata_for_user(user_a)
.is_empty());
@ -630,7 +629,7 @@ async fn test_offline_projects(
cx_a.foreground().advance_clock(rpc::RECEIVE_TIMEOUT);
assert!(server
.store
.read()
.lock()
.await
.project_metadata_for_user(user_a)
.is_empty());
@ -1491,7 +1490,7 @@ async fn test_collaborating_with_diagnostics(
// Wait for server to see the diagnostics update.
deterministic.run_until_parked();
{
let store = server.store.read().await;
let store = server.store.lock().await;
let project = store.project(ProjectId::from_proto(project_id)).unwrap();
let worktree = project.worktrees.get(&worktree_id.to_proto()).unwrap();
assert!(!worktree.diagnostic_summaries.is_empty());
@ -1517,6 +1516,7 @@ async fn test_collaborating_with_diagnostics(
// Join project as client C and observe the diagnostics.
let project_c = client_c.build_remote_project(&project_a, cx_a, cx_c).await;
deterministic.run_until_parked();
project_c.read_with(cx_c, |project, cx| {
assert_eq!(
project.diagnostic_summaries(cx).collect::<Vec<_>>(),
@ -3216,7 +3216,7 @@ async fn test_basic_chat(cx_a: &mut TestAppContext, cx_b: &mut TestAppContext) {
assert_eq!(
server
.state()
.store()
.await
.channel(channel_id)
.unwrap()
@ -4470,8 +4470,16 @@ async fn test_random_collaboration(
let mut server = TestServer::start(cx.foreground(), cx.background()).await;
let db = server.app_state.db.clone();
let host_user_id = db.create_user("host", None, false).await.unwrap();
for username in ["guest-1", "guest-2", "guest-3", "guest-4"] {
let mut available_guests = vec![
"guest-1".to_string(),
"guest-2".to_string(),
"guest-3".to_string(),
"guest-4".to_string(),
];
for username in &available_guests {
let guest_user_id = db.create_user(username, None, false).await.unwrap();
assert_eq!(*username, format!("guest-{}", guest_user_id));
server
.app_state
.db
@ -4665,12 +4673,7 @@ async fn test_random_collaboration(
} else {
max_operations
};
let mut available_guests = vec![
"guest-1".to_string(),
"guest-2".to_string(),
"guest-3".to_string(),
"guest-4".to_string(),
];
let mut operations = 0;
while operations < max_operations {
if operations == disconnect_host_at {
@ -4701,7 +4704,7 @@ async fn test_random_collaboration(
.unwrap();
let contacts = server
.store
.read()
.lock()
.await
.build_initial_contacts_update(contacts)
.contacts;
@ -4773,6 +4776,7 @@ async fn test_random_collaboration(
server.disconnect_client(removed_guest_id);
deterministic.advance_clock(RECEIVE_TIMEOUT);
deterministic.start_waiting();
log::info!("Waiting for guest {} to exit...", removed_guest_id);
let (guest, guest_project, mut guest_cx, guest_err) = guest.await;
deterministic.finish_waiting();
server.allow_connections();
@ -4785,7 +4789,7 @@ async fn test_random_collaboration(
let contacts = server.app_state.db.get_contacts(*user_id).await.unwrap();
let contacts = server
.store
.read()
.lock()
.await
.build_initial_contacts_update(contacts)
.contacts;
@ -4989,6 +4993,7 @@ impl TestServer {
Arc::get_mut(&mut client)
.unwrap()
.set_id(user_id.0 as usize)
.override_authenticate(move |cx| {
cx.spawn(|_| async move {
let access_token = "the-token".to_string();
@ -5116,10 +5121,6 @@ impl TestServer {
})
}
async fn state<'a>(&'a self) -> RwLockReadGuard<'a, Store> {
self.server.store.read().await
}
async fn condition<F>(&mut self, mut predicate: F)
where
F: FnMut(&Store) -> bool,
@ -5128,7 +5129,7 @@ impl TestServer {
self.foreground.parking_forbidden(),
"you must call forbid_parking to use server conditions so we don't block indefinitely"
);
while !(predicate)(&*self.server.store.read().await) {
while !(predicate)(&*self.server.store.lock().await) {
self.foreground.start_waiting();
self.notifications.next().await;
self.foreground.finish_waiting();

169
crates/collab/src/rpc.rs
View file

@ -51,7 +51,7 @@ use std::{
};
use time::OffsetDateTime;
use tokio::{
sync::{RwLock, RwLockReadGuard, RwLockWriteGuard},
sync::{Mutex, MutexGuard},
time::Sleep,
};
use tower::ServiceBuilder;
@ -97,7 +97,7 @@ impl<R: RequestMessage> Response<R> {
pub struct Server {
peer: Arc<Peer>,
pub(crate) store: RwLock<Store>,
pub(crate) store: Mutex<Store>,
app_state: Arc<AppState>,
handlers: HashMap<TypeId, MessageHandler>,
notifications: Option<mpsc::UnboundedSender<()>>,
@ -115,13 +115,8 @@ pub struct RealExecutor;
const MESSAGE_COUNT_PER_PAGE: usize = 100;
const MAX_MESSAGE_LEN: usize = 1024;
struct StoreReadGuard<'a> {
guard: RwLockReadGuard<'a, Store>,
_not_send: PhantomData<Rc<()>>,
}
struct StoreWriteGuard<'a> {
guard: RwLockWriteGuard<'a, Store>,
pub(crate) struct StoreGuard<'a> {
guard: MutexGuard<'a, Store>,
_not_send: PhantomData<Rc<()>>,
}
@ -129,7 +124,7 @@ struct StoreWriteGuard<'a> {
pub struct ServerSnapshot<'a> {
peer: &'a Peer,
#[serde(serialize_with = "serialize_deref")]
store: RwLockReadGuard<'a, Store>,
store: StoreGuard<'a>,
}
pub fn serialize_deref<S, T, U>(value: &T, serializer: S) -> Result<S::Ok, S::Error>
@ -385,7 +380,7 @@ impl Server {
).await?;
{
let mut store = this.store_mut().await;
let mut store = this.store().await;
store.add_connection(connection_id, user_id, user.admin);
this.peer.send(connection_id, store.build_initial_contacts_update(contacts))?;
@ -472,7 +467,7 @@ impl Server {
let mut projects_to_unregister = Vec::new();
let removed_user_id;
{
let mut store = self.store_mut().await;
let mut store = self.store().await;
let removed_connection = store.remove_connection(connection_id)?;
for (project_id, project) in removed_connection.hosted_projects {
@ -606,7 +601,7 @@ impl Server {
.await
.user_id_for_connection(request.sender_id)?;
let project_id = self.app_state.db.register_project(user_id).await?;
self.store_mut().await.register_project(
self.store().await.register_project(
request.sender_id,
project_id,
request.payload.online,
@ -626,7 +621,7 @@ impl Server {
) -> Result<()> {
let project_id = ProjectId::from_proto(request.payload.project_id);
let (user_id, project) = {
let mut state = self.store_mut().await;
let mut state = self.store().await;
let project = state.unregister_project(project_id, request.sender_id)?;
(state.user_id_for_connection(request.sender_id)?, project)
};
@ -728,7 +723,7 @@ impl Server {
return Err(anyhow!("no such project"))?;
}
self.store_mut().await.request_join_project(
self.store().await.request_join_project(
guest_user_id,
project_id,
response.into_receipt(),
@ -750,7 +745,7 @@ impl Server {
let host_user_id;
{
let mut state = self.store_mut().await;
let mut state = self.store().await;
let project_id = ProjectId::from_proto(request.payload.project_id);
let project = state.project(project_id)?;
if project.host_connection_id != request.sender_id {
@ -794,20 +789,10 @@ impl Server {
let worktrees = project
.worktrees
.iter()
.filter_map(|(id, shared_worktree)| {
let worktree = project.worktrees.get(&id)?;
Some(proto::Worktree {
id: *id,
root_name: worktree.root_name.clone(),
entries: shared_worktree.entries.values().cloned().collect(),
diagnostic_summaries: shared_worktree
.diagnostic_summaries
.values()
.cloned()
.collect(),
visible: worktree.visible,
scan_id: shared_worktree.scan_id,
})
.map(|(id, worktree)| proto::WorktreeMetadata {
id: *id,
root_name: worktree.root_name.clone(),
visible: worktree.visible,
})
.collect::<Vec<_>>();
@ -843,14 +828,15 @@ impl Server {
}
}
for (receipt, replica_id) in receipts_with_replica_ids {
// First, we send the metadata associated with each worktree.
for (receipt, replica_id) in &receipts_with_replica_ids {
self.peer.respond(
receipt,
receipt.clone(),
proto::JoinProjectResponse {
variant: Some(proto::join_project_response::Variant::Accept(
proto::join_project_response::Accept {
worktrees: worktrees.clone(),
replica_id: replica_id as u32,
replica_id: *replica_id as u32,
collaborators: collaborators.clone(),
language_servers: project.language_servers.clone(),
},
@ -858,6 +844,43 @@ impl Server {
},
)?;
}
for (worktree_id, worktree) in &project.worktrees {
#[cfg(any(test, feature = "test-support"))]
const MAX_CHUNK_SIZE: usize = 2;
#[cfg(not(any(test, feature = "test-support")))]
const MAX_CHUNK_SIZE: usize = 256;
// Stream this worktree's entries.
let message = proto::UpdateWorktree {
project_id: project_id.to_proto(),
worktree_id: *worktree_id,
root_name: worktree.root_name.clone(),
updated_entries: worktree.entries.values().cloned().collect(),
removed_entries: Default::default(),
scan_id: worktree.scan_id,
is_last_update: worktree.is_complete,
};
for update in proto::split_worktree_update(message, MAX_CHUNK_SIZE) {
for (receipt, _) in &receipts_with_replica_ids {
self.peer.send(receipt.sender_id, update.clone())?;
}
}
// Stream this worktree's diagnostics.
for summary in worktree.diagnostic_summaries.values() {
for (receipt, _) in &receipts_with_replica_ids {
self.peer.send(
receipt.sender_id,
proto::UpdateDiagnosticSummary {
project_id: project_id.to_proto(),
worktree_id: *worktree_id,
summary: Some(summary.clone()),
},
)?;
}
}
}
}
self.update_user_contacts(host_user_id).await?;
@ -872,7 +895,7 @@ impl Server {
let project_id = ProjectId::from_proto(request.payload.project_id);
let project;
{
let mut store = self.store_mut().await;
let mut store = self.store().await;
project = store.leave_project(sender_id, project_id)?;
tracing::info!(
%project_id,
@ -923,7 +946,7 @@ impl Server {
let project_id = ProjectId::from_proto(request.payload.project_id);
let user_id;
{
let mut state = self.store_mut().await;
let mut state = self.store().await;
user_id = state.user_id_for_connection(request.sender_id)?;
let guest_connection_ids = state
.read_project(project_id, request.sender_id)?
@ -983,7 +1006,7 @@ impl Server {
self: Arc<Server>,
request: TypedEnvelope<proto::RegisterProjectActivity>,
) -> Result<()> {
self.store_mut().await.register_project_activity(
self.store().await.register_project_activity(
ProjectId::from_proto(request.payload.project_id),
request.sender_id,
)?;
@ -998,7 +1021,7 @@ impl Server {
let project_id = ProjectId::from_proto(request.payload.project_id);
let worktree_id = request.payload.worktree_id;
let (connection_ids, metadata_changed) = {
let mut store = self.store_mut().await;
let mut store = self.store().await;
let (connection_ids, metadata_changed) = store.update_worktree(
request.sender_id,
project_id,
@ -1007,6 +1030,7 @@ impl Server {
&request.payload.removed_entries,
&request.payload.updated_entries,
request.payload.scan_id,
request.payload.is_last_update,
)?;
(connection_ids, metadata_changed)
};
@ -1054,7 +1078,7 @@ impl Server {
.summary
.clone()
.ok_or_else(|| anyhow!("invalid summary"))?;
let receiver_ids = self.store_mut().await.update_diagnostic_summary(
let receiver_ids = self.store().await.update_diagnostic_summary(
ProjectId::from_proto(request.payload.project_id),
request.payload.worktree_id,
request.sender_id,
@ -1072,7 +1096,7 @@ impl Server {
self: Arc<Server>,
request: TypedEnvelope<proto::StartLanguageServer>,
) -> Result<()> {
let receiver_ids = self.store_mut().await.start_language_server(
let receiver_ids = self.store().await.start_language_server(
ProjectId::from_proto(request.payload.project_id),
request.sender_id,
request
@ -1111,20 +1135,23 @@ impl Server {
where
T: EntityMessage + RequestMessage,
{
let project_id = ProjectId::from_proto(request.payload.remote_entity_id());
let host_connection_id = self
.store()
.await
.read_project(
ProjectId::from_proto(request.payload.remote_entity_id()),
request.sender_id,
)?
.read_project(project_id, request.sender_id)?
.host_connection_id;
let payload = self
.peer
.forward_request(request.sender_id, host_connection_id, request.payload)
.await?;
response.send(
self.peer
.forward_request(request.sender_id, host_connection_id, request.payload)
.await?,
)?;
// Ensure project still exists by the time we get the response from the host.
self.store()
.await
.read_project(project_id, request.sender_id)?;
response.send(payload)?;
Ok(())
}
@ -1165,7 +1192,7 @@ impl Server {
) -> Result<()> {
let project_id = ProjectId::from_proto(request.payload.project_id);
let receiver_ids = {
let mut store = self.store_mut().await;
let mut store = self.store().await;
store.register_project_activity(project_id, request.sender_id)?;
store.project_connection_ids(project_id, request.sender_id)?
};
@ -1232,7 +1259,7 @@ impl Server {
let leader_id = ConnectionId(request.payload.leader_id);
let follower_id = request.sender_id;
{
let mut store = self.store_mut().await;
let mut store = self.store().await;
if !store
.project_connection_ids(project_id, follower_id)?
.contains(&leader_id)
@ -1257,7 +1284,7 @@ impl Server {
async fn unfollow(self: Arc<Self>, request: TypedEnvelope<proto::Unfollow>) -> Result<()> {
let project_id = ProjectId::from_proto(request.payload.project_id);
let leader_id = ConnectionId(request.payload.leader_id);
let mut store = self.store_mut().await;
let mut store = self.store().await;
if !store
.project_connection_ids(project_id, request.sender_id)?
.contains(&leader_id)
@ -1275,7 +1302,7 @@ impl Server {
request: TypedEnvelope<proto::UpdateFollowers>,
) -> Result<()> {
let project_id = ProjectId::from_proto(request.payload.project_id);
let mut store = self.store_mut().await;
let mut store = self.store().await;
store.register_project_activity(project_id, request.sender_id)?;
let connection_ids = store.project_connection_ids(project_id, request.sender_id)?;
let leader_id = request
@ -1533,7 +1560,7 @@ impl Server {
Err(anyhow!("access denied"))?;
}
self.store_mut()
self.store()
.await
.join_channel(request.sender_id, channel_id);
let messages = self
@ -1575,7 +1602,7 @@ impl Server {
Err(anyhow!("access denied"))?;
}
self.store_mut()
self.store()
.await
.leave_channel(request.sender_id, channel_id);
@ -1683,25 +1710,13 @@ impl Server {
Ok(())
}
async fn store<'a>(self: &'a Arc<Self>) -> StoreReadGuard<'a> {
pub(crate) async fn store<'a>(&'a self) -> StoreGuard<'a> {
#[cfg(test)]
tokio::task::yield_now().await;
let guard = self.store.read().await;
let guard = self.store.lock().await;
#[cfg(test)]
tokio::task::yield_now().await;
StoreReadGuard {
guard,
_not_send: PhantomData,
}
}
async fn store_mut<'a>(self: &'a Arc<Self>) -> StoreWriteGuard<'a> {
#[cfg(test)]
tokio::task::yield_now().await;
let guard = self.store.write().await;
#[cfg(test)]
tokio::task::yield_now().await;
StoreWriteGuard {
StoreGuard {
guard,
_not_send: PhantomData,
}
@ -1709,13 +1724,13 @@ impl Server {
pub async fn snapshot<'a>(self: &'a Arc<Self>) -> ServerSnapshot<'a> {
ServerSnapshot {
store: self.store.read().await,
store: self.store().await,
peer: &self.peer,
}
}
}
impl<'a> Deref for StoreReadGuard<'a> {
impl<'a> Deref for StoreGuard<'a> {
type Target = Store;
fn deref(&self) -> &Self::Target {
@ -1723,21 +1738,13 @@ impl<'a> Deref for StoreReadGuard<'a> {
}
}
impl<'a> Deref for StoreWriteGuard<'a> {
type Target = Store;
fn deref(&self) -> &Self::Target {
&*self.guard
}
}
impl<'a> DerefMut for StoreWriteGuard<'a> {
impl<'a> DerefMut for StoreGuard<'a> {
fn deref_mut(&mut self) -> &mut Self::Target {
&mut *self.guard
}
}
impl<'a> Drop for StoreWriteGuard<'a> {
impl<'a> Drop for StoreGuard<'a> {
fn drop(&mut self) {
#[cfg(test)]
self.check_invariants();

3
crates/collab/src/rpc/store.rs
View file

@ -56,6 +56,7 @@ pub struct Worktree {
#[serde(skip)]
pub diagnostic_summaries: BTreeMap<PathBuf, proto::DiagnosticSummary>,
pub scan_id: u64,
pub is_complete: bool,
}
#[derive(Default)]
@ -646,6 +647,7 @@ impl Store {
removed_entries: &[u64],
updated_entries: &[proto::Entry],
scan_id: u64,
is_last_update: bool,
) -> Result<(Vec<ConnectionId>, bool)> {
let project = self.write_project(project_id, connection_id)?;
if !project.online {
@ -666,6 +668,7 @@ impl Store {
}
worktree.scan_id = scan_id;
worktree.is_complete = is_last_update;
Ok((connection_ids, metadata_changed))
}