From 0f04ba51bc3fe2fa358cd3b3adbd3065f7e567da Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Mon, 31 Aug 2026 17:45:21 +0000 Subject: [PATCH] feat(hotpath): fill session-temporal-store query/persist gaps Co-authored-by: Zack Jackson --- .../tracedecay-session-temporal-store/src/cursor_keys.rs | 4 ++++ crates/tracedecay-session-temporal-store/src/execution.rs | 2 ++ crates/tracedecay-session-temporal-store/src/hydration.rs | 7 +++++++ .../src/operations/message_anchor.rs | 3 +++ .../src/projection/persist.rs | 8 ++++++++ crates/tracedecay-session-temporal-store/src/query.rs | 4 ++++ .../src/retrieval/candidates.rs | 2 ++ crates/tracedecay-session-temporal-store/src/sql.rs | 1 + crates/tracedecay-session-temporal-store/src/store.rs | 7 +++++++ 9 files changed, 38 insertions(+) diff --git a/crates/tracedecay-session-temporal-store/src/cursor_keys.rs b/crates/tracedecay-session-temporal-store/src/cursor_keys.rs index 102075f963..6ad51a629a 100644 --- a/crates/tracedecay-session-temporal-store/src/cursor_keys.rs +++ b/crates/tracedecay-session-temporal-store/src/cursor_keys.rs @@ -59,6 +59,7 @@ pub struct GlobalDbCursorKeyProvider { authenticators: Vec<(SignedCursorKeyRefV1, InMemoryCursorAuthenticator)>, } +#[hotpath::measure(future = true, label = "session_temporal.cursor_key.provision")] pub(super) async fn ensure_active_session_cursor_key_in_transaction( transaction: &impl crate::handle::SessionTemporalExec, ) -> SessionStoreResult { @@ -195,6 +196,7 @@ pub(super) async fn ensure_active_session_cursor_key_in_transaction( } impl GlobalDbCursorKeyProvider { + #[hotpath::measure(future = true, label = "session_temporal.cursor_key.load_active")] pub async fn from_registered_active( read: &DatabaseEngineReadSnapshot, ) -> Result { @@ -252,6 +254,7 @@ impl GlobalDbCursorKeyProvider { Self::from_registered_key_ref_at(read, expected, now_micros().0).await } + #[hotpath::measure(future = true, label = "session_temporal.cursor_key.load")] async fn from_registered_key_ref_at( read: &DatabaseEngineReadSnapshot, expected: SignedCursorKeyRefV1, @@ -307,6 +310,7 @@ impl GlobalDbCursorKeyProvider { authenticators, }) } + #[hotpath::measure(label = "session_temporal.cursor_key.keyring")] pub fn retrieval_keyring( &self, privacy_domain: PrivacyDomainId, diff --git a/crates/tracedecay-session-temporal-store/src/execution.rs b/crates/tracedecay-session-temporal-store/src/execution.rs index 4ffd147674..771e3c8f22 100644 --- a/crates/tracedecay-session-temporal-store/src/execution.rs +++ b/crates/tracedecay-session-temporal-store/src/execution.rs @@ -131,6 +131,7 @@ impl AuthorizedTemporalExecutionRequest { } } + #[hotpath::measure(label = "session_temporal.execution.validate_report")] pub fn validates_report(&self, report: &SessionTemporalExecutionReport) -> bool { let snapshot = &report.result().snapshot; let actual = snapshot.request(); @@ -319,6 +320,7 @@ pub struct AuthorizedTaskSessionExecutionRequestV1 { impl AuthorizedTaskSessionExecutionRequestV1 { #[allow(clippy::too_many_arguments)] + #[hotpath::measure(label = "session_temporal.execution.authorize_task_session")] pub fn new( temporal: AuthorizedTemporalExecutionRequest, retrieval: RetrievalRequest, diff --git a/crates/tracedecay-session-temporal-store/src/hydration.rs b/crates/tracedecay-session-temporal-store/src/hydration.rs index f9e049645d..6a8d7a42c6 100644 --- a/crates/tracedecay-session-temporal-store/src/hydration.rs +++ b/crates/tracedecay-session-temporal-store/src/hydration.rs @@ -126,6 +126,7 @@ impl SessionTemporalHydrationAdapter { } impl SessionTemporalHydrationAdapter { + #[hotpath::measure(future = true, label = "session_temporal.hydrate.authorize")] async fn authorize( &self, snapshot: &TemporalExecutionSnapshot, @@ -142,6 +143,7 @@ impl SessionTemporalHydrationAdapter { } } + #[hotpath::measure(future = true, label = "session_temporal.hydrate.read_authorized")] async fn read_after_recheck( &self, snapshot: &TemporalExecutionSnapshot, @@ -345,6 +347,7 @@ impl<'snapshot> SessionTemporalHydrationAdapter, snapshot: &TemporalExecutionSnapshot, @@ -621,6 +624,7 @@ impl TemporalHydrationBackend for GlobalDbHydrationBackend<'_> { } #[allow(clippy::too_many_arguments)] +#[hotpath::measure(future = true, label = "session_temporal.hydrate.read_occurrence")] async fn read_occurrence_content( conn: &TemporalSqlRead<'_>, storage_root: &Path, @@ -815,6 +819,7 @@ async fn resolve_current( )) } +#[hotpath::measure(future = true, label = "session_temporal.hydrate.resolve_occurrence")] async fn resolve_occurrence( conn: &TemporalSqlRead<'_>, snapshot: &TemporalExecutionSnapshot, @@ -933,6 +938,7 @@ async fn resolve_occurrence( }))) } +#[hotpath::measure(future = true, label = "session_temporal.hydrate.resolve_summary")] async fn resolve_summary( conn: &TemporalSqlRead<'_>, relation_authority: Option<&SessionHydrationRelationAuthority<'_>>, @@ -1081,6 +1087,7 @@ async fn resolve_summary( } #[allow(clippy::too_many_arguments)] +#[hotpath::measure(future = true, label = "session_temporal.hydrate.summary_evidence")] async fn summary_has_provider_evidence( conn: &TemporalSqlRead<'_>, relation_store: &SessionRelationGraphStore, diff --git a/crates/tracedecay-session-temporal-store/src/operations/message_anchor.rs b/crates/tracedecay-session-temporal-store/src/operations/message_anchor.rs index 589301ed93..3ea187f9cb 100644 --- a/crates/tracedecay-session-temporal-store/src/operations/message_anchor.rs +++ b/crates/tracedecay-session-temporal-store/src/operations/message_anchor.rs @@ -31,6 +31,7 @@ pub(super) type ResolvedMessageAnchor = (String, bool, i64); /// `Ok(None)` means the message has no canonical anchor in this store at all — /// the only case in which the publication falls back to a legacy compatibility /// anchor. +#[hotpath::measure(future = true, label = "session_temporal.publication.resolve_anchor")] pub(super) async fn resolve_message_anchor( conn: &impl crate::handle::SessionTemporalExec, provider: &str, @@ -47,6 +48,7 @@ pub(super) async fn resolve_message_anchor( } /// Resolves through the message's occurrence in the active temporal generation. +#[hotpath::measure(future = true, label = "session_temporal.publication.anchor_occurrence")] async fn resolve_materialized_occurrence( conn: &impl crate::handle::SessionTemporalExec, provider: &str, @@ -124,6 +126,7 @@ async fn resolve_materialized_occurrence( /// Resolves through the durable observation authority, which retains the /// exact-observation anchor before any generation materializes the occurrence. +#[hotpath::measure(future = true, label = "session_temporal.publication.anchor_observation")] async fn resolve_canonical_observation( conn: &impl crate::handle::SessionTemporalExec, provider: &str, diff --git a/crates/tracedecay-session-temporal-store/src/projection/persist.rs b/crates/tracedecay-session-temporal-store/src/projection/persist.rs index 17b0a0337c..8897887c8b 100644 --- a/crates/tracedecay-session-temporal-store/src/projection/persist.rs +++ b/crates/tracedecay-session-temporal-store/src/projection/persist.rs @@ -41,6 +41,7 @@ fn observation_envelope( .map_err(|error| storage(PERSIST_OPERATION, error)) } +#[hotpath::measure(future = true, label = "session_temporal.projection.record_count")] pub async fn session_temporal_projection_record_count( conn: &impl crate::handle::SessionTemporalQuery, session_id: &SessionId, @@ -266,6 +267,7 @@ async fn canonical_occurrence_projection( }) } +#[hotpath::measure(future = true, label = "session_temporal.persist.occurrences")] pub(super) async fn persist_occurrences( conn: &impl crate::handle::SessionTemporalExec, batch: &SessionTemporalProjectionBatchV1, @@ -509,6 +511,7 @@ async fn persist_occurrence( } /// Derives the canonical occurrence for one already-resolved projection output. +#[hotpath::measure(future = true, label = "session_temporal.projection.canonical_occurrence")] pub(super) async fn canonical_occurrence( conn: &impl crate::handle::SessionTemporalQuery, observation: &tracedecay_domain::DurableObservationV1, @@ -703,6 +706,7 @@ pub(super) async fn ensure_agent( Ok(()) } +#[hotpath::measure(future = true, label = "session_temporal.projection.require_occurrence")] pub(super) async fn require_exact_occurrence( conn: &impl crate::handle::SessionTemporalExec, batch: &SessionTemporalProjectionBatchV1, @@ -791,6 +795,7 @@ pub(super) async fn require_exact_occurrence( Ok(()) } +#[hotpath::measure(future = true, label = "session_temporal.projection.validate_copy")] pub(super) async fn validate_copy( conn: &impl crate::handle::SessionTemporalExec, batch: &SessionTemporalProjectionBatchV1, @@ -908,6 +913,7 @@ pub(super) async fn occurrence_observation_and_anchor( Ok((observation, envelope, anchor_id)) } +#[hotpath::measure(future = true, label = "session_temporal.projection.validate_copy_proof")] pub(super) async fn validate_copy_proof( conn: &impl crate::handle::SessionTemporalExec, batch: &SessionTemporalProjectionBatchV1, @@ -1005,6 +1011,7 @@ pub(super) async fn validate_copy_proof( Ok(()) } +#[hotpath::measure(future = true, label = "session_temporal.persist.assertion")] pub(super) async fn persist_assertion( conn: &impl crate::handle::SessionTemporalExec, batch: &SessionTemporalProjectionBatchV1, @@ -1094,6 +1101,7 @@ pub(super) async fn persist_assertion( Ok(inserted) } +#[hotpath::measure(future = true, label = "session_temporal.projection.validate_assertion")] pub(super) async fn validate_assertion( conn: &impl crate::handle::SessionTemporalExec, batch: &SessionTemporalProjectionBatchV1, diff --git a/crates/tracedecay-session-temporal-store/src/query.rs b/crates/tracedecay-session-temporal-store/src/query.rs index 8dd5a56501..82e081e73c 100644 --- a/crates/tracedecay-session-temporal-store/src/query.rs +++ b/crates/tracedecay-session-temporal-store/src/query.rs @@ -77,6 +77,7 @@ pub(super) fn encode_watermarks( .map_err(|error| storage(operation, error)) } +#[hotpath::measure(future = true, label = "session_temporal.query.read_generation")] pub(super) async fn read_generation( conn: &impl crate::handle::SessionTemporalQuery, session_id: &SessionId, @@ -106,6 +107,7 @@ fn decode_generation(row: &Row, operation: &'static str) -> SessionStoreResult, candidates: &[RankingCandidate], @@ -343,6 +344,7 @@ pub(super) async fn require_candidate_root_authority( } #[allow(clippy::too_many_arguments)] +#[hotpath::measure(future = true, label = "session_temporal.query.candidate_clause")] pub(super) async fn query_candidate_clause( conn: &TemporalSqlRead<'_>, scope: &TemporalRetrievalScope, diff --git a/crates/tracedecay-session-temporal-store/src/sql.rs b/crates/tracedecay-session-temporal-store/src/sql.rs index 54e7175338..2b086c3685 100644 --- a/crates/tracedecay-session-temporal-store/src/sql.rs +++ b/crates/tracedecay-session-temporal-store/src/sql.rs @@ -102,6 +102,7 @@ impl<'a> TemporalSqlRead<'a> { Self::Registered(read) } + #[hotpath::measure(future = true, label = "session_temporal.sql.query")] pub(super) async fn query

(&self, sql: &str, params: P) -> engine::Result where P: engine::IntoParams, diff --git a/crates/tracedecay-session-temporal-store/src/store.rs b/crates/tracedecay-session-temporal-store/src/store.rs index e417516400..b1a2866987 100644 --- a/crates/tracedecay-session-temporal-store/src/store.rs +++ b/crates/tracedecay-session-temporal-store/src/store.rs @@ -68,6 +68,7 @@ impl<'a, D: SessionTemporalRegisteredDb + Sync> GlobalDbSessionTemporalStore<'a, SessionTemporalAccess::new(self.db) } + #[hotpath::measure(future = true, label = "session_temporal.store.persist_refresh_batch")] pub async fn persist_session_refresh_projection_batch( &self, progress: SessionRefreshProgressV1, @@ -81,6 +82,10 @@ impl<'a, D: SessionTemporalRegisteredDb + Sync> GlobalDbSessionTemporalStore<'a, .await } + #[hotpath::measure( + future = true, + label = "session_temporal.store.persist_refresh_batch_controlled" + )] pub async fn persist_session_refresh_projection_batch_controlled( &self, progress: SessionRefreshProgressV1, @@ -99,6 +104,7 @@ impl<'a, D: SessionTemporalRegisteredDb + Sync> GlobalDbSessionTemporalStore<'a, .await } + #[hotpath::measure(future = true, label = "session_temporal.store.refresh_recovery")] pub async fn session_refresh_recovery( &self, session_id: &tracedecay_domain::SessionId, @@ -108,6 +114,7 @@ impl<'a, D: SessionTemporalRegisteredDb + Sync> GlobalDbSessionTemporalStore<'a, .await } + #[hotpath::measure(future = true, label = "session_temporal.store.refresh_running")] pub async fn running_session_refreshes( &self, ) -> SessionStoreResult> {