diff --git a/Cargo.lock b/Cargo.lock index 94d55ed6005..7cdc3ac11cc 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3613,6 +3613,7 @@ dependencies = [ "error-stack", "hash-codec", "hash-codegen", + "hash-graph-temporal-versioning", "indoc", "insta", "postgres-types", diff --git a/libs/@local/graph/authorization/Cargo.toml b/libs/@local/graph/authorization/Cargo.toml index a2ca133044a..502dd975b5f 100644 --- a/libs/@local/graph/authorization/Cargo.toml +++ b/libs/@local/graph/authorization/Cargo.toml @@ -8,8 +8,9 @@ authors.workspace = true [dependencies] # Public workspace dependencies -error-stack = { workspace = true, public = true, features = ["unstable"] } -type-system = { workspace = true, public = true } +error-stack = { workspace = true, public = true, features = ["unstable"] } +hash-graph-temporal-versioning = { workspace = true, public = true } +type-system = { workspace = true, public = true } # Public third-party dependencies postgres-types = { workspace = true, public = true, features = ["derive", "with-uuid-1"], optional = true } diff --git a/libs/@local/graph/authorization/package.json b/libs/@local/graph/authorization/package.json index 9959bf15033..edb9bca1607 100644 --- a/libs/@local/graph/authorization/package.json +++ b/libs/@local/graph/authorization/package.json @@ -26,7 +26,8 @@ "dependencies": { "@blockprotocol/type-system-rs": "workspace:*", "@rust/error-stack": "workspace:*", - "@rust/hash-codec": "workspace:*" + "@rust/hash-codec": "workspace:*", + "@rust/hash-graph-temporal-versioning": "workspace:*" }, "devDependencies": { "@local/tsconfig": "workspace:*", diff --git a/libs/@local/graph/authorization/src/policies/components.rs b/libs/@local/graph/authorization/src/policies/components.rs index 1dcdac1d281..245e360836a 100644 --- a/libs/@local/graph/authorization/src/policies/components.rs +++ b/libs/@local/graph/authorization/src/policies/components.rs @@ -2,6 +2,7 @@ use alloc::borrow::Cow; use std::collections::{HashMap, HashSet}; use error_stack::{Report, ResultExt as _}; +use hash_graph_temporal_versioning::Timestamp; use type_system::{ knowledge::entity::{ EntityId, @@ -65,6 +66,7 @@ pub enum MergePolicies { #[derive(Debug)] pub struct PolicyComponents { actor_id: Option, + timestamp: Timestamp<()>, is_instance_admin: bool, policies: Vec, tracked_actions: HashMap>, @@ -86,6 +88,18 @@ impl PolicyComponents { self.actor_id } + /// Returns the store's clock reading captured while building these components. + /// + /// Components always carry a reading, whether or not the operation goes on to use one. + /// + /// This value is the time authority for the operation these components were built for: it + /// should be used to resolve temporal axes and to derive written timestamps, so that all + /// timestamps within one operation agree with each other and with the store's clock. + #[must_use] + pub const fn timestamp(&self) -> Timestamp<()> { + self.timestamp + } + /// Returns `true` if the actor is an instance admin. /// /// Instance admins have elevated privileges, such as bypassing filter protection @@ -319,6 +333,7 @@ impl PolicyComponents { pub struct PolicyComponentsBuilder<'a, S> { store: &'a S, actor: AuthenticatedActor, + timestamp: Option>, context: ContextBuilder, entity_type_ids: HashSet>, property_type_ids: HashSet>, @@ -334,6 +349,7 @@ impl<'a, S> PolicyComponentsBuilder<'a, S> { Self { store, actor: AuthenticatedActor::Uuid(ActorEntityUuid::public_actor()), + timestamp: None, context: ContextBuilder::default(), entity_type_ids: HashSet::new(), property_type_ids: HashSet::new(), @@ -353,6 +369,28 @@ impl<'a, S> PolicyComponentsBuilder<'a, S> { self } + /// Provides the store's clock reading captured for the surrounding operation. + /// + /// Callers which already hold a clock reading from the store — e.g. because an earlier + /// statement of the same operation returned one — should pass it here so the components share + /// the operation's timestamp. When absent, a reading is captured while building the + /// components: on the actor lookup where the actor still needs resolving, otherwise through a + /// statement of its own. + pub const fn set_timestamp(&mut self, timestamp: Timestamp<()>) { + self.timestamp = Some(timestamp); + } + + /// Provides the store's clock reading captured for the surrounding operation. + /// + /// See [`set_timestamp`] for details. + /// + /// [`set_timestamp`]: Self::set_timestamp + #[must_use] + pub const fn with_timestamp(mut self, timestamp: Timestamp<()>) -> Self { + self.set_timestamp(timestamp); + self + } + pub fn add_entity_type_id(&mut self, entity_type: &'a VersionedUrl) { self.entity_type_ids.insert(Cow::Borrowed(entity_type)); } @@ -584,11 +622,31 @@ where #[tracing::instrument(level = "info", skip(self))] fn into_future(mut self) -> Self::IntoFuture { async move { - let actor_id = match self.actor { - AuthenticatedActor::Id(actor_id) => Some(actor_id), - AuthenticatedActor::Uuid(actor_uuid) => self + // The components always carry a clock reading, so each arm resolves one: a reading the + // caller supplied is taken as-is, an actor which still needs resolving has the reading + // captured by its lookup statement, and only an already-resolved actor without a + // supplied reading pays for a statement of its own. + let (actor_id, timestamp) = match (self.actor, self.timestamp) { + (AuthenticatedActor::Id(actor_id), Some(timestamp)) => (Some(actor_id), timestamp), + (AuthenticatedActor::Id(actor_id), None) => ( + Some(actor_id), + self.store + .current_timestamp() + .await + .change_context(ContextCreationError::StoreError)?, + ), + (AuthenticatedActor::Uuid(actor_uuid), Some(timestamp)) => ( + self.store + .determine_actor(actor_uuid) + .await + .change_context(ContextCreationError::DetermineActor { + actor_id: actor_uuid, + })?, + timestamp, + ), + (AuthenticatedActor::Uuid(actor_uuid), None) => self .store - .determine_actor(actor_uuid) + .determine_actor_with_timestamp(actor_uuid) .await .change_context(ContextCreationError::DetermineActor { actor_id: actor_uuid, @@ -707,6 +765,7 @@ where let mut policy_components = PolicyComponents { actor_id, + timestamp, is_instance_admin: self.context.is_instance_admin(), policies, tracked_actions: actions.iter().map(|action| (*action, None)).collect(), @@ -735,6 +794,7 @@ where mod tests { use std::collections::{HashMap, HashSet}; + use hash_graph_temporal_versioning::Timestamp; use type_system::{knowledge::entity::id::EntityUuid, principal::actor::ActorId}; use uuid::Uuid; @@ -772,6 +832,7 @@ mod tests { let mut policy_components = PolicyComponents { actor_id: None, + timestamp: Timestamp::UNIX_EPOCH, is_instance_admin: false, policies, tracked_actions: HashMap::from([(ActionName::View, None)]), @@ -826,6 +887,7 @@ mod tests { let policy_components_without_optimization = PolicyComponents { actor_id: None, + timestamp: Timestamp::UNIX_EPOCH, is_instance_admin: false, policies: policies_without_optimization, tracked_actions: HashMap::from([(ActionName::View, None)]), @@ -854,6 +916,7 @@ mod tests { let mut policy_components_with_optimization = PolicyComponents { actor_id: None, + timestamp: Timestamp::UNIX_EPOCH, is_instance_admin: false, policies: policies_with_optimization, tracked_actions: HashMap::new(), @@ -890,6 +953,7 @@ mod tests { actor_id: Some(ActorId::User(type_system::principal::actor::UserId::new( Uuid::new_v4(), ))), + timestamp: Timestamp::UNIX_EPOCH, is_instance_admin: false, policies: vec![policy], tracked_actions: HashMap::from([(ActionName::View, None)]), diff --git a/libs/@local/graph/authorization/src/policies/store/error.rs b/libs/@local/graph/authorization/src/policies/store/error.rs index 3953d9e1579..d749ed0090e 100644 --- a/libs/@local/graph/authorization/src/policies/store/error.rs +++ b/libs/@local/graph/authorization/src/policies/store/error.rs @@ -272,6 +272,12 @@ pub enum DetermineActorError { impl Error for DetermineActorError {} +#[derive(Debug, derive_more::Display)] +#[display("Could not read the current timestamp from the store")] +pub struct CurrentTimestampError; + +impl Error for CurrentTimestampError {} + #[derive(Debug, derive_more::Display)] #[display("Could not build principal context for actor with ID `{actor_id}`")] pub enum BuildPrincipalContextError { diff --git a/libs/@local/graph/authorization/src/policies/store/mod.rs b/libs/@local/graph/authorization/src/policies/store/mod.rs index 94420fb68f9..54477863892 100644 --- a/libs/@local/graph/authorization/src/policies/store/mod.rs +++ b/libs/@local/graph/authorization/src/policies/store/mod.rs @@ -8,6 +8,7 @@ use std::collections::{ }; use error_stack::{Report, bail, ensure}; +use hash_graph_temporal_versioning::Timestamp; use type_system::{ knowledge::{entity::id::EntityEditionId, property::PropertyObjectWithMetadata}, ontology::VersionedUrl, @@ -22,10 +23,10 @@ use uuid::Uuid; use self::error::{ ActorCreationError, BuildDataTypeContextError, BuildEntityContextError, BuildEntityTypeContextError, BuildPrincipalContextError, BuildPropertyTypeContextError, - ContextCreationError, CreatePolicyError, DetermineActorError, EnsureSystemPoliciesError, - GetPoliciesError, GetSystemAccountError, PolicyStoreError, RemovePolicyError, - RoleAssignmentError, TeamCreationError, TeamRoleCreationError, TeamRoleError, - UpdatePolicyError, WebCreationError, WebRoleCreationError, WebRoleError, + ContextCreationError, CreatePolicyError, CurrentTimestampError, DetermineActorError, + EnsureSystemPoliciesError, GetPoliciesError, GetSystemAccountError, PolicyStoreError, + RemovePolicyError, RoleAssignmentError, TeamCreationError, TeamRoleCreationError, + TeamRoleError, UpdatePolicyError, WebCreationError, WebRoleCreationError, WebRoleError, }; use super::{ ContextBuilder, Effect, Policy, PolicyId, ResolvedPolicy, @@ -534,6 +535,42 @@ pub trait PrincipalStore { actor_entity_uuid: ActorEntityUuid, ) -> Result, Report>; + /// Determines the type of an actor by its ID, additionally reading the store's clock. + /// + /// The timestamp is captured by the same statement that looks up the actor, so both are + /// obtained in a single round trip. + /// + /// Where an actor must be resolved but no clock reading is required, [`determine_actor`] + /// performs the lookup alone. + /// + /// [`determine_actor`]: Self::determine_actor + /// + /// # Errors + /// + /// - [`StoreError`] if a database error occurs + /// - [`ActorNotFound`] if the actor with the given ID doesn't exist + /// + /// [`StoreError`]: DetermineActorError::StoreError + /// [`ActorNotFound`]: DetermineActorError::ActorNotFound + async fn determine_actor_with_timestamp( + &self, + actor_entity_uuid: ActorEntityUuid, + ) -> Result<(Option, Timestamp<()>), Report>; + + /// Reads the current timestamp from the store's clock. + /// + /// The store's clock is the single time authority for all query-relevant timestamps, so + /// callers which need a timestamp — e.g. to resolve temporal axes or to stamp written + /// records — should use this (or a value derived from another statement's clock reading, such + /// as [`determine_actor_with_timestamp`]) rather than the host's clock. + /// + /// [`determine_actor_with_timestamp`]: Self::determine_actor_with_timestamp + /// + /// # Errors + /// + /// - [`CurrentTimestampError`] if a database error occurs + async fn current_timestamp(&self) -> Result, Report>; + /// Builds a context used to evaluate policies for an actor. /// /// # Errors diff --git a/libs/@local/graph/postgres-store/src/snapshot/entity/batch.rs b/libs/@local/graph/postgres-store/src/snapshot/entity/batch.rs index 280facff88e..ec1734267a9 100644 --- a/libs/@local/graph/postgres-store/src/snapshot/entity/batch.rs +++ b/libs/@local/graph/postgres-store/src/snapshot/entity/batch.rs @@ -2,6 +2,7 @@ use std::collections::HashMap; use error_stack::{Report, ResultExt as _, ensure}; use futures::{StreamExt as _, TryStreamExt as _, stream}; +use hash_graph_authorization::policies::store::PrincipalStore as _; use hash_graph_store::{ entity::{EntityValidationReport, ValidateEntityComponents}, error::InsertionError, @@ -257,10 +258,15 @@ where .await .change_context(InsertionError)?; + let timestamp = postgres_client + .current_timestamp() + .await + .change_context(InsertionError)?; let validator_provider = StoreProvider { store: postgres_client, cache: Box::new(StoreCache::default()), policy_components: None, + timestamp, }; let mut edition_ids_updates = Vec::new(); diff --git a/libs/@local/graph/postgres-store/src/store/postgres/knowledge/entity/delete.rs b/libs/@local/graph/postgres-store/src/store/postgres/knowledge/entity/delete.rs index ad02ed11990..5365821b82a 100644 --- a/libs/@local/graph/postgres-store/src/store/postgres/knowledge/entity/delete.rs +++ b/libs/@local/graph/postgres-store/src/store/postgres/knowledge/entity/delete.rs @@ -3,6 +3,7 @@ use std::collections::{HashMap, HashSet, hash_map::Entry}; use error_stack::{Report, ResultExt as _}; use futures::TryStreamExt as _; +use hash_graph_authorization::policies::store::PrincipalStore as _; use hash_graph_store::{ entity::{ DeleteEntitiesParams, DeletionScope, DeletionSummary, EntityQueryPath, LinkDeletionBehavior, @@ -896,12 +897,17 @@ where /// /// [`IncomingLinksExist`]: DeletionError::IncomingLinksExist /// [`Store`]: DeletionError::Store + #[expect(clippy::too_many_lines)] pub(super) async fn execute_entity_deletion( &mut self, actor_id: ActorEntityUuid, params: DeleteEntitiesParams<'_>, ) -> Result> { - let transaction_time = Timestamp::::now(); + let transaction_time = Timestamp::::from_anonymous( + self.current_timestamp() + .await + .change_context(DeletionError::Store)?, + ); let decision_time = params .decision_time .unwrap_or_else(|| transaction_time.cast()); diff --git a/libs/@local/graph/postgres-store/src/store/postgres/knowledge/entity/mod.rs b/libs/@local/graph/postgres-store/src/store/postgres/knowledge/entity/mod.rs index 5a36120b4a8..e1740f9244b 100644 --- a/libs/@local/graph/postgres-store/src/store/postgres/knowledge/entity/mod.rs +++ b/libs/@local/graph/postgres-store/src/store/postgres/knowledge/entity/mod.rs @@ -33,7 +33,7 @@ use hash_graph_store::{ SearchEntitiesResponse, SummarizeEntitiesParams, SummarizeEntitiesResponse, UpdateEntityEmbeddingsParams, ValidateEntityComponents, ValidateEntityParams, }, - entity_type::{EntityTypeStore as _, IncludeEntityTypeOption}, + entity_type::IncludeEntityTypeOption, error::{ CheckPermissionError, ClusterError, DeletionError, InsertionError, QueryError, UpdateError, }, @@ -686,14 +686,15 @@ where Ok(QueryEntitiesResponse { closed_multi_entity_types: if params.include_entity_types.is_some() { Some( - self.get_closed_multi_entity_types( + self.get_closed_multi_entity_types_impl( policy_components .actor_id() .map_or_else(ActorEntityUuid::public_actor, ActorEntityUuid::from), entities .iter() .map(|entity| entity.metadata.entity_type_ids.clone()), - QueryTemporalAxesUnresolved::live_only(), + &QueryTemporalAxesUnresolved::live_only() + .resolve_with(policy_components.timestamp()), None, ) .await? @@ -767,7 +768,9 @@ where .await .change_context(QueryError)?; - let temporal_axes = params.temporal_axes.resolve(); + let temporal_axes = params + .temporal_axes + .resolve_with(policy_components.timestamp()); let mut response = self .read_entities_impl(¶ms, &temporal_axes, &policy_components) @@ -797,6 +800,7 @@ where temporal_axes: params.temporal_axes, include_drafts: params.include_drafts, }, + Some(policy_components.timestamp()), ) .await .change_context(QueryError)?; @@ -843,7 +847,8 @@ where .await .change_context(QueryError)?; - let temporal_axes = request.temporal_axes.resolve(); + let timestamp = policy_components.timestamp(); + let temporal_axes = request.temporal_axes.resolve_with(timestamp); let time_axis = temporal_axes.variable_time_axis(); let QueryEntitiesResponse { @@ -943,14 +948,14 @@ where Ok(QueryEntitySubgraphResponse { closed_multi_entity_types: if request.include_entity_types.is_some() { Some( - self.get_closed_multi_entity_types( + self.get_closed_multi_entity_types_impl( actor_id, subgraph .vertices .entities .values() .map(|entity| entity.metadata.entity_type_ids.clone()), - QueryTemporalAxesUnresolved::live_only(), + &QueryTemporalAxesUnresolved::live_only().resolve_with(timestamp), None, ) .await? @@ -1010,6 +1015,7 @@ where temporal_axes: request.temporal_axes, include_drafts: request.include_drafts, }, + Some(timestamp), ) .await .change_context(QueryError)?; @@ -1105,6 +1111,11 @@ where /// Returns the entity editions among `params.entity_ids` on which `authenticated_actor` may /// perform `params.action`. /// + /// `timestamp` is the store's clock reading of the surrounding operation. It is forwarded to + /// the policy components, so the permission check resolves its temporal axes to the same point + /// in time as the operation's other statements without taking a further clock reading. When + /// absent, the reading captured while building this check's policy components is used. + /// /// This is inherent rather than only an [`EntityStore`] method because the snapshot-consistent /// read implementations invoke it on the [`InTransaction`] store, where the [`EntityStore`] /// impl — bounded on [`BeginReadOnlyTransaction`] — is not available. @@ -1117,8 +1128,24 @@ where &self, authenticated_actor: AuthenticatedActor, params: HasPermissionForEntitiesParams<'_>, + timestamp: Option>, ) -> Result>, Report> { - let temporal_axes = params.temporal_axes.resolve(); + let mut policy_components_builder = PolicyComponents::builder(self) + .with_actor(authenticated_actor) + .with_action(params.action, MergePolicies::Yes); + if let Some(timestamp) = timestamp { + // An already-resolved actor has no lookup statement for the builder to capture a + // reading on, so forwarding the operation's own keeps it from querying the clock + // again — and keeps the permission check on the instant its caller reads at. + policy_components_builder.set_timestamp(timestamp); + } + let policy_components = policy_components_builder + .await + .change_context(CheckPermissionError::BuildPolicyContext)?; + + let temporal_axes = params + .temporal_axes + .resolve_with(policy_components.timestamp()); let mut compiler = SelectCompiler::new(Some(&temporal_axes), params.include_drafts); let entity_uuids = params @@ -1138,12 +1165,6 @@ where compiler .add_filter(&entity_filter) .change_context(CheckPermissionError::CompileFilter)?; - - let policy_components = PolicyComponents::builder(self) - .with_actor(authenticated_actor) - .with_action(params.action, MergePolicies::Yes) - .await - .change_context(CheckPermissionError::BuildPolicyContext)?; let policy_filter = Filter::::for_policies( policy_components.extract_filter_policies(params.action), policy_components.actor_id(), @@ -1208,7 +1229,6 @@ where actor_uuid: ActorEntityUuid, params: Vec, ) -> Result, Report> { - let transaction_time = Timestamp::::now().remove_nanosecond(); let mut entity_edition_ids = Vec::with_capacity(params.len()); let mut entity_id_rows = Vec::with_capacity(params.len()); @@ -1230,13 +1250,19 @@ where .await .change_context(InsertionError)?; - let actor_id = transaction - .determine_actor(actor_uuid) + let (actor_id, timestamp) = transaction + .determine_actor_with_timestamp(actor_uuid) .await - .change_context(InsertionError)? - .ok_or_else(|| Report::new(InsertionError).attach("Actor not found"))?; + .change_context(InsertionError)?; + let actor_id = + actor_id.ok_or_else(|| Report::new(InsertionError).attach("Actor not found"))?; + // The transaction time is read from the database clock by the transaction's first + // statement, so it is consistent with the timestamps of concurrently committed data + // while every statement of this operation shares the single value. + let transaction_time = Timestamp::::from_anonymous(timestamp); - let mut policy_components_builder = PolicyComponents::builder(&transaction); + let mut policy_components_builder = + PolicyComponents::builder(&transaction).with_timestamp(timestamp); let mut entity_ids = Vec::with_capacity(params.len()); @@ -2036,7 +2062,9 @@ where ¶ms.filter }; - let temporal_axes = params.temporal_axes.resolve(); + let temporal_axes = params + .temporal_axes + .resolve_with(policy_components.timestamp()); let mut compiler = SelectCompiler::new(Some(&temporal_axes), params.include_drafts); compiler .add_filter(&policy_filter) @@ -2128,7 +2156,7 @@ where transaction_time.map(LimitedTemporalBound::Inclusive), ), } - .resolve(); + .resolve_with(policy_components.timestamp()); Read::::read_one( self, @@ -2146,15 +2174,18 @@ where actor_id: ActorEntityUuid, mut params: PatchEntityParams, ) -> Result> { - let transaction_time = Timestamp::now().remove_nanosecond(); - let decision_time = params - .decision_time - .map_or_else(|| transaction_time.cast(), Timestamp::remove_nanosecond); - let transaction = self.begin_transaction().await.change_context(UpdateError)?; + // The transaction time is read from the database clock by the locking statement — the + // transaction's first statement — so it is consistent with the timestamps of + // concurrently committed data while every statement of this operation shares the single + // value. let locked_row = transaction - .lock_entity_edition(params.entity_id, transaction_time, decision_time) + .lock_entity_edition( + params.entity_id, + None, + params.decision_time.map(Timestamp::remove_nanosecond), + ) .await? .ok_or_else(|| { Report::new(EntityDoesNotExist) @@ -2162,6 +2193,10 @@ where .attach(params.entity_id) .change_context(UpdateError) })?; + let transaction_time = locked_row.locked_at; + let decision_time = params + .decision_time + .map_or_else(|| transaction_time.cast(), Timestamp::remove_nanosecond); let ClosedTemporalBound::Inclusive(locked_transaction_time) = *locked_row.transaction_time.start(); let ClosedTemporalBound::Inclusive(locked_decision_time) = @@ -2193,6 +2228,7 @@ where let policy_components = PolicyComponents::builder(&transaction) .with_actor(actor_id) + .with_timestamp(transaction_time.cast()) .with_entity_edition_id(previous_entity.metadata.record_id.edition_id) .with_entity_type_ids(¶ms.entity_type_ids) .with_actions( @@ -2568,7 +2604,11 @@ where } if let Some(previous_live_entity) = transaction - .lock_entity_edition(params.entity_id, transaction_time, decision_time) + .lock_entity_edition( + params.entity_id, + Some(transaction_time), + Some(decision_time), + ) .await? { transaction @@ -2833,7 +2873,7 @@ where // Delegates to the inherent method on `PostgresStore`, so the permission check is // also reachable where the `EntityStore` impl — bounded on `BeginReadOnlyTransaction` — // is unavailable. - self.has_permission_for_entities_impl(authenticated_actor, params) + self.has_permission_for_entities_impl(authenticated_actor, params, None) .await } @@ -2886,6 +2926,7 @@ where }, include_drafts: false, }, + None, ) .await .change_context(ClusterError::Store)?; @@ -3047,6 +3088,9 @@ struct LockedEntityEdition { entity_edition_id: EntityEditionId, decision_time: LeftClosedTemporalInterval, transaction_time: LeftClosedTemporalInterval, + /// The database clock reading of the locking statement, which callers use as the operation's + /// transaction time when they did not supply their own timestamps to the lock. + locked_at: Timestamp, } /// Builds the statement populating `entity_edition_cache` by aggregating the editions' @@ -3269,12 +3313,17 @@ where Ok(edition_id) } + /// Locks the entity's edition which is current at the given timestamps. + /// + /// A `transaction_time` or `decision_time` of `None` falls back to the database clock, whose + /// reading is returned as [`LockedEntityEdition::locked_at`] so callers can reuse it for the + /// remainder of the operation. #[tracing::instrument(level = "info", skip(self))] async fn lock_entity_edition( &self, entity_id: EntityId, - transaction_time: Timestamp, - decision_time: Timestamp, + transaction_time: Option>, + decision_time: Option>, ) -> Result, Report> { let current_data = if let Some(draft_id) = entity_id.draft_id { self.as_client() @@ -3283,13 +3332,16 @@ where SELECT entity_temporal_metadata.entity_edition_id, entity_temporal_metadata.transaction_time, - entity_temporal_metadata.decision_time + entity_temporal_metadata.decision_time, + statement_timestamp() FROM entity_temporal_metadata WHERE entity_temporal_metadata.web_id = $1 AND entity_temporal_metadata.entity_uuid = $2 AND entity_temporal_metadata.draft_id = $3 - AND entity_temporal_metadata.transaction_time @> $4::timestamptz - AND entity_temporal_metadata.decision_time @> $5::timestamptz + AND entity_temporal_metadata.transaction_time + @> COALESCE($4::timestamptz, statement_timestamp()) + AND entity_temporal_metadata.decision_time + @> COALESCE($5::timestamptz, statement_timestamp()) FOR NO KEY UPDATE NOWAIT;", &[ &entity_id.web_id, @@ -3313,13 +3365,16 @@ where SELECT entity_temporal_metadata.entity_edition_id, entity_temporal_metadata.transaction_time, - entity_temporal_metadata.decision_time + entity_temporal_metadata.decision_time, + statement_timestamp() FROM entity_temporal_metadata WHERE entity_temporal_metadata.web_id = $1 AND entity_temporal_metadata.entity_uuid = $2 AND entity_temporal_metadata.draft_id IS NULL - AND entity_temporal_metadata.transaction_time @> $3::timestamptz - AND entity_temporal_metadata.decision_time @> $4::timestamptz + AND entity_temporal_metadata.transaction_time + @> COALESCE($3::timestamptz, statement_timestamp()) + AND entity_temporal_metadata.decision_time + @> COALESCE($4::timestamptz, statement_timestamp()) FOR NO KEY UPDATE NOWAIT;", &[ &entity_id.web_id, @@ -3344,6 +3399,7 @@ where entity_edition_id: row.get(0), transaction_time: row.get(1), decision_time: row.get(2), + locked_at: row.get(3), }) }) .map_err(|error| match error.code() { diff --git a/libs/@local/graph/postgres-store/src/store/postgres/knowledge/entity/table.rs b/libs/@local/graph/postgres-store/src/store/postgres/knowledge/entity/table.rs index 8a90616917f..7911824eea8 100644 --- a/libs/@local/graph/postgres-store/src/store/postgres/knowledge/entity/table.rs +++ b/libs/@local/graph/postgres-store/src/store/postgres/knowledge/entity/table.rs @@ -24,7 +24,7 @@ use hash_graph_store::{ EntityTableSummary, EntityTableWebScope, QueryEntitiesTableParams, QueryEntitiesTableResponse, TYPE_UNIVERSE_LIMIT, }, - entity_type::{EntityTypeQueryPath, EntityTypeStore as _, IncludeEntityTypeOption}, + entity_type::{EntityTypeQueryPath, IncludeEntityTypeOption}, error::QueryError, filter::{ Filter, FilterExpression, FilterExpressionList, JsonPath, Parameter, ParameterList, @@ -33,10 +33,7 @@ use hash_graph_store::{ query::CursorField, subgraph::{ edges::{EdgeDirection, KnowledgeGraphEdgeKind, SharedEdgeKind}, - temporal_axes::{ - PinnedTemporalAxis, QueryTemporalAxes, QueryTemporalAxesUnresolved, - VariableTemporalAxis, - }, + temporal_axes::{PinnedTemporalAxis, QueryTemporalAxes, VariableTemporalAxis}, }, }; use hash_graph_temporal_versioning::{ @@ -366,14 +363,15 @@ where .change_context(QueryError)?; // The sort shapes the keyset, so a continuation reads it from its - // token instead of trusting the re-sent request. The snapshot instants - // are minted here for a first page — a shade after the transaction's - // own snapshot, so a writer committing in between stays out of this - // page's aggregates but can still surface on a later page. + // token instead of trusting the re-sent request. A first page takes its + // snapshot instants from the operation's database clock reading — a + // shade after the transaction's own snapshot, so a writer committing in + // between stays out of this page's aggregates but can still surface on + // a later page. let (transaction_time, decision_time, mut type_universe, sort, position) = match ¶ms.cursor { None => { - let now: Timestamp<()> = Timestamp::now(); + let now = policy_components.timestamp(); (now.cast(), now.cast(), None, params.sort, None) } Some(cursor) => ( @@ -642,7 +640,7 @@ where .map(|endpoint| endpoint.entity_type_ids.clone()) }; Some( - self.get_closed_multi_entity_types( + self.get_closed_multi_entity_types_impl( actor_id, rows.iter() .map(|row| row.entity_type_ids.clone()) @@ -654,7 +652,9 @@ where rows.iter() .filter_map(|row| endpoint_types(&row.target_entity)), ), - QueryTemporalAxesUnresolved::live_only(), + // The types are read at the page's instant rather than at a clock reading of + // their own, so the chips describe the rows the page actually returned. + &temporal_axes, None, ) .await? @@ -1029,7 +1029,9 @@ mod tests { use core::str::FromStr as _; use hash_codec::numeric::Real; - use hash_graph_store::entity::EntityTableFilter; + use hash_graph_store::{ + entity::EntityTableFilter, subgraph::temporal_axes::QueryTemporalAxesUnresolved, + }; use type_system::{ontology::BaseUrl, principal::actor_group::WebId}; use uuid::Uuid; @@ -1134,7 +1136,7 @@ mod tests { // out. #[test] fn table_exclusion_scopes_statement() { - let temporal_axes = QueryTemporalAxesUnresolved::all().resolve(); + let temporal_axes = QueryTemporalAxesUnresolved::all().resolve_with(Timestamp::now()); let mut params = params(None); params.filter.webs = EntityTableWebScope::Exclude { webs: vec![WebId::new(Uuid::nil())], @@ -1231,7 +1233,7 @@ mod tests { let filter = property_filters_filter(&property_filters).expect("the filters should build a filter"); - let temporal_axes = QueryTemporalAxesUnresolved::all().resolve(); + let temporal_axes = QueryTemporalAxesUnresolved::all().resolve_with(Timestamp::now()); let mut compiler = SelectCompiler::::new(Some(&temporal_axes), INCLUDE_DRAFTS); compiler .add_filter(&filter) @@ -1276,7 +1278,7 @@ mod tests { EntityTableSortKey::Archived, ] .map(|key| { - let temporal_axes = QueryTemporalAxesUnresolved::all().resolve(); + let temporal_axes = QueryTemporalAxesUnresolved::all().resolve_with(Timestamp::now()); let mut compiler = SelectCompiler::::new(Some(&temporal_axes), INCLUDE_DRAFTS); compiler.set_limit(10); compiler.set_statement_shape(StatementShape::KeysFirst); @@ -1319,7 +1321,7 @@ mod tests { #[test] fn table_page_statement() { - let temporal_axes = QueryTemporalAxesUnresolved::all().resolve(); + let temporal_axes = QueryTemporalAxesUnresolved::all().resolve_with(Timestamp::now()); let params = params(Some(vec![WebId::new(Uuid::nil())])); let scope = scope_filter(¶ms); diff --git a/libs/@local/graph/postgres-store/src/store/postgres/mod.rs b/libs/@local/graph/postgres-store/src/store/postgres/mod.rs index cb58ef2d4f7..d0a9758258d 100644 --- a/libs/@local/graph/postgres-store/src/store/postgres/mod.rs +++ b/libs/@local/graph/postgres-store/src/store/postgres/mod.rs @@ -29,9 +29,9 @@ use hash_graph_authorization::policies::{ error::{ BuildDataTypeContextError, BuildEntityContextError, BuildEntityTypeContextError, BuildPrincipalContextError, BuildPropertyTypeContextError, CreatePolicyError, - DetermineActorError, EnsureSystemPoliciesError, GetPoliciesError, - GetSystemAccountError, RemovePolicyError, RoleAssignmentError, TeamRoleError, - UpdatePolicyError, WebCreationError, WebRoleError, + CurrentTimestampError, DetermineActorError, EnsureSystemPoliciesError, + GetPoliciesError, GetSystemAccountError, RemovePolicyError, RoleAssignmentError, + TeamRoleError, UpdatePolicyError, WebCreationError, WebRoleError, }, }, }; @@ -47,7 +47,7 @@ use hash_graph_store::{ filter::protection::PropertyProtectionFilterConfig, query::ConflictBehavior, }; -use hash_graph_temporal_versioning::{LeftClosedTemporalInterval, TransactionTime}; +use hash_graph_temporal_versioning::{LeftClosedTemporalInterval, Timestamp, TransactionTime}; use hash_status::StatusCode; use hash_temporal_client::TemporalClient; use postgres_types::{Json, ToSql}; @@ -809,6 +809,23 @@ where } } +fn actor_id_from_principal_type( + principal_type: PrincipalType, + actor_entity_uuid: ActorEntityUuid, +) -> ActorId { + match principal_type { + PrincipalType::User => ActorId::User(UserId::new(actor_entity_uuid)), + PrincipalType::Machine => ActorId::Machine(MachineId::new(actor_entity_uuid)), + PrincipalType::Ai => ActorId::Ai(AiId::new(actor_entity_uuid)), + principal_type @ (PrincipalType::Web + | PrincipalType::Team + | PrincipalType::WebRole + | PrincipalType::TeamRole) => { + unreachable!("Unexpected actor type: {principal_type:?}") + } + } +} + impl PrincipalStore for PostgresStore where C: AsClient, @@ -1251,17 +1268,65 @@ where .change_context(DetermineActorError::StoreError)? .ok_or(DetermineActorError::ActorNotFound { actor_entity_uuid })?; - Ok(Some(match row.get(0) { - PrincipalType::User => ActorId::User(UserId::new(actor_entity_uuid)), - PrincipalType::Machine => ActorId::Machine(MachineId::new(actor_entity_uuid)), - PrincipalType::Ai => ActorId::Ai(AiId::new(actor_entity_uuid)), - principal_type @ (PrincipalType::Web - | PrincipalType::Team - | PrincipalType::WebRole - | PrincipalType::TeamRole) => { - unreachable!("Unexpected actor type: {principal_type:?}") - } - })) + Ok(Some(actor_id_from_principal_type( + row.get(0), + actor_entity_uuid, + ))) + } + + #[tracing::instrument(skip(self))] + async fn determine_actor_with_timestamp( + &self, + actor_entity_uuid: ActorEntityUuid, + ) -> Result<(Option, Timestamp<()>), Report> { + if actor_entity_uuid.is_public_actor() { + let timestamp = self + .current_timestamp() + .await + .change_context(DetermineActorError::StoreError)?; + return Ok((None, timestamp)); + } + + let row = self + .as_client() + .query_opt( + "SELECT principal_type, statement_timestamp() FROM actor WHERE id = $1", + &[&actor_entity_uuid], + ) + .instrument(tracing::info_span!( + "SELECT", + otel.kind = "client", + db.system = "postgresql", + peer.service = "Postgres", + )) + .await + .change_context(DetermineActorError::StoreError)? + .ok_or(DetermineActorError::ActorNotFound { actor_entity_uuid })?; + + Ok(( + Some(actor_id_from_principal_type(row.get(0), actor_entity_uuid)), + row.get(1), + )) + } + + #[tracing::instrument(skip(self))] + async fn current_timestamp(&self) -> Result, Report> { + // `statement_timestamp()` rather than `now()`: it advances per statement even inside a + // transaction, so operations sharing one transaction still observe distinct timestamps, + // while a transaction's first statement yields a reading aligned with the transaction's + // snapshot. + Ok(self + .as_client() + .query_one("SELECT statement_timestamp()", &[]) + .instrument(tracing::info_span!( + "SELECT", + otel.kind = "client", + db.system = "postgresql", + peer.service = "Postgres", + )) + .await + .change_context(CurrentTimestampError)? + .get(0)) } #[tracing::instrument(level = "info", skip(self, context_builder))] @@ -2681,18 +2746,19 @@ where &self, ontology_id: OntologyTypeUuid, provenance: &OntologyEditionProvenance, + transaction_time: Timestamp, ) -> Result, Report> { let query = " INSERT INTO ontology_temporal_metadata ( ontology_id, transaction_time, provenance - ) VALUES ($1, tstzrange(now(), NULL, '[)'), $2) + ) VALUES ($1, tstzrange($3, NULL, '[)'), $2) RETURNING transaction_time; "; self.as_client() - .query_one(query, &[&ontology_id, &provenance]) + .query_one(query, &[&ontology_id, &provenance, &transaction_time]) .instrument(tracing::info_span!( "INSERT", otel.kind = "client", @@ -2708,11 +2774,12 @@ where &self, id: &VersionedUrl, archived_by_id: ActorEntityUuid, + transaction_time: Timestamp, ) -> Result> { let query = " UPDATE ontology_temporal_metadata SET - transaction_time = tstzrange(lower(transaction_time), now(), '[)'), + transaction_time = tstzrange(lower(transaction_time), $4, '[)'), provenance = provenance || JSONB_BUILD_OBJECT( 'archivedById', $3::UUID ) @@ -2720,13 +2787,22 @@ where SELECT ontology_id FROM ontology_ids WHERE base_url = $1 AND version = $2 - ) AND transaction_time @> now() + ) AND upper(transaction_time) IS NULL + AND transaction_time @> $4::timestamptz RETURNING transaction_time; "; let optional = self .as_client() - .query_opt(query, &[&id.base_url, &id.version, &archived_by_id]) + .query_opt( + query, + &[ + &id.base_url, + &id.version, + &archived_by_id, + &transaction_time, + ], + ) .instrument(tracing::info_span!( "UPDATE", otel.kind = "client", @@ -2778,6 +2854,7 @@ where &self, id: &VersionedUrl, provenance: &OntologyEditionProvenance, + transaction_time: Timestamp, ) -> Result> { let query = " INSERT INTO ontology_temporal_metadata ( @@ -2786,7 +2863,7 @@ where provenance ) VALUES ( (SELECT ontology_id FROM ontology_ids WHERE base_url = $1 AND version = $2), - tstzrange(now(), NULL, '[)'), + tstzrange($4, NULL, '[)'), $3 ) RETURNING transaction_time; @@ -2795,7 +2872,10 @@ where Ok(OntologyTemporalMetadata { transaction_time: self .as_client() - .query_one(query, &[&id.base_url, &id.version, &provenance]) + .query_one( + query, + &[&id.base_url, &id.version, &provenance, &transaction_time], + ) .instrument(tracing::info_span!( "INSERT", otel.kind = "client", @@ -3391,6 +3471,7 @@ where ownership: &OntologyOwnership, on_conflict: ConflictBehavior, provenance: &OntologyProvenance, + transaction_time: Timestamp, ) -> Result, Report> { match ownership { OntologyOwnership::Local { web_id } => { @@ -3399,7 +3480,11 @@ where let ontology_id = self.create_ontology_id(ontology_id, on_conflict).await?; if let Some(ontology_id) = ontology_id { let transaction_time = self - .create_ontology_temporal_metadata(ontology_id, &provenance.edition) + .create_ontology_temporal_metadata( + ontology_id, + &provenance.edition, + transaction_time, + ) .await?; self.create_ontology_owned_metadata(ontology_id, *web_id) .await?; @@ -3421,7 +3506,11 @@ where let ontology_id = self.create_ontology_id(ontology_id, on_conflict).await?; if let Some(ontology_id) = ontology_id { let transaction_time = self - .create_ontology_temporal_metadata(ontology_id, &provenance.edition) + .create_ontology_temporal_metadata( + ontology_id, + &provenance.edition, + transaction_time, + ) .await?; self.create_ontology_external_metadata(ontology_id, *fetched_at) .await?; @@ -3449,6 +3538,7 @@ where &self, url: &VersionedUrl, provenance: &OntologyEditionProvenance, + transaction_time: Timestamp, ) -> Result<(OntologyTypeUuid, WebId, OntologyTemporalMetadata), Report> { let previous_version = OntologyTypeVersion { @@ -3520,7 +3610,7 @@ where .expect("ontology id should have been created"); let transaction_time = self - .create_ontology_temporal_metadata(ontology_id, provenance) + .create_ontology_temporal_metadata(ontology_id, provenance, transaction_time) .await .change_context(UpdateError)?; self.create_ontology_owned_metadata(ontology_id, web_id) diff --git a/libs/@local/graph/postgres-store/src/store/postgres/ontology/data_type.rs b/libs/@local/graph/postgres-store/src/store/postgres/ontology/data_type.rs index 79bd5797e53..958f70f500d 100644 --- a/libs/@local/graph/postgres-store/src/store/postgres/ontology/data_type.rs +++ b/libs/@local/graph/postgres-store/src/store/postgres/ontology/data_type.rs @@ -6,7 +6,7 @@ use error_stack::{Report, ResultExt as _}; use futures::{StreamExt as _, TryStreamExt as _}; use hash_graph_authorization::policies::{ Authorized, MergePolicies, PolicyComponents, Request, RequestContext, ResourceId, - action::ActionName, principal::actor::AuthenticatedActor, + action::ActionName, principal::actor::AuthenticatedActor, store::PrincipalStore as _, }; use hash_graph_migrations::Transaction as _; use hash_graph_store::{ @@ -31,7 +31,9 @@ use hash_graph_store::{ temporal_axes::{QueryTemporalAxes, QueryTemporalAxesUnresolved, VariableAxis}, }, }; -use hash_graph_temporal_versioning::RightBoundedTemporalInterval; +use hash_graph_temporal_versioning::{ + RightBoundedTemporalInterval, TemporalTagged as _, Timestamp, TransactionTime, +}; use hash_status::StatusCode; use postgres_types::{Json, ToSql}; use tokio_postgres::{GenericClient as _, Row}; @@ -510,12 +512,20 @@ where .await .change_context(InsertionError)?; + let transaction_time = Timestamp::::from_anonymous( + transaction + .current_timestamp() + .await + .change_context(InsertionError)?, + ); + let mut inserted_data_type_metadata = Vec::new(); let mut inserted_data_types = Vec::new(); let mut data_type_reference_ids = HashSet::new(); let mut data_type_conversions_rows = Vec::new(); - let mut policy_components_builder = PolicyComponents::builder(&transaction); + let mut policy_components_builder = + PolicyComponents::builder(&transaction).with_timestamp(transaction_time.cast()); for parameters in params { let provenance = OntologyProvenance { @@ -540,6 +550,7 @@ where ¶meters.ownership, parameters.conflict_behavior, &provenance, + transaction_time, ) .await? { @@ -763,7 +774,9 @@ where .await .change_context(QueryError)?; - let temporal_axes = params.temporal_axes.resolve(); + let temporal_axes = params + .temporal_axes + .resolve_with(policy_components.timestamp()); self.query_data_types_impl(params, &temporal_axes, &policy_components) .await } @@ -790,7 +803,11 @@ where Ok(self .read( &[params.filter], - Some(¶ms.temporal_axes.resolve()), + Some( + ¶ms + .temporal_axes + .resolve_with(policy_components.timestamp()), + ), false, ) .await? @@ -820,7 +837,9 @@ where .await .change_context(QueryError)?; - let temporal_axes = request.temporal_axes.resolve(); + let temporal_axes = request + .temporal_axes + .resolve_with(policy_components.timestamp()); let time_axis = temporal_axes.variable_time_axis(); let mut subgraph = Subgraph::new(request.temporal_axes, temporal_axes.clone()); @@ -936,6 +955,13 @@ where { let transaction = self.begin_transaction().await.change_context(UpdateError)?; + let transaction_time = Timestamp::::from_anonymous( + transaction + .current_timestamp() + .await + .change_context(UpdateError)?, + ); + let mut updated_data_type_metadata = Vec::new(); let mut inserted_data_types = Vec::new(); let mut data_type_reference_ids = HashSet::new(); @@ -973,7 +999,11 @@ where let data_type_id = DataTypeUuid::from_url(¶meters.schema.id); let (_ontology_id, web_id, temporal_versioning) = transaction - .update_owned_ontology_id(¶meters.schema.id, &provenance.edition) + .update_owned_ontology_id( + ¶meters.schema.id, + &provenance.edition, + transaction_time, + ) .await?; data_type_reference_ids.extend( @@ -1002,6 +1032,7 @@ where let policy_components = PolicyComponents::builder(&transaction) .with_actor(actor_id) + .with_timestamp(transaction_time.cast()) .with_data_type_ids(&old_data_type_ids) .with_actions([ActionName::UpdateDataType], MergePolicies::No) .await @@ -1220,8 +1251,12 @@ where } } - self.archive_ontology_type(¶ms.data_type_id, actor_id) - .await + self.archive_ontology_type( + ¶ms.data_type_id, + actor_id, + Timestamp::from_anonymous(policy_components.timestamp()), + ) + .await } #[tracing::instrument(level = "info", skip(self))] @@ -1269,6 +1304,7 @@ where archived_by_id: None, user_defined: params.provenance, }, + Timestamp::from_anonymous(policy_components.timestamp()), ) .await } @@ -1614,7 +1650,14 @@ where authenticated_actor: AuthenticatedActor, params: HasPermissionForDataTypesParams<'_>, ) -> Result, Report> { - let temporal_axes = QueryTemporalAxesUnresolved::live_only().resolve(); + let policy_components = PolicyComponents::builder(self) + .with_actor(authenticated_actor) + .with_action(params.action, MergePolicies::Yes) + .await + .change_context(CheckPermissionError::BuildPolicyContext)?; + + let temporal_axes = + QueryTemporalAxesUnresolved::live_only().resolve_with(policy_components.timestamp()); let mut compiler = SelectCompiler::new(Some(&temporal_axes), true); let data_type_uuids = params @@ -1627,12 +1670,6 @@ where compiler .add_filter(&data_type_filter) .change_context(CheckPermissionError::CompileFilter)?; - - let policy_components = PolicyComponents::builder(self) - .with_actor(authenticated_actor) - .with_action(params.action, MergePolicies::Yes) - .await - .change_context(CheckPermissionError::BuildPolicyContext)?; let policy_filter = Filter::::for_policies( policy_components.extract_filter_policies(params.action), policy_components.optimization_data(params.action), diff --git a/libs/@local/graph/postgres-store/src/store/postgres/ontology/entity_type.rs b/libs/@local/graph/postgres-store/src/store/postgres/ontology/entity_type.rs index 2d1d7c8a508..63ead160007 100644 --- a/libs/@local/graph/postgres-store/src/store/postgres/ontology/entity_type.rs +++ b/libs/@local/graph/postgres-store/src/store/postgres/ontology/entity_type.rs @@ -7,7 +7,7 @@ use futures::{StreamExt as _, TryStreamExt as _}; use hash_codec::numeric::Real; use hash_graph_authorization::policies::{ Authorized, MergePolicies, PolicyComponents, Request, RequestContext, ResourceId, - action::ActionName, principal::actor::AuthenticatedActor, + action::ActionName, principal::actor::AuthenticatedActor, store::PrincipalStore as _, }; use hash_graph_migrations::Transaction as _; use hash_graph_store::{ @@ -38,7 +38,9 @@ use hash_graph_store::{ temporal_axes::{QueryTemporalAxes, QueryTemporalAxesUnresolved, VariableAxis}, }, }; -use hash_graph_temporal_versioning::RightBoundedTemporalInterval; +use hash_graph_temporal_versioning::{ + RightBoundedTemporalInterval, TemporalTagged as _, Timestamp, TransactionTime, +}; use hash_graph_types::ontology::OntologyTypeProvider; use hash_status::StatusCode; use postgres_types::Json; @@ -586,11 +588,9 @@ where pub(crate) async fn query_closed_entity_types( &self, filter: &Filter<'_, EntityTypeWithMetadata>, - temporal_axes: QueryTemporalAxesUnresolved, + temporal_axes: &QueryTemporalAxes, ) -> Result, Report> { - let resolved_temporal_axes = temporal_axes.resolve(); - - let mut compiler = SelectCompiler::new(Some(&resolved_temporal_axes), false); + let mut compiler = SelectCompiler::new(Some(temporal_axes), false); compiler.add_filter(filter).change_context(QueryError)?; let closed_schema_idx = compiler.add_selection_path(&EntityTypeQueryPath::ClosedSchema(None)); @@ -620,6 +620,135 @@ where .change_context(QueryError) } + #[tracing::instrument( + level = "info", + skip(self, actor_id, entity_type_ids, temporal_axes, include_resolved) + )] + pub(crate) async fn get_closed_multi_entity_types_impl( + &self, + actor_id: ActorEntityUuid, + entity_type_ids: I, + temporal_axes: &QueryTemporalAxes, + include_resolved: Option, + ) -> Result> + where + I: IntoIterator + Send, + J: IntoIterator + Send, + { + let mut response = GetClosedMultiEntityTypesResponse { + entity_types: HashMap::new(), + definitions: None, + }; + + // Collect all unique entity type IDs that need resolution + let mut entity_type_ids_to_resolve = HashSet::new(); + let all_multi_entity_type_ids = entity_type_ids + .into_iter() + .map(|entity_type_ids| { + entity_type_ids + .into_iter() + .inspect(|id| { + entity_type_ids_to_resolve.insert(EntityTypeUuid::from_url(id)); + }) + .collect::>() + }) + .collect::>(); + + // Convert entity type IDs to database-specific UUID references + let entity_type_uuids = entity_type_ids_to_resolve.into_iter().collect::>(); + + // Fetch all closed entity types in a single database query for efficiency + let closed_types = self + .query_closed_entity_types( + &Filter::for_entity_type_uuids(&entity_type_uuids), + temporal_axes, + ) + .await? + .into_iter() + .map(|closed_entity_type| (closed_entity_type.id.clone(), closed_entity_type)) + .collect::>(); + + // Build the nested hierarchical structure for each set of entity types + for entity_multi_type_ids in all_multi_entity_type_ids { + // Get the first entity type to serve as the root of the hierarchy + let mut entity_type_id_iter = entity_multi_type_ids.into_iter(); + let Some(first_entity_type_id) = entity_type_id_iter.next() else { + continue; // Skip empty sets + }; + + // Create or retrieve the entry for the first entity type + let mut map_ref = response + .entity_types + .entry(first_entity_type_id.clone()) + .or_insert_with(|| ClosedMultiEntityTypeMap { + schema: ClosedMultiEntityType::from_closed_schema( + closed_types + .get(&first_entity_type_id) + .expect( + "The entity type was already resolved, so it should be present in \ + the closed types", + ) + .clone(), + ), + inner: HashMap::new(), + }); + + // Process remaining entity types in the set, creating a nested structure + for entity_type_id in entity_type_id_iter { + // For each additional entity type, create a deeper level in the hierarchy + let new_map = map_ref + .inner + .entry(entity_type_id.clone()) + .or_insert_with(|| { + let mut closed_parent = map_ref.schema.clone(); + closed_parent + .add_closed_entity_type( + closed_types + .get(&entity_type_id) + .expect( + "The entity type was already resolved, so it should be \ + present in the closed types", + ) + .clone(), + ) + .expect("The entity type was constructed before so it has to be valid"); + ClosedMultiEntityTypeMap { + schema: closed_parent, + inner: HashMap::new(), + } + }); + map_ref = new_map; + } + } + + if let Some(include_entity_types) = include_resolved { + match include_entity_types { + IncludeResolvedEntityTypeOption::Resolved => { + response.definitions = Some( + self.get_entity_type_resolve_definitions( + actor_id, + &entity_type_uuids, + false, + ) + .await?, + ); + } + IncludeResolvedEntityTypeOption::ResolvedWithDataTypeChildren => { + response.definitions = Some( + self.get_entity_type_resolve_definitions( + actor_id, + &entity_type_uuids, + true, + ) + .await?, + ); + } + } + } + + Ok(response) + } + /// Internal method to read a [`EntityTypeWithMetadata`] into four [`TraversalContext`]s. /// /// This is used to recursively resolve a type, so the result can be reused. @@ -899,11 +1028,19 @@ where .await .change_context(InsertionError)?; + let transaction_time = Timestamp::::from_anonymous( + transaction + .current_timestamp() + .await + .change_context(InsertionError)?, + ); + let mut inserted_entity_type_metadata = Vec::new(); let mut inserted_entity_types = Vec::new(); let mut entity_type_reference_ids = Vec::new(); - let mut policy_components_builder = PolicyComponents::builder(&transaction); + let mut policy_components_builder = + PolicyComponents::builder(&transaction).with_timestamp(transaction_time.cast()); for parameters in params { let provenance = OntologyProvenance { @@ -929,6 +1066,7 @@ where ¶meters.ownership, parameters.conflict_behavior, &provenance, + transaction_time, ) .await? { @@ -1126,7 +1264,9 @@ where policy_components.optimization_data(ActionName::ViewEntityType), ); - let temporal_axes = params.temporal_axes.resolve(); + let temporal_axes = params + .temporal_axes + .resolve_with(policy_components.timestamp()); let mut compiler = SelectCompiler::new(Some(&temporal_axes), false); compiler .add_filter(&policy_filter) @@ -1172,7 +1312,7 @@ where .change_context(QueryError)?; let temporal_axes = params.request.temporal_axes; - let resolved_temporal_axes = temporal_axes.resolve(); + let resolved_temporal_axes = temporal_axes.resolve_with(policy_components.timestamp()); let mut response = self .query_entity_types_impl(params.request, &resolved_temporal_axes, &policy_components) .await?; @@ -1185,8 +1325,11 @@ where .collect::>(); response.closed_entity_types = Some( - self.query_closed_entity_types(&Filter::for_entity_type_uuids(&ids), temporal_axes) - .await?, + self.query_closed_entity_types( + &Filter::for_entity_type_uuids(&ids), + &resolved_temporal_axes, + ) + .await?, ); match include_entity_types { @@ -1277,121 +1420,18 @@ where include_resolved: Option, ) -> Result> where - I: IntoIterator, - J: IntoIterator, + I: IntoIterator + Send, + J: IntoIterator + Send, { - let mut response = GetClosedMultiEntityTypesResponse { - entity_types: HashMap::new(), - definitions: None, - }; - - // Collect all unique entity type IDs that need resolution - let mut entity_type_ids_to_resolve = HashSet::new(); - let all_multi_entity_type_ids = entity_type_ids - .into_iter() - .map(|entity_type_ids| { - entity_type_ids - .into_iter() - .inspect(|id| { - entity_type_ids_to_resolve.insert(EntityTypeUuid::from_url(id)); - }) - .collect::>() - }) - .collect::>(); - - // Convert entity type IDs to database-specific UUID references - let entity_type_uuids = entity_type_ids_to_resolve.into_iter().collect::>(); - - // Fetch all closed entity types in a single database query for efficiency - let closed_types = self - .query_closed_entity_types( - &Filter::for_entity_type_uuids(&entity_type_uuids), - temporal_axes, - ) - .await? - .into_iter() - .map(|closed_entity_type| (closed_entity_type.id.clone(), closed_entity_type)) - .collect::>(); - - // Build the nested hierarchical structure for each set of entity types - for entity_multi_type_ids in all_multi_entity_type_ids { - // Get the first entity type to serve as the root of the hierarchy - let mut entity_type_id_iter = entity_multi_type_ids.into_iter(); - let Some(first_entity_type_id) = entity_type_id_iter.next() else { - continue; // Skip empty sets - }; - - // Create or retrieve the entry for the first entity type - let mut map_ref = response - .entity_types - .entry(first_entity_type_id.clone()) - .or_insert_with(|| ClosedMultiEntityTypeMap { - schema: ClosedMultiEntityType::from_closed_schema( - closed_types - .get(&first_entity_type_id) - .expect( - "The entity type was already resolved, so it should be present in \ - the closed types", - ) - .clone(), - ), - inner: HashMap::new(), - }); - - // Process remaining entity types in the set, creating a nested structure - for entity_type_id in entity_type_id_iter { - // For each additional entity type, create a deeper level in the hierarchy - let new_map = map_ref - .inner - .entry(entity_type_id.clone()) - .or_insert_with(|| { - let mut closed_parent = map_ref.schema.clone(); - closed_parent - .add_closed_entity_type( - closed_types - .get(&entity_type_id) - .expect( - "The entity type was already resolved, so it should be \ - present in the closed types", - ) - .clone(), - ) - .expect("The entity type was constructed before so it has to be valid"); - ClosedMultiEntityTypeMap { - schema: closed_parent, - inner: HashMap::new(), - } - }); - map_ref = new_map; - } - } - - if let Some(include_entity_types) = include_resolved { - match include_entity_types { - IncludeResolvedEntityTypeOption::Resolved => { - response.definitions = Some( - self.get_entity_type_resolve_definitions( - actor_id, - &entity_type_uuids, - false, - ) - .await?, - ); - } - IncludeResolvedEntityTypeOption::ResolvedWithDataTypeChildren => { - response.definitions = Some( - self.get_entity_type_resolve_definitions( - actor_id, - &entity_type_uuids, - true, - ) - .await?, - ); - } - } - } - - Ok(response) + let temporal_axes = + temporal_axes.resolve_with(self.current_timestamp().await.change_context(QueryError)?); + self.get_closed_multi_entity_types_impl( + actor_id, + entity_type_ids, + &temporal_axes, + include_resolved, + ) + .await } #[tracing::instrument(level = "info", skip(self))] @@ -1418,7 +1458,9 @@ where .await .change_context(QueryError)?; - let temporal_axes = request.temporal_axes.resolve(); + let temporal_axes = request + .temporal_axes + .resolve_with(policy_components.timestamp()); let time_axis = temporal_axes.variable_time_axis(); let mut subgraph = Subgraph::new(request.temporal_axes, temporal_axes.clone()); @@ -1541,6 +1583,13 @@ where { let transaction = self.begin_transaction().await.change_context(UpdateError)?; + let transaction_time = Timestamp::::from_anonymous( + transaction + .current_timestamp() + .await + .change_context(UpdateError)?, + ); + let mut updated_entity_type_metadata = Vec::new(); let mut inserted_entity_types = Vec::new(); let mut entity_type_reference_ids = Vec::new(); @@ -1578,7 +1627,11 @@ where let entity_type_id = EntityTypeUuid::from_url(¶meters.schema.id); let (_ontology_id, web_id, temporal_versioning) = transaction - .update_owned_ontology_id(¶meters.schema.id, &provenance.edition) + .update_owned_ontology_id( + ¶meters.schema.id, + &provenance.edition, + transaction_time, + ) .await?; entity_type_reference_ids.extend( @@ -1598,6 +1651,7 @@ where let policy_components = PolicyComponents::builder(&transaction) .with_actor(actor_id) + .with_timestamp(transaction_time.cast()) .with_entity_type_ids(&old_entity_type_ids) .with_actions([ActionName::UpdateEntityType], MergePolicies::No) .await @@ -1795,8 +1849,12 @@ where } } - self.archive_ontology_type(¶ms.entity_type_id, actor_id) - .await + self.archive_ontology_type( + ¶ms.entity_type_id, + actor_id, + Timestamp::from_anonymous(policy_components.timestamp()), + ) + .await } #[tracing::instrument(level = "info", skip(self))] @@ -1846,6 +1904,7 @@ where archived_by_id: None, user_defined: params.provenance, }, + Timestamp::from_anonymous(policy_components.timestamp()), ) .await } @@ -2111,7 +2170,14 @@ where }) .collect() } else { - let temporal_axes = QueryTemporalAxesUnresolved::live_only().resolve(); + let policy_components = PolicyComponents::builder(self) + .with_actor(authenticated_actor) + .with_action(params.action, MergePolicies::Yes) + .await + .change_context(CheckPermissionError::BuildPolicyContext)?; + + let temporal_axes = QueryTemporalAxesUnresolved::live_only() + .resolve_with(policy_components.timestamp()); let mut compiler = SelectCompiler::new(Some(&temporal_axes), true); let entity_type_uuids = params @@ -2124,12 +2190,6 @@ where compiler .add_filter(&entity_type_filter) .change_context(CheckPermissionError::CompileFilter)?; - - let policy_components = PolicyComponents::builder(self) - .with_actor(authenticated_actor) - .with_action(params.action, MergePolicies::Yes) - .await - .change_context(CheckPermissionError::BuildPolicyContext)?; let policy_filter = Filter::::for_policies( policy_components.extract_filter_policies(params.action), policy_components.optimization_data(params.action), diff --git a/libs/@local/graph/postgres-store/src/store/postgres/ontology/property_type.rs b/libs/@local/graph/postgres-store/src/store/postgres/ontology/property_type.rs index 4aeeffd5573..6b40a5856d0 100644 --- a/libs/@local/graph/postgres-store/src/store/postgres/ontology/property_type.rs +++ b/libs/@local/graph/postgres-store/src/store/postgres/ontology/property_type.rs @@ -5,7 +5,7 @@ use error_stack::{Report, ResultExt as _}; use futures::{StreamExt as _, TryStreamExt as _}; use hash_graph_authorization::policies::{ Authorized, MergePolicies, PolicyComponents, Request, RequestContext, ResourceId, - action::ActionName, principal::actor::AuthenticatedActor, + action::ActionName, principal::actor::AuthenticatedActor, store::PrincipalStore as _, }; use hash_graph_migrations::Transaction as _; use hash_graph_store::{ @@ -29,7 +29,9 @@ use hash_graph_store::{ temporal_axes::{QueryTemporalAxes, QueryTemporalAxesUnresolved, VariableAxis}, }, }; -use hash_graph_temporal_versioning::RightBoundedTemporalInterval; +use hash_graph_temporal_versioning::{ + RightBoundedTemporalInterval, TemporalTagged as _, Timestamp, TransactionTime, +}; use hash_status::StatusCode; use postgres_types::Json; use tokio_postgres::{GenericClient as _, Row}; @@ -508,11 +510,19 @@ where .await .change_context(InsertionError)?; + let transaction_time = Timestamp::::from_anonymous( + transaction + .current_timestamp() + .await + .change_context(InsertionError)?, + ); + let mut inserted_property_type_metadata = Vec::new(); let mut inserted_property_types = Vec::new(); let mut inserted_ontology_ids = Vec::new(); - let mut policy_components_builder = PolicyComponents::builder(&transaction); + let mut policy_components_builder = + PolicyComponents::builder(&transaction).with_timestamp(transaction_time.cast()); let property_type_validator = PropertyTypeValidator; @@ -539,6 +549,7 @@ where ¶meters.ownership, parameters.conflict_behavior, &provenance, + transaction_time, ) .await? { @@ -657,7 +668,9 @@ where policy_components.optimization_data(ActionName::ViewPropertyType), ); - let temporal_axes = params.temporal_axes.resolve(); + let temporal_axes = params + .temporal_axes + .resolve_with(policy_components.timestamp()); let mut compiler = SelectCompiler::new(Some(&temporal_axes), false); compiler .add_filter(&policy_filter) @@ -701,7 +714,9 @@ where .await .change_context(QueryError)?; - let temporal_axes = params.temporal_axes.resolve(); + let temporal_axes = params + .temporal_axes + .resolve_with(policy_components.timestamp()); self.query_property_types_impl(params, &temporal_axes, &policy_components) .await } @@ -730,7 +745,9 @@ where .await .change_context(QueryError)?; - let temporal_axes = request.temporal_axes.resolve(); + let temporal_axes = request + .temporal_axes + .resolve_with(policy_components.timestamp()); let time_axis = temporal_axes.variable_time_axis(); let mut subgraph = Subgraph::new(request.temporal_axes, temporal_axes.clone()); @@ -846,6 +863,13 @@ where { let transaction = self.begin_transaction().await.change_context(UpdateError)?; + let transaction_time = Timestamp::::from_anonymous( + transaction + .current_timestamp() + .await + .change_context(UpdateError)?, + ); + let mut updated_property_type_metadata = Vec::new(); let mut inserted_property_types = Vec::new(); let mut inserted_ontology_ids = Vec::new(); @@ -884,7 +908,11 @@ where let record_id = OntologyTypeRecordId::from(parameters.schema.id.clone()); let (ontology_id, web_id, temporal_versioning) = transaction - .update_owned_ontology_id(¶meters.schema.id, &provenance.edition) + .update_owned_ontology_id( + ¶meters.schema.id, + &provenance.edition, + transaction_time, + ) .await?; transaction @@ -913,6 +941,7 @@ where let policy_components = PolicyComponents::builder(&transaction) .with_actor(actor_id) + .with_timestamp(transaction_time.cast()) .with_property_type_ids(&old_property_type_ids) .with_actions([ActionName::UpdatePropertyType], MergePolicies::No) .await @@ -1018,8 +1047,12 @@ where } } - self.archive_ontology_type(¶ms.property_type_id, actor_id) - .await + self.archive_ontology_type( + ¶ms.property_type_id, + actor_id, + Timestamp::from_anonymous(policy_components.timestamp()), + ) + .await } #[tracing::instrument(level = "info", skip(self))] @@ -1069,6 +1102,7 @@ where archived_by_id: None, user_defined: params.provenance, }, + Timestamp::from_anonymous(policy_components.timestamp()), ) .await } @@ -1159,7 +1193,14 @@ where authenticated_actor: AuthenticatedActor, params: HasPermissionForPropertyTypesParams<'_>, ) -> Result, Report> { - let temporal_axes = QueryTemporalAxesUnresolved::live_only().resolve(); + let policy_components = PolicyComponents::builder(self) + .with_actor(authenticated_actor) + .with_action(params.action, MergePolicies::Yes) + .await + .change_context(CheckPermissionError::BuildPolicyContext)?; + + let temporal_axes = + QueryTemporalAxesUnresolved::live_only().resolve_with(policy_components.timestamp()); let mut compiler = SelectCompiler::new(Some(&temporal_axes), true); let property_type_uuids = params @@ -1172,12 +1213,6 @@ where compiler .add_filter(&property_type_filter) .change_context(CheckPermissionError::CompileFilter)?; - - let policy_components = PolicyComponents::builder(self) - .with_actor(authenticated_actor) - .with_action(params.action, MergePolicies::Yes) - .await - .change_context(CheckPermissionError::BuildPolicyContext)?; let policy_filter = Filter::::for_policies( policy_components.extract_filter_policies(params.action), policy_components.optimization_data(params.action), diff --git a/libs/@local/graph/postgres-store/src/store/postgres/query/compile/tests.rs b/libs/@local/graph/postgres-store/src/store/postgres/query/compile/tests.rs index f7f3cc53110..06236b1acba 100644 --- a/libs/@local/graph/postgres-store/src/store/postgres/query/compile/tests.rs +++ b/libs/@local/graph/postgres-store/src/store/postgres/query/compile/tests.rs @@ -17,6 +17,7 @@ use hash_graph_store::{ temporal_axes::QueryTemporalAxesUnresolved, }, }; +use hash_graph_temporal_versioning::Timestamp; use hash_graph_types::Embedding; use postgres_types::ToSql; use type_system::{ @@ -62,7 +63,7 @@ fn test_compilation<'p, 'q: 'p, T: PostgresRecord + 'static>( #[test] fn asterisk() { - let temporal_axes = QueryTemporalAxesUnresolved::all().resolve(); + let temporal_axes = QueryTemporalAxesUnresolved::all().resolve_with(Timestamp::now()); test_compilation( &SelectCompiler::::with_asterisk(Some(&temporal_axes), false), r#"SELECT * FROM "ontology_temporal_metadata" AS "ontology_temporal_metadata_0_0_0""#, @@ -72,7 +73,7 @@ fn asterisk() { #[test] fn simple_expression() { - let temporal_axes = QueryTemporalAxesUnresolved::all().resolve(); + let temporal_axes = QueryTemporalAxesUnresolved::all().resolve_with(Timestamp::now()); let pinned_timestamp = temporal_axes.pinned_timestamp(); let mut compiler = SelectCompiler::::with_asterisk(Some(&temporal_axes), false); @@ -108,7 +109,7 @@ fn simple_expression() { #[test] fn limited_temporal() { - let temporal_axes = QueryTemporalAxesUnresolved::all().resolve(); + let temporal_axes = QueryTemporalAxesUnresolved::all().resolve_with(Timestamp::now()); let mut compiler = SelectCompiler::::with_asterisk(Some(&temporal_axes), false); let filter = Filter::Equal( FilterExpression::Path { @@ -165,7 +166,7 @@ fn full_temporal() { #[test] fn specific_version() { - let temporal_axes = QueryTemporalAxesUnresolved::all().resolve(); + let temporal_axes = QueryTemporalAxesUnresolved::all().resolve_with(Timestamp::now()); let pinned_timestamp = temporal_axes.pinned_timestamp(); let mut compiler = SelectCompiler::::with_asterisk(Some(&temporal_axes), false); @@ -214,7 +215,7 @@ fn specific_version() { #[test] fn latest_version() { - let temporal_axes = QueryTemporalAxesUnresolved::all().resolve(); + let temporal_axes = QueryTemporalAxesUnresolved::all().resolve_with(Timestamp::now()); let pinned_timestamp = temporal_axes.pinned_timestamp(); let mut compiler = SelectCompiler::::with_asterisk(Some(&temporal_axes), false); @@ -249,7 +250,7 @@ fn latest_version() { #[test] fn not_latest_version() { - let temporal_axes = QueryTemporalAxesUnresolved::all().resolve(); + let temporal_axes = QueryTemporalAxesUnresolved::all().resolve_with(Timestamp::now()); let pinned_timestamp = temporal_axes.pinned_timestamp(); let mut compiler = SelectCompiler::::with_asterisk(Some(&temporal_axes), false); @@ -284,7 +285,7 @@ fn not_latest_version() { #[test] fn property_type_by_referenced_data_types() { - let temporal_axes = QueryTemporalAxesUnresolved::all().resolve(); + let temporal_axes = QueryTemporalAxesUnresolved::all().resolve_with(Timestamp::now()); let pinned_timestamp = temporal_axes.pinned_timestamp(); let mut compiler = SelectCompiler::::with_asterisk(Some(&temporal_axes), false); @@ -386,7 +387,7 @@ fn property_type_by_referenced_data_types() { #[test] fn property_type_by_referenced_property_types() { - let temporal_axes = QueryTemporalAxesUnresolved::all().resolve(); + let temporal_axes = QueryTemporalAxesUnresolved::all().resolve_with(Timestamp::now()); let pinned_timestamp = temporal_axes.pinned_timestamp(); let mut compiler = SelectCompiler::::with_asterisk(Some(&temporal_axes), false); @@ -427,7 +428,7 @@ fn property_type_by_referenced_property_types() { #[test] fn entity_type_by_referenced_property_types() { - let temporal_axes = QueryTemporalAxesUnresolved::all().resolve(); + let temporal_axes = QueryTemporalAxesUnresolved::all().resolve_with(Timestamp::now()); let pinned_timestamp = temporal_axes.pinned_timestamp(); let mut compiler = SelectCompiler::::with_asterisk(Some(&temporal_axes), false); @@ -469,7 +470,7 @@ fn entity_type_by_referenced_property_types() { #[test] fn entity_type_by_referenced_link_types() { - let temporal_axes = QueryTemporalAxesUnresolved::all().resolve(); + let temporal_axes = QueryTemporalAxesUnresolved::all().resolve_with(Timestamp::now()); let pinned_timestamp = temporal_axes.pinned_timestamp(); let mut compiler = SelectCompiler::::with_asterisk(Some(&temporal_axes), false); @@ -523,7 +524,7 @@ fn entity_type_by_referenced_link_types() { #[test] fn entity_type_by_inheritance() { - let temporal_axes = QueryTemporalAxesUnresolved::all().resolve(); + let temporal_axes = QueryTemporalAxesUnresolved::all().resolve_with(Timestamp::now()); let pinned_timestamp = temporal_axes.pinned_timestamp(); let mut compiler = SelectCompiler::::with_asterisk(Some(&temporal_axes), false); @@ -571,7 +572,7 @@ fn entity_type_by_inheritance() { #[test] fn entity_simple_query() { - let temporal_axes = QueryTemporalAxesUnresolved::all().resolve(); + let temporal_axes = QueryTemporalAxesUnresolved::all().resolve_with(Timestamp::now()); let pinned_timestamp = temporal_axes.pinned_timestamp(); let mut compiler = SelectCompiler::::with_asterisk(Some(&temporal_axes), false); @@ -606,7 +607,7 @@ fn entity_simple_query() { #[test] fn filter_entity_by_created_by_id() { - let temporal_axes = QueryTemporalAxesUnresolved::all().resolve(); + let temporal_axes = QueryTemporalAxesUnresolved::all().resolve_with(Timestamp::now()); let pinned_timestamp = temporal_axes.pinned_timestamp(); let mut compiler = SelectCompiler::::with_asterisk(Some(&temporal_axes), false); @@ -645,7 +646,7 @@ fn filter_entity_by_created_by_id() { #[test] fn sort_entity_by_created_at_transaction_time() { - let temporal_axes = QueryTemporalAxesUnresolved::all().resolve(); + let temporal_axes = QueryTemporalAxesUnresolved::all().resolve_with(Timestamp::now()); let pinned_timestamp = temporal_axes.pinned_timestamp(); let mut compiler = SelectCompiler::::new(Some(&temporal_axes), true); compiler.add_distinct_selection_with_ordering( @@ -672,7 +673,7 @@ fn sort_entity_by_created_at_transaction_time() { #[test] fn entity_with_manual_selection() { - let temporal_axes = QueryTemporalAxesUnresolved::all().resolve(); + let temporal_axes = QueryTemporalAxesUnresolved::all().resolve_with(Timestamp::now()); let pinned_timestamp = temporal_axes.pinned_timestamp(); let mut compiler = SelectCompiler::::new(Some(&temporal_axes), true); compiler.add_distinct_selection_with_ordering( @@ -720,7 +721,7 @@ fn entity_with_manual_selection() { #[test] fn entity_property_query() { - let temporal_axes = QueryTemporalAxesUnresolved::all().resolve(); + let temporal_axes = QueryTemporalAxesUnresolved::all().resolve_with(Timestamp::now()); let pinned_timestamp = temporal_axes.pinned_timestamp(); let mut compiler = SelectCompiler::::with_asterisk(Some(&temporal_axes), false); let json_path = JsonPath::from_path_tokens(vec![PathToken::Field(Cow::Borrowed( @@ -761,7 +762,7 @@ fn entity_property_query() { #[test] fn entity_property_null_query() { - let temporal_axes = QueryTemporalAxesUnresolved::all().resolve(); + let temporal_axes = QueryTemporalAxesUnresolved::all().resolve_with(Timestamp::now()); let pinned_timestamp = temporal_axes.pinned_timestamp(); let mut compiler = SelectCompiler::::with_asterisk(Some(&temporal_axes), false); let json_path = JsonPath::from_path_tokens(vec![PathToken::Field(Cow::Borrowed( @@ -795,7 +796,7 @@ fn entity_property_null_query() { #[test] fn entity_outgoing_link_query() { - let temporal_axes = QueryTemporalAxesUnresolved::all().resolve(); + let temporal_axes = QueryTemporalAxesUnresolved::all().resolve_with(Timestamp::now()); let pinned_timestamp = temporal_axes.pinned_timestamp(); let mut compiler = SelectCompiler::::with_asterisk(Some(&temporal_axes), false); @@ -856,7 +857,7 @@ fn entity_outgoing_link_query() { #[test] fn has_to_many_join_flag() { - let temporal_axes = QueryTemporalAxesUnresolved::all().resolve(); + let temporal_axes = QueryTemporalAxesUnresolved::all().resolve_with(Timestamp::now()); // A type-URL filter resolves to the edition cache (to-one join) — no fan-out. let mut to_one = SelectCompiler::::with_asterisk(Some(&temporal_axes), false); @@ -907,7 +908,7 @@ fn has_to_many_join_flag() { #[test] fn entity_incoming_link_query() { - let temporal_axes = QueryTemporalAxesUnresolved::all().resolve(); + let temporal_axes = QueryTemporalAxesUnresolved::all().resolve_with(Timestamp::now()); let pinned_timestamp = temporal_axes.pinned_timestamp(); let mut compiler = SelectCompiler::::with_asterisk(Some(&temporal_axes), false); @@ -968,7 +969,7 @@ fn entity_incoming_link_query() { #[test] fn link_entity_left_right_id() { - let temporal_axes = QueryTemporalAxesUnresolved::all().resolve(); + let temporal_axes = QueryTemporalAxesUnresolved::all().resolve_with(Timestamp::now()); let pinned_timestamp = temporal_axes.pinned_timestamp(); let mut compiler = SelectCompiler::::with_asterisk(Some(&temporal_axes), false); @@ -1058,7 +1059,7 @@ fn link_entity_left_right_id() { #[test] #[expect(clippy::similar_names)] fn two_linked_entities() { - let temporal_axes = QueryTemporalAxesUnresolved::all().resolve(); + let temporal_axes = QueryTemporalAxesUnresolved::all().resolve_with(Timestamp::now()); let pinned_timestamp = temporal_axes.pinned_timestamp(); let entity_a_uuid = Uuid::new_v4(); @@ -1157,7 +1158,7 @@ fn two_linked_entities() { #[test] fn filter_left_and_right() { - let temporal_axes = QueryTemporalAxesUnresolved::all().resolve(); + let temporal_axes = QueryTemporalAxesUnresolved::all().resolve_with(Timestamp::now()); let pinned_timestamp = temporal_axes.pinned_timestamp(); let mut compiler = SelectCompiler::::with_asterisk(Some(&temporal_axes), false); @@ -1258,7 +1259,7 @@ fn filter_left_and_right() { #[test] fn filter_entity_by_type_versioned_url() { - let temporal_axes = QueryTemporalAxesUnresolved::all().resolve(); + let temporal_axes = QueryTemporalAxesUnresolved::all().resolve_with(Timestamp::now()); let pinned_timestamp = temporal_axes.pinned_timestamp(); let mut compiler = SelectCompiler::::with_asterisk(Some(&temporal_axes), false); @@ -1290,7 +1291,7 @@ fn filter_entity_by_type_versioned_url() { #[test] fn filter_entity_by_any_type_versioned_url() { - let temporal_axes = QueryTemporalAxesUnresolved::all().resolve(); + let temporal_axes = QueryTemporalAxesUnresolved::all().resolve_with(Timestamp::now()); let pinned_timestamp = temporal_axes.pinned_timestamp(); let mut compiler = SelectCompiler::::with_asterisk(Some(&temporal_axes), false); @@ -1329,7 +1330,7 @@ fn filter_entity_by_any_type_versioned_url() { #[test] fn filter_entity_by_all_type_versioned_url() { - let temporal_axes = QueryTemporalAxesUnresolved::all().resolve(); + let temporal_axes = QueryTemporalAxesUnresolved::all().resolve_with(Timestamp::now()); let pinned_timestamp = temporal_axes.pinned_timestamp(); let mut compiler = SelectCompiler::::with_asterisk(Some(&temporal_axes), false); @@ -1368,7 +1369,7 @@ fn filter_entity_by_all_type_versioned_url() { #[test] fn filter_entity_own_and_linked_type_stay_separate() { - let temporal_axes = QueryTemporalAxesUnresolved::all().resolve(); + let temporal_axes = QueryTemporalAxesUnresolved::all().resolve_with(Timestamp::now()); let pinned_timestamp = temporal_axes.pinned_timestamp(); let mut compiler = SelectCompiler::::with_asterisk(Some(&temporal_axes), false); @@ -1447,7 +1448,7 @@ fn filter_entity_own_and_linked_type_stay_separate() { #[test] fn filter_entity_by_no_type_versioned_url() { - let temporal_axes = QueryTemporalAxesUnresolved::all().resolve(); + let temporal_axes = QueryTemporalAxesUnresolved::all().resolve_with(Timestamp::now()); let pinned_timestamp = temporal_axes.pinned_timestamp(); let mut compiler = SelectCompiler::::with_asterisk(Some(&temporal_axes), false); @@ -1495,7 +1496,7 @@ fn filter_entity_by_no_type_versioned_url() { #[test] fn filter_entity_by_type_starts_with_rejected() { - let temporal_axes = QueryTemporalAxesUnresolved::all().resolve(); + let temporal_axes = QueryTemporalAxesUnresolved::all().resolve_with(Timestamp::now()); let mut compiler = SelectCompiler::::with_asterisk(Some(&temporal_axes), false); // String operations have no scalar to operate on once the path resolves to the @@ -1531,7 +1532,7 @@ fn filter_entity_by_type_starts_with_rejected() { #[test] fn filter_entity_by_type_versioned_url_not_equal() { - let temporal_axes = QueryTemporalAxesUnresolved::all().resolve(); + let temporal_axes = QueryTemporalAxesUnresolved::all().resolve_with(Timestamp::now()); let pinned_timestamp = temporal_axes.pinned_timestamp(); let mut compiler = SelectCompiler::::with_asterisk(Some(&temporal_axes), false); @@ -1574,7 +1575,7 @@ fn filter_entity_by_type_versioned_url_not_equal() { #[test] fn filter_entity_by_type_base_url() { - let temporal_axes = QueryTemporalAxesUnresolved::all().resolve(); + let temporal_axes = QueryTemporalAxesUnresolved::all().resolve_with(Timestamp::now()); let pinned_timestamp = temporal_axes.pinned_timestamp(); let mut compiler = SelectCompiler::::with_asterisk(Some(&temporal_axes), false); @@ -1646,7 +1647,7 @@ fn filter_embedding_distance() { #[test] fn sort_by_label_and_type_title() { - let temporal_axes = QueryTemporalAxesUnresolved::all().resolve(); + let temporal_axes = QueryTemporalAxesUnresolved::all().resolve_with(Timestamp::now()); let pinned_timestamp = temporal_axes.pinned_timestamp(); let mut compiler = SelectCompiler::::new(Some(&temporal_axes), true); compiler.add_distinct_selection_with_ordering( @@ -1711,7 +1712,7 @@ mod predefined { }, }; - let temporal_axes = QueryTemporalAxesUnresolved::all().resolve(); + let temporal_axes = QueryTemporalAxesUnresolved::all().resolve_with(Timestamp::now()); let pinned_timestamp = temporal_axes.pinned_timestamp(); let mut compiler = SelectCompiler::::with_asterisk(Some(&temporal_axes), false); @@ -1741,7 +1742,7 @@ mod predefined { draft_id: None, }; - let temporal_axes = QueryTemporalAxesUnresolved::all().resolve(); + let temporal_axes = QueryTemporalAxesUnresolved::all().resolve_with(Timestamp::now()); let pinned_timestamp = temporal_axes.pinned_timestamp(); let mut compiler = SelectCompiler::::with_asterisk(Some(&temporal_axes), false); @@ -1993,7 +1994,7 @@ fn cursor_after_embedding_filter_is_rejected() { #[test] fn entity_cursor_pagination() { - let temporal_axes = QueryTemporalAxesUnresolved::all().resolve(); + let temporal_axes = QueryTemporalAxesUnresolved::all().resolve_with(Timestamp::now()); let pinned_timestamp = temporal_axes.pinned_timestamp(); let mut compiler = SelectCompiler::::with_asterisk(Some(&temporal_axes), false); @@ -2028,7 +2029,7 @@ fn entity_cursor_pagination() { #[test] fn fetch_keys_then_hydrate_entity_query() { - let temporal_axes = QueryTemporalAxesUnresolved::all().resolve(); + let temporal_axes = QueryTemporalAxesUnresolved::all().resolve_with(Timestamp::now()); let pinned_timestamp = temporal_axes.pinned_timestamp(); let mut compiler = SelectCompiler::::new(Some(&temporal_axes), true); compiler.add_distinct_selection_with_ordering( @@ -2085,7 +2086,7 @@ fn fetch_keys_then_hydrate_entity_query() { #[test] fn filter_joined_tables_stay_out_of_the_key_projection() { - let temporal_axes = QueryTemporalAxesUnresolved::all().resolve(); + let temporal_axes = QueryTemporalAxesUnresolved::all().resolve_with(Timestamp::now()); let pinned_timestamp = temporal_axes.pinned_timestamp(); let mut compiler = SelectCompiler::::new(Some(&temporal_axes), true); @@ -2142,7 +2143,7 @@ fn filter_joined_tables_stay_out_of_the_key_projection() { #[test] fn keys_first_omits_the_fence() { - let temporal_axes = QueryTemporalAxesUnresolved::all().resolve(); + let temporal_axes = QueryTemporalAxesUnresolved::all().resolve_with(Timestamp::now()); let pinned_timestamp = temporal_axes.pinned_timestamp(); let mut compiler = SelectCompiler::::new(Some(&temporal_axes), true); compiler.add_distinct_selection_with_ordering( @@ -2179,7 +2180,7 @@ fn keys_first_omits_the_fence() { #[test] fn fetch_keys_then_hydrate_requires_sort_and_limit() { - let temporal_axes = QueryTemporalAxesUnresolved::all().resolve(); + let temporal_axes = QueryTemporalAxesUnresolved::all().resolve_with(Timestamp::now()); // Sorted but unlimited: nothing to fence off, the layout stays single-pass. let mut compiler = SelectCompiler::::new(Some(&temporal_axes), true); @@ -2215,7 +2216,7 @@ fn fetch_keys_then_hydrate_requires_sort_and_limit() { #[test] fn fetch_keys_then_hydrate_asterisk_falls_back() { - let temporal_axes = QueryTemporalAxesUnresolved::all().resolve(); + let temporal_axes = QueryTemporalAxesUnresolved::all().resolve_with(Timestamp::now()); let mut compiler = SelectCompiler::::with_asterisk(Some(&temporal_axes), true); compiler.add_distinct_selection_with_ordering( &EntityQueryPath::Uuid, @@ -2236,7 +2237,7 @@ fn fetch_keys_then_hydrate_asterisk_falls_back() { #[test] fn fetch_keys_then_hydrate_to_many_filter_falls_back() { - let temporal_axes = QueryTemporalAxesUnresolved::all().resolve(); + let temporal_axes = QueryTemporalAxesUnresolved::all().resolve_with(Timestamp::now()); let mut compiler = SelectCompiler::::new(Some(&temporal_axes), true); compiler.add_distinct_selection_with_ordering( &EntityQueryPath::Uuid, @@ -2274,7 +2275,7 @@ fn fetch_keys_then_hydrate_to_many_filter_falls_back() { #[test] fn fetch_keys_then_hydrate_sorts_across_joins() { - let temporal_axes = QueryTemporalAxesUnresolved::all().resolve(); + let temporal_axes = QueryTemporalAxesUnresolved::all().resolve_with(Timestamp::now()); let pinned_timestamp = temporal_axes.pinned_timestamp(); let mut compiler = SelectCompiler::::new(Some(&temporal_axes), true); compiler.add_distinct_selection_with_ordering( @@ -2318,7 +2319,7 @@ fn fetch_keys_then_hydrate_sorts_across_joins() { #[test] fn fetch_keys_then_hydrate_with_cursor() { - let temporal_axes = QueryTemporalAxesUnresolved::all().resolve(); + let temporal_axes = QueryTemporalAxesUnresolved::all().resolve_with(Timestamp::now()); let pinned_timestamp = temporal_axes.pinned_timestamp(); let mut compiler = SelectCompiler::::new(Some(&temporal_axes), true); @@ -2402,7 +2403,7 @@ fn fetch_keys_then_hydrate_embedding_distance() { #[test] fn fetch_keys_then_hydrate_carries_existing_ctes() { - let temporal_axes = QueryTemporalAxesUnresolved::all().resolve(); + let temporal_axes = QueryTemporalAxesUnresolved::all().resolve_with(Timestamp::now()); let pinned_timestamp = temporal_axes.pinned_timestamp(); let mut compiler = SelectCompiler::::new(Some(&temporal_axes), false); compiler.add_distinct_selection_with_ordering( @@ -2531,7 +2532,7 @@ fn fetch_keys_then_hydrate_generating_key_join_falls_back() { #[test] fn fetch_keys_then_hydrate_reused_to_many_join_falls_back() { - let temporal_axes = QueryTemporalAxesUnresolved::all().resolve(); + let temporal_axes = QueryTemporalAxesUnresolved::all().resolve_with(Timestamp::now()); let mut compiler = SelectCompiler::::new(Some(&temporal_axes), true); compiler.add_distinct_selection_with_ordering( &EntityQueryPath::Uuid, diff --git a/libs/@local/graph/postgres-store/src/store/validation.rs b/libs/@local/graph/postgres-store/src/store/validation.rs index ac889db9ee6..b2cd86f0110 100644 --- a/libs/@local/graph/postgres-store/src/store/validation.rs +++ b/libs/@local/graph/postgres-store/src/store/validation.rs @@ -9,6 +9,7 @@ use hash_graph_store::{ error::QueryError, filter::Filter, query::Read as _, subgraph::temporal_axes::QueryTemporalAxesUnresolved, }; +use hash_graph_temporal_versioning::Timestamp; use hash_graph_types::ontology::{DataTypeLookup, OntologyTypeProvider}; use hash_graph_validation::EntityProvider; use hash_status::StatusCode; @@ -127,6 +128,9 @@ pub struct StoreProvider<'a, S> { pub store: &'a S, pub cache: Box, pub policy_components: Option<&'a PolicyComponents>, + /// The store's clock reading for the surrounding operation, used to resolve the temporal + /// axes of the lookups performed by this provider. + pub timestamp: Timestamp<()>, } impl<'a, S> StoreProvider<'a, S> { @@ -135,6 +139,7 @@ impl<'a, S> StoreProvider<'a, S> { store, cache: Box::new(StoreCache::default()), policy_components: Some(policy_components), + timestamp: policy_components.timestamp(), } } } @@ -158,7 +163,7 @@ where .store .read_one( &filters, - Some(&QueryTemporalAxesUnresolved::live_only().resolve()), + Some(&QueryTemporalAxesUnresolved::live_only().resolve_with(self.timestamp)), false, ) .await; @@ -208,7 +213,7 @@ where .store .read_one( &filters, - Some(&QueryTemporalAxesUnresolved::live_only().resolve()), + Some(&QueryTemporalAxesUnresolved::live_only().resolve_with(self.timestamp)), false, ) .await?; @@ -392,7 +397,7 @@ where .store .read( &filters, - Some(&QueryTemporalAxesUnresolved::live_only().resolve()), + Some(&QueryTemporalAxesUnresolved::live_only().resolve_with(self.timestamp)), false, ) .await @@ -481,7 +486,7 @@ where .store .read_closed_schemas( &filters, - Some(&QueryTemporalAxesUnresolved::live_only().resolve()), + Some(&QueryTemporalAxesUnresolved::live_only().resolve_with(self.timestamp)), ) .await .change_context(QueryError)? @@ -615,7 +620,7 @@ where .store .read_one( &filters, - Some(&QueryTemporalAxesUnresolved::live_only().resolve()), + Some(&QueryTemporalAxesUnresolved::live_only().resolve_with(self.timestamp)), entity_id.draft_id.is_some(), ) .await?; diff --git a/libs/@local/graph/store/src/subgraph/temporal_axes.rs b/libs/@local/graph/store/src/subgraph/temporal_axes.rs index 41799f60ea7..84b282b58d7 100644 --- a/libs/@local/graph/store/src/subgraph/temporal_axes.rs +++ b/libs/@local/graph/store/src/subgraph/temporal_axes.rs @@ -293,16 +293,6 @@ impl QueryTemporalAxesUnresolved { }, } } - - /// Resolves temporal axes using the current timestamp. - /// - /// Convenience method that resolves temporal axes against the current time. - /// Equivalent to calling `resolve_relative_to(Timestamp::now())`. - #[must_use] - pub fn resolve(self) -> QueryTemporalAxes { - let now = Timestamp::now(); - self.resolve_with(now) - } } /// A representation of a "pinned" temporal axis, used to project another temporal axis along the diff --git a/libs/@local/graph/type-fetcher/src/store.rs b/libs/@local/graph/type-fetcher/src/store.rs index 7be7aa00f9a..0751a37aede 100644 --- a/libs/@local/graph/type-fetcher/src/store.rs +++ b/libs/@local/graph/type-fetcher/src/store.rs @@ -14,9 +14,9 @@ use hash_graph_authorization::policies::{ error::{ BuildDataTypeContextError, BuildEntityContextError, BuildEntityTypeContextError, BuildPrincipalContextError, BuildPropertyTypeContextError, CreatePolicyError, - DetermineActorError, EnsureSystemPoliciesError, GetPoliciesError, - GetSystemAccountError, RemovePolicyError, RoleAssignmentError, TeamRoleError, - UpdatePolicyError, WebCreationError, WebRoleError, + CurrentTimestampError, DetermineActorError, EnsureSystemPoliciesError, + GetPoliciesError, GetSystemAccountError, RemovePolicyError, RoleAssignmentError, + TeamRoleError, UpdatePolicyError, WebCreationError, WebRoleError, }, }, }; @@ -254,6 +254,19 @@ where self.store.determine_actor(actor_entity_uuid).await } + async fn determine_actor_with_timestamp( + &self, + actor_entity_uuid: ActorEntityUuid, + ) -> Result<(Option, Timestamp<()>), Report> { + self.store + .determine_actor_with_timestamp(actor_entity_uuid) + .await + } + + async fn current_timestamp(&self) -> Result, Report> { + self.store.current_timestamp().await + } + async fn build_principal_context( &self, actor_id: ActorId, diff --git a/tests/graph/integration/postgres/data_type.rs b/tests/graph/integration/postgres/data_type.rs index f7b57301681..1b253fe730d 100644 --- a/tests/graph/integration/postgres/data_type.rs +++ b/tests/graph/integration/postgres/data_type.rs @@ -1,20 +1,27 @@ +use alloc::borrow::Cow; use core::str::FromStr as _; use std::collections::{HashMap, HashSet}; use hash_codec::numeric::Real; -use hash_graph_postgres_store::store::error::{ - BaseUrlAlreadyExists, OntologyTypeIsNotOwned, OntologyVersionDoesNotExist, - VersionedUrlAlreadyExists, +use hash_graph_postgres_store::store::{ + AsClient as _, + error::{ + BaseUrlAlreadyExists, OntologyTypeIsNotOwned, OntologyVersionDoesNotExist, + VersionedUrlAlreadyExists, + }, }; use hash_graph_store::{ data_type::{ - CreateDataTypeParams, DataTypeStore as _, QueryDataTypesParams, UpdateDataTypesParams, + ArchiveDataTypeParams, CreateDataTypeParams, DataTypeStore as _, + QueryDataTypeSubgraphParams, QueryDataTypesParams, UnarchiveDataTypeParams, + UpdateDataTypesParams, }, entity::{CreateEntityParams, EntityStore as _}, filter::Filter, query::ConflictBehavior, - subgraph::temporal_axes::QueryTemporalAxesUnresolved, + subgraph::temporal_axes::{QueryTemporalAxes, QueryTemporalAxesUnresolved}, }; +use hash_graph_temporal_versioning::{TemporalTagged as _, Timestamp}; use time::OffsetDateTime; use type_system::{ knowledge::{ @@ -35,7 +42,7 @@ use type_system::{ provenance::{OriginProvenance, OriginType}, }; -use crate::DatabaseTestWrapper; +use crate::{DatabaseApi, DatabaseTestWrapper}; #[tokio::test] async fn insert() { @@ -838,3 +845,149 @@ async fn update_external_with_owned() { "wrong error, expected `OntologyTypeIsNotOwned`, got {report:?}" ); } + +async fn is_queryable(api: &DatabaseApi<'_>, data_type_id: &VersionedUrl) -> bool { + !api.query_data_types( + api.account_id, + QueryDataTypesParams { + filter: Filter::for_versioned_url(data_type_id), + temporal_axes: QueryTemporalAxesUnresolved::live_only(), + after: None, + limit: None, + include_count: false, + }, + ) + .await + .expect("could not query data types") + .data_types + .is_empty() +} + +#[tokio::test] +async fn archive_unarchive_round_trip() { + let list_v1: DataType = serde_json::from_str(hash_graph_test_data::data_type::LIST_V1) + .expect("could not parse data type representation"); + + let mut database = DatabaseTestWrapper::new().await; + let mut api = database + .seed([hash_graph_test_data::data_type::VALUE_V1], [], []) + .await + .expect("could not seed database"); + + api.create_data_type( + api.account_id, + CreateDataTypeParams { + schema: list_v1.clone(), + ownership: OntologyOwnership::Local { + web_id: WebId::new(api.account_id), + }, + conflict_behavior: ConflictBehavior::Fail, + provenance: ProvidedOntologyEditionProvenance { + actor_type: ActorType::User, + origin: OriginProvenance::from_empty_type(OriginType::Api), + sources: Vec::new(), + }, + conversions: HashMap::new(), + }, + ) + .await + .expect("could not create data type"); + + assert!( + is_queryable(&api, &list_v1.id).await, + "data type should be queryable after creation" + ); + + api.archive_data_type( + api.account_id, + ArchiveDataTypeParams { + data_type_id: Cow::Borrowed(&list_v1.id), + }, + ) + .await + .expect("could not archive data type"); + + assert!( + !is_queryable(&api, &list_v1.id).await, + "data type should not be queryable after archival" + ); + + api.unarchive_data_type( + api.account_id, + UnarchiveDataTypeParams { + data_type_id: list_v1.id.clone(), + provenance: ProvidedOntologyEditionProvenance { + actor_type: ActorType::User, + origin: OriginProvenance::from_empty_type(OriginType::Api), + sources: Vec::new(), + }, + }, + ) + .await + .expect("could not unarchive data type"); + + assert!( + is_queryable(&api, &list_v1.id).await, + "data type should be queryable after unarchival" + ); +} + +/// The resolved temporal axes of a subgraph response are taken from the database clock: database +/// clock readings bracketing the query must also bracket the resolved pinned timestamp, +/// regardless of the host clock. +#[tokio::test] +async fn resolved_temporal_axes_use_database_clock() { + let value_v1: DataType = serde_json::from_str(hash_graph_test_data::data_type::VALUE_V1) + .expect("could not parse data type representation"); + + let mut database = DatabaseTestWrapper::new().await; + let api = database + .seed([hash_graph_test_data::data_type::VALUE_V1], [], []) + .await + .expect("could not seed database"); + + let db_time_before: Timestamp<()> = api + .store + .as_client() + .query_one("SELECT statement_timestamp();", &[]) + .await + .expect("could not read the database clock") + .get(0); + + let response = api + .query_data_type_subgraph( + api.account_id, + QueryDataTypeSubgraphParams::Paths { + traversal_paths: Vec::new(), + request: QueryDataTypesParams { + filter: Filter::for_versioned_url(&value_v1.id), + temporal_axes: QueryTemporalAxesUnresolved::live_only(), + after: None, + limit: None, + include_count: false, + }, + }, + ) + .await + .expect("could not query data type subgraph"); + + let db_time_after: Timestamp<()> = api + .store + .as_client() + .query_one("SELECT statement_timestamp();", &[]) + .await + .expect("could not read the database clock") + .get(0); + + let QueryTemporalAxes::DecisionTime { pinned, .. } = response.subgraph.temporal_axes.resolved + else { + panic!("expected decision-time temporal axes"); + }; + let pinned_timestamp: Timestamp<()> = pinned.timestamp.cast(); + + assert!( + db_time_before <= pinned_timestamp && pinned_timestamp <= db_time_after, + "resolved pinned timestamp should come from the database clock: expected {db_time_before} \ + <= {pinned_timestamp} <= {db_time_after}" + ); +} diff --git a/yarn.lock b/yarn.lock index bc2358930be..ee9043e258f 100644 --- a/yarn.lock +++ b/yarn.lock @@ -13586,6 +13586,7 @@ __metadata: "@rust/error-stack": "workspace:*" "@rust/hash-codec": "workspace:*" "@rust/hash-codegen": "workspace:*" + "@rust/hash-graph-temporal-versioning": "workspace:*" typescript: "npm:5.9.3" languageName: unknown linkType: soft