From 8ec7ed1a33372d840e6aa700083f1a617a832e80 Mon Sep 17 00:00:00 2001 From: Sophia Tevosyan Date: Fri, 11 Jul 2025 12:50:48 -0700 Subject: [PATCH 01/10] first commit --- src/Worker/Grpc/GrpcEntityRunner.cs | 61 ++++++++++++++++++++++++----- 1 file changed, 51 insertions(+), 10 deletions(-) diff --git a/src/Worker/Grpc/GrpcEntityRunner.cs b/src/Worker/Grpc/GrpcEntityRunner.cs index f35366df..5db05323 100644 --- a/src/Worker/Grpc/GrpcEntityRunner.cs +++ b/src/Worker/Grpc/GrpcEntityRunner.cs @@ -3,9 +3,11 @@ using DurableTask.Core.Entities; using DurableTask.Core.Entities.OperationFormat; +using DurableTask.Core.Exceptions; using Google.Protobuf; using Microsoft.DurableTask.Entities; -using Microsoft.DurableTask.Worker.Shims; +using Microsoft.DurableTask.Worker.Shims; +using Microsoft.Extensions.Caching.Memory; using Microsoft.Extensions.DependencyInjection; using P = Microsoft.DurableTask.Protobuf; @@ -25,7 +27,9 @@ namespace Microsoft.DurableTask.Worker.Grpc; /// /// public static class GrpcEntityRunner -{ +{ + const int DefaultExtendedSessionIdleTimeoutInSeconds = 30; + /// /// Deserializes entity batch request from and uses it to invoke the /// requested operations implemented by . @@ -50,24 +54,61 @@ public static class GrpcEntityRunner /// Thrown if contains invalid data. /// public static async Task LoadAndRunAsync( - string encodedEntityRequest, ITaskEntity implementation, IServiceProvider? services = null) + string encodedEntityRequest, ITaskEntity implementation, IMemoryCache entityStates, IServiceProvider? services = null) { Check.NotNullOrEmpty(encodedEntityRequest); Check.NotNull(implementation); P.EntityBatchRequest request = P.EntityBatchRequest.Parser.Base64Decode( - encodedEntityRequest); + encodedEntityRequest); + Dictionary properties = request.Properties.ToDictionary( + pair => pair.Key, + pair => ProtoUtils.ConvertValueToObject(pair.Value)); EntityBatchRequest batch = request.ToEntityBatchRequest(); EntityId id = EntityId.FromString(batch.InstanceId!); - TaskName entityName = new(id.Name); - + TaskName entityName = new(id.Name); + + string? entityState = null; + bool entityStateFound = false; + bool addToEntityStates = false; + + if (properties.TryGetValue("ExtendedSession", out object? isExtendedSession) && (bool)isExtendedSession!) + { + addToEntityStates = true; + if (entityStates.TryGetValue(request.InstanceId, out entityState)) + { + entityStateFound = true; + batch.EntityState = entityState; + } + } + + int extendedSessionIdleTimeoutInSeconds = DefaultExtendedSessionIdleTimeoutInSeconds; + if (properties.TryGetValue("ExtendedSessionIdleTimeoutInSeconds", out object? extendedSessionIdleTimeout) + && (int)extendedSessionIdleTimeout! >= 0) + { + extendedSessionIdleTimeoutInSeconds = (int)extendedSessionIdleTimeout!; + } + + if (!entityStateFound && (properties.TryGetValue("IncludeEntityState", out object? entityStateIncluded) && (bool)entityStateIncluded!)) + { + throw new SessionAbortedException($"The worker has since ended the extended session due to its being idle longer than the maximum of {extendedSessionIdleTimeoutInSeconds} seconds."); + } + DurableTaskShimFactory factory = services is null ? DurableTaskShimFactory.Default - : ActivatorUtilities.GetServiceOrCreateInstance(services); - - TaskEntity entity = factory.CreateEntity(entityName, implementation, id); - EntityBatchResult result = await entity.ExecuteOperationBatchAsync(batch); + : ActivatorUtilities.GetServiceOrCreateInstance(services); + + TaskEntity entity = factory.CreateEntity(entityName, implementation, id); + EntityBatchResult result = await entity.ExecuteOperationBatchAsync(batch); + + if (addToEntityStates) + { + entityStates.Set( + request.InstanceId, + result.EntityState, + new MemoryCacheEntryOptions { SlidingExpiration = TimeSpan.FromSeconds(extendedSessionIdleTimeoutInSeconds) }); + } P.EntityBatchResult response = result.ToEntityBatchResult(); byte[] responseBytes = response.ToByteArray(); From 7b84f2ed7319d39ee33b80055e7108b35a7ed1af Mon Sep 17 00:00:00 2001 From: Sophia Tevosyan Date: Thu, 13 Nov 2025 15:08:24 -0800 Subject: [PATCH 02/10] updated implementation to match the orchestration implementation and added tests --- src/Worker/Grpc/GrpcEntityRunner.cs | 98 +++-- .../Grpc.Tests/GrpcEntityRunnerTests.cs | 380 ++++++++++++++++++ .../GrpcOrchestrationRunnerTests.cs | 7 +- 3 files changed, 458 insertions(+), 27 deletions(-) create mode 100644 test/Worker/Grpc.Tests/GrpcEntityRunnerTests.cs diff --git a/src/Worker/Grpc/GrpcEntityRunner.cs b/src/Worker/Grpc/GrpcEntityRunner.cs index 5db05323..5124b129 100644 --- a/src/Worker/Grpc/GrpcEntityRunner.cs +++ b/src/Worker/Grpc/GrpcEntityRunner.cs @@ -3,7 +3,6 @@ using DurableTask.Core.Entities; using DurableTask.Core.Entities.OperationFormat; -using DurableTask.Core.Exceptions; using Google.Protobuf; using Microsoft.DurableTask.Entities; using Microsoft.DurableTask.Worker.Shims; @@ -28,8 +27,6 @@ namespace Microsoft.DurableTask.Worker.Grpc; /// public static class GrpcEntityRunner { - const int DefaultExtendedSessionIdleTimeoutInSeconds = 30; - /// /// Deserializes entity batch request from and uses it to invoke the /// requested operations implemented by . @@ -54,7 +51,39 @@ public static class GrpcEntityRunner /// Thrown if contains invalid data. /// public static async Task LoadAndRunAsync( - string encodedEntityRequest, ITaskEntity implementation, IMemoryCache entityStates, IServiceProvider? services = null) + string encodedEntityRequest, ITaskEntity implementation, IServiceProvider? services = null) + { + return await LoadAndRunAsync(encodedEntityRequest, implementation, extendedSessionsCache: null, services: services); + } + + /// + /// Deserializes entity batch request from and uses it to invoke the + /// requested operations implemented by . + /// + /// + /// The encoded protobuf payload representing an entity batch request. This is a base64-encoded string. + /// + /// + /// An implementation that defines the entity logic. + /// + /// + /// The cache of entity states which can be used to retrieve the entity state if this request is from within an extended session. + /// + /// + /// Optional from which injected dependencies can be retrieved. + /// + /// + /// Returns a serialized result of the entity batch that should be used as the return value of the entity function + /// trigger. + /// + /// + /// Thrown if or is null. + /// + /// + /// Thrown if contains invalid data. + /// + public static async Task LoadAndRunAsync( + string encodedEntityRequest, ITaskEntity implementation, ExtendedSessionsCache? extendedSessionsCache, IServiceProvider? services = null) { Check.NotNullOrEmpty(encodedEntityRequest); Check.NotNull(implementation); @@ -69,30 +98,49 @@ public static async Task LoadAndRunAsync( EntityId id = EntityId.FromString(batch.InstanceId!); TaskName entityName = new(id.Name); - string? entityState = null; - bool entityStateFound = false; - bool addToEntityStates = false; + MemoryCache? extendedSessions = null; - if (properties.TryGetValue("ExtendedSession", out object? isExtendedSession) && (bool)isExtendedSession!) + // If any of the request parameters are malformed, we assume the default - extended sessions are not enabled and the orchestration history is attached + bool addToExtendedSessions = false; + bool entityStateIncluded = true; + bool entityStateCached = false; + bool isExtendedSession = false; + double extendedSessionIdleTimeoutInSeconds = 0; + + // Only attempt to initialize the extended sessions cache if all the parameters are correctly specified + if (properties.TryGetValue("ExtendedSessionIdleTimeoutInSeconds", out object? extendedSessionIdleTimeoutObj) + && extendedSessionIdleTimeoutObj is double extendedSessionIdleTimeout + && extendedSessionIdleTimeout > 0 + && properties.TryGetValue("IsExtendedSession", out object? extendedSessionObj) + && extendedSessionObj is bool extendedSession) { - addToEntityStates = true; - if (entityStates.TryGetValue(request.InstanceId, out entityState)) - { - entityStateFound = true; - batch.EntityState = entityState; - } + extendedSessionIdleTimeoutInSeconds = extendedSessionIdleTimeout; + isExtendedSession = extendedSession; + extendedSessions = extendedSessionsCache?.GetOrInitializeCache(extendedSessionIdleTimeoutInSeconds); + } + + if (properties.TryGetValue("IncludeEntityState", out object? includeEntityStateObj) + && includeEntityStateObj is bool includeEntityState) + { + entityStateIncluded = includeEntityState; } - int extendedSessionIdleTimeoutInSeconds = DefaultExtendedSessionIdleTimeoutInSeconds; - if (properties.TryGetValue("ExtendedSessionIdleTimeoutInSeconds", out object? extendedSessionIdleTimeout) - && (int)extendedSessionIdleTimeout! >= 0) + if (isExtendedSession && extendedSessions != null) { - extendedSessionIdleTimeoutInSeconds = (int)extendedSessionIdleTimeout!; + addToExtendedSessions = true; + + // If an entity state was provided, even if we already have one stored, we always want to use the provided state. + if (!entityStateIncluded && extendedSessions.TryGetValue(request.InstanceId, out string? entityState)) + { + batch.EntityState = entityState; + entityStateCached = true; + } } - if (!entityStateFound && (properties.TryGetValue("IncludeEntityState", out object? entityStateIncluded) && (bool)entityStateIncluded!)) + if (!entityStateCached && !entityStateIncluded) { - throw new SessionAbortedException($"The worker has since ended the extended session due to its being idle longer than the maximum of {extendedSessionIdleTimeoutInSeconds} seconds."); + // No state was provided, and we do not have one cached, so we cannot execute the batch request. + return Convert.ToBase64String(new P.EntityBatchResult { RequiresState = true }.ToByteArray()); } DurableTaskShimFactory factory = services is null @@ -102,14 +150,18 @@ public static async Task LoadAndRunAsync( TaskEntity entity = factory.CreateEntity(entityName, implementation, id); EntityBatchResult result = await entity.ExecuteOperationBatchAsync(batch); - if (addToEntityStates) + if (addToExtendedSessions) { - entityStates.Set( + extendedSessions.Set( request.InstanceId, result.EntityState, new MemoryCacheEntryOptions { SlidingExpiration = TimeSpan.FromSeconds(extendedSessionIdleTimeoutInSeconds) }); } - + else + { + extendedSessions?.Remove(request.InstanceId); + } + P.EntityBatchResult response = result.ToEntityBatchResult(); byte[] responseBytes = response.ToByteArray(); return Convert.ToBase64String(responseBytes); diff --git a/test/Worker/Grpc.Tests/GrpcEntityRunnerTests.cs b/test/Worker/Grpc.Tests/GrpcEntityRunnerTests.cs new file mode 100644 index 00000000..dd20d87c --- /dev/null +++ b/test/Worker/Grpc.Tests/GrpcEntityRunnerTests.cs @@ -0,0 +1,380 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +using Google.Protobuf; +using Google.Protobuf.Collections; +using Google.Protobuf.WellKnownTypes; +using Microsoft.DurableTask.Entities; + +namespace Microsoft.DurableTask.Worker.Grpc.Tests; + +public class GrpcEntityRunnerTests +{ + const string TestInstanceId = "@instance_id@my_entity"; + const int DefaultExtendedSessionIdleTimeoutInSeconds = 30; + + [Fact] + public async Task EmptyOrNullParameters_Throw_Exceptions() + { + Func act = async () => + await GrpcEntityRunner.LoadAndRunAsync(string.Empty, new SimpleEntity(), new ExtendedSessionsCache()); + await act.Should().ThrowExactlyAsync().WithParameterName("encodedEntityRequest"); + + act = () => + GrpcEntityRunner.LoadAndRunAsync(null!, new SimpleEntity(), new ExtendedSessionsCache()); + await act.Should().ThrowExactlyAsync().WithParameterName("encodedEntityRequest"); + + act = () => + GrpcEntityRunner.LoadAndRunAsync("request", null!, new ExtendedSessionsCache()); + await act.Should().ThrowExactlyAsync().WithParameterName("implementation"); + } + + [Fact] + public async Task EmptyState_Returns_NeedsStateInResponse() + { + using var extendedSessions = new ExtendedSessionsCache(); + + // No state and without extended sessions enabled + Protobuf.EntityBatchRequest entityRequest = CreateEntityRequest([]); + entityRequest.Properties.Add(new MapField() { + { "IncludeEntityState", Value.ForBool(false) }}); + byte[] requestBytes = entityRequest.ToByteArray(); + string requestString = Convert.ToBase64String(requestBytes); + string stringResponse = await GrpcEntityRunner.LoadAndRunAsync(requestString, new SimpleEntity(), extendedSessions); + Protobuf.EntityBatchResult response = Protobuf.EntityBatchResult.Parser.ParseFrom(Convert.FromBase64String(stringResponse)); + Assert.True(response.RequiresState); + Assert.False(extendedSessions.IsInitialized); + + // No state but with extended sessions enabled + entityRequest.Properties.Add(new MapField() { + { "IsExtendedSession", Value.ForBool(true) }, + { "ExtendedSessionIdleTimeoutInSeconds", Value.ForNumber(DefaultExtendedSessionIdleTimeoutInSeconds) } }); + requestBytes = entityRequest.ToByteArray(); + requestString = Convert.ToBase64String(requestBytes); + stringResponse = await GrpcEntityRunner.LoadAndRunAsync(requestString, new SimpleEntity(), extendedSessions); + response = Protobuf.EntityBatchResult.Parser.ParseFrom(Convert.FromBase64String(stringResponse)); + Assert.True(response.RequiresState); + Assert.True(extendedSessions.IsInitialized); + } + + [Fact] + public async Task MalformedRequestParameters_Means_CacheNotInitialized() + { + using var extendedSessions = new ExtendedSessionsCache(); + Protobuf.EntityBatchRequest entityRequest = CreateEntityRequest([]); + + // Misspelled extended session timeout key + entityRequest.Properties.Add(new MapField() { + { "IncludeEntityState", Value.ForBool(false) }, + { "IsExtendedSession", Value.ForBool(true) }, + { "ExtendedSessionsIdleTimeoutInSeconds", Value.ForNumber(DefaultExtendedSessionIdleTimeoutInSeconds) } }); + byte[] requestBytes = entityRequest.ToByteArray(); + string requestString = Convert.ToBase64String(requestBytes); + string stringResponse = await GrpcEntityRunner.LoadAndRunAsync(requestString, new SimpleEntity(), extendedSessions); + Protobuf.OrchestratorResponse response = Protobuf.OrchestratorResponse.Parser.ParseFrom(Convert.FromBase64String(stringResponse)); + Assert.False(extendedSessions.IsInitialized); + + // Wrong value type for extended session timeout key + entityRequest.Properties.Clear(); + entityRequest.Properties.Add(new MapField() { + { "IncludeEntityState", Value.ForBool(false) }, + { "IsExtendedSession", Value.ForBool(true) }, + { "ExtendedSessionIdleTimeoutInSeconds", Value.ForString("hi") } }); + requestBytes = entityRequest.ToByteArray(); + requestString = Convert.ToBase64String(requestBytes); + await GrpcEntityRunner.LoadAndRunAsync(requestString, new SimpleEntity(), extendedSessions); + Assert.False(extendedSessions.IsInitialized); + + // Invalid number for extended session timeout key (must be > 0) + entityRequest.Properties.Clear(); + entityRequest.Properties.Add(new MapField() { + { "IncludeEntityState", Value.ForBool(false) }, + { "IsExtendedSession", Value.ForBool(true) }, + { "ExtendedSessionIdleTimeoutInSeconds", Value.ForNumber(0) } }); + requestBytes = entityRequest.ToByteArray(); + requestString = Convert.ToBase64String(requestBytes); + await GrpcEntityRunner.LoadAndRunAsync(requestString, new SimpleEntity(), extendedSessions); + Assert.False(extendedSessions.IsInitialized); + + // Invalid number for extended session timeout key (must be > 0) + entityRequest.Properties.Clear(); + entityRequest.Properties.Add(new MapField() { + { "IncludeEntityState", Value.ForBool(false) }, + { "IsExtendedSession", Value.ForBool(true) }, + { "ExtendedSessionIdleTimeoutInSeconds", Value.ForNumber(-1) } }); + requestBytes = entityRequest.ToByteArray(); + requestString = Convert.ToBase64String(requestBytes); + await GrpcEntityRunner.LoadAndRunAsync(requestString, new SimpleEntity(), extendedSessions); + Assert.False(extendedSessions.IsInitialized); + + // No extended session timeout key + entityRequest.Properties.Clear(); + entityRequest.Properties.Add(new MapField() { + { "IncludeEntityState", Value.ForBool(false) }, + { "IsExtendedSession", Value.ForBool(true) } }); + requestBytes = entityRequest.ToByteArray(); + requestString = Convert.ToBase64String(requestBytes); + await GrpcEntityRunner.LoadAndRunAsync(requestString, new SimpleEntity(), extendedSessions); + Assert.False(extendedSessions.IsInitialized); + + // Misspelled extended session key + entityRequest.Properties.Clear(); + entityRequest.Properties.Add(new MapField() { + { "IncludeEntityState", Value.ForBool(false) }, + { "isExtendedSession", Value.ForBool(true) }, + { "ExtendedSessionIdleTimeoutInSeconds", Value.ForNumber(DefaultExtendedSessionIdleTimeoutInSeconds) } }); + requestBytes = entityRequest.ToByteArray(); + requestString = Convert.ToBase64String(requestBytes); + await GrpcEntityRunner.LoadAndRunAsync(requestString, new SimpleEntity(), extendedSessions); + Assert.False(extendedSessions.IsInitialized); + + // Wrong value type for extended session key + entityRequest.Properties.Clear(); + entityRequest.Properties.Add(new MapField() { + { "IncludeEntityState", Value.ForBool(false) }, + { "IsExtendedSession", Value.ForNumber(1) }, + { "ExtendedSessionIdleTimeoutInSeconds", Value.ForNumber(DefaultExtendedSessionIdleTimeoutInSeconds) } }); + requestBytes = entityRequest.ToByteArray(); + requestString = Convert.ToBase64String(requestBytes); + await GrpcEntityRunner.LoadAndRunAsync(requestString, new SimpleEntity(), extendedSessions); + Assert.False(extendedSessions.IsInitialized); + + // No extended session key + entityRequest.Properties.Clear(); + entityRequest.Properties.Add(new MapField() { + { "IncludeEntityState", Value.ForBool(false) }, + { "ExtendedSessionIdleTimeoutInSeconds", Value.ForNumber(DefaultExtendedSessionIdleTimeoutInSeconds) } }); + requestBytes = entityRequest.ToByteArray(); + requestString = Convert.ToBase64String(requestBytes); + await GrpcEntityRunner.LoadAndRunAsync(requestString, new SimpleEntity(), extendedSessions); + Assert.False(extendedSessions.IsInitialized); + } + + [Fact] + public async Task IsExtendedSessionFalse_Means_NoExtendedSessionStored() + { + using var extendedSessions = new ExtendedSessionsCache(); + Protobuf.EntityBatchRequest entityRequest = CreateEntityRequest([new Protobuf.OperationRequest() { RequestId = Guid.NewGuid().ToString(), Input = 1.ToString(), Operation = "Set"}]); + + entityRequest.Properties.Add(new MapField() { + { "IncludeEntityState", Value.ForBool(false) }, + { "IsExtendedSession", Value.ForBool(false) }, + { "ExtendedSessionIdleTimeoutInSeconds", Value.ForNumber(DefaultExtendedSessionIdleTimeoutInSeconds) } }); + byte[] requestBytes = entityRequest.ToByteArray(); + string requestString = Convert.ToBase64String(requestBytes); + await GrpcEntityRunner.LoadAndRunAsync(requestString, new SimpleEntity(), extendedSessions); + Assert.True(extendedSessions.IsInitialized); + Assert.False(extendedSessions.GetOrInitializeCache(DefaultExtendedSessionIdleTimeoutInSeconds).TryGetValue(TestInstanceId, out object? extendedSession)); + } + + /// + /// These tests verify that a malformed/nonexistent "IncludeEntityState" parameter means that the worker will attempt to + /// fulfill the entity request and not request a state for it. However, it is of course very undesirable that a + /// state is not attached to the request, but no state is requested by the worker due to a malformed "IncludeEntityState" parameter + /// even when it needs one to fulfill the request. This would need to be checked on whatever side is calling this SDK. + /// + [Fact] + public async Task MalformedIncludeEntityStateParameter_Means_NoStateRequired() + { + using var extendedSessions = new ExtendedSessionsCache(); + Protobuf.EntityBatchRequest entityRequest = CreateEntityRequest( + [new Protobuf.OperationRequest() { RequestId = Guid.NewGuid().ToString(), Input = 1.ToString(), Operation = "Add" }], + entityState: 1.ToString()); + + // Misspelled include entity state key + entityRequest.Properties.Add(new MapField() { + { "IncludeENtityState", Value.ForBool(false) }, + { "IsExtendedSession", Value.ForBool(false) }, + { "ExtendedSessionIdleTimeoutInSeconds", Value.ForNumber(DefaultExtendedSessionIdleTimeoutInSeconds) } }); + byte[] requestBytes = entityRequest.ToByteArray(); + string requestString = Convert.ToBase64String(requestBytes); + string stringResponse = await GrpcEntityRunner.LoadAndRunAsync(requestString, new SimpleEntity(), extendedSessions); + Protobuf.EntityBatchResult response = Protobuf.EntityBatchResult.Parser.ParseFrom(Convert.FromBase64String(stringResponse)); + Assert.False(response.RequiresState); + Assert.Equal("2", response.EntityState); + + // Wrong value type for include entity state key + entityRequest.Properties.Clear(); + entityRequest.Properties.Add(new MapField() { + { "IncludeEntityState", Value.ForString("no") }, + { "IsExtendedSession", Value.ForBool(false) }, + { "ExtendedSessionIdleTimeoutInSeconds", Value.ForNumber(DefaultExtendedSessionIdleTimeoutInSeconds) } }); + requestBytes = entityRequest.ToByteArray(); + requestString = Convert.ToBase64String(requestBytes); + stringResponse = await GrpcEntityRunner.LoadAndRunAsync(requestString, new SimpleEntity(), extendedSessions); + response = Protobuf.EntityBatchResult.Parser.ParseFrom(Convert.FromBase64String(stringResponse)); + Assert.False(response.RequiresState); + Assert.Equal("2", response.EntityState); + + // No include entity state key + entityRequest.Properties.Clear(); + entityRequest.Properties.Add(new MapField() { + { "IsExtendedSession", Value.ForBool(false) }, + { "ExtendedSessionIdleTimeoutInSeconds", Value.ForNumber(DefaultExtendedSessionIdleTimeoutInSeconds) } }); + requestBytes = entityRequest.ToByteArray(); + requestString = Convert.ToBase64String(requestBytes); + stringResponse = await GrpcEntityRunner.LoadAndRunAsync(requestString, new SimpleEntity(), extendedSessions); + response = Protobuf.EntityBatchResult.Parser.ParseFrom(Convert.FromBase64String(stringResponse)); + Assert.False(response.RequiresState); + Assert.Equal("2", response.EntityState); + } + + [Fact] + public async Task Entity_State_Stored() + { + using var extendedSessions = new ExtendedSessionsCache(); + Protobuf.EntityBatchRequest entityRequest = CreateEntityRequest([new Protobuf.OperationRequest() { RequestId = Guid.NewGuid().ToString(), Input = 1.ToString(), Operation = "Set" }]); + entityRequest.Properties.Add(new MapField() { + { "IncludeEntityState", Value.ForBool(true) }, + { "IsExtendedSession", Value.ForBool(true) }, + { "ExtendedSessionIdleTimeoutInSeconds", Value.ForNumber(DefaultExtendedSessionIdleTimeoutInSeconds) } }); + byte[] requestBytes = entityRequest.ToByteArray(); + string requestString = Convert.ToBase64String(requestBytes); + await GrpcEntityRunner.LoadAndRunAsync(requestString, new SimpleEntity(), extendedSessions); + Assert.True(extendedSessions.IsInitialized); + Assert.True(extendedSessions.GetOrInitializeCache(DefaultExtendedSessionIdleTimeoutInSeconds).TryGetValue(TestInstanceId, out object? extendedSession)); + Assert.Equal("1", extendedSession); + } + + [Fact] + public async Task ExternallyEndedExtendedSession_Evicted() + { + using var extendedSessions = new ExtendedSessionsCache(); + Protobuf.EntityBatchRequest entityRequest = CreateEntityRequest([new Protobuf.OperationRequest() { RequestId = Guid.NewGuid().ToString(), Input = 1.ToString(), Operation = "Set" }]); + entityRequest.Properties.Add(new MapField() { + { "IncludeEntityState", Value.ForBool(true) }, + { "IsExtendedSession", Value.ForBool(true) }, + { "ExtendedSessionIdleTimeoutInSeconds", Value.ForNumber(DefaultExtendedSessionIdleTimeoutInSeconds) } }); + byte[] requestBytes = entityRequest.ToByteArray(); + string requestString = Convert.ToBase64String(requestBytes); + await GrpcEntityRunner.LoadAndRunAsync(requestString, new SimpleEntity(), extendedSessions); + Assert.True(extendedSessions.IsInitialized); + Assert.True(extendedSessions.GetOrInitializeCache(DefaultExtendedSessionIdleTimeoutInSeconds).TryGetValue(TestInstanceId, out object? extendedSession)); + Assert.Equal("1", extendedSession); + + // Now set the extended session flag to false for this instance + entityRequest.Properties.Clear(); + entityRequest.Properties.Add(new MapField() { + { "IncludeEntityState", Value.ForBool(true) }, + { "IsExtendedSession", Value.ForBool(false) }, + { "ExtendedSessionIdleTimeoutInSeconds", Value.ForNumber(DefaultExtendedSessionIdleTimeoutInSeconds) } }); + requestBytes = entityRequest.ToByteArray(); + requestString = Convert.ToBase64String(requestBytes); + await GrpcEntityRunner.LoadAndRunAsync(requestString, new SimpleEntity(), extendedSessions); + Assert.True(extendedSessions.IsInitialized); + Assert.False(extendedSessions.GetOrInitializeCache(DefaultExtendedSessionIdleTimeoutInSeconds).TryGetValue(TestInstanceId, out extendedSession)); + } + + [Fact] + public async Task Stale_ExtendedSessions_Evicted_Async() + { + using var extendedSessions = new ExtendedSessionsCache(); + int extendedSessionIdleTimeout = 5; + Protobuf.EntityBatchRequest entityRequest = CreateEntityRequest([new Protobuf.OperationRequest() { RequestId = Guid.NewGuid().ToString(), Input = 1.ToString(), Operation = "Set" }]); + entityRequest.Properties.Add(new MapField() { + { "IncludeEntityState", Value.ForBool(true) }, + { "IsExtendedSession", Value.ForBool(true) }, + { "ExtendedSessionIdleTimeoutInSeconds", Value.ForNumber(extendedSessionIdleTimeout) } }); + byte[] requestBytes = entityRequest.ToByteArray(); + string requestString = Convert.ToBase64String(requestBytes); + await GrpcEntityRunner.LoadAndRunAsync(requestString, new SimpleEntity(), extendedSessions); + Assert.True(extendedSessions.IsInitialized); + Assert.True(extendedSessions.GetOrInitializeCache(extendedSessionIdleTimeout).TryGetValue(TestInstanceId, out object? extendedSession)); + Assert.Equal("1", extendedSession); + + // Wait for longer than the timeout to account for finite cache scan for stale items frequency + await Task.Delay(extendedSessionIdleTimeout * 1000 * 2); + Assert.False(extendedSessions.GetOrInitializeCache(extendedSessionIdleTimeout).TryGetValue(TestInstanceId, out extendedSession)); + + // Now that the entry was evicted from the cache, the entity runner needs an entity state to complete the request + entityRequest.Properties.Clear(); + entityRequest.Properties.Add(new MapField() { + { "IncludeEntityState", Value.ForBool(false) }, + { "IsExtendedSession", Value.ForBool(true) }, + { "ExtendedSessionIdleTimeoutInSeconds", Value.ForNumber(extendedSessionIdleTimeout) } }); + requestBytes = entityRequest.ToByteArray(); + requestString = Convert.ToBase64String(requestBytes); + string stringResponse = await GrpcEntityRunner.LoadAndRunAsync(requestString, new SimpleEntity(), extendedSessions); + Protobuf.EntityBatchResult response = Protobuf.EntityBatchResult.Parser.ParseFrom(Convert.FromBase64String(stringResponse)); + Assert.True(response.RequiresState); + } + + [Fact] + public async Task EntityStateIncluded_Means_ExtendedSession_Evicted() + { + using var extendedSessions = new ExtendedSessionsCache(); + Protobuf.OperationRequest operationRequest = new() { RequestId = Guid.NewGuid().ToString(), Input = 1.ToString(), Operation = "Add" }; + Protobuf.EntityBatchRequest entityRequest = CreateEntityRequest([operationRequest], entityState: 1.ToString()); + entityRequest.Properties.Add(new MapField() { + { "IncludeEntityState", Value.ForBool(true) }, + { "IsExtendedSession", Value.ForBool(true) }, + { "ExtendedSessionIdleTimeoutInSeconds", Value.ForNumber(DefaultExtendedSessionIdleTimeoutInSeconds) } }); + byte[] requestBytes = entityRequest.ToByteArray(); + string requestString = Convert.ToBase64String(requestBytes); + await GrpcEntityRunner.LoadAndRunAsync(requestString, new SimpleEntity(), extendedSessions); + Assert.True(extendedSessions.IsInitialized); + Assert.True(extendedSessions.GetOrInitializeCache(DefaultExtendedSessionIdleTimeoutInSeconds).TryGetValue(TestInstanceId, out object? extendedSession)); + + // Now we will retry the same request, but with a different value for the state. If the extended session is not evicted, then the result will be incorrect. + entityRequest = CreateEntityRequest([operationRequest], entityState: 5.ToString()); + entityRequest.Properties.Add(new MapField() { + { "IncludeEntityState", Value.ForBool(true) }, + { "IsExtendedSession", Value.ForBool(true) }, + { "ExtendedSessionIdleTimeoutInSeconds", Value.ForNumber(DefaultExtendedSessionIdleTimeoutInSeconds) } }); + requestBytes = entityRequest.ToByteArray(); + requestString = Convert.ToBase64String(requestBytes); + await GrpcEntityRunner.LoadAndRunAsync(requestString, new SimpleEntity(), extendedSessions); + Assert.True(extendedSessions.GetOrInitializeCache(DefaultExtendedSessionIdleTimeoutInSeconds).TryGetValue(TestInstanceId, out extendedSession)); + Assert.Equal("6", extendedSession); + } + + [Fact] + public async Task Null_ExtendedSessionsCache_IsOkay() + { + Protobuf.EntityBatchRequest entityRequest = CreateEntityRequest([new Protobuf.OperationRequest() { RequestId = Guid.NewGuid().ToString(), Input = 1.ToString(), Operation = "Set" }]); + + // Set up the parameters as if extended sessions are enabled, but do not pass an extended session cache to the request. + // The request should still be successful. + entityRequest.Properties.Add(new MapField() { + { "IncludeEntityState", Value.ForBool(true) }, + { "IsExtendedSession", Value.ForBool(true) }, + { "ExtendedSessionIdleTimeoutInSeconds", Value.ForNumber(DefaultExtendedSessionIdleTimeoutInSeconds) } }); + byte[] requestBytes = entityRequest.ToByteArray(); + string requestString = Convert.ToBase64String(requestBytes); + string stringResponse = await GrpcEntityRunner.LoadAndRunAsync(requestString, new SimpleEntity()); + Protobuf.EntityBatchResult response = Protobuf.EntityBatchResult.Parser.ParseFrom(Convert.FromBase64String(stringResponse)); + Assert.Equal("1", response.EntityState); + + // Now try it again without any properties specified. The request should still be successful. + entityRequest.Properties.Clear(); + requestBytes = entityRequest.ToByteArray(); + requestString = Convert.ToBase64String(requestBytes); + stringResponse = await GrpcEntityRunner.LoadAndRunAsync(requestString, new SimpleEntity()); + response = Protobuf.EntityBatchResult.Parser.ParseFrom(Convert.FromBase64String(stringResponse)); + Assert.Equal("1", response.EntityState); + } + + static Protobuf.EntityBatchRequest CreateEntityRequest(IEnumerable requests, string? entityState = null) + { + var entityBatchRequest = new Protobuf.EntityBatchRequest() + { + InstanceId = TestInstanceId, + EntityState = entityState, + Operations = { requests } + }; + return entityBatchRequest; + } + + class SimpleEntity : TaskEntity + { + public async Task Set(int value) + { + this.State = value; + } + + public async Task Add(int value) + { + this.State += value; + } + } +} diff --git a/test/Worker/Grpc.Tests/GrpcOrchestrationRunnerTests.cs b/test/Worker/Grpc.Tests/GrpcOrchestrationRunnerTests.cs index 6c317921..b701dd08 100644 --- a/test/Worker/Grpc.Tests/GrpcOrchestrationRunnerTests.cs +++ b/test/Worker/Grpc.Tests/GrpcOrchestrationRunnerTests.cs @@ -376,7 +376,6 @@ public async void Stale_ExtendedSessions_Evicted_Async() public void PastEventIncluded_Means_ExtendedSession_Evicted() { using var extendedSessions = new ExtendedSessionsCache(); - int extendedSessionIdleTimeout = 5; var historyEvent = new Protobuf.HistoryEvent { EventId = -1, @@ -394,17 +393,17 @@ public void PastEventIncluded_Means_ExtendedSession_Evicted() orchestratorRequest.Properties.Add(new MapField() { { "IncludePastEvents", Value.ForBool(true) }, { "IsExtendedSession", Value.ForBool(true) }, - { "ExtendedSessionIdleTimeoutInSeconds", Value.ForNumber(extendedSessionIdleTimeout) } }); + { "ExtendedSessionIdleTimeoutInSeconds", Value.ForNumber(DefaultExtendedSessionIdleTimeoutInSeconds) } }); byte[] requestBytes = orchestratorRequest.ToByteArray(); string requestString = Convert.ToBase64String(requestBytes); GrpcOrchestrationRunner.LoadAndRun(requestString, new CallSubOrchestrationOrchestrator(), extendedSessions); Assert.True(extendedSessions.IsInitialized); - Assert.True(extendedSessions.GetOrInitializeCache(extendedSessionIdleTimeout).TryGetValue(TestInstanceId, out object? extendedSession)); + Assert.True(extendedSessions.GetOrInitializeCache(DefaultExtendedSessionIdleTimeoutInSeconds).TryGetValue(TestInstanceId, out object? extendedSession)); // Now we will retry the same exact request. If the extended session is not evicted, then the request will fail due to duplicate ExecutionStarted events being detected // If the extended session is evicted because IncludePastEvents is true, then the request will succeed and a new extended session will be stored GrpcOrchestrationRunner.LoadAndRun(requestString, new CallSubOrchestrationOrchestrator(), extendedSessions); - Assert.True(extendedSessions.GetOrInitializeCache(extendedSessionIdleTimeout).TryGetValue(TestInstanceId, out extendedSession)); + Assert.True(extendedSessions.GetOrInitializeCache(DefaultExtendedSessionIdleTimeoutInSeconds).TryGetValue(TestInstanceId, out extendedSession)); } [Fact] From 464ad237a632845ee1f5f82e3c9b5356ffd2ec6c Mon Sep 17 00:00:00 2001 From: Sophia Tevosyan Date: Thu, 13 Nov 2025 16:45:43 -0800 Subject: [PATCH 03/10] updated tests --- src/Worker/Grpc/GrpcEntityRunner.cs | 9 +- .../Grpc.Tests/GrpcEntityRunnerTests.cs | 138 +++++++++++++----- .../GrpcOrchestrationRunnerTests.cs | 86 ++++++++--- 3 files changed, 174 insertions(+), 59 deletions(-) diff --git a/src/Worker/Grpc/GrpcEntityRunner.cs b/src/Worker/Grpc/GrpcEntityRunner.cs index 5124b129..c1fb1294 100644 --- a/src/Worker/Grpc/GrpcEntityRunner.cs +++ b/src/Worker/Grpc/GrpcEntityRunner.cs @@ -103,7 +103,6 @@ public static async Task LoadAndRunAsync( // If any of the request parameters are malformed, we assume the default - extended sessions are not enabled and the orchestration history is attached bool addToExtendedSessions = false; bool entityStateIncluded = true; - bool entityStateCached = false; bool isExtendedSession = false; double extendedSessionIdleTimeoutInSeconds = 0; @@ -130,14 +129,13 @@ public static async Task LoadAndRunAsync( addToExtendedSessions = true; // If an entity state was provided, even if we already have one stored, we always want to use the provided state. - if (!entityStateIncluded && extendedSessions.TryGetValue(request.InstanceId, out string? entityState)) + if (!entityStateIncluded && extendedSessions.TryGetValue(request.InstanceId, out string? entityState) && entityState is not null) { batch.EntityState = entityState; - entityStateCached = true; } } - if (!entityStateCached && !entityStateIncluded) + if (batch.EntityState == null && !entityStateIncluded) { // No state was provided, and we do not have one cached, so we cannot execute the batch request. return Convert.ToBase64String(new P.EntityBatchResult { RequiresState = true }.ToByteArray()); @@ -150,7 +148,8 @@ public static async Task LoadAndRunAsync( TaskEntity entity = factory.CreateEntity(entityName, implementation, id); EntityBatchResult result = await entity.ExecuteOperationBatchAsync(batch); - if (addToExtendedSessions) + // An entity with a null state will be deleted, so we should not cache it. + if (addToExtendedSessions && result.EntityState != null) { extendedSessions.Set( request.InstanceId, diff --git a/test/Worker/Grpc.Tests/GrpcEntityRunnerTests.cs b/test/Worker/Grpc.Tests/GrpcEntityRunnerTests.cs index dd20d87c..89a73f5e 100644 --- a/test/Worker/Grpc.Tests/GrpcEntityRunnerTests.cs +++ b/test/Worker/Grpc.Tests/GrpcEntityRunnerTests.cs @@ -5,6 +5,7 @@ using Google.Protobuf.Collections; using Google.Protobuf.WellKnownTypes; using Microsoft.DurableTask.Entities; +using Microsoft.Extensions.Caching.Memory; namespace Microsoft.DurableTask.Worker.Grpc.Tests; @@ -12,6 +13,8 @@ public class GrpcEntityRunnerTests { const string TestInstanceId = "@instance_id@my_entity"; const int DefaultExtendedSessionIdleTimeoutInSeconds = 30; + static readonly Protobuf.OperationRequest setOperation = new() { RequestId = Guid.NewGuid().ToString(), Input = 1.ToString(), Operation = "Set" }; + static readonly Protobuf.OperationRequest addOperation = new() { RequestId = Guid.NewGuid().ToString(), Input = 1.ToString(), Operation = "Add" }; [Fact] public async Task EmptyOrNullParameters_Throw_Exceptions() @@ -29,6 +32,43 @@ public async Task EmptyOrNullParameters_Throw_Exceptions() await act.Should().ThrowExactlyAsync().WithParameterName("implementation"); } + [Fact] + public async Task NullStateStored_Means_StoredStateNotUsed() + { + using var extendedSessions = new ExtendedSessionsCache(); + extendedSessions.GetOrInitializeCache(DefaultExtendedSessionIdleTimeoutInSeconds).Set( + TestInstanceId, + null!, + new MemoryCacheEntryOptions { SlidingExpiration = TimeSpan.FromSeconds(DefaultExtendedSessionIdleTimeoutInSeconds) }); + + // No state, so response indicates that a state is required + Protobuf.EntityBatchRequest entityRequest = CreateEntityRequest([]); + entityRequest.Properties.Add(new MapField() { + { "IncludeEntityState", Value.ForBool(false) }, + { "IsExtendedSession", Value.ForBool(true) }, + { "ExtendedSessionIdleTimeoutInSeconds", Value.ForNumber(DefaultExtendedSessionIdleTimeoutInSeconds) } }); + byte[] requestBytes = entityRequest.ToByteArray(); + string requestString = Convert.ToBase64String(requestBytes); + string responseString = await GrpcEntityRunner.LoadAndRunAsync(requestString, new SimpleEntity(), extendedSessions); + Protobuf.EntityBatchResult response = Protobuf.EntityBatchResult.Parser.ParseFrom(Convert.FromBase64String(responseString)); + Assert.True(response.RequiresState); + + // State provided so request can be fulfilled, new state stored + entityRequest = CreateEntityRequest([setOperation]); + entityRequest.Properties.Add(new MapField() { + { "IncludeEntityState", Value.ForBool(true) }, + { "IsExtendedSession", Value.ForBool(true) }, + { "ExtendedSessionIdleTimeoutInSeconds", Value.ForNumber(DefaultExtendedSessionIdleTimeoutInSeconds) } }); + requestBytes = entityRequest.ToByteArray(); + requestString = Convert.ToBase64String(requestBytes); + responseString = await GrpcEntityRunner.LoadAndRunAsync(requestString, new SimpleEntity(), extendedSessions); + response = Protobuf.EntityBatchResult.Parser.ParseFrom(Convert.FromBase64String(responseString)); + Assert.False(response.RequiresState); + Assert.True(extendedSessions.GetOrInitializeCache(DefaultExtendedSessionIdleTimeoutInSeconds).TryGetValue(TestInstanceId, out object? extendedSession)); + Assert.Equal("1", extendedSession); + Assert.Equal("1", response.EntityState); + } + [Fact] public async Task EmptyState_Returns_NeedsStateInResponse() { @@ -40,8 +80,8 @@ public async Task EmptyState_Returns_NeedsStateInResponse() { "IncludeEntityState", Value.ForBool(false) }}); byte[] requestBytes = entityRequest.ToByteArray(); string requestString = Convert.ToBase64String(requestBytes); - string stringResponse = await GrpcEntityRunner.LoadAndRunAsync(requestString, new SimpleEntity(), extendedSessions); - Protobuf.EntityBatchResult response = Protobuf.EntityBatchResult.Parser.ParseFrom(Convert.FromBase64String(stringResponse)); + string responseString = await GrpcEntityRunner.LoadAndRunAsync(requestString, new SimpleEntity(), extendedSessions); + Protobuf.EntityBatchResult response = Protobuf.EntityBatchResult.Parser.ParseFrom(Convert.FromBase64String(responseString)); Assert.True(response.RequiresState); Assert.False(extendedSessions.IsInitialized); @@ -51,8 +91,8 @@ public async Task EmptyState_Returns_NeedsStateInResponse() { "ExtendedSessionIdleTimeoutInSeconds", Value.ForNumber(DefaultExtendedSessionIdleTimeoutInSeconds) } }); requestBytes = entityRequest.ToByteArray(); requestString = Convert.ToBase64String(requestBytes); - stringResponse = await GrpcEntityRunner.LoadAndRunAsync(requestString, new SimpleEntity(), extendedSessions); - response = Protobuf.EntityBatchResult.Parser.ParseFrom(Convert.FromBase64String(stringResponse)); + responseString = await GrpcEntityRunner.LoadAndRunAsync(requestString, new SimpleEntity(), extendedSessions); + response = Protobuf.EntityBatchResult.Parser.ParseFrom(Convert.FromBase64String(responseString)); Assert.True(response.RequiresState); Assert.True(extendedSessions.IsInitialized); } @@ -70,8 +110,8 @@ public async Task MalformedRequestParameters_Means_CacheNotInitialized() { "ExtendedSessionsIdleTimeoutInSeconds", Value.ForNumber(DefaultExtendedSessionIdleTimeoutInSeconds) } }); byte[] requestBytes = entityRequest.ToByteArray(); string requestString = Convert.ToBase64String(requestBytes); - string stringResponse = await GrpcEntityRunner.LoadAndRunAsync(requestString, new SimpleEntity(), extendedSessions); - Protobuf.OrchestratorResponse response = Protobuf.OrchestratorResponse.Parser.ParseFrom(Convert.FromBase64String(stringResponse)); + string responseString = await GrpcEntityRunner.LoadAndRunAsync(requestString, new SimpleEntity(), extendedSessions); + Protobuf.OrchestratorResponse response = Protobuf.OrchestratorResponse.Parser.ParseFrom(Convert.FromBase64String(responseString)); Assert.False(extendedSessions.IsInitialized); // Wrong value type for extended session timeout key @@ -154,7 +194,7 @@ public async Task MalformedRequestParameters_Means_CacheNotInitialized() public async Task IsExtendedSessionFalse_Means_NoExtendedSessionStored() { using var extendedSessions = new ExtendedSessionsCache(); - Protobuf.EntityBatchRequest entityRequest = CreateEntityRequest([new Protobuf.OperationRequest() { RequestId = Guid.NewGuid().ToString(), Input = 1.ToString(), Operation = "Set"}]); + Protobuf.EntityBatchRequest entityRequest = CreateEntityRequest([setOperation]); entityRequest.Properties.Add(new MapField() { { "IncludeEntityState", Value.ForBool(false) }, @@ -177,9 +217,7 @@ public async Task IsExtendedSessionFalse_Means_NoExtendedSessionStored() public async Task MalformedIncludeEntityStateParameter_Means_NoStateRequired() { using var extendedSessions = new ExtendedSessionsCache(); - Protobuf.EntityBatchRequest entityRequest = CreateEntityRequest( - [new Protobuf.OperationRequest() { RequestId = Guid.NewGuid().ToString(), Input = 1.ToString(), Operation = "Add" }], - entityState: 1.ToString()); + Protobuf.EntityBatchRequest entityRequest = CreateEntityRequest([addOperation], entityState: 1.ToString()); // Misspelled include entity state key entityRequest.Properties.Add(new MapField() { @@ -188,8 +226,8 @@ [new Protobuf.OperationRequest() { RequestId = Guid.NewGuid().ToString(), Input { "ExtendedSessionIdleTimeoutInSeconds", Value.ForNumber(DefaultExtendedSessionIdleTimeoutInSeconds) } }); byte[] requestBytes = entityRequest.ToByteArray(); string requestString = Convert.ToBase64String(requestBytes); - string stringResponse = await GrpcEntityRunner.LoadAndRunAsync(requestString, new SimpleEntity(), extendedSessions); - Protobuf.EntityBatchResult response = Protobuf.EntityBatchResult.Parser.ParseFrom(Convert.FromBase64String(stringResponse)); + string responseString = await GrpcEntityRunner.LoadAndRunAsync(requestString, new SimpleEntity(), extendedSessions); + Protobuf.EntityBatchResult response = Protobuf.EntityBatchResult.Parser.ParseFrom(Convert.FromBase64String(responseString)); Assert.False(response.RequiresState); Assert.Equal("2", response.EntityState); @@ -201,8 +239,8 @@ [new Protobuf.OperationRequest() { RequestId = Guid.NewGuid().ToString(), Input { "ExtendedSessionIdleTimeoutInSeconds", Value.ForNumber(DefaultExtendedSessionIdleTimeoutInSeconds) } }); requestBytes = entityRequest.ToByteArray(); requestString = Convert.ToBase64String(requestBytes); - stringResponse = await GrpcEntityRunner.LoadAndRunAsync(requestString, new SimpleEntity(), extendedSessions); - response = Protobuf.EntityBatchResult.Parser.ParseFrom(Convert.FromBase64String(stringResponse)); + responseString = await GrpcEntityRunner.LoadAndRunAsync(requestString, new SimpleEntity(), extendedSessions); + response = Protobuf.EntityBatchResult.Parser.ParseFrom(Convert.FromBase64String(responseString)); Assert.False(response.RequiresState); Assert.Equal("2", response.EntityState); @@ -213,8 +251,8 @@ [new Protobuf.OperationRequest() { RequestId = Guid.NewGuid().ToString(), Input { "ExtendedSessionIdleTimeoutInSeconds", Value.ForNumber(DefaultExtendedSessionIdleTimeoutInSeconds) } }); requestBytes = entityRequest.ToByteArray(); requestString = Convert.ToBase64String(requestBytes); - stringResponse = await GrpcEntityRunner.LoadAndRunAsync(requestString, new SimpleEntity(), extendedSessions); - response = Protobuf.EntityBatchResult.Parser.ParseFrom(Convert.FromBase64String(stringResponse)); + responseString = await GrpcEntityRunner.LoadAndRunAsync(requestString, new SimpleEntity(), extendedSessions); + response = Protobuf.EntityBatchResult.Parser.ParseFrom(Convert.FromBase64String(responseString)); Assert.False(response.RequiresState); Assert.Equal("2", response.EntityState); } @@ -223,24 +261,26 @@ [new Protobuf.OperationRequest() { RequestId = Guid.NewGuid().ToString(), Input public async Task Entity_State_Stored() { using var extendedSessions = new ExtendedSessionsCache(); - Protobuf.EntityBatchRequest entityRequest = CreateEntityRequest([new Protobuf.OperationRequest() { RequestId = Guid.NewGuid().ToString(), Input = 1.ToString(), Operation = "Set" }]); + Protobuf.EntityBatchRequest entityRequest = CreateEntityRequest([setOperation]); entityRequest.Properties.Add(new MapField() { { "IncludeEntityState", Value.ForBool(true) }, { "IsExtendedSession", Value.ForBool(true) }, { "ExtendedSessionIdleTimeoutInSeconds", Value.ForNumber(DefaultExtendedSessionIdleTimeoutInSeconds) } }); byte[] requestBytes = entityRequest.ToByteArray(); string requestString = Convert.ToBase64String(requestBytes); - await GrpcEntityRunner.LoadAndRunAsync(requestString, new SimpleEntity(), extendedSessions); + string responseString = await GrpcEntityRunner.LoadAndRunAsync(requestString, new SimpleEntity(), extendedSessions); + Protobuf.EntityBatchResult response = Protobuf.EntityBatchResult.Parser.ParseFrom(Convert.FromBase64String(responseString)); Assert.True(extendedSessions.IsInitialized); Assert.True(extendedSessions.GetOrInitializeCache(DefaultExtendedSessionIdleTimeoutInSeconds).TryGetValue(TestInstanceId, out object? extendedSession)); Assert.Equal("1", extendedSession); + Assert.Equal("1", response.EntityState); } [Fact] - public async Task ExternallyEndedExtendedSession_Evicted() + public async Task Null_Entity_State_NotStored() { using var extendedSessions = new ExtendedSessionsCache(); - Protobuf.EntityBatchRequest entityRequest = CreateEntityRequest([new Protobuf.OperationRequest() { RequestId = Guid.NewGuid().ToString(), Input = 1.ToString(), Operation = "Set" }]); + Protobuf.EntityBatchRequest entityRequest = CreateEntityRequest([new Protobuf.OperationRequest() { RequestId = Guid.NewGuid().ToString(), Operation = "Delete" }]); entityRequest.Properties.Add(new MapField() { { "IncludeEntityState", Value.ForBool(true) }, { "IsExtendedSession", Value.ForBool(true) }, @@ -249,8 +289,26 @@ public async Task ExternallyEndedExtendedSession_Evicted() string requestString = Convert.ToBase64String(requestBytes); await GrpcEntityRunner.LoadAndRunAsync(requestString, new SimpleEntity(), extendedSessions); Assert.True(extendedSessions.IsInitialized); + Assert.False(extendedSessions.GetOrInitializeCache(DefaultExtendedSessionIdleTimeoutInSeconds).TryGetValue(TestInstanceId, out object? extendedSession)); + } + + [Fact] + public async Task ExternallyEndedExtendedSession_Evicted() + { + using var extendedSessions = new ExtendedSessionsCache(); + Protobuf.EntityBatchRequest entityRequest = CreateEntityRequest([setOperation]); + entityRequest.Properties.Add(new MapField() { + { "IncludeEntityState", Value.ForBool(true) }, + { "IsExtendedSession", Value.ForBool(true) }, + { "ExtendedSessionIdleTimeoutInSeconds", Value.ForNumber(DefaultExtendedSessionIdleTimeoutInSeconds) } }); + byte[] requestBytes = entityRequest.ToByteArray(); + string requestString = Convert.ToBase64String(requestBytes); + string responseString = await GrpcEntityRunner.LoadAndRunAsync(requestString, new SimpleEntity(), extendedSessions); + Protobuf.EntityBatchResult response = Protobuf.EntityBatchResult.Parser.ParseFrom(Convert.FromBase64String(responseString)); + Assert.True(extendedSessions.IsInitialized); Assert.True(extendedSessions.GetOrInitializeCache(DefaultExtendedSessionIdleTimeoutInSeconds).TryGetValue(TestInstanceId, out object? extendedSession)); Assert.Equal("1", extendedSession); + Assert.Equal("1", response.EntityState); // Now set the extended session flag to false for this instance entityRequest.Properties.Clear(); @@ -270,17 +328,19 @@ public async Task Stale_ExtendedSessions_Evicted_Async() { using var extendedSessions = new ExtendedSessionsCache(); int extendedSessionIdleTimeout = 5; - Protobuf.EntityBatchRequest entityRequest = CreateEntityRequest([new Protobuf.OperationRequest() { RequestId = Guid.NewGuid().ToString(), Input = 1.ToString(), Operation = "Set" }]); + Protobuf.EntityBatchRequest entityRequest = CreateEntityRequest([setOperation]); entityRequest.Properties.Add(new MapField() { { "IncludeEntityState", Value.ForBool(true) }, { "IsExtendedSession", Value.ForBool(true) }, { "ExtendedSessionIdleTimeoutInSeconds", Value.ForNumber(extendedSessionIdleTimeout) } }); byte[] requestBytes = entityRequest.ToByteArray(); string requestString = Convert.ToBase64String(requestBytes); - await GrpcEntityRunner.LoadAndRunAsync(requestString, new SimpleEntity(), extendedSessions); + string responseString = await GrpcEntityRunner.LoadAndRunAsync(requestString, new SimpleEntity(), extendedSessions); + Protobuf.EntityBatchResult response = Protobuf.EntityBatchResult.Parser.ParseFrom(Convert.FromBase64String(responseString)); Assert.True(extendedSessions.IsInitialized); Assert.True(extendedSessions.GetOrInitializeCache(extendedSessionIdleTimeout).TryGetValue(TestInstanceId, out object? extendedSession)); Assert.Equal("1", extendedSession); + Assert.Equal("1", response.EntityState); // Wait for longer than the timeout to account for finite cache scan for stale items frequency await Task.Delay(extendedSessionIdleTimeout * 1000 * 2); @@ -294,8 +354,8 @@ public async Task Stale_ExtendedSessions_Evicted_Async() { "ExtendedSessionIdleTimeoutInSeconds", Value.ForNumber(extendedSessionIdleTimeout) } }); requestBytes = entityRequest.ToByteArray(); requestString = Convert.ToBase64String(requestBytes); - string stringResponse = await GrpcEntityRunner.LoadAndRunAsync(requestString, new SimpleEntity(), extendedSessions); - Protobuf.EntityBatchResult response = Protobuf.EntityBatchResult.Parser.ParseFrom(Convert.FromBase64String(stringResponse)); + responseString = await GrpcEntityRunner.LoadAndRunAsync(requestString, new SimpleEntity(), extendedSessions); + response = Protobuf.EntityBatchResult.Parser.ParseFrom(Convert.FromBase64String(responseString)); Assert.True(response.RequiresState); } @@ -303,8 +363,7 @@ public async Task Stale_ExtendedSessions_Evicted_Async() public async Task EntityStateIncluded_Means_ExtendedSession_Evicted() { using var extendedSessions = new ExtendedSessionsCache(); - Protobuf.OperationRequest operationRequest = new() { RequestId = Guid.NewGuid().ToString(), Input = 1.ToString(), Operation = "Add" }; - Protobuf.EntityBatchRequest entityRequest = CreateEntityRequest([operationRequest], entityState: 1.ToString()); + Protobuf.EntityBatchRequest entityRequest = CreateEntityRequest([addOperation], entityState: 1.ToString()); entityRequest.Properties.Add(new MapField() { { "IncludeEntityState", Value.ForBool(true) }, { "IsExtendedSession", Value.ForBool(true) }, @@ -316,22 +375,24 @@ public async Task EntityStateIncluded_Means_ExtendedSession_Evicted() Assert.True(extendedSessions.GetOrInitializeCache(DefaultExtendedSessionIdleTimeoutInSeconds).TryGetValue(TestInstanceId, out object? extendedSession)); // Now we will retry the same request, but with a different value for the state. If the extended session is not evicted, then the result will be incorrect. - entityRequest = CreateEntityRequest([operationRequest], entityState: 5.ToString()); + entityRequest = CreateEntityRequest([addOperation], entityState: 5.ToString()); entityRequest.Properties.Add(new MapField() { { "IncludeEntityState", Value.ForBool(true) }, { "IsExtendedSession", Value.ForBool(true) }, { "ExtendedSessionIdleTimeoutInSeconds", Value.ForNumber(DefaultExtendedSessionIdleTimeoutInSeconds) } }); requestBytes = entityRequest.ToByteArray(); requestString = Convert.ToBase64String(requestBytes); - await GrpcEntityRunner.LoadAndRunAsync(requestString, new SimpleEntity(), extendedSessions); + string responseString = await GrpcEntityRunner.LoadAndRunAsync(requestString, new SimpleEntity(), extendedSessions); + Protobuf.EntityBatchResult response = Protobuf.EntityBatchResult.Parser.ParseFrom(Convert.FromBase64String(responseString)); Assert.True(extendedSessions.GetOrInitializeCache(DefaultExtendedSessionIdleTimeoutInSeconds).TryGetValue(TestInstanceId, out extendedSession)); Assert.Equal("6", extendedSession); + Assert.Equal("6", response.EntityState); } [Fact] public async Task Null_ExtendedSessionsCache_IsOkay() { - Protobuf.EntityBatchRequest entityRequest = CreateEntityRequest([new Protobuf.OperationRequest() { RequestId = Guid.NewGuid().ToString(), Input = 1.ToString(), Operation = "Set" }]); + Protobuf.EntityBatchRequest entityRequest = CreateEntityRequest([setOperation]); // Set up the parameters as if extended sessions are enabled, but do not pass an extended session cache to the request. // The request should still be successful. @@ -341,16 +402,16 @@ public async Task Null_ExtendedSessionsCache_IsOkay() { "ExtendedSessionIdleTimeoutInSeconds", Value.ForNumber(DefaultExtendedSessionIdleTimeoutInSeconds) } }); byte[] requestBytes = entityRequest.ToByteArray(); string requestString = Convert.ToBase64String(requestBytes); - string stringResponse = await GrpcEntityRunner.LoadAndRunAsync(requestString, new SimpleEntity()); - Protobuf.EntityBatchResult response = Protobuf.EntityBatchResult.Parser.ParseFrom(Convert.FromBase64String(stringResponse)); + string responseString = await GrpcEntityRunner.LoadAndRunAsync(requestString, new SimpleEntity()); + Protobuf.EntityBatchResult response = Protobuf.EntityBatchResult.Parser.ParseFrom(Convert.FromBase64String(responseString)); Assert.Equal("1", response.EntityState); // Now try it again without any properties specified. The request should still be successful. entityRequest.Properties.Clear(); requestBytes = entityRequest.ToByteArray(); requestString = Convert.ToBase64String(requestBytes); - stringResponse = await GrpcEntityRunner.LoadAndRunAsync(requestString, new SimpleEntity()); - response = Protobuf.EntityBatchResult.Parser.ParseFrom(Convert.FromBase64String(stringResponse)); + responseString = await GrpcEntityRunner.LoadAndRunAsync(requestString, new SimpleEntity()); + response = Protobuf.EntityBatchResult.Parser.ParseFrom(Convert.FromBase64String(responseString)); Assert.Equal("1", response.EntityState); } @@ -365,16 +426,21 @@ static Protobuf.EntityBatchRequest CreateEntityRequest(IEnumerable + class SimpleEntity : TaskEntity { - public async Task Set(int value) + public void Set(int value) { this.State = value; } - public async Task Add(int value) + public void Add(int value) { this.State += value; } + + public void Delete() + { + this.State = null; + } } } diff --git a/test/Worker/Grpc.Tests/GrpcOrchestrationRunnerTests.cs b/test/Worker/Grpc.Tests/GrpcOrchestrationRunnerTests.cs index b701dd08..15d6e186 100644 --- a/test/Worker/Grpc.Tests/GrpcOrchestrationRunnerTests.cs +++ b/test/Worker/Grpc.Tests/GrpcOrchestrationRunnerTests.cs @@ -4,6 +4,7 @@ using Google.Protobuf; using Google.Protobuf.Collections; using Google.Protobuf.WellKnownTypes; +using Microsoft.Extensions.Caching.Memory; namespace Microsoft.DurableTask.Worker.Grpc.Tests; @@ -40,8 +41,8 @@ public void EmptyHistory_Returns_NeedsHistoryInResponse() { "IncludePastEvents", Value.ForBool(false) }}); byte[] requestBytes = orchestratorRequest.ToByteArray(); string requestString = Convert.ToBase64String(requestBytes); - string stringResponse = GrpcOrchestrationRunner.LoadAndRun(requestString, new SimpleOrchestrator(), extendedSessions); - Protobuf.OrchestratorResponse response = Protobuf.OrchestratorResponse.Parser.ParseFrom(Convert.FromBase64String(stringResponse)); + string responseString = GrpcOrchestrationRunner.LoadAndRun(requestString, new SimpleOrchestrator(), extendedSessions); + Protobuf.OrchestratorResponse response = Protobuf.OrchestratorResponse.Parser.ParseFrom(Convert.FromBase64String(responseString)); Assert.True(response.RequiresHistory); Assert.False(extendedSessions.IsInitialized); @@ -51,12 +52,61 @@ public void EmptyHistory_Returns_NeedsHistoryInResponse() { "ExtendedSessionIdleTimeoutInSeconds", Value.ForNumber(DefaultExtendedSessionIdleTimeoutInSeconds) } }); requestBytes = orchestratorRequest.ToByteArray(); requestString = Convert.ToBase64String(requestBytes); - stringResponse = GrpcOrchestrationRunner.LoadAndRun(requestString, new SimpleOrchestrator(), extendedSessions); - response = Protobuf.OrchestratorResponse.Parser.ParseFrom(Convert.FromBase64String(stringResponse)); + responseString = GrpcOrchestrationRunner.LoadAndRun(requestString, new SimpleOrchestrator(), extendedSessions); + response = Protobuf.OrchestratorResponse.Parser.ParseFrom(Convert.FromBase64String(responseString)); Assert.True(response.RequiresHistory); Assert.True(extendedSessions.IsInitialized); } + [Fact] + public void NullExtendedSessionStored_Means_NeedsExtendedSessionNotUsed() + { + using var extendedSessions = new ExtendedSessionsCache(); + extendedSessions.GetOrInitializeCache(DefaultExtendedSessionIdleTimeoutInSeconds).Set( + TestInstanceId, + null!, + new MemoryCacheEntryOptions { SlidingExpiration = TimeSpan.FromSeconds(DefaultExtendedSessionIdleTimeoutInSeconds) }); + + // No history, so response indicates that a history is required + Protobuf.OrchestratorRequest orchestratorRequest = CreateOrchestratorRequest([]); + orchestratorRequest.Properties.Add(new MapField() { + { "IncludePastEvents", Value.ForBool(false) }, + { "IsExtendedSession", Value.ForBool(true) }, + { "ExtendedSessionIdleTimeoutInSeconds", Value.ForNumber(DefaultExtendedSessionIdleTimeoutInSeconds) } }); + byte[] requestBytes = orchestratorRequest.ToByteArray(); + string requestString = Convert.ToBase64String(requestBytes); + string responseString = GrpcOrchestrationRunner.LoadAndRun(requestString, new SimpleOrchestrator(), extendedSessions); + Protobuf.OrchestratorResponse response = Protobuf.OrchestratorResponse.Parser.ParseFrom(Convert.FromBase64String(responseString)); + Assert.True(response.RequiresHistory); + + // History provided so the request can be fulfilled and an extended session is stored + var historyEvent = new Protobuf.HistoryEvent + { + EventId = -1, + Timestamp = Timestamp.FromDateTime(DateTime.UtcNow), + ExecutionStarted = new Protobuf.ExecutionStartedEvent() + { + OrchestrationInstance = new Protobuf.OrchestrationInstance + { + InstanceId = TestInstanceId, + ExecutionId = TestExecutionId, + }, + } + }; + orchestratorRequest = CreateOrchestratorRequest([historyEvent]); + orchestratorRequest.Properties.Add(new MapField() { + { "IncludePastEvents", Value.ForBool(true) }, + { "IsExtendedSession", Value.ForBool(true) }, + { "ExtendedSessionIdleTimeoutInSeconds", Value.ForNumber(DefaultExtendedSessionIdleTimeoutInSeconds) } }); + requestBytes = orchestratorRequest.ToByteArray(); + requestString = Convert.ToBase64String(requestBytes); + responseString = GrpcOrchestrationRunner.LoadAndRun(requestString, new CallSubOrchestrationOrchestrator(), extendedSessions); + response = Protobuf.OrchestratorResponse.Parser.ParseFrom(Convert.FromBase64String(responseString)); + Assert.False(response.RequiresHistory); + Assert.True(extendedSessions.GetOrInitializeCache(DefaultExtendedSessionIdleTimeoutInSeconds).TryGetValue(TestInstanceId, out object? extendedSession)); + Assert.NotNull(extendedSession); + } + [Fact] public void MalformedRequestParameters_Means_CacheNotInitialized() { @@ -70,8 +120,8 @@ public void MalformedRequestParameters_Means_CacheNotInitialized() { "ExtendedSessionsIdleTimeoutInSeconds", Value.ForNumber(DefaultExtendedSessionIdleTimeoutInSeconds) } }); byte[] requestBytes = orchestratorRequest.ToByteArray(); string requestString = Convert.ToBase64String(requestBytes); - string stringResponse = GrpcOrchestrationRunner.LoadAndRun(requestString, new SimpleOrchestrator(), extendedSessions); - Protobuf.OrchestratorResponse response = Protobuf.OrchestratorResponse.Parser.ParseFrom(Convert.FromBase64String(stringResponse)); + string responseString = GrpcOrchestrationRunner.LoadAndRun(requestString, new SimpleOrchestrator(), extendedSessions); + Protobuf.OrchestratorResponse response = Protobuf.OrchestratorResponse.Parser.ParseFrom(Convert.FromBase64String(responseString)); Assert.False(extendedSessions.IsInitialized); // Wrong value type for extended session timeout key @@ -199,8 +249,8 @@ public void MalformedPastEventsParameter_Means_NoHistoryRequired() { "ExtendedSessionIdleTimeoutInSeconds", Value.ForNumber(DefaultExtendedSessionIdleTimeoutInSeconds) } }); byte[] requestBytes = orchestratorRequest.ToByteArray(); string requestString = Convert.ToBase64String(requestBytes); - string stringResponse = GrpcOrchestrationRunner.LoadAndRun(requestString, new SimpleOrchestrator(), extendedSessions); - Protobuf.OrchestratorResponse response = Protobuf.OrchestratorResponse.Parser.ParseFrom(Convert.FromBase64String(stringResponse)); + string responseString = GrpcOrchestrationRunner.LoadAndRun(requestString, new SimpleOrchestrator(), extendedSessions); + Protobuf.OrchestratorResponse response = Protobuf.OrchestratorResponse.Parser.ParseFrom(Convert.FromBase64String(responseString)); Assert.False(response.RequiresHistory); // Wrong value type for include past events key @@ -211,8 +261,8 @@ public void MalformedPastEventsParameter_Means_NoHistoryRequired() { "ExtendedSessionIdleTimeoutInSeconds", Value.ForNumber(DefaultExtendedSessionIdleTimeoutInSeconds) } }); requestBytes = orchestratorRequest.ToByteArray(); requestString = Convert.ToBase64String(requestBytes); - stringResponse = GrpcOrchestrationRunner.LoadAndRun(requestString, new SimpleOrchestrator(), extendedSessions); - response = Protobuf.OrchestratorResponse.Parser.ParseFrom(Convert.FromBase64String(stringResponse)); + responseString = GrpcOrchestrationRunner.LoadAndRun(requestString, new SimpleOrchestrator(), extendedSessions); + response = Protobuf.OrchestratorResponse.Parser.ParseFrom(Convert.FromBase64String(responseString)); Assert.False(response.RequiresHistory); // No include past events key @@ -222,8 +272,8 @@ public void MalformedPastEventsParameter_Means_NoHistoryRequired() { "ExtendedSessionIdleTimeoutInSeconds", Value.ForNumber(DefaultExtendedSessionIdleTimeoutInSeconds) } }); requestBytes = orchestratorRequest.ToByteArray(); requestString = Convert.ToBase64String(requestBytes); - stringResponse = GrpcOrchestrationRunner.LoadAndRun(requestString, new SimpleOrchestrator(), extendedSessions); - response = Protobuf.OrchestratorResponse.Parser.ParseFrom(Convert.FromBase64String(stringResponse)); + responseString = GrpcOrchestrationRunner.LoadAndRun(requestString, new SimpleOrchestrator(), extendedSessions); + response = Protobuf.OrchestratorResponse.Parser.ParseFrom(Convert.FromBase64String(responseString)); Assert.False(response.RequiresHistory); } @@ -367,8 +417,8 @@ public async void Stale_ExtendedSessions_Evicted_Async() { "ExtendedSessionIdleTimeoutInSeconds", Value.ForNumber(extendedSessionIdleTimeout) } }); requestBytes = orchestratorRequest.ToByteArray(); requestString = Convert.ToBase64String(requestBytes); - string stringResponse = GrpcOrchestrationRunner.LoadAndRun(requestString, new CallSubOrchestrationOrchestrator(), extendedSessions); - Protobuf.OrchestratorResponse response = Protobuf.OrchestratorResponse.Parser.ParseFrom(Convert.FromBase64String(stringResponse)); + string responseString = GrpcOrchestrationRunner.LoadAndRun(requestString, new CallSubOrchestrationOrchestrator(), extendedSessions); + Protobuf.OrchestratorResponse response = Protobuf.OrchestratorResponse.Parser.ParseFrom(Convert.FromBase64String(responseString)); Assert.True(response.RequiresHistory); } @@ -432,8 +482,8 @@ public void Null_ExtendedSessionsCache_IsOkay() { "ExtendedSessionIdleTimeoutInSeconds", Value.ForNumber(DefaultExtendedSessionIdleTimeoutInSeconds) } }); byte[] requestBytes = orchestratorRequest.ToByteArray(); string requestString = Convert.ToBase64String(requestBytes); - string stringResponse = GrpcOrchestrationRunner.LoadAndRun(requestString, new SimpleOrchestrator()); - Protobuf.OrchestratorResponse response = Protobuf.OrchestratorResponse.Parser.ParseFrom(Convert.FromBase64String(stringResponse)); + string responseString = GrpcOrchestrationRunner.LoadAndRun(requestString, new SimpleOrchestrator()); + Protobuf.OrchestratorResponse response = Protobuf.OrchestratorResponse.Parser.ParseFrom(Convert.FromBase64String(responseString)); Assert.Single(response.Actions); Assert.NotNull(response.Actions[0].CompleteOrchestration); Assert.Equal(Protobuf.OrchestrationStatus.Completed, response.Actions[0].CompleteOrchestration.OrchestrationStatus); @@ -442,8 +492,8 @@ public void Null_ExtendedSessionsCache_IsOkay() orchestratorRequest.Properties.Clear(); requestBytes = orchestratorRequest.ToByteArray(); requestString = Convert.ToBase64String(requestBytes); - stringResponse = GrpcOrchestrationRunner.LoadAndRun(requestString, new SimpleOrchestrator()); - response = Protobuf.OrchestratorResponse.Parser.ParseFrom(Convert.FromBase64String(stringResponse)); + responseString = GrpcOrchestrationRunner.LoadAndRun(requestString, new SimpleOrchestrator()); + response = Protobuf.OrchestratorResponse.Parser.ParseFrom(Convert.FromBase64String(responseString)); Assert.Single(response.Actions); Assert.NotNull(response.Actions[0].CompleteOrchestration); Assert.Equal(Protobuf.OrchestrationStatus.Completed, response.Actions[0].CompleteOrchestration.OrchestrationStatus); From 21512fd57f8aa84313b1f8a59c36429589fba080 Mon Sep 17 00:00:00 2001 From: Sophia Tevosyan Date: Thu, 13 Nov 2025 23:09:03 -0800 Subject: [PATCH 04/10] refactoring --- src/Worker/Grpc/GrpcEntityRunner.cs | 33 +++------- src/Worker/Grpc/GrpcInstanceRunnerUtils.cs | 71 ++++++++++++++++++++++ src/Worker/Grpc/GrpcOrchestrationRunner.cs | 30 +++------ 3 files changed, 86 insertions(+), 48 deletions(-) create mode 100644 src/Worker/Grpc/GrpcInstanceRunnerUtils.cs diff --git a/src/Worker/Grpc/GrpcEntityRunner.cs b/src/Worker/Grpc/GrpcEntityRunner.cs index c1fb1294..db99c863 100644 --- a/src/Worker/Grpc/GrpcEntityRunner.cs +++ b/src/Worker/Grpc/GrpcEntityRunner.cs @@ -98,31 +98,14 @@ public static async Task LoadAndRunAsync( EntityId id = EntityId.FromString(batch.InstanceId!); TaskName entityName = new(id.Name); - MemoryCache? extendedSessions = null; - - // If any of the request parameters are malformed, we assume the default - extended sessions are not enabled and the orchestration history is attached bool addToExtendedSessions = false; - bool entityStateIncluded = true; - bool isExtendedSession = false; - double extendedSessionIdleTimeoutInSeconds = 0; - - // Only attempt to initialize the extended sessions cache if all the parameters are correctly specified - if (properties.TryGetValue("ExtendedSessionIdleTimeoutInSeconds", out object? extendedSessionIdleTimeoutObj) - && extendedSessionIdleTimeoutObj is double extendedSessionIdleTimeout - && extendedSessionIdleTimeout > 0 - && properties.TryGetValue("IsExtendedSession", out object? extendedSessionObj) - && extendedSessionObj is bool extendedSession) - { - extendedSessionIdleTimeoutInSeconds = extendedSessionIdleTimeout; - isExtendedSession = extendedSession; - extendedSessions = extendedSessionsCache?.GetOrInitializeCache(extendedSessionIdleTimeoutInSeconds); - } - - if (properties.TryGetValue("IncludeEntityState", out object? includeEntityStateObj) - && includeEntityStateObj is bool includeEntityState) - { - entityStateIncluded = includeEntityState; - } + GrpcInstanceRunnerUtils.ParseRequestPropertiesAndInitializeCache( + properties, + extendedSessionsCache, + out double extendedSessionIdleTimeoutInSeconds, + out bool isExtendedSession, + out bool entityStateIncluded, + out MemoryCache? extendedSessions); if (isExtendedSession && extendedSessions != null) { @@ -149,7 +132,7 @@ public static async Task LoadAndRunAsync( EntityBatchResult result = await entity.ExecuteOperationBatchAsync(batch); // An entity with a null state will be deleted, so we should not cache it. - if (addToExtendedSessions && result.EntityState != null) + if (addToExtendedSessions && result.EntityState is not null) { extendedSessions.Set( request.InstanceId, diff --git a/src/Worker/Grpc/GrpcInstanceRunnerUtils.cs b/src/Worker/Grpc/GrpcInstanceRunnerUtils.cs new file mode 100644 index 00000000..6d395bc8 --- /dev/null +++ b/src/Worker/Grpc/GrpcInstanceRunnerUtils.cs @@ -0,0 +1,71 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +using System; +using System.Collections.Generic; +using System.Text; +using Microsoft.Extensions.Caching.Memory; + +namespace Microsoft.DurableTask.Worker.Grpc; + +/// +/// Utility methods for the and classes. +/// +static class GrpcInstanceRunnerUtils +{ + /// + /// Parses request properties to determine extended session settings and initializes the extended sessions cache if + /// the settings are properly enabled. + /// + /// + /// If any request property is missing or invalid (i.e. the key is misspelled or the value is of the wrong type), + /// extended sessions are not enabled and default values are assigned are assigned to the returns. + /// + /// + /// A dictionary containing request properties used to configure extended session behavior. + /// + /// The extended sessions cache manager. + /// + /// When the method returns, contains the idle timeout value for extended sessions, in seconds. Cache entries that + /// have not been accessed in this timeframe are evicted from . + /// Set to zero if extended sessions are not enabled. + /// + /// When the method returns, indicates whether this request is from within an extended session. + /// When the method returns, indicates whether instance state is included in the request. + /// When the method returns, contains the extended sessions cache initialized from + /// if and + /// are correctly specified in the ; otherwise, null. + /// + internal static void ParseRequestPropertiesAndInitializeCache( + Dictionary properties, + ExtendedSessionsCache? extendedSessionsCache, + out double extendedSessionIdleTimeoutInSeconds, + out bool isExtendedSession, + out bool stateIncluded, + out MemoryCache? extendedSessions) + { + // If any of the request parameters are malformed, we assume the default - extended sessions are not enabled and the instance state is attached + extendedSessions = null; + stateIncluded = true; + isExtendedSession = false; + extendedSessionIdleTimeoutInSeconds = 0; + + // Only attempt to initialize the extended sessions cache if all the parameters are correctly specified + if (properties.TryGetValue("ExtendedSessionIdleTimeoutInSeconds", out object? extendedSessionIdleTimeoutObj) + && extendedSessionIdleTimeoutObj is double extendedSessionIdleTimeout + && extendedSessionIdleTimeout > 0 + && properties.TryGetValue("IsExtendedSession", out object? extendedSessionObj) + && extendedSessionObj is bool extendedSession) + { + extendedSessionIdleTimeoutInSeconds = extendedSessionIdleTimeout; + isExtendedSession = extendedSession; + extendedSessions = extendedSessionsCache?.GetOrInitializeCache(extendedSessionIdleTimeoutInSeconds); + } + + if (properties.TryGetValue("IncludeState", out object? includeStateObj) + && includeStateObj is bool includeState) + { + stateIncluded = includeState; + } + } +} diff --git a/src/Worker/Grpc/GrpcOrchestrationRunner.cs b/src/Worker/Grpc/GrpcOrchestrationRunner.cs index d8d04dea..77e1c5fd 100644 --- a/src/Worker/Grpc/GrpcOrchestrationRunner.cs +++ b/src/Worker/Grpc/GrpcOrchestrationRunner.cs @@ -131,32 +131,16 @@ public static string LoadAndRun( pair => ProtoUtils.ConvertValueToObject(pair.Value)); OrchestratorExecutionResult? result = null; - MemoryCache? extendedSessions = null; - // If any of the request parameters are malformed, we assume the default - extended sessions are not enabled and the orchestration history is attached bool addToExtendedSessions = false; bool requiresHistory = false; - bool pastEventsIncluded = true; - bool isExtendedSession = false; - double extendedSessionIdleTimeoutInSeconds = 0; - - // Only attempt to initialize the extended sessions cache if all the parameters are correctly specified - if (properties.TryGetValue("ExtendedSessionIdleTimeoutInSeconds", out object? extendedSessionIdleTimeoutObj) - && extendedSessionIdleTimeoutObj is double extendedSessionIdleTimeout - && extendedSessionIdleTimeout > 0 - && properties.TryGetValue("IsExtendedSession", out object? extendedSessionObj) - && extendedSessionObj is bool extendedSession) - { - extendedSessionIdleTimeoutInSeconds = extendedSessionIdleTimeout; - isExtendedSession = extendedSession; - extendedSessions = extendedSessionsCache?.GetOrInitializeCache(extendedSessionIdleTimeoutInSeconds); - } - - if (properties.TryGetValue("IncludePastEvents", out object? includePastEventsObj) - && includePastEventsObj is bool includePastEvents) - { - pastEventsIncluded = includePastEvents; - } + GrpcInstanceRunnerUtils.ParseRequestPropertiesAndInitializeCache( + properties, + extendedSessionsCache, + out double extendedSessionIdleTimeoutInSeconds, + out bool isExtendedSession, + out bool pastEventsIncluded, + out MemoryCache? extendedSessions); if (isExtendedSession && extendedSessions != null) { From e03691480e36e065a9984f5393848347b3fd2b7a Mon Sep 17 00:00:00 2001 From: Sophia Tevosyan Date: Fri, 14 Nov 2025 13:39:14 -0800 Subject: [PATCH 05/10] removed assumption that null entity state is invalid --- src/Worker/Grpc/GrpcEntityRunner.cs | 9 +- .../Grpc.Tests/GrpcEntityRunnerTests.cs | 113 +++++------------- .../GrpcOrchestrationRunnerTests.cs | 50 ++++---- 3 files changed, 60 insertions(+), 112 deletions(-) diff --git a/src/Worker/Grpc/GrpcEntityRunner.cs b/src/Worker/Grpc/GrpcEntityRunner.cs index db99c863..e5f6763e 100644 --- a/src/Worker/Grpc/GrpcEntityRunner.cs +++ b/src/Worker/Grpc/GrpcEntityRunner.cs @@ -99,6 +99,7 @@ public static async Task LoadAndRunAsync( TaskName entityName = new(id.Name); bool addToExtendedSessions = false; + bool stateCached = false; GrpcInstanceRunnerUtils.ParseRequestPropertiesAndInitializeCache( properties, extendedSessionsCache, @@ -112,13 +113,14 @@ public static async Task LoadAndRunAsync( addToExtendedSessions = true; // If an entity state was provided, even if we already have one stored, we always want to use the provided state. - if (!entityStateIncluded && extendedSessions.TryGetValue(request.InstanceId, out string? entityState) && entityState is not null) + if (!entityStateIncluded && extendedSessions.TryGetValue(request.InstanceId, out string? entityState)) { batch.EntityState = entityState; + stateCached = true; } } - if (batch.EntityState == null && !entityStateIncluded) + if (!stateCached && !entityStateIncluded) { // No state was provided, and we do not have one cached, so we cannot execute the batch request. return Convert.ToBase64String(new P.EntityBatchResult { RequiresState = true }.ToByteArray()); @@ -131,8 +133,7 @@ public static async Task LoadAndRunAsync( TaskEntity entity = factory.CreateEntity(entityName, implementation, id); EntityBatchResult result = await entity.ExecuteOperationBatchAsync(batch); - // An entity with a null state will be deleted, so we should not cache it. - if (addToExtendedSessions && result.EntityState is not null) + if (addToExtendedSessions) { extendedSessions.Set( request.InstanceId, diff --git a/test/Worker/Grpc.Tests/GrpcEntityRunnerTests.cs b/test/Worker/Grpc.Tests/GrpcEntityRunnerTests.cs index 89a73f5e..b63711d4 100644 --- a/test/Worker/Grpc.Tests/GrpcEntityRunnerTests.cs +++ b/test/Worker/Grpc.Tests/GrpcEntityRunnerTests.cs @@ -33,51 +33,14 @@ public async Task EmptyOrNullParameters_Throw_Exceptions() } [Fact] - public async Task NullStateStored_Means_StoredStateNotUsed() - { - using var extendedSessions = new ExtendedSessionsCache(); - extendedSessions.GetOrInitializeCache(DefaultExtendedSessionIdleTimeoutInSeconds).Set( - TestInstanceId, - null!, - new MemoryCacheEntryOptions { SlidingExpiration = TimeSpan.FromSeconds(DefaultExtendedSessionIdleTimeoutInSeconds) }); - - // No state, so response indicates that a state is required - Protobuf.EntityBatchRequest entityRequest = CreateEntityRequest([]); - entityRequest.Properties.Add(new MapField() { - { "IncludeEntityState", Value.ForBool(false) }, - { "IsExtendedSession", Value.ForBool(true) }, - { "ExtendedSessionIdleTimeoutInSeconds", Value.ForNumber(DefaultExtendedSessionIdleTimeoutInSeconds) } }); - byte[] requestBytes = entityRequest.ToByteArray(); - string requestString = Convert.ToBase64String(requestBytes); - string responseString = await GrpcEntityRunner.LoadAndRunAsync(requestString, new SimpleEntity(), extendedSessions); - Protobuf.EntityBatchResult response = Protobuf.EntityBatchResult.Parser.ParseFrom(Convert.FromBase64String(responseString)); - Assert.True(response.RequiresState); - - // State provided so request can be fulfilled, new state stored - entityRequest = CreateEntityRequest([setOperation]); - entityRequest.Properties.Add(new MapField() { - { "IncludeEntityState", Value.ForBool(true) }, - { "IsExtendedSession", Value.ForBool(true) }, - { "ExtendedSessionIdleTimeoutInSeconds", Value.ForNumber(DefaultExtendedSessionIdleTimeoutInSeconds) } }); - requestBytes = entityRequest.ToByteArray(); - requestString = Convert.ToBase64String(requestBytes); - responseString = await GrpcEntityRunner.LoadAndRunAsync(requestString, new SimpleEntity(), extendedSessions); - response = Protobuf.EntityBatchResult.Parser.ParseFrom(Convert.FromBase64String(responseString)); - Assert.False(response.RequiresState); - Assert.True(extendedSessions.GetOrInitializeCache(DefaultExtendedSessionIdleTimeoutInSeconds).TryGetValue(TestInstanceId, out object? extendedSession)); - Assert.Equal("1", extendedSession); - Assert.Equal("1", response.EntityState); - } - - [Fact] - public async Task EmptyState_Returns_NeedsStateInResponse() + public async Task EmptyHistory_Returns_NeedsHistoryInResponse_Async() { using var extendedSessions = new ExtendedSessionsCache(); // No state and without extended sessions enabled Protobuf.EntityBatchRequest entityRequest = CreateEntityRequest([]); entityRequest.Properties.Add(new MapField() { - { "IncludeEntityState", Value.ForBool(false) }}); + { "IncludeState", Value.ForBool(false) }}); byte[] requestBytes = entityRequest.ToByteArray(); string requestString = Convert.ToBase64String(requestBytes); string responseString = await GrpcEntityRunner.LoadAndRunAsync(requestString, new SimpleEntity(), extendedSessions); @@ -98,14 +61,14 @@ public async Task EmptyState_Returns_NeedsStateInResponse() } [Fact] - public async Task MalformedRequestParameters_Means_CacheNotInitialized() + public async Task MalformedRequestParameters_Means_CacheNotInitialized_Async() { using var extendedSessions = new ExtendedSessionsCache(); Protobuf.EntityBatchRequest entityRequest = CreateEntityRequest([]); // Misspelled extended session timeout key entityRequest.Properties.Add(new MapField() { - { "IncludeEntityState", Value.ForBool(false) }, + { "IncludeState", Value.ForBool(false) }, { "IsExtendedSession", Value.ForBool(true) }, { "ExtendedSessionsIdleTimeoutInSeconds", Value.ForNumber(DefaultExtendedSessionIdleTimeoutInSeconds) } }); byte[] requestBytes = entityRequest.ToByteArray(); @@ -117,7 +80,7 @@ public async Task MalformedRequestParameters_Means_CacheNotInitialized() // Wrong value type for extended session timeout key entityRequest.Properties.Clear(); entityRequest.Properties.Add(new MapField() { - { "IncludeEntityState", Value.ForBool(false) }, + { "IncludeState", Value.ForBool(false) }, { "IsExtendedSession", Value.ForBool(true) }, { "ExtendedSessionIdleTimeoutInSeconds", Value.ForString("hi") } }); requestBytes = entityRequest.ToByteArray(); @@ -128,7 +91,7 @@ public async Task MalformedRequestParameters_Means_CacheNotInitialized() // Invalid number for extended session timeout key (must be > 0) entityRequest.Properties.Clear(); entityRequest.Properties.Add(new MapField() { - { "IncludeEntityState", Value.ForBool(false) }, + { "IncludeState", Value.ForBool(false) }, { "IsExtendedSession", Value.ForBool(true) }, { "ExtendedSessionIdleTimeoutInSeconds", Value.ForNumber(0) } }); requestBytes = entityRequest.ToByteArray(); @@ -139,7 +102,7 @@ public async Task MalformedRequestParameters_Means_CacheNotInitialized() // Invalid number for extended session timeout key (must be > 0) entityRequest.Properties.Clear(); entityRequest.Properties.Add(new MapField() { - { "IncludeEntityState", Value.ForBool(false) }, + { "IncludeState", Value.ForBool(false) }, { "IsExtendedSession", Value.ForBool(true) }, { "ExtendedSessionIdleTimeoutInSeconds", Value.ForNumber(-1) } }); requestBytes = entityRequest.ToByteArray(); @@ -150,7 +113,7 @@ public async Task MalformedRequestParameters_Means_CacheNotInitialized() // No extended session timeout key entityRequest.Properties.Clear(); entityRequest.Properties.Add(new MapField() { - { "IncludeEntityState", Value.ForBool(false) }, + { "IncludeState", Value.ForBool(false) }, { "IsExtendedSession", Value.ForBool(true) } }); requestBytes = entityRequest.ToByteArray(); requestString = Convert.ToBase64String(requestBytes); @@ -160,7 +123,7 @@ public async Task MalformedRequestParameters_Means_CacheNotInitialized() // Misspelled extended session key entityRequest.Properties.Clear(); entityRequest.Properties.Add(new MapField() { - { "IncludeEntityState", Value.ForBool(false) }, + { "IncludeState", Value.ForBool(false) }, { "isExtendedSession", Value.ForBool(true) }, { "ExtendedSessionIdleTimeoutInSeconds", Value.ForNumber(DefaultExtendedSessionIdleTimeoutInSeconds) } }); requestBytes = entityRequest.ToByteArray(); @@ -171,7 +134,7 @@ public async Task MalformedRequestParameters_Means_CacheNotInitialized() // Wrong value type for extended session key entityRequest.Properties.Clear(); entityRequest.Properties.Add(new MapField() { - { "IncludeEntityState", Value.ForBool(false) }, + { "IncludeState", Value.ForBool(false) }, { "IsExtendedSession", Value.ForNumber(1) }, { "ExtendedSessionIdleTimeoutInSeconds", Value.ForNumber(DefaultExtendedSessionIdleTimeoutInSeconds) } }); requestBytes = entityRequest.ToByteArray(); @@ -182,7 +145,7 @@ public async Task MalformedRequestParameters_Means_CacheNotInitialized() // No extended session key entityRequest.Properties.Clear(); entityRequest.Properties.Add(new MapField() { - { "IncludeEntityState", Value.ForBool(false) }, + { "IncludeState", Value.ForBool(false) }, { "ExtendedSessionIdleTimeoutInSeconds", Value.ForNumber(DefaultExtendedSessionIdleTimeoutInSeconds) } }); requestBytes = entityRequest.ToByteArray(); requestString = Convert.ToBase64String(requestBytes); @@ -191,13 +154,13 @@ public async Task MalformedRequestParameters_Means_CacheNotInitialized() } [Fact] - public async Task IsExtendedSessionFalse_Means_NoExtendedSessionStored() + public async Task IsExtendedSessionFalse_Means_NoExtendedSessionStored_Async() { using var extendedSessions = new ExtendedSessionsCache(); Protobuf.EntityBatchRequest entityRequest = CreateEntityRequest([setOperation]); entityRequest.Properties.Add(new MapField() { - { "IncludeEntityState", Value.ForBool(false) }, + { "IncludeState", Value.ForBool(false) }, { "IsExtendedSession", Value.ForBool(false) }, { "ExtendedSessionIdleTimeoutInSeconds", Value.ForNumber(DefaultExtendedSessionIdleTimeoutInSeconds) } }); byte[] requestBytes = entityRequest.ToByteArray(); @@ -208,20 +171,20 @@ public async Task IsExtendedSessionFalse_Means_NoExtendedSessionStored() } /// - /// These tests verify that a malformed/nonexistent "IncludeEntityState" parameter means that the worker will attempt to + /// These tests verify that a malformed/nonexistent "IncludeState" parameter means that the worker will attempt to /// fulfill the entity request and not request a state for it. However, it is of course very undesirable that a - /// state is not attached to the request, but no state is requested by the worker due to a malformed "IncludeEntityState" parameter + /// state is not attached to the request, but no state is requested by the worker due to a malformed "IncludeState" parameter /// even when it needs one to fulfill the request. This would need to be checked on whatever side is calling this SDK. /// [Fact] - public async Task MalformedIncludeEntityStateParameter_Means_NoStateRequired() + public async Task MalformedIncludeStateParameter_Means_NoStateRequired_Async() { using var extendedSessions = new ExtendedSessionsCache(); Protobuf.EntityBatchRequest entityRequest = CreateEntityRequest([addOperation], entityState: 1.ToString()); // Misspelled include entity state key entityRequest.Properties.Add(new MapField() { - { "IncludeENtityState", Value.ForBool(false) }, + { "IncludeSTate", Value.ForBool(false) }, { "IsExtendedSession", Value.ForBool(false) }, { "ExtendedSessionIdleTimeoutInSeconds", Value.ForNumber(DefaultExtendedSessionIdleTimeoutInSeconds) } }); byte[] requestBytes = entityRequest.ToByteArray(); @@ -234,7 +197,7 @@ public async Task MalformedIncludeEntityStateParameter_Means_NoStateRequired() // Wrong value type for include entity state key entityRequest.Properties.Clear(); entityRequest.Properties.Add(new MapField() { - { "IncludeEntityState", Value.ForString("no") }, + { "IncludeState", Value.ForString("no") }, { "IsExtendedSession", Value.ForBool(false) }, { "ExtendedSessionIdleTimeoutInSeconds", Value.ForNumber(DefaultExtendedSessionIdleTimeoutInSeconds) } }); requestBytes = entityRequest.ToByteArray(); @@ -258,12 +221,12 @@ public async Task MalformedIncludeEntityStateParameter_Means_NoStateRequired() } [Fact] - public async Task Entity_State_Stored() + public async Task Entity_State_Stored_Async() { using var extendedSessions = new ExtendedSessionsCache(); Protobuf.EntityBatchRequest entityRequest = CreateEntityRequest([setOperation]); entityRequest.Properties.Add(new MapField() { - { "IncludeEntityState", Value.ForBool(true) }, + { "IncludeState", Value.ForBool(true) }, { "IsExtendedSession", Value.ForBool(true) }, { "ExtendedSessionIdleTimeoutInSeconds", Value.ForNumber(DefaultExtendedSessionIdleTimeoutInSeconds) } }); byte[] requestBytes = entityRequest.ToByteArray(); @@ -277,28 +240,12 @@ public async Task Entity_State_Stored() } [Fact] - public async Task Null_Entity_State_NotStored() - { - using var extendedSessions = new ExtendedSessionsCache(); - Protobuf.EntityBatchRequest entityRequest = CreateEntityRequest([new Protobuf.OperationRequest() { RequestId = Guid.NewGuid().ToString(), Operation = "Delete" }]); - entityRequest.Properties.Add(new MapField() { - { "IncludeEntityState", Value.ForBool(true) }, - { "IsExtendedSession", Value.ForBool(true) }, - { "ExtendedSessionIdleTimeoutInSeconds", Value.ForNumber(DefaultExtendedSessionIdleTimeoutInSeconds) } }); - byte[] requestBytes = entityRequest.ToByteArray(); - string requestString = Convert.ToBase64String(requestBytes); - await GrpcEntityRunner.LoadAndRunAsync(requestString, new SimpleEntity(), extendedSessions); - Assert.True(extendedSessions.IsInitialized); - Assert.False(extendedSessions.GetOrInitializeCache(DefaultExtendedSessionIdleTimeoutInSeconds).TryGetValue(TestInstanceId, out object? extendedSession)); - } - - [Fact] - public async Task ExternallyEndedExtendedSession_Evicted() + public async Task ExternallyEndedExtendedSession_Evicted_Async() { using var extendedSessions = new ExtendedSessionsCache(); Protobuf.EntityBatchRequest entityRequest = CreateEntityRequest([setOperation]); entityRequest.Properties.Add(new MapField() { - { "IncludeEntityState", Value.ForBool(true) }, + { "IncludeState", Value.ForBool(true) }, { "IsExtendedSession", Value.ForBool(true) }, { "ExtendedSessionIdleTimeoutInSeconds", Value.ForNumber(DefaultExtendedSessionIdleTimeoutInSeconds) } }); byte[] requestBytes = entityRequest.ToByteArray(); @@ -313,7 +260,7 @@ public async Task ExternallyEndedExtendedSession_Evicted() // Now set the extended session flag to false for this instance entityRequest.Properties.Clear(); entityRequest.Properties.Add(new MapField() { - { "IncludeEntityState", Value.ForBool(true) }, + { "IncludeState", Value.ForBool(true) }, { "IsExtendedSession", Value.ForBool(false) }, { "ExtendedSessionIdleTimeoutInSeconds", Value.ForNumber(DefaultExtendedSessionIdleTimeoutInSeconds) } }); requestBytes = entityRequest.ToByteArray(); @@ -330,7 +277,7 @@ public async Task Stale_ExtendedSessions_Evicted_Async() int extendedSessionIdleTimeout = 5; Protobuf.EntityBatchRequest entityRequest = CreateEntityRequest([setOperation]); entityRequest.Properties.Add(new MapField() { - { "IncludeEntityState", Value.ForBool(true) }, + { "IncludeState", Value.ForBool(true) }, { "IsExtendedSession", Value.ForBool(true) }, { "ExtendedSessionIdleTimeoutInSeconds", Value.ForNumber(extendedSessionIdleTimeout) } }); byte[] requestBytes = entityRequest.ToByteArray(); @@ -349,7 +296,7 @@ public async Task Stale_ExtendedSessions_Evicted_Async() // Now that the entry was evicted from the cache, the entity runner needs an entity state to complete the request entityRequest.Properties.Clear(); entityRequest.Properties.Add(new MapField() { - { "IncludeEntityState", Value.ForBool(false) }, + { "IncludeState", Value.ForBool(false) }, { "IsExtendedSession", Value.ForBool(true) }, { "ExtendedSessionIdleTimeoutInSeconds", Value.ForNumber(extendedSessionIdleTimeout) } }); requestBytes = entityRequest.ToByteArray(); @@ -360,12 +307,12 @@ public async Task Stale_ExtendedSessions_Evicted_Async() } [Fact] - public async Task EntityStateIncluded_Means_ExtendedSession_Evicted() + public async Task EntityStateIncluded_Means_ExtendedSession_Evicted_Async() { using var extendedSessions = new ExtendedSessionsCache(); Protobuf.EntityBatchRequest entityRequest = CreateEntityRequest([addOperation], entityState: 1.ToString()); entityRequest.Properties.Add(new MapField() { - { "IncludeEntityState", Value.ForBool(true) }, + { "IncludeState", Value.ForBool(true) }, { "IsExtendedSession", Value.ForBool(true) }, { "ExtendedSessionIdleTimeoutInSeconds", Value.ForNumber(DefaultExtendedSessionIdleTimeoutInSeconds) } }); byte[] requestBytes = entityRequest.ToByteArray(); @@ -377,7 +324,7 @@ public async Task EntityStateIncluded_Means_ExtendedSession_Evicted() // Now we will retry the same request, but with a different value for the state. If the extended session is not evicted, then the result will be incorrect. entityRequest = CreateEntityRequest([addOperation], entityState: 5.ToString()); entityRequest.Properties.Add(new MapField() { - { "IncludeEntityState", Value.ForBool(true) }, + { "IncludeState", Value.ForBool(true) }, { "IsExtendedSession", Value.ForBool(true) }, { "ExtendedSessionIdleTimeoutInSeconds", Value.ForNumber(DefaultExtendedSessionIdleTimeoutInSeconds) } }); requestBytes = entityRequest.ToByteArray(); @@ -390,14 +337,14 @@ public async Task EntityStateIncluded_Means_ExtendedSession_Evicted() } [Fact] - public async Task Null_ExtendedSessionsCache_IsOkay() + public async Task Null_ExtendedSessionsCache_IsOkay_Async() { Protobuf.EntityBatchRequest entityRequest = CreateEntityRequest([setOperation]); // Set up the parameters as if extended sessions are enabled, but do not pass an extended session cache to the request. // The request should still be successful. entityRequest.Properties.Add(new MapField() { - { "IncludeEntityState", Value.ForBool(true) }, + { "IncludeState", Value.ForBool(true) }, { "IsExtendedSession", Value.ForBool(true) }, { "ExtendedSessionIdleTimeoutInSeconds", Value.ForNumber(DefaultExtendedSessionIdleTimeoutInSeconds) } }); byte[] requestBytes = entityRequest.ToByteArray(); diff --git a/test/Worker/Grpc.Tests/GrpcOrchestrationRunnerTests.cs b/test/Worker/Grpc.Tests/GrpcOrchestrationRunnerTests.cs index 15d6e186..983deaef 100644 --- a/test/Worker/Grpc.Tests/GrpcOrchestrationRunnerTests.cs +++ b/test/Worker/Grpc.Tests/GrpcOrchestrationRunnerTests.cs @@ -38,7 +38,7 @@ public void EmptyHistory_Returns_NeedsHistoryInResponse() // No history and without extended sessions enabled Protobuf.OrchestratorRequest orchestratorRequest = CreateOrchestratorRequest([]); orchestratorRequest.Properties.Add(new MapField() { - { "IncludePastEvents", Value.ForBool(false) }}); + { "IncludeState", Value.ForBool(false) }}); byte[] requestBytes = orchestratorRequest.ToByteArray(); string requestString = Convert.ToBase64String(requestBytes); string responseString = GrpcOrchestrationRunner.LoadAndRun(requestString, new SimpleOrchestrator(), extendedSessions); @@ -70,7 +70,7 @@ public void NullExtendedSessionStored_Means_NeedsExtendedSessionNotUsed() // No history, so response indicates that a history is required Protobuf.OrchestratorRequest orchestratorRequest = CreateOrchestratorRequest([]); orchestratorRequest.Properties.Add(new MapField() { - { "IncludePastEvents", Value.ForBool(false) }, + { "IncludeState", Value.ForBool(false) }, { "IsExtendedSession", Value.ForBool(true) }, { "ExtendedSessionIdleTimeoutInSeconds", Value.ForNumber(DefaultExtendedSessionIdleTimeoutInSeconds) } }); byte[] requestBytes = orchestratorRequest.ToByteArray(); @@ -95,7 +95,7 @@ public void NullExtendedSessionStored_Means_NeedsExtendedSessionNotUsed() }; orchestratorRequest = CreateOrchestratorRequest([historyEvent]); orchestratorRequest.Properties.Add(new MapField() { - { "IncludePastEvents", Value.ForBool(true) }, + { "IncludeState", Value.ForBool(true) }, { "IsExtendedSession", Value.ForBool(true) }, { "ExtendedSessionIdleTimeoutInSeconds", Value.ForNumber(DefaultExtendedSessionIdleTimeoutInSeconds) } }); requestBytes = orchestratorRequest.ToByteArray(); @@ -115,7 +115,7 @@ public void MalformedRequestParameters_Means_CacheNotInitialized() // Misspelled extended session timeout key orchestratorRequest.Properties.Add(new MapField() { - { "IncludePastEvents", Value.ForBool(false) }, + { "IncludeState", Value.ForBool(false) }, { "IsExtendedSession", Value.ForBool(true) }, { "ExtendedSessionsIdleTimeoutInSeconds", Value.ForNumber(DefaultExtendedSessionIdleTimeoutInSeconds) } }); byte[] requestBytes = orchestratorRequest.ToByteArray(); @@ -127,7 +127,7 @@ public void MalformedRequestParameters_Means_CacheNotInitialized() // Wrong value type for extended session timeout key orchestratorRequest.Properties.Clear(); orchestratorRequest.Properties.Add(new MapField() { - { "IncludePastEvents", Value.ForBool(false) }, + { "IncludeState", Value.ForBool(false) }, { "IsExtendedSession", Value.ForBool(true) }, { "ExtendedSessionIdleTimeoutInSeconds", Value.ForString("hi") } }); requestBytes = orchestratorRequest.ToByteArray(); @@ -138,7 +138,7 @@ public void MalformedRequestParameters_Means_CacheNotInitialized() // Invalid number for extended session timeout key (must be > 0) orchestratorRequest.Properties.Clear(); orchestratorRequest.Properties.Add(new MapField() { - { "IncludePastEvents", Value.ForBool(false) }, + { "IncludeState", Value.ForBool(false) }, { "IsExtendedSession", Value.ForBool(true) }, { "ExtendedSessionIdleTimeoutInSeconds", Value.ForNumber(0) } }); requestBytes = orchestratorRequest.ToByteArray(); @@ -149,7 +149,7 @@ public void MalformedRequestParameters_Means_CacheNotInitialized() // Invalid number for extended session timeout key (must be > 0) orchestratorRequest.Properties.Clear(); orchestratorRequest.Properties.Add(new MapField() { - { "IncludePastEvents", Value.ForBool(false) }, + { "IncludeState", Value.ForBool(false) }, { "IsExtendedSession", Value.ForBool(true) }, { "ExtendedSessionIdleTimeoutInSeconds", Value.ForNumber(-1) } }); requestBytes = orchestratorRequest.ToByteArray(); @@ -160,7 +160,7 @@ public void MalformedRequestParameters_Means_CacheNotInitialized() // No extended session timeout key orchestratorRequest.Properties.Clear(); orchestratorRequest.Properties.Add(new MapField() { - { "IncludePastEvents", Value.ForBool(false) }, + { "IncludeState", Value.ForBool(false) }, { "IsExtendedSession", Value.ForBool(true) } }); requestBytes = orchestratorRequest.ToByteArray(); requestString = Convert.ToBase64String(requestBytes); @@ -170,7 +170,7 @@ public void MalformedRequestParameters_Means_CacheNotInitialized() // Misspelled extended session key orchestratorRequest.Properties.Clear(); orchestratorRequest.Properties.Add(new MapField() { - { "IncludePastEvents", Value.ForBool(false) }, + { "IncludeState", Value.ForBool(false) }, { "isExtendedSession", Value.ForBool(true) }, { "ExtendedSessionIdleTimeoutInSeconds", Value.ForNumber(DefaultExtendedSessionIdleTimeoutInSeconds) } }); requestBytes = orchestratorRequest.ToByteArray(); @@ -181,7 +181,7 @@ public void MalformedRequestParameters_Means_CacheNotInitialized() // Wrong value type for extended session key orchestratorRequest.Properties.Clear(); orchestratorRequest.Properties.Add(new MapField() { - { "IncludePastEvents", Value.ForBool(false) }, + { "IncludeState", Value.ForBool(false) }, { "IsExtendedSession", Value.ForNumber(1) }, { "ExtendedSessionIdleTimeoutInSeconds", Value.ForNumber(DefaultExtendedSessionIdleTimeoutInSeconds) } }); requestBytes = orchestratorRequest.ToByteArray(); @@ -192,7 +192,7 @@ public void MalformedRequestParameters_Means_CacheNotInitialized() // No extended session key orchestratorRequest.Properties.Clear(); orchestratorRequest.Properties.Add(new MapField() { - { "IncludePastEvents", Value.ForBool(false) }, + { "IncludeState", Value.ForBool(false) }, { "ExtendedSessionIdleTimeoutInSeconds", Value.ForNumber(DefaultExtendedSessionIdleTimeoutInSeconds) } }); requestBytes = orchestratorRequest.ToByteArray(); requestString = Convert.ToBase64String(requestBytes); @@ -207,7 +207,7 @@ public void IsExtendedSessionFalse_Means_NoExtendedSessionStored() Protobuf.OrchestratorRequest orchestratorRequest = CreateOrchestratorRequest([]); orchestratorRequest.Properties.Add(new MapField() { - { "IncludePastEvents", Value.ForBool(false) }, + { "IncludeState", Value.ForBool(false) }, { "IsExtendedSession", Value.ForBool(false) }, { "ExtendedSessionIdleTimeoutInSeconds", Value.ForNumber(DefaultExtendedSessionIdleTimeoutInSeconds) } }); byte[] requestBytes = orchestratorRequest.ToByteArray(); @@ -218,9 +218,9 @@ public void IsExtendedSessionFalse_Means_NoExtendedSessionStored() } /// - /// These tests verify that a malformed/nonexistent "IncludePastEvents" parameter means that the worker will attempt to + /// These tests verify that a malformed/nonexistent "IncludeState" parameter means that the worker will attempt to /// fulfill the orchestration request and not request a history for it. However, it is of course very undesirable that a - /// history is not attached to the request, but no history is requested by the worker due to a malformed "IncludePastEvents" parameter + /// history is not attached to the request, but no history is requested by the worker due to a malformed "IncludeState" parameter /// even when it needs one to fulfill the request. This would need to be checked on whatever side is calling this SDK. /// [Fact] @@ -244,7 +244,7 @@ public void MalformedPastEventsParameter_Means_NoHistoryRequired() // Misspelled include past events key orchestratorRequest.Properties.Add(new MapField() { - { "INcludePastEvents", Value.ForBool(false) }, + { "IncludeSTate", Value.ForBool(false) }, { "IsExtendedSession", Value.ForBool(false) }, { "ExtendedSessionIdleTimeoutInSeconds", Value.ForNumber(DefaultExtendedSessionIdleTimeoutInSeconds) } }); byte[] requestBytes = orchestratorRequest.ToByteArray(); @@ -256,7 +256,7 @@ public void MalformedPastEventsParameter_Means_NoHistoryRequired() // Wrong value type for include past events key orchestratorRequest.Properties.Clear(); orchestratorRequest.Properties.Add(new MapField() { - { "IncludePastEvents", Value.ForString("no") }, + { "IncludeState", Value.ForString("no") }, { "IsExtendedSession", Value.ForBool(false) }, { "ExtendedSessionIdleTimeoutInSeconds", Value.ForNumber(DefaultExtendedSessionIdleTimeoutInSeconds) } }); requestBytes = orchestratorRequest.ToByteArray(); @@ -296,7 +296,7 @@ public void Incomplete_Orchestration_Stored() }; Protobuf.OrchestratorRequest orchestratorRequest = CreateOrchestratorRequest([historyEvent]); orchestratorRequest.Properties.Add(new MapField() { - { "IncludePastEvents", Value.ForBool(true) }, + { "IncludeState", Value.ForBool(true) }, { "IsExtendedSession", Value.ForBool(true) }, { "ExtendedSessionIdleTimeoutInSeconds", Value.ForNumber(DefaultExtendedSessionIdleTimeoutInSeconds) } }); byte[] requestBytes = orchestratorRequest.ToByteArray(); @@ -325,7 +325,7 @@ public void Complete_Orchestration_NotStored() }; Protobuf.OrchestratorRequest orchestratorRequest = CreateOrchestratorRequest([historyEvent]); orchestratorRequest.Properties.Add(new MapField() { - { "IncludePastEvents", Value.ForBool(true) }, + { "IncludeState", Value.ForBool(true) }, { "IsExtendedSession", Value.ForBool(true) }, { "ExtendedSessionIdleTimeoutInSeconds", Value.ForNumber(DefaultExtendedSessionIdleTimeoutInSeconds) } }); byte[] requestBytes = orchestratorRequest.ToByteArray(); @@ -354,7 +354,7 @@ public void ExternallyEndedExtendedSession_Evicted() }; Protobuf.OrchestratorRequest orchestratorRequest = CreateOrchestratorRequest([historyEvent]); orchestratorRequest.Properties.Add(new MapField() { - { "IncludePastEvents", Value.ForBool(true) }, + { "IncludeState", Value.ForBool(true) }, { "IsExtendedSession", Value.ForBool(true) }, { "ExtendedSessionIdleTimeoutInSeconds", Value.ForNumber(DefaultExtendedSessionIdleTimeoutInSeconds) } }); byte[] requestBytes = orchestratorRequest.ToByteArray(); @@ -366,7 +366,7 @@ public void ExternallyEndedExtendedSession_Evicted() // Now set the extended session flag to false for this instance orchestratorRequest.Properties.Clear(); orchestratorRequest.Properties.Add(new MapField() { - { "IncludePastEvents", Value.ForBool(true) }, + { "IncludeState", Value.ForBool(true) }, { "IsExtendedSession", Value.ForBool(false) }, { "ExtendedSessionIdleTimeoutInSeconds", Value.ForNumber(DefaultExtendedSessionIdleTimeoutInSeconds) } }); requestBytes = orchestratorRequest.ToByteArray(); @@ -396,7 +396,7 @@ public async void Stale_ExtendedSessions_Evicted_Async() }; Protobuf.OrchestratorRequest orchestratorRequest = CreateOrchestratorRequest([historyEvent]); orchestratorRequest.Properties.Add(new MapField() { - { "IncludePastEvents", Value.ForBool(true) }, + { "IncludeState", Value.ForBool(true) }, { "IsExtendedSession", Value.ForBool(true) }, { "ExtendedSessionIdleTimeoutInSeconds", Value.ForNumber(extendedSessionIdleTimeout) } }); byte[] requestBytes = orchestratorRequest.ToByteArray(); @@ -412,7 +412,7 @@ public async void Stale_ExtendedSessions_Evicted_Async() // Now that the entry was evicted from the cache, the orchestration runner needs an orchestration history to complete the request orchestratorRequest.Properties.Clear(); orchestratorRequest.Properties.Add(new MapField() { - { "IncludePastEvents", Value.ForBool(false) }, + { "IncludeState", Value.ForBool(false) }, { "IsExtendedSession", Value.ForBool(true) }, { "ExtendedSessionIdleTimeoutInSeconds", Value.ForNumber(extendedSessionIdleTimeout) } }); requestBytes = orchestratorRequest.ToByteArray(); @@ -441,7 +441,7 @@ public void PastEventIncluded_Means_ExtendedSession_Evicted() }; Protobuf.OrchestratorRequest orchestratorRequest = CreateOrchestratorRequest([historyEvent]); orchestratorRequest.Properties.Add(new MapField() { - { "IncludePastEvents", Value.ForBool(true) }, + { "IncludeState", Value.ForBool(true) }, { "IsExtendedSession", Value.ForBool(true) }, { "ExtendedSessionIdleTimeoutInSeconds", Value.ForNumber(DefaultExtendedSessionIdleTimeoutInSeconds) } }); byte[] requestBytes = orchestratorRequest.ToByteArray(); @@ -451,7 +451,7 @@ public void PastEventIncluded_Means_ExtendedSession_Evicted() Assert.True(extendedSessions.GetOrInitializeCache(DefaultExtendedSessionIdleTimeoutInSeconds).TryGetValue(TestInstanceId, out object? extendedSession)); // Now we will retry the same exact request. If the extended session is not evicted, then the request will fail due to duplicate ExecutionStarted events being detected - // If the extended session is evicted because IncludePastEvents is true, then the request will succeed and a new extended session will be stored + // If the extended session is evicted because IncludeState is true, then the request will succeed and a new extended session will be stored GrpcOrchestrationRunner.LoadAndRun(requestString, new CallSubOrchestrationOrchestrator(), extendedSessions); Assert.True(extendedSessions.GetOrInitializeCache(DefaultExtendedSessionIdleTimeoutInSeconds).TryGetValue(TestInstanceId, out extendedSession)); } @@ -477,7 +477,7 @@ public void Null_ExtendedSessionsCache_IsOkay() // Set up the parameters as if extended sessions are enabled, but do not pass an extended session cache to the request. // The request should still be successful. orchestratorRequest.Properties.Add(new MapField() { - { "IncludePastEvents", Value.ForBool(true) }, + { "IncludeState", Value.ForBool(true) }, { "IsExtendedSession", Value.ForBool(true) }, { "ExtendedSessionIdleTimeoutInSeconds", Value.ForNumber(DefaultExtendedSessionIdleTimeoutInSeconds) } }); byte[] requestBytes = orchestratorRequest.ToByteArray(); From 2862d4d89737b924e4836289d9b0e8d6740e9f6c Mon Sep 17 00:00:00 2001 From: Sophia Tevosyan Date: Thu, 18 Dec 2025 11:29:02 -0800 Subject: [PATCH 06/10] addressing copilot comments --- src/Worker/Grpc/GrpcEntityRunner.cs | 2 +- src/Worker/Grpc/GrpcInstanceRunnerUtils.cs | 5 +---- src/Worker/Grpc/GrpcOrchestrationRunner.cs | 2 +- test/Worker/Grpc.Tests/GrpcEntityRunnerTests.cs | 5 ++--- test/Worker/Grpc.Tests/GrpcOrchestrationRunnerTests.cs | 3 +-- 5 files changed, 6 insertions(+), 11 deletions(-) diff --git a/src/Worker/Grpc/GrpcEntityRunner.cs b/src/Worker/Grpc/GrpcEntityRunner.cs index e5f6763e..9bce664f 100644 --- a/src/Worker/Grpc/GrpcEntityRunner.cs +++ b/src/Worker/Grpc/GrpcEntityRunner.cs @@ -135,7 +135,7 @@ public static async Task LoadAndRunAsync( if (addToExtendedSessions) { - extendedSessions.Set( + extendedSessions!.Set( request.InstanceId, result.EntityState, new MemoryCacheEntryOptions { SlidingExpiration = TimeSpan.FromSeconds(extendedSessionIdleTimeoutInSeconds) }); diff --git a/src/Worker/Grpc/GrpcInstanceRunnerUtils.cs b/src/Worker/Grpc/GrpcInstanceRunnerUtils.cs index 6d395bc8..abc2e2e2 100644 --- a/src/Worker/Grpc/GrpcInstanceRunnerUtils.cs +++ b/src/Worker/Grpc/GrpcInstanceRunnerUtils.cs @@ -1,9 +1,6 @@ // Copyright (c) Microsoft Corporation. // Licensed under the MIT License. -using System; -using System.Collections.Generic; -using System.Text; using Microsoft.Extensions.Caching.Memory; namespace Microsoft.DurableTask.Worker.Grpc; @@ -19,7 +16,7 @@ static class GrpcInstanceRunnerUtils /// /// /// If any request property is missing or invalid (i.e. the key is misspelled or the value is of the wrong type), - /// extended sessions are not enabled and default values are assigned are assigned to the returns. + /// extended sessions are not enabled and default values are assigned to the returns. /// /// /// A dictionary containing request properties used to configure extended session behavior. diff --git a/src/Worker/Grpc/GrpcOrchestrationRunner.cs b/src/Worker/Grpc/GrpcOrchestrationRunner.cs index 77e1c5fd..d0d422bd 100644 --- a/src/Worker/Grpc/GrpcOrchestrationRunner.cs +++ b/src/Worker/Grpc/GrpcOrchestrationRunner.cs @@ -207,7 +207,7 @@ public static string LoadAndRun( if (addToExtendedSessions && !executor.IsCompleted) { - extendedSessions.Set( + extendedSessions!.Set( request.InstanceId, new(runtimeState, shim, executor), new MemoryCacheEntryOptions { SlidingExpiration = TimeSpan.FromSeconds(extendedSessionIdleTimeoutInSeconds) }); diff --git a/test/Worker/Grpc.Tests/GrpcEntityRunnerTests.cs b/test/Worker/Grpc.Tests/GrpcEntityRunnerTests.cs index b63711d4..bfdb3718 100644 --- a/test/Worker/Grpc.Tests/GrpcEntityRunnerTests.cs +++ b/test/Worker/Grpc.Tests/GrpcEntityRunnerTests.cs @@ -73,8 +73,7 @@ public async Task MalformedRequestParameters_Means_CacheNotInitialized_Async() { "ExtendedSessionsIdleTimeoutInSeconds", Value.ForNumber(DefaultExtendedSessionIdleTimeoutInSeconds) } }); byte[] requestBytes = entityRequest.ToByteArray(); string requestString = Convert.ToBase64String(requestBytes); - string responseString = await GrpcEntityRunner.LoadAndRunAsync(requestString, new SimpleEntity(), extendedSessions); - Protobuf.OrchestratorResponse response = Protobuf.OrchestratorResponse.Parser.ParseFrom(Convert.FromBase64String(responseString)); + await GrpcEntityRunner.LoadAndRunAsync(requestString, new SimpleEntity(), extendedSessions); Assert.False(extendedSessions.IsInitialized); // Wrong value type for extended session timeout key @@ -373,7 +372,7 @@ static Protobuf.EntityBatchRequest CreateEntityRequest(IEnumerable + sealed class SimpleEntity : TaskEntity { public void Set(int value) { diff --git a/test/Worker/Grpc.Tests/GrpcOrchestrationRunnerTests.cs b/test/Worker/Grpc.Tests/GrpcOrchestrationRunnerTests.cs index 5e18f838..78dc0805 100644 --- a/test/Worker/Grpc.Tests/GrpcOrchestrationRunnerTests.cs +++ b/test/Worker/Grpc.Tests/GrpcOrchestrationRunnerTests.cs @@ -120,8 +120,7 @@ public void MalformedRequestParameters_Means_CacheNotInitialized() { "ExtendedSessionsIdleTimeoutInSeconds", Value.ForNumber(DefaultExtendedSessionIdleTimeoutInSeconds) } }); byte[] requestBytes = orchestratorRequest.ToByteArray(); string requestString = Convert.ToBase64String(requestBytes); - string responseString = GrpcOrchestrationRunner.LoadAndRun(requestString, new SimpleOrchestrator(), extendedSessions); - Protobuf.OrchestratorResponse response = Protobuf.OrchestratorResponse.Parser.ParseFrom(Convert.FromBase64String(responseString)); + GrpcOrchestrationRunner.LoadAndRun(requestString, new SimpleOrchestrator(), extendedSessions); Assert.False(extendedSessions.IsInitialized); // Wrong value type for extended session timeout key From 2c84986f28c87196000fce382310a38f778e1a6c Mon Sep 17 00:00:00 2001 From: Sophia Tevosyan Date: Thu, 18 Dec 2025 11:44:16 -0800 Subject: [PATCH 07/10] addressing more comments --- test/Worker/Grpc.Tests/GrpcEntityRunnerTests.cs | 2 +- test/Worker/Grpc.Tests/GrpcOrchestrationRunnerTests.cs | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/test/Worker/Grpc.Tests/GrpcEntityRunnerTests.cs b/test/Worker/Grpc.Tests/GrpcEntityRunnerTests.cs index bfdb3718..bec7228a 100644 --- a/test/Worker/Grpc.Tests/GrpcEntityRunnerTests.cs +++ b/test/Worker/Grpc.Tests/GrpcEntityRunnerTests.cs @@ -33,7 +33,7 @@ public async Task EmptyOrNullParameters_Throw_Exceptions() } [Fact] - public async Task EmptyHistory_Returns_NeedsHistoryInResponse_Async() + public async Task EmptyState_Returns_NeedsStateInResponse_Async() { using var extendedSessions = new ExtendedSessionsCache(); diff --git a/test/Worker/Grpc.Tests/GrpcOrchestrationRunnerTests.cs b/test/Worker/Grpc.Tests/GrpcOrchestrationRunnerTests.cs index 78dc0805..4d5d9cd9 100644 --- a/test/Worker/Grpc.Tests/GrpcOrchestrationRunnerTests.cs +++ b/test/Worker/Grpc.Tests/GrpcOrchestrationRunnerTests.cs @@ -59,7 +59,7 @@ public void EmptyHistory_Returns_NeedsHistoryInResponse() } [Fact] - public void NullExtendedSessionStored_Means_NeedsExtendedSessionNotUsed() + public void NullExtendedSessionStored_Means_ExtendedSessionNotUsed() { using var extendedSessions = new ExtendedSessionsCache(); extendedSessions.GetOrInitializeCache(DefaultExtendedSessionIdleTimeoutInSeconds).Set( From a33a32958328a8b777859d758aa4ab98621ffe70 Mon Sep 17 00:00:00 2001 From: Sophia Tevosyan Date: Thu, 18 Dec 2025 15:01:18 -0800 Subject: [PATCH 08/10] updated comments --- src/Worker/Grpc/GrpcEntityRunner.cs | 1 + src/Worker/Grpc/GrpcOrchestrationRunner.cs | 1 + 2 files changed, 2 insertions(+) diff --git a/src/Worker/Grpc/GrpcEntityRunner.cs b/src/Worker/Grpc/GrpcEntityRunner.cs index 9bce664f..28ebc67e 100644 --- a/src/Worker/Grpc/GrpcEntityRunner.cs +++ b/src/Worker/Grpc/GrpcEntityRunner.cs @@ -135,6 +135,7 @@ public static async Task LoadAndRunAsync( if (addToExtendedSessions) { + // addToExtendedSessions can only be set to true if extendedSessions is not null extendedSessions!.Set( request.InstanceId, result.EntityState, diff --git a/src/Worker/Grpc/GrpcOrchestrationRunner.cs b/src/Worker/Grpc/GrpcOrchestrationRunner.cs index d0d422bd..5fbe2228 100644 --- a/src/Worker/Grpc/GrpcOrchestrationRunner.cs +++ b/src/Worker/Grpc/GrpcOrchestrationRunner.cs @@ -207,6 +207,7 @@ public static string LoadAndRun( if (addToExtendedSessions && !executor.IsCompleted) { + // addToExtendedSessions can only be set to true if extendedSessions is not null extendedSessions!.Set( request.InstanceId, new(runtimeState, shim, executor), From f28bdeb9162267f568ccf4020310db916ac8b837 Mon Sep 17 00:00:00 2001 From: Sophia Tevosyan Date: Mon, 29 Dec 2025 14:14:19 -0800 Subject: [PATCH 09/10] pulled protos from main --- src/Grpc/orchestrator_service.proto | 3 +++ src/Grpc/versions.txt | 4 ++-- 2 files changed, 5 insertions(+), 2 deletions(-) diff --git a/src/Grpc/orchestrator_service.proto b/src/Grpc/orchestrator_service.proto index 196c88da..e36c4f01 100644 --- a/src/Grpc/orchestrator_service.proto +++ b/src/Grpc/orchestrator_service.proto @@ -599,6 +599,7 @@ message EntityBatchRequest { string instanceId = 1; google.protobuf.StringValue entityState = 2; repeated OperationRequest operations = 3; + map properties = 4; } message EntityBatchResult { @@ -608,6 +609,8 @@ message EntityBatchResult { TaskFailureDetails failureDetails = 4; string completionToken = 5; repeated OperationInfo operationInfos = 6; // used only with DTS + // Whether or not an entity state is required to complete the original EntityBatchRequest and none was provided. + bool requiresState = 7; } message EntityRequest { diff --git a/src/Grpc/versions.txt b/src/Grpc/versions.txt index 3b7d5121..ea0ccc96 100644 --- a/src/Grpc/versions.txt +++ b/src/Grpc/versions.txt @@ -1,2 +1,2 @@ -# The following files were downloaded from branch main at 2025-12-12 01:49:06 UTC -https://raw.githubusercontent.com/microsoft/durabletask-protobuf/b03c06dea21952dcf2a86551fd761b6b78c64d15/protos/orchestrator_service.proto +# The following files were downloaded from branch main at 2025-12-29 22:13:55 UTC +https://raw.githubusercontent.com/microsoft/durabletask-protobuf/b7e260ad7b84740a2ed5cb4600ce73bef702a979/protos/orchestrator_service.proto From 2286bf54340149591ddd62e387c70476e3938be7 Mon Sep 17 00:00:00 2001 From: Sophia Tevosyan Date: Mon, 29 Dec 2025 14:26:42 -0800 Subject: [PATCH 10/10] copilot comments --- test/Worker/Grpc.Tests/GrpcEntityRunnerTests.cs | 2 +- test/Worker/Grpc.Tests/GrpcOrchestrationRunnerTests.cs | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/test/Worker/Grpc.Tests/GrpcEntityRunnerTests.cs b/test/Worker/Grpc.Tests/GrpcEntityRunnerTests.cs index bec7228a..248b79d5 100644 --- a/test/Worker/Grpc.Tests/GrpcEntityRunnerTests.cs +++ b/test/Worker/Grpc.Tests/GrpcEntityRunnerTests.cs @@ -42,7 +42,7 @@ public async Task EmptyState_Returns_NeedsStateInResponse_Async() entityRequest.Properties.Add(new MapField() { { "IncludeState", Value.ForBool(false) }}); byte[] requestBytes = entityRequest.ToByteArray(); - string requestString = Convert.ToBase64String(requestBytes); + string requestString = Convert.ToBase64String(requestBytes); string responseString = await GrpcEntityRunner.LoadAndRunAsync(requestString, new SimpleEntity(), extendedSessions); Protobuf.EntityBatchResult response = Protobuf.EntityBatchResult.Parser.ParseFrom(Convert.FromBase64String(responseString)); Assert.True(response.RequiresState); diff --git a/test/Worker/Grpc.Tests/GrpcOrchestrationRunnerTests.cs b/test/Worker/Grpc.Tests/GrpcOrchestrationRunnerTests.cs index 4d5d9cd9..0bb6be24 100644 --- a/test/Worker/Grpc.Tests/GrpcOrchestrationRunnerTests.cs +++ b/test/Worker/Grpc.Tests/GrpcOrchestrationRunnerTests.cs @@ -40,7 +40,7 @@ public void EmptyHistory_Returns_NeedsHistoryInResponse() orchestratorRequest.Properties.Add(new MapField() { { "IncludeState", Value.ForBool(false) }}); byte[] requestBytes = orchestratorRequest.ToByteArray(); - string requestString = Convert.ToBase64String(requestBytes); + string requestString = Convert.ToBase64String(requestBytes); string responseString = GrpcOrchestrationRunner.LoadAndRun(requestString, new SimpleOrchestrator(), extendedSessions); Protobuf.OrchestratorResponse response = Protobuf.OrchestratorResponse.Parser.ParseFrom(Convert.FromBase64String(responseString)); Assert.True(response.RequiresHistory); @@ -223,7 +223,7 @@ public void IsExtendedSessionFalse_Means_NoExtendedSessionStored() /// even when it needs one to fulfill the request. This would need to be checked on whatever side is calling this SDK. /// [Fact] - public void MalformedPastEventsParameter_Means_NoHistoryRequired() + public void MalformedIncludeStateParameter_Means_NoHistoryRequired() { using var extendedSessions = new ExtendedSessionsCache(); var historyEvent = new Protobuf.HistoryEvent