From 0bfef64fea4cbdb08fa2d20741b831cf043de71f Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Mon, 31 Aug 2026 17:41:48 +0000 Subject: [PATCH] feat(hotpath): fill global-db schema/session/transcript gaps Co-authored-by: Zack Jackson --- .../src/configuration/store/audit.rs | 10 ++++++ .../src/configuration/store/mutation.rs | 22 ++++++++++++ .../src/observation_projection/rebuild.rs | 8 +++++ .../src/registered/delivery_settlement.rs | 35 +++++++++++++++++++ .../src/registered_sessions.rs | 11 ++++++ .../src/schema_contract/invariants/repair.rs | 23 ++++++++++++ .../src/schema_contract/invariants/rows.rs | 15 ++++++++ .../schema_contract/invariants/triggers.rs | 24 +++++++++++++ .../src/schema_contract/validation.rs | 29 +++++++++++++++ .../src/session_temporal_handle.rs | 33 +++++++++++++++++ crates/tracedecay-global-db/src/transcript.rs | 22 ++++++++++++ 11 files changed, 232 insertions(+) diff --git a/crates/tracedecay-global-db/src/configuration/store/audit.rs b/crates/tracedecay-global-db/src/configuration/store/audit.rs index 2f27fca729..89c32b1744 100644 --- a/crates/tracedecay-global-db/src/configuration/store/audit.rs +++ b/crates/tracedecay-global-db/src/configuration/store/audit.rs @@ -29,6 +29,7 @@ pub(super) fn encode_audit_payload( pub(super) const CONFIGURATION_AUDIT_REDACTION_KEY_BYTES: usize = 32; +#[hotpath::measure(future = true, label = "global_db.configuration.audit.read_key")] pub(super) async fn read_audit_redaction_key( transaction: &impl QueryExecutor, ) -> ConfigurationStoreResult>>> { @@ -55,6 +56,7 @@ pub(super) async fn read_audit_redaction_key( Ok(Some(material)) } +#[hotpath::measure(future = true, label = "global_db.configuration.audit.ensure_key")] pub(super) async fn ensure_audit_redaction_key( transaction: &impl Executor, created_at: UtcMicros, @@ -93,6 +95,7 @@ pub(super) fn audit_target_commitment( .map_err(ConfigurationStoreError::from) } +#[hotpath::measure(future = true, label = "global_db.configuration.audit.seal_target")] pub(super) async fn seal_audit_target( transaction: &impl Executor, event_id: &ConfigurationAuditEventId, @@ -109,6 +112,7 @@ pub(super) async fn seal_audit_target( Ok((sealed, commitment)) } +#[hotpath::measure(future = true, label = "global_db.configuration.audit.validate_seal")] pub(super) async fn validate_sealed_audit_target( transaction: &impl QueryExecutor, event: &ConfigurationAuditEvent, @@ -129,6 +133,7 @@ pub(super) async fn validate_sealed_audit_target( Ok(()) } +#[hotpath::measure(future = true, label = "global_db.configuration.audit.insert_dry_run")] pub(super) async fn insert_dry_run_audit_event( transaction: &impl Executor, record: &ConfigurationProtectedPlanRecordV1, @@ -267,6 +272,7 @@ pub(super) fn decode_audit_row( Ok((event, sealed_target_reference)) } +#[hotpath::measure(future = true, label = "global_db.configuration.audit.read_event")] pub(super) async fn read_audit_event_from_transaction( transaction: &impl QueryExecutor, event_id: &ConfigurationAuditEventId, @@ -295,6 +301,7 @@ pub(super) async fn read_audit_event_from_transaction( Ok(Some(event)) } +#[hotpath::measure(future = true, label = "global_db.configuration.audit.insert_event")] pub(super) async fn insert_audit_event_with_receipt_digest( transaction: &impl Executor, event: &ConfigurationAuditEvent, @@ -354,6 +361,7 @@ pub(super) fn is_terminal_plan_event(event_kind: &str) -> bool { matches!(event_kind, "applied" | "rollback_applied") } +#[hotpath::measure(future = true, label = "global_db.configuration.audit.append_terminal")] pub(super) async fn append_terminal_plan_event( transaction: &impl Executor, plan: &ProtectedChangePlan, @@ -423,6 +431,7 @@ pub(super) async fn append_terminal_plan_event( Ok(()) } +#[hotpath::measure(future = true, label = "global_db.configuration.audit.match_terminal")] pub(super) async fn has_matching_terminal_plan_event( transaction: &impl QueryExecutor, plan: &ProtectedChangePlan, @@ -454,6 +463,7 @@ pub(super) async fn has_matching_terminal_plan_event( Ok(terminal_count == 1 && matched) } +#[hotpath::measure(future = true, label = "global_db.configuration.audit.read_page")] pub(super) async fn audit_from_transaction( transaction: &impl QueryExecutor, after: Option<&ConfigurationAuditEventId>, diff --git a/crates/tracedecay-global-db/src/configuration/store/mutation.rs b/crates/tracedecay-global-db/src/configuration/store/mutation.rs index f4355cab41..5f9461b6ed 100644 --- a/crates/tracedecay-global-db/src/configuration/store/mutation.rs +++ b/crates/tracedecay-global-db/src/configuration/store/mutation.rs @@ -144,6 +144,10 @@ pub(super) fn decode_stored_mutation_receipt( }) } +#[hotpath::measure( + future = true, + label = "global_db.configuration.query.idempotency_receipt" +)] pub(super) async fn receipt_for_idempotency_from_transaction( transaction: &impl QueryExecutor, actor_id: &ActorId, @@ -174,6 +178,10 @@ pub(super) async fn receipt_for_idempotency_from_transaction( Ok(Some(receipt)) } +#[hotpath::measure( + future = true, + label = "global_db.configuration.persist.mutation_receipt" +)] pub(super) async fn insert_mutation_receipt( transaction: &impl Executor, commit: &ConfigurationCommitV1, @@ -257,6 +265,7 @@ pub(super) fn validate_commit_bindings( Ok(()) } +#[hotpath::measure(future = true, label = "global_db.configuration.query.replay_match")] pub(super) async fn replay_matches_commit( transaction: &impl QueryExecutor, stored: &StoredMutationReceipt, @@ -292,6 +301,7 @@ pub(super) async fn replay_matches_commit( Ok(true) } +#[hotpath::measure(future = true, label = "global_db.configuration.persist.commit")] pub(super) async fn commit_configuration_transaction( transaction: &impl Executor, commit: &ConfigurationCommitV1, @@ -310,6 +320,10 @@ pub(super) async fn commit_configuration_transaction( .await } +#[hotpath::measure( + future = true, + label = "global_db.configuration.persist.commit_with_registry" +)] pub(super) async fn commit_configuration_transaction_with_registry( transaction: &impl Executor, commit: &ConfigurationCommitV1, @@ -596,6 +610,7 @@ pub(super) struct ConfigurationCommitDraft<'a, T> { pub(super) target: &'a T, } +#[hotpath::measure(future = true, label = "global_db.configuration.persist.build_commit")] pub(super) async fn build_configuration_commit( transaction: &impl Executor, draft: ConfigurationCommitDraft<'_, T>, @@ -780,6 +795,7 @@ pub(super) fn rollback_redacted_changes( .collect() } +#[hotpath::measure(future = true, label = "global_db.configuration.query.current_state")] pub(super) async fn current_state_from_transaction( transaction: &impl QueryExecutor, ) -> Result { @@ -798,6 +814,7 @@ pub(super) async fn current_state_from_transaction( }) } +#[hotpath::measure(future = true, label = "global_db.configuration.query.replay_control")] pub(super) async fn replay_control_receipt( transaction: &impl QueryExecutor, actor_id: &ActorId, @@ -848,6 +865,7 @@ pub struct ConfigurationDirectCommitOutcomeV1 { pub current: ConfigurationCurrentStateV1, } +#[hotpath::measure(future = true, label = "global_db.configuration.persist.commit_direct")] pub async fn commit_direct_in_transaction( transaction: &E, authority: &ConfigurationMutationAuthority, @@ -867,6 +885,10 @@ where .await } +#[hotpath::measure( + future = true, + label = "global_db.configuration.persist.commit_direct_registry" +)] pub(super) async fn commit_direct_in_transaction_with_registry( transaction: &E, authority: &ConfigurationMutationAuthority, diff --git a/crates/tracedecay-global-db/src/observation_projection/rebuild.rs b/crates/tracedecay-global-db/src/observation_projection/rebuild.rs index 0bfd1fa3fc..fa8a0ae3b7 100644 --- a/crates/tracedecay-global-db/src/observation_projection/rebuild.rs +++ b/crates/tracedecay-global-db/src/observation_projection/rebuild.rs @@ -263,6 +263,10 @@ fn projection_retry_delay_micros(attempt_count: u32) -> i64 { .min(PROJECTION_RETRY_MAX_MICROS) } +#[hotpath::measure( + future = true, + label = "global_db.observation_projection.persist.retry" +)] async fn persist_projection_retry_on_database( database: &Database, observation_id: &CanonicalObservationIdV1, @@ -293,6 +297,10 @@ async fn persist_projection_retry_on_database( .map_err(|error| storage("commit projection retry transaction", error)) } +#[hotpath::measure( + future = true, + label = "global_db.observation_projection.persist.rejection" +)] async fn persist_projection_rejection_on_database( database: &Database, observation_id: &CanonicalObservationIdV1, diff --git a/crates/tracedecay-global-db/src/registered/delivery_settlement.rs b/crates/tracedecay-global-db/src/registered/delivery_settlement.rs index 3eec135196..937160535f 100644 --- a/crates/tracedecay-global-db/src/registered/delivery_settlement.rs +++ b/crates/tracedecay-global-db/src/registered/delivery_settlement.rs @@ -54,6 +54,7 @@ impl RegisteredGlobalDb { /// bound to `work_attempt`. The query is authorization-scoped by project, /// keyed by a canonical identity digest, and capped before any data leaves /// the reader. It never derives a Work identity from `owner_event_id`. + #[hotpath::measure(future = true, label = "global_db.delivery_settlement.query.censuses")] pub async fn work_attempt_delivery_censuses( &self, project_id: &str, @@ -122,6 +123,7 @@ impl RegisteredGlobalDb { /// Durably records one concrete recipient at the completion boundary that /// observed it. Callers choose whether that boundary is pre-write or /// post-hoc and must describe their timing truthfully. + #[hotpath::measure(future = true, label = "global_db.delivery_settlement.persist.begin")] pub async fn begin_delivery_attempt( &self, project_id: &str, @@ -133,6 +135,10 @@ impl RegisteredGlobalDb { /// Durably binds an opaque source acknowledgement token to the exact /// admitted recipient in the same transaction as attempt admission. + #[hotpath::measure( + future = true, + label = "global_db.delivery_settlement.persist.begin_receipted" + )] pub async fn begin_receipted_delivery_attempt( &self, project_id: &str, @@ -208,6 +214,10 @@ impl RegisteredGlobalDb { /// Resolves a project-scoped source acknowledgement token to the exact /// durable attempt admitted for it. Unknown tokens remain a typed absence. + #[hotpath::measure( + future = true, + label = "global_db.delivery_settlement.query.source_receipt" + )] pub async fn delivery_attempt_for_source_receipt( &self, project_id: &str, @@ -283,6 +293,10 @@ impl RegisteredGlobalDb { /// Reads one bounded due page of pending opaque source receipts. This is /// an indexed deadline scan; callers advance durable state by settling the /// returned exact attempts and may repeat until the page is empty. + #[hotpath::measure( + future = true, + label = "global_db.delivery_settlement.query.pending_due" + )] pub async fn pending_receipted_delivery_attempts_due( &self, project_id: &str, @@ -348,6 +362,7 @@ impl RegisteredGlobalDb { /// CASes an admitted recipient to one immutable terminal outcome and /// returns the exact bounded surface census from the same transaction. + #[hotpath::measure(future = true, label = "global_db.delivery_settlement.persist.settle")] pub async fn settle_delivery_attempt( &self, project_id: &str, @@ -496,6 +511,10 @@ fn validate_source_receipt_ref(source_receipt_ref: &str) -> Result<(), String> { Ok(()) } +#[hotpath::measure( + future = true, + label = "global_db.delivery_settlement.persist.bind_receipt" +)] async fn bind_source_receipt( transaction: &RegisteredGlobalDbWriteTransaction<'_>, project_id: &str, @@ -586,6 +605,10 @@ async fn bind_source_receipt( Ok(()) } +#[hotpath::measure( + future = true, + label = "global_db.delivery_settlement.persist.fanout_identity" +)] async fn ensure_fanout_identity( transaction: &RegisteredGlobalDbWriteTransaction<'_>, project_id: &str, @@ -696,6 +719,10 @@ impl StoredDeliveryAttempt { } } +#[hotpath::measure( + future = true, + label = "global_db.delivery_settlement.query.read_attempt" +)] async fn read_delivery_attempt( transaction: &RegisteredGlobalDbWriteTransaction<'_>, project_id: &str, @@ -755,6 +782,10 @@ async fn read_delivery_attempt( })) } +#[hotpath::measure( + future = true, + label = "global_db.delivery_settlement.query.count_attempts" +)] async fn count_attempts( transaction: &RegisteredGlobalDbWriteTransaction<'_>, project_id: &str, @@ -784,6 +815,10 @@ async fn count_attempts( }) } +#[hotpath::measure( + future = true, + label = "global_db.delivery_settlement.query.pending_census" +)] async fn read_pending_delivery_census( transaction: &RegisteredGlobalDbWriteTransaction<'_>, project_id: &str, diff --git a/crates/tracedecay-global-db/src/registered_sessions.rs b/crates/tracedecay-global-db/src/registered_sessions.rs index c7f687c5ff..46fd2b8585 100644 --- a/crates/tracedecay-global-db/src/registered_sessions.rs +++ b/crates/tracedecay-global-db/src/registered_sessions.rs @@ -13,12 +13,14 @@ pub(crate) use super::{SessionProviderCoverage, SessionProviderCoverageState}; pub(crate) use tracedecay_sessions::runtime::store_access::SESSION_MESSAGES_AFTER_SQL; impl RegisteredGlobalDb { + #[hotpath::measure(future = true, label = "global_db.sessions.ingest_health")] pub async fn cursor_session_ingest_health(&self) -> Result { SessionStoreAccess::new(self) .cursor_session_ingest_health() .await } + #[hotpath::measure(future = true, label = "global_db.sessions.ingest_health_provider")] pub async fn session_ingest_health_for_provider( &self, provider: Option<&str>, @@ -28,6 +30,7 @@ impl RegisteredGlobalDb { .await } + #[hotpath::measure(future = true, label = "global_db.sessions.has_message")] pub async fn has_session_message( &self, provider: &str, @@ -38,10 +41,12 @@ impl RegisteredGlobalDb { .await } + #[hotpath::measure(future = true, label = "global_db.sessions.message_count")] pub async fn session_message_count(&self) -> Result { SessionStoreAccess::new(self).session_message_count().await } + #[hotpath::measure(future = true, label = "global_db.sessions.project_message_count")] pub async fn session_message_count_for_project( &self, project_key: &str, @@ -51,6 +56,7 @@ impl RegisteredGlobalDb { .await } + #[hotpath::measure(future = true, label = "global_db.sessions.messages_after")] pub async fn session_messages_after( &self, provider: &str, @@ -68,6 +74,7 @@ impl RegisteredGlobalDb { /// `Ok(None)` is the truthful "this store holds no timestamped messages"; /// a failed query or an unreadable timestamp stays an error rather than /// masquerading as an idle store. + #[hotpath::measure(future = true, label = "global_db.sessions.latest_activity")] pub async fn latest_session_activity_secs( &self, ) -> tracedecay_domain::errors::Result> { @@ -78,6 +85,7 @@ impl RegisteredGlobalDb { /// Reads one message by provider and id. `Ok(None)` is truthful absence; /// snapshot, query, and row-decode failures stay typed errors. + #[hotpath::measure(future = true, label = "global_db.sessions.get_message")] pub async fn get_session_message( &self, provider: &str, @@ -92,6 +100,7 @@ impl RegisteredGlobalDb { /// /// `Ok(vec![])` is the truthful "nothing matched"; snapshot, query, and /// row-decode failures are typed errors instead of an empty result page. + #[hotpath::measure(future = true, label = "global_db.sessions.search_messages")] pub async fn search_session_messages( &self, provider: &str, @@ -107,6 +116,7 @@ impl RegisteredGlobalDb { /// Lists each session's latest canonical goal state, newest first. /// Goals with no native timestamp rank after all timestamped goals /// instead of being assigned a fabricated epoch-zero time. + #[hotpath::measure(future = true, label = "global_db.sessions.recent_goals")] pub async fn recent_session_goals( &self, project_key: Option<&str>, @@ -117,6 +127,7 @@ impl RegisteredGlobalDb { .await } + #[hotpath::measure(future = true, label = "global_db.sessions.workflow_facts")] pub async fn workflow_fact_rows( &self, ) -> Result, Option)>, TraceDecayError> { diff --git a/crates/tracedecay-global-db/src/schema_contract/invariants/repair.rs b/crates/tracedecay-global-db/src/schema_contract/invariants/repair.rs index 4fa7154abe..339c76e25f 100644 --- a/crates/tracedecay-global-db/src/schema_contract/invariants/repair.rs +++ b/crates/tracedecay-global-db/src/schema_contract/invariants/repair.rs @@ -25,6 +25,7 @@ struct CommittedCursorCandidate { /// `observation_projection::rebuild::read_observation_frontier` for why a /// caller doing further reads or writes on the same connection depends on /// that. +#[hotpath::measure(future = true, label = "global_db.schema_contract.repair.read_frontier")] async fn read_observation_frontier( conn: &impl QueryExecutor, ) -> tracedecay_domain::errors::Result { @@ -40,6 +41,10 @@ async fn read_observation_frontier( .map_err(|error| global_db_operation_error(OPERATION, error)) } +#[hotpath::measure( + future = true, + label = "global_db.schema_contract.repair.projection_frontier" +)] pub(super) async fn repair_projection_frontier( conn: &impl Executor, trusted_checkpoint: i64, @@ -165,6 +170,10 @@ pub(super) async fn repair_projection_frontier( Ok(repaired_checkpoint) } +#[hotpath::measure( + future = true, + label = "global_db.schema_contract.repair.source_cursors" +)] pub(super) async fn repair_committed_source_cursors( conn: &impl Executor, after_sequence: i64, @@ -216,6 +225,10 @@ pub(super) async fn repair_committed_source_cursors( Ok(()) } +#[hotpath::measure( + future = true, + label = "global_db.schema_contract.repair.scan_committed" +)] async fn latest_committed_source_cursors( conn: &impl QueryExecutor, after_sequence: i64, @@ -284,6 +297,7 @@ fn is_new_generation_frontier( && stored.position() == 0 } +#[hotpath::measure(future = true, label = "global_db.schema_contract.repair.read_cursor")] async fn read_source_cursor( conn: &impl QueryExecutor, source_json: &str, @@ -309,6 +323,7 @@ async fn read_source_cursor( .transpose() } +#[hotpath::measure(future = true, label = "global_db.schema_contract.repair.write_cursor")] async fn write_source_cursor( conn: &impl Executor, candidate: &CommittedCursorCandidate, @@ -329,6 +344,10 @@ async fn write_source_cursor( .map_err(|error| global_db_operation_error(OPERATION, error)) } +#[hotpath::measure( + future = true, + label = "global_db.schema_contract.repair.check_advance_receipt" +)] async fn cursor_has_exact_advance_receipt( conn: &impl QueryExecutor, source_json: &str, @@ -363,6 +382,10 @@ async fn cursor_has_exact_advance_receipt( Ok(false) } +#[hotpath::measure( + future = true, + label = "global_db.schema_contract.repair.cursor_coverage" +)] pub(super) async fn validate_observation_cursor_coverage( conn: &impl QueryExecutor, after_sequence: i64, diff --git a/crates/tracedecay-global-db/src/schema_contract/invariants/rows.rs b/crates/tracedecay-global-db/src/schema_contract/invariants/rows.rs index a9b6cea54f..ad6755734c 100644 --- a/crates/tracedecay-global-db/src/schema_contract/invariants/rows.rs +++ b/crates/tracedecay-global-db/src/schema_contract/invariants/rows.rs @@ -68,6 +68,7 @@ fn bounded_row_audit_invariants() -> impl Iterator { }) } +#[hotpath::measure(future = true, label = "global_db.schema_contract.audit.probe")] pub(super) async fn query_has_rows( conn: &impl QueryExecutor, query: &str, @@ -112,6 +113,7 @@ pub(super) fn encode_authority_json( .map_err(|error| authority_violation(format!("cannot encode {authority}: {error}"))) } +#[hotpath::measure(future = true, label = "global_db.schema_contract.audit.receipt_rows")] pub(super) async fn validate_receipt_authority_rows( conn: &impl QueryExecutor, after_rowid: i64, @@ -201,6 +203,10 @@ pub(super) async fn validate_receipt_authority_rows( } } +#[hotpath::measure( + future = true, + label = "global_db.schema_contract.audit.observation_rows" +)] pub(super) async fn validate_observation_authority_rows( conn: &impl QueryExecutor, after_sequence: i64, @@ -289,6 +295,10 @@ pub(super) async fn validate_observation_authority_rows( } } +#[hotpath::measure( + future = true, + label = "global_db.schema_contract.audit.source_cursor_rows" +)] pub(super) async fn validate_source_cursor_authority_rows( conn: &impl QueryExecutor, ) -> tracedecay_domain::errors::Result<()> { @@ -305,6 +315,10 @@ pub(super) async fn validate_source_cursor_authority_rows( } } +#[hotpath::measure( + future = true, + label = "global_db.schema_contract.audit.source_cursor_chunk" +)] pub(super) async fn validate_source_cursor_authority_chunk( conn: &impl QueryExecutor, mut cursor_rowid: i64, @@ -441,6 +455,7 @@ pub(super) async fn validate_source_cursor_authority_chunk( Ok((cursor_rowid, advance_rowid, page_rows < AUDIT_PAGE_ROWS)) } +#[hotpath::measure(future = true, label = "global_db.schema_contract.audit.mutable_rows")] pub(super) async fn validate_mutable_invariant_rows( conn: &impl QueryExecutor, ) -> tracedecay_domain::errors::Result<()> { diff --git a/crates/tracedecay-global-db/src/schema_contract/invariants/triggers.rs b/crates/tracedecay-global-db/src/schema_contract/invariants/triggers.rs index f6e970a038..dbc00bbe74 100644 --- a/crates/tracedecay-global-db/src/schema_contract/invariants/triggers.rs +++ b/crates/tracedecay-global-db/src/schema_contract/invariants/triggers.rs @@ -1604,6 +1604,10 @@ pub(crate) fn invariant_trigger_names_for_tables(tables: &[&str]) -> Vec<&'stati .collect() } +#[hotpath::measure( + future = true, + label = "global_db.schema_contract.triggers.replace" +)] pub(super) async fn replace_trigger( conn: &impl Executor, trigger: &Trigger, @@ -1616,6 +1620,10 @@ pub(super) async fn replace_trigger( .map_err(|error| global_db_operation_error(OPERATION, error)) } +#[hotpath::measure( + future = true, + label = "global_db.schema_contract.triggers.contracts_intact" +)] pub(super) async fn trigger_contracts_intact( conn: &impl QueryExecutor, ) -> tracedecay_domain::errors::Result { @@ -1629,6 +1637,10 @@ pub(super) async fn trigger_contracts_intact( Ok(true) } +#[hotpath::measure( + future = true, + label = "global_db.schema_contract.triggers.released_v3_intact" +)] pub async fn released_v3_invariant_triggers_intact( conn: &impl QueryExecutor, ) -> tracedecay_domain::errors::Result { @@ -1669,6 +1681,10 @@ async fn trigger_matches( trigger_matches_sql(conn, trigger, trigger.create_sql).await } +#[hotpath::measure( + future = true, + label = "global_db.schema_contract.triggers.match_sql" +)] async fn trigger_matches_sql( conn: &impl QueryExecutor, trigger: &Trigger, @@ -1699,6 +1715,10 @@ async fn trigger_matches_sql( && normalize_trigger_sql(&sql) == normalize_trigger_sql(expected_sql)) } +#[hotpath::measure( + future = true, + label = "global_db.schema_contract.triggers.suspend_immutability" +)] pub async fn suspend_immutability_for_canonical_repair( conn: &impl Executor, ) -> tracedecay_domain::errors::Result<()> { @@ -1718,6 +1738,10 @@ pub async fn suspend_immutability_for_canonical_repair( Ok(()) } +#[hotpath::measure( + future = true, + label = "global_db.schema_contract.triggers.restore_immutability" +)] pub async fn restore_immutability_after_canonical_repair( conn: &impl Executor, ) -> tracedecay_domain::errors::Result<()> { diff --git a/crates/tracedecay-global-db/src/schema_contract/validation.rs b/crates/tracedecay-global-db/src/schema_contract/validation.rs index a41bf643d4..eeef92f218 100644 --- a/crates/tracedecay-global-db/src/schema_contract/validation.rs +++ b/crates/tracedecay-global-db/src/schema_contract/validation.rs @@ -134,6 +134,7 @@ fn normalize_default(value: Option<&str>) -> Option { }) } +#[hotpath::measure(future = true, label = "global_db.schema_contract.validate.table")] async fn validate_table( conn: &impl QueryExecutor, contract: &Table, @@ -265,6 +266,7 @@ fn primary_key_index_matches(actual: &ActualIndex, expected_columns: &[&str]) -> }) } +#[hotpath::measure(future = true, label = "global_db.schema_contract.validate.indexes")] async fn validate_indexes_for_table( conn: &impl QueryExecutor, table: &str, @@ -324,6 +326,7 @@ async fn validate_indexes_for_table( Ok(()) } +#[hotpath::measure(future = true, label = "global_db.schema_contract.validate.trigger")] async fn validate_trigger( conn: &impl QueryExecutor, trigger: &super::invariants::Trigger, @@ -365,6 +368,10 @@ async fn validate_trigger( } } +#[hotpath::measure( + future = true, + label = "global_db.schema_contract.validate.autoincrement" +)] async fn validate_observation_autoincrement( conn: &impl QueryExecutor, ) -> tracedecay_domain::errors::Result<()> { @@ -438,6 +445,10 @@ async fn validate_named_tables_and_indexes( Ok(()) } +#[hotpath::measure( + future = true, + label = "global_db.schema_contract.validate.session_temporal" +)] pub async fn validate_session_temporal_schema_contract( conn: &impl QueryExecutor, table_names: &[&str], @@ -445,6 +456,10 @@ pub async fn validate_session_temporal_schema_contract( validate_named_tables_and_indexes(conn, table_names).await } +#[hotpath::measure( + future = true, + label = "global_db.schema_contract.validate.projection_receipts" +)] pub async fn validate_released_v3_temporal_projection_receipt_contract( conn: &impl QueryExecutor, ) -> tracedecay_domain::errors::Result<()> { @@ -452,6 +467,10 @@ pub async fn validate_released_v3_temporal_projection_receipt_contract( validate_indexes_for_table(conn, SESSION_TEMPORAL_PROJECTION_RECEIPTS_V3.name).await } +#[hotpath::measure( + future = true, + label = "global_db.schema_contract.validate.graph_publication" +)] pub async fn validate_session_graph_publication_schema_contract( conn: &impl QueryExecutor, ) -> tracedecay_domain::errors::Result<()> { @@ -495,6 +514,10 @@ pub async fn validate_session_graph_publication_schema_contract( Ok(()) } +#[hotpath::measure( + future = true, + label = "global_db.schema_contract.query.graph_inventory" +)] async fn read_graph_publication_inventory( conn: &impl QueryExecutor, ) -> tracedecay_domain::errors::Result { @@ -557,12 +580,17 @@ fn belongs_to_graph_publication_namespace(name: &str, table: &str) -> bool { }) } +#[hotpath::measure(future = true, label = "global_db.schema_contract.validate.registry")] pub async fn validate_registry_schema_contract( conn: &impl QueryExecutor, ) -> tracedecay_domain::errors::Result<()> { validate_named_tables_and_indexes(conn, REGISTRY_TABLE_NAMES).await } +#[hotpath::measure( + future = true, + label = "global_db.schema_contract.validate.remote_deletion" +)] pub async fn validate_remote_deletion_schema_contract( conn: &impl QueryExecutor, ) -> tracedecay_domain::errors::Result<()> { @@ -573,6 +601,7 @@ pub async fn validate_remote_deletion_schema_contract( /// /// Transcript, LCM, git-correlation, and workflow-index tables are independently owned by their /// schema modules; this validator intentionally neither claims nor validates those domains. +#[hotpath::measure(future = true, label = "global_db.schema_contract.validate.authority")] pub async fn validate_authority_schema_contract( conn: &impl QueryExecutor, ) -> tracedecay_domain::errors::Result<()> { diff --git a/crates/tracedecay-global-db/src/session_temporal_handle.rs b/crates/tracedecay-global-db/src/session_temporal_handle.rs index f05db3a85f..fe62d7f506 100644 --- a/crates/tracedecay-global-db/src/session_temporal_handle.rs +++ b/crates/tracedecay-global-db/src/session_temporal_handle.rs @@ -101,6 +101,7 @@ impl RegisteredGlobalDb { SessionTemporalAccess::new(self).git_scope_session_ids(filter) } + #[hotpath::measure(future = true, label = "global_db.session_temporal.doctor_health")] pub async fn session_temporal_doctor_health( &self, ) -> tracedecay_session_temporal_store::SessionTemporalHealthReport { @@ -109,6 +110,7 @@ impl RegisteredGlobalDb { .await } + #[hotpath::measure(future = true, label = "global_db.session_temporal.ensure_cursor_key")] pub async fn ensure_active_session_cursor_key_result( &self, ) -> tracedecay_store::SessionStoreResult { @@ -117,6 +119,10 @@ impl RegisteredGlobalDb { .await } + #[hotpath::measure( + future = true, + label = "global_db.session_temporal.load_cursor_key_provider" + )] pub async fn load_session_cursor_key_provider_result( &self, ) -> Result< @@ -128,6 +134,10 @@ impl RegisteredGlobalDb { .await } + #[hotpath::measure( + future = true, + label = "global_db.session_temporal.load_preprovisioned_cursor_key" + )] pub async fn load_preprovisioned_session_cursor_key_provider_result( &self, ) -> Result< @@ -139,6 +149,10 @@ impl RegisteredGlobalDb { .await } + #[hotpath::measure( + future = true, + label = "global_db.session_temporal.pending_refresh_page" + )] pub async fn pending_session_temporal_refresh_page_result( &self, limit: usize, @@ -152,6 +166,10 @@ impl RegisteredGlobalDb { .await } + #[hotpath::measure( + future = true, + label = "global_db.session_temporal.materialize_refresh_batch" + )] pub async fn materialize_session_temporal_refresh_batch_result( &self, recovery: &tracedecay_session_temporal_store::SessionRefreshRecoveryV1, @@ -166,6 +184,7 @@ impl RegisteredGlobalDb { .await } + #[hotpath::measure(future = true, label = "global_db.session_temporal.freeze_snapshot")] pub async fn freeze_session_temporal_snapshot_result( &self, request: tracedecay_store::SessionTemporalSnapshotRequestV1, @@ -175,6 +194,10 @@ impl RegisteredGlobalDb { .await } + #[hotpath::measure( + future = true, + label = "global_db.session_temporal.summary_relations" + )] pub async fn active_session_summary_relations( &self, session_id: &tracedecay_domain::SessionId, @@ -190,6 +213,10 @@ impl RegisteredGlobalDb { .await } + #[hotpath::measure( + future = true, + label = "global_db.session_temporal.apply_relation_projection" + )] pub async fn apply_active_session_relation_projection( &self, session_id: &tracedecay_domain::SessionId, @@ -200,6 +227,10 @@ impl RegisteredGlobalDb { .await } + #[hotpath::measure( + future = true, + label = "global_db.session_temporal.recover_relation_projections" + )] pub async fn recover_pending_session_relation_projections( &self, limit: usize, @@ -210,6 +241,7 @@ impl RegisteredGlobalDb { .await } + #[hotpath::measure(future = true, label = "global_db.session_temporal.refresh_recovery")] pub async fn session_refresh_recovery_result( &self, session_id: &tracedecay_domain::SessionId, @@ -221,6 +253,7 @@ impl RegisteredGlobalDb { .await } + #[hotpath::measure(future = true, label = "global_db.session_temporal.complete_refresh")] pub async fn complete_session_refresh_result( &self, request: tracedecay_store::SessionRefreshCompletionRequestV1, diff --git a/crates/tracedecay-global-db/src/transcript.rs b/crates/tracedecay-global-db/src/transcript.rs index 4e519cf8b5..319cac1ac8 100644 --- a/crates/tracedecay-global-db/src/transcript.rs +++ b/crates/tracedecay-global-db/src/transcript.rs @@ -8,16 +8,19 @@ pub(super) use tracedecay_sessions::runtime::store_access::{ }; impl RegisteredGlobalDb { + #[hotpath::measure(future = true, label = "global_db.transcript.upsert_session")] pub async fn upsert_session(&self, session: &SessionRecord) -> bool { SessionStoreAccess::new(self).upsert_session(session).await } + #[hotpath::measure(future = true, label = "global_db.transcript.get_session")] pub async fn get_session(&self, provider: &str, session_id: &str) -> Option { SessionStoreAccess::new(self) .get_session(provider, session_id) .await } + #[hotpath::measure(future = true, label = "global_db.transcript.get_session_result")] pub async fn get_session_result( &self, provider: &str, @@ -28,6 +31,7 @@ impl RegisteredGlobalDb { .await } + #[hotpath::measure(future = true, label = "global_db.transcript.upsert_batch")] pub async fn upsert_transcript_batch( &self, session: &SessionRecord, @@ -40,6 +44,7 @@ impl RegisteredGlobalDb { .await } + #[hotpath::measure(future = true, label = "global_db.transcript.persist_batch")] pub async fn persist_transcript_batch_result( &self, session: &SessionRecord, @@ -59,6 +64,7 @@ impl RegisteredGlobalDb { .await } + #[hotpath::measure(future = true, label = "global_db.transcript.persist_offset")] pub async fn persist_transcript_offset_result( &self, parse_offset_path: &str, @@ -70,6 +76,10 @@ impl RegisteredGlobalDb { .await } + #[hotpath::measure( + future = true, + label = "global_db.transcript.upsert_projection_batches" + )] pub async fn upsert_transcript_projection_batches( &self, batches: &[TranscriptBatch], @@ -81,10 +91,15 @@ impl RegisteredGlobalDb { .await } + #[hotpath::measure(future = true, label = "global_db.transcript.get_parse_offset")] pub async fn get_parse_offset(&self, path: &str) -> Option { SessionStoreAccess::new(self).get_parse_offset(path).await } + #[hotpath::measure( + future = true, + label = "global_db.transcript.get_parse_offset_result" + )] pub async fn get_parse_offset_result( &self, path: &str, @@ -94,12 +109,14 @@ impl RegisteredGlobalDb { .await } + #[hotpath::measure(future = true, label = "global_db.transcript.set_parse_offset")] pub async fn set_parse_offset(&self, path: &str, offset: ParseOffset) -> Result<(), String> { SessionStoreAccess::new(self) .set_parse_offset(path, offset) .await } + #[hotpath::measure(future = true, label = "global_db.transcript.advance_parse_offset")] pub async fn advance_parse_offset_result( &self, path: &str, @@ -110,6 +127,7 @@ impl RegisteredGlobalDb { .await } + #[hotpath::measure(future = true, label = "global_db.transcript.replace_parse_offset")] pub async fn replace_parse_offset_result( &self, path: &str, @@ -121,6 +139,10 @@ impl RegisteredGlobalDb { .await } + #[hotpath::measure( + future = true, + label = "global_db.transcript.replace_parse_offset_pair" + )] pub async fn replace_parse_offset_pair_result( &self, first: (&str, ParseOffset, ParseOffset),