chore: don't try to prune empty redis stream

This commit is contained in:
Bartosz Sypytkowski 2024-10-17 10:34:48 +02:00
parent 13d657861a
commit 15f438ea05
2 changed files with 25 additions and 13 deletions

View File

@ -167,14 +167,18 @@ impl CollabRedisStream {
.into_iter() .into_iter()
.map(|stream_id| stream_id.id) .map(|stream_id| stream_id.id)
.collect(); .collect();
let count: usize = conn.xdel(stream_key, &msg_ids).await?; if !msg_ids.is_empty() {
drop(conn); let count: usize = conn.xdel(stream_key, &msg_ids).await?;
tracing::debug!( drop(conn);
"Pruned redis stream `{}` <= `{}` ({} objects)", tracing::debug!(
stream_key, "Pruned redis stream `{}` <= `{}` ({} objects)",
message_id, stream_key,
count message_id,
); count
Ok(count) );
Ok(count)
} else {
Ok(0)
}
} }
} }

View File

@ -612,26 +612,34 @@ impl CollabGroup {
if let Ok(sv) = state.state_vector.try_read() { if let Ok(sv) = state.state_vector.try_read() {
// we optimistically try to obtain state vector lock for a fast track: // we optimistically try to obtain state vector lock for a fast track:
// if we remote sv is up-to-date with current one, we don't need to do anything // if we remote sv is up-to-date with current one, we don't need to do anything
if remote_sv == &*sv { match sv.partial_cmp(&remote_sv) {
tracing::trace!("remote client is up to date - nothing to send"); Some(std::cmp::Ordering::Equal) => return Ok(None), // client and server are in sync
return Ok(None); Some(std::cmp::Ordering::Less) => {
// server is behind client
let msg = Message::Sync(SyncMessage::SyncStep1(sv.clone()));
return Ok(Some(msg.encode_v1()));
},
Some(std::cmp::Ordering::Greater) | None => { /* server has some new updates */ },
} }
} }
// we need to reconstruct document state on the server side
let snapshot = state let snapshot = state
.persister .persister
.load() .load()
.await .await
.map_err(|err| RTProtocolError::Internal(err.into()))?; .map_err(|err| RTProtocolError::Internal(err.into()))?;
// prepare document state update and state vector
let tx = snapshot.collab.transact(); let tx = snapshot.collab.transact();
let doc_state = tx.encode_state_as_update_v1(remote_sv); let doc_state = tx.encode_state_as_update_v1(remote_sv);
let local_sv = tx.state_vector(); let local_sv = tx.state_vector();
drop(tx); drop(tx);
// Retrieve the latest document state from the client after they return online from offline editing. // Retrieve the latest document state from the client after they return online from offline editing.
tracing::trace!("sending missing data to client ({} bytes)", doc_state.len()); tracing::trace!("sending missing data to client ({} bytes)", doc_state.len());
let mut encoder = EncoderV1::new(); let mut encoder = EncoderV1::new();
Message::Sync(SyncMessage::SyncStep2(doc_state)).encode(&mut encoder); Message::Sync(SyncMessage::SyncStep2(doc_state)).encode(&mut encoder);
//FIXME: this should never happen as response to sync step 1 from the client, but rather be //FIXME: this should never happen as response to sync step 1 from the client, but rather be
// send when a connection is established // send when a connection is established
Message::Sync(SyncMessage::SyncStep1(local_sv)).encode(&mut encoder); Message::Sync(SyncMessage::SyncStep1(local_sv)).encode(&mut encoder);