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

Use "id" nomenclature more consistently

Browse source
This commit is contained in:
Antonio Scandurra 2022-12-15 10:15:59 +01:00
parent af77f1188a
commit 688f179256
8 changed files with 127 additions and 94 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -125,11 +125,13 @@ impl Database {
pub async fn delete_stale_projects( pub async fn delete_stale_projects(
&self, &self,
new_epoch: ServerId,
environment: &str, environment: &str,
new_server_id: ServerId,
) -> Result<()> { ) -> Result<()> {
self.transaction(|tx| async move { self.transaction(|tx| async move {
let stale_server_epochs = self.stale_server_ids(environment, new_epoch, &tx).await?; let stale_server_epochs = self
.stale_server_ids(environment, new_server_id, &tx)
.await?;
project_collaborator::Entity::delete_many() project_collaborator::Entity::delete_many()
.filter( .filter(
project_collaborator::Column::ConnectionServerId project_collaborator::Column::ConnectionServerId
@ -151,8 +153,8 @@ impl Database {
pub async fn stale_room_ids( pub async fn stale_room_ids(
&self, &self,
new_epoch: ServerId,
environment: &str, environment: &str,
new_server_id: ServerId,
) -> Result<Vec<RoomId>> { ) -> Result<Vec<RoomId>> {
self.transaction(|tx| async move { self.transaction(|tx| async move {
#[derive(Copy, Clone, Debug, EnumIter, DeriveColumn)] #[derive(Copy, Clone, Debug, EnumIter, DeriveColumn)]
@ -160,7 +162,9 @@ impl Database {
RoomId, RoomId,
} }
let stale_server_epochs = self.stale_server_ids(environment, new_epoch, &tx).await?; let stale_server_epochs = self
.stale_server_ids(environment, new_server_id, &tx)
.await?;
Ok(room_participant::Entity::find() Ok(room_participant::Entity::find()
.select_only() .select_only()
.column(room_participant::Column::RoomId) .column(room_participant::Column::RoomId)
@ -179,13 +183,13 @@ impl Database {
pub async fn refresh_room( pub async fn refresh_room(
&self, &self,
room_id: RoomId, room_id: RoomId,
new_epoch: ServerId, new_server_id: ServerId,
) -> Result<RoomGuard<RefreshedRoom>> { ) -> Result<RoomGuard<RefreshedRoom>> {
self.room_transaction(|tx| async move { self.room_transaction(|tx| async move {
let stale_participant_filter = Condition::all() let stale_participant_filter = Condition::all()
.add(room_participant::Column::RoomId.eq(room_id)) .add(room_participant::Column::RoomId.eq(room_id))
.add(room_participant::Column::AnsweringConnectionId.is_not_null()) .add(room_participant::Column::AnsweringConnectionId.is_not_null())
.add(room_participant::Column::AnsweringConnectionServerId.ne(new_epoch)); .add(room_participant::Column::AnsweringConnectionServerId.ne(new_server_id));
let stale_participant_user_ids = room_participant::Entity::find() let stale_participant_user_ids = room_participant::Entity::find()
.filter(stale_participant_filter.clone()) .filter(stale_participant_filter.clone())
@ -229,13 +233,17 @@ impl Database {
.await .await
} }
pub async fn delete_stale_servers(&self, new_epoch: ServerId, environment: &str) -> Result<()> { pub async fn delete_stale_servers(
&self,
new_server_id: ServerId,
environment: &str,
) -> Result<()> {
self.transaction(|tx| async move { self.transaction(|tx| async move {
server::Entity::delete_many() server::Entity::delete_many()
.filter( .filter(
Condition::all() Condition::all()
.add(server::Column::Environment.eq(environment)) .add(server::Column::Environment.eq(environment))
.add(server::Column::Id.ne(new_epoch)), .add(server::Column::Id.ne(new_server_id)),
) )
.exec(&*tx) .exec(&*tx)
.await?; .await?;
@ -247,14 +255,14 @@ impl Database {
async fn stale_server_ids( async fn stale_server_ids(
&self, &self,
environment: &str, environment: &str,
new_epoch: ServerId, new_server_id: ServerId,
tx: &DatabaseTransaction, tx: &DatabaseTransaction,
) -> Result<Vec<ServerId>> { ) -> Result<Vec<ServerId>> {
let stale_servers = server::Entity::find() let stale_servers = server::Entity::find()
.filter( .filter(
Condition::all() Condition::all()
.add(server::Column::Environment.eq(environment)) .add(server::Column::Environment.eq(environment))
.add(server::Column::Id.ne(new_epoch)), .add(server::Column::Id.ne(new_server_id)),
) )
.all(&*tx) .all(&*tx)
.await?; .await?;
@ -1155,13 +1163,13 @@ impl Database {
user_id: ActiveValue::set(user_id), user_id: ActiveValue::set(user_id),
answering_connection_id: ActiveValue::set(Some(connection.id as i32)), answering_connection_id: ActiveValue::set(Some(connection.id as i32)),
answering_connection_server_id: ActiveValue::set(Some(ServerId( answering_connection_server_id: ActiveValue::set(Some(ServerId(
connection.epoch as i32, connection.owner_id as i32,
))), ))),
answering_connection_lost: ActiveValue::set(false), answering_connection_lost: ActiveValue::set(false),
calling_user_id: ActiveValue::set(user_id), calling_user_id: ActiveValue::set(user_id),
calling_connection_id: ActiveValue::set(connection.id as i32), calling_connection_id: ActiveValue::set(connection.id as i32),
calling_connection_server_id: ActiveValue::set(Some(ServerId( calling_connection_server_id: ActiveValue::set(Some(ServerId(
connection.epoch as i32, connection.owner_id as i32,
))), ))),
..Default::default() ..Default::default()
} }
@ -1190,7 +1198,7 @@ impl Database {
calling_user_id: ActiveValue::set(calling_user_id), calling_user_id: ActiveValue::set(calling_user_id),
calling_connection_id: ActiveValue::set(calling_connection.id as i32), calling_connection_id: ActiveValue::set(calling_connection.id as i32),
calling_connection_server_id: ActiveValue::set(Some(ServerId( calling_connection_server_id: ActiveValue::set(Some(ServerId(
calling_connection.epoch as i32, calling_connection.owner_id as i32,
))), ))),
initial_project_id: ActiveValue::set(initial_project_id), initial_project_id: ActiveValue::set(initial_project_id),
..Default::default() ..Default::default()
@ -1274,7 +1282,7 @@ impl Database {
) )
.add( .add(
room_participant::Column::CallingConnectionServerId room_participant::Column::CallingConnectionServerId
.eq(calling_connection.epoch as i32), .eq(calling_connection.owner_id as i32),
) )
.add(room_participant::Column::AnsweringConnectionId.is_null()), .add(room_participant::Column::AnsweringConnectionId.is_null()),
) )
@ -1314,14 +1322,14 @@ impl Database {
.add(room_participant::Column::AnsweringConnectionLost.eq(true)) .add(room_participant::Column::AnsweringConnectionLost.eq(true))
.add( .add(
room_participant::Column::AnsweringConnectionServerId room_participant::Column::AnsweringConnectionServerId
.ne(connection.epoch as i32), .ne(connection.owner_id as i32),
), ),
), ),
) )
.set(room_participant::ActiveModel { .set(room_participant::ActiveModel {
answering_connection_id: ActiveValue::set(Some(connection.id as i32)), answering_connection_id: ActiveValue::set(Some(connection.id as i32)),
answering_connection_server_id: ActiveValue::set(Some(ServerId( answering_connection_server_id: ActiveValue::set(Some(ServerId(
connection.epoch as i32, connection.owner_id as i32,
))), ))),
answering_connection_lost: ActiveValue::set(false), answering_connection_lost: ActiveValue::set(false),
..Default::default() ..Default::default()
@ -1349,7 +1357,7 @@ impl Database {
) )
.add( .add(
room_participant::Column::AnsweringConnectionServerId room_participant::Column::AnsweringConnectionServerId
.eq(connection.epoch as i32), .eq(connection.owner_id as i32),
), ),
) )
.one(&*tx) .one(&*tx)
@ -1372,7 +1380,7 @@ impl Database {
) )
.add( .add(
room_participant::Column::CallingConnectionServerId room_participant::Column::CallingConnectionServerId
.eq(connection.epoch as i32), .eq(connection.owner_id as i32),
) )
.add(room_participant::Column::AnsweringConnectionId.is_null()), .add(room_participant::Column::AnsweringConnectionId.is_null()),
) )
@ -1408,7 +1416,7 @@ impl Database {
) )
.add( .add(
project_collaborator::Column::ConnectionServerId project_collaborator::Column::ConnectionServerId
.eq(connection.epoch as i32), .eq(connection.owner_id as i32),
), ),
) )
.into_values::<_, QueryProjectIds>() .into_values::<_, QueryProjectIds>()
@ -1432,7 +1440,7 @@ impl Database {
}); });
let collaborator_connection_id = ConnectionId { let collaborator_connection_id = ConnectionId {
epoch: collaborator.connection_server_id.0 as u32, owner_id: collaborator.connection_server_id.0 as u32,
id: collaborator.connection_id as u32, id: collaborator.connection_id as u32,
}; };
if collaborator_connection_id != connection { if collaborator_connection_id != connection {
@ -1455,7 +1463,7 @@ impl Database {
) )
.add( .add(
project_collaborator::Column::ConnectionServerId project_collaborator::Column::ConnectionServerId
.eq(connection.epoch as i32), .eq(connection.owner_id as i32),
), ),
) )
.exec(&*tx) .exec(&*tx)
@ -1468,7 +1476,8 @@ impl Database {
.add(project::Column::RoomId.eq(room_id)) .add(project::Column::RoomId.eq(room_id))
.add(project::Column::HostConnectionId.eq(connection.id as i32)) .add(project::Column::HostConnectionId.eq(connection.id as i32))
.add( .add(
project::Column::HostConnectionServerId.eq(connection.epoch as i32), project::Column::HostConnectionServerId
.eq(connection.owner_id as i32),
), ),
) )
.exec(&*tx) .exec(&*tx)
@ -1536,7 +1545,7 @@ impl Database {
) )
.add( .add(
room_participant::Column::AnsweringConnectionServerId room_participant::Column::AnsweringConnectionServerId
.eq(connection.epoch as i32), .eq(connection.owner_id as i32),
), ),
) )
.set(room_participant::ActiveModel { .set(room_participant::ActiveModel {
@ -1571,7 +1580,7 @@ impl Database {
) )
.add( .add(
room_participant::Column::AnsweringConnectionServerId room_participant::Column::AnsweringConnectionServerId
.eq(connection.epoch as i32), .eq(connection.owner_id as i32),
), ),
) )
.one(&*tx) .one(&*tx)
@ -1593,7 +1602,7 @@ impl Database {
.add(project_collaborator::Column::ConnectionId.eq(connection.id as i32)) .add(project_collaborator::Column::ConnectionId.eq(connection.id as i32))
.add( .add(
project_collaborator::Column::ConnectionServerId project_collaborator::Column::ConnectionServerId
.eq(connection.epoch as i32), .eq(connection.owner_id as i32),
), ),
) )
.all(&*tx) .all(&*tx)
@ -1604,7 +1613,7 @@ impl Database {
.add(project_collaborator::Column::ConnectionId.eq(connection.id as i32)) .add(project_collaborator::Column::ConnectionId.eq(connection.id as i32))
.add( .add(
project_collaborator::Column::ConnectionServerId project_collaborator::Column::ConnectionServerId
.eq(connection.epoch as i32), .eq(connection.owner_id as i32),
), ),
) )
.exec(&*tx) .exec(&*tx)
@ -1621,7 +1630,7 @@ impl Database {
.into_iter() .into_iter()
.map(|collaborator| ConnectionId { .map(|collaborator| ConnectionId {
id: collaborator.connection_id as u32, id: collaborator.connection_id as u32,
epoch: collaborator.connection_server_id.0 as u32, owner_id: collaborator.connection_server_id.0 as u32,
}) })
.collect(); .collect();
@ -1630,7 +1639,7 @@ impl Database {
host_user_id: project.host_user_id, host_user_id: project.host_user_id,
host_connection_id: ConnectionId { host_connection_id: ConnectionId {
id: project.host_connection_id as u32, id: project.host_connection_id as u32,
epoch: project.host_connection_server_id.0 as u32, owner_id: project.host_connection_server_id.0 as u32,
}, },
connection_ids, connection_ids,
}); });
@ -1641,7 +1650,9 @@ impl Database {
.filter( .filter(
Condition::all() Condition::all()
.add(project::Column::HostConnectionId.eq(connection.id as i32)) .add(project::Column::HostConnectionId.eq(connection.id as i32))
.add(project::Column::HostConnectionServerId.eq(connection.epoch as i32)), .add(
project::Column::HostConnectionServerId.eq(connection.owner_id as i32),
),
) )
.exec(&*tx) .exec(&*tx)
.await?; .await?;
@ -1717,7 +1728,7 @@ impl Database {
}; };
let answering_connection = ConnectionId { let answering_connection = ConnectionId {
epoch: answering_connection_server_id.0 as u32, owner_id: answering_connection_server_id.0 as u32,
id: answering_connection_id as u32, id: answering_connection_id as u32,
}; };
participants.insert( participants.insert(
@ -1748,7 +1759,7 @@ impl Database {
while let Some(row) = db_projects.next().await { while let Some(row) = db_projects.next().await {
let (db_project, db_worktree) = row?; let (db_project, db_worktree) = row?;
let host_connection = ConnectionId { let host_connection = ConnectionId {
epoch: db_project.host_connection_server_id.0 as u32, owner_id: db_project.host_connection_server_id.0 as u32,
id: db_project.host_connection_id as u32, id: db_project.host_connection_id as u32,
}; };
if let Some(participant) = participants.get_mut(&host_connection) { if let Some(participant) = participants.get_mut(&host_connection) {
@ -1818,7 +1829,7 @@ impl Database {
) )
.add( .add(
room_participant::Column::AnsweringConnectionServerId room_participant::Column::AnsweringConnectionServerId
.eq(connection.epoch as i32), .eq(connection.owner_id as i32),
), ),
) )
.one(&*tx) .one(&*tx)
@ -1832,7 +1843,7 @@ impl Database {
room_id: ActiveValue::set(participant.room_id), room_id: ActiveValue::set(participant.room_id),
host_user_id: ActiveValue::set(participant.user_id), host_user_id: ActiveValue::set(participant.user_id),
host_connection_id: ActiveValue::set(connection.id as i32), host_connection_id: ActiveValue::set(connection.id as i32),
host_connection_server_id: ActiveValue::set(ServerId(connection.epoch as i32)), host_connection_server_id: ActiveValue::set(ServerId(connection.owner_id as i32)),
..Default::default() ..Default::default()
} }
.insert(&*tx) .insert(&*tx)
@ -1857,7 +1868,7 @@ impl Database {
project_collaborator::ActiveModel { project_collaborator::ActiveModel {
project_id: ActiveValue::set(project.id), project_id: ActiveValue::set(project.id),
connection_id: ActiveValue::set(connection.id as i32), connection_id: ActiveValue::set(connection.id as i32),
connection_server_id: ActiveValue::set(ServerId(connection.epoch as i32)), connection_server_id: ActiveValue::set(ServerId(connection.owner_id as i32)),
user_id: ActiveValue::set(participant.user_id), user_id: ActiveValue::set(participant.user_id),
replica_id: ActiveValue::set(ReplicaId(0)), replica_id: ActiveValue::set(ReplicaId(0)),
is_host: ActiveValue::set(true), is_host: ActiveValue::set(true),
@ -1885,7 +1896,7 @@ impl Database {
.await? .await?
.ok_or_else(|| anyhow!("project not found"))?; .ok_or_else(|| anyhow!("project not found"))?;
let host_connection = ConnectionId { let host_connection = ConnectionId {
epoch: project.host_connection_server_id.0 as u32, owner_id: project.host_connection_server_id.0 as u32,
id: project.host_connection_id as u32, id: project.host_connection_id as u32,
}; };
if host_connection == connection { if host_connection == connection {
@ -1913,7 +1924,9 @@ impl Database {
.filter( .filter(
Condition::all() Condition::all()
.add(project::Column::HostConnectionId.eq(connection.id as i32)) .add(project::Column::HostConnectionId.eq(connection.id as i32))
.add(project::Column::HostConnectionServerId.eq(connection.epoch as i32)), .add(
project::Column::HostConnectionServerId.eq(connection.owner_id as i32),
),
) )
.one(&*tx) .one(&*tx)
.await? .await?
@ -1971,7 +1984,9 @@ impl Database {
.filter( .filter(
Condition::all() Condition::all()
.add(project::Column::HostConnectionId.eq(connection.id as i32)) .add(project::Column::HostConnectionId.eq(connection.id as i32))
.add(project::Column::HostConnectionServerId.eq(connection.epoch as i32)), .add(
project::Column::HostConnectionServerId.eq(connection.owner_id as i32),
),
) )
.one(&*tx) .one(&*tx)
.await? .await?
@ -2068,7 +2083,7 @@ impl Database {
.await? .await?
.ok_or_else(|| anyhow!("no such project"))?; .ok_or_else(|| anyhow!("no such project"))?;
let host_connection = ConnectionId { let host_connection = ConnectionId {
epoch: project.host_connection_server_id.0 as u32, owner_id: project.host_connection_server_id.0 as u32,
id: project.host_connection_id as u32, id: project.host_connection_id as u32,
}; };
if host_connection != connection { if host_connection != connection {
@ -2125,7 +2140,7 @@ impl Database {
.await? .await?
.ok_or_else(|| anyhow!("no such project"))?; .ok_or_else(|| anyhow!("no such project"))?;
let host_connection = ConnectionId { let host_connection = ConnectionId {
epoch: project.host_connection_server_id.0 as u32, owner_id: project.host_connection_server_id.0 as u32,
id: project.host_connection_id as u32, id: project.host_connection_id as u32,
}; };
if host_connection != connection { if host_connection != connection {
@ -2171,7 +2186,7 @@ impl Database {
) )
.add( .add(
room_participant::Column::AnsweringConnectionServerId room_participant::Column::AnsweringConnectionServerId
.eq(connection.epoch as i32), .eq(connection.owner_id as i32),
), ),
) )
.one(&*tx) .one(&*tx)
@ -2201,7 +2216,7 @@ impl Database {
let new_collaborator = project_collaborator::ActiveModel { let new_collaborator = project_collaborator::ActiveModel {
project_id: ActiveValue::set(project_id), project_id: ActiveValue::set(project_id),
connection_id: ActiveValue::set(connection.id as i32), connection_id: ActiveValue::set(connection.id as i32),
connection_server_id: ActiveValue::set(ServerId(connection.epoch as i32)), connection_server_id: ActiveValue::set(ServerId(connection.owner_id as i32)),
user_id: ActiveValue::set(participant.user_id), user_id: ActiveValue::set(participant.user_id),
replica_id: ActiveValue::set(replica_id), replica_id: ActiveValue::set(replica_id),
is_host: ActiveValue::set(false), is_host: ActiveValue::set(false),
@ -2313,7 +2328,7 @@ impl Database {
.add(project_collaborator::Column::ConnectionId.eq(connection.id as i32)) .add(project_collaborator::Column::ConnectionId.eq(connection.id as i32))
.add( .add(
project_collaborator::Column::ConnectionServerId project_collaborator::Column::ConnectionServerId
.eq(connection.epoch as i32), .eq(connection.owner_id as i32),
), ),
) )
.exec(&*tx) .exec(&*tx)
@ -2333,7 +2348,7 @@ impl Database {
let connection_ids = collaborators let connection_ids = collaborators
.into_iter() .into_iter()
.map(|collaborator| ConnectionId { .map(|collaborator| ConnectionId {
epoch: collaborator.connection_server_id.0 as u32, owner_id: collaborator.connection_server_id.0 as u32,
id: collaborator.connection_id as u32, id: collaborator.connection_id as u32,
}) })
.collect(); .collect();
@ -2342,7 +2357,7 @@ impl Database {
id: project_id, id: project_id,
host_user_id: project.host_user_id, host_user_id: project.host_user_id,
host_connection_id: ConnectionId { host_connection_id: ConnectionId {
epoch: project.host_connection_server_id.0 as u32, owner_id: project.host_connection_server_id.0 as u32,
id: project.host_connection_id as u32, id: project.host_connection_id as u32,
}, },
connection_ids, connection_ids,
@ -2369,7 +2384,7 @@ impl Database {
if collaborators.iter().any(|collaborator| { if collaborators.iter().any(|collaborator| {
let collaborator_connection = ConnectionId { let collaborator_connection = ConnectionId {
epoch: collaborator.connection_server_id.0 as u32, owner_id: collaborator.connection_server_id.0 as u32,
id: collaborator.connection_id as u32, id: collaborator.connection_id as u32,
}; };
collaborator_connection == connection collaborator_connection == connection
@ -2401,7 +2416,7 @@ impl Database {
while let Some(participant) = participants.next().await { while let Some(participant) = participants.next().await {
let participant = participant?; let participant = participant?;
connection_ids.insert(ConnectionId { connection_ids.insert(ConnectionId {
epoch: participant.connection_server_id.0 as u32, owner_id: participant.connection_server_id.0 as u32,
id: participant.connection_id as u32, id: participant.connection_id as u32,
}); });
} }
@ -2433,7 +2448,7 @@ impl Database {
while let Some(participant) = participants.next().await { while let Some(participant) = participants.next().await {
let participant = participant?; let participant = participant?;
guest_connection_ids.push(ConnectionId { guest_connection_ids.push(ConnectionId {
epoch: participant.connection_server_id.0 as u32, owner_id: participant.connection_server_id.0 as u32,
id: participant.connection_id as u32, id: participant.connection_id as u32,
}); });
} }

18
crates/collab/src/db/tests.rs
View file

@ -410,7 +410,7 @@ test_both_dbs!(
test_project_count_sqlite, test_project_count_sqlite,
db, db,
{ {
let epoch = db.create_server("test").await.unwrap().0 as u32; let owner_id = db.create_server("test").await.unwrap().0 as u32;
let user1 = db let user1 = db
.create_user( .create_user(
@ -438,7 +438,7 @@ test_both_dbs!(
.unwrap(); .unwrap();
let room_id = RoomId::from_proto( let room_id = RoomId::from_proto(
db.create_room(user1.user_id, ConnectionId { epoch, id: 0 }, "") db.create_room(user1.user_id, ConnectionId { owner_id, id: 0 }, "")
.await .await
.unwrap() .unwrap()
.id, .id,
@ -446,34 +446,36 @@ test_both_dbs!(
db.call( db.call(
room_id, room_id,
user1.user_id, user1.user_id,
ConnectionId { epoch, id: 0 }, ConnectionId { owner_id, id: 0 },
user2.user_id, user2.user_id,
None, None,
) )
.await .await
.unwrap(); .unwrap();
db.join_room(room_id, user2.user_id, ConnectionId { epoch, id: 1 }) db.join_room(room_id, user2.user_id, ConnectionId { owner_id, id: 1 })
.await .await
.unwrap(); .unwrap();
assert_eq!(db.project_count_excluding_admins().await.unwrap(), 0); assert_eq!(db.project_count_excluding_admins().await.unwrap(), 0);
db.share_project(room_id, ConnectionId { epoch, id: 1 }, &[]) db.share_project(room_id, ConnectionId { owner_id, id: 1 }, &[])
.await .await
.unwrap(); .unwrap();
assert_eq!(db.project_count_excluding_admins().await.unwrap(), 1); assert_eq!(db.project_count_excluding_admins().await.unwrap(), 1);
db.share_project(room_id, ConnectionId { epoch, id: 1 }, &[]) db.share_project(room_id, ConnectionId { owner_id, id: 1 }, &[])
.await .await
.unwrap(); .unwrap();
assert_eq!(db.project_count_excluding_admins().await.unwrap(), 2); assert_eq!(db.project_count_excluding_admins().await.unwrap(), 2);
// Projects shared by admins aren't counted. // Projects shared by admins aren't counted.
db.share_project(room_id, ConnectionId { epoch, id: 0 }, &[]) db.share_project(room_id, ConnectionId { owner_id, id: 0 }, &[])
.await .await
.unwrap(); .unwrap();
assert_eq!(db.project_count_excluding_admins().await.unwrap(), 2); assert_eq!(db.project_count_excluding_admins().await.unwrap(), 2);
db.leave_room(ConnectionId { epoch, id: 1 }).await.unwrap(); db.leave_room(ConnectionId { owner_id, id: 1 })
.await
.unwrap();
assert_eq!(db.project_count_excluding_admins().await.unwrap(), 0); assert_eq!(db.project_count_excluding_admins().await.unwrap(), 0);
} }
); );

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

@ -6123,9 +6123,17 @@ async fn test_random_collaboration(
30..=34 => { 30..=34 => {
log::info!("Simulating server restart"); log::info!("Simulating server restart");
server.reset().await; server.reset().await;
deterministic.advance_clock(RECEIVE_TIMEOUT + RECONNECT_TIMEOUT); deterministic.advance_clock(RECEIVE_TIMEOUT);
server.start().await.unwrap(); server.start().await.unwrap();
deterministic.advance_clock(CLEANUP_TIMEOUT); deterministic.advance_clock(CLEANUP_TIMEOUT);
let environment = &server.app_state.config.zed_environment;
let stale_room_ids = server
.app_state
.db
.stale_room_ids(environment, server.id())
.await
.unwrap();
assert_eq!(stale_room_ids, vec![]);
} }
_ if !op_start_signals.is_empty() => { _ if !op_start_signals.is_empty() => {
while operations < max_operations && rng.lock().gen_bool(0.7) { while operations < max_operations && rng.lock().gen_bool(0.7) {

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

@ -138,7 +138,7 @@ impl Deref for DbHandle {
} }
pub struct Server { pub struct Server {
epoch: parking_lot::Mutex<ServerId>, id: parking_lot::Mutex<ServerId>,
peer: Arc<Peer>, peer: Arc<Peer>,
pub(crate) connection_pool: Arc<parking_lot::Mutex<ConnectionPool>>, pub(crate) connection_pool: Arc<parking_lot::Mutex<ConnectionPool>>,
app_state: Arc<AppState>, app_state: Arc<AppState>,
@ -169,10 +169,10 @@ where
} }
impl Server { impl Server {
pub fn new(epoch: ServerId, app_state: Arc<AppState>, executor: Executor) -> Arc<Self> { pub fn new(id: ServerId, app_state: Arc<AppState>, executor: Executor) -> Arc<Self> {
let mut server = Self { let mut server = Self {
epoch: parking_lot::Mutex::new(epoch), id: parking_lot::Mutex::new(id),
peer: Peer::new(epoch.0 as u32), peer: Peer::new(id.0 as u32),
app_state, app_state,
executor, executor,
connection_pool: Default::default(), connection_pool: Default::default(),
@ -241,7 +241,7 @@ impl Server {
} }
pub async fn start(&self) -> Result<()> { pub async fn start(&self) -> Result<()> {
let epoch = *self.epoch.lock(); let server_id = *self.id.lock();
let app_state = self.app_state.clone(); let app_state = self.app_state.clone();
let peer = self.peer.clone(); let peer = self.peer.clone();
let timeout = self.executor.sleep(CLEANUP_TIMEOUT); let timeout = self.executor.sleep(CLEANUP_TIMEOUT);
@ -254,7 +254,7 @@ impl Server {
tracing::info!("begin deleting stale projects"); tracing::info!("begin deleting stale projects");
app_state app_state
.db .db
.delete_stale_projects(epoch, &app_state.config.zed_environment) .delete_stale_projects(&app_state.config.zed_environment, server_id)
.await?; .await?;
tracing::info!("finish deleting stale projects"); tracing::info!("finish deleting stale projects");
@ -266,7 +266,7 @@ impl Server {
tracing::info!("cleanup timeout expired, retrieving stale rooms"); tracing::info!("cleanup timeout expired, retrieving stale rooms");
if let Some(room_ids) = app_state if let Some(room_ids) = app_state
.db .db
.stale_room_ids(epoch, &app_state.config.zed_environment) .stale_room_ids(&app_state.config.zed_environment, server_id)
.await .await
.trace_err() .trace_err()
{ {
@ -278,7 +278,7 @@ impl Server {
let mut delete_live_kit_room = false; let mut delete_live_kit_room = false;
if let Ok(mut refreshed_room) = if let Ok(mut refreshed_room) =
app_state.db.refresh_room(room_id, epoch).await app_state.db.refresh_room(room_id, server_id).await
{ {
tracing::info!( tracing::info!(
room_id = room_id.0, room_id = room_id.0,
@ -354,7 +354,7 @@ impl Server {
app_state app_state
.db .db
.delete_stale_servers(epoch, &app_state.config.zed_environment) .delete_stale_servers(server_id, &app_state.config.zed_environment)
.await .await
.trace_err(); .trace_err();
} }
@ -370,10 +370,15 @@ impl Server {
} }
#[cfg(test)] #[cfg(test)]
pub fn reset(&self, epoch: ServerId) { pub fn reset(&self, id: ServerId) {
self.teardown(); self.teardown();
*self.epoch.lock() = epoch; *self.id.lock() = id;
self.peer.reset(epoch.0 as u32); self.peer.reset(id.0 as u32);
}
#[cfg(test)]
pub fn id(&self) -> ServerId {
*self.id.lock()
} }
fn add_handler<F, Fut, M>(&mut self, handler: F) -> &mut Self fn add_handler<F, Fut, M>(&mut self, handler: F) -> &mut Self
@ -1156,7 +1161,7 @@ async fn join_project(
.iter() .iter()
.map(|collaborator| { .map(|collaborator| {
let peer_id = proto::PeerId { let peer_id = proto::PeerId {
epoch: collaborator.connection_server_id.0 as u32, owner_id: collaborator.connection_server_id.0 as u32,
id: collaborator.connection_id as u32, id: collaborator.connection_id as u32,
}; };
proto::Collaborator { proto::Collaborator {
@ -1412,7 +1417,7 @@ where
.find(|collaborator| collaborator.is_host) .find(|collaborator| collaborator.is_host)
.ok_or_else(|| anyhow!("host not found"))?; .ok_or_else(|| anyhow!("host not found"))?;
ConnectionId { ConnectionId {
epoch: host.connection_server_id.0 as u32, owner_id: host.connection_server_id.0 as u32,
id: host.connection_id as u32, id: host.connection_id as u32,
} }
}; };
@ -1443,7 +1448,7 @@ async fn save_buffer(
.find(|collaborator| collaborator.is_host) .find(|collaborator| collaborator.is_host)
.ok_or_else(|| anyhow!("host not found"))?; .ok_or_else(|| anyhow!("host not found"))?;
ConnectionId { ConnectionId {
epoch: host.connection_server_id.0 as u32, owner_id: host.connection_server_id.0 as u32,
id: host.connection_id as u32, id: host.connection_id as u32,
} }
}; };
@ -1459,13 +1464,13 @@ async fn save_buffer(
.await?; .await?;
collaborators.retain(|collaborator| { collaborators.retain(|collaborator| {
let collaborator_connection = ConnectionId { let collaborator_connection = ConnectionId {
epoch: collaborator.connection_server_id.0 as u32, owner_id: collaborator.connection_server_id.0 as u32,
id: collaborator.connection_id as u32, id: collaborator.connection_id as u32,
}; };
collaborator_connection != session.connection_id collaborator_connection != session.connection_id
}); });
let project_connection_ids = collaborators.iter().map(|collaborator| ConnectionId { let project_connection_ids = collaborators.iter().map(|collaborator| ConnectionId {
epoch: collaborator.connection_server_id.0 as u32, owner_id: collaborator.connection_server_id.0 as u32,
id: collaborator.connection_id as u32, id: collaborator.connection_id as u32,
}); });
broadcast(host_connection_id, project_connection_ids, |conn_id| { broadcast(host_connection_id, project_connection_ids, |conn_id| {

2
crates/rpc/proto/zed.proto
View file

@ -2,7 +2,7 @@ syntax = "proto3";
package zed.messages; package zed.messages;
message PeerId { message PeerId {
uint32 epoch = 1; uint32 owner_id = 1;
uint32 id = 2; uint32 id = 2;
} }

2
crates/rpc/src/macros.rs
View file

@ -7,7 +7,7 @@ macro_rules! messages {
Some(Box::new(TypedEnvelope { Some(Box::new(TypedEnvelope {
sender_id, sender_id,
original_sender_id: envelope.original_sender_id.map(|original_sender| PeerId { original_sender_id: envelope.original_sender_id.map(|original_sender| PeerId {
epoch: original_sender.epoch, owner_id: original_sender.owner_id,
id: original_sender.id id: original_sender.id
}), }),
message_id: envelope.id, message_id: envelope.id,

10
crates/rpc/src/peer.rs
View file

@ -25,14 +25,14 @@ use tracing::instrument;
#[derive(Clone, Copy, Default, PartialEq, Eq, PartialOrd, Ord, Hash, Debug, Serialize)] #[derive(Clone, Copy, Default, PartialEq, Eq, PartialOrd, Ord, Hash, Debug, Serialize)]
pub struct ConnectionId { pub struct ConnectionId {
pub epoch: u32, pub owner_id: u32,
pub id: u32, pub id: u32,
} }
impl Into<PeerId> for ConnectionId { impl Into<PeerId> for ConnectionId {
fn into(self) -> PeerId { fn into(self) -> PeerId {
PeerId { PeerId {
epoch: self.epoch, owner_id: self.owner_id,
id: self.id, id: self.id,
} }
} }
@ -41,7 +41,7 @@ impl Into<PeerId> for ConnectionId {
impl From<PeerId> for ConnectionId { impl From<PeerId> for ConnectionId {
fn from(peer_id: PeerId) -> Self { fn from(peer_id: PeerId) -> Self {
Self { Self {
epoch: peer_id.epoch, owner_id: peer_id.owner_id,
id: peer_id.id, id: peer_id.id,
} }
} }
@ -49,7 +49,7 @@ impl From<PeerId> for ConnectionId {
impl fmt::Display for ConnectionId { impl fmt::Display for ConnectionId {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(f, "{}/{}", self.epoch, self.id) write!(f, "{}/{}", self.owner_id, self.id)
} }
} }
@ -157,7 +157,7 @@ impl Peer {
let (outgoing_tx, mut outgoing_rx) = mpsc::unbounded(); let (outgoing_tx, mut outgoing_rx) = mpsc::unbounded();
let connection_id = ConnectionId { let connection_id = ConnectionId {
epoch: self.epoch.load(SeqCst), owner_id: self.epoch.load(SeqCst),
id: self.next_connection_id.fetch_add(1, SeqCst), id: self.next_connection_id.fetch_add(1, SeqCst),
}; };
let connection_state = ConnectionState { let connection_state = ConnectionState {

31
crates/rpc/src/proto.rs
View file

@ -78,13 +78,13 @@ impl<T: EnvelopedMessage> AnyTypedEnvelope for TypedEnvelope<T> {
impl PeerId { impl PeerId {
pub fn from_u64(peer_id: u64) -> Self { pub fn from_u64(peer_id: u64) -> Self {
let epoch = (peer_id >> 32) as u32; let owner_id = (peer_id >> 32) as u32;
let id = peer_id as u32; let id = peer_id as u32;
Self { epoch, id } Self { owner_id, id }
} }
pub fn as_u64(self) -> u64 { pub fn as_u64(self) -> u64 {
((self.epoch as u64) << 32) | (self.id as u64) ((self.owner_id as u64) << 32) | (self.id as u64)
} }
} }
@ -94,8 +94,8 @@ impl Eq for PeerId {}
impl Ord for PeerId { impl Ord for PeerId {
fn cmp(&self, other: &Self) -> cmp::Ordering { fn cmp(&self, other: &Self) -> cmp::Ordering {
self.epoch self.owner_id
.cmp(&other.epoch) .cmp(&other.owner_id)
.then_with(|| self.id.cmp(&other.id)) .then_with(|| self.id.cmp(&other.id))
} }
} }
@ -108,14 +108,14 @@ impl PartialOrd for PeerId {
impl std::hash::Hash for PeerId { impl std::hash::Hash for PeerId {
fn hash<H: std::hash::Hasher>(&self, state: &mut H) { fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
self.epoch.hash(state); self.owner_id.hash(state);
self.id.hash(state); self.id.hash(state);
} }
} }
impl fmt::Display for PeerId { impl fmt::Display for PeerId {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(f, "{}/{}", self.epoch, self.id) write!(f, "{}/{}", self.owner_id, self.id)
} }
} }
@ -124,7 +124,7 @@ impl FromStr for PeerId {
fn from_str(s: &str) -> Result<Self, Self::Err> { fn from_str(s: &str) -> Result<Self, Self::Err> {
let mut components = s.split('/'); let mut components = s.split('/');
let epoch = components let owner_id = components
.next() .next()
.ok_or_else(|| anyhow!("invalid peer id {:?}", s))? .ok_or_else(|| anyhow!("invalid peer id {:?}", s))?
.parse()?; .parse()?;
@ -132,7 +132,7 @@ impl FromStr for PeerId {
.next() .next()
.ok_or_else(|| anyhow!("invalid peer id {:?}", s))? .ok_or_else(|| anyhow!("invalid peer id {:?}", s))?
.parse()?; .parse()?;
Ok(PeerId { epoch, id }) Ok(PeerId { owner_id, id })
} }
} }
@ -542,20 +542,23 @@ mod tests {
#[gpui::test] #[gpui::test]
fn test_converting_peer_id_from_and_to_u64() { fn test_converting_peer_id_from_and_to_u64() {
let peer_id = PeerId { epoch: 10, id: 3 };
assert_eq!(PeerId::from_u64(peer_id.as_u64()), peer_id);
let peer_id = PeerId { let peer_id = PeerId {
epoch: u32::MAX, owner_id: 10,
id: 3, id: 3,
}; };
assert_eq!(PeerId::from_u64(peer_id.as_u64()), peer_id); assert_eq!(PeerId::from_u64(peer_id.as_u64()), peer_id);
let peer_id = PeerId { let peer_id = PeerId {
epoch: 10, owner_id: u32::MAX,
id: 3,
};
assert_eq!(PeerId::from_u64(peer_id.as_u64()), peer_id);
let peer_id = PeerId {
owner_id: 10,
id: u32::MAX, id: u32::MAX,
}; };
assert_eq!(PeerId::from_u64(peer_id.as_u64()), peer_id); assert_eq!(PeerId::from_u64(peer_id.as_u64()), peer_id);
let peer_id = PeerId { let peer_id = PeerId {
epoch: u32::MAX, owner_id: u32::MAX,
id: u32::MAX, id: u32::MAX,
}; };
assert_eq!(PeerId::from_u64(peer_id.as_u64()), peer_id); assert_eq!(PeerId::from_u64(peer_id.as_u64()), peer_id);