Versioning support for activity tasks (#14)

microsoft · Apr 6, 2021 · 6257830 · 6257830
1 parent def9b32
commit 6257830
Show file tree

Hide file tree

Showing 7 changed files with 45 additions and 9 deletions.
diff --git a/src/DurableTask.SqlServer/DTUtils.cs b/src/DurableTask.SqlServer/DTUtils.cs
@@ -147,6 +147,7 @@ public static bool TryGetPayloadText(HistoryEvent e, out string? payloadText)
             return historyEvent.EventType switch
             {
                 EventType.ExecutionStarted => ((ExecutionStartedEvent)historyEvent).Version,
+                EventType.TaskScheduled => ((TaskScheduledEvent)historyEvent).Version,
                 _ => null,
             };
         }

diff --git a/src/DurableTask.SqlServer/Scripts/logic.sql b/src/DurableTask.SqlServer/Scripts/logic.sql
@@ -813,7 +813,8 @@ BEGIN
         [VisibleTime],
         [LockedBy],
         [LockExpiration],
-        [PayloadID]
+        [PayloadID],
+        [Version]
     )
     OUTPUT
         INSERTED.[SequenceNumber],
@@ -827,7 +828,8 @@ BEGIN
         [VisibleTime],
         [LockedBy],
         [LockExpiration],
-        [PayloadID]
+        [PayloadID],
+        [Version]
     FROM @NewTaskEvents
 
     COMMIT TRANSACTION
@@ -999,6 +1001,7 @@ BEGIN
         [VisibleTime],
         [Timestamp],
         [DequeueCount],
+        [Version],
         (SELECT TOP 1 [Text] FROM Payloads P WHERE
             P.[TaskHub] = @TaskHub AND
             P.[InstanceID] = N.[InstanceID] AND

diff --git a/src/DurableTask.SqlServer/Scripts/schema-0.2.0.sql b/src/DurableTask.SqlServer/Scripts/schema-0.2.0.sql
@@ -73,7 +73,8 @@ IF TYPE_ID(N'dt.TaskEvents') IS NULL
         [LockExpiration] datetime2 NULL,
         [Reason] varchar(max) NULL,
         [PayloadText] varchar(max) NULL,
-        [PayloadID] uniqueidentifier NULL
+        [PayloadID] uniqueidentifier NULL,
+        [Version] varchar(100) NULL
     )
 GO
 
@@ -193,6 +194,7 @@ IF OBJECT_ID(N'dt.NewTasks', 'U') IS NULL
         [LockedBy] varchar(100) NULL,
         [LockExpiration] datetime2 NULL,
         [PayloadID] uniqueidentifier NULL,
+        [Version] varchar(100) NULL,
 
         CONSTRAINT PK_NewTasks PRIMARY KEY (TaskHub, SequenceNumber),
         CONSTRAINT FK_NewTasks_Instances FOREIGN KEY (TaskHub, InstanceID) REFERENCES dt.Instances(TaskHub, InstanceID) ON DELETE CASCADE,

diff --git a/src/DurableTask.SqlServer/SqlTypes/TaskEventSqlType.cs b/src/DurableTask.SqlServer/SqlTypes/TaskEventSqlType.cs
@@ -30,6 +30,7 @@ static class TaskEventSqlType
             new SqlMetaData("Reason", SqlDbType.VarChar, -1 /* max */),
             new SqlMetaData("PayloadText", SqlDbType.VarChar, -1 /* max */),
             new SqlMetaData("PayloadID", SqlDbType.UniqueIdentifier),
+            new SqlMetaData("Version", SqlDbType.VarChar, 100),
         };
 
         static class ColumnOrdinals
@@ -46,6 +47,7 @@ static class ColumnOrdinals
             public const int Reason = 8;
             public const int PayloadText = 9;
             public const int PayloadId = 10;
+            public const int Version = 11;
         }
 
         public static SqlParameter AddTaskEventsParameter(
@@ -119,6 +121,8 @@ static SqlDataRecord PopulateTaskMessageRecord(TaskMessage msg, SqlDataRecord re
             // Optionally, the LockedBy and LockExpiration fields can be specified
             // to pre-lock task work items for this particular node.
 
+            record.SetSqlString(ColumnOrdinals.Version, SqlUtils.GetVersion(msg.Event));
+
             return record;
         }
     }

diff --git a/src/DurableTask.SqlServer/SqlUtils.cs b/src/DurableTask.SqlServer/SqlUtils.cs
@@ -156,7 +156,7 @@ public static HistoryEvent GetHistoryEvent(this DbDataReader reader, bool isOrch
                     {
                         Input = GetPayloadText(reader),
                         Name = GetName(reader),
-                        Version = null, // TODO
+                        Version = GetVersion(reader),
                     };
                     break;
                 case EventType.TimerCreated:

diff --git a/test/DurableTask.SqlServer.Tests/Integration/Orchestrations.cs b/test/DurableTask.SqlServer.Tests/Integration/Orchestrations.cs
@@ -456,5 +456,30 @@ public async Task VersionedSubOrchestration()
                 });
             await parentInstance.WaitForCompletion(waitTimeout, expectedOutput: version1 + version2);
         }
+
+        [Fact]
+        public async Task VersionedActivity()
+        {
+            string activityName = "VersionedActivityTest";
+            string version1 = "V1";
+            string version2 = "V2";
+            var waitTimeout = TimeSpan.FromSeconds(30);
+
+            this.testService.RegisterInlineActivity(
+                activityName, version1, TestService.MakeActivity<string, string>((ctx, input) => version1));
+            this.testService.RegisterInlineActivity(
+                activityName, version2, TestService.MakeActivity<string, string>((ctx, input) => version2));
+
+            var instance = await this.testService.RunOrchestration<string, string>(
+                null,
+                "OrchestrationWithVersionedActivities",
+                implementation: async (ctx, input) =>
+                {
+                    var result1 = await ctx.ScheduleTask<string>(activityName, version1, input);
+                    var result2 = await ctx.ScheduleTask<string>(activityName, version2, input);
+                    return result1 + result2;
+                });
+            await instance.WaitForCompletion(waitTimeout, expectedOutput: version1 + version2);
+        }
     }
 }
diff --git a/test/DurableTask.SqlServer.Tests/Utils/TestService.cs b/test/DurableTask.SqlServer.Tests/Utils/TestService.cs
@@ -130,7 +130,7 @@ public async Task<List<TestInstance<TInput>>> RunOrchestrations<TOutput, TInput>
 
             foreach ((string name, TaskActivity activity) in activities)
             {
-                this.worker.AddTaskActivities(new TestObjectCreator<TaskActivity>(name, activity));
+                RegisterInlineActivity(name, string.Empty, activity);
             }
 
             DateTime utcNow = DateTime.UtcNow;
@@ -154,6 +154,11 @@ public async Task<List<TestInstance<TInput>>> RunOrchestrations<TOutput, TInput>
             return instances;
         }
 
+        public void RegisterInlineActivity(string name, string version, TaskActivity activity)
+        {
+            this.worker.AddTaskActivities(new TestObjectCreator<TaskActivity>(name, version, activity));
+        }
+
         public void RegisterInlineOrchestration<TOutput, TInput>(
             string orchestrationName,
             string version,
@@ -364,10 +369,6 @@ class TestObjectCreator<T> : ObjectCreator<T>
         {
             readonly T obj;
 
-            public TestObjectCreator(string name, T obj) : this(name, string.Empty, obj)
-            {
-            }
-
             public TestObjectCreator(string name, string version, T obj)
             {
                 this.Name = name;