zed-industries/zed
1
1
Fork 0
mirror of https://github.com/zed-industries/zed.git synced 2024-12-26 08:02:22 +03:00
Code Issues Packages Projects Releases Wiki Activity

WIP: improve move and link handling around 'root paths', currently very incorrect and in need of a deeper rework

Browse Source
This commit is contained in:
Mikayla 2023-09-10 22:48:25 -07:00
parent cda54b8b5f
commit 7fa68a9aa4
No known key found for this signature in database
6 changed files with 362 additions and 167 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

66
crates/channel/src/channel_store/channel_index.rs
View File

@ -1,10 +1,11 @@
use std::{sync::Arc, ops::Deref};
use std::{ops::Deref, sync::Arc};
use collections::HashMap;
use rpc::proto;
use serde_derive::{Serialize, Deserialize};
use serde_derive::{Deserialize, Serialize};
use crate::{ChannelId, Channel};
use crate::{Channel, ChannelId};
pub type ChannelsById = HashMap<ChannelId, Arc<Channel>>;
@ -21,9 +22,7 @@ impl Deref for ChannelPath {
impl ChannelPath {
pub fn parent_id(&self) -> Option<ChannelId> {
self.0.len().checked_sub(2).map(|i| {
self.0[i]
})
self.0.len().checked_sub(2).map(|i| self.0[i])
}
}
@ -39,7 +38,6 @@ pub struct ChannelIndex {
channels_by_id: ChannelsById,
}
impl ChannelIndex {
pub fn by_id(&self) -> &ChannelsById {
&self.channels_by_id
@ -62,24 +60,53 @@ impl ChannelIndex {
self.paths.iter()
}
/// Remove the given edge from this index. This will not remove the channel
/// and may result in dangling channels.
pub fn delete_edge(&mut self, parent_id: ChannelId, channel_id: ChannelId) {
self.paths.retain(|path| {
!path
.windows(2)
.any(|window| window == [parent_id, channel_id])
});
/// Remove the given edge from this index. This will not remove the channel.
/// If this operation would result in a dangling edge, re-insert it.
pub fn delete_edge(&mut self, parent_id: Option<ChannelId>, channel_id: ChannelId) {
if let Some(parent_id) = parent_id {
self.paths.retain(|path| {
!path
.windows(2)
.any(|window| window == [parent_id, channel_id])
});
} else {
self.paths.retain(|path| path.first() != Some(&channel_id));
}
// Ensure that there is at least one channel path in the index
if !self
.paths
.iter()
.any(|path| path.iter().any(|id| id == &channel_id))
{
let path = ChannelPath(Arc::from([channel_id]));
let current_item: Vec<_> =
channel_path_sorting_key(&path, &self.channels_by_id).collect();
match self.paths.binary_search_by(|channel_path| {
current_item
.iter()
.copied()
.cmp(channel_path_sorting_key(channel_path, &self.channels_by_id))
}) {
Ok(ix) => self.paths.insert(ix, path),
Err(ix) => self.paths.insert(ix, path),
}
}
}
/// Delete the given channels from this index.
pub fn delete_channels(&mut self, channels: &[ChannelId]) {
self.channels_by_id.retain(|channel_id, _| !channels.contains(channel_id));
self.paths.retain(|channel_path| !channel_path.iter().any(|channel_id| {channels.contains(channel_id)}))
self.channels_by_id
.retain(|channel_id, _| !channels.contains(channel_id));
self.paths.retain(|channel_path| {
!channel_path
.iter()
.any(|channel_id| channels.contains(channel_id))
})
}
/// Upsert one or more channels into this index.
pub fn start_upsert(& mut self) -> ChannelPathsUpsertGuard {
pub fn start_upsert(&mut self) -> ChannelPathsUpsertGuard {
ChannelPathsUpsertGuard {
paths: &mut self.paths,
channels_by_id: &mut self.channels_by_id,
@ -90,7 +117,7 @@ impl ChannelIndex {
/// A guard for ensuring that the paths index maintains its sort and uniqueness
/// invariants after a series of insertions
pub struct ChannelPathsUpsertGuard<'a> {
paths: &'a mut Vec<ChannelPath>,
paths: &'a mut Vec<ChannelPath>,
channels_by_id: &'a mut ChannelsById,
}
@ -151,7 +178,6 @@ impl<'a> Drop for ChannelPathsUpsertGuard<'a> {
}
}
fn channel_path_sorting_key<'a>(
path: &'a [ChannelId],
channels_by_id: &'a ChannelsById,

38
crates/collab/src/db/queries/channels.rs
View File

@ -870,36 +870,22 @@ impl Database {
&self,
user: UserId,
channel: ChannelId,
from: Option<ChannelId>,
from: ChannelId,
tx: &DatabaseTransaction,
) -> Result<()> {
if let Some(from) = from {
self.check_user_is_channel_admin(from, user, &*tx).await?;
self.check_user_is_channel_admin(from, user, &*tx).await?;
let sql = r#"
let sql = r#"
DELETE FROM channel_paths
WHERE
id_path LIKE '%' || $1 || '/' || $2 || '%'
"#;
let channel_paths_stmt = Statement::from_sql_and_values(
self.pool.get_database_backend(),
sql,
[from.to_proto().into(), channel.to_proto().into()],
);
tx.execute(channel_paths_stmt).await?;
} else {
let sql = r#"
DELETE FROM channel_paths
WHERE
id_path = '/' || $1 || '/'
"#;
let channel_paths_stmt = Statement::from_sql_and_values(
self.pool.get_database_backend(),
sql,
[channel.to_proto().into()],
);
tx.execute(channel_paths_stmt).await?;
}
let channel_paths_stmt = Statement::from_sql_and_values(
self.pool.get_database_backend(),
sql,
[from.to_proto().into(), channel.to_proto().into()],
);
tx.execute(channel_paths_stmt).await?;
// Make sure that there is always at least one path to the channel
let sql = r#"
@ -929,7 +915,7 @@ impl Database {
&self,
user: UserId,
channel: ChannelId,
from: Option<ChannelId>,
from: ChannelId,
to: ChannelId,
) -> Result<Vec<Channel>> {
self.transaction(|tx| async move {
@ -941,6 +927,10 @@ impl Database {
self.unlink_channel_internal(user, channel, from, &*tx)
.await?;
dbg!(channel_path::Entity::find().all(&*tx).await);
dbg!(&moved_channels);
Ok(moved_channels)
})
.await

42
crates/collab/src/rpc.rs
View File

@ -2435,22 +2435,23 @@ async fn unlink_channel(
let db = session.db().await;
let channel_id = ChannelId::from_proto(request.channel_id);
let from = request.from.map(ChannelId::from_proto);
// Get the members before we remove it, so we know who to notify
let members = db.get_channel_members(channel_id).await?;
db.unlink_channel(session.user_id, channel_id, from).await?;
if let Some(from_parent) = from {
let members = db.get_channel_members(from_parent).await?;
let update = proto::UpdateChannels {
delete_channel_edge: vec![proto::ChannelEdge {
channel_id: channel_id.to_proto(),
parent_id: from_parent.to_proto(),
}],
..Default::default()
};
let connection_pool = session.connection_pool().await;
for member_id in members {
for connection_id in connection_pool.user_connection_ids(member_id) {
session.peer.send(connection_id, update.clone())?;
}
let update = proto::UpdateChannels {
delete_channel_edge: vec![proto::ChannelEdge {
channel_id: channel_id.to_proto(),
parent_id: from.map(ChannelId::to_proto),
}],
..Default::default()
};
let connection_pool = session.connection_pool().await;
for member_id in members {
for connection_id in connection_pool.user_connection_ids(member_id) {
session.peer.send(connection_id, update.clone())?;
}
}
@ -2468,16 +2469,24 @@ async fn move_channel(
let channel_id = ChannelId::from_proto(request.channel_id);
let from_parent = request.from.map(ChannelId::from_proto);
let to = ChannelId::from_proto(request.to);
let mut members = db.get_channel_members(channel_id).await?;
let channels_to_send: Vec<Channel> = db
.move_channel(session.user_id, channel_id, from_parent, to)
.await?;
let members_after = db.get_channel_members(channel_id).await?;
members.extend(members_after);
members.sort();
members.dedup();
if let Some(from_parent) = from_parent {
let members = db.get_channel_members(from_parent).await?;
let update = proto::UpdateChannels {
delete_channel_edge: vec![proto::ChannelEdge {
channel_id: channel_id.to_proto(),
parent_id: from_parent.to_proto(),
parent_id: Some(from_parent.to_proto()),
}],
..Default::default()
};
@ -2489,7 +2498,6 @@ async fn move_channel(
}
}
let members = db.get_channel_members(to).await?;
let connection_pool = session.connection_pool().await;
let update = proto::UpdateChannels {
channels: channels_to_send

10
crates/collab/src/tests/channel_buffer_tests.rs
View File

@ -25,7 +25,7 @@ async fn test_core_channel_buffers(
let client_b = server.create_client(cx_b, "user_b").await;
let channel_id = server
.make_channel("zed", (&client_a, cx_a), &mut [(&client_b, cx_b)])
.make_channel("zed", None, (&client_a, cx_a), &mut [(&client_b, cx_b)])
.await;
// Client A joins the channel buffer
@ -135,6 +135,7 @@ async fn test_channel_buffer_replica_ids(
let channel_id = server
.make_channel(
"the-channel",
None,
(&client_a, cx_a),
&mut [(&client_b, cx_b), (&client_c, cx_c)],
)
@ -279,7 +280,7 @@ async fn test_reopen_channel_buffer(deterministic: Arc<Deterministic>, cx_a: &mu
let client_a = server.create_client(cx_a, "user_a").await;
let channel_id = server
.make_channel("the-channel", (&client_a, cx_a), &mut [])
.make_channel("the-channel", None, (&client_a, cx_a), &mut [])
.await;
let channel_buffer_1 = client_a
@ -341,7 +342,7 @@ async fn test_channel_buffer_disconnect(
let client_b = server.create_client(cx_b, "user_b").await;
let channel_id = server
.make_channel("the-channel", (&client_a, cx_a), &mut [(&client_b, cx_b)])
.make_channel("the-channel", None, (&client_a, cx_a), &mut [(&client_b, cx_b)])
.await;
let channel_buffer_a = client_a
@ -411,7 +412,7 @@ async fn test_rejoin_channel_buffer(
let client_b = server.create_client(cx_b, "user_b").await;
let channel_id = server
.make_channel("the-channel", (&client_a, cx_a), &mut [(&client_b, cx_b)])
.make_channel("the-channel", None, (&client_a, cx_a), &mut [(&client_b, cx_b)])
.await;
let channel_buffer_a = client_a
@ -491,6 +492,7 @@ async fn test_channel_buffers_and_server_restarts(
let channel_id = server
.make_channel(
"the-channel",
None,
(&client_a, cx_a),
&mut [(&client_b, cx_b), (&client_c, cx_c)],
)

302
crates/collab/src/tests/channel_tests.rs
View File

@ -326,7 +326,7 @@ async fn test_joining_channel_ancestor_member(
let client_b = server.create_client(cx_b, "user_b").await;
let parent_id = server
.make_channel("parent", (&client_a, cx_a), &mut [(&client_b, cx_b)])
.make_channel("parent", None, (&client_a, cx_a), &mut [(&client_b, cx_b)])
.await;
let sub_id = client_a
@ -361,6 +361,7 @@ async fn test_channel_room(
let zed_id = server
.make_channel(
"zed",
None,
(&client_a, cx_a),
&mut [(&client_b, cx_b), (&client_c, cx_c)],
)
@ -544,9 +545,11 @@ async fn test_channel_jumping(deterministic: Arc<Deterministic>, cx_a: &mut Test
let mut server = TestServer::start(&deterministic).await;
let client_a = server.create_client(cx_a, "user_a").await;
let zed_id = server.make_channel("zed", (&client_a, cx_a), &mut []).await;
let zed_id = server
.make_channel("zed", None, (&client_a, cx_a), &mut [])
.await;
let rust_id = server
.make_channel("rust", (&client_a, cx_a), &mut [])
.make_channel("rust", None, (&client_a, cx_a), &mut [])
.await;
let active_call_a = cx_a.read(ActiveCall::global);
@ -597,7 +600,7 @@ async fn test_permissions_update_while_invited(
let client_b = server.create_client(cx_b, "user_b").await;
let rust_id = server
.make_channel("rust", (&client_a, cx_a), &mut [])
.make_channel("rust", None, (&client_a, cx_a), &mut [])
.await;
client_a
@ -658,7 +661,7 @@ async fn test_channel_rename(
let client_b = server.create_client(cx_b, "user_b").await;
let rust_id = server
.make_channel("rust", (&client_a, cx_a), &mut [(&client_b, cx_b)])
.make_channel("rust", None, (&client_a, cx_a), &mut [(&client_b, cx_b)])
.await;
// Rename the channel
@ -716,6 +719,7 @@ async fn test_call_from_channel(
let channel_id = server
.make_channel(
"x",
None,
(&client_a, cx_a),
&mut [(&client_b, cx_b), (&client_c, cx_c)],
)
@ -776,6 +780,7 @@ async fn test_lost_channel_creation(
deterministic: Arc<Deterministic>,
cx_a: &mut TestAppContext,
cx_b: &mut TestAppContext,
cx_c: &mut TestAppContext,
) {
deterministic.forbid_parking();
let mut server = TestServer::start(&deterministic).await;
@ -786,7 +791,9 @@ async fn test_lost_channel_creation(
.make_contacts(&mut [(&client_a, cx_a), (&client_b, cx_b)])
.await;
let channel_id = server.make_channel("x", (&client_a, cx_a), &mut []).await;
let channel_id = server
.make_channel("x", None, (&client_a, cx_a), &mut [])
.await;
// Invite a member
client_a
@ -875,140 +882,216 @@ async fn test_lost_channel_creation(
}
#[gpui::test]
async fn test_channel_moving(deterministic: Arc<Deterministic>, cx_a: &mut TestAppContext) {
async fn test_channel_moving(
deterministic: Arc<Deterministic>,
cx_a: &mut TestAppContext,
cx_b: &mut TestAppContext,
cx_c: &mut TestAppContext,
) {
deterministic.forbid_parking();
let mut server = TestServer::start(&deterministic).await;
let client_a = server.create_client(cx_a, "user_a").await;
let client_b = server.create_client(cx_b, "user_b").await;
let client_c = server.create_client(cx_c, "user_c").await;
let channel_a_id = client_a
.channel_store()
.update(cx_a, |channel_store, cx| {
channel_store.create_channel("channel-a", None, cx)
})
.await
.unwrap();
let channel_b_id = client_a
.channel_store()
.update(cx_a, |channel_store, cx| {
channel_store.create_channel("channel-b", Some(channel_a_id), cx)
})
.await
.unwrap();
let channel_c_id = client_a
.channel_store()
.update(cx_a, |channel_store, cx| {
channel_store.create_channel("channel-c", Some(channel_b_id), cx)
})
.await
.unwrap();
let channels = server
.make_channel_tree(
&[
("channel-a", None),
("channel-b", Some("channel-a")),
("channel-c", Some("channel-b")),
("channel-d", Some("channel-c")),
],
(&client_a, cx_a),
)
.await;
let channel_a_a_id = channels[0];
let channel_a_b_id = channels[1];
let channel_a_c_id = channels[2];
let channel_a_d_id = channels[3];
// Current shape:
// a - b - c
deterministic.run_until_parked();
assert_channels(
// a - b - c - d
assert_channels_list_shape(
client_a.channel_store(),
cx_a,
&[
ExpectedChannel {
id: channel_a_id,
name: "channel-a".to_string(),
depth: 0,
user_is_admin: true,
},
ExpectedChannel {
id: channel_b_id,
name: "channel-b".to_string(),
depth: 1,
user_is_admin: true,
},
ExpectedChannel {
id: channel_c_id,
name: "channel-c".to_string(),
depth: 2,
user_is_admin: true,
},
(channel_a_a_id, 0),
(channel_a_b_id, 1),
(channel_a_c_id, 2),
(channel_a_d_id, 3),
],
);
client_a
.channel_store()
.update(cx_a, |channel_store, cx| {
channel_store.move_channel(channel_c_id, Some(channel_b_id), channel_a_id, cx)
channel_store.move_channel(channel_a_d_id, Some(channel_a_c_id), channel_a_b_id, cx)
})
.await
.unwrap();
// Current shape:
// /- c
// a -- b
deterministic.run_until_parked();
assert_channels(
// /- d
// a - b -- c
assert_channels_list_shape(
client_a.channel_store(),
cx_a,
&[
ExpectedChannel {
id: channel_a_id,
name: "channel-a".to_string(),
depth: 0,
user_is_admin: true,
},
ExpectedChannel {
id: channel_b_id,
name: "channel-b".to_string(),
depth: 1,
user_is_admin: true,
},
ExpectedChannel {
id: channel_c_id,
name: "channel-c".to_string(),
depth: 1,
user_is_admin: true,
},
(channel_a_a_id, 0),
(channel_a_b_id, 1),
(channel_a_c_id, 2),
(channel_a_d_id, 2),
],
);
client_a
.channel_store()
.update(cx_a, |channel_store, cx| {
channel_store.link_channel(channel_c_id, channel_b_id, cx)
channel_store.link_channel(channel_a_d_id, channel_a_c_id, cx)
})
.await
.unwrap();
// Current shape:
// /------\
// a -- b -- c
deterministic.run_until_parked();
assert_channels(
// /------\
// a - b -- c -- d
assert_channels_list_shape(
client_a.channel_store(),
cx_a,
&[
ExpectedChannel {
id: channel_a_id,
name: "channel-a".to_string(),
depth: 0,
user_is_admin: true,
},
ExpectedChannel {
id: channel_b_id,
name: "channel-b".to_string(),
depth: 1,
user_is_admin: true,
},
ExpectedChannel {
id: channel_c_id,
name: "channel-c".to_string(),
depth: 2,
user_is_admin: true,
},
ExpectedChannel {
id: channel_c_id,
name: "channel-c".to_string(),
depth: 1,
user_is_admin: true,
},
(channel_a_a_id, 0),
(channel_a_b_id, 1),
(channel_a_c_id, 2),
(channel_a_d_id, 3),
(channel_a_d_id, 2),
],
);
let b_channels = server
.make_channel_tree(
&[
("channel-mu", None),
("channel-gamma", Some("channel-mu")),
("channel-epsilon", Some("channel-mu")),
],
(&client_b, cx_b),
)
.await;
let channel_b_mu_id = b_channels[0];
let channel_b_gamma_id = b_channels[1];
let channel_b_epsilon_id = b_channels[2];
// Current shape for B:
// /- ep
// mu -- ga
assert_channels_list_shape(
client_b.channel_store(),
cx_b,
&[
(channel_b_mu_id, 0),
(channel_b_gamma_id, 1),
(channel_b_epsilon_id, 1)
],
);
client_a.add_admin_to_channel((&client_b, cx_b), channel_a_b_id, cx_a).await;
// Current shape for B:
// /- ep
// mu -- ga
// /---------\
// b -- c -- d
assert_channels_list_shape(
client_b.channel_store(),
cx_b,
&[
// B's old channels
(channel_b_mu_id, 0),
(channel_b_gamma_id, 1),
(channel_b_epsilon_id, 1),
// New channels from a
(channel_a_b_id, 0),
(channel_a_c_id, 1),
(channel_a_d_id, 1),
(channel_a_d_id, 2),
],
);
client_b
.channel_store()
.update(cx_a, |channel_store, cx| {
channel_store.move_channel(channel_a_b_id, None, channel_b_epsilon_id, cx)
})
.await
.unwrap();
// Current shape for B:
// /---------\
// /- ep -- b -- c -- d
// mu -- ga
assert_channels_list_shape(
client_b.channel_store(),
cx_b,
&[
// B's old channels
(channel_b_mu_id, 0),
(channel_b_gamma_id, 1),
(channel_b_epsilon_id, 1),
// New channels from a, now under epsilon
(channel_a_b_id, 2),
(channel_a_c_id, 3),
(channel_a_d_id, 3),
(channel_a_d_id, 4),
],
);
client_b
.channel_store()
.update(cx_a, |channel_store, cx| {
channel_store.link_channel(channel_b_gamma_id, channel_a_b_id, cx)
})
.await
.unwrap();
// Current shape for B:
// /---------\
// /- ep -- b -- c -- d
// / \
// mu ---------- ga
assert_channels_list_shape(
client_b.channel_store(),
cx_b,
&[
// B's old channels
(channel_b_mu_id, 0),
(channel_b_gamma_id, 1),
(channel_b_epsilon_id, 1),
// New channels from a, now under epsilon, with gamma
(channel_a_b_id, 2),
(channel_b_gamma_id, 3),
(channel_a_c_id, 3),
(channel_a_d_id, 3),
(channel_a_d_id, 4),
],
);
// Current shape for A:
assert_channels_list_shape(
client_a.channel_store(),
cx_a,
&[
(channel_a_a_id, 0),
(channel_a_b_id, 1),
(channel_b_gamma_id, 1),
(channel_a_c_id, 2),
(channel_a_d_id, 3),
(channel_a_d_id, 2),
],
);
// TODO: Make sure to test that non-local root removing problem I was thinking about
}
#[derive(Debug, PartialEq)]
@ -1059,3 +1142,20 @@ fn assert_channels(
});
pretty_assertions::assert_eq!(actual, expected_channels);
}
#[track_caller]
fn assert_channels_list_shape(
channel_store: &ModelHandle<ChannelStore>,
cx: &TestAppContext,
expected_channels: &[(u64, usize)],
) {
cx.foreground().run_until_parked();
let actual = channel_store.read_with(cx, |store, _| {
store
.channels()
.map(|(depth, channel)| (channel.id, depth))
.collect::<Vec<_>>()
});
pretty_assertions::assert_eq!(actual, expected_channels);
}

71
crates/collab/src/tests/test_server.rs
View File

@ -288,6 +288,7 @@ impl TestServer {
pub async fn make_channel(
&self,
channel: &str,
parent: Option<u64>,
admin: (&TestClient, &mut TestAppContext),
members: &mut [(&TestClient, &mut TestAppContext)],
) -> u64 {
@ -296,7 +297,7 @@ impl TestServer {
.app_state
.channel_store
.update(admin_cx, |channel_store, cx| {
channel_store.create_channel(channel, None, cx)
channel_store.create_channel(channel, parent, cx)
})
.await
.unwrap();
@ -331,6 +332,39 @@ impl TestServer {
channel_id
}
pub async fn make_channel_tree(
&self,
channels: &[(&str, Option<&str>)],
creator: (&TestClient, &mut TestAppContext),
) -> Vec<u64> {
let mut observed_channels = HashMap::default();
let mut result = Vec::new();
for (channel, parent) in channels {
let id;
if let Some(parent) = parent {
if let Some(parent_id) = observed_channels.get(parent) {
id = self
.make_channel(channel, Some(*parent_id), (creator.0, creator.1), &mut [])
.await;
} else {
panic!(
"Edge {}->{} referenced before {} was created",
parent, channel, parent
)
}
} else {
id = self
.make_channel(channel, None, (creator.0, creator.1), &mut [])
.await;
}
observed_channels.insert(channel, id);
result.push(id);
}
result
}
pub async fn create_room(&self, clients: &mut [(&TestClient, &mut TestAppContext)]) {
self.make_contacts(clients).await;
@ -549,6 +583,41 @@ impl TestClient {
) -> WindowHandle<Workspace> {
cx.add_window(|cx| Workspace::new(0, project.clone(), self.app_state.clone(), cx))
}
pub async fn add_admin_to_channel(
&self,
user: (&TestClient, &mut TestAppContext),
channel: u64,
cx_self: &mut TestAppContext,
) {
let (other_client, other_cx) = user;
self
.app_state
.channel_store
.update(cx_self, |channel_store, cx| {
channel_store.invite_member(
channel,
other_client.user_id().unwrap(),
true,
cx,
)
})
.await
.unwrap();
cx_self.foreground().run_until_parked();
other_client
.app_state
.channel_store
.update(other_cx, |channels, _| {
channels.respond_to_channel_invite(channel, true)
})
.await
.unwrap();
}
}
impl Drop for TestClient {