Allow unaltered payload response from codec (temporalio#425)

Author: travismcchesney

src/Temporalio/Worker/WorkflowCodecHelper.cs

@@ -275,11 +275,16 @@ private static async Task EncodeAsync(IPayloadCodec codec, Payload payload)
         {
             // We are gonna require a single result here. It is important that we do Single() call
             // before clearing out payload to merge with since underlying enumerable may be lazy.
+            // If the returned payload is literally the same object as the one sent to the codec,
+            // we leave it alone.
             var encodedList = await codec.EncodeAsync(new Payload[] { payload }).ConfigureAwait(false);
             var encoded = encodedList.Single();
-            payload.Metadata.Clear();
-            payload.Data = ByteString.Empty;
-            payload.MergeFrom(encoded);
+            if (!ReferenceEquals(encoded, payload))
+            {
+                payload.Metadata.Clear();
+                payload.Data = ByteString.Empty;
+                payload.MergeFrom(encoded);
+            }
         }
 
         private static async Task DecodeAsync(IPayloadCodec codec, ActivityResolution res)
@@ -377,11 +382,16 @@ private static async Task DecodeAsync(IPayloadCodec codec, RepeatedField<Payload
 
         private static async Task DecodeAsync(IPayloadCodec codec, Payload payload)
         {
-            // We are gonna require a single result here
+            // We are gonna require a single result here.
+            // Similarly with encode, we leave the payload alone if it's exactly the same object as the original.
             var decoded = await codec.DecodeAsync(new Payload[] { payload }).ConfigureAwait(false);
-            payload.Metadata.Clear();
-            payload.Data = ByteString.Empty;
-            payload.MergeFrom(decoded.Single());
+            var decodedPayload = decoded.Single();
+            if (!ReferenceEquals(decodedPayload, payload))
+            {
+                payload.Metadata.Clear();
+                payload.Data = ByteString.Empty;
+                payload.MergeFrom(decodedPayload);
+            }
         }
     }
 }

tests/Temporalio.Tests/Worker/WorkflowCodecHelperTests.cs

@@ -40,7 +40,7 @@ await CreateAndVisitPayload(
     public async Task EncodeAsync_AllPayloads_EncodesAll()
     {
         var comp = new WorkflowActivationCompletion();
-        var codec = new MarkerPayloadCodec();
+        var codecs = new List<IPayloadCodec> { new MarkerPayloadCodec(), new MarkerNoClonePayloadCodec() };
         await CreateAndVisitPayload(new(), comp, async (ctx, payload) =>
         {
             // We don't check search attributes on purpose
@@ -49,10 +49,13 @@ await CreateAndVisitPayload(new(), comp, async (ctx, payload) =>
                 return;
             }
             Assert.DoesNotContain("encoded", payload().Metadata.Keys);
-            await WorkflowCodecHelper.EncodeAsync(codec, comp);
-            if (!payload().Metadata.ContainsKey("encoded"))
+            foreach (var codec in codecs)
             {
-                Assert.Fail($"Payload at path {ctx.Path} not encoded");
+                await WorkflowCodecHelper.EncodeAsync(codec, comp);
+                if (!payload().Metadata.ContainsKey("encoded"))
+                {
+                    Assert.Fail($"Payload at path {ctx.Path} not encoded with codec {codec}");
+                }
             }
         });
     }
@@ -61,7 +64,7 @@ await CreateAndVisitPayload(new(), comp, async (ctx, payload) =>
     public async Task DecodeAsync_AllPayloads_DecodesAll()
     {
         var act = new WorkflowActivation();
-        var codec = new MarkerPayloadCodec();
+        var codecs = new List<IPayloadCodec> { new MarkerPayloadCodec(), new MarkerNoClonePayloadCodec() };
         await CreateAndVisitPayload(new(), act, async (ctx, payload) =>
         {
             // We don't check search attributes on purpose
@@ -70,10 +73,13 @@ await CreateAndVisitPayload(new(), act, async (ctx, payload) =>
                 return;
             }
             Assert.DoesNotContain("decoded", payload().Metadata.Keys);
-            await WorkflowCodecHelper.DecodeAsync(codec, act);
-            if (!payload().Metadata.ContainsKey("decoded"))
+            foreach (var codec in codecs)
             {
-                Assert.Fail($"Payload at path {ctx.Path} not decoded");
+                await WorkflowCodecHelper.DecodeAsync(codec, act);
+                if (!payload().Metadata.ContainsKey("decoded"))
+                {
+                    Assert.Fail($"Payload at path {ctx.Path} not decoded with codec {codec}");
+                }
             }
         });
     }
@@ -189,4 +195,23 @@ public Task<IReadOnlyCollection<Payload>> DecodeAsync(IReadOnlyCollection<Payloa
                 return newP;
             }).ToList());
     }
+
+    private class MarkerNoClonePayloadCodec : IPayloadCodec
+    {
+        public Task<IReadOnlyCollection<Payload>> EncodeAsync(IReadOnlyCollection<Payload> payloads) =>
+            Task.FromResult<IReadOnlyCollection<Payload>>(payloads.Select(p =>
+            {
+                var newP = p;
+                newP.Metadata["encoded"] = ByteString.Empty;
+                return newP;
+            }).ToList());
+
+        public Task<IReadOnlyCollection<Payload>> DecodeAsync(IReadOnlyCollection<Payload> payloads) =>
+            Task.FromResult<IReadOnlyCollection<Payload>>(payloads.Select(p =>
+            {
+                var newP = p;
+                newP.Metadata["decoded"] = ByteString.Empty;
+                return newP;
+            }).ToList());
+    }
 }