Feature/workflow record

.github/workflows/build_and_it.yml

@@ -51,7 +51,8 @@ jobs:
           tar -xvf dtm_1.18.0_linux_amd64.tar.gz
           pwd
           mkdir /home/runner/work/client-csharp/client-csharp/logs
-          nohup ./dtm > /home/runner/work/client-csharp/client-csharp/logs/dtm.log 2>&1 &
+          echo "UpdateBranchSync: 1" > ./config.yml
+          nohup ./dtm -c ./config.yml > /home/runner/work/client-csharp/client-csharp/logs/dtm.log 2>&1 &
           sleep 5
           curl "127.0.0.1:36789/api/dtmsvr/newGid"
       - name: Setup Busi Service

samples/DtmSample/Controllers/WfTestController.cs

@@ -7,12 +7,10 @@
 using Microsoft.Extensions.Options;
 using System;
 using System.IO;
-using System.Diagnostics;
 using System.Net.Http;
 using System.Net.Http.Headers;
 using System.Text;
 using System.Text.Json;
-using System.Text.Unicode;
 using System.Threading;
 using System.Threading.Tasks;
 using Exception = System.Exception;
@@ -257,5 +255,84 @@ public async Task<IActionResult> TccRollBack(CancellationToken cancellationToken
                 return Ok(TransResponse.BuildFailureResponse());
             }
         }
+
+
+        private static readonly string wfNameForResume = "wfNameForResume";
+
+        /// <summary>
+        /// 
+        /// </summary>
+        /// <param name="cancellationToken"></param>
+        /// <returns></returns>
+        [HttpPost("wf-crash")]
+        public async Task<IActionResult> Crash(CancellationToken cancellationToken)
+        {
+            if (!_globalTransaction.Exists(wfNameForResume))
+            {
+                _globalTransaction.Register(wfNameForResume, async (wf, data) =>
+                {
+                    var content = new ByteArrayContent(data);
+                    content.Headers.ContentType = new MediaTypeHeaderValue("application/json");
+
+                    var outClient = wf.NewBranch().NewRequest();
+                    await outClient.PostAsync(_settings.BusiUrl + "/TransOut", content);
+
+                    // the first branch succeed, then crashed, the dtm server will call back the flowing wf-call-back
+                    // manual stop application
+                    Environment.Exit(0);
+
+                    var inClient = wf.NewBranch().NewRequest();
+                    await inClient.PostAsync(_settings.BusiUrl + "/TransIn", content);
+
+                    return null;
+                });
+            }
+
+            var req = JsonSerializer.Serialize(new TransRequest("1", -30));
+            await _globalTransaction.Execute(wfNameForResume, Guid.NewGuid().ToString("N"), Encoding.UTF8.GetBytes(req), true);
+
+            return Ok(TransResponse.BuildSucceedResponse());
+        }
+
+        [HttpPost("wf-resume")]
+        public async Task<IActionResult> WfResume(CancellationToken cancellationToken)
+        {
+            try
+            {
+                if (!_globalTransaction.Exists(wfNameForResume))
+                {
+                    // register again after manual crash by Environment.Exit(0);
+                    _globalTransaction.Register(wfNameForResume, async (wf, data) =>
+                    {
+                        var content = new ByteArrayContent(data);
+                        content.Headers.ContentType = new MediaTypeHeaderValue("application/json");
+
+                        var outClient = wf.NewBranch().NewRequest();
+                        await outClient.PostAsync(_settings.BusiUrl + "/TransOut", content);
+
+                        var inClient = wf.NewBranch().NewRequest();
+                        await inClient.PostAsync(_settings.BusiUrl + "/TransIn", content);
+
+                        return null;
+                    });
+                }
+
+                // prepared call ExecuteByQS
+                using var bodyMemoryStream = new MemoryStream();
+                await Request.Body.CopyToAsync(bodyMemoryStream, cancellationToken);
+                byte[] bytes = bodyMemoryStream.ToArray();
+                string body = Encoding.UTF8.GetString(bytes);
+                _logger.LogDebug($"body: {body}");
+
+                await _globalTransaction.ExecuteByQS(Request.Query, bodyMemoryStream.ToArray());
+
+                return Ok(TransResponse.BuildSucceedResponse());
+            }
+            catch (Exception ex)
+            {
+                _logger.LogError(ex, "Workflow Error");
+                return Ok(TransResponse.BuildFailureResponse());
+            }
+        }
     }
 }

src/DtmCommon/Barrier/BranchBarrier.cs

@@ -42,7 +42,7 @@
 
         public int BarrierID { get; set; }
 
-        public async Task Call(DbConnection db, Func<DbTransaction, Task> busiCall)
+        public async Task<(bool done, string reason)> Call(DbConnection db, Func<DbTransaction, Task> busiCall)
         {
             this.BarrierID = this.BarrierID + 1;
             var bid = this.BarrierID.ToString().PadLeft(2, '0');
@@ -91,7 +91,7 @@
 #else
                     await tx.CommitAsync();
 #endif
-                    return;
+                    return (false, isNullCompensation ? "isNullCompensation" : "isDuplicateOrPend");
                 }
 
                 await busiCall.Invoke(tx);
@@ -118,9 +118,10 @@
 
                 throw;
             }
+            return (true, string.Empty);
         }
 
-        public async Task Call(DbConnection db, Func<Task> busiCall, TransactionScopeOption transactionScope = TransactionScopeOption.Required, IsolationLevel isolationLevel = IsolationLevel.Serializable)
+        public async Task<(bool done, string reason)> Call(DbConnection db, Func<Task> busiCall, TransactionScopeOption transactionScope = TransactionScopeOption.Required, IsolationLevel isolationLevel = IsolationLevel.Serializable)
         {
             this.BarrierID = this.BarrierID + 1;
             var bid = this.BarrierID.ToString().PadLeft(2, '0');
@@ -158,7 +159,7 @@
                     if (isNullCompensation || isDuplicateOrPend)
                     {
                         Logger?.LogInformation("Will not exec busiCall, isNullCompensation={isNullCompensation}, isDuplicateOrPend={isDuplicateOrPend}", isNullCompensation, isDuplicateOrPend);
-                        return;
+                        return (false, isNullCompensation ? "isNullCompensation" : "isDuplicateOrPend");
                     }
                     await busiCall.Invoke();
                     scope.Complete();
@@ -174,7 +175,7 @@
                     throw;
                 }
             }
-
+            return (true, string.Empty);
         }
 
         public async Task<string> QueryPrepared(DbConnection db)

src/DtmCommon/Constant.cs

@@ -67,8 +67,9 @@ public class Barrier
 
             public static readonly Dictionary<string, string> OpDict = new Dictionary<string, string>()
             {
-                { "cancel", "try" },
-                { "compensate", "action" },
+                { "cancel", "try" }, // tcc
+                { "compensate", "action" }, // saga
+                { "rollback", "action" }, // workflow
             };
             public static readonly string REDIS_LUA_CheckAdjustAmount = @" -- RedisCheckAdjustAmount
 local v = redis.call('GET', KEYS[1])

src/Dtmcli/Constant.cs

@@ -4,7 +4,8 @@ internal static class Constant
     {
         internal const string DtmClientHttpName = "dtmClient";
         internal const string BranchClientHttpName = "branchClient";
-
+        internal const string WorkflowBranchClientHttpName = "WF";
+
         internal static class Request
         {
             internal const string CONTENT_TYPE = "application/json";

src/Dtmcli/DtmClient.cs

@@ -148,7 +148,7 @@ public async Task<TransGlobal> Query(string gid, CancellationToken cancellationT
             var client = _httpClientFactory.CreateClient(Constant.DtmClientHttpName);
             var response = await client.GetAsync(url, cancellationToken).ConfigureAwait(false);
             var dtmContent = await response.Content.ReadAsStringAsync().ConfigureAwait(false);
-            DtmImp.Utils.CheckStatus(response.StatusCode, dtmContent);
+            DtmImp.Utils.CheckStatusCode(response.StatusCode);
             return JsonSerializer.Deserialize<TransGlobal>(dtmContent, _jsonOptions);
         }
 
@@ -167,7 +167,7 @@ public async Task<string> QueryStatus(string gid, CancellationToken cancellation
             var client = _httpClientFactory.CreateClient(Constant.DtmClientHttpName);
             var response = await client.GetAsync(url, cancellationToken).ConfigureAwait(false);
             var dtmContent = await response.Content.ReadAsStringAsync().ConfigureAwait(false);
-            DtmImp.Utils.CheckStatus(response.StatusCode, dtmContent);
+            DtmImp.Utils.CheckStatusCode(response.StatusCode);
             var graph = JsonSerializer.Deserialize<TransGlobalForStatus>(dtmContent, _jsonOptions);
             return graph.Transaction == null
                 ? string.Empty

src/Dtmcli/DtmImp/Utils.cs

@@ -40,6 +40,14 @@ public static void CheckStatus(HttpStatusCode status, string dtmResult)
             }
         }
 
+        public static void CheckStatusCode(HttpStatusCode status)
+        {
+            if (status != HttpStatusCode.OK)
+            {
+                throw new DtmException(string.Format(CheckStatusMsgFormat, status.ToString(), string.Empty));
+            }
+        }
+
         /// <summary>
         /// OrString return the first not null or not empty string
         /// </summary>

src/Dtmcli/TransGlobal.cs

@@ -28,7 +28,7 @@ public class DtmTransaction
     {
         [JsonPropertyName("id")] public int Id { get; set; }
 
-        [JsonPropertyName("create_time")] public DateTimeOffset CreateTime { get; set; }
+        [JsonPropertyName("create_time")] public DateTimeOffset? CreateTime { get; set; }
 
         [JsonPropertyName("update_time")] public DateTimeOffset UpdateTime { get; set; }
 
@@ -64,9 +64,9 @@ public class DtmBranch
     {
         [JsonPropertyName("id")] public int Id { get; set; }
 
-        [JsonPropertyName("create_time")] public DateTimeOffset CreateTime { get; set; }
+        [JsonPropertyName("create_time")] public DateTimeOffset? CreateTime { get; set; }
 
-        [JsonPropertyName("update_time")] public DateTimeOffset UpdateTime { get; set; }
+        [JsonPropertyName("update_time")] public DateTimeOffset? UpdateTime { get; set; }
 
         [JsonPropertyName("gid")] public string Gid { get; set; }
 

src/Dtmworkflow/ServiceCollectionExtensions.cs

@@ -23,6 +23,8 @@ public static IServiceCollection AddDtmWorkflow(this IServiceCollection services
             services.TryAddSingleton<IWorkflowFactory, WorkflowFactory>();
             services.TryAddSingleton<WorkflowGlobalTransaction>();
 
+            // AddHttpClient(services);
+
             return services;
         }
 
@@ -33,8 +35,22 @@ public static IServiceCollection AddDtmWorkflow(this IServiceCollection services
 
             services.TryAddSingleton<IWorkflowFactory, WorkflowFactory>();
             services.TryAddSingleton<WorkflowGlobalTransaction>();
+
+            // AddHttpClient(services);
 
             return services;
         }
+
+        // private static void AddHttpClient(IServiceCollection services /*, DtmOptions options*/)
+        // {
+        //     services.AddHttpClient(Dtmcli.Constant.WorkflowBranchClientHttpName, client =>
+        //     {
+        //         // TODO DtmOptions
+        //         // client.Timeout = TimeSpan.FromMilliseconds(options.BranchTimeout);                
+        //     }).AddHttpMessageHandler<WorkflowHttpInterceptor>();
+        //
+        //     // TODO how to inject workflow instance?
+        //     services.AddTransient<WorkflowHttpInterceptor>();
+        // }
     }
-}
+}

src/Dtmworkflow/Workflow.Imp.cs

@@ -64,7 +64,7 @@
             }
 
             err = Utils.GrpcError2DtmError(err);
-
+            
             if (err != null && err is not DtmCommon.DtmFailureException) throw err;
 
             try
@@ -164,23 +164,23 @@
             };
         }
 
-        private Exception StepResultToGrpc(StepResult r, IMessage reply)
+        internal Exception StepResultToGrpc(StepResult r, IMessage reply)
         {
             if (r.Error == null && r.Status == DtmCommon.Constant.StatusSucceed)
             {
-                // Check 
-
+                // TODO Check 
+                // dtmgimp.MustProtoUnmarshal(s.Data, reply.(protoreflect.ProtoMessage));
             }
 
             return r.Error;
         }
 
-        private StepResult StepResultFromGrpc(IMessage reply, Exception err)
+        internal StepResult StepResultFromGrpc(IMessage reply, Exception err)
         {
             var sr = new StepResult
             {
-                // GRPCError2DtmError
-                Error = null,
+                // TODO GRPCError2DtmError
+                Error = Utils.GrpcError2DtmError(err),
             };
 
             sr.Status = WfErrorToStatus(sr.Error);
@@ -196,7 +196,7 @@
             return sr;
         }
 
-        private HttpResponseMessage StepResultToHttp(StepResult r)
+        internal HttpResponseMessage StepResultToHttp(StepResult r)
         {
             if (r.Error != null)
             {
@@ -206,7 +206,7 @@
             return Utils.NewJSONResponse(HttpStatusCode.OK, r.Data);
         }
 
-        private StepResult StepResultFromHTTP(HttpResponseMessage resp, Exception err)
+        internal StepResult StepResultFromHTTP(HttpResponseMessage resp, Exception err)
         {
             var sr = new StepResult
             {
@@ -215,7 +215,7 @@
 
             if (err == null)
             {
-                // HTTPResp2DtmError
+                (sr.Data, sr.Error) = Utils.HTTPResp2DtmError(resp); // TODO go used this.Options.HTTPResp2DtmError(resp), for custom
                 sr.Status = WfErrorToStatus(sr.Error);
             }
 
@@ -237,9 +237,9 @@
         }
 
 
-        private async Task<StepResult> RecordedDo(Func<DtmCommon.BranchBarrier, Task<StepResult>> fn)
+        internal async Task<StepResult> RecordedDo(Func<DtmCommon.BranchBarrier, Task<StepResult>> fn)
         {
-            var sr = await this.RecordedDoInner(fn);
+            StepResult sr = await this.RecordedDoInner(fn);
 
             // do not compensate the failed branch if !CompensateErrorBranch
             if (this.Options.CompensateErrorBranch && sr.Status == DtmCommon.Constant.StatusFailed)

src/Dtmworkflow/Workflow.cs

@@ -3,7 +3,9 @@
 using Microsoft.Extensions.Logging;
 using System;
 using System.Collections.Generic;
+using System.Net.Http;
 using System.Threading.Tasks;
+using Microsoft.Extensions.DependencyInjection;
 
 namespace Dtmworkflow
 {
@@ -32,7 +34,8 @@ public Workflow(IDtmClient httpClient, IDtmgRPCClient grpcClient, Dtmcli.IBranch
 
         public System.Net.Http.HttpClient NewRequest()
         {
-            return _httpClient.GetHttpClient("WF");
+            // return _httpClient.GetHttpClient("WF");
+            return new HttpClient(new WorkflowHttpInterceptor(this));
         }
 
         /// <summary>

src/Dtmworkflow/WorkflowGlobalTransaction.cs

@@ -52,15 +52,13 @@ public void Register(string name, WfFunc2 handler, params Action<Workflow>[] cus
             });
         }
 
-#if NET5_0_OR_GREATER
         public async Task ExecuteByQS(Microsoft.AspNetCore.Http.IQueryCollection query, byte[] body)
         {
             _ = query.TryGetValue("gid", out var gid);
             _ = query.TryGetValue("op", out var op);
 
             await Execute(op, gid, body, true);
         }
-#endif
 
         #if DEBUG // for sample only
         public bool Exists(string name)