Method apply_record

nodedb/src/engine/array/recovery.rs:73–120

(&mut self, rec: RecoveryRecord)

    }

    pub fn apply_record(&mut self, rec: RecoveryRecord) -> Result<(), RecoveryError> {
        let durable = self.store.manifest().durable_lsn;
        match rec {
            RecoveryRecord::Put { lsn, payload } => {
                if lsn <= durable {
                    self.stats.puts_skipped += 1;
                    return Ok(());
                }
                let cells: Vec<ArrayPutCell> = payload.cells;
                // Invariant: every cell's `system_from_ms` must be the
                // leader-stamped value from the WAL payload. The recovery
                // path MUST NOT call `HlcClock::now()` or any wall-clock
                // function to derive a replacement — that would cause the
                // replay to assign a different stamp than the leader did,
                // breaking cross-replica consistency at any `system_as_of`
                // cutoff that falls between the two stamps.
                debug_assert!(
                    cells.iter().all(|c| c.system_from_ms > 0),
                    "recovery: ArrayPutCell.system_from_ms must be the leader-stamped \
                     value (> 0), not a default — check that the WAL decoder does not \
                     zero-fill this field on version mismatch"
                );
                stamp_put_cells(self.store, cells, lsn)?;
                self.stats.puts_applied += 1;
            }
            RecoveryRecord::Delete { lsn, payload } => {
                if lsn <= durable {
                    self.stats.deletes_skipped += 1;
                    return Ok(());
                }
                // Same invariant for deletes: `system_from_ms` is the
                // tombstone version key; it must come from the WAL payload,
                // not from the local clock.
                debug_assert!(
                    payload.cells.iter().all(|c| c.system_from_ms > 0),
                    "recovery: ArrayDeleteCell.system_from_ms must be the leader-stamped \
                     value (> 0), not a default"
                );
                stamp_delete_cells(self.store, payload.cells, lsn)?;
                self.stats.deletes_applied += 1;
            }
            RecoveryRecord::Flush { lsn, .. } => {
                let m = self.store.manifest_mut();
                m.durable_lsn = m.durable_lsn.max(lsn);
            }
        }
        Ok(())
    }
}