use super::*;
#[cfg(test)]
#[path = "graph_recovery/checkpoint_capture_shadow_tests.rs"]
mod checkpoint_capture_shadow_tests;
#[cfg(test)]
#[path = "graph_recovery/checkpoint_capture_projection_tests.rs"]
mod checkpoint_capture_projection_tests;
#[cfg(test)]
#[path = "graph_recovery/checkpoint_capture_metadata_tests.rs"]
mod checkpoint_capture_metadata_tests;
#[cfg(test)]
#[path = "graph_recovery/checkpoint_delta_tests.rs"]
mod checkpoint_delta_tests;
#[cfg(test)]
#[path = "graph_recovery/checkpoint_delta_related_tests.rs"]
mod checkpoint_delta_related_tests;
#[cfg(test)]
#[path = "graph_recovery/checkpoint_native_read_tests.rs"]
mod checkpoint_native_read_tests;
#[cfg(test)]
#[path = "graph_recovery/checkpoint_native_read_related_tests.rs"]
mod checkpoint_native_read_related_tests;
#[cfg(test)]
#[path = "graph_recovery/checkpoint_native_decode_tests.rs"]
mod checkpoint_native_decode_tests;
#[cfg(test)]
#[path = "graph_recovery/checkpoint_native_decode_related_tests.rs"]
mod checkpoint_native_decode_related_tests;
impl GraphStore {
pub fn import_graph_snapshot_rows(
&mut self,
catalog: &mut Catalog,
nodes: Vec<GraphSnapshotNodeImport>,
relationships: Vec<GraphSnapshotRelationshipImport>,
) -> Result<()> {
if self.basic_statistics.node_count != 0 || self.basic_statistics.relationship_count != 0 {
return Err(HawDBError::Storage(
"HawDB Lightning initial import requires an empty target graph".to_string(),
));
}
let mut node_ids = BTreeSet::new();
for (id, label, _) in &nodes {
if label.is_empty() {
return Err(HawDBError::Storage(
"HawDB Lightning initial import node label is empty".to_string(),
));
}
if !node_ids.insert(*id) {
return Err(HawDBError::Storage(format!(
"HawDB Lightning initial import duplicate node id {}",
id.0
)));
}
}
let mut relationship_ids = BTreeSet::new();
for (id, source, target, rel_type, _) in &relationships {
if rel_type.is_empty() {
return Err(HawDBError::Storage(
"HawDB Lightning initial import relationship type is empty".to_string(),
));
}
if !relationship_ids.insert(*id) {
return Err(HawDBError::Storage(format!(
"HawDB Lightning initial import duplicate relationship id {}",
id.0
)));
}
if !node_ids.contains(source) {
return Err(HawDBError::Storage(format!(
"HawDB Lightning initial import relationship {} references missing source node {}",
id.0, source.0
)));
}
if !node_ids.contains(target) {
return Err(HawDBError::Storage(format!(
"HawDB Lightning initial import relationship {} references missing target node {}",
id.0, target.0
)));
}
}
if nodes.is_empty() && relationships.is_empty() {
return Ok(());
}
let mut working_catalog = catalog.clone();
for (_, label, _) in &nodes {
working_catalog.get_or_create_label(label);
}
for (_, _, _, rel_type, _) in &relationships {
working_catalog.get_or_create_rel_type(rel_type);
}
let mut ops = Vec::with_capacity(nodes.len() + relationships.len());
ops.extend(
nodes
.iter()
.map(|(id, label, properties)| WalOp::CreateNode {
id: *id,
label: label.clone(),
properties: properties.clone(),
}),
);
ops.extend(
relationships
.iter()
.map(
|(id, source, target, rel_type, properties)| WalOp::CreateRelationship {
id: *id,
source: *source,
target: *target,
rel_type: rel_type.clone(),
properties: properties.clone(),
},
),
);
self.validate_constraints_for_ops(&working_catalog, &ops)?;
self.append_graph_wal_and_capture(&working_catalog, &ops, false)?;
for op in ops {
self.apply_wal_op(catalog, op)?;
}
self.finish_non_relational_commit();
Ok(())
}
#[doc(hidden)]
pub fn import_hawdb_snapshot_rows_with_source_fingerprint(
&mut self,
catalog: &mut Catalog,
import: HawDBSnapshotRowsImport,
) -> Result<()> {
let HawDBSnapshotRowsImport {
stable_id_mapping,
source_fingerprint,
nodes,
relationships,
relational_state,
target_has_only_engine_bootstrap,
} = import;
if self.initial_import_source_fingerprint.is_some()
|| !self.nodes.is_empty()
|| !self.relationships.is_empty()
|| (!self.relational_state.is_empty() && !target_has_only_engine_bootstrap)
|| !catalog.is_empty()
{
return Err(HawDBError::Storage(
"hawdb lightning initial import requires an empty target database".to_string(),
));
}
let mut node_ids = BTreeSet::new();
for (id, label, _) in &nodes {
if label.is_empty() {
return Err(HawDBError::Storage(
"hawdb lightning initial import node label is empty".to_string(),
));
}
if !node_ids.insert(*id) {
return Err(HawDBError::Storage(format!(
"hawdb lightning initial import duplicate node id {}",
id.0
)));
}
}
let mut relationship_ids = BTreeSet::new();
for (id, source, target, rel_type, _) in &relationships {
if rel_type.is_empty() {
return Err(HawDBError::Storage(
"hawdb lightning initial import relationship type is empty".to_string(),
));
}
if !relationship_ids.insert(*id) {
return Err(HawDBError::Storage(format!(
"hawdb lightning initial import duplicate relationship id {}",
id.0
)));
}
if !node_ids.contains(source) || !node_ids.contains(target) {
return Err(HawDBError::Storage(format!(
"hawdb lightning initial import relationship {} references a missing endpoint",
id.0
)));
}
}
let mut working_catalog = catalog.clone();
for (_, label, _) in &nodes {
working_catalog.get_or_create_label(label);
}
for (_, _, _, rel_type, _) in &relationships {
working_catalog.get_or_create_rel_type(rel_type);
}
let target_commit_epoch = self
.commit_epoch
.checked_add(1)
.ok_or_else(|| HawDBError::Storage("commit epoch overflow".to_string()))?;
let relational_record =
encode_relational_checkpoint(target_commit_epoch, &relational_state)
.map_err(HawDBError::from_storage_error)?;
let mut ops = Vec::with_capacity(nodes.len() + relationships.len() + 2);
ops.push(WalOp::MarkInitialImportSource { source_fingerprint });
ops.push(WalOp::RelationalSnapshot {
record: Arc::from(relational_record),
});
ops.extend(
nodes
.iter()
.map(|(id, label, properties)| WalOp::CreateNode {
id: *id,
label: label.clone(),
properties: properties.clone(),
}),
);
ops.extend(
relationships
.iter()
.map(
|(id, source, target, rel_type, properties)| WalOp::CreateRelationship {
id: *id,
source: *source,
target: *target,
rel_type: rel_type.clone(),
properties: properties.clone(),
},
),
);
self.validate_constraints_for_ops(&working_catalog, &ops)?;
let staged_projection_change = self.prepare_search_projection_changes_for_ops(
&working_catalog,
target_commit_epoch,
&ops,
None,
)?;
self.replace_stable_id_mapping_for_epoch(stable_id_mapping, target_commit_epoch)?;
self.append_durable_wal_batch(&ops)?;
self.install_search_projection_change(staged_projection_change);
for op in ops {
self.apply_wal_op(catalog, op)?;
}
self.finish_non_relational_commit();
Ok(())
}
pub(super) fn load_source_scan_manifest(&mut self) -> Result<()> {
let Some(durable) = &self.durable else {
return Ok(());
};
self.source_scan_manifest = durable.load_source_scan_manifest(self.commit_epoch)?.into();
Ok(())
}
pub(super) fn load_projected_graph_artifacts(&mut self) -> Result<()> {
if self.canonical_base_out_of_core {
self.projected_graph_artifacts = BTreeMap::new().into();
return Ok(());
}
let Some(durable) = &self.durable else {
return Ok(());
};
self.projected_graph_artifacts = durable
.load_projected_graph_artifacts()?
.into_iter()
.filter(|(name, artifact)| {
artifact.commit_epoch == self.commit_epoch
&& self
.projected_graphs
.get(name)
.is_some_and(|definition| definition == &artifact.definition)
})
.collect::<BTreeMap<_, _>>()
.into();
Ok(())
}
pub(super) fn load_stable_id_mapping(&mut self) -> Result<()> {
let Some(durable) = &mut self.durable else {
return Ok(());
};
durable.load_stable_id_mapping()?;
self.stable_id_mapping = CowSegment::default();
Ok(())
}
pub(super) fn load_checkpoint(
&mut self,
catalog: &mut Catalog,
config: WalReplayConfig,
) -> Result<()> {
let Some(durable) = &self.durable else {
return Ok(());
};
if !durable.checkpoint_path.exists() {
if durable.checkpoint_commit_epoch != 0 {
return Err(HawDBError::Storage(format!(
"manifest checkpoint generation {} is missing",
durable.checkpoint_epoch
)));
}
return Ok(());
}
let expected_generation = durable.checkpoint_epoch;
let expected_commit_epoch = durable.checkpoint_commit_epoch;
let durable_root_path = durable.root_path.clone();
let text = durable.read_checkpoint_text(config)?;
let (body, checksum) = split_checkpoint_checksum(&text)?;
let actual = checksum_bytes(body.as_bytes());
if checksum != actual {
return Err(HawDBError::Storage(format!(
"checkpoint checksum mismatch: expected {checksum}, got {actual}"
)));
}
let relational_checkpoint = relational_checkpoint_metadata(body)?;
let canonical_relational_manifest = if config.residency_mode
== StorageResidencyMode::OutOfCore
&& config
.relational_index_mode
.requires_authoritative_indexes()
{
let overflow = durable.open_bound_relational_overflow()?;
Some(
durable
.open_bound_relational_row_pages(&overflow)?
.manifest()
.clone(),
)
} else {
None
};
if relational_checkpoint.is_none() && canonical_relational_manifest.is_none() {
let overflow = durable.open_bound_relational_overflow()?;
let rows = durable.open_bound_relational_row_pages(&overflow)?;
if !rows.manifest().tables.is_empty() {
return Err(HawDBError::Storage(
"checkpoint stores canonical metadata-only relational rows; reopen requires OutOfCore residency with Authoritative relational indexes"
.to_string(),
));
}
}
let mut decoded = hawdb_storage::checkpoint::DecodedCheckpoint::default();
hawdb_storage::checkpoint::parse_checkpoint(body, catalog, &mut decoded)?;
self.next_node_id = decoded.next_node_id;
self.next_rel_id = decoded.next_rel_id;
if let Some(commit_epoch) = decoded.commit_epoch {
self.commit_epoch = commit_epoch;
}
if let Some(start_epoch) = decoded.search_projection_change_log_start_epoch {
self.search_projection_change_log_start_epoch = start_epoch;
}
self.search_projection_change_log_retained_bytes =
decoded.search_projection_change_log_retained_bytes;
self.search_projection_database_identity = decoded.search_projection_database_identity;
self.initial_import_source_fingerprint = decoded.initial_import_source_fingerprint;
self.search_projection_graph_changes = decoded
.search_projection_graph_changes
.into_iter()
.map(Arc::new)
.collect();
self.basic_statistics = decoded.basic_statistics.into();
self.checkpoint_statistics = decoded.checkpoint_statistics.into();
for (name, definition) in decoded.projected_graphs {
self.apply_project_graph_definition(name, definition)?;
}
if !decoded.nodes.is_empty() || !decoded.relationships.is_empty() {
self.basic_statistics = BasicStatisticsState::default();
}
for node in decoded.nodes {
self.apply_create_node_with_labels(catalog, node.id, node.labels, node.properties);
}
for relationship in decoded.relationships {
self.apply_create_relationship(
relationship.id,
relationship.source,
relationship.target,
relationship.rel_type,
relationship.properties,
);
}
let loaded_search_projection_change_log_start_epoch =
decoded.search_projection_change_log_start_epoch;
let loaded_generation = decoded.generation;
let loaded_commit_epoch = decoded.commit_epoch;
let loaded_statistics_complete = decoded.statistics_complete;
let saw_checkpoint_statistics = decoded.saw_checkpoint_statistics;
let canonical_records = decoded.canonical_records;
if saw_checkpoint_statistics {
self.checkpoint_statistics.advanced_statistics_complete =
loaded_statistics_complete.unwrap_or(true);
retain_supported_property_statistics(&mut self.checkpoint_statistics, Some(catalog));
self.checkpoint_statistics
.retain_valid_index_samples(catalog);
}
if loaded_generation != Some(expected_generation) {
return Err(HawDBError::Storage(format!(
"checkpoint generation {:?} does not match manifest generation {expected_generation}",
loaded_generation
)));
}
if loaded_commit_epoch != Some(expected_commit_epoch) {
return Err(HawDBError::Storage(format!(
"checkpoint commit epoch {:?} does not match manifest commit epoch {expected_commit_epoch}",
loaded_commit_epoch
)));
}
if let Some(metadata) = relational_checkpoint {
let path =
durable_root_path.join(relational_checkpoint_generation_file(expected_generation));
let max_bytes = RelationalDecodeLimits::checkpoint().max_record_bytes;
let file_len = fs::metadata(&path)?.len();
if file_len > max_bytes as u64 {
return Err(HawDBError::Storage(format!(
"relational checkpoint contains {file_len} bytes, exceeding max_record_bytes {max_bytes}"
)));
}
verify_file_integrity(
&path,
metadata.encoded_len,
metadata.encoded_checksum,
metadata.encoded_sha256,
"relational checkpoint",
)?;
if canonical_relational_manifest.is_none() {
let index_load = if config
.relational_index_mode
.requires_authoritative_indexes()
{
RelationalCheckpointIndexLoad::OmitMaterializedPostings
} else {
RelationalCheckpointIndexLoad::MaterializedPostings
};
let checkpoint = decode_relational_checkpoint_file_with_index_load(
&path,
RelationalDecodeLimits::checkpoint(),
index_load,
)
.map_err(HawDBError::from_storage_error)?;
if checkpoint.epoch != self.commit_epoch {
return Err(HawDBError::Storage(format!(
"relational checkpoint epoch {} does not match graph commit epoch {}",
checkpoint.epoch, self.commit_epoch
)));
}
self.relational_state = checkpoint.state;
}
}
if let Some(row_root) = canonical_relational_manifest.as_ref() {
if row_root.source_commit_epoch != self.commit_epoch {
return Err(HawDBError::Storage(format!(
"relational row root epoch {} does not match graph commit epoch {}",
row_root.source_commit_epoch, self.commit_epoch
)));
}
self.relational_state = RelationalState::from_canonical_row_root(row_root)
.map_err(HawDBError::from_storage_error)?;
}
match loaded_search_projection_change_log_start_epoch {
Some(start_epoch) => {
validate_search_projection_checkpoint_changes(
start_epoch,
self.commit_epoch,
self.search_projection_graph_changes.iter().map(Arc::as_ref),
)?;
}
None => {
return Err(HawDBError::Storage(
"checkpoint search projection changes are missing their start epoch"
.to_string(),
));
}
}
if canonical_records {
let reader = self
.durable
.as_ref()
.and_then(|durable| durable.canonical_segments.clone())
.ok_or_else(|| {
HawDBError::Storage(
"checkpoint delegates records to missing canonical segments".to_string(),
)
})?;
let materialize = match config.residency_mode {
StorageResidencyMode::Materialized => true,
StorageResidencyMode::OutOfCore => false,
StorageResidencyMode::Auto => {
reader.manifest().artifact_len <= config.auto_materialize_checkpoint_bytes
}
};
if materialize {
self.basic_statistics = BasicStatisticsState::default();
reader
.scan_nodes(|node| {
self.apply_create_node_with_labels(
catalog,
node.id,
node.labels,
node.properties,
);
Ok(())
})
.map_err(HawDBError::from_storage_error)?;
reader
.scan_relationships(|relationship| {
self.apply_create_relationship(
relationship.id,
relationship.source,
relationship.target,
relationship.rel_type,
relationship.properties,
);
Ok(())
})
.map_err(HawDBError::from_storage_error)?;
} else {
self.basic_statistics.node_count = reader.manifest().node_count;
self.basic_statistics.relationship_count = reader.manifest().relationship_count;
self.canonical_base = Some(reader);
self.canonical_adjacency = self
.durable
.as_ref()
.and_then(|durable| durable.canonical_adjacency.clone());
self.persistent_property_projection = self
.durable
.as_ref()
.and_then(|durable| durable.persistent_property_projection.clone());
self.canonical_base_out_of_core = true;
}
}
if let Some(durable) = &mut self.durable {
durable.relational_checkpoint_encoded_len =
relational_checkpoint.map(|metadata| metadata.encoded_len);
durable.relational_checkpoint_encoded_checksum =
relational_checkpoint.map(|metadata| metadata.encoded_checksum);
durable.relational_checkpoint_encoded_sha256 =
relational_checkpoint.map(|metadata| metadata.encoded_sha256);
}
Ok(())
}
pub(super) fn replay_wal(
&mut self,
catalog: &mut Catalog,
config: WalReplayConfig,
) -> Result<StorageRecoveryReport> {
let Some(durable) = &self.durable else {
return Ok(StorageRecoveryReport::default());
};
let mut source = RelationalRecoverySourceBuilder::new(
durable.wal_generation,
durable.wal_replay_start_lsn,
);
let report = self.replay_wal_interval(catalog, config, &mut source)?;
self.finish_wal_recovery(source, report.replayed_wal_entries)?;
Ok(report)
}
pub(super) fn finish_wal_recovery(
&mut self,
source: RelationalRecoverySourceBuilder,
replayed_entries: usize,
) -> Result<()> {
if let Some(durable) = &mut self.durable
&& self.commit_epoch > durable.checkpoint_commit_epoch
{
durable
.wal_uncheckpointed_since
.get_or_insert_with(std::time::Instant::now);
}
let identity = if replayed_entries == 0 {
None
} else {
Some(
source
.finish()
.map_err(|reason| HawDBError::Storage(reason.to_string()))?,
)
};
self.finish_relational_row_page_recovery(identity);
self.finish_relational_index_recovery(identity);
Ok(())
}
pub(super) fn finish_private_wal_recovery(
&mut self,
source: &RelationalRecoverySourceBuilder,
replayed_entries: usize,
work: &crate::background::CheckpointWorkContext,
) -> Result<[Option<crate::relational::PreparedRelationalRecoverySelector>; 2]> {
let identity = if replayed_entries == 0 {
None
} else {
Some(
source
.prefix_identity()
.map_err(|reason| HawDBError::Storage(reason.to_string()))?,
)
};
let row = self.finish_private_relational_row_page_recovery(identity, work)?;
let index = self.finish_private_relational_index_recovery(identity, work)?;
Ok([row, index])
}
pub(super) fn replay_wal_interval(
&mut self,
catalog: &mut Catalog,
config: WalReplayConfig,
relational_recovery_source: &mut RelationalRecoverySourceBuilder,
) -> Result<StorageRecoveryReport> {
self.replay_wal_interval_with_file(catalog, config, relational_recovery_source, &mut None)
}
pub(super) fn replay_wal_interval_with_file(
&mut self,
catalog: &mut Catalog,
config: WalReplayConfig,
relational_recovery_source: &mut RelationalRecoverySourceBuilder,
captured_file: &mut Option<fs::File>,
) -> Result<StorageRecoveryReport> {
let captured = captured_file.is_some();
let Some(durable) = &self.durable else {
return Ok(StorageRecoveryReport::default());
};
let wal_path = durable.wal_path.clone();
let checkpoint_epoch = durable.checkpoint_epoch;
let checkpoint_commit_epoch = durable.checkpoint_commit_epoch;
let wal_replay_start_lsn = durable.wal_replay_start_lsn;
let wal_generation = durable.wal_generation;
let read_only = durable.read_only;
let mut tail_repair = durable.wal_tail_repair.clone();
let mut torn_tail_reason = tail_repair
.as_ref()
.map(|_| "resumed interrupted repair of an incomplete final WAL record".to_string());
let checkpoint_present = durable.checkpoint_path.exists();
let mut replayed_entries = 0_usize;
let mut replayed_bytes = 0u64;
let wal_present = captured || wal_path.exists();
if !wal_present {
if checkpoint_epoch > 0 {
return Err(HawDBError::Storage(format!(
"manifest WAL generation {wal_generation} is missing"
)));
}
return Ok(StorageRecoveryReport {
open_timings: Default::default(),
durable: true,
recovery_mode: config.recovery_mode,
max_wal_replay_entries: config.max_entries,
max_wal_replay_bytes: config.max_bytes,
max_wal_record_bytes: config.max_record_bytes,
checkpoint_epoch: checkpoint_present.then_some(checkpoint_epoch),
checkpoint_commit_epoch: checkpoint_present.then_some(checkpoint_commit_epoch),
wal_present,
wal_generation: Some(wal_generation),
wal_replay_start_lsn: Some(wal_replay_start_lsn),
next_lsn_after_replay: Some(wal_replay_start_lsn),
replayed_wal_entries: replayed_entries,
replayed_wal_bytes: replayed_bytes,
torn_tail_ignored: false,
torn_tail_repaired: false,
discarded_wal_tail_bytes: 0,
torn_tail_reason: None,
recovered_commit_epoch: self.commit_epoch,
});
}
let wal_len = match captured_file.as_ref() {
Some(file) => file.metadata()?.len(),
None => fs::metadata(&wal_path)?.len(),
};
if config.max_bytes.is_some_and(|limit| wal_len > limit) {
return Err(HawDBError::Storage(format!(
"WAL replay byte limit exceeded: max_wal_replay_bytes={}",
config.max_bytes.unwrap_or_default()
)));
}
let opened = match captured_file.take() {
Some(file) => WalRecordCursor::from_file(file, config.max_record_bytes)?,
None => WalRecordCursor::open(&wal_path, config.max_record_bytes)?,
};
let cursor = match opened {
WalOpenOutcome::Cursor(cursor) => cursor,
WalOpenOutcome::MissingHeader => {
return Err(HawDBError::Storage(format!(
"WAL generation {wal_generation} is missing its header"
)));
}
WalOpenOutcome::HeaderTorn { reason } => {
return Err(HawDBError::Storage(reason));
}
WalOpenOutcome::HeaderCorrupt { reason } => {
return reject_wal_source_corruption(
captured,
&wal_path,
wal_generation,
read_only,
config.max_quarantine_bytes,
0,
reason,
);
}
};
if cursor.generation() != wal_generation || cursor.start_lsn() != wal_replay_start_lsn {
return Err(HawDBError::Storage(format!(
"WAL header generation/start ({}, {}) does not match manifest ({wal_generation}, {wal_replay_start_lsn})",
cursor.generation(),
cursor.start_lsn()
)));
}
let mut cursor = Some(cursor);
let mut expected_lsn = wal_replay_start_lsn;
loop {
let (entry, record_start, record_encoded_len, payload_len, payload_sha256) =
match cursor.as_mut().expect("active WAL cursor").next()? {
WalCursorEvent::Eof => break,
WalCursorEvent::TornTail { reason, .. }
if config.recovery_mode == RecoveryMode::AutoRepairTornTail
&& !read_only =>
{
let owned = cursor.take().expect("active WAL cursor");
if captured {
owned.require_native_path(&wal_path)?;
*captured_file = Some(owned.into_file());
} else {
drop(owned);
}
let durable = self.durable.as_mut().expect("durable WAL replay");
let repair_directory = durable.wal_path.parent().ok_or_else(|| {
HawDBError::StorageIntegrity(
"WAL repair has no parent directory".into(),
)
})?;
let repair = super::doctor::automatic_wal_tail_repair_locked(
repair_directory,
config,
)?;
durable.wal_bytes = repair.retained_wal_len;
durable.wal_tail_repair = Some(repair.clone());
tail_repair = Some(repair);
torn_tail_reason = Some(reason);
break;
}
WalCursorEvent::TornTail { reason, .. } => {
return Err(HawDBError::Storage(format!(
"strict WAL recovery rejected torn tail: {reason}; use DatabaseDoctor to inspect and explicitly repair the incomplete final record"
)));
}
WalCursorEvent::Corrupt { offset, reason } => {
return reject_wal_source_corruption(
captured,
&wal_path,
wal_generation,
read_only,
config.max_quarantine_bytes,
offset,
reason,
);
}
WalCursorEvent::Entry {
entry,
start_offset,
encoded_len,
payload_len,
payload_sha256,
} => (
entry,
start_offset,
encoded_len,
payload_len,
payload_sha256,
),
};
if entry.lsn != expected_lsn {
if !captured {
quarantine_corrupt_wal(
&wal_path,
wal_generation,
read_only,
config.max_quarantine_bytes,
)?;
}
return Err(HawDBError::Storage(format!(
"WAL LSN sequence mismatch at byte offset {record_start}: expected {expected_lsn}, got {}",
entry.lsn
)));
}
if let Some(max_entries) = config.max_entries
&& replayed_entries >= max_entries
{
return Err(HawDBError::Storage(format!(
"WAL replay entry limit exceeded: max_wal_replay_entries={max_entries}"
)));
}
if let WalOp::Batch(ops) = &entry.op
&& config
.max_batch_operations
.is_some_and(|limit| ops.len() > limit)
{
return Err(HawDBError::Storage(format!(
"WAL batch operation limit exceeded: max_wal_batch_operations={}",
config.max_batch_operations.unwrap_or_default()
)));
}
replayed_entries += 1;
replayed_bytes = replayed_bytes.saturating_add(record_encoded_len);
relational_recovery_source
.record(entry.lsn, payload_len, payload_sha256)
.map_err(|reason| HawDBError::Storage(reason.to_string()))?;
expected_lsn = expected_lsn
.checked_add(1)
.ok_or_else(|| HawDBError::Storage("WAL LSN overflow during replay".to_string()))?;
self.apply_replayed_wal_transaction(catalog, entry.op)?;
}
if captured && captured_file.is_none() {
*captured_file = cursor.map(WalRecordCursor::into_file);
}
if let Some(durable) = &mut self.durable {
durable.next_lsn = expected_lsn;
durable.wal_commit_epoch = self.commit_epoch;
}
Ok(StorageRecoveryReport {
open_timings: Default::default(),
durable: true,
recovery_mode: config.recovery_mode,
max_wal_replay_entries: config.max_entries,
max_wal_replay_bytes: config.max_bytes,
max_wal_record_bytes: config.max_record_bytes,
checkpoint_epoch: checkpoint_present.then_some(checkpoint_epoch),
checkpoint_commit_epoch: checkpoint_present.then_some(checkpoint_commit_epoch),
wal_present,
wal_generation: Some(wal_generation),
wal_replay_start_lsn: Some(wal_replay_start_lsn),
next_lsn_after_replay: Some(expected_lsn),
replayed_wal_entries: replayed_entries,
replayed_wal_bytes: replayed_bytes,
torn_tail_ignored: false,
torn_tail_repaired: tail_repair.is_some(),
discarded_wal_tail_bytes: tail_repair
.as_ref()
.map_or(0, |repair| repair.discarded_wal_tail_bytes),
torn_tail_reason,
recovered_commit_epoch: self.commit_epoch,
})
}
pub(crate) fn retain_decoded_checkpoint_memory(
&mut self,
incoming: &mut crate::background::CheckpointAllocationOwner,
work: &crate::background::CheckpointWorkContext,
) -> Result<()> {
self.checkpoint_allocations
.append_retained(incoming, work)
.map_err(HawDBError::from_storage_error)
}
pub(crate) fn apply_replayed_wal_transaction(
&mut self,
catalog: &mut Catalog,
operation: WalOp,
) -> Result<()> {
self.apply_replayed_wal_transaction_with_work_context(catalog, operation, None, None, None)
}
#[cfg(test)]
pub(crate) fn apply_replayed_checkpoint_wal_transaction(
&mut self,
catalog: &mut Catalog,
operation: WalOp,
work: &crate::background::CheckpointWorkContext,
) -> Result<()> {
self.apply_replayed_checkpoint_wal_transaction_with_boundary(
catalog, operation, work, &mut false, None,
)
}
pub(crate) fn apply_replayed_checkpoint_wal_transaction_with_boundary(
&mut self,
catalog: &mut Catalog,
operation: WalOp,
work: &crate::background::CheckpointWorkContext,
mutation_started: &mut bool,
replay_memory: Option<&mut crate::background::CheckpointAllocationOwner>,
) -> Result<()> {
self.apply_replayed_wal_transaction_with_work_context(
catalog,
operation,
Some(work),
Some(mutation_started),
replay_memory,
)
}
fn apply_replayed_wal_transaction_with_work_context(
&mut self,
catalog: &mut Catalog,
operation: WalOp,
work: Option<&crate::background::CheckpointWorkContext>,
mutation_started: Option<&mut bool>,
replay_memory: Option<&mut crate::background::CheckpointAllocationOwner>,
) -> Result<()> {
let operations = match &operation {
WalOp::Batch(operations) => operations.as_slice(),
operation => std::slice::from_ref(operation),
};
if let Some(work) = work {
work.checkpoint().map_err(HawDBError::from_storage_error)?;
}
if let Some(work) = work {
self.ensure_checkpoint_out_of_core_delta_replay_admission(operations, work)?;
} else {
self.ensure_out_of_core_delta_replay_admission(operations)?;
}
if let Some(work) = work {
work.checkpoint().map_err(HawDBError::from_storage_error)?;
}
let epoch = self.commit_epoch.checked_add(1).ok_or_else(|| {
HawDBError::StorageIntegrity("commit epoch overflow during WAL replay".into())
})?;
let allocations = work.map(|work| crate::background::CheckpointDecodeContext {
work: work.clone(),
memory: Default::default(),
});
let mut relational_records = if let Some(allocations) = &allocations {
self.prepare_replayed_relational_records(operations, epoch, allocations)?
} else {
Vec::new()
};
let relational_changes = if let Some(work) = work {
Self::relational_changes_from_prepared_records(
operations,
&mut relational_records,
work,
)?
} else {
self.relational_primary_key_changes_from_wal_ops(operations)?
};
if let Some(work) = work {
work.checkpoint().map_err(HawDBError::from_storage_error)?;
}
if let Some(work) = work {
self.prepare_replayed_checkpoint_record_copies(operations, work)?;
}
match (replay_memory, allocations.as_ref()) {
(Some(memory), Some(allocations)) => {
memory
.append_retained(&mut allocations.memory.borrow_mut(), allocations)
.map_err(HawDBError::from_storage_error)?;
self.retain_decoded_checkpoint_memory(memory, allocations)?;
}
(None, Some(allocations)) => self.retain_decoded_checkpoint_memory(
&mut allocations.memory.borrow_mut(),
allocations,
)?,
(None, None) => {}
(Some(_), None) => unreachable!("decoded checkpoint memory requires a work context"),
}
if let Some(mutation_started) = mutation_started {
*mutation_started = true;
}
if let Some(work) = work {
self.record_checkpoint_search_projection_changes_for_ops(
catalog,
epoch,
operations,
relational_changes,
work,
)?;
let mut relational_records = relational_records.into_iter();
self.apply_replayed_checkpoint_operations(
catalog,
operation,
work,
&mut relational_records,
)?;
if relational_records.next().is_some() {
return Err(HawDBError::StorageIntegrity(
"unused prepared relational WAL record".into(),
));
}
work.checkpoint().map_err(HawDBError::from_storage_error)?;
} else {
self.record_search_projection_changes_for_ops(
catalog,
epoch,
operations,
relational_changes,
)?;
match operation {
WalOp::Batch(operations) => {
for operation in operations {
self.apply_wal_op(catalog, operation)?;
}
}
operation => self.apply_wal_op(catalog, operation)?,
}
}
self.commit_epoch = epoch;
self.advance_relational_row_recovery_epoch(epoch);
Ok(())
}
fn prepare_replayed_checkpoint_record_copies(
&mut self,
operations: &[WalOp],
work: &crate::background::CheckpointWorkContext,
) -> Result<()> {
fn count_insertions(
operations: &[WalOp],
work: &crate::background::CheckpointWorkContext,
count: &mut (usize, usize),
) -> Result<()> {
for operation in operations {
let unit = work.start_unit().map_err(HawDBError::from_storage_error)?;
match operation {
WalOp::Batch(operations) => {
unit.finish();
count_insertions(operations, work, count)?;
continue;
}
WalOp::CreateNode { .. } => count.0 += 1,
WalOp::CreateRelationship { .. } => count.1 += 1,
_ => {}
}
unit.finish();
}
Ok(())
}
let mut insertions = (0, 0);
count_insertions(operations, work, &mut insertions)?;
self.prepare_replayed_checkpoint_record_copies_inner(operations, insertions, work)
}
fn prepare_replayed_checkpoint_record_copies_inner(
&mut self,
operations: &[WalOp],
insertions: (usize, usize),
work: &crate::background::CheckpointWorkContext,
) -> Result<()> {
for operation in operations {
work.checkpoint().map_err(HawDBError::from_storage_error)?;
match operation {
WalOp::Batch(operations) => {
self.prepare_replayed_checkpoint_record_copies_inner(
operations, insertions, work,
)?;
}
WalOp::CreateNode { id, .. } => {
let result = if insertions.0 > 1 {
self.nodes
.prepare_checkpoint_insertions_for_key(id, insertions.0, work)
} else {
self.nodes.prepare_checkpoint_insert_copy_for_key(id, work)
};
result.map_err(HawDBError::from_storage_error)?;
}
WalOp::CreateRelationship { id, .. } => {
let result = if insertions.1 > 1 {
self.relationships.prepare_checkpoint_insertions_for_key(
id,
insertions.1,
work,
)
} else {
self.relationships
.prepare_checkpoint_insert_copy_for_key(id, work)
};
result.map_err(HawDBError::from_storage_error)?;
}
WalOp::SetNodeProperty { id, .. } | WalOp::DeleteNode { id } => self
.nodes
.prepare_checkpoint_copy_for_key(id, work)
.map_err(HawDBError::from_storage_error)?,
WalOp::SetRelationshipProperty { id, .. } | WalOp::DeleteRelationship { id } => {
self.relationships
.prepare_checkpoint_copy_for_key(id, work)
.map_err(HawDBError::from_storage_error)?
}
_ => {}
}
}
work.checkpoint().map_err(HawDBError::from_storage_error)
}
fn relational_changes_from_prepared_records(
operations: &[WalOp],
records: &mut [crate::relational::RelationalWalBatch],
work: &crate::background::CheckpointWorkContext,
) -> Result<Option<crate::relational::RelationalPrimaryKeyChangeCapture>> {
use crate::relational::{
RelationalPrimaryKeyChangeCapture as Capture,
RelationalPrimaryKeyChangeRebuildReason as Reason,
};
fn snapshot_count(
operations: &[WalOp],
work: &crate::background::CheckpointWorkContext,
) -> Result<usize> {
let mut count = 0usize;
for operation in operations {
let unit = work.start_unit().map_err(HawDBError::from_storage_error)?;
unit.finish();
let incoming = match operation {
WalOp::Batch(operations) => snapshot_count(operations, work)?,
WalOp::RelationalSnapshot { .. } => 1,
_ => 0,
};
count = count.checked_add(incoming).ok_or_else(|| {
HawDBError::StorageIntegrity("relational snapshot count overflow".into())
})?;
}
Ok(count)
}
let snapshots = snapshot_count(operations, work)?;
if records.len().saturating_add(snapshots) > 1 {
return Ok(Some(Capture::RequiresRebuild {
reason: Reason::MultipleRelationalTransactions,
}));
}
if snapshots != 0 {
return Ok(Some(Capture::RequiresRebuild {
reason: Reason::SnapshotReplacement,
}));
}
Ok(records.first_mut().map(|record| {
record
.primary_key_changes
.take()
.unwrap_or(Capture::RequiresRebuild {
reason: Reason::MissingWalCapture,
})
}))
}
fn prepare_replayed_relational_records(
&self,
operations: &[WalOp],
epoch: u64,
allocations: &crate::background::CheckpointDecodeContext,
) -> Result<Vec<crate::relational::RelationalWalBatch>> {
fn count(
operations: &[WalOp],
work: &crate::background::CheckpointWorkContext,
) -> Result<usize> {
let mut total = 0usize;
for operation in operations {
let unit = work.start_unit().map_err(HawDBError::from_storage_error)?;
unit.finish();
let incoming = match operation {
WalOp::Batch(operations) => count(operations, work)?,
WalOp::Relational { .. } => 1,
_ => 0,
};
total = total.checked_add(incoming).ok_or_else(|| {
HawDBError::StorageIntegrity("relational WAL record count overflow".into())
})?;
}
Ok(total)
}
fn decode(
operations: &[WalOp],
epoch: u64,
allocations: &crate::background::CheckpointDecodeContext,
output: &mut Vec<crate::relational::RelationalWalBatch>,
) -> Result<()> {
for operation in operations {
allocations
.checkpoint()
.map_err(HawDBError::from_storage_error)?;
match operation {
WalOp::Batch(operations) => decode(operations, epoch, allocations, output)?,
WalOp::Relational { record } => {
let batch = crate::relational::decode_relational_wal_batch_owned(
record,
RelationalDecodeLimits::wal(),
allocations,
)
.map_err(crate::relational::RelationalError::into_wal_recovery_error)?;
if batch.epoch != epoch {
return Err(HawDBError::StorageIntegrity(format!(
"relational WAL epoch mismatch: expected {epoch}, got {}",
batch.epoch
)));
}
output.push(batch);
}
_ => {}
}
}
Ok(())
}
let count = count(operations, allocations)?;
allocations
.checkpoint()
.map_err(HawDBError::from_storage_error)?;
if count == 0 {
return Ok(Vec::new());
}
let mut records = allocations.vector_capacity(count)?;
decode(operations, epoch, allocations, &mut records)?;
Ok(records)
}
fn apply_replayed_checkpoint_operations(
&mut self,
catalog: &mut Catalog,
operation: WalOp,
work: &crate::background::CheckpointWorkContext,
relational_records: &mut std::vec::IntoIter<crate::relational::RelationalWalBatch>,
) -> Result<()> {
match operation {
WalOp::Batch(operations) => {
for operation in operations {
self.apply_replayed_checkpoint_operations(
catalog,
operation,
work,
relational_records,
)?;
}
Ok(())
}
WalOp::Relational { .. } => {
let batch = relational_records.next().ok_or_else(|| {
HawDBError::StorageIntegrity("missing prepared relational WAL record".into())
})?;
self.apply_recovered_relational_batch(batch, Some(work))
}
operation @ WalOp::ProjectGraph { .. } => {
self.apply_wal_op_with_work_context(catalog, operation, Some(work))
}
operation => {
let unit = work.start_unit().map_err(HawDBError::from_storage_error)?;
self.apply_wal_op(catalog, operation)?;
unit.finish();
Ok(())
}
}
}
}
fn reject_wal_source_corruption<T>(
captured: bool,
path: &Path,
generation: u64,
read_only: bool,
max_quarantine_bytes: u64,
record_start: u64,
reason: impl std::fmt::Display,
) -> Result<T> {
if captured {
Err(HawDBError::Storage(format!(
"WAL corruption at byte offset {record_start}: {reason}"
)))
} else {
reject_corrupt_wal_record(
path,
generation,
read_only,
max_quarantine_bytes,
record_start,
reason,
)
}
}