use crate::{
db::{MutationJobPhase, MutationJobRestartReason, MutationJobTargetFailureReason},
metrics::state as metrics,
};
use super::counters::*;
use super::{ExecKind, MetricsEvent, MetricsSink, MutationJobLifecycleEvent, SaveMutationKind};
pub(crate) struct GlobalMetricsSink;
impl MetricsSink for GlobalMetricsSink {
#[remain::check]
#[expect(clippy::too_many_lines)]
fn record(&self, event: MetricsEvent) {
#[remain::sorted]
match event {
MetricsEvent::AcceptedSchemaFootprint {
entity_path,
fields,
nested_leaf_facts,
} => {
metrics::with_state_mut(|m| {
let (previous_fields, previous_nested_leaf_facts) = {
let entry = m.entities.entry(entity_path.to_string()).or_default();
let previous = (
entry.accepted_schema_fields,
entry.accepted_schema_nested_leaf_facts,
);
entry.accepted_schema_fields = fields;
entry.accepted_schema_nested_leaf_facts = nested_leaf_facts;
previous
};
replace_gauge_total(&mut m.ops.accepted_schema_fields, previous_fields, fields);
replace_gauge_total(
&mut m.ops.accepted_schema_nested_leaf_facts,
previous_nested_leaf_facts,
nested_leaf_facts,
);
});
}
MetricsEvent::Cache {
entity_path,
kind,
outcome,
} => {
metrics::with_state_mut(|m| {
record_global_cache_outcome(&mut m.ops, kind, outcome);
let entry = m.entities.entry(entity_path.to_string()).or_default();
record_entity_cache_outcome(entry, kind, outcome);
});
}
MetricsEvent::CacheEntries { kind, entries } => {
metrics::with_state_mut(|m| {
record_global_cache_entries(&mut m.ops, kind, entries);
});
}
MetricsEvent::CacheMissReason {
entity_path,
kind,
reason,
} => {
metrics::with_state_mut(|m| {
record_global_cache_miss_reason(&mut m.ops, kind, reason);
let entry = m.entities.entry(entity_path.to_string()).or_default();
record_entity_cache_miss_reason(entry, kind, reason);
});
}
MetricsEvent::ExecError {
kind,
entity_path,
outcome,
} => {
metrics::with_state_mut(|m| {
record_global_exec_start(&mut m.ops, kind);
record_global_exec_outcome(&mut m.ops, outcome);
let entry = m.entities.entry(entity_path.to_string()).or_default();
record_entity_exec_start(entry, kind);
record_entity_exec_outcome(entry, outcome);
});
}
MetricsEvent::ExecFinish {
kind,
entity_path,
rows_touched,
inst_delta,
outcome,
} => {
metrics::with_state_mut(|m| {
record_global_exec_outcome(&mut m.ops, outcome);
#[remain::sorted]
match kind {
ExecKind::Delete => {
m.ops.rows_deleted = m.ops.rows_deleted.saturating_add(rows_touched);
m.ops.write_rows_touched =
m.ops.write_rows_touched.saturating_add(rows_touched);
metrics::add_instructions(
&mut m.perf.delete_inst_total,
&mut m.perf.delete_inst_max,
inst_delta,
);
}
ExecKind::Load => {
m.ops.rows_loaded = m.ops.rows_loaded.saturating_add(rows_touched);
metrics::add_instructions(
&mut m.perf.load_inst_total,
&mut m.perf.load_inst_max,
inst_delta,
);
}
ExecKind::Save => {
m.ops.rows_saved = m.ops.rows_saved.saturating_add(rows_touched);
m.ops.write_rows_touched =
m.ops.write_rows_touched.saturating_add(rows_touched);
metrics::add_instructions(
&mut m.perf.save_inst_total,
&mut m.perf.save_inst_max,
inst_delta,
);
}
}
let entry = m.entities.entry(entity_path.to_string()).or_default();
record_entity_exec_outcome(entry, outcome);
#[remain::sorted]
match kind {
ExecKind::Delete => {
entry.rows_deleted = entry.rows_deleted.saturating_add(rows_touched);
entry.write_rows_touched =
entry.write_rows_touched.saturating_add(rows_touched);
}
ExecKind::Load => {
entry.rows_loaded = entry.rows_loaded.saturating_add(rows_touched);
}
ExecKind::Save => {
entry.rows_saved = entry.rows_saved.saturating_add(rows_touched);
entry.write_rows_touched =
entry.write_rows_touched.saturating_add(rows_touched);
}
}
});
}
MetricsEvent::ExecStart { kind, entity_path } => {
metrics::with_state_mut(|m| {
record_global_exec_start(&mut m.ops, kind);
let entry = m.entities.entry(entity_path.to_string()).or_default();
record_entity_exec_start(entry, kind);
});
}
MetricsEvent::IndexDelta {
entity_path,
inserts,
removes,
} => {
metrics::with_state_mut(|m| {
m.ops.index_inserts = m.ops.index_inserts.saturating_add(inserts);
m.ops.index_removes = m.ops.index_removes.saturating_add(removes);
let changed = inserts.saturating_add(removes);
m.ops.write_index_entries_changed =
m.ops.write_index_entries_changed.saturating_add(changed);
let entry = m.entities.entry(entity_path.to_string()).or_default();
entry.index_inserts = entry.index_inserts.saturating_add(inserts);
entry.index_removes = entry.index_removes.saturating_add(removes);
entry.write_index_entries_changed =
entry.write_index_entries_changed.saturating_add(changed);
});
}
MetricsEvent::LoadRowEfficiency {
entity_path,
candidate_rows_scanned,
candidate_rows_filtered,
result_rows_emitted,
} => {
metrics::with_state_mut(|m| {
m.ops.load_candidate_rows_scanned = m
.ops
.load_candidate_rows_scanned
.saturating_add(candidate_rows_scanned);
m.ops.load_candidate_rows_filtered = m
.ops
.load_candidate_rows_filtered
.saturating_add(candidate_rows_filtered);
m.ops.load_result_rows_emitted = m
.ops
.load_result_rows_emitted
.saturating_add(result_rows_emitted);
let entry = m.entities.entry(entity_path.to_string()).or_default();
entry.load_candidate_rows_scanned = entry
.load_candidate_rows_scanned
.saturating_add(candidate_rows_scanned);
entry.load_candidate_rows_filtered = entry
.load_candidate_rows_filtered
.saturating_add(candidate_rows_filtered);
entry.load_result_rows_emitted = entry
.load_result_rows_emitted
.saturating_add(result_rows_emitted);
});
}
MetricsEvent::MutationCommitPlan { .. } => {}
MetricsEvent::MutationJobCapacity {
retained_count,
hard_limit,
reserved_integrity_headroom,
integrity_count,
resumable_count,
mutation_count,
retained_record_bytes,
} => {
metrics::with_state_mut(|m| {
let jobs = &mut m.ops.mutation_jobs;
jobs.retained_count = retained_count;
jobs.hard_limit = hard_limit;
jobs.reserved_integrity_headroom = reserved_integrity_headroom;
jobs.integrity_count = integrity_count;
jobs.resumable_count = resumable_count;
jobs.mutation_count = mutation_count;
jobs.retained_record_bytes = retained_record_bytes;
});
}
MetricsEvent::MutationJobLifecycle { event } => {
metrics::with_state_mut(|m| {
let jobs = &mut m.ops.mutation_jobs;
#[remain::sorted]
let counter = match event {
MutationJobLifecycleEvent::AdvanceExactReplay => {
&mut jobs.advances_exact_replayed
}
MutationJobLifecycleEvent::CancelUnadvanced => &mut jobs.cancellations,
MutationJobLifecycleEvent::Complete => &mut jobs.completions,
MutationJobLifecycleEvent::ForwardToVerify => {
&mut jobs.forward_to_verify_transitions
}
MutationJobLifecycleEvent::InventoryLoaded => &mut jobs.inventories_loaded,
MutationJobLifecycleEvent::StartExactReplay => {
&mut jobs.starts_exact_replayed
}
MutationJobLifecycleEvent::StartInserted => &mut jobs.starts_inserted,
MutationJobLifecycleEvent::StateLoaded => &mut jobs.states_loaded,
MutationJobLifecycleEvent::TerminalAcknowledged => {
&mut jobs.terminal_acknowledgements
}
MutationJobLifecycleEvent::VerifyRestartResidualWork => {
&mut jobs.verify_restarts_residual_work
}
MutationJobLifecycleEvent::VerifyRestartRevisionDrift => {
&mut jobs.verify_restarts_revision_drift
}
};
*counter = counter.saturating_add(1);
});
}
MetricsEvent::MutationJobRestart { reason } => {
metrics::with_state_mut(|m| {
let jobs = &mut m.ops.mutation_jobs;
let counter = match reason {
MutationJobRestartReason::AcceptedSchemaChanged => {
&mut jobs.restart_accepted_schema_changed
}
MutationJobRestartReason::BatchPolicyChanged => {
&mut jobs.restart_batch_policy_changed
}
MutationJobRestartReason::CandidateExceedsBatchPolicy => {
&mut jobs.restart_candidate_exceeds_batch_policy
}
MutationJobRestartReason::ExecutionBudgetPolicyExceeded => {
&mut jobs.restart_execution_budget_policy_exceeded
}
MutationJobRestartReason::IntentIneligible => {
&mut jobs.restart_intent_ineligible
}
MutationJobRestartReason::ManagedTimestampRegression => {
&mut jobs.restart_managed_timestamp_regression
}
MutationJobRestartReason::TargetAllocationChanged => {
&mut jobs.restart_target_allocation_changed
}
MutationJobRestartReason::UnsupportedContinuation => {
&mut jobs.restart_unsupported_continuation
}
};
*counter = counter.saturating_add(1);
});
}
MetricsEvent::MutationJobStep {
phase,
keys_scanned,
rows_updated,
scan_bytes,
staged_bytes,
keys_scanned_total,
rows_updated_total,
verify_restarts_total,
} => {
metrics::with_state_mut(|m| {
let jobs = &mut m.ops.mutation_jobs;
match phase {
MutationJobPhase::Forward => {
jobs.forward_steps_committed =
jobs.forward_steps_committed.saturating_add(1);
}
MutationJobPhase::Verify => {
jobs.verify_steps_committed =
jobs.verify_steps_committed.saturating_add(1);
}
}
jobs.keys_scanned = jobs.keys_scanned.saturating_add(keys_scanned);
jobs.rows_updated = jobs.rows_updated.saturating_add(rows_updated);
jobs.scan_bytes = jobs.scan_bytes.saturating_add(scan_bytes);
jobs.staged_bytes = jobs.staged_bytes.saturating_add(staged_bytes);
jobs.keys_scanned_cumulative_max =
jobs.keys_scanned_cumulative_max.max(keys_scanned_total);
jobs.rows_updated_cumulative_max =
jobs.rows_updated_cumulative_max.max(rows_updated_total);
jobs.verify_restarts_cumulative_max = jobs
.verify_restarts_cumulative_max
.max(verify_restarts_total);
});
}
MetricsEvent::MutationJobTargetFailure { reason } => {
metrics::with_state_mut(|m| {
let jobs = &mut m.ops.mutation_jobs;
let counter = match reason {
MutationJobTargetFailureReason::StagingByteBudgetExceeded => {
&mut jobs.target_failure_staging_byte_budget_exceeded
}
MutationJobTargetFailureReason::Other => &mut jobs.target_failure_other,
};
*counter = counter.saturating_add(1);
});
}
MetricsEvent::NonAtomicPartialCommit {
entity_path,
committed_rows,
} => {
metrics::with_state_mut(|m| {
m.ops.non_atomic_partial_commits =
m.ops.non_atomic_partial_commits.saturating_add(1);
m.ops.non_atomic_partial_rows_committed = m
.ops
.non_atomic_partial_rows_committed
.saturating_add(committed_rows);
let entry = m.entities.entry(entity_path.to_string()).or_default();
entry.non_atomic_partial_commits =
entry.non_atomic_partial_commits.saturating_add(1);
entry.non_atomic_partial_rows_committed = entry
.non_atomic_partial_rows_committed
.saturating_add(committed_rows);
});
}
MetricsEvent::Plan {
entity_path,
kind,
grouped_execution_mode,
} => {
metrics::with_state_mut(|m| {
record_global_plan_kind(&mut m.ops, kind);
record_global_grouped_plan_mode(&mut m.ops, grouped_execution_mode);
let entry = m.entities.entry(entity_path.to_string()).or_default();
record_entity_plan_kind(entry, kind);
record_entity_grouped_plan_mode(entry, grouped_execution_mode);
});
}
MetricsEvent::PlanChoice {
entity_path,
reason,
} => {
metrics::with_state_mut(|m| {
record_global_plan_choice_reason(&mut m.ops, reason);
let entry = m.entities.entry(entity_path.to_string()).or_default();
record_entity_plan_choice_reason(entry, reason);
});
}
MetricsEvent::PreparedShapeAlreadyFinalized { entity_path } => {
metrics::with_state_mut(|m| {
m.ops.prepared_shape_already_finalized =
m.ops.prepared_shape_already_finalized.saturating_add(1);
let entry = m.entities.entry(entity_path.to_string()).or_default();
entry.prepared_shape_already_finalized =
entry.prepared_shape_already_finalized.saturating_add(1);
});
}
MetricsEvent::RelationValidation {
entity_path,
reverse_lookups,
blocked_deletes,
} => {
metrics::with_state_mut(|m| {
m.ops.relation_reverse_lookups = m
.ops
.relation_reverse_lookups
.saturating_add(reverse_lookups);
m.ops.relation_delete_blocks =
m.ops.relation_delete_blocks.saturating_add(blocked_deletes);
m.ops.write_relation_checks =
m.ops.write_relation_checks.saturating_add(reverse_lookups);
let entry = m.entities.entry(entity_path.to_string()).or_default();
entry.relation_reverse_lookups = entry
.relation_reverse_lookups
.saturating_add(reverse_lookups);
entry.relation_delete_blocks =
entry.relation_delete_blocks.saturating_add(blocked_deletes);
entry.write_relation_checks =
entry.write_relation_checks.saturating_add(reverse_lookups);
});
}
MetricsEvent::ReverseIndexDelta {
entity_path,
inserts,
removes,
} => {
metrics::with_state_mut(|m| {
m.ops.reverse_index_inserts =
m.ops.reverse_index_inserts.saturating_add(inserts);
m.ops.reverse_index_removes =
m.ops.reverse_index_removes.saturating_add(removes);
let changed = inserts.saturating_add(removes);
m.ops.write_reverse_index_entries_changed = m
.ops
.write_reverse_index_entries_changed
.saturating_add(changed);
let entry = m.entities.entry(entity_path.to_string()).or_default();
entry.reverse_index_inserts =
entry.reverse_index_inserts.saturating_add(inserts);
entry.reverse_index_removes =
entry.reverse_index_removes.saturating_add(removes);
entry.write_reverse_index_entries_changed = entry
.write_reverse_index_entries_changed
.saturating_add(changed);
});
}
MetricsEvent::RowsAggregated {
entity_path,
rows_aggregated,
} => {
metrics::with_state_mut(|m| {
m.ops.rows_aggregated = m.ops.rows_aggregated.saturating_add(rows_aggregated);
let entry = m.entities.entry(entity_path.to_string()).or_default();
entry.rows_aggregated = entry.rows_aggregated.saturating_add(rows_aggregated);
});
}
MetricsEvent::RowsEmitted {
entity_path,
rows_emitted,
} => {
metrics::with_state_mut(|m| {
m.ops.rows_emitted = m.ops.rows_emitted.saturating_add(rows_emitted);
let entry = m.entities.entry(entity_path.to_string()).or_default();
entry.rows_emitted = entry.rows_emitted.saturating_add(rows_emitted);
});
}
MetricsEvent::RowsFiltered {
entity_path,
rows_filtered,
} => {
metrics::with_state_mut(|m| {
m.ops.rows_filtered = m.ops.rows_filtered.saturating_add(rows_filtered);
let entry = m.entities.entry(entity_path.to_string()).or_default();
entry.rows_filtered = entry.rows_filtered.saturating_add(rows_filtered);
});
}
MetricsEvent::RowsScanned {
entity_path,
rows_scanned,
} => {
metrics::with_state_mut(|m| {
m.ops.rows_scanned = m.ops.rows_scanned.saturating_add(rows_scanned);
let entry = m.entities.entry(entity_path.to_string()).or_default();
entry.rows_scanned = entry.rows_scanned.saturating_add(rows_scanned);
});
}
MetricsEvent::SaveMutation {
entity_path,
kind,
rows_touched,
} => {
metrics::with_state_mut(|m| {
#[remain::sorted]
match kind {
SaveMutationKind::Insert => {
m.ops.save_insert_calls = m.ops.save_insert_calls.saturating_add(1);
m.ops.rows_inserted = m.ops.rows_inserted.saturating_add(rows_touched);
}
SaveMutationKind::Replace => {
m.ops.save_replace_calls = m.ops.save_replace_calls.saturating_add(1);
m.ops.rows_replaced = m.ops.rows_replaced.saturating_add(rows_touched);
}
SaveMutationKind::Update => {
m.ops.save_update_calls = m.ops.save_update_calls.saturating_add(1);
m.ops.rows_updated = m.ops.rows_updated.saturating_add(rows_touched);
}
}
let entry = m.entities.entry(entity_path.to_string()).or_default();
#[remain::sorted]
match kind {
SaveMutationKind::Insert => {
entry.save_insert_calls = entry.save_insert_calls.saturating_add(1);
entry.rows_inserted = entry.rows_inserted.saturating_add(rows_touched);
}
SaveMutationKind::Replace => {
entry.save_replace_calls = entry.save_replace_calls.saturating_add(1);
entry.rows_replaced = entry.rows_replaced.saturating_add(rows_touched);
}
SaveMutationKind::Update => {
entry.save_update_calls = entry.save_update_calls.saturating_add(1);
entry.rows_updated = entry.rows_updated.saturating_add(rows_touched);
}
}
});
}
MetricsEvent::SchemaReconcile {
entity_path,
outcome,
} => {
metrics::with_state_mut(|m| {
record_global_schema_reconcile_outcome(&mut m.ops, outcome);
let entry = m.entities.entry(entity_path.to_string()).or_default();
record_entity_schema_reconcile_outcome(entry, outcome);
});
}
MetricsEvent::SchemaStoreFootprint {
encoded_bytes,
entity_path,
latest_snapshot_bytes,
snapshots,
} => {
metrics::with_state_mut(|m| {
let (
previous_snapshots,
previous_encoded_bytes,
previous_latest_snapshot_bytes,
) = {
let entry = m.entities.entry(entity_path.to_string()).or_default();
let previous = (
entry.schema_store_snapshots,
entry.schema_store_encoded_bytes,
entry.schema_store_latest_snapshot_bytes,
);
entry.schema_store_snapshots = snapshots;
entry.schema_store_encoded_bytes = encoded_bytes;
entry.schema_store_latest_snapshot_bytes = latest_snapshot_bytes;
previous
};
replace_gauge_total(
&mut m.ops.schema_store_snapshots,
previous_snapshots,
snapshots,
);
replace_gauge_total(
&mut m.ops.schema_store_encoded_bytes,
previous_encoded_bytes,
encoded_bytes,
);
replace_gauge_total(
&mut m.ops.schema_store_latest_snapshot_bytes,
previous_latest_snapshot_bytes,
latest_snapshot_bytes,
);
});
}
MetricsEvent::SchemaTransition {
entity_path,
outcome,
} => {
metrics::with_state_mut(|m| {
record_global_schema_transition_outcome(&mut m.ops, outcome);
let entry = m.entities.entry(entity_path.to_string()).or_default();
record_entity_schema_transition_outcome(entry, outcome);
});
}
MetricsEvent::SqlCompileReject { entity_path, phase } => {
metrics::with_state_mut(|m| {
record_global_sql_compile_reject_phase(&mut m.ops, phase);
let entry = m.entities.entry(entity_path.to_string()).or_default();
record_entity_sql_compile_reject_phase(entry, phase);
});
}
MetricsEvent::SqlWrite {
entity_path,
kind,
staged_rows,
matched_rows,
mutated_rows,
returning_rows,
} => {
metrics::with_state_mut(|m| {
record_global_sql_write_kind(&mut m.ops, kind);
m.ops.sql_write_staged_rows =
m.ops.sql_write_staged_rows.saturating_add(staged_rows);
m.ops.sql_write_matched_rows =
m.ops.sql_write_matched_rows.saturating_add(matched_rows);
m.ops.sql_write_mutated_rows =
m.ops.sql_write_mutated_rows.saturating_add(mutated_rows);
m.ops.sql_write_returning_rows = m
.ops
.sql_write_returning_rows
.saturating_add(returning_rows);
let entry = m.entities.entry(entity_path.to_string()).or_default();
record_entity_sql_write_kind(entry, kind);
entry.sql_write_staged_rows =
entry.sql_write_staged_rows.saturating_add(staged_rows);
entry.sql_write_matched_rows =
entry.sql_write_matched_rows.saturating_add(matched_rows);
entry.sql_write_mutated_rows =
entry.sql_write_mutated_rows.saturating_add(mutated_rows);
entry.sql_write_returning_rows = entry
.sql_write_returning_rows
.saturating_add(returning_rows);
});
}
MetricsEvent::SqlWriteError {
entity_path,
kind,
class,
} => {
metrics::with_state_mut(|m| {
record_global_sql_write_error_kind(&mut m.ops, kind);
record_global_sql_write_error_class(&mut m.ops, class);
let entry = m.entities.entry(entity_path.to_string()).or_default();
record_entity_sql_write_error_kind(entry, kind);
record_entity_sql_write_error_class(entry, class);
});
}
MetricsEvent::UniqueViolation { entity_path } => {
metrics::with_state_mut(|m| {
m.ops.unique_violations = m.ops.unique_violations.saturating_add(1);
let entry = m.entities.entry(entity_path.to_string()).or_default();
entry.unique_violations = entry.unique_violations.saturating_add(1);
});
}
}
}
}
pub(crate) const GLOBAL_METRICS_SINK: GlobalMetricsSink = GlobalMetricsSink;