Add journal, query and timestamp to ApplicadtionApi::execute_public_query (#26115)

Those are needed by the sync worker.

GitOrigin-RevId: d2888bd2ad04ff9a34b2e79c6bd81d033d4d1564

Author: Preslav Le

crates/application/src/api.rs

@@ -7,25 +7,36 @@ use common::{
         AllowedVisibility,
         FunctionCaller,
     },
-    value::ConvexValue,
     RequestId,
 };
 use model::session_requests::types::SessionRequestIdentifier;
 use serde_json::Value as JsonValue;
-use sync_types::AuthenticationToken;
+use sync_types::{
+    AuthenticationToken,
+    SerializedQueryJournal,
+    Timestamp,
+};
 
 use crate::{
-    redaction::{
-        RedactedJsError,
-        RedactedLogLines,
-    },
     Application,
     RedactedActionError,
     RedactedActionReturn,
     RedactedMutationError,
     RedactedMutationReturn,
+    RedactedQueryReturn,
 };
 
+#[cfg_attr(
+    any(test, feature = "testing"),
+    derive(proptest_derive::Arbitrary, Debug, Clone, PartialEq)
+)]
+pub enum ExecuteQueryTimestamp {
+    // Execute the query at the latest timestamp.
+    Latest,
+    // Execute the query at a given timestamp.
+    At(Timestamp),
+}
+
 // A trait that abstracts the backend API. It all state and validation logic
 // so http routes can be kept thin and stateless. The implementor is also
 // responsible for routing the request to the appropriate backend in the hosted
@@ -40,8 +51,9 @@ pub trait ApplicationApi: Send + Sync {
         path: ComponentFunctionPath,
         args: Vec<JsonValue>,
         caller: FunctionCaller,
-        // TODO(presley): Replace this with RedactedQueryReturn.
-    ) -> anyhow::Result<(Result<ConvexValue, RedactedJsError>, RedactedLogLines)>;
+        ts: ExecuteQueryTimestamp,
+        journal: Option<SerializedQueryJournal>,
+    ) -> anyhow::Result<RedactedQueryReturn>;
 
     async fn execute_public_mutation(
         &self,
@@ -77,7 +89,9 @@ impl<RT: Runtime> ApplicationApi for Application<RT> {
         path: ComponentFunctionPath,
         args: Vec<JsonValue>,
         caller: FunctionCaller,
-    ) -> anyhow::Result<(Result<ConvexValue, RedactedJsError>, RedactedLogLines)> {
+        ts: ExecuteQueryTimestamp,
+        journal: Option<SerializedQueryJournal>,
+    ) -> anyhow::Result<RedactedQueryReturn> {
         anyhow::ensure!(
             caller.allowed_visibility() == AllowedVisibility::PublicOnly,
             "This method should not be used by internal callers."
@@ -86,14 +100,12 @@ impl<RT: Runtime> ApplicationApi for Application<RT> {
         let validate_time = self.runtime().system_time();
         let identity = self.authenticate(auth_token, validate_time).await?;
 
-        let ts = *self.now_ts_for_reads();
-        let journal = None;
-
-        let query_return = self
-            .read_only_udf_at_ts(request_id, path, args, identity, ts, journal, caller)
-            .await?;
-
-        Ok((query_return.result, query_return.log_lines))
+        let ts = match ts {
+            ExecuteQueryTimestamp::Latest => *self.now_ts_for_reads(),
+            ExecuteQueryTimestamp::At(ts) => ts,
+        };
+        self.read_only_udf_at_ts(request_id, path, args, identity, ts, journal, caller)
+            .await
     }
 
     async fn execute_public_mutation(

crates/database/src/token.rs

@@ -19,7 +19,7 @@ pub type SerializedToken = String;
 /// for a transaction. This can be externalized to a user and used to represent
 /// current transaction state.
 #[derive(Clone, Debug, Eq, PartialEq)]
-#[cfg_attr(test, derive(proptest_derive::Arbitrary))]
+#[cfg_attr(any(test, feature = "testing"), derive(proptest_derive::Arbitrary))]
 pub struct Token {
     read_set: ReadSet,
     ts: Timestamp,
@@ -68,7 +68,7 @@ impl Token {
         )
     }
 
-    pub(crate) fn new(read_set: ReadSet, ts: Timestamp) -> Self {
+    pub fn new(read_set: ReadSet, ts: Timestamp) -> Self {
         Self { read_set, ts }
     }
 

crates/local_backend/src/public_api.rs

@@ -2,7 +2,10 @@ use std::sync::Arc;
 
 use anyhow::anyhow;
 use application::{
-    api::ApplicationApi,
+    api::{
+        ApplicationApi,
+        ExecuteQueryTimestamp,
+    },
     redaction::{
         RedactedJsError,
         RedactedLogLines,
@@ -219,7 +222,8 @@ pub async fn public_query_get(
 ) -> Result<impl IntoResponse, HttpResponseError> {
     let udf_path = parse_udf_path(&req.path)?;
     let args = req.args.into_arg_vec();
-    let (result, log_lines) = api
+    let journal = None;
+    let query_result = api
         .execute_public_query(
             host,
             request_id,
@@ -230,10 +234,13 @@ pub async fn public_query_get(
             },
             args,
             FunctionCaller::HttpApi(client_version.clone()),
+            ExecuteQueryTimestamp::Latest,
+            journal,
         )
         .await?;
     let value_format = req.format.as_ref().map(|f| f.parse()).transpose()?;
-    let response = match result {
+    let log_lines = query_result.log_lines;
+    let response = match query_result.result {
         Ok(value) => UdfResponse::Success {
             value: export_value(value, value_format, client_version)?,
             log_lines,
@@ -253,7 +260,8 @@ pub async fn public_query_post(
     Json(req): Json<UdfPostRequest>,
 ) -> Result<impl IntoResponse, HttpResponseError> {
     let udf_path = parse_udf_path(&req.path)?;
-    let (result, log_lines) = api
+    let journal = None;
+    let query_return = api
         .execute_public_query(
             host,
             request_id,
@@ -264,15 +272,19 @@ pub async fn public_query_post(
             },
             req.args.into_arg_vec(),
             FunctionCaller::HttpApi(client_version.clone()),
+            ExecuteQueryTimestamp::Latest,
+            journal,
         )
         .await?;
     let value_format = req.format.as_ref().map(|f| f.parse()).transpose()?;
-    let response = match result {
+    let response = match query_return.result {
         Ok(value) => UdfResponse::Success {
             value: export_value(value, value_format, client_version)?,
-            log_lines,
+            log_lines: query_return.log_lines,
+        },
+        Err(error) => {
+            UdfResponse::error(error, query_return.log_lines, value_format, client_version)?
         },
-        Err(error) => UdfResponse::error(error, log_lines, value_format, client_version)?,
     };
     Ok(Json(response))
 }