refactor: Make submit return ok if parameters are the same

tomasol · tomasol · commit 944e78e2adab · 2025-12-16T21:55:49.000+01:00
diff --git a/proto/obelisk.proto b/proto/obelisk.proto
@@ -175,6 +175,12 @@ message SubmitRequest {
 }
 
 message SubmitResponse {
+    enum Outcome {
+        CREATED = 0;
+        EXISTS_WITH_SAME_PARAMETERS = 1;
+    }
+
+    Outcome outcome = 1;
 }
 
 message StubRequest {
diff --git a/src/command/server.rs b/src/command/server.rs
@@ -54,6 +54,7 @@ use concepts::component_id::InputContentDigest;
 use concepts::storage::CreateRequest;
 use concepts::storage::DbConnection;
 use concepts::storage::DbErrorWrite;
+use concepts::storage::DbErrorWriteNonRetriable;
 use concepts::storage::DbExecutor;
 use concepts::storage::DbPool;
 use concepts::storage::DbPoolCloseable;
@@ -81,7 +82,6 @@ use grpc::grpc_gen::GetStatusResponse;
 use grpc::grpc_gen::get_status_response::Message;
 use grpc::grpc_mapping::TonicServerResultExt;
 use grpc::grpc_mapping::db_error_read_to_status;
-use grpc::grpc_mapping::db_error_write_to_status;
 use hashbrown::HashMap;
 use itertools::Either;
 use serde_json::json;
@@ -192,20 +192,15 @@ pub(crate) enum SubmitError {
     FunctionNotFound,
     #[error("{0}")]
     ParamsInvalid(String),
+    #[error("execution already exists with the same id and different parameters")]
+    Conflict,
     #[error(transparent)]
     DbErrorWrite(DbErrorWrite),
 }
-impl From<SubmitError> for tonic::Status {
-    fn from(value: SubmitError) -> Self {
-        match value {
-            SubmitError::ExecutionIdMustBeTopLevel => tonic::Status::invalid_argument(
-                "argument `execution_id` must be a top-level execution id",
-            ),
-            SubmitError::FunctionNotFound => tonic::Status::not_found("function not found"),
-            SubmitError::ParamsInvalid(reason) => tonic::Status::invalid_argument(reason),
-            SubmitError::DbErrorWrite(db_err) => db_error_write_to_status(&db_err),
-        }
-    }
+
+pub(crate) enum SubmitOutcome {
+    Created,
+    ExistsWithSameParameters,
 }
 
 pub(crate) async fn submit(
@@ -214,7 +209,7 @@ pub(crate) async fn submit(
     ffqn: FunctionFqn,
     mut params: Vec<serde_json::Value>,
     component_registry_ro: &ComponentConfigRegistryRO,
-) -> Result<(), SubmitError> {
+) -> Result<SubmitOutcome, SubmitError> {
     let span = Span::current();
     span.record("execution_id", tracing::field::display(&execution_id));
     if !execution_id.is_top_level() {
@@ -317,21 +312,34 @@ pub(crate) async fn submit(
 
     // Associate the (root) request execution with the request span. Makes possible to find the trace by execution id.
     let metadata = concepts::ExecutionMetadata::from_parent_span(&span);
-    db_connection
+    let res = db_connection
         .create(CreateRequest {
             created_at,
             execution_id: execution_id.clone(),
             metadata,
             ffqn: ffqn.clone(),
-            params,
+            params: params.clone(),
             parent: None,
             scheduled_at,
             component_id: component_id.clone(),
             scheduled_by: None,
         })
-        .await
-        .map_err(SubmitError::DbErrorWrite)?;
-    Ok(())
+        .await;
+    match res {
+        Ok(_) => Ok(SubmitOutcome::Created),
+        Err(DbErrorWrite::NonRetriable(DbErrorWriteNonRetriable::Conflict)) => {
+            let create_req = db_connection
+                .get_create_request(&execution_id)
+                .await
+                .map_err(|err| SubmitError::DbErrorWrite(err.into()))?;
+            if create_req.params == params {
+                Ok(SubmitOutcome::ExistsWithSameParameters)
+            } else {
+                Err(SubmitError::Conflict)
+            }
+        }
+        Err(err) => Err(SubmitError::DbErrorWrite(err)),
+    }
 }
 
 pub(crate) async fn poll_status(
diff --git a/src/server/grpc.rs b/src/server/grpc.rs
@@ -1,6 +1,7 @@
 use crate::command::server;
 use crate::command::server::ComponentConfigRegistryRO;
 use crate::command::server::ComponentSourceMap;
+use crate::command::server::SubmitError;
 use chrono::DateTime;
 use chrono::Utc;
 use concepts::ComponentId;
@@ -30,6 +31,7 @@ use grpc::grpc_gen::get_status_response::Message;
 use grpc::grpc_mapping::TonicServerOptionExt;
 use grpc::grpc_mapping::TonicServerResultExt;
 use grpc::grpc_mapping::convert_length;
+use grpc::grpc_mapping::db_error_write_to_status;
 use grpc::grpc_mapping::from_execution_event_to_grpc;
 use grpc_gen::ExecutionSummary;
 use serde::Deserialize;
@@ -131,7 +133,7 @@ impl grpc_gen::execution_repository_server::ExecutionRepository for GrpcServer {
                 .vec
         };
 
-        server::submit(
+        let outcome = server::submit(
             self.db_pool.connection().as_ref(),
             execution_id,
             ffqn,
@@ -140,7 +142,15 @@ impl grpc_gen::execution_repository_server::ExecutionRepository for GrpcServer {
         )
         .await?;
 
-        let resp = grpc_gen::SubmitResponse {};
+        let resp = grpc_gen::SubmitResponse {
+            outcome: match outcome {
+                server::SubmitOutcome::Created => grpc_gen::submit_response::Outcome::Created,
+                server::SubmitOutcome::ExistsWithSameParameters => {
+                    grpc_gen::submit_response::Outcome::ExistsWithSameParameters
+                }
+            }
+            .into(),
+        };
         Ok(tonic::Response::new(resp))
     }
 
@@ -755,3 +765,17 @@ fn list_fns(functions: Vec<FunctionMetadata>) -> Vec<grpc_gen::FunctionDetail> {
     }
     vec
 }
+
+impl From<SubmitError> for tonic::Status {
+    fn from(value: SubmitError) -> Self {
+        match value {
+            SubmitError::ExecutionIdMustBeTopLevel => tonic::Status::invalid_argument(
+                "argument `execution_id` must be a top-level execution id",
+            ),
+            SubmitError::FunctionNotFound => tonic::Status::not_found("function not found"),
+            SubmitError::ParamsInvalid(reason) => tonic::Status::invalid_argument(reason),
+            err @ SubmitError::Conflict => tonic::Status::already_exists(err.to_string()),
+            SubmitError::DbErrorWrite(db_err) => db_error_write_to_status(&db_err),
+        }
+    }
+}
diff --git a/src/server/web_api.rs b/src/server/web_api.rs
@@ -1,5 +1,5 @@
 use crate::{
-    command::server::{self, ComponentConfigRegistryRO, SubmitError},
+    command::server::{self, ComponentConfigRegistryRO, SubmitError, SubmitOutcome},
     server::web_api::components::{component_wit, components_list},
 };
 use axum::{
@@ -15,9 +15,9 @@ use concepts::{
     component_id::InputContentDigest,
     prefixed_ulid::{DelayId, ExecutionIdDerived},
     storage::{
-        self, CancelOutcome, DbErrorGeneric, DbErrorRead, DbErrorWrite, DbErrorWriteNonRetriable,
-        DbPool, ExecutionListPagination, ExecutionRequest, ExecutionWithState, Pagination,
-        PendingState, Version, VersionType,
+        self, CancelOutcome, DbErrorGeneric, DbErrorRead, DbErrorWrite, DbPool,
+        ExecutionListPagination, ExecutionRequest, ExecutionWithState, Pagination, PendingState,
+        Version, VersionType,
     },
     time::{ClockFn as _, Now},
 };
@@ -49,7 +49,7 @@ fn v1_router() -> Router<Arc<WebApiState>> {
         .route("/delays/{delay-id}/cancel", routing::put(delay_cancel))
         .route("/execution-id", routing::get(execution_id_generate))
         .route("/executions", routing::get(executions_list))
-        .route("/executions/submit", routing::post(execution_submit_post))
+        .route("/executions", routing::post(execution_submit_post))
         .route(
             "/executions/{execution-id}/cancel",
             routing::put(execution_cancel),
@@ -513,13 +513,7 @@ async fn execution_submit_put(
     accept: AcceptHeader,
     Json(payload): Json<ExecutionPutPayload>,
 ) -> Result<Response, HttpResponse> {
-    execution_submit(execution_id.clone(), state, payload, accept).await?;
-    Ok(HttpResponse {
-        status: StatusCode::CREATED,
-        message: execution_id.to_string(),
-        accept,
-    }
-    .into_response())
+    execution_submit(execution_id, state, payload, accept).await
 }
 
 async fn execution_submit_post(
@@ -528,43 +522,40 @@ async fn execution_submit_post(
     Json(payload): Json<ExecutionPutPayload>,
 ) -> Result<Response, HttpResponse> {
     let execution_id = ExecutionId::generate();
-    execution_submit(execution_id.clone(), state, payload, accept).await?;
-    Ok(HttpResponse {
-        status: StatusCode::CREATED,
-        message: execution_id.to_string(),
-        accept,
-    }
-    .into_response())
+    execution_submit(execution_id, state, payload, accept).await
 }
 
 async fn execution_submit(
     execution_id: ExecutionId,
     state: State<Arc<WebApiState>>,
     payload: ExecutionPutPayload,
     accept: AcceptHeader,
-) -> Result<(), HttpResponse> {
-    server::submit(
-        state.db_pool.connection().as_ref(),
-        execution_id,
+) -> Result<Response, HttpResponse> {
+    let conn = state.db_pool.connection();
+    let res = server::submit(
+        conn.as_ref(),
+        execution_id.clone(),
         payload.ffqn,
         payload.params,
         &state.component_registry_ro,
     )
-    .await
-    .map_err(|err| match err {
-        SubmitError::DbErrorWrite(DbErrorWrite::NonRetriable(
-            DbErrorWriteNonRetriable::Conflict,
-        )) => HttpResponse {
-            status: StatusCode::CONFLICT,
-            message: "already exists".to_string(),
+    .await;
+
+    match res {
+        Ok(SubmitOutcome::Created) => Ok(HttpResponse {
+            status: StatusCode::CREATED,
+            message: execution_id.to_string(),
             accept,
-        },
-        err => HttpResponse {
-            status: StatusCode::INTERNAL_SERVER_ERROR,
-            message: err.to_string(),
+        }
+        .into_response()),
+        Ok(SubmitOutcome::ExistsWithSameParameters) => Ok(HttpResponse {
+            status: StatusCode::OK,
+            message: execution_id.to_string(),
             accept,
-        },
-    })
+        }
+        .into_response()),
+        Err(err) => Err(ErrorWrapper(err, accept).into()),
+    }
 }
 
 pub(crate) mod components {
@@ -601,7 +592,10 @@ pub(crate) mod components {
         state: State<Arc<WebApiState>>,
     ) -> Result<Response, HttpResponse> {
         let Some(wit) = state.component_registry_ro.get_wit(&digest) else {
-            return Err(HttpResponse::not_found(AcceptHeader::Text));
+            return Err(HttpResponse::not_found(
+                AcceptHeader::Text,
+                Some("component"),
+            ));
         };
         Ok(if let Some(wit) = wit {
             wit.to_string().into_response()
@@ -778,10 +772,14 @@ impl HttpResponse {
         }
     }
 
-    fn not_found(accept: AcceptHeader) -> Self {
+    fn not_found(accept: AcceptHeader, what: Option<&str>) -> Self {
         HttpResponse {
             status: StatusCode::NOT_FOUND,
-            message: "not found".to_string(),
+            message: if let Some(what) = what {
+                format!("{what} not found")
+            } else {
+                "not found".to_string()
+            },
             accept,
         }
     }
@@ -817,7 +815,7 @@ impl From<ErrorWrapper<DbErrorRead>> for HttpResponse {
     fn from(value: ErrorWrapper<DbErrorRead>) -> Self {
         let accept = value.1;
         let (status, message) = match value.0 {
-            DbErrorRead::NotFound => return HttpResponse::not_found(accept),
+            DbErrorRead::NotFound => return HttpResponse::not_found(accept, None),
             DbErrorRead::Generic(err) => (StatusCode::SERVICE_UNAVAILABLE, err.to_string()),
         };
         HttpResponse {
@@ -831,7 +829,7 @@ impl From<ErrorWrapper<DbErrorWrite>> for HttpResponse {
     fn from(value: ErrorWrapper<DbErrorWrite>) -> Self {
         let accept = value.1;
         let (status, message) = match value.0 {
-            DbErrorWrite::NotFound => return HttpResponse::not_found(accept),
+            DbErrorWrite::NotFound => return HttpResponse::not_found(accept, None),
             DbErrorWrite::Generic(err) => (StatusCode::SERVICE_UNAVAILABLE, err.to_string()),
             DbErrorWrite::NonRetriable(err) => (StatusCode::INTERNAL_SERVER_ERROR, err.to_string()),
         };
@@ -842,3 +840,26 @@ impl From<ErrorWrapper<DbErrorWrite>> for HttpResponse {
         }
     }
 }
+impl From<ErrorWrapper<SubmitError>> for HttpResponse {
+    fn from(value: ErrorWrapper<SubmitError>) -> Self {
+        let accept = value.1;
+        match value.0 {
+            err @ SubmitError::Conflict => HttpResponse {
+                status: StatusCode::CONFLICT,
+                message: err.to_string(),
+                accept,
+            },
+            SubmitError::FunctionNotFound => HttpResponse::not_found(accept, Some("ffqn")),
+            SubmitError::DbErrorWrite(db_error_write) => {
+                HttpResponse::from(ErrorWrapper(db_error_write, accept))
+            }
+            err @ (SubmitError::ExecutionIdMustBeTopLevel | SubmitError::ParamsInvalid(_)) => {
+                HttpResponse {
+                    status: StatusCode::BAD_REQUEST,
+                    message: err.to_string(),
+                    accept,
+                }
+            }
+        }
+    }
+}

Original file line number	Diff line number	Diff line change
`@@ -175,6 +175,12 @@ message SubmitRequest {`
`175`	`175`	`}`
`176`	`176`
`177`	`177`	`message SubmitResponse {`
	`178`	`+ enum Outcome {`
	`179`	`+ CREATED = 0;`
	`180`	`+ EXISTS_WITH_SAME_PARAMETERS = 1;`
	`181`	`+ }`
	`182`	`+`
	`183`	`+ Outcome outcome = 1;`
`178`	`184`	`}`
`179`	`185`
`180`	`186`	`message StubRequest {`