Repository: qdrant/qdrant
File: lib/collection/src/shards/local_shard/wal_ops.rs (Rust)

Code before:
        let update_handler = self.update_handler.lock().await;

        // Next, clear the update worker channel.
        // To do this, set the skip updates flag. The guard resets it when dropped.
        // New updates won't be skipped because the update_lock is held until we're done.
        let _skip_updates_guard = SkipUpdatesGuard::new(update_handler.skip_updates.clone());

        // Then, Send the plunger signal to the update handler.
        // It a marker that all previous updates are processed or skipped.
        let (tx, rx) = oneshot::channel();
        let plunger = UpdateSignal::Plunger(tx);
        self.update_sender.load().send(plunger).await?;
        rx.await?;

        // The update worker is now idle, and no new updates are being processed.
        // It's safe to lock and drop WAL now.
        let mut wal_lock = Mutex::lock_owned(self.wal.wal.clone()).await;
        let last_wal_op_num = wal_lock.last_index();

        let applied_seq = update_handler.applied_seq();
        let applied_seq_num = applied_seq.op_num().unwrap_or(last_wal_op_num);

        // `applied_seq_num` is persisted with `APPLIED_SEQ_SAVE_INTERVAL` step.
        // So WAL can be dropped from `applied_seq_num + APPLIED_SEQ_SAVE_INTERVAL`.
        // Add 1 also because the last applied record must stay in the WAL.
        let safe_drop_seq_num = applied_seq_num + APPLIED_SEQ_SAVE_INTERVAL + 1;

        let removed_records = if safe_drop_seq_num < last_wal_op_num {
            wal_lock.drop_from(safe_drop_seq_num)?;
            wal_lock.flush()?;

            // To calculate removed records, add 1 because both `last_wal_op_num` and `safe_drop_seq_num` are inclusive.
            (last_wal_op_num - safe_drop_seq_num + 1) as usize
        } else {
            0
        };

        Ok(removed_records)
    }
}


Review comment (refactor):
This is problematic.

As far as I can see we:
- drain the full channel
- drain the same **or less** entries from the WAL

If that is true, we must submit the difference of those two into the channel again. We currently don't, which means we'll be missing operations that remain persisted.

I again strongly prefer not to rely on `applied_seq` here:
- the `applied_seq` is an extra abstraction that hides details
- we can grab the correct `op_num` from the channel/WAL itself
- we don't have to handle `APPLIED_SEQ_SAVE_INTERVAL`
- we eliminate a potential future foot-gun of the applied seq not being exactly in sync

Let me give a quick attempt to implement this idea. 👍 

Code after:

        // Next, clear the update worker channel.
        // To do this, set the skip updates flag. The guard resets it when dropped.
        // New updates won't be skipped because the update_lock is held until we're done.
        let _skip_updates_guard = SkipUpdatesGuard::new(update_handler.skip_updates.clone());

        // Then, Send the plunger signal to the update handler.
        // It's a marker that all previous updates are processed or skipped.
        let (tx, rx) = oneshot::channel();
        let plunger = UpdateSignal::Plunger(tx);
        self.update_sender.load().send(plunger).await?;
        let truncate_from_op_num = rx.await?;

        // The update worker is now idle, and no new updates are being processed.
        // It's safe to lock and drop WAL now.
        let mut wal_lock = Mutex::lock_owned(self.wal.wal.clone()).await;
        let last_wal_op_num = wal_lock.last_index();

        let Some(truncate_from_op_num) = truncate_from_op_num else {
            return Ok(0);
        };

        // A corner case: truncate was called twice in a row without new updates in between.
        // In this case, `truncate_from_op_num` is equal to last_wal_op_num + 1`.
        // Because previously unapplied WAL records were already truncated up to `truncate_from_op_num` inclusive.
        if truncate_from_op_num == last_wal_op_num + 1 {
            return Ok(0);
        }

        debug_assert!(truncate_from_op_num <= last_wal_op_num);

        wal_lock.drop_from(truncate_from_op_num)?;
        wal_lock.flush()?;

        // To calculate removed records, add 1 because both `last_wal_op_num` and `truncate_from_op_num` are inclusive.
        let removed_records = (last_wal_op_num + 1 - truncate_from_op_num) as usize;

        Ok(removed_records)
    }
}