Merge pull request #709 from obeli-sk/deployment-idempotent

refactor(api,cli): Add optional deployment ID for idempotent submission

Author: tomasol

assets/schemas/cli.json

@@ -644,6 +644,10 @@
               "name": "--description",
               "help": "Optional human-readable description"
             },
+            {
+              "name": "--deployment-id",
+              "help": "Optional client-supplied deployment ID for idempotent submission"
+            },
             {
               "name": "--api-url",
               "short": "a",
@@ -676,6 +680,10 @@
               "name": "--description",
               "help": "Optional human-readable description for a newly submitted deployment"
             },
+            {
+              "name": "--deployment-id",
+              "help": "Optional client-supplied deployment ID for idempotent submission"
+            },
             {
               "name": "--api-url",
               "short": "a",
@@ -701,6 +709,10 @@
               "name": "--description",
               "help": "Optional human-readable description for a newly submitted deployment"
             },
+            {
+              "name": "--deployment-id",
+              "help": "Optional client-supplied deployment ID for idempotent submission"
+            },
             {
               "name": "--api-url",
               "short": "a",
@@ -1010,6 +1022,17 @@
             }
           ]
         },
+        {
+          "name": "deployment-id",
+          "about": "Generate a fresh random deployment ID and print it to stdout",
+          "options": [
+            {
+              "name": "--json",
+              "short": "j",
+              "help": "Output as JSON instead of plain text"
+            }
+          ]
+        },
         {
           "name": "prompt",
           "about": "Print a prompt context for authoring an Obelisk application with a coding agent",

assets/schemas/openapi.json

@@ -2087,6 +2087,13 @@
             "type": "string",
             "description": "Deployment config as JSON string"
           },
+          "deployment_id": {
+            "type": [
+              "string",
+              "null"
+            ],
+            "description": "Optional client-supplied deployment ID for idempotent submission.\nIf a deployment with this ID already exists and its content digest\nmatches, the submission is a no-op; a digest mismatch is rejected."
+          },
           "description": {
             "type": [
               "string",

proto/obelisk.proto

@@ -1100,6 +1100,11 @@ message SubmitDeploymentRequest {
     // Verify all environment variables before persisting the deployment.
     bool verify = 3;
     optional string description = 4;
+    // Optional client-supplied deployment ID for idempotent submission.
+    // If a deployment with this ID already exists and its content digest matches
+    // the submitted config, the submission is a no-op and returns this ID.
+    // A digest mismatch is rejected. When unset, a fresh ID is generated.
+    optional DeploymentId deployment_id = 5;
 }
 message SubmitDeploymentResponse {
     DeploymentId deployment_id = 1;

src/args.rs

@@ -108,6 +108,13 @@ pub(crate) enum Deployment {
         /// Optional human-readable description.
         #[arg(long)]
         description: Option<String>,
+        /// Optional client-supplied deployment ID for idempotent submission.
+        ///
+        /// If a deployment with this ID already exists and its content digest
+        /// matches, the submission is a no-op; a digest mismatch is an error.
+        /// Generate one with `obelisk generate deployment-id`.
+        #[arg(long)]
+        deployment_id: Option<DeploymentId>,
         /// Address of the obelisk server
         #[arg(short, long, default_value = "http://127.0.0.1:5005")]
         api_url: String,
@@ -130,6 +137,13 @@ pub(crate) enum Deployment {
         /// Optional human-readable description for a newly submitted deployment.
         #[arg(long)]
         description: Option<String>,
+        /// Optional client-supplied deployment ID for idempotent submission.
+        ///
+        /// Only valid when submitting a file/empty deployment, not when `source`
+        /// is an existing deployment ID. Generate one with
+        /// `obelisk generate deployment-id`.
+        #[arg(long)]
+        deployment_id: Option<DeploymentId>,
         /// Address of the obelisk server
         #[arg(short, long, default_value = "http://127.0.0.1:5005")]
         api_url: String,
@@ -149,6 +163,13 @@ pub(crate) enum Deployment {
         /// Optional human-readable description for a newly submitted deployment.
         #[arg(long)]
         description: Option<String>,
+        /// Optional client-supplied deployment ID for idempotent submission.
+        ///
+        /// Only valid when submitting a file/empty deployment, not when `source`
+        /// is an existing deployment ID. Generate one with
+        /// `obelisk generate deployment-id`.
+        #[arg(long)]
+        deployment_id: Option<DeploymentId>,
         /// Address of the obelisk server
         #[arg(short, long, default_value = "http://127.0.0.1:5005")]
         api_url: String,
@@ -288,6 +309,15 @@ pub(crate) enum Generate {
         #[arg(short, long)]
         json: bool,
     },
+    /// Generate a fresh random deployment ID and print it to stdout.
+    ///
+    /// The ID can be passed to `deployment submit/enqueue/apply --deployment-id`
+    /// for idempotent submission.
+    DeploymentId {
+        /// Output as JSON instead of plain text.
+        #[arg(short, long)]
+        json: bool,
+    },
     /// Print a prompt context for authoring an Obelisk application with a coding agent.
     ///
     /// Usage: obelisk generate prompt description of what to build | claude

src/command/deployment.rs

@@ -19,6 +19,7 @@ impl args::Deployment {
                 empty,
                 verify,
                 description,
+                deployment_id,
                 api_url,
             } => {
                 let config_json = load_config_json_from_file_or_empty(file, empty).await?;
@@ -30,6 +31,7 @@ impl args::Deployment {
                         created_by: Some("cli".to_string()),
                         verify,
                         description,
+                        deployment_id: deployment_id.map(grpc_gen::DeploymentId::from),
                     })
                     .await?
                     .into_inner();
@@ -43,6 +45,7 @@ impl args::Deployment {
                 empty,
                 verify,
                 description,
+                deployment_id,
                 api_url,
             } => {
                 let channel = to_channel(&api_url).await?;
@@ -53,6 +56,7 @@ impl args::Deployment {
                     empty,
                     false, // will be verified in switch
                     description,
+                    deployment_id,
                 )
                 .await?;
                 switch_deployment(
@@ -68,6 +72,7 @@ impl args::Deployment {
                 source,
                 empty,
                 description,
+                deployment_id,
                 api_url,
             } => {
                 let channel = to_channel(&api_url).await?;
@@ -78,6 +83,7 @@ impl args::Deployment {
                     empty,
                     false, // will be verified in switch
                     description,
+                    deployment_id,
                 )
                 .await?;
                 switch_deployment(
@@ -167,13 +173,17 @@ async fn submit_deployment(
     empty: bool,
     verify: bool,
     description: Option<String>,
+    deployment_id: Option<DeploymentId>,
 ) -> anyhow::Result<DeploymentId> {
     assert_ne!(source.is_some(), empty);
     let config_json = match source {
         Some(DeploymentSource::Id(id)) => {
             if description.is_some() {
                 bail!("--description cannot be used with an existing deployment ID");
             }
+            if deployment_id.is_some() {
+                bail!("--deployment-id cannot be used with an existing deployment ID source");
+            }
             return Ok(id);
         }
         Some(DeploymentSource::File(path)) => load_config_json(path).await?,
@@ -185,6 +195,7 @@ async fn submit_deployment(
             created_by: Some("cli".to_string()),
             verify,
             description,
+            deployment_id: deployment_id.map(grpc_gen::DeploymentId::from),
         })
         .await?
         .into_inner();

src/command/generate.rs

@@ -130,6 +130,15 @@ impl Generate {
                 }
                 Ok(())
             }
+            Generate::DeploymentId { json } => {
+                let deployment_id = DeploymentId::generate();
+                if json {
+                    println!("{}", serde_json::to_string_pretty(&deployment_id)?);
+                } else {
+                    println!("{deployment_id}");
+                }
+                Ok(())
+            }
             Generate::Prompt { description } => {
                 let version = format!("v{PKG_VERSION}");
                 let description = description.join(" ");

src/command/integration_tests.rs

@@ -875,6 +875,26 @@ impl TestServer {
             .expect("webapi submit deployment: invalid deployment id")
     }
 
+    /// Submit a deployment via the Web API with an explicit (idempotency) deployment ID,
+    /// returning the raw response so callers can assert on success or conflict.
+    async fn webapi_submit_deployment_with_id(
+        &self,
+        config_json: &str,
+        deployment_id: DeploymentId,
+    ) -> reqwest::Response {
+        self.client
+            .post(format!("{}/v1/deployments", self.base_url))
+            .header("Accept", "application/json")
+            .json(&json!({
+                "config_json": config_json,
+                "verify": false,
+                "deployment_id": deployment_id.to_string(),
+            }))
+            .send()
+            .await
+            .expect("webapi submit deployment request failed")
+    }
+
     /// Hot-redeploy to the given deployment via the Web API.
     async fn webapi_switch_hot_redeploy(&self, deployment_id: DeploymentId) {
         let resp = self
@@ -939,6 +959,7 @@ impl TestDeployClient {
                         created_by: Some("test".to_string()),
                         verify: false,
                         description: None,
+                        deployment_id: None,
                     })
                     .await
                     .unwrap()
@@ -1657,6 +1678,79 @@ async fn list_components_grpc_filters_with_explicit_deployment_id() {
     server.shutdown().await;
 }
 
+#[tokio::test]
+async fn submit_deployment_is_idempotent_by_id_and_digest_webapi() {
+    let server = TestServer::start(test_addr!(40_104)).await;
+
+    // Read the active deployment's canonical config so we can resubmit it verbatim.
+    let pool = SqlitePool::new(&server.sqlite_file, SqliteConfig::default())
+        .await
+        .unwrap();
+    let conn = pool.external_api_conn().await.unwrap();
+    let active = conn.get_active_deployment().await.unwrap().unwrap();
+    pool.close().await;
+    let canonical: DeploymentCanonical = serde_json::from_str(&active.config_json).unwrap();
+    let config_json = crate::config::toml::compute_config_json(&canonical);
+
+    let deployment_id = DeploymentId::generate();
+
+    // First submission under the explicit ID creates the deployment.
+    let resp1 = server
+        .webapi_submit_deployment_with_id(&config_json, deployment_id)
+        .await;
+    assert!(
+        resp1.status().is_success(),
+        "first submit failed: {}",
+        resp1.status()
+    );
+    let body1: Value = resp1.json().await.unwrap();
+    let returned1: DeploymentId = body1["ok"].as_str().unwrap().parse().unwrap();
+    assert_eq!(deployment_id, returned1, "explicit ID must be honored");
+
+    // Resubmitting the identical config under the same ID is an idempotent no-op.
+    let resp2 = server
+        .webapi_submit_deployment_with_id(&config_json, deployment_id)
+        .await;
+    assert!(
+        resp2.status().is_success(),
+        "idempotent resubmit failed: {}",
+        resp2.status()
+    );
+    let body2: Value = resp2.json().await.unwrap();
+    let returned2: DeploymentId = body2["ok"].as_str().unwrap().parse().unwrap();
+    assert_eq!(
+        deployment_id, returned2,
+        "no-op resubmit must return same ID"
+    );
+
+    // Submitting a different config under the same ID is rejected as a digest conflict.
+    let mut mutated: DeploymentCanonical = serde_json::from_str(&active.config_json).unwrap();
+    mutated
+        .activities_stub
+        .push(ActivityStubComponentConfigCanonical::Inline(
+            ActivityStubExtInlineConfigCanonical {
+                name: ConfigName::new(concepts::StrVariant::from("idempotency_conflict_stub"))
+                    .unwrap(),
+                ffqn: "testing:integration/stubs.idempotency-conflict"
+                    .parse()
+                    .unwrap(),
+                params: Some(vec![]),
+                return_type: Some("result<string, string>".to_string()),
+            },
+        ));
+    let mutated_json = crate::config::toml::compute_config_json(&mutated);
+    let resp3 = server
+        .webapi_submit_deployment_with_id(&mutated_json, deployment_id)
+        .await;
+    assert_eq!(
+        resp3.status(),
+        reqwest::StatusCode::BAD_REQUEST,
+        "different config under same ID must be rejected as a digest conflict"
+    );
+
+    server.shutdown().await;
+}
+
 // ---- Activity: submit + result ----
 
 #[tokio::test]

src/command/server.rs

@@ -1504,6 +1504,7 @@ pub(crate) async fn submit_deployment(
     verify: bool,
     created_by: Option<String>,
     description: Option<String>,
+    requested_deployment_id: Option<DeploymentId>,
     prepared_dirs: &PreparedDirs,
     db_pool: Arc<dyn DbPool>,
     termination_watcher: &mut watch::Receiver<()>,
@@ -1513,6 +1514,26 @@ pub(crate) async fn submit_deployment(
         serde_json::from_str(config_json).with_context(|| "cannot parse config_json")?;
 
     let canonical_config = crate::config::toml::compute_config_json(&deployment);
+    let digest = DeploymentRecord::compute_digest(&canonical_config);
+
+    let conn = db_pool.external_api_conn().await?;
+
+    // Idempotent submission: if the caller supplied a deployment ID that already
+    // exists, return it as a no-op when the content digest matches (skipping the
+    // expensive verify/compile/link), and reject a digest mismatch as a conflict.
+    if let Some(requested_deployment_id) = requested_deployment_id
+        && let Some(existing) = conn.get_deployment(requested_deployment_id).await?
+    {
+        if existing.digest == digest {
+            info!(%requested_deployment_id, "Deployment already exists with matching digest, returning existing ID");
+            return Ok(requested_deployment_id);
+        }
+        bail!(
+            "deployment {requested_deployment_id} already exists with a different content digest \
+             (existing {}, submitted {digest}); use a fresh deployment ID",
+            existing.digest
+        );
+    }
 
     let verify_deployment_id = DeploymentId::generate();
     let server_compiled = deployment_verify_config_compile_link(
@@ -1533,10 +1554,8 @@ pub(crate) async fn submit_deployment(
     )
     .await?;
 
-    let deployment_id = DeploymentId::generate();
-    let conn = db_pool.external_api_conn().await?;
+    let deployment_id = requested_deployment_id.unwrap_or_else(DeploymentId::generate);
     let now = chrono::Utc::now();
-    let digest = DeploymentRecord::compute_digest(&canonical_config);
 
     conn.insert_deployment(DeploymentRecord {
         deployment_id,