oak-team/oak-gpui
3
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

clock: Cleanup ReplicaId, Lamport and Global (#40600)

Browse Source 
- Notable change is the use of a newtype for `ReplicaId`
- Fixes `WorktreeStore::create_remote_worktree` creating a remote
worktree with the local replica id, though this is not currently used
- Fixes observing the `Agent` (that is following the agent) causing
global clocks to allocate 65535 elements
- Shrinks the size of `Global` a bit. In a local or non-collab remote
session it won't ever allocate still.

Release Notes:

- N/A *or* Added/Fixed/Improved ...
This commit is contained in:
Lukas Wirth
2025-10-20 13:26:20 +02:00
committed by GitHub
parent 37e264ab99
commit 43a9368dff
33 changed files with 427 additions and 284 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/collab/src/db/queries/buffers.rs
+13 -9
View File
@@ -62,9 +62,9 @@ impl Database {
.iter()
.map(|c| c.replica_id)
.collect::<HashSet<_>>();
let mut replica_id = ReplicaId(0);
let mut replica_id = ReplicaId(clock::ReplicaId::FIRST_COLLAB_ID.as_u16() as i32);
while replica_ids.contains(&replica_id) {
replica_id.0 += 1;
replica_id = ReplicaId(replica_id.0 + 1);
}
let collaborator = channel_buffer_collaborator::ActiveModel {
channel_id: ActiveValue::Set(channel_id),
@@ -203,7 +203,7 @@ impl Database {
while let Some(row) = rows.next().await {
let row = row?;
let timestamp = clock::Lamport {
replica_id: row.replica_id as u16,
replica_id: clock::ReplicaId::new(row.replica_id as u16),
value: row.lamport_timestamp as u32,
};
server_version.observe(timestamp);
@@ -701,7 +701,11 @@ impl Database {
return Ok(());
}
let mut text_buffer = text::Buffer::new(0, text::BufferId::new(1).unwrap(), base_text);
let mut text_buffer = text::Buffer::new(
clock::ReplicaId::LOCAL,
text::BufferId::new(1).unwrap(),
base_text,
);
text_buffer.apply_ops(operations.into_iter().filter_map(operation_from_wire));
let base_text = text_buffer.text();
@@ -934,7 +938,7 @@ pub fn operation_from_wire(operation: proto::Operation) -> Option<text::Operatio
match operation.variant? {
proto::operation::Variant::Edit(edit) => Some(text::Operation::Edit(EditOperation {
timestamp: clock::Lamport {
replica_id: edit.replica_id as text::ReplicaId,
replica_id: clock::ReplicaId::new(edit.replica_id as u16),
value: edit.lamport_timestamp,
},
version: version_from_wire(&edit.version),
@@ -949,7 +953,7 @@ pub fn operation_from_wire(operation: proto::Operation) -> Option<text::Operatio
})),
proto::operation::Variant::Undo(undo) => Some(text::Operation::Undo(UndoOperation {
timestamp: clock::Lamport {
replica_id: undo.replica_id as text::ReplicaId,
replica_id: clock::ReplicaId::new(undo.replica_id as u16),
value: undo.lamport_timestamp,
},
version: version_from_wire(&undo.version),
@@ -959,7 +963,7 @@ pub fn operation_from_wire(operation: proto::Operation) -> Option<text::Operatio
.map(|c| {
(
clock::Lamport {
replica_id: c.replica_id as text::ReplicaId,
replica_id: clock::ReplicaId::new(c.replica_id as u16),
value: c.lamport_timestamp,
},
c.count,
@@ -975,7 +979,7 @@ fn version_from_wire(message: &[proto::VectorClockEntry]) -> clock::Global {
let mut version = clock::Global::new();
for entry in message {
version.observe(clock::Lamport {
replica_id: entry.replica_id as text::ReplicaId,
replica_id: clock::ReplicaId::new(entry.replica_id as u16),
value: entry.timestamp,
});
}
@@ -986,7 +990,7 @@ fn version_to_wire(version: &clock::Global) -> Vec<proto::VectorClockEntry> {
let mut message = Vec::new();
for entry in version.iter() {
message.push(proto::VectorClockEntry {
replica_id: entry.replica_id as u32,
replica_id: entry.replica_id.as_u16() as u32,
timestamp: entry.value,
});
}
crates/collab/src/db/queries/projects.rs
+7 -3
View File
@@ -91,14 +91,18 @@ impl Database {
.await?;
}
let replica_id = if is_ssh_project { 1 } else { 0 };
let replica_id = if is_ssh_project {
clock::ReplicaId::REMOTE_SERVER
} else {
clock::ReplicaId::LOCAL
};
project_collaborator::ActiveModel {
project_id: ActiveValue::set(project.id),
connection_id: ActiveValue::set(connection.id as i32),
connection_server_id: ActiveValue::set(ServerId(connection.owner_id as i32)),
user_id: ActiveValue::set(participant.user_id),
replica_id: ActiveValue::set(ReplicaId(replica_id)),
replica_id: ActiveValue::set(ReplicaId(replica_id.as_u16() as i32)),
is_host: ActiveValue::set(true),
id: ActiveValue::NotSet,
committer_name: ActiveValue::Set(None),
@@ -841,7 +845,7 @@ impl Database {
.iter()
.map(|c| c.replica_id)
.collect::<HashSet<_>>();
let mut replica_id = ReplicaId(1);
let mut replica_id = ReplicaId(clock::ReplicaId::FIRST_COLLAB_ID.as_u16() as i32);
while replica_ids.contains(&replica_id) {
replica_id.0 += 1;
}
crates/collab/src/db/tests/buffer_tests.rs
+33 -11
View File
@@ -1,7 +1,7 @@
use super::*;
use crate::test_both_dbs;
use language::proto::{self, serialize_version};
use text::Buffer;
use text::{Buffer, ReplicaId};
test_both_dbs!(
test_channel_buffers,
@@ -70,7 +70,11 @@ async fn test_channel_buffers(db: &Arc<Database>) {
.await
.unwrap();
let mut buffer_a = Buffer::new(0, text::BufferId::new(1).unwrap(), "".to_string());
let mut buffer_a = Buffer::new(
ReplicaId::new(0),
text::BufferId::new(1).unwrap(),
"".to_string(),
);
let operations = vec![
buffer_a.edit([(0..0, "hello world")]),
buffer_a.edit([(5..5, ", cruel")]),
@@ -95,7 +99,7 @@ async fn test_channel_buffers(db: &Arc<Database>) {
.unwrap();
let mut buffer_b = Buffer::new(
0,
ReplicaId::new(0),
text::BufferId::new(1).unwrap(),
buffer_response_b.base_text,
);
@@ -124,7 +128,7 @@ async fn test_channel_buffers(db: &Arc<Database>) {
rpc::proto::Collaborator {
user_id: a_id.to_proto(),
peer_id: Some(rpc::proto::PeerId { id: 1, owner_id }),
replica_id: 0,
replica_id: ReplicaId::FIRST_COLLAB_ID.as_u16() as u32,
is_host: false,
committer_name: None,
committer_email: None,
@@ -132,7 +136,7 @@ async fn test_channel_buffers(db: &Arc<Database>) {
rpc::proto::Collaborator {
user_id: b_id.to_proto(),
peer_id: Some(rpc::proto::PeerId { id: 2, owner_id }),
replica_id: 1,
replica_id: ReplicaId::FIRST_COLLAB_ID.as_u16() as u32 + 1,
is_host: false,
committer_name: None,
committer_email: None,
@@ -228,7 +232,8 @@ async fn test_channel_buffers_last_operations(db: &Database) {
.await
.unwrap();
db.join_channel_buffer(channel, user_id, connection_id)
let res = db
.join_channel_buffer(channel, user_id, connection_id)
.await
.unwrap();
@@ -239,7 +244,7 @@ async fn test_channel_buffers_last_operations(db: &Database) {
);
text_buffers.push(Buffer::new(
0,
ReplicaId::new(res.replica_id as u16),
text::BufferId::new(1).unwrap(),
"".to_string(),
));
@@ -276,7 +281,12 @@ async fn test_channel_buffers_last_operations(db: &Database) {
db.join_channel_buffer(buffers[1].channel_id, user_id, connection_id)
.await
.unwrap();
text_buffers[1] = Buffer::new(1, text::BufferId::new(1).unwrap(), "def".to_string());
let replica_id = text_buffers[1].replica_id();
text_buffers[1] = Buffer::new(
replica_id,
text::BufferId::new(1).unwrap(),
"def".to_string(),
);
update_buffer(
buffers[1].channel_id,
user_id,
@@ -304,20 +314,32 @@ async fn test_channel_buffers_last_operations(db: &Database) {
rpc::proto::ChannelBufferVersion {
channel_id: buffers[0].channel_id.to_proto(),
epoch: 0,
version: serialize_version(&text_buffers[0].version()),
version: serialize_version(&text_buffers[0].version())
.into_iter()
.filter(
|vector| vector.replica_id == text_buffers[0].replica_id().as_u16() as u32
)
.collect::<Vec<_>>(),
},
rpc::proto::ChannelBufferVersion {
channel_id: buffers[1].channel_id.to_proto(),
epoch: 1,
version: serialize_version(&text_buffers[1].version())
.into_iter()
.filter(|vector| vector.replica_id == text_buffers[1].replica_id() as u32)
.filter(
|vector| vector.replica_id == text_buffers[1].replica_id().as_u16() as u32
)
.collect::<Vec<_>>(),
},
rpc::proto::ChannelBufferVersion {
channel_id: buffers[2].channel_id.to_proto(),
epoch: 0,
version: serialize_version(&text_buffers[2].version()),
version: serialize_version(&text_buffers[2].version())
.into_iter()
.filter(
|vector| vector.replica_id == text_buffers[2].replica_id().as_u16() as u32
)
.collect::<Vec<_>>(),
},
]
);