Repository: qdrant/qdrant
File: lib/storage/src/content_manager/consensus_manager.rs (Rust)

Code before:
        propose_sender: OperationSender,
        storage_path: &Path,
    ) -> Self {
        let mut wal = ConsensusOpWal::new(storage_path);

        // If WAL has more entries than we have applied, truncate them and resync
        if let Ok(Some(last)) = wal.last_entry() {
            let last_committed = persistent_state.state().hard_state.commit;
            let last_wal_commit_applied =
                last_committed.saturating_sub(persistent_state.latest_snapshot_meta.index);
            let last_wal_commit = last.index;

            debug_assert!(
                last_wal_commit >= last_wal_commit_applied,
                "consensus WAL is missing entries, last committed WAL index is {last_wal_commit} but WAL only goes up to {last_wal_commit_applied}",
            );

            let extra_entries = last_wal_commit.saturating_sub(last_wal_commit_applied);
            if extra_entries > 0 {
                log::warn!(
                    "Consensus WAL has {extra_entries} unapplied entries, truncating from index {last_wal_commit_applied} onwards"
                );
                wal.truncate(last_wal_commit_applied)
                    .expect("Failed to truncate WAL on startup");
            }
        }

        Self {
            persistent: RwLock::new(persistent_state),
            is_leader_established: Arc::new(IsReady::default()),
            wal: Mutex::new(wal),
            soft_state: RwLock::new(None),
            toc,
            on_consensus_op_apply: Default::default(),
            propose_sender,
            consensus_thread_status: RwLock::new(ConsensusThreadStatus::Working {
                last_update: Utc::now(),
            }),
            message_send_failures: Default::default(),
            next_peer_metadata_update_attempt: Mutex::new(Instant::now()),
        }
    }

    pub fn report_snapshot(
        &self,
        peer_id: u64,
        status: impl Into<SnapshotStatus>,
    ) -> Result<(), StorageError> {
        self.propose_sender
            .send(ConsensusOperations::report_snapshot(peer_id, status))
            .map_err(|_err| {

Review comment (bug):
I'd say it's wrong to rely on `hard_state.commit`. We need to rely on `latest_snapshot_metadata.index`.

Code after:
        propose_sender: OperationSender,
        storage_path: &Path,
    ) -> Result<Self, StorageError> {
        let mut wal = ConsensusOpWal::new(storage_path);

        // When our Raft index and last snapshot index match, the last thing we did is apply a Raft
        // snapshot. It is possible that we crashed before clearing the WAL, so we still do it now.
        // Specifically, if the last operation was applying a snapshot and our WAL does still have
        // older Raft entries, we clear the whole WAL. Consensus will take care of us catching up
        // with the rest.
        // See `apply_snapshot` function and <https://github.com/qdrant/qdrant/pull/7577>.
        let raft_index = persistent_state.state().hard_state.commit;
        let snapshot_index = persistent_state.latest_snapshot_meta.index;
        let last_operation_was_snapshot = raft_index == persistent_state.latest_snapshot_meta.index;
        if last_operation_was_snapshot
            && let Ok(Some(last)) = wal.last_entry()
            && last.index < snapshot_index
        {
            log::warn!(
                "Consensus WAL was not cleared after applying consensus snapshot, clearing it now"
            );
            wal.clear()?;
        }

        Ok(Self {
            persistent: RwLock::new(persistent_state),
            is_leader_established: Arc::new(IsReady::default()),
            wal: Mutex::new(wal),
            soft_state: RwLock::new(None),
            toc,
            on_consensus_op_apply: Default::default(),
            propose_sender,
            consensus_thread_status: RwLock::new(ConsensusThreadStatus::Working {
                last_update: Utc::now(),
            }),
            message_send_failures: Default::default(),
            next_peer_metadata_update_attempt: Mutex::new(Instant::now()),
        })
    }

    pub fn report_snapshot(
        &self,
        peer_id: u64,
        status: impl Into<SnapshotStatus>,
    ) -> Result<(), StorageError> {
        self.propose_sender
            .send(ConsensusOperations::report_snapshot(peer_id, status))
            .map_err(|_err| {
                StorageError::service_error(
                    "failed to send ReportSnapshot message to consensus thread",
                )