Fix some bugs and add some logs

Author: jsdt

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

@@ -20,8 +20,8 @@ use axum::response::{ErrorResponse, IntoResponse};
 use axum::routing::MethodRouter;
 use axum::Extension;
 use axum_extra::TypedHeader;
-use http::HeaderMap;
 use futures::TryStreamExt;
+use http::HeaderMap;
 use http::StatusCode;
 use log::{info, warn};
 use serde::Deserialize;
@@ -421,7 +421,12 @@ pub async fn wound_2pc<S: ControlStateDelegate + NodeDelegate>(
     let tx_id = global_tx_id
         .parse::<GlobalTxId>()
         .map_err(|e| (StatusCode::BAD_REQUEST, e).into_response())?;
-    let (module, _database) = find_module_and_database(&worker_ctx, name_or_identity).await?;
+    let (module, database) = find_module_and_database(&worker_ctx, name_or_identity).await?;
+
+    log::info!(
+        "received 2PC wound request for transaction {tx_id} on database {}",
+        database.database_identity
+    );
 
     module.wound_global_tx(tx_id).await.map_err(|e| {
         log::warn!("2PC wound failed for {tx_id}: {e}");

crates/core/src/client/client_connection.rs

@@ -14,7 +14,7 @@ use crate::error::DBError;
 use crate::host::module_host::ClientConnectedError;
 use crate::host::{
     CallProcedureReturn, FunctionArgs, ModuleHost, NoSuchModule, ProcedureCallResult, ReducerCallError,
-    ReducerCallResult,
+    ReducerCallResult, ReducerOutcome,
 };
 use crate::subscription::module_subscription_manager::BroadcastError;
 use crate::subscription::row_list_builder_pool::JsonRowListBuilderFakePool;
@@ -870,18 +870,33 @@ impl ClientConnection {
         timer: Instant,
         _flags: ws_v2::CallReducerFlags,
     ) -> Result<ReducerCallResult, ReducerCallError> {
-        self.module()
-            .call_reducer(
-                self.id.identity,
-                Some(self.id.connection_id),
-                None,
-                Some(self.sender()),
-                Some(request_id),
-                Some(timer),
-                reducer,
-                FunctionArgs::Bsatn(args),
-            )
-            .await
+        const MAX_WOUNDED_RETRIES: usize = 3;
+
+        let module = self.module();
+        let mut tx_id = module.replica_ctx().mint_global_tx_id(Timestamp::now());
+
+        for attempt in 0..=MAX_WOUNDED_RETRIES {
+            let result = module
+                .call_reducer(
+                    self.id.identity,
+                    Some(self.id.connection_id),
+                    Some(tx_id),
+                    Some(self.sender()),
+                    Some(request_id),
+                    Some(timer),
+                    reducer,
+                    FunctionArgs::Bsatn(args.clone()),
+                )
+                .await?;
+
+            if !matches!(result.outcome, ReducerOutcome::Wounded(_)) || attempt == MAX_WOUNDED_RETRIES {
+                return Ok(result);
+            }
+
+            tx_id = tx_id.next_attempt();
+        }
+
+        unreachable!("retry loop should return before exhausting attempts")
     }
 
     pub async fn call_procedure(

crates/core/src/client/messages.rs

@@ -381,9 +381,7 @@ impl ToProtocol for TransactionUpdateMessage {
                 EventStatus::Committed(_) => ws_v1::UpdateStatus::Committed(update),
                 EventStatus::FailedUser(errmsg)
                 | EventStatus::FailedInternal(errmsg)
-                | EventStatus::Wounded(errmsg) => {
-                    ws_v1::UpdateStatus::Failed(errmsg.clone().into())
-                }
+                | EventStatus::Wounded(errmsg) => ws_v1::UpdateStatus::Failed(errmsg.clone().into()),
                 EventStatus::OutOfEnergy => ws_v1::UpdateStatus::OutOfEnergy,
             };