BE-2: Use the Postgres clock for writing timestamps and resolving temporal axes

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 }

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:*",

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<ActorId>,
+    timestamp: Timestamp<()>,
     is_instance_admin: bool,
     policies: Vec<ResolvedPolicy>,
     tracked_actions: HashMap<ActionName, Option<OptimizationData>>,
@@ -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<Timestamp<()>>,
     context: ContextBuilder,
     entity_type_ids: HashSet<Cow<'a, VersionedUrl>>,
     property_type_ids: HashSet<Cow<'a, VersionedUrl>>,
@@ -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)]),

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 {

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<Option<ActorId>, Report<DetermineActorError>>;
 
+    /// 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<ActorId>, Timestamp<()>), Report<DetermineActorError>>;
+
+    /// 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<Timestamp<()>, Report<CurrentTimestampError>>;
+
     /// Builds a context used to evaluate policies for an actor.
     ///
     /// # Errors

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();

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<DeletionSummary, Report<DeletionError>> {
-        let transaction_time = Timestamp::<TransactionTime>::now();
+        let transaction_time = Timestamp::<TransactionTime>::from_anonymous(
+            self.current_timestamp()
+                .await
+                .change_context(DeletionError::Store)?,
+        );
         let decision_time = params
             .decision_time
             .unwrap_or_else(|| transaction_time.cast());