ack

Author: Shubham8287

crates/client-api/src/routes/database.rs

@@ -374,6 +374,30 @@ pub async fn status_2pc<S: ControlStateDelegate + NodeDelegate>(
     Ok((StatusCode::OK, decision))
 }
 
+/// 2PC commit-ack endpoint.
+///
+/// Called by participant B after it commits via the status-poll recovery path,
+/// so that the coordinator can delete its `st_2pc_coordinator_log` entry.
+///
+/// `POST /v1/database/:name_or_identity/2pc/ack-commit/:prepare_id`
+pub async fn ack_commit_2pc<S: ControlStateDelegate + NodeDelegate>(
+    State(worker_ctx): State<S>,
+    Extension(_auth): Extension<SpacetimeAuth>,
+    Path(TwoPcParams {
+        name_or_identity,
+        prepare_id,
+    }): Path<TwoPcParams>,
+) -> axum::response::Result<impl IntoResponse> {
+    let (module, _database) = find_module_and_database(&worker_ctx, name_or_identity).await?;
+
+    module.ack_2pc_coordinator_commit(&prepare_id).map_err(|e| {
+        log::error!("2PC ack-commit failed: {e}");
+        (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()).into_response()
+    })?;
+
+    Ok(StatusCode::OK)
+}
+
 fn reducer_outcome_response(
     module: &ModuleHost,
     owner_identity: &Identity,
@@ -1388,6 +1412,8 @@ pub struct DatabaseRoutes<S> {
     pub abort_2pc_post: MethodRouter<S>,
     /// GET: /database/:name_or_identity/2pc/status/:prepare_id
     pub status_2pc_get: MethodRouter<S>,
+    /// POST: /database/:name_or_identity/2pc/ack-commit/:prepare_id
+    pub ack_commit_2pc_post: MethodRouter<S>,
 }
 
 impl<S> Default for DatabaseRoutes<S>
@@ -1417,6 +1443,7 @@ where
             commit_2pc_post: post(commit_2pc::<S>),
             abort_2pc_post: post(abort_2pc::<S>),
             status_2pc_get: get(status_2pc::<S>),
+            ack_commit_2pc_post: post(ack_commit_2pc::<S>),
         }
     }
 }
@@ -1445,7 +1472,8 @@ where
             .route("/prepare/:reducer", self.prepare_post)
             .route("/2pc/commit/:prepare_id", self.commit_2pc_post)
             .route("/2pc/abort/:prepare_id", self.abort_2pc_post)
-            .route("/2pc/status/:prepare_id", self.status_2pc_get);
+            .route("/2pc/status/:prepare_id", self.status_2pc_get)
+            .route("/2pc/ack-commit/:prepare_id", self.ack_commit_2pc_post);
 
         axum::Router::new()
             .route("/", self.root_post)

crates/core/src/host/module_host.rs

@@ -1875,6 +1875,16 @@ impl ModuleHost {
         Ok(())
     }
 
+    /// Delete a coordinator log entry for `prepare_id`.
+    /// Called when B has confirmed it committed, so A can stop retransmitting.
+    pub fn ack_2pc_coordinator_commit(&self, prepare_id: &str) -> Result<(), anyhow::Error> {
+        let db = self.relational_db().clone();
+        db.with_auto_commit::<_, _, anyhow::Error>(Workload::Internal, |tx| {
+            tx.delete_st_2pc_coordinator_log(prepare_id)
+                .map_err(anyhow::Error::from)
+        })
+    }
+
     /// Check whether `prepare_id` is present in the coordinator log of this database.
     /// Used by participant B to ask coordinator A: "did you commit?"
     pub fn has_2pc_coordinator_commit(&self, prepare_id: &str) -> bool {
@@ -2039,6 +2049,14 @@ impl ModuleHost {
                             Some(commit) => {
                                 if commit {
                                     let _ = this2.commit_prepared(&new_prepare_id);
+                                    // Tell A we committed so it can delete its coordinator log entry.
+                                    Self::send_ack_commit_to_coordinator(
+                                        &client,
+                                        &router,
+                                        auth_token.clone(),
+                                        coordinator_identity,
+                                        &original_prepare_id,
+                                    ).await;
                                 } else {
                                     let _ = this2.abort_prepared(&new_prepare_id);
                                 }
@@ -2099,6 +2117,45 @@ impl ModuleHost {
         }
     }
 
+    /// POST `POST /v1/database/{coordinator}/2pc/ack-commit/{prepare_id}` to tell A that
+    /// B has committed, so A can delete its coordinator log entry.
+    async fn send_ack_commit_to_coordinator(
+        client: &reqwest::Client,
+        router: &std::sync::Arc<dyn crate::host::reducer_router::ReducerCallRouter>,
+        auth_token: Option<String>,
+        coordinator_identity: Identity,
+        prepare_id: &str,
+    ) {
+        let base_url = match router.resolve_base_url(coordinator_identity).await {
+            Ok(url) => url,
+            Err(e) => {
+                log::warn!("2PC ack-commit: cannot resolve coordinator URL: {e}");
+                return;
+            }
+        };
+        let url = format!(
+            "{}/v1/database/{}/2pc/ack-commit/{}",
+            base_url,
+            coordinator_identity.to_hex(),
+            prepare_id,
+        );
+        let mut req = client.post(&url);
+        if let Some(token) = &auth_token {
+            req = req.header(http::header::AUTHORIZATION, format!("Bearer {token}"));
+        }
+        match req.send().await {
+            Ok(resp) if resp.status().is_success() => {
+                log::info!("2PC ack-commit: notified coordinator for {prepare_id}");
+            }
+            Ok(resp) => {
+                log::warn!("2PC ack-commit: coordinator returned {} for {prepare_id}", resp.status());
+            }
+            Err(e) => {
+                log::warn!("2PC ack-commit: transport error for {prepare_id}: {e}");
+            }
+        }
+    }
+
     pub async fn call_view_add_single_subscription(
         &self,
         sender: Arc<ClientConnectionSender>,