From 59150bfdde013bff433a5a748ba0f9c4f74c2093 Mon Sep 17 00:00:00 2001 From: Reuben Bond Date: Wed, 16 Sep 2026 13:25:48 -0700 Subject: [PATCH 1/7] feat(journaling): add grain feature participants --- src/Orleans.Journaling/DurableGrain.cs | 13 +- .../IJournaledGrainParticipant.cs | 23 ++ .../Orleans.Journaling/Orleans.Journaling.cs | 5 + .../JournaledGrainParticipantTests.cs | 328 ++++++++++++++++++ 4 files changed, 368 insertions(+), 1 deletion(-) create mode 100644 src/Orleans.Journaling/IJournaledGrainParticipant.cs create mode 100644 test/Orleans.Journaling.Tests/JournaledGrainParticipantTests.cs diff --git a/src/Orleans.Journaling/DurableGrain.cs b/src/Orleans.Journaling/DurableGrain.cs index c403f36f37e..00699001b03 100644 --- a/src/Orleans.Journaling/DurableGrain.cs +++ b/src/Orleans.Journaling/DurableGrain.cs @@ -8,8 +8,14 @@ namespace Orleans.Journaling; public abstract class DurableGrain : Grain, IGrainBase { /// - /// Initializes a new instance of the class and associates its state manager with the grain lifecycle. + /// Initializes a new instance of the class, associates its state manager + /// with the grain lifecycle, and initializes its journaled feature participants. /// + /// + /// All registered services are constructed before their + /// initializers run in registration order. Initialization completes before journal recovery begins. + /// Participant construction and initialization failures propagate to the activation caller. + /// protected DurableGrain() { StateManager = ServiceProvider.GetRequiredService(); @@ -17,6 +23,11 @@ protected DurableGrain() { participant.Participate(((IGrainBase)this).GrainContext.ObservableLifecycle); } + + foreach (var feature in ServiceProvider.GetServices().ToArray()) + { + feature.Initialize(); + } } /// diff --git a/src/Orleans.Journaling/IJournaledGrainParticipant.cs b/src/Orleans.Journaling/IJournaledGrainParticipant.cs new file mode 100644 index 00000000000..13ffad91ab0 --- /dev/null +++ b/src/Orleans.Journaling/IJournaledGrainParticipant.cs @@ -0,0 +1,23 @@ +namespace Orleans.Journaling; + +/// +/// Initializes a feature which contributes journaled state to a grain activation. +/// +/// +/// Register implementations as grain-scoped services. The constructor +/// resolves all registered participants before invoking their methods +/// in registration order, once per activation. This makes each feature's state available when +/// the state manager recovers the journal during grain lifecycle setup. +/// +public interface IJournaledGrainParticipant +{ + /// + /// Materializes the feature's grain-scoped services before journal recovery begins. + /// + /// + /// Use this method to resolve and register the feature's journaled state. All participant + /// constructors have completed when this method runs. A participant construction or + /// initialization exception propagates to the activation caller and fails activation. + /// + void Initialize(); +} diff --git a/src/api/Orleans.Journaling/Orleans.Journaling.cs b/src/api/Orleans.Journaling/Orleans.Journaling.cs index c41714fc25e..93d6c12d1f0 100644 --- a/src/api/Orleans.Journaling/Orleans.Journaling.cs +++ b/src/api/Orleans.Journaling/Orleans.Journaling.cs @@ -215,6 +215,11 @@ public partial interface IDurableValue T? Value { get; set; } } + public partial interface IJournaledGrainParticipant + { + void Initialize(); + } + public partial interface IJournaledState { void AppendEntries(JournalStreamWriter writer); diff --git a/test/Orleans.Journaling.Tests/JournaledGrainParticipantTests.cs b/test/Orleans.Journaling.Tests/JournaledGrainParticipantTests.cs new file mode 100644 index 00000000000..bbeb550aba9 --- /dev/null +++ b/test/Orleans.Journaling.Tests/JournaledGrainParticipantTests.cs @@ -0,0 +1,328 @@ +using System.Collections.Concurrent; +using Microsoft.Extensions.DependencyInjection; +using Orleans.Core.Internal; +using Orleans.TestingHost; +using Xunit; + +namespace Orleans.Journaling.Tests; + +[TestCategory("BVT")] +[TestSuite("BVT")] +[TestProvider("None")] +[TestArea("Journaling")] +public sealed class JournaledGrainParticipantTests(JournaledGrainParticipantFixture fixture) + : IClassFixture +{ + private static readonly string[] SuccessfulActivationEvents = + [ + "first constructed", "second constructed", "failure constructed", + "first initialized", "second initialized", "failure initialized", + "before recovery", "activated" + ]; + + [Fact] + public async Task DurableGrain_InitializesComposedParticipantsExactlyOnceBeforeRecovery() + { + var grain = fixture.Client.GetGrain(Guid.NewGuid()); + + var activationId = await grain.GetActivationId(); + Assert.Equal(SuccessfulActivationEvents, Assert.Single(fixture.GetEvents(grain.GetGrainId()))); + Assert.Equal(new string?[] { null, null }, await grain.GetActivationValues()); + await grain.SetValues("one", "two"); + Assert.Equal(new string?[] { "one", "two" }, await grain.GetValues()); + Assert.Equal(SuccessfulActivationEvents, Assert.Single(fixture.GetEvents(grain.GetGrainId()))); + + await grain.Cast().DeactivateOnIdle(TestContext.Current.CancellationToken); + + Assert.NotEqual(activationId, await grain.GetActivationId()); + Assert.Equal(new string?[] { "one", "two" }, await grain.GetActivationValues()); + Assert.Equal(new string?[] { "one", "two" }, await grain.GetValues()); + var activations = fixture.GetEvents(grain.GetGrainId()); + Assert.Equal(2, activations.Length); + Assert.All(activations, events => Assert.Equal(SuccessfulActivationEvents, events)); + } + + [Fact] + public async Task DurableGrain_PropagatesParticipantConstructorFailure() + { + var grain = fixture.Client.GetGrain(Guid.NewGuid()); + + var exception = await Assert.ThrowsAnyAsync(() => grain.Ping()); + + Assert.Contains(ParticipantFailure.ConstructorMessage, exception.ToString(), StringComparison.Ordinal); + var activations = fixture.GetEvents(grain.GetGrainId()); + Assert.NotEmpty(activations); + Assert.All(activations, events => Assert.Equal( + ["first constructed", "second constructed", "failure constructed"], events)); + } + + [Fact] + public async Task DurableGrain_PropagatesParticipantInitializeFailure() + { + var grain = fixture.Client.GetGrain(Guid.NewGuid()); + + var exception = await Assert.ThrowsAnyAsync(() => grain.Ping()); + + Assert.Contains(ParticipantFailure.InitializeMessage, exception.ToString(), StringComparison.Ordinal); + var activations = fixture.GetEvents(grain.GetGrainId()); + Assert.NotEmpty(activations); + Assert.All(activations, events => Assert.Equal( + ["first constructed", "second constructed", "failure constructed", "first initialized"], events)); + } + + [Fact] + public async Task DurableGrain_InitializesParticipantsBeforePropagatingActivationFailure() + { + var grain = fixture.Client.GetGrain(Guid.NewGuid()); + + var exception = await Assert.ThrowsAnyAsync(() => grain.Ping()); + + Assert.Contains(ParticipantFailure.ActivationMessage, exception.ToString(), StringComparison.Ordinal); + var activations = fixture.GetEvents(grain.GetGrainId()); + Assert.NotEmpty(activations); + Assert.All(activations, events => Assert.Equal( + [.. SuccessfulActivationEvents[..^1], "activation failed"], events)); + } + + [Fact] + public async Task StateManagerFactory_RecoversExplicitJournalWithGrainParticipantsRegistered() + { + var services = fixture.Cluster.Silos.First().ServiceProvider; + var factory = services.GetRequiredService(); + var shared = services.GetRequiredService(); + var journalId = new JournalId($"participant-factory-{Guid.NewGuid():N}"); + var cancellationToken = TestContext.Current.CancellationToken; + + await using (var manager = factory.Create(journalId)) + { + var value = new DurableValue("value", manager, shared, services); + await manager.InitializeAsync(cancellationToken); + value.Value = "persisted outside a grain"; + await manager.WriteStateAsync(cancellationToken); + } + + await using (var manager = factory.Create(journalId)) + { + var value = new DurableValue("value", manager, shared, services); + await manager.InitializeAsync(cancellationToken); + Assert.Equal("persisted outside a grain", value.Value); + } + } +} + +public sealed class JournaledGrainParticipantFixture : IntegrationTestFixture +{ + protected override void ConfigureTestCluster(InProcessTestClusterBuilder builder) + { + builder.ConfigureSilo((_, siloBuilder) => + { + siloBuilder.Services.AddSingleton(); + siloBuilder.Services.AddScoped(); + siloBuilder.Services.AddScoped(); + siloBuilder.Services.AddScoped(); + }); + } + + public string[][] GetEvents(GrainId grainId) => + Cluster.Silos.SelectMany(silo => + silo.ServiceProvider.GetRequiredService().GetEvents(grainId)).ToArray(); +} + +internal sealed class FirstJournaledParticipant : IJournaledGrainParticipant +{ + private readonly IGrainContext _grainContext; + private readonly IServiceProvider _serviceProvider; + private readonly ParticipantRecorder _recorder; + + public FirstJournaledParticipant( + IGrainContext grainContext, + IServiceProvider serviceProvider, + ParticipantRecorder recorder) + { + _grainContext = grainContext; + _serviceProvider = serviceProvider; + _recorder = recorder; + recorder.Record(grainContext, "first constructed"); + } + + public void Initialize() + { + _recorder.Record(_grainContext, "first initialized"); + if (_grainContext.GrainId.Type.ToString() == InitializeFailureParticipantGrain.GrainTypeName) + { + throw new InvalidOperationException(ParticipantFailure.InitializeMessage); + } + + _ = _serviceProvider.GetRequiredKeyedService>("participant-one"); + } +} + +internal sealed class SecondJournaledParticipant : IJournaledGrainParticipant +{ + private readonly IGrainContext _grainContext; + private readonly IServiceProvider _serviceProvider; + private readonly ParticipantRecorder _recorder; + + public SecondJournaledParticipant( + IGrainContext grainContext, + IServiceProvider serviceProvider, + ParticipantRecorder recorder) + { + _grainContext = grainContext; + _serviceProvider = serviceProvider; + _recorder = recorder; + recorder.Record(grainContext, "second constructed"); + } + + public void Initialize() + { + _recorder.Record(_grainContext, "second initialized"); + _ = _serviceProvider.GetRequiredKeyedService>("participant-two"); + } +} + +internal sealed class ParticipantFailure : IJournaledGrainParticipant +{ + public const string ConstructorMessage = "Expected participant constructor failure."; + public const string InitializeMessage = "Expected participant initialization failure."; + public const string ActivationMessage = "Expected grain activation failure."; + private readonly IGrainContext _grainContext; + private readonly ParticipantRecorder _recorder; + + public ParticipantFailure(IGrainContext grainContext, ParticipantRecorder recorder) + { + _grainContext = grainContext; + _recorder = recorder; + recorder.Record(grainContext, "failure constructed"); + if (grainContext.GrainId.Type.ToString() == ConstructorFailureParticipantGrain.GrainTypeName) + { + throw new InvalidOperationException(ConstructorMessage); + } + + grainContext.ObservableLifecycle.Subscribe( + nameof(ParticipantFailure), + GrainLifecycleStage.SetupState - 1, + cancellationToken => + { + var manager = grainContext.ActivationServices.GetRequiredService(); + Assert.True(manager.TryGetState("participant-one", out _)); + Assert.True(manager.TryGetState("participant-two", out _)); + recorder.Record(grainContext, "before recovery"); + return Task.CompletedTask; + }); + } + + public void Initialize() => _recorder.Record(_grainContext, "failure initialized"); +} + +public sealed class ParticipantRecorder +{ + private readonly ConcurrentDictionary>> _events = new(); + + public void Record(IGrainContext grainContext, string value) => + _events.GetOrAdd(grainContext.GrainId, static _ => new()) + .GetOrAdd(grainContext.ActivationId, static _ => new()).Enqueue(value); + + public IEnumerable GetEvents(GrainId grainId) => + _events.TryGetValue(grainId, out var activations) + ? activations.Values.Select(events => events.ToArray()) + : []; +} + +[GrainType(GrainTypeName)] +public sealed class ComposedParticipantGrain : DurableGrain, IComposedParticipantGrain +{ + public const string GrainTypeName = "journaling-composed-participant"; + private readonly Guid _activationId = Guid.NewGuid(); + private string?[] _activationValues = []; + + public override Task OnActivateAsync(CancellationToken cancellationToken) + { + _activationValues = ReadValues(); + ServiceProvider.GetRequiredService().Record(GrainContext, "activated"); + return Task.CompletedTask; + } + + public async Task SetValues(string first, string second) + { + GetValue("participant-one").Value = first; + GetValue("participant-two").Value = second; + await WriteStateAsync(); + } + + public Task GetValues() => Task.FromResult(ReadValues()); + + public Task GetActivationValues() => Task.FromResult(_activationValues); + + private string?[] ReadValues() => [GetValue("participant-one").Value, GetValue("participant-two").Value]; + + public Task GetActivationId() => Task.FromResult(_activationId); + + private IDurableValue GetValue(string name) + { + AssertStateExists(name, StateManager.TryGetState(name, out var state)); + return state as IDurableValue + ?? throw new InvalidOperationException($"The participant state '{name}' has an unexpected type."); + + static void AssertStateExists(string name, bool exists) + { + if (!exists) + { + throw new InvalidOperationException($"The participant state '{name}' was not initialized."); + } + } + } +} + +[GrainType(GrainTypeName)] +public sealed class ConstructorFailureParticipantGrain : DurableGrain, IConstructorFailureParticipantGrain +{ + public const string GrainTypeName = "journaling-constructor-failure-participant"; + + public Task Ping() => Task.CompletedTask; +} + +[GrainType(GrainTypeName)] +public sealed class InitializeFailureParticipantGrain : DurableGrain, IInitializeFailureParticipantGrain +{ + public const string GrainTypeName = "journaling-initialize-failure-participant"; + + public Task Ping() => Task.CompletedTask; +} + +[GrainType(GrainTypeName)] +public sealed class ActivationFailureParticipantGrain : DurableGrain, IActivationFailureParticipantGrain +{ + public const string GrainTypeName = "journaling-activation-failure-participant"; + + public override Task OnActivateAsync(CancellationToken cancellationToken) + { + ServiceProvider.GetRequiredService().Record(GrainContext, "activation failed"); + return Task.FromException(new InvalidOperationException(ParticipantFailure.ActivationMessage)); + } + + public Task Ping() => Task.CompletedTask; +} + +public interface IComposedParticipantGrain : IGrainWithGuidKey +{ + Task SetValues(string first, string second); + Task GetValues(); + Task GetActivationValues(); + Task GetActivationId(); +} + +public interface IConstructorFailureParticipantGrain : IGrainWithGuidKey +{ + Task Ping(); +} + +public interface IInitializeFailureParticipantGrain : IGrainWithGuidKey +{ + Task Ping(); +} + +public interface IActivationFailureParticipantGrain : IGrainWithGuidKey +{ + Task Ping(); +} From c866a3e243b4b22274c0fb17433522caff0b3ba1 Mon Sep 17 00:00:00 2001 From: Reuben Bond Date: Thu, 17 Sep 2026 11:16:19 -0700 Subject: [PATCH 2/7] feat(journaling): support composition on plain grains --- src/Orleans.Journaling/DurableGrain.cs | 17 +- src/Orleans.Journaling/HostingExtensions.cs | 22 +- .../IJournaledGrainParticipant.cs | 23 - .../Orleans.Journaling/Orleans.Journaling.cs | 5 - .../JournaledGrainCompositionTests.cs | 608 ++++++++++++++++++ .../JournaledGrainParticipantTests.cs | 328 ---------- .../KeyedJournalingRegistrationTests.cs | 17 +- 7 files changed, 647 insertions(+), 373 deletions(-) delete mode 100644 src/Orleans.Journaling/IJournaledGrainParticipant.cs create mode 100644 test/Orleans.Journaling.Tests/JournaledGrainCompositionTests.cs delete mode 100644 test/Orleans.Journaling.Tests/JournaledGrainParticipantTests.cs diff --git a/src/Orleans.Journaling/DurableGrain.cs b/src/Orleans.Journaling/DurableGrain.cs index 00699001b03..06ef8e8a602 100644 --- a/src/Orleans.Journaling/DurableGrain.cs +++ b/src/Orleans.Journaling/DurableGrain.cs @@ -3,31 +3,24 @@ namespace Orleans.Journaling; /// -/// Provides a base class for grains which manage journaled durable state. +/// Provides convenience methods for grains which manage journaled durable state. /// public abstract class DurableGrain : Grain, IGrainBase { /// - /// Initializes a new instance of the class, associates its state manager - /// with the grain lifecycle, and initializes its journaled feature participants. + /// Initializes a new instance of the class and resolves its state manager. /// /// - /// All registered services are constructed before their - /// initializers run in registration order. Initialization completes before journal recovery begins. - /// Participant construction and initialization failures propagate to the activation caller. + /// The standard grain-scoped state manager enrolls in the lifecycle through its hosting factory. + /// This base class enrolls custom state managers which implement . /// protected DurableGrain() { StateManager = ServiceProvider.GetRequiredService(); - if (StateManager is ILifecycleParticipant participant) + if (StateManager is not JournaledStateManager && StateManager is ILifecycleParticipant participant) { participant.Participate(((IGrainBase)this).GrainContext.ObservableLifecycle); } - - foreach (var feature in ServiceProvider.GetServices().ToArray()) - { - feature.Initialize(); - } } /// diff --git a/src/Orleans.Journaling/HostingExtensions.cs b/src/Orleans.Journaling/HostingExtensions.cs index ef2b07fec51..3adf01a41a3 100644 --- a/src/Orleans.Journaling/HostingExtensions.cs +++ b/src/Orleans.Journaling/HostingExtensions.cs @@ -17,6 +17,11 @@ public static class HostingExtensions /// /// The silo builder. /// The silo builder. + /// + /// Resolving the standard grain-scoped enrolls it in the grain lifecycle. + /// Its registered durable states recover during , before grain activation. + /// Managers created through have caller-owned initialization and disposal. + /// public static ISiloBuilder AddJournalStorage(this ISiloBuilder builder) { builder.Services.AddOptions(); @@ -26,10 +31,23 @@ public static ISiloBuilder AddJournalStorage(this ISiloBuilder builder) : JournalingInstruments.CreateForDirectConstruction()); builder.Services.TryAddSingleton(); builder.Services.TryAddScoped(static services => - new JournaledStateManager( + { + var context = services.GetRequiredService(); + var manager = new JournaledStateManager( services.GetRequiredService(), services.GetRequiredService(), - services.GetRequiredService())); + context); + try + { + ((ILifecycleParticipant)manager).Participate(context.ObservableLifecycle); + return manager; + } + catch + { + ((IDisposable)manager).Dispose(); + throw; + } + }); builder.Services.TryAddSingleton(static services => services.GetKeyedService(ProviderConstants.DEFAULT_STORAGE_PROVIDER_NAME) ?? ActivatorUtilities.CreateInstance(services)); diff --git a/src/Orleans.Journaling/IJournaledGrainParticipant.cs b/src/Orleans.Journaling/IJournaledGrainParticipant.cs deleted file mode 100644 index 13ffad91ab0..00000000000 --- a/src/Orleans.Journaling/IJournaledGrainParticipant.cs +++ /dev/null @@ -1,23 +0,0 @@ -namespace Orleans.Journaling; - -/// -/// Initializes a feature which contributes journaled state to a grain activation. -/// -/// -/// Register implementations as grain-scoped services. The constructor -/// resolves all registered participants before invoking their methods -/// in registration order, once per activation. This makes each feature's state available when -/// the state manager recovers the journal during grain lifecycle setup. -/// -public interface IJournaledGrainParticipant -{ - /// - /// Materializes the feature's grain-scoped services before journal recovery begins. - /// - /// - /// Use this method to resolve and register the feature's journaled state. All participant - /// constructors have completed when this method runs. A participant construction or - /// initialization exception propagates to the activation caller and fails activation. - /// - void Initialize(); -} diff --git a/src/api/Orleans.Journaling/Orleans.Journaling.cs b/src/api/Orleans.Journaling/Orleans.Journaling.cs index 93d6c12d1f0..c41714fc25e 100644 --- a/src/api/Orleans.Journaling/Orleans.Journaling.cs +++ b/src/api/Orleans.Journaling/Orleans.Journaling.cs @@ -215,11 +215,6 @@ public partial interface IDurableValue T? Value { get; set; } } - public partial interface IJournaledGrainParticipant - { - void Initialize(); - } - public partial interface IJournaledState { void AppendEntries(JournalStreamWriter writer); diff --git a/test/Orleans.Journaling.Tests/JournaledGrainCompositionTests.cs b/test/Orleans.Journaling.Tests/JournaledGrainCompositionTests.cs new file mode 100644 index 00000000000..3f6627627f2 --- /dev/null +++ b/test/Orleans.Journaling.Tests/JournaledGrainCompositionTests.cs @@ -0,0 +1,608 @@ +using System.Buffers; +using System.Collections.Concurrent; +using System.Threading.Channels; +using Microsoft.Extensions.Configuration; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging.Abstractions; +using NSubstitute; +using Orleans.Journaling.Json; +using Orleans.Metadata; +using Orleans.Runtime; +using Orleans.TestingHost; +using Xunit; + +namespace Orleans.Journaling.Tests; + +[TestCategory("BVT"), TestSuite("BVT"), TestProvider("None"), TestArea("Journaling")] +public sealed class JournaledGrainCompositionTests(JournalCompositionFixture fixture) + : IClassFixture +{ + private static readonly TimeSpan Timeout = TimeSpan.FromSeconds(30); + private static CancellationToken Cancellation => TestContext.Current.CancellationToken; + + [Theory] + [InlineData(typeof(PlainJournalGrain))] + [InlineData(typeof(ApplicationJournalGrain))] + [InlineData(typeof(InjectedJournalGrain))] + [InlineData(typeof(ConvenienceJournalGrain))] + public async Task Composition_RecoversBeforeActivationAndIsolatesFreshScopes(Type grainClass) + { + var grain = fixture.GetGrain(grainClass); + Assert.Equal(new string?[] { null, null }, await grain.GetActivationValues()); + var first = await fixture.ReadProbe(grain); + Assert.Same(first.Context.GrainInstance, first.ConstructorInstance); + Assert.Equal(1, fixture.Storage.Get(first.Context.GrainId).Reads); + Assert.Equal(grainClass == typeof(InjectedJournalGrain) ? 0 : 1, first.SetupCount); + Assert.Equal(first.SetupCount, first.Feature?.ParticipationCount ?? 0); + Assert.True(first.Manager!.TryGetState("one", out var state)); + Assert.Same(first.First, state); + + await grain.SetValues("one", "two"); + Assert.Equal(new string?[] { "one", "two" }, await grain.GetValues()); + Assert.Equal(1, fixture.Storage.Get(first.Context.GrainId).Writes); + Assert.Equal(1, fixture.Storage.Get(first.Context.GrainId).Managers); + + var other = fixture.GetGrain(grainClass); + Assert.Equal(new string?[] { null, null }, await other.GetActivationValues()); + var isolated = await fixture.ReadProbe(other); + Assert.NotSame(first.Manager, isolated.Manager); + Assert.NotSame(first.First, isolated.First); + Assert.NotSame(first.Context.ActivationServices, isolated.Context.ActivationServices); + + await Deactivate(first); + Assert.Equal(1, first.Disposals); + Assert.Equal(first.SetupCount, first.Feature?.Disposals ?? 0); + await Assert.ThrowsAsync(() => first.Manager.InitializeAsync(Cancellation).AsTask()); + + Assert.Equal(new string?[] { "one", "two" }, await grain.GetActivationValues()); + Assert.Equal(new string?[] { "one", "two" }, await grain.GetValues()); + var replacement = await fixture.ReadProbe(grain); + Assert.NotSame(first.Manager, replacement.Manager); + Assert.NotSame(first.First, replacement.First); + Assert.Equal(first.SetupCount, replacement.SetupCount); + Assert.Equal(2, fixture.Storage.Get(first.Context.GrainId).Reads); + Assert.Equal(2, fixture.Storage.Get(first.Context.GrainId).Managers); + if (first.SetupCount == 1) + { + Assert.NotSame(first.Feature, replacement.Feature); + Assert.Equal(1, fixture.Registry.Configurations[first.Context.GrainId.Type]); + Assert.Equal( + new[] { "constructed", "setup", "feature constructed", "before recovery", "after recovery", "activated" }, + replacement.Events); + } + + await Deactivate(isolated); + await Deactivate(replacement); + } + + [Theory] + [InlineData(typeof(JournalConstructorFailureGrain), "constructor", 0)] + [InlineData(typeof(JournalFeatureConstructorFailureGrain), "feature constructor", 0)] + [InlineData(typeof(JournalSetupFailureGrain), "setup", 0)] + [InlineData(typeof(JournalActivationFailureGrain), "activation", 1)] + public async Task Composition_PropagatesFailuresAndDisposesScope(Type grainClass, string phase, int reads) + { + var grain = fixture.GetGrain(grainClass); + var call = grain.GetValues(); + var probe = await fixture.ReadProbe(grain); + var exception = await Assert.ThrowsAnyAsync(() => call.WaitAsync(Timeout, Cancellation)); + Assert.Contains($"Expected {phase} failure.", exception.ToString(), StringComparison.Ordinal); + await probe.Context.Deactivated.WaitAsync(Timeout, Cancellation); + Assert.Equal(1, probe.Disposals); + Assert.Equal(reads, fixture.Storage.Get(probe.Context.GrainId).Reads); + Assert.False(probe.Activated); + if (probe.Feature is { } feature) + { + Assert.Equal(1, feature.Disposals); + } + + await Assert.ThrowsAsync(() => probe.Manager!.InitializeAsync(Cancellation).AsTask()); + } + + [Fact] + public async Task UninvolvedGrain_CreatesNoJournalManagerOrFeature() + { + var grain = fixture.GetGrain(typeof(UninvolvedJournalGrain)); + Assert.Empty(await grain.GetValues()); + var probe = await fixture.ReadProbe(grain); + Assert.Null(probe.Manager); + Assert.Null(probe.Feature); + Assert.Equal(0, probe.SetupCount); + Assert.False(fixture.Storage.Contains(JournalId.FromGrainId(grain.GetGrainId()))); + await Deactivate(probe); + Assert.Equal(1, probe.Disposals); + } + + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task PlainGrain_TerminalWriteFailureReactivatesAndReplaysActualCommit(bool committed) + { + var grain = fixture.GetGrain(typeof(PlainJournalGrain)); + await grain.SetValues("old one", "old two"); + var first = await fixture.ReadProbe(grain); + var storage = fixture.Storage.Get(first.Context.GrainId); + storage.FailNextWrite = committed; + var error = await Assert.ThrowsAsync(() => grain.SetValues("new one", "new two")); + Assert.Equal(JournalCompositionStorage.FailureMessage, error.Message); + await first.Context.Deactivated.WaitAsync(Timeout, Cancellation); + Assert.Equal(1, first.Disposals); + var expected = committed ? new[] { "new one", "new two" } : new[] { "old one", "old two" }; + Assert.Equal(expected, await grain.GetActivationValues()); + Assert.Equal(expected, await grain.GetValues()); + var replacement = await fixture.ReadProbe(grain); + Assert.NotSame(first.Manager, replacement.Manager); + Assert.NotSame(first.First, replacement.First); + Assert.Equal(2, storage.Managers); + Assert.Equal(2, storage.Reads); + await Deactivate(replacement); + } + + [Fact] + public async Task ExplicitFactory_WithAmbientGrainContextKeepsManualIndependentLifecycle() + { + var grain = fixture.GetGrain(typeof(PlainJournalGrain)); + await grain.SetValues("grain one", "grain two"); + var probe = await fixture.ReadProbe(grain); + var id = new JournalId($"standalone/{Guid.NewGuid():N}"); + var factory = fixture.Services.GetRequiredService(); + var codec = fixture.Services.GetRequiredKeyedService>(JsonJournalExtensions.JournalFormatKey); + IJournaledStateManager manager; + RuntimeContext.SetExecutionContext(probe.Context, out var previous); + try + { + manager = factory.Create(id); + } + finally + { + RuntimeContext.SetExecutionContext(previous!, out _); + } + + await using (manager) + { + var value = new DurableValue("value", manager, codec); + Assert.Equal(0, fixture.Storage.Get(id).Reads); + await manager.InitializeAsync(Cancellation); + value.Value = "standalone"; + await manager.WriteStateAsync(Cancellation); + fixture.Storage.Get(id).FailNextWrite = false; + value.Value = "uncommitted"; + await Assert.ThrowsAsync(() => manager.WriteStateAsync(Cancellation).AsTask()); + Assert.False(probe.Context.Deactivated.IsCompleted); + Assert.Equal(new[] { "grain one", "grain two" }, await grain.GetValues()); + } + + await using (var recovered = factory.Create(id)) + { + var value = new DurableValue("value", recovered, codec); + await recovered.InitializeAsync(Cancellation); + Assert.Equal("standalone", value.Value); + } + + Assert.Equal(2, fixture.Storage.Get(id).Managers); + Assert.Equal(2, fixture.Storage.Get(id).Reads); + await Deactivate(probe); + } + + [Fact] + public async Task HostingFactory_EnrollsExactlyOnceAndDurableGrainRetainsHelpers() + { + var builder = CreateBuilder(); + var lifecycle = new CompositionTestLifecycle(); + var context = Substitute.For(); + context.GrainId.Returns(GrainId.Create("composition", "factory-enrollment")); + context.ObservableLifecycle.Returns(lifecycle); + builder.Services.AddScoped(_ => context); + await using var services = builder.Services.BuildServiceProvider(new ServiceProviderOptions { ValidateScopes = true }); + await using var scope = services.CreateAsyncScope(); + context.ActivationServices.Returns(scope.ServiceProvider); + RuntimeContext.SetExecutionContext(context, out var previous); + HelperJournalGrain grain; + try + { + grain = new HelperJournalGrain(); + } + finally + { + RuntimeContext.SetExecutionContext(previous!, out _); + } + + Assert.Equal(1, lifecycle.Subscriptions); + var value = grain.Value; + Assert.Same(value, grain.Value); + Assert.Same(grain.State, grain.State); + Assert.Same(scope.ServiceProvider.GetRequiredService(), grain.Manager); + await lifecycle.OnStart(Cancellation); + value.Value = "helper"; + await grain.Commit(); + await lifecycle.OnStop(Cancellation); + } + + [Fact] + public void DurableGrain_EnrollsCustomLifecycleParticipant() + { + var manager = Substitute.For>(); + var context = Substitute.For(); + var lifecycle = new CompositionTestLifecycle(); + using var services = new ServiceCollection().AddSingleton(manager).BuildServiceProvider(); + context.ActivationServices.Returns(services); + context.ObservableLifecycle.Returns(lifecycle); + RuntimeContext.SetExecutionContext(context, out var previous); + try + { + var grain = new HelperJournalGrain(); + Assert.Same(manager, grain.Manager); + } + finally + { + RuntimeContext.SetExecutionContext(previous!, out _); + } + + ((ILifecycleParticipant)manager).Received(1).Participate(lifecycle); + } + + [Fact] + public void HostingFactory_EnrollmentFailureDisposesManagerAndPreservesException() + { + var builder = CreateBuilder(); + builder.Services.AddSingleton(); + builder.Services.AddKeyedSingleton(JsonJournalExtensions.JournalFormatKey, + static (services, _) => services.GetRequiredService()); + var context = Substitute.For(); + context.GrainId.Returns(GrainId.Create("composition", "enrollment-failure")); + var lifecycle = Substitute.For(); + var expected = new InvalidOperationException("Expected enrollment failure."); + lifecycle.When(x => x.Subscribe(Arg.Any(), Arg.Any(), Arg.Any())) + .Do(_ => throw expected); + context.ObservableLifecycle.Returns(lifecycle); + builder.Services.AddScoped(_ => context); + using var services = builder.Services.BuildServiceProvider(); + using var scope = services.CreateScope(); + var error = Assert.Throws(() => scope.ServiceProvider.GetRequiredService()); + Assert.Same(expected, error); + var writer = Assert.Single(services.GetRequiredService().Writers); + Assert.Throws(() => writer.GetBuffer()); + } + + internal static async Task Deactivate(JournalCompositionProbe probe) + { + probe.Context.Deactivate(new(DeactivationReasonCode.ApplicationRequested, "Journaling composition test")); + await probe.Context.Deactivated.WaitAsync(Timeout, Cancellation); + } + + private static CompositionSiloBuilder CreateBuilder() + { + var builder = new CompositionSiloBuilder(); + builder.Services.AddLogging(); + builder.Services.AddSingleton(TimeProvider.System); + builder.Services.AddKeyedSingleton(KeyedService.AnyKey, static (services, _) => services.GetRequiredService()); + builder.AddVolatileJournalStorage().UseJsonJournalFormat(JournalingTestsJsonContext.Default); + return builder; + } + + private sealed class CompositionSiloBuilder : ISiloBuilder + { + public IServiceCollection Services { get; } = new ServiceCollection(); + public IConfiguration Configuration { get; } = new ConfigurationBuilder().Build(); + } + + private sealed class TrackingJournalFormat(JsonLinesJournalFormat inner) : IJournalFormat + { + public List Writers { get; } = []; + public string FormatKey => inner.FormatKey; + public string? MimeType => inner.MimeType; + public JournalBufferWriter CreateWriter() + { + var writer = inner.CreateWriter(); + Writers.Add(writer); + return writer; + } + public void Replay(JournalBufferReader input, JournalReplayContext context) => inner.Replay(input, context); + } + + private sealed class HelperJournalGrain : DurableGrain + { + public IJournaledStateManager Manager => StateManager; + public IDurableValue Value => ServiceProvider.GetRequiredKeyedService>("helper"); + public IJournaledState State => GetOrCreateState("helper-state", static _ => Substitute.For(), 0); + public ValueTask Commit() => WriteStateAsync(); + } +} + +internal sealed class CompositionTestLifecycle() : LifecycleSubject(NullLogger.Instance), IGrainLifecycle +{ + public int Subscriptions { get; private set; } + public override IDisposable Subscribe(string observerName, int stage, ILifecycleObserver observer) + { + Subscriptions++; + return base.Subscribe(observerName, stage, observer); + } + public void AddMigrationParticipant(IGrainMigrationParticipant participant) { } + public void RemoveMigrationParticipant(IGrainMigrationParticipant participant) { } +} + +public sealed class JournalCompositionFixture : IntegrationTestFixture +{ + public IServiceProvider Services => Cluster.Silos[0].ServiceProvider; + public JournalCompositionRegistry Registry => Services.GetRequiredService(); + public JournalCompositionStorageProvider Storage => Services.GetRequiredService(); + + protected override void ConfigureTestCluster(InProcessTestClusterBuilder builder) + { + builder.Options.InitialSilosCount = 1; + builder.ConfigureSilo((_, silo) => + { + silo.Services.AddSingleton(); + silo.Services.AddScoped(); + silo.Services.AddScoped(); + silo.Services.AddSingleton(); + silo.Services.AddSingleton(static services => services.GetRequiredService()); + silo.Services.AddSingleton(); + }); + } + + public IJournalCompositionGrain GetGrain(Type type) => Client.GetGrain(Guid.NewGuid(), type.FullName); + public async Task ReadProbe(IJournalCompositionGrain grain) => + await Registry.Get(grain.GetGrainId()).Reader.ReadAsync(TestContext.Current.CancellationToken) + .AsTask().WaitAsync(TimeSpan.FromSeconds(30), TestContext.Current.CancellationToken); +} + +public sealed class JournalCompositionRegistry +{ + private readonly ConcurrentDictionary> _probes = new(); + public ConcurrentDictionary Configurations { get; } = new(); + public Channel Get(GrainId id) => _probes.GetOrAdd(id, static _ => Channel.CreateUnbounded()); +} + +public sealed class JournalCompositionProbe : IDisposable +{ + public JournalCompositionProbe(IGrainContext context, JournalCompositionRegistry registry) + { + Context = context; + Assert.True(registry.Get(context.GrainId).Writer.TryWrite(this)); + } + + public IGrainContext Context { get; } + public IJournaledStateManager? Manager { get; set; } + public IDurableValue? First { get; set; } + public IDurableValue? Second { get; set; } + public JournalCompositionFeature? Feature { get; set; } + public object? ConstructorInstance { get; private set; } + public List Events { get; } = []; + public int SetupCount { get; set; } + public int Disposals { get; private set; } + public bool Activated { get; private set; } + public string?[] ActivationValues { get; private set; } = []; + public void Constructed(object instance) + { + ConstructorInstance = instance; + Events.Add("constructed"); + } + public Task Activate() + { + if (Context.GrainInstance is JournalActivationFailureGrain) + { + throw new InvalidOperationException("Expected activation failure."); + } + + ActivationValues = Read(); + Activated = true; + Events.Add("activated"); + return Task.CompletedTask; + } + public string?[] Read() => [First!.Value, Second!.Value]; + public async Task SetValues(string first, string second) + { + First!.Value = first; + Second!.Value = second; + await Manager!.WriteStateAsync(CancellationToken.None); + } + public void Dispose() => Disposals++; +} + +public interface IJournalFeatureGrain; + +internal sealed class JournalCompositionConfigurator(GrainClassMap map, JournalCompositionRegistry registry) : IConfigureGrainTypeComponents +{ + public void Configure(GrainType grainType, GrainProperties properties, GrainTypeSharedContext shared) + { + if (map.TryGetGrainClass(grainType, out var type) && typeof(IJournalFeatureGrain).IsAssignableFrom(type)) + { + registry.Configurations.AddOrUpdate(grainType, 1, static (_, count) => count + 1); + shared.AddActivationSetup(static context => + { + var probe = context.ActivationServices.GetRequiredService(); + Assert.Same(probe.ConstructorInstance, context.GrainInstance); + Assert.NotNull(context.GrainInstance); + probe.SetupCount++; + probe.Events.Add("setup"); + context.ActivationServices.GetRequiredService().Participate(context.ObservableLifecycle); + if (context.GrainInstance is JournalSetupFailureGrain) + { + throw new InvalidOperationException("Expected setup failure."); + } + }); + } + } +} + +public sealed class JournalCompositionFeature : ILifecycleParticipant, IDisposable +{ + private readonly JournalCompositionProbe _probe; + public JournalCompositionFeature(JournalCompositionProbe probe, IJournaledStateManager manager, + [FromKeyedServices("one")] IDurableValue first, + [FromKeyedServices("two")] IDurableValue second) + { + _probe = probe; + probe.Manager = manager; + probe.First = first; + probe.Second = second; + if (probe.Context.GrainInstance is JournalFeatureConstructorFailureGrain) + { + throw new InvalidOperationException("Expected feature constructor failure."); + } + probe.Feature = this; + probe.Events.Add("feature constructed"); + } + public int ParticipationCount { get; private set; } + public int Disposals { get; private set; } + public void Participate(IGrainLifecycle lifecycle) + { + ParticipationCount++; + lifecycle.Subscribe(GrainLifecycleStage.SetupState - 1, _ => + { + Assert.True(_probe.Manager!.TryGetState("one", out var first)); + Assert.True(_probe.Manager.TryGetState("two", out var second)); + Assert.Same(_probe.First, first); + Assert.Same(_probe.Second, second); + _probe.Events.Add("before recovery"); + return Task.CompletedTask; + }); + lifecycle.Subscribe(GrainLifecycleStage.SetupState + 1, _ => + { + _probe.Events.Add("after recovery"); + return Task.CompletedTask; + }); + } + public void Dispose() => Disposals++; +} + +public interface IJournalCompositionGrain : IGrainWithGuidKey +{ + Task SetValues(string first, string second); + Task GetValues(); + Task GetActivationValues(); +} + +public class PlainJournalGrain : Grain, IJournalCompositionGrain, IJournalFeatureGrain +{ + protected JournalCompositionProbe Probe { get; } + public PlainJournalGrain(JournalCompositionProbe probe) + { + Probe = probe; + probe.Constructed(this); + } + public override Task OnActivateAsync(CancellationToken cancellationToken) => Probe.Activate(); + public Task SetValues(string first, string second) => Probe.SetValues(first, second); + public Task GetValues() => Task.FromResult(Probe.Read()); + public Task GetActivationValues() => Task.FromResult(Probe.ActivationValues); +} + +public abstract class JournalingApplicationBase; + +public sealed class ApplicationJournalGrain : JournalingApplicationBase, IGrainBase, IJournalCompositionGrain, IJournalFeatureGrain +{ + private readonly JournalCompositionProbe _probe; + public ApplicationJournalGrain(JournalCompositionProbe probe) + { + _probe = probe; + probe.Constructed(this); + } + public IGrainContext GrainContext => _probe.Context; + public Task OnActivateAsync(CancellationToken cancellationToken) => _probe.Activate(); + public Task SetValues(string first, string second) => _probe.SetValues(first, second); + public Task GetValues() => Task.FromResult(_probe.Read()); + public Task GetActivationValues() => Task.FromResult(_probe.ActivationValues); +} + +public sealed class InjectedJournalGrain : Grain, IJournalCompositionGrain +{ + private readonly JournalCompositionProbe _probe; + public InjectedJournalGrain(JournalCompositionProbe probe, IJournaledStateManager manager, + [FromKeyedServices("one")] IDurableValue first, + [FromKeyedServices("two")] IDurableValue second) + { + _probe = probe; + probe.Manager = manager; + probe.First = first; + probe.Second = second; + probe.Constructed(this); + } + public override Task OnActivateAsync(CancellationToken cancellationToken) => _probe.Activate(); + public Task SetValues(string first, string second) => _probe.SetValues(first, second); + public Task GetValues() => Task.FromResult(_probe.Read()); + public Task GetActivationValues() => Task.FromResult(_probe.ActivationValues); +} + +public sealed class ConvenienceJournalGrain : DurableGrain, IJournalCompositionGrain, IJournalFeatureGrain +{ + private readonly JournalCompositionProbe _probe; + public ConvenienceJournalGrain(JournalCompositionProbe probe) + { + _probe = probe; + probe.Constructed(this); + } + public override Task OnActivateAsync(CancellationToken cancellationToken) => _probe.Activate(); + public Task SetValues(string first, string second) => _probe.SetValues(first, second); + public Task GetValues() => Task.FromResult(_probe.Read()); + public Task GetActivationValues() => Task.FromResult(_probe.ActivationValues); +} + +public sealed class JournalConstructorFailureGrain : PlainJournalGrain +{ + public JournalConstructorFailureGrain(JournalCompositionProbe probe, IJournaledStateManager manager) : base(probe) + { + probe.Manager = manager; + throw new InvalidOperationException("Expected constructor failure."); + } +} +public sealed class JournalFeatureConstructorFailureGrain(JournalCompositionProbe probe) : PlainJournalGrain(probe); +public sealed class JournalSetupFailureGrain(JournalCompositionProbe probe) : PlainJournalGrain(probe); +public sealed class JournalActivationFailureGrain(JournalCompositionProbe probe) : PlainJournalGrain(probe); + +public sealed class UninvolvedJournalGrain(JournalCompositionProbe probe) : Grain, IJournalCompositionGrain +{ + public Task SetValues(string first, string second) => throw new NotSupportedException(); + public Task GetValues() + { + probe.Constructed(this); + return Task.FromResult(Array.Empty()); + } + public Task GetActivationValues() => Task.FromResult(Array.Empty()); +} + +public sealed class JournalCompositionStorageProvider : IJournalStorageProvider +{ + private readonly ConcurrentDictionary _storage = new(); + public IJournalStorage CreateStorage(JournalId journalId) + { + var result = _storage.GetOrAdd(journalId, static _ => new JournalCompositionStorage()); + Interlocked.Increment(ref result.Managers); + return result; + } + public JournalCompositionStorage Get(GrainId id) => Get(JournalId.FromGrainId(id)); + public JournalCompositionStorage Get(JournalId id) => _storage[id]; + public bool Contains(JournalId id) => _storage.ContainsKey(id); +} + +public sealed class JournalCompositionStorage : IJournalStorage +{ + public const string FailureMessage = "Expected journal commit acknowledgement failure."; + private readonly VolatileJournalStorage _inner = new(JsonJournalExtensions.JournalFormatKey); + public int Managers; + public int Reads; + public int Writes; + public bool? FailNextWrite { get; set; } + public bool IsCompactionRequested => false; + public ValueTask ReadAsync(IJournalStorageConsumer consumer, CancellationToken cancellationToken) + { + Interlocked.Increment(ref Reads); + return _inner.ReadAsync(consumer, cancellationToken); + } + public async ValueTask AppendAsync(ReadOnlySequence value, CancellationToken cancellationToken) + { + Interlocked.Increment(ref Writes); + var failure = FailNextWrite; + FailNextWrite = null; + if (failure != false) + { + await _inner.AppendAsync(value, cancellationToken); + } + if (failure.HasValue) + { + throw new IOException(FailureMessage); + } + } + public ValueTask ReplaceAsync(ReadOnlySequence value, CancellationToken cancellationToken) => _inner.ReplaceAsync(value, cancellationToken); + public ValueTask DeleteAsync(CancellationToken cancellationToken) => _inner.DeleteAsync(cancellationToken); +} diff --git a/test/Orleans.Journaling.Tests/JournaledGrainParticipantTests.cs b/test/Orleans.Journaling.Tests/JournaledGrainParticipantTests.cs deleted file mode 100644 index bbeb550aba9..00000000000 --- a/test/Orleans.Journaling.Tests/JournaledGrainParticipantTests.cs +++ /dev/null @@ -1,328 +0,0 @@ -using System.Collections.Concurrent; -using Microsoft.Extensions.DependencyInjection; -using Orleans.Core.Internal; -using Orleans.TestingHost; -using Xunit; - -namespace Orleans.Journaling.Tests; - -[TestCategory("BVT")] -[TestSuite("BVT")] -[TestProvider("None")] -[TestArea("Journaling")] -public sealed class JournaledGrainParticipantTests(JournaledGrainParticipantFixture fixture) - : IClassFixture -{ - private static readonly string[] SuccessfulActivationEvents = - [ - "first constructed", "second constructed", "failure constructed", - "first initialized", "second initialized", "failure initialized", - "before recovery", "activated" - ]; - - [Fact] - public async Task DurableGrain_InitializesComposedParticipantsExactlyOnceBeforeRecovery() - { - var grain = fixture.Client.GetGrain(Guid.NewGuid()); - - var activationId = await grain.GetActivationId(); - Assert.Equal(SuccessfulActivationEvents, Assert.Single(fixture.GetEvents(grain.GetGrainId()))); - Assert.Equal(new string?[] { null, null }, await grain.GetActivationValues()); - await grain.SetValues("one", "two"); - Assert.Equal(new string?[] { "one", "two" }, await grain.GetValues()); - Assert.Equal(SuccessfulActivationEvents, Assert.Single(fixture.GetEvents(grain.GetGrainId()))); - - await grain.Cast().DeactivateOnIdle(TestContext.Current.CancellationToken); - - Assert.NotEqual(activationId, await grain.GetActivationId()); - Assert.Equal(new string?[] { "one", "two" }, await grain.GetActivationValues()); - Assert.Equal(new string?[] { "one", "two" }, await grain.GetValues()); - var activations = fixture.GetEvents(grain.GetGrainId()); - Assert.Equal(2, activations.Length); - Assert.All(activations, events => Assert.Equal(SuccessfulActivationEvents, events)); - } - - [Fact] - public async Task DurableGrain_PropagatesParticipantConstructorFailure() - { - var grain = fixture.Client.GetGrain(Guid.NewGuid()); - - var exception = await Assert.ThrowsAnyAsync(() => grain.Ping()); - - Assert.Contains(ParticipantFailure.ConstructorMessage, exception.ToString(), StringComparison.Ordinal); - var activations = fixture.GetEvents(grain.GetGrainId()); - Assert.NotEmpty(activations); - Assert.All(activations, events => Assert.Equal( - ["first constructed", "second constructed", "failure constructed"], events)); - } - - [Fact] - public async Task DurableGrain_PropagatesParticipantInitializeFailure() - { - var grain = fixture.Client.GetGrain(Guid.NewGuid()); - - var exception = await Assert.ThrowsAnyAsync(() => grain.Ping()); - - Assert.Contains(ParticipantFailure.InitializeMessage, exception.ToString(), StringComparison.Ordinal); - var activations = fixture.GetEvents(grain.GetGrainId()); - Assert.NotEmpty(activations); - Assert.All(activations, events => Assert.Equal( - ["first constructed", "second constructed", "failure constructed", "first initialized"], events)); - } - - [Fact] - public async Task DurableGrain_InitializesParticipantsBeforePropagatingActivationFailure() - { - var grain = fixture.Client.GetGrain(Guid.NewGuid()); - - var exception = await Assert.ThrowsAnyAsync(() => grain.Ping()); - - Assert.Contains(ParticipantFailure.ActivationMessage, exception.ToString(), StringComparison.Ordinal); - var activations = fixture.GetEvents(grain.GetGrainId()); - Assert.NotEmpty(activations); - Assert.All(activations, events => Assert.Equal( - [.. SuccessfulActivationEvents[..^1], "activation failed"], events)); - } - - [Fact] - public async Task StateManagerFactory_RecoversExplicitJournalWithGrainParticipantsRegistered() - { - var services = fixture.Cluster.Silos.First().ServiceProvider; - var factory = services.GetRequiredService(); - var shared = services.GetRequiredService(); - var journalId = new JournalId($"participant-factory-{Guid.NewGuid():N}"); - var cancellationToken = TestContext.Current.CancellationToken; - - await using (var manager = factory.Create(journalId)) - { - var value = new DurableValue("value", manager, shared, services); - await manager.InitializeAsync(cancellationToken); - value.Value = "persisted outside a grain"; - await manager.WriteStateAsync(cancellationToken); - } - - await using (var manager = factory.Create(journalId)) - { - var value = new DurableValue("value", manager, shared, services); - await manager.InitializeAsync(cancellationToken); - Assert.Equal("persisted outside a grain", value.Value); - } - } -} - -public sealed class JournaledGrainParticipantFixture : IntegrationTestFixture -{ - protected override void ConfigureTestCluster(InProcessTestClusterBuilder builder) - { - builder.ConfigureSilo((_, siloBuilder) => - { - siloBuilder.Services.AddSingleton(); - siloBuilder.Services.AddScoped(); - siloBuilder.Services.AddScoped(); - siloBuilder.Services.AddScoped(); - }); - } - - public string[][] GetEvents(GrainId grainId) => - Cluster.Silos.SelectMany(silo => - silo.ServiceProvider.GetRequiredService().GetEvents(grainId)).ToArray(); -} - -internal sealed class FirstJournaledParticipant : IJournaledGrainParticipant -{ - private readonly IGrainContext _grainContext; - private readonly IServiceProvider _serviceProvider; - private readonly ParticipantRecorder _recorder; - - public FirstJournaledParticipant( - IGrainContext grainContext, - IServiceProvider serviceProvider, - ParticipantRecorder recorder) - { - _grainContext = grainContext; - _serviceProvider = serviceProvider; - _recorder = recorder; - recorder.Record(grainContext, "first constructed"); - } - - public void Initialize() - { - _recorder.Record(_grainContext, "first initialized"); - if (_grainContext.GrainId.Type.ToString() == InitializeFailureParticipantGrain.GrainTypeName) - { - throw new InvalidOperationException(ParticipantFailure.InitializeMessage); - } - - _ = _serviceProvider.GetRequiredKeyedService>("participant-one"); - } -} - -internal sealed class SecondJournaledParticipant : IJournaledGrainParticipant -{ - private readonly IGrainContext _grainContext; - private readonly IServiceProvider _serviceProvider; - private readonly ParticipantRecorder _recorder; - - public SecondJournaledParticipant( - IGrainContext grainContext, - IServiceProvider serviceProvider, - ParticipantRecorder recorder) - { - _grainContext = grainContext; - _serviceProvider = serviceProvider; - _recorder = recorder; - recorder.Record(grainContext, "second constructed"); - } - - public void Initialize() - { - _recorder.Record(_grainContext, "second initialized"); - _ = _serviceProvider.GetRequiredKeyedService>("participant-two"); - } -} - -internal sealed class ParticipantFailure : IJournaledGrainParticipant -{ - public const string ConstructorMessage = "Expected participant constructor failure."; - public const string InitializeMessage = "Expected participant initialization failure."; - public const string ActivationMessage = "Expected grain activation failure."; - private readonly IGrainContext _grainContext; - private readonly ParticipantRecorder _recorder; - - public ParticipantFailure(IGrainContext grainContext, ParticipantRecorder recorder) - { - _grainContext = grainContext; - _recorder = recorder; - recorder.Record(grainContext, "failure constructed"); - if (grainContext.GrainId.Type.ToString() == ConstructorFailureParticipantGrain.GrainTypeName) - { - throw new InvalidOperationException(ConstructorMessage); - } - - grainContext.ObservableLifecycle.Subscribe( - nameof(ParticipantFailure), - GrainLifecycleStage.SetupState - 1, - cancellationToken => - { - var manager = grainContext.ActivationServices.GetRequiredService(); - Assert.True(manager.TryGetState("participant-one", out _)); - Assert.True(manager.TryGetState("participant-two", out _)); - recorder.Record(grainContext, "before recovery"); - return Task.CompletedTask; - }); - } - - public void Initialize() => _recorder.Record(_grainContext, "failure initialized"); -} - -public sealed class ParticipantRecorder -{ - private readonly ConcurrentDictionary>> _events = new(); - - public void Record(IGrainContext grainContext, string value) => - _events.GetOrAdd(grainContext.GrainId, static _ => new()) - .GetOrAdd(grainContext.ActivationId, static _ => new()).Enqueue(value); - - public IEnumerable GetEvents(GrainId grainId) => - _events.TryGetValue(grainId, out var activations) - ? activations.Values.Select(events => events.ToArray()) - : []; -} - -[GrainType(GrainTypeName)] -public sealed class ComposedParticipantGrain : DurableGrain, IComposedParticipantGrain -{ - public const string GrainTypeName = "journaling-composed-participant"; - private readonly Guid _activationId = Guid.NewGuid(); - private string?[] _activationValues = []; - - public override Task OnActivateAsync(CancellationToken cancellationToken) - { - _activationValues = ReadValues(); - ServiceProvider.GetRequiredService().Record(GrainContext, "activated"); - return Task.CompletedTask; - } - - public async Task SetValues(string first, string second) - { - GetValue("participant-one").Value = first; - GetValue("participant-two").Value = second; - await WriteStateAsync(); - } - - public Task GetValues() => Task.FromResult(ReadValues()); - - public Task GetActivationValues() => Task.FromResult(_activationValues); - - private string?[] ReadValues() => [GetValue("participant-one").Value, GetValue("participant-two").Value]; - - public Task GetActivationId() => Task.FromResult(_activationId); - - private IDurableValue GetValue(string name) - { - AssertStateExists(name, StateManager.TryGetState(name, out var state)); - return state as IDurableValue - ?? throw new InvalidOperationException($"The participant state '{name}' has an unexpected type."); - - static void AssertStateExists(string name, bool exists) - { - if (!exists) - { - throw new InvalidOperationException($"The participant state '{name}' was not initialized."); - } - } - } -} - -[GrainType(GrainTypeName)] -public sealed class ConstructorFailureParticipantGrain : DurableGrain, IConstructorFailureParticipantGrain -{ - public const string GrainTypeName = "journaling-constructor-failure-participant"; - - public Task Ping() => Task.CompletedTask; -} - -[GrainType(GrainTypeName)] -public sealed class InitializeFailureParticipantGrain : DurableGrain, IInitializeFailureParticipantGrain -{ - public const string GrainTypeName = "journaling-initialize-failure-participant"; - - public Task Ping() => Task.CompletedTask; -} - -[GrainType(GrainTypeName)] -public sealed class ActivationFailureParticipantGrain : DurableGrain, IActivationFailureParticipantGrain -{ - public const string GrainTypeName = "journaling-activation-failure-participant"; - - public override Task OnActivateAsync(CancellationToken cancellationToken) - { - ServiceProvider.GetRequiredService().Record(GrainContext, "activation failed"); - return Task.FromException(new InvalidOperationException(ParticipantFailure.ActivationMessage)); - } - - public Task Ping() => Task.CompletedTask; -} - -public interface IComposedParticipantGrain : IGrainWithGuidKey -{ - Task SetValues(string first, string second); - Task GetValues(); - Task GetActivationValues(); - Task GetActivationId(); -} - -public interface IConstructorFailureParticipantGrain : IGrainWithGuidKey -{ - Task Ping(); -} - -public interface IInitializeFailureParticipantGrain : IGrainWithGuidKey -{ - Task Ping(); -} - -public interface IActivationFailureParticipantGrain : IGrainWithGuidKey -{ - Task Ping(); -} diff --git a/test/Orleans.Journaling.Tests/KeyedJournalingRegistrationTests.cs b/test/Orleans.Journaling.Tests/KeyedJournalingRegistrationTests.cs index f4d501db160..3b7de5c9843 100644 --- a/test/Orleans.Journaling.Tests/KeyedJournalingRegistrationTests.cs +++ b/test/Orleans.Journaling.Tests/KeyedJournalingRegistrationTests.cs @@ -36,7 +36,14 @@ public async Task DefaultProviderOverride_GrainAndFactoryManagersShareStorageAnd var replacement = new LifecycleJournalStorageProvider(); var grainId = GrainId.Create("test-grain", "default-override"); var journalId = JournalId.FromGrainId(grainId); - builder.Services.AddScoped(_ => new JournalBatchTests.TestGrainContext(grainId)); + builder.Services.AddScoped(); + builder.Services.AddScoped(services => + { + var context = Substitute.For(); + context.GrainId.Returns(grainId); + context.ObservableLifecycle.Returns(services.GetRequiredService()); + return context; + }); if (registerOverrideFirst) { builder.Services.AddSingleton(replacement); @@ -55,7 +62,9 @@ public async Task DefaultProviderOverride_GrainAndFactoryManagersShareStorageAnd { var manager = scope.ServiceProvider.GetRequiredService(); var value = scope.ServiceProvider.GetRequiredKeyedService>("value"); - await manager.InitializeAsync(token); + var grainLifecycle = scope.ServiceProvider.GetRequiredService(); + Assert.Equal(1, grainLifecycle.Subscriptions); + await grainLifecycle.OnStart(token); value.Value = 42; await manager.WriteStateAsync(token); } @@ -77,7 +86,9 @@ public async Task DefaultProviderOverride_GrainAndFactoryManagersShareStorageAnd { var manager = scope.ServiceProvider.GetRequiredService(); var value = scope.ServiceProvider.GetRequiredKeyedService>("value"); - await manager.InitializeAsync(token); + var grainLifecycle = scope.ServiceProvider.GetRequiredService(); + Assert.Equal(1, grainLifecycle.Subscriptions); + await grainLifecycle.OnStart(token); Assert.Equal(43, value.Value); } From 95aae345b67286e19d2943d76d332269b6789da7 Mon Sep 17 00:00:00 2001 From: Reuben Bond Date: Thu, 17 Sep 2026 11:27:49 -0700 Subject: [PATCH 3/7] fix(journaling): preserve explicitly supplied manager enrollment --- src/Orleans.Journaling/DurableGrain.cs | 6 +++-- .../JournaledStateManager.cs | 8 ++++++- .../JournaledGrainCompositionTests.cs | 24 ++++++++++++++++--- 3 files changed, 32 insertions(+), 6 deletions(-) diff --git a/src/Orleans.Journaling/DurableGrain.cs b/src/Orleans.Journaling/DurableGrain.cs index 06ef8e8a602..d65cb0bedd8 100644 --- a/src/Orleans.Journaling/DurableGrain.cs +++ b/src/Orleans.Journaling/DurableGrain.cs @@ -12,12 +12,14 @@ public abstract class DurableGrain : Grain, IGrainBase /// /// /// The standard grain-scoped state manager enrolls in the lifecycle through its hosting factory. - /// This base class enrolls custom state managers which implement . + /// This base class enrolls explicitly supplied state managers which implement + /// , including factory-created managers. /// protected DurableGrain() { StateManager = ServiceProvider.GetRequiredService(); - if (StateManager is not JournaledStateManager && StateManager is ILifecycleParticipant participant) + if (StateManager is not JournaledStateManager { IsLifecycleEnrolled: true } + && StateManager is ILifecycleParticipant participant) { participant.Participate(((IGrainBase)this).GrainContext.ObservableLifecycle); } diff --git a/src/Orleans.Journaling/JournaledStateManager.cs b/src/Orleans.Journaling/JournaledStateManager.cs index 8c30a7811e3..81636d19794 100644 --- a/src/Orleans.Journaling/JournaledStateManager.cs +++ b/src/Orleans.Journaling/JournaledStateManager.cs @@ -88,6 +88,8 @@ private static IJournalStorage CreateStorage(IJournalStorageProvider storageProv internal IServiceProvider ServiceProvider => _shared.ServiceProvider; + internal bool IsLifecycleEnrolled { get; private set; } + public void RegisterState(string name, IJournaledState state) { ArgumentNullException.ThrowIfNullOrEmpty(name); @@ -992,7 +994,11 @@ private void BindState(string name, uint id) public long PendingWriteByteCount => _journalWriter.CommittedLength; - void ILifecycleParticipant.Participate(IGrainLifecycle observer) => observer.Subscribe(GrainLifecycleStage.SetupState, this); + void ILifecycleParticipant.Participate(IGrainLifecycle observer) + { + observer.Subscribe(GrainLifecycleStage.SetupState, this); + IsLifecycleEnrolled = true; + } Task ILifecycleObserver.OnStart(CancellationToken cancellationToken) => InitializeAsync(cancellationToken).AsTask(); async Task ILifecycleObserver.OnStop(CancellationToken cancellationToken) => await StopAsync(cancellationToken).ConfigureAwait(false); diff --git a/test/Orleans.Journaling.Tests/JournaledGrainCompositionTests.cs b/test/Orleans.Journaling.Tests/JournaledGrainCompositionTests.cs index 3f6627627f2..90f0041916c 100644 --- a/test/Orleans.Journaling.Tests/JournaledGrainCompositionTests.cs +++ b/test/Orleans.Journaling.Tests/JournaledGrainCompositionTests.cs @@ -184,8 +184,10 @@ public async Task ExplicitFactory_WithAmbientGrainContextKeepsManualIndependentL await Deactivate(probe); } - [Fact] - public async Task HostingFactory_EnrollsExactlyOnceAndDurableGrainRetainsHelpers() + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task HostingFactory_EnrollsExactlyOnceAndDurableGrainRetainsHelpers(bool overrideWithExplicitFactory) { var builder = CreateBuilder(); var lifecycle = new CompositionTestLifecycle(); @@ -193,9 +195,19 @@ public async Task HostingFactory_EnrollsExactlyOnceAndDurableGrainRetainsHelpers context.GrainId.Returns(GrainId.Create("composition", "factory-enrollment")); context.ObservableLifecycle.Returns(lifecycle); builder.Services.AddScoped(_ => context); + var journalId = overrideWithExplicitFactory + ? new JournalId("explicit/scoped-override") + : JournalId.FromGrainId(context.GrainId); + if (overrideWithExplicitFactory) + { + builder.Services.AddScoped(services => services.GetRequiredService().Create(journalId)); + } + await using var services = builder.Services.BuildServiceProvider(new ServiceProviderOptions { ValidateScopes = true }); await using var scope = services.CreateAsyncScope(); context.ActivationServices.Returns(scope.ServiceProvider); + var scopedManager = scope.ServiceProvider.GetRequiredService(); + Assert.Equal(overrideWithExplicitFactory ? 0 : 1, lifecycle.Subscriptions); RuntimeContext.SetExecutionContext(context, out var previous); HelperJournalGrain grain; try @@ -211,11 +223,17 @@ public async Task HostingFactory_EnrollsExactlyOnceAndDurableGrainRetainsHelpers var value = grain.Value; Assert.Same(value, grain.Value); Assert.Same(grain.State, grain.State); - Assert.Same(scope.ServiceProvider.GetRequiredService(), grain.Manager); + Assert.Same(scopedManager, grain.Manager); await lifecycle.OnStart(Cancellation); value.Value = "helper"; await grain.Commit(); await lifecycle.OnStop(Cancellation); + + await using var recovered = services.GetRequiredService().Create(journalId); + var codec = services.GetRequiredKeyedService>(JsonJournalExtensions.JournalFormatKey); + var recoveredValue = new DurableValue("helper", recovered, codec); + await recovered.InitializeAsync(Cancellation); + Assert.Equal("helper", recoveredValue.Value); } [Fact] From 15be673aaa396621f2be41a442b8e4025133e019 Mon Sep 17 00:00:00 2001 From: Reuben Bond Date: Thu, 17 Sep 2026 11:27:49 -0700 Subject: [PATCH 4/7] docs(journaling): demonstrate plain grain composition --- .../docs/grains/journaling/configuration.md | 14 ++- .../docs/grains/journaling/durable-state.md | 30 +++++-- .../content/docs/grains/journaling/index.md | 15 ++-- .../grains/journaling/runtime-behavior.md | 26 ++++-- .../CompiledDocumentationSnippets.csproj | 1 + .../compiled/Grains/JournalingSnippets.cs | 87 +++++++++++++++++++ src/Orleans.Journaling/README.md | 12 ++- 7 files changed, 164 insertions(+), 21 deletions(-) create mode 100644 docs/site/src/content/docs/snippets/compiled/Grains/JournalingSnippets.cs diff --git a/docs/site/src/content/docs/grains/journaling/configuration.md b/docs/site/src/content/docs/grains/journaling/configuration.md index f7b1591b3c8..0568b7a8889 100644 --- a/docs/site/src/content/docs/grains/journaling/configuration.md +++ b/docs/site/src/content/docs/grains/journaling/configuration.md @@ -7,10 +7,20 @@ ms.topic: how-to # Configure Orleans Journaling -Configure a default journal storage provider on every silo that can activate a . Additional named providers give journal consumers independent storage namespaces alongside default grain journaling. Provider registration also adds the core Journaling services, durable-state keyed services, JSON format, and Orleans binary reader. +Configure a default journal storage provider on every silo that hosts grains or activation-scoped features using durable state. Additional named providers give journal consumers independent storage namespaces alongside default grain journaling. Provider registration also adds the core Journaling services, durable-state keyed services, JSON format, and Orleans binary reader. The Journaling packages are pre-release alpha packages and their APIs carry diagnostic `ORLEANSEXP005`. +## Configure grain-scoped composition + +The standard registration supplies one scoped per activation. Its service factory enrolls the concrete manager in the grain lifecycle before returning it. Inject that manager and keyed durable states into an ordinary , an application-owned grain base, or an activation-scoped feature. Recovery then completes before and requests. + +Storage registration makes these services available across the silo. Per-grain journal reads and writes are triggered by activations which resolve the manager directly or through a durable-state dependency. Unrelated grain activations retain their existing persistence behavior. + + provides optional protected helpers for the same standard manager and lifecycle enrollment for explicitly supplied managers exposing a lifecycle participant. Custom managers used with plain grains assign lifecycle enrollment to their service factory or a [shared activation setup action](../grain-lifecycle.md#shared-activation-setup). Explicit factories and manually constructed managers retain caller-owned initialization and disposal unless the caller deliberately assigns lifecycle ownership. + +See [Use durable state](durable-state.md) for source-backed constructor injection and feature setup examples, and [Activation and recovery](runtime-behavior.md#activation-and-recovery) for lifecycle ordering. + ## Choose a storage provider | Provider | Package | Storage model | Compaction trigger | @@ -131,6 +141,6 @@ The default minimum is seven days. Removal is persisted by a compaction after th ## Development storage - registers core services and resolves an . Runtime tests and disposable development hosts can use with a provider name. Its contents live in process memory, so use persistent emulator storage to validate restart recovery and provider migration. + registers core services which use an . Runtime tests and disposable development hosts can use with a provider name. Its contents live in process memory, so use persistent emulator storage to validate restart recovery and provider migration. Use the same durable provider category in staging that production uses so recovery, compaction, concurrency, and backup procedures receive realistic validation. diff --git a/docs/site/src/content/docs/grains/journaling/durable-state.md b/docs/site/src/content/docs/grains/journaling/durable-state.md index 5c7cf40339e..10aa8be84b6 100644 --- a/docs/site/src/content/docs/grains/journaling/durable-state.md +++ b/docs/site/src/content/docs/grains/journaling/durable-state.md @@ -7,17 +7,23 @@ ms.topic: how-to # Use durable state -Install the pre-release [`Microsoft.Orleans.Journaling`](https://www.nuget.org/packages/Microsoft.Orleans.Journaling) package in the silo project. Install a [journal storage provider](configuration.md#choose-a-storage-provider) and configure it before activating a . +Install the pre-release [`Microsoft.Orleans.Journaling`](https://www.nuget.org/packages/Microsoft.Orleans.Journaling) package in the silo project. Install a [journal storage provider](configuration.md#choose-a-storage-provider) and configure it on the silos which host grains using durable state. All Journaling APIs are experimental and carry diagnostic `ORLEANSEXP005`. ## Define a durable grain -Inject durable states with . The service key becomes the state's stable name in the grain journal: +Compose an ordinary with an injected and durable states. Inject durable states with . The service key becomes the state's stable name in the grain journal: -:::code language="csharp" source="./snippets/journaling/JournalingBasics.cs" id="durable_shopping_cart"::: +:::code language="csharp" source="../../snippets/compiled/Grains/JournalingSnippets.cs" id="composed_shopping_cart"::: -The dictionary mutation is immediately visible to the current activation. Awaiting establishes the durability point for every pending durable-state mutation on that grain. +The standard grain-scoped service factory enrolls the manager in the grain lifecycle when constructor injection resolves it. Durable states register during construction, and recovery finishes before and request processing. The same composition works with an application-owned grain base class. + +The dictionary mutation is immediately visible to the current activation. Awaiting establishes the durability point for every pending durable-state mutation on that grain. The manager's asynchronous APIs require a argument; pass a token with the operation's lifetime or . + + remains a convenience base exposing its protected , , and members. Choose that base when those helpers fit the application; constructor-injected composition gives existing grain hierarchies the same standard recovery behavior. + +The composition example is compiled against repository source so it exercises the factory-owned lifecycle behavior. The format and collection snippets use the approved published package versions. ## Select a state type @@ -31,7 +37,7 @@ The dictionary mutation is immediately visible to the current activation. Awaiti | | Durable task completion | Complete, fault, or cancel | | | Record-style state | Set or clear a versioned state value | -All named states in one grain share the grain's journal and participate in the same write. This makes a single the atomic storage boundary for their pending changes. Coordination with another grain or an external service requires an application protocol such as idempotency, an inbox, or an outbox. +All named states in one grain share the grain's journal and participate in the same write. This makes a single the atomic storage boundary for their pending changes. Coordination with another grain or an external service requires an application protocol such as idempotency, an inbox, or an outbox. An changes status in memory when `TrySetResult`, `TrySetException`, or `TrySetCanceled` succeeds. Its `Task` completes after a write acknowledges that status or recovery replays it, allowing waiters to observe a durable completion. @@ -47,6 +53,20 @@ The keyed service name identifies a durable state across activations and deploym Registering two states with the same name fails activation. Registering a state after activation setup also fails because recovery has already assigned journal stream identities. +## Compose an activation-scoped feature + +A reusable feature can own durable state independently of the grain's constructor and base class. Select the grain implementation once per grain type using in an implementation, then register a shared setup action. This example selects the existing shopping-cart interface and records an activation count: + +:::code language="csharp" source="../../snippets/compiled/Grains/JournalingSnippets.cs" id="journaled_feature"::: + +Register the feature as scoped and the configurator as singleton: + +:::code language="csharp" source="../../snippets/compiled/Grains/JournalingSnippets.cs" id="journaled_feature_registration"::: + +The action resolves the feature from and enrolls its lifecycle participant. Resolving the feature's dependencies also resolves the standard state manager, whose factory owns the manager's enrollment. The feature uses a stage after and before so it increments recovered state before application activation begins. + +Keep one enrollment owner per participant. Setup actions are shared across concurrent activations; resolve activation-specific data from the supplied context and keep shared callbacks stateless. See [Shared activation setup](../grain-lifecycle.md#shared-activation-setup) and [Journaling activation and recovery](runtime-behavior.md#activation-and-recovery) for ordering and failure behavior. + ## Use journal-backed persistent state Journaling registers keyed services. Its familiar `State`, `WriteStateAsync`, and `ClearStateAsync` members write through the same journal manager as the durable collections. `ReadStateAsync` completes from the already-recovered in-memory state because activation setup replayed the grain journal. diff --git a/docs/site/src/content/docs/grains/journaling/index.md b/docs/site/src/content/docs/grains/journaling/index.md index f91cc12b6a2..b84bac80f25 100644 --- a/docs/site/src/content/docs/grains/journaling/index.md +++ b/docs/site/src/content/docs/grains/journaling/index.md @@ -14,7 +14,9 @@ Orleans Journaling is an experimental persistence model that records mutations t ## Programming model -A journaling grain derives from and receives named durable states through keyed dependency injection. Orleans currently provides: +A journaling grain composes with named durable states received through keyed dependency injection. The standard manager factory owns lifecycle enrollment, so constructor-injected state recovers before application activation and requests. This works with an ordinary , an application-owned grain base, or a reusable activation-scoped feature. remains an optional convenience base with protected state-manager helpers. + +Orleans currently provides: - - @@ -24,7 +26,7 @@ A journaling grain derives from and recei - - backed by the grain's journal -Mutations update the activation's in-memory state and add encoded operations to its pending journal buffer. Await at the application durability point. The returned task completes after the storage provider acknowledges the append or snapshot replacement. +Mutations update the activation's in-memory state and add encoded operations to its pending journal buffer. Await at the application durability point. The returned task completes after the storage provider acknowledges the append or snapshot replacement. Prepare fallible work in operation-local data and stage mutations once they are safe to commit. Interleaved calls share the pending journal. A failed journal operation fences the manager and requests grain deactivation; @@ -34,13 +36,14 @@ Each named state has a stable stream identity within the grain journal. Keep tho ## Journal lifecycle -1. During activation setup, Orleans reads the journal in order and replays each state stream. +1. Resolving the grain-scoped manager enrolls it in the lifecycle. Constructor injection and shared setup register states before lifecycle startup. +1. During the setup-state stage, Orleans reads the journal in order and replays each state stream before application activation and requests. 1. Grain code synchronously mutates durable values and collections during a grain turn. -1. gathers pending operations for the grain and submits one atomic journal append or replacement to storage. +1. gathers pending operations for the grain and submits one atomic journal append or replacement to storage. 1. The storage provider can request compaction when its configured size or row threshold is reached. The next write creates a snapshot of the current durable states and atomically replaces the journal. 1. A later activation replays the latest snapshot and subsequent operations to restore the same durable state. -For the detailed guarantees, see [Runtime behavior and consistency](runtime-behavior.md). +Registering a provider makes the services available; activations which resolve the manager trigger per-grain journal I/O. For the detailed guarantees and caller-owned manager lifetimes, see [Runtime behavior and consistency](runtime-behavior.md). ## Journal formats @@ -56,7 +59,7 @@ Orleans offers two separate journal-oriented programming models: | Model | Application state model | Persistence coordination | | --- | --- | --- | -| Orleans Journaling | Mutable durable values and collections on | One per-grain journal managed by `Microsoft.Orleans.Journaling` | +| Orleans Journaling | Mutable durable values and collections composed with a grain or feature | One per-grain journal managed by `Microsoft.Orleans.Journaling` | | [Orleans Event Sourcing](../event-sourcing/index.md) | Application-defined events applied to `JournaledGrain` | Log-consistency providers confirm, persist, and synchronize events | Choose Journaling when evaluating operation-based persistence for built-in mutable state structures. Choose Event Sourcing when domain events, event history, and the supported log-consistency programming model are application requirements. diff --git a/docs/site/src/content/docs/grains/journaling/runtime-behavior.md b/docs/site/src/content/docs/grains/journaling/runtime-behavior.md index 09194a43383..458a50aa708 100644 --- a/docs/site/src/content/docs/grains/journaling/runtime-behavior.md +++ b/docs/site/src/content/docs/grains/journaling/runtime-behavior.md @@ -11,7 +11,11 @@ Orleans Journaling assigns one to each grain ## Activation and recovery -The constructor attaches its state manager to the grain lifecycle. During the setup-state stage, the manager: +The standard grain-scoped factory registered by creates the concrete manager and enrolls it in the grain lifecycle before returning it. Resolving a keyed durable state also resolves this manager. Each activation shares one manager across its injected states and features. + +Constructor injection registers durable states before lifecycle startup. Orleans completes the grain constructor and assigns , then runs any shared activation setup actions before calling the grain object's `Participate` method and starting the lifecycle. A feature can resolve additional activation-scoped states and enroll its own participant in those actions. All subscriptions are established before lifecycle startup. + +During , the manager: 1. Reads the journal as an ordered byte stream. 1. Selects the reader named by the stored format metadata. @@ -19,15 +23,25 @@ The constructor attaches its state manage 1. Resets and replays each registered durable state. 1. Completes activation setup after replay finishes. -Requests reach the grain after setup succeeds. A storage read, format, codec, or malformed-data failure fails activation and preserves the stored journal for diagnosis and recovery. + and requests observe recovered durable state after setup succeeds, whether the grain derives directly from , from an application-owned base, or from . A storage read, format, codec, or malformed-data failure fails activation and preserves the stored journal for diagnosis and recovery. + +Provider registration makes Journaling services available. Per-grain journal I/O begins only for activations which resolve the manager, directly or through durable-state dependencies. Grains which use other persistence models keep their existing activation behavior. + +Lifecycle stages execute in order, with callbacks at the same stage eligible to run concurrently. A feature which reads recovered state subscribes after `SetupState`; a feature which must finish before `OnActivateAsync` subscribes before . Shared setup actions execute after construction and are reused concurrently across activations, so they resolve scoped state through the supplied context. See [Shared activation setup](../grain-lifecycle.md#shared-activation-setup). Storage providers can split reads at arbitrary byte boundaries. The journal format buffers incomplete entries and only applies complete ordered records. +### Custom and caller-owned managers + +Assign lifecycle enrollment to the owner which creates or selects a manager. `DurableGrain` enrolls explicitly supplied managers implementing for , including a standard manager returned by a custom registration. It preserves the standard hosting factory's completed enrollment so that manager participates once. A custom manager injected into a plain grain uses its service factory or an explicit shared activation setup action to enroll. Ordinary DI registration makes a service resolvable; its factory or setup action establishes lifecycle subscriptions. + +Managers created through with an explicit , and manually constructed managers, retain caller-owned initialization and disposal. This applies inside grain calls as well as outside the runtime. Register their states, await with a cancellation token, and dispose the manager when processing ends. A caller can deliberately assign lifecycle ownership by supplying a manager through its scoped registration to `DurableGrain` or explicitly enrolling it. Creation through the explicit-journal factory keeps failure handling independent of the ambient activation, including after the caller enrolls that manager in a lifecycle. + ## Mutation and write acknowledgement Durable collections encode their operation before applying it to the in-memory collection. A codec failure therefore leaves both the journal buffer and collection unchanged. - gathers pending entries from all named states and queues one storage operation: + gathers pending entries from all named states and queues one storage operation: - **Append** adds the encoded operation batch atomically. - **Snapshot replacement** writes the current state of every registered stream and atomically publishes it as the new journal generation. @@ -35,7 +49,7 @@ Durable collections encode their operation before applying it to the in-memory c Concurrent calls made while the same kind of write is queued can share that queued operation. Each caller observes its completion or failure. Calls made after a storage operation starts are processed by a later operation. > [!IMPORTANT] -> In-memory mutation is visible before storage acknowledgement. Return success to a caller only after the required completes. Recovery reconstructs durable state in a new activation. +> In-memory mutation is visible before storage acknowledgement. Return success to a caller only after the required completes. Recovery reconstructs durable state in a new activation. ## Safe-to-commit staging @@ -79,7 +93,7 @@ or repair the backing data before creating a fresh manager or retrying activatio ## Compaction -Each provider reports when its journal crosses a configured storage threshold. The next : +Each provider reports when its journal crosses a configured storage threshold. The next : 1. Builds a snapshot containing the state directory and every active durable state. 1. Atomically replaces the published journal with the snapshot. @@ -97,4 +111,4 @@ This behavior supports staged deployments and rollback. Keep the previous format ## Deactivation and shutdown -Grain deactivation stops the state manager's work loop. Completed writes are the durability barrier. Await every required write during the grain call which made the mutation, and size host shutdown grace periods for writes already in progress. +Grain deactivation stops the lifecycle-bound state manager's work loop, and activation-scope disposal releases its resources. Completed writes are the durability barrier. Await every required write during the grain call which made the mutation, and size host shutdown grace periods for writes already in progress. Owners of explicit-journal or manually constructed managers arrange their own shutdown and disposal. diff --git a/docs/site/src/content/docs/snippets/compiled/CompiledDocumentationSnippets.csproj b/docs/site/src/content/docs/snippets/compiled/CompiledDocumentationSnippets.csproj index 26edfaa39ba..d91767cf717 100644 --- a/docs/site/src/content/docs/snippets/compiled/CompiledDocumentationSnippets.csproj +++ b/docs/site/src/content/docs/snippets/compiled/CompiledDocumentationSnippets.csproj @@ -7,6 +7,7 @@ + diff --git a/docs/site/src/content/docs/snippets/compiled/Grains/JournalingSnippets.cs b/docs/site/src/content/docs/snippets/compiled/Grains/JournalingSnippets.cs new file mode 100644 index 00000000000..62c996c736a --- /dev/null +++ b/docs/site/src/content/docs/snippets/compiled/Grains/JournalingSnippets.cs @@ -0,0 +1,87 @@ +using Microsoft.Extensions.DependencyInjection; +using Orleans; +using Orleans.Hosting; +using Orleans.Journaling; +using Orleans.Metadata; +using Orleans.Runtime; + +#pragma warning disable ORLEANSEXP005 + +namespace Documentation.Grains.Journaling; + +// +public interface IShoppingCartGrain : IGrainWithStringKey +{ + ValueTask AddItem(string itemId, int quantity); + ValueTask> GetItems(); +} + +public sealed class ShoppingCartGrain( + IJournaledStateManager stateManager, + [FromKeyedServices("cart")] IDurableDictionary cart) + : Grain, IShoppingCartGrain +{ + public async ValueTask AddItem(string itemId, int quantity) + { + cart[itemId] = quantity; + await stateManager.WriteStateAsync(CancellationToken.None); + } + + public ValueTask> GetItems() => + new(cart.ToDictionary()); +} +// + +// +public sealed class CartActivationCounter( + IJournaledStateManager stateManager, + [FromKeyedServices("activation-count")] IDurableValue count) + : ILifecycleParticipant +{ + public void Participate(IGrainLifecycle lifecycle) + { + lifecycle.Subscribe( + GrainLifecycleStage.Activate - 1, + async cancellationToken => + { + count.Value++; + await stateManager.WriteStateAsync(cancellationToken); + }); + } +} + +public sealed class CartFeatureConfigurator(GrainClassMap grainClasses) + : IConfigureGrainTypeComponents +{ + public void Configure( + GrainType grainType, + GrainProperties properties, + GrainTypeSharedContext shared) + { + if (grainClasses.TryGetGrainClass(grainType, out var grainClass) + && typeof(IShoppingCartGrain).IsAssignableFrom(grainClass)) + { + shared.AddActivationSetup(static context => + { + var feature = context.ActivationServices + .GetRequiredService(); + feature.Participate(context.ObservableLifecycle); + }); + } + } +} +// + +internal static class FeatureConfiguration +{ + internal static void Configure(ISiloBuilder siloBuilder) + { + // + siloBuilder.ConfigureServices(services => + { + services.AddScoped(); + services.AddSingleton(); + }); + // + } +} diff --git a/src/Orleans.Journaling/README.md b/src/Orleans.Journaling/README.md index d70daff3ab5..1fb7c22647e 100644 --- a/src/Orleans.Journaling/README.md +++ b/src/Orleans.Journaling/README.md @@ -68,6 +68,7 @@ To migrate to JSON, configure `JournaledStateManagerOptions.JournalFormatKey` to ## Example - Using durable states ```csharp using Microsoft.Extensions.DependencyInjection; +using Orleans; using Orleans.Journaling; public interface IShoppingCartGrain : IGrainWithStringKey @@ -77,19 +78,26 @@ public interface IShoppingCartGrain : IGrainWithStringKey } public sealed class ShoppingCartGrain( + IJournaledStateManager stateManager, [FromKeyedServices("cart")] IDurableDictionary cart) - : DurableGrain, IShoppingCartGrain + : Grain, IShoppingCartGrain { public async ValueTask AddItem(string itemId, int quantity) { cart[itemId] = quantity; - await WriteStateAsync(); + await stateManager.WriteStateAsync(CancellationToken.None); } public ValueTask> GetItems() => new(cart.ToDictionary()); } ``` +The standard grain-scoped factory registered by `AddJournalStorage` enrolls the state manager in the grain lifecycle before returning it. Constructor-injected durable states register with that manager, and recovery completes before `OnActivateAsync` and grain requests. This works with `Grain` or an application-owned grain base class. Registering journal storage makes the services available; only activations which resolve the manager perform per-grain journal I/O. + +`DurableGrain` remains an optional convenience base exposing `StateManager`, `GetOrCreateState`, and `WriteStateAsync`. It also enrolls explicitly supplied managers implementing `ILifecycleParticipant`, including a standard manager supplied by a custom registration, while preserving the standard hosting factory's completed enrollment. A custom manager used with an ordinary grain assigns enrollment to its service factory or an explicit shared activation setup action. + +Managers created with an explicit `JournalId` through `IJournaledStateManagerFactory`, or constructed manually, retain caller-owned initialization and disposal even when created inside a grain call. Register their states and await `InitializeAsync(CancellationToken.None)` before use, or deliberately supply them through a registration which assigns lifecycle ownership. Creation through the explicit-journal factory keeps failure handling independent of the ambient grain context, including when the caller subsequently enrolls the manager in a lifecycle. The asynchronous `IJournaledStateManager` methods require a cancellation-token argument. + All durable state types use the configured JSON codec automatically. Configure `JsonJournalOptions` to control the `JsonSerializerOptions` instance used for entry payloads. Journaling command names and record shape are fixed by the storage format, so serializer naming policies only affect user payload values. For trimming and Native AOT, use `Configure(...)` to configure `SerializerOptions.TypeInfoResolver`, `SerializerOptions.TypeInfoResolverChain`, or `JsonJournalOptions.AddTypeInfoResolver(...)` with source-generated metadata for every journaled key, value, and state type. The `UseJsonJournalFormat(JournalJsonContext.Default)` overload is the recommended low-friction path when you also want to enable the JSON format explicitly. If metadata is unavailable, the JSON durable entry codecs fail with a configuration error instead of falling back to reflection-based serialization. From 9962bfc9083c94b7e9f061f5c87f8cef53534c40 Mon Sep 17 00:00:00 2001 From: Reuben Bond Date: Thu, 17 Sep 2026 13:16:30 -0700 Subject: [PATCH 5/7] refactor(journaling): rely on factory-owned lifecycle enrollment --- .../docs/grains/journaling/configuration.md | 2 +- .../grains/journaling/runtime-behavior.md | 4 +-- src/Orleans.Journaling/DurableGrain.cs | 9 +------ .../JournaledStateManager.cs | 8 +----- src/Orleans.Journaling/README.md | 2 +- .../JournaledGrainCompositionTests.cs | 25 +++++++++++++++---- 6 files changed, 26 insertions(+), 24 deletions(-) diff --git a/docs/site/src/content/docs/grains/journaling/configuration.md b/docs/site/src/content/docs/grains/journaling/configuration.md index 0568b7a8889..6fd93273c38 100644 --- a/docs/site/src/content/docs/grains/journaling/configuration.md +++ b/docs/site/src/content/docs/grains/journaling/configuration.md @@ -17,7 +17,7 @@ The standard regi Storage registration makes these services available across the silo. Per-grain journal reads and writes are triggered by activations which resolve the manager directly or through a durable-state dependency. Unrelated grain activations retain their existing persistence behavior. - provides optional protected helpers for the same standard manager and lifecycle enrollment for explicitly supplied managers exposing a lifecycle participant. Custom managers used with plain grains assign lifecycle enrollment to their service factory or a [shared activation setup action](../grain-lifecycle.md#shared-activation-setup). Explicit factories and manually constructed managers retain caller-owned initialization and disposal unless the caller deliberately assigns lifecycle ownership. + provides optional protected helpers for the same scoped manager. Every grain-scoped manager registration owns lifecycle enrollment before returning the manager, including custom registrations. Reusable features compose their own lifecycle work through a [shared activation setup action](../grain-lifecycle.md#shared-activation-setup). Explicit factories and manually constructed managers retain caller-owned initialization and disposal unless the caller deliberately assigns lifecycle ownership. See [Use durable state](durable-state.md) for source-backed constructor injection and feature setup examples, and [Activation and recovery](runtime-behavior.md#activation-and-recovery) for lifecycle ordering. diff --git a/docs/site/src/content/docs/grains/journaling/runtime-behavior.md b/docs/site/src/content/docs/grains/journaling/runtime-behavior.md index 458a50aa708..89d7ca0d335 100644 --- a/docs/site/src/content/docs/grains/journaling/runtime-behavior.md +++ b/docs/site/src/content/docs/grains/journaling/runtime-behavior.md @@ -33,9 +33,9 @@ Storage providers can split reads at arbitrary byte boundaries. The journal form ### Custom and caller-owned managers -Assign lifecycle enrollment to the owner which creates or selects a manager. `DurableGrain` enrolls explicitly supplied managers implementing for , including a standard manager returned by a custom registration. It preserves the standard hosting factory's completed enrollment so that manager participates once. A custom manager injected into a plain grain uses its service factory or an explicit shared activation setup action to enroll. Ordinary DI registration makes a service resolvable; its factory or setup action establishes lifecycle subscriptions. +Every grain-scoped manager registration owns lifecycle enrollment before returning the manager. Custom registrations follow the same contract by enrolling their for , or subscribing initialization and shutdown callbacks directly. This gives ordinary grains, application-owned bases, and `DurableGrain` the same recovery and shutdown lifecycle for their injected managers. -Managers created through with an explicit , and manually constructed managers, retain caller-owned initialization and disposal. This applies inside grain calls as well as outside the runtime. Register their states, await with a cancellation token, and dispose the manager when processing ends. A caller can deliberately assign lifecycle ownership by supplying a manager through its scoped registration to `DurableGrain` or explicitly enrolling it. Creation through the explicit-journal factory keeps failure handling independent of the ambient activation, including after the caller enrolls that manager in a lifecycle. +Managers created through with an explicit , and manually constructed managers, retain caller-owned initialization and disposal. This applies inside grain calls as well as outside the runtime. Register their states, await with a cancellation token, and dispose the manager when processing ends. A caller can deliberately assign lifecycle ownership by enrolling the manager in a grain-scoped registration factory. Creation through the explicit-journal factory keeps failure handling independent of the ambient activation, including after the caller enrolls that manager in a lifecycle. ## Mutation and write acknowledgement diff --git a/src/Orleans.Journaling/DurableGrain.cs b/src/Orleans.Journaling/DurableGrain.cs index d65cb0bedd8..85c19a899d9 100644 --- a/src/Orleans.Journaling/DurableGrain.cs +++ b/src/Orleans.Journaling/DurableGrain.cs @@ -11,18 +11,11 @@ public abstract class DurableGrain : Grain, IGrainBase /// Initializes a new instance of the class and resolves its state manager. /// /// - /// The standard grain-scoped state manager enrolls in the lifecycle through its hosting factory. - /// This base class enrolls explicitly supplied state managers which implement - /// , including factory-created managers. + /// The grain-scoped service factory enrolls the state manager in the grain lifecycle before returning it. /// protected DurableGrain() { StateManager = ServiceProvider.GetRequiredService(); - if (StateManager is not JournaledStateManager { IsLifecycleEnrolled: true } - && StateManager is ILifecycleParticipant participant) - { - participant.Participate(((IGrainBase)this).GrainContext.ObservableLifecycle); - } } /// diff --git a/src/Orleans.Journaling/JournaledStateManager.cs b/src/Orleans.Journaling/JournaledStateManager.cs index 81636d19794..8c30a7811e3 100644 --- a/src/Orleans.Journaling/JournaledStateManager.cs +++ b/src/Orleans.Journaling/JournaledStateManager.cs @@ -88,8 +88,6 @@ private static IJournalStorage CreateStorage(IJournalStorageProvider storageProv internal IServiceProvider ServiceProvider => _shared.ServiceProvider; - internal bool IsLifecycleEnrolled { get; private set; } - public void RegisterState(string name, IJournaledState state) { ArgumentNullException.ThrowIfNullOrEmpty(name); @@ -994,11 +992,7 @@ private void BindState(string name, uint id) public long PendingWriteByteCount => _journalWriter.CommittedLength; - void ILifecycleParticipant.Participate(IGrainLifecycle observer) - { - observer.Subscribe(GrainLifecycleStage.SetupState, this); - IsLifecycleEnrolled = true; - } + void ILifecycleParticipant.Participate(IGrainLifecycle observer) => observer.Subscribe(GrainLifecycleStage.SetupState, this); Task ILifecycleObserver.OnStart(CancellationToken cancellationToken) => InitializeAsync(cancellationToken).AsTask(); async Task ILifecycleObserver.OnStop(CancellationToken cancellationToken) => await StopAsync(cancellationToken).ConfigureAwait(false); diff --git a/src/Orleans.Journaling/README.md b/src/Orleans.Journaling/README.md index 1fb7c22647e..29fedef9b49 100644 --- a/src/Orleans.Journaling/README.md +++ b/src/Orleans.Journaling/README.md @@ -94,7 +94,7 @@ public sealed class ShoppingCartGrain( The standard grain-scoped factory registered by `AddJournalStorage` enrolls the state manager in the grain lifecycle before returning it. Constructor-injected durable states register with that manager, and recovery completes before `OnActivateAsync` and grain requests. This works with `Grain` or an application-owned grain base class. Registering journal storage makes the services available; only activations which resolve the manager perform per-grain journal I/O. -`DurableGrain` remains an optional convenience base exposing `StateManager`, `GetOrCreateState`, and `WriteStateAsync`. It also enrolls explicitly supplied managers implementing `ILifecycleParticipant`, including a standard manager supplied by a custom registration, while preserving the standard hosting factory's completed enrollment. A custom manager used with an ordinary grain assigns enrollment to its service factory or an explicit shared activation setup action. +`DurableGrain` remains an optional convenience base exposing `StateManager`, `GetOrCreateState`, and `WriteStateAsync`. Every grain-scoped manager registration owns lifecycle enrollment before returning the manager. Custom registrations follow the same contract, enrolling their `ILifecycleParticipant` or subscribing their initialization and shutdown callbacks directly. Managers created with an explicit `JournalId` through `IJournaledStateManagerFactory`, or constructed manually, retain caller-owned initialization and disposal even when created inside a grain call. Register their states and await `InitializeAsync(CancellationToken.None)` before use, or deliberately supply them through a registration which assigns lifecycle ownership. Creation through the explicit-journal factory keeps failure handling independent of the ambient grain context, including when the caller subsequently enrolls the manager in a lifecycle. The asynchronous `IJournaledStateManager` methods require a cancellation-token argument. diff --git a/test/Orleans.Journaling.Tests/JournaledGrainCompositionTests.cs b/test/Orleans.Journaling.Tests/JournaledGrainCompositionTests.cs index 90f0041916c..be58babbe85 100644 --- a/test/Orleans.Journaling.Tests/JournaledGrainCompositionTests.cs +++ b/test/Orleans.Journaling.Tests/JournaledGrainCompositionTests.cs @@ -200,14 +200,20 @@ public async Task HostingFactory_EnrollsExactlyOnceAndDurableGrainRetainsHelpers : JournalId.FromGrainId(context.GrainId); if (overrideWithExplicitFactory) { - builder.Services.AddScoped(services => services.GetRequiredService().Create(journalId)); + builder.Services.AddScoped(services => + { + var manager = services.GetRequiredService().Create(journalId); + ((ILifecycleParticipant)manager).Participate( + services.GetRequiredService().ObservableLifecycle); + return manager; + }); } await using var services = builder.Services.BuildServiceProvider(new ServiceProviderOptions { ValidateScopes = true }); await using var scope = services.CreateAsyncScope(); context.ActivationServices.Returns(scope.ServiceProvider); var scopedManager = scope.ServiceProvider.GetRequiredService(); - Assert.Equal(overrideWithExplicitFactory ? 0 : 1, lifecycle.Subscriptions); + Assert.Equal(1, lifecycle.Subscriptions); RuntimeContext.SetExecutionContext(context, out var previous); HelperJournalGrain grain; try @@ -237,14 +243,23 @@ public async Task HostingFactory_EnrollsExactlyOnceAndDurableGrainRetainsHelpers } [Fact] - public void DurableGrain_EnrollsCustomLifecycleParticipant() + public async Task DurableGrain_UsesFactoryEnrolledCustomManager() { var manager = Substitute.For>(); var context = Substitute.For(); var lifecycle = new CompositionTestLifecycle(); - using var services = new ServiceCollection().AddSingleton(manager).BuildServiceProvider(); - context.ActivationServices.Returns(services); + await using var services = new ServiceCollection() + .AddScoped(_ => + { + ((ILifecycleParticipant)manager).Participate(lifecycle); + return manager; + }) + .BuildServiceProvider(new ServiceProviderOptions { ValidateScopes = true }); + await using var scope = services.CreateAsyncScope(); + context.ActivationServices.Returns(scope.ServiceProvider); context.ObservableLifecycle.Returns(lifecycle); + Assert.Same(manager, scope.ServiceProvider.GetRequiredService()); + ((ILifecycleParticipant)manager).Received(1).Participate(lifecycle); RuntimeContext.SetExecutionContext(context, out var previous); try { From cee434f796484e45607cc9d3d723dce6274065dc Mon Sep 17 00:00:00 2001 From: Reuben Bond Date: Thu, 17 Sep 2026 14:43:04 -0700 Subject: [PATCH 6/7] refactor(journaling): enroll grain managers during construction --- src/Orleans.Journaling/DurableGrain.cs | 2 +- src/Orleans.Journaling/HostingExtensions.cs | 17 ++----------- .../JournaledStateManager.cs | 9 +++++++ .../JournalBatchTests.cs | 2 +- .../JournaledGrainCompositionTests.cs | 25 +++++++++++++------ .../KeyedJournalingRegistrationTests.cs | 2 ++ 6 files changed, 33 insertions(+), 24 deletions(-) diff --git a/src/Orleans.Journaling/DurableGrain.cs b/src/Orleans.Journaling/DurableGrain.cs index 85c19a899d9..09e3136728f 100644 --- a/src/Orleans.Journaling/DurableGrain.cs +++ b/src/Orleans.Journaling/DurableGrain.cs @@ -11,7 +11,7 @@ public abstract class DurableGrain : Grain, IGrainBase /// Initializes a new instance of the class and resolves its state manager. /// /// - /// The grain-scoped service factory enrolls the state manager in the grain lifecycle before returning it. + /// The standard state manager enrolls in the grain lifecycle during grain-bound construction. /// protected DurableGrain() { diff --git a/src/Orleans.Journaling/HostingExtensions.cs b/src/Orleans.Journaling/HostingExtensions.cs index 3adf01a41a3..6aa57c66b21 100644 --- a/src/Orleans.Journaling/HostingExtensions.cs +++ b/src/Orleans.Journaling/HostingExtensions.cs @@ -31,23 +31,10 @@ public static ISiloBuilder AddJournalStorage(this ISiloBuilder builder) : JournalingInstruments.CreateForDirectConstruction()); builder.Services.TryAddSingleton(); builder.Services.TryAddScoped(static services => - { - var context = services.GetRequiredService(); - var manager = new JournaledStateManager( + new JournaledStateManager( services.GetRequiredService(), services.GetRequiredService(), - context); - try - { - ((ILifecycleParticipant)manager).Participate(context.ObservableLifecycle); - return manager; - } - catch - { - ((IDisposable)manager).Dispose(); - throw; - } - }); + services.GetRequiredService())); builder.Services.TryAddSingleton(static services => services.GetKeyedService(ProviderConstants.DEFAULT_STORAGE_PROVIDER_NAME) ?? ActivatorUtilities.CreateInstance(services)); diff --git a/src/Orleans.Journaling/JournaledStateManager.cs b/src/Orleans.Journaling/JournaledStateManager.cs index 8c30a7811e3..c62c4e1f157 100644 --- a/src/Orleans.Journaling/JournaledStateManager.cs +++ b/src/Orleans.Journaling/JournaledStateManager.cs @@ -38,6 +38,15 @@ public JournaledStateManager(JournaledStateManagerShared shared, IJournalStorage : this(shared, CreateStorage(storageProvider, CreateJournalId(grainContext))) { _grainContext = grainContext; + try + { + ((ILifecycleParticipant)this).Participate(grainContext.ObservableLifecycle); + } + catch + { + ((IDisposable)this).Dispose(); + throw; + } } public JournaledStateManager(JournaledStateManagerShared shared, IJournalStorageProvider storageProvider, JournalId journalId) diff --git a/test/Orleans.Journaling.Tests/JournalBatchTests.cs b/test/Orleans.Journaling.Tests/JournalBatchTests.cs index 50b9f1dff73..92dd85f2e74 100644 --- a/test/Orleans.Journaling.Tests/JournalBatchTests.cs +++ b/test/Orleans.Journaling.Tests/JournalBatchTests.cs @@ -409,7 +409,7 @@ internal sealed class TestGrainContext(GrainId grainId) : IGrainContext public ActivationId ActivationId => throw new NotImplementedException(); public GrainAddress Address => throw new NotImplementedException(); public IServiceProvider ActivationServices => throw new NotImplementedException(); - public IGrainLifecycle ObservableLifecycle => throw new NotImplementedException(); + public IGrainLifecycle ObservableLifecycle { get; } = new CompositionTestLifecycle(); public IWorkItemScheduler Scheduler => throw new NotImplementedException(); public Task Deactivated => throw new NotImplementedException(); diff --git a/test/Orleans.Journaling.Tests/JournaledGrainCompositionTests.cs b/test/Orleans.Journaling.Tests/JournaledGrainCompositionTests.cs index be58babbe85..a299578b565 100644 --- a/test/Orleans.Journaling.Tests/JournaledGrainCompositionTests.cs +++ b/test/Orleans.Journaling.Tests/JournaledGrainCompositionTests.cs @@ -185,9 +185,10 @@ public async Task ExplicitFactory_WithAmbientGrainContextKeepsManualIndependentL } [Theory] - [InlineData(false)] - [InlineData(true)] - public async Task HostingFactory_EnrollsExactlyOnceAndDurableGrainRetainsHelpers(bool overrideWithExplicitFactory) + [InlineData("hosting")] + [InlineData("implementation")] + [InlineData("explicit")] + public async Task ManagerResolution_EnrollsExactlyOnceAndDurableGrainRetainsHelpers(string registration) { var builder = CreateBuilder(); var lifecycle = new CompositionTestLifecycle(); @@ -195,10 +196,14 @@ public async Task HostingFactory_EnrollsExactlyOnceAndDurableGrainRetainsHelpers context.GrainId.Returns(GrainId.Create("composition", "factory-enrollment")); context.ObservableLifecycle.Returns(lifecycle); builder.Services.AddScoped(_ => context); - var journalId = overrideWithExplicitFactory + var journalId = registration == "explicit" ? new JournalId("explicit/scoped-override") : JournalId.FromGrainId(context.GrainId); - if (overrideWithExplicitFactory) + if (registration == "implementation") + { + builder.Services.AddScoped(); + } + else if (registration == "explicit") { builder.Services.AddScoped(services => { @@ -274,10 +279,16 @@ public async Task DurableGrain_UsesFactoryEnrolledCustomManager() ((ILifecycleParticipant)manager).Received(1).Participate(lifecycle); } - [Fact] - public void HostingFactory_EnrollmentFailureDisposesManagerAndPreservesException() + [Theory] + [InlineData(false)] + [InlineData(true)] + public void GrainBoundConstruction_EnrollmentFailureDisposesManagerAndPreservesException(bool useImplementationRegistration) { var builder = CreateBuilder(); + if (useImplementationRegistration) + { + builder.Services.AddScoped(); + } builder.Services.AddSingleton(); builder.Services.AddKeyedSingleton(JsonJournalExtensions.JournalFormatKey, static (services, _) => services.GetRequiredService()); diff --git a/test/Orleans.Journaling.Tests/KeyedJournalingRegistrationTests.cs b/test/Orleans.Journaling.Tests/KeyedJournalingRegistrationTests.cs index 3b7de5c9843..ff73f40f198 100644 --- a/test/Orleans.Journaling.Tests/KeyedJournalingRegistrationTests.cs +++ b/test/Orleans.Journaling.Tests/KeyedJournalingRegistrationTests.cs @@ -377,6 +377,8 @@ public void DurableService_ResolvesCommandCodecFromJournalFormatKey() _ = scope.ServiceProvider.GetRequiredKeyedService>("value"); Assert.True(wasUsed); + Assert.Equal(1, Assert.IsType( + scope.ServiceProvider.GetRequiredService().ObservableLifecycle).Subscriptions); } [Fact] From a2ebe0d35c91466c69a867b8f655372d27812227 Mon Sep 17 00:00:00 2001 From: Reuben Bond Date: Thu, 17 Sep 2026 14:43:05 -0700 Subject: [PATCH 7/7] docs(journaling): propagate operation cancellation in examples --- .../docs/grains/journaling/configuration.md | 4 +- .../docs/grains/journaling/durable-state.md | 8 +- .../content/docs/grains/journaling/index.md | 2 +- .../grains/journaling/runtime-behavior.md | 8 +- .../snippets/journaling/JournalingBasics.cs | 16 ++-- .../snippets/redis-journaling/Program.cs | 57 ++++++++----- .../compiled/Grains/JournalingSnippets.cs | 16 ++-- .../JournalingAzureBlobJson/Program.cs | 79 +++++++++++-------- samples/JournalingAzureBlobJson/README.md | 3 + src/Orleans.Journaling/README.md | 38 +++++---- 10 files changed, 140 insertions(+), 91 deletions(-) diff --git a/docs/site/src/content/docs/grains/journaling/configuration.md b/docs/site/src/content/docs/grains/journaling/configuration.md index 6fd93273c38..a9b61f7c7ea 100644 --- a/docs/site/src/content/docs/grains/journaling/configuration.md +++ b/docs/site/src/content/docs/grains/journaling/configuration.md @@ -13,11 +13,11 @@ The Journaling packages are pre-release alpha packages and their APIs carry diag ## Configure grain-scoped composition -The standard registration supplies one scoped per activation. Its service factory enrolls the concrete manager in the grain lifecycle before returning it. Inject that manager and keyed durable states into an ordinary , an application-owned grain base, or an activation-scoped feature. Recovery then completes before and requests. +The standard registration supplies one scoped per activation. The manager enrolls itself in the grain lifecycle when constructed with the activation's . Inject that manager and keyed durable states into an ordinary , an application-owned grain base, or an activation-scoped feature. Recovery then completes before and requests. Storage registration makes these services available across the silo. Per-grain journal reads and writes are triggered by activations which resolve the manager directly or through a durable-state dependency. Unrelated grain activations retain their existing persistence behavior. - provides optional protected helpers for the same scoped manager. Every grain-scoped manager registration owns lifecycle enrollment before returning the manager, including custom registrations. Reusable features compose their own lifecycle work through a [shared activation setup action](../grain-lifecycle.md#shared-activation-setup). Explicit factories and manually constructed managers retain caller-owned initialization and disposal unless the caller deliberately assigns lifecycle ownership. + provides optional protected helpers for the same scoped manager. Grain-scoped managers are enrolled before resolution returns; custom managers establish this in their constructor or registration factory. Reusable features compose their own lifecycle work through a [shared activation setup action](../grain-lifecycle.md#shared-activation-setup). Explicit factories and managers constructed directly from storage retain caller-owned initialization and disposal unless the caller deliberately assigns lifecycle ownership. See [Use durable state](durable-state.md) for source-backed constructor injection and feature setup examples, and [Activation and recovery](runtime-behavior.md#activation-and-recovery) for lifecycle ordering. diff --git a/docs/site/src/content/docs/grains/journaling/durable-state.md b/docs/site/src/content/docs/grains/journaling/durable-state.md index 10aa8be84b6..fb9790165ab 100644 --- a/docs/site/src/content/docs/grains/journaling/durable-state.md +++ b/docs/site/src/content/docs/grains/journaling/durable-state.md @@ -17,13 +17,13 @@ Compose an ordinary with an injected and request processing. The same composition works with an application-owned grain base class. +The standard manager enrolls itself in the grain lifecycle when constructed with the activation's . Durable states register during construction, and recovery finishes before and request processing. The same composition works with an application-owned grain base class. -The dictionary mutation is immediately visible to the current activation. Awaiting establishes the durability point for every pending durable-state mutation on that grain. The manager's asynchronous APIs require a argument; pass a token with the operation's lifetime or . +The dictionary mutation is immediately visible to the current activation. Awaiting establishes the durability point for every pending durable-state mutation on that grain. Accept a on grain operations and flow it through state-manager calls so cancellation follows the caller's operation lifetime. remains a convenience base exposing its protected , , and members. Choose that base when those helpers fit the application; constructor-injected composition gives existing grain hierarchies the same standard recovery behavior. -The composition example is compiled against repository source so it exercises the factory-owned lifecycle behavior. The format and collection snippets use the approved published package versions. +The composition example is compiled against repository source so it exercises constructor-owned lifecycle enrollment. ## Select a state type @@ -63,7 +63,7 @@ Register the feature as scoped and the configurator as singleton: :::code language="csharp" source="../../snippets/compiled/Grains/JournalingSnippets.cs" id="journaled_feature_registration"::: -The action resolves the feature from and enrolls its lifecycle participant. Resolving the feature's dependencies also resolves the standard state manager, whose factory owns the manager's enrollment. The feature uses a stage after and before so it increments recovered state before application activation begins. +The action resolves the feature from and enrolls its lifecycle participant. Resolving the feature's dependencies also constructs the standard state manager, which enrolls itself in the grain lifecycle. The feature uses a stage after and before so it increments recovered state before application activation begins. Keep one enrollment owner per participant. Setup actions are shared across concurrent activations; resolve activation-specific data from the supplied context and keep shared callbacks stateless. See [Shared activation setup](../grain-lifecycle.md#shared-activation-setup) and [Journaling activation and recovery](runtime-behavior.md#activation-and-recovery) for ordering and failure behavior. diff --git a/docs/site/src/content/docs/grains/journaling/index.md b/docs/site/src/content/docs/grains/journaling/index.md index b84bac80f25..442bdb47789 100644 --- a/docs/site/src/content/docs/grains/journaling/index.md +++ b/docs/site/src/content/docs/grains/journaling/index.md @@ -14,7 +14,7 @@ Orleans Journaling is an experimental persistence model that records mutations t ## Programming model -A journaling grain composes with named durable states received through keyed dependency injection. The standard manager factory owns lifecycle enrollment, so constructor-injected state recovers before application activation and requests. This works with an ordinary , an application-owned grain base, or a reusable activation-scoped feature. remains an optional convenience base with protected state-manager helpers. +A journaling grain composes with named durable states received through keyed dependency injection. The standard manager enrolls itself in the lifecycle during grain-bound construction, so constructor-injected state recovers before application activation and requests. This works with an ordinary , an application-owned grain base, or a reusable activation-scoped feature. remains an optional convenience base with protected state-manager helpers. Orleans currently provides: diff --git a/docs/site/src/content/docs/grains/journaling/runtime-behavior.md b/docs/site/src/content/docs/grains/journaling/runtime-behavior.md index 89d7ca0d335..cf56d93d3e7 100644 --- a/docs/site/src/content/docs/grains/journaling/runtime-behavior.md +++ b/docs/site/src/content/docs/grains/journaling/runtime-behavior.md @@ -11,7 +11,7 @@ Orleans Journaling assigns one to each grain ## Activation and recovery -The standard grain-scoped factory registered by creates the concrete manager and enrolls it in the grain lifecycle before returning it. Resolving a keyed durable state also resolves this manager. Each activation shares one manager across its injected states and features. +The standard grain-scoped registered by enrolls itself in the grain lifecycle when constructed with the activation's . Resolving a keyed durable state also resolves this manager. Each activation shares one manager across its injected states and features. Constructor injection registers durable states before lifecycle startup. Orleans completes the grain constructor and assigns , then runs any shared activation setup actions before calling the grain object's `Participate` method and starting the lifecycle. A feature can resolve additional activation-scoped states and enroll its own participant in those actions. All subscriptions are established before lifecycle startup. @@ -33,9 +33,9 @@ Storage providers can split reads at arbitrary byte boundaries. The journal form ### Custom and caller-owned managers -Every grain-scoped manager registration owns lifecycle enrollment before returning the manager. Custom registrations follow the same contract by enrolling their for , or subscribing initialization and shutdown callbacks directly. This gives ordinary grains, application-owned bases, and `DurableGrain` the same recovery and shutdown lifecycle for their injected managers. +Grain-scoped managers are enrolled before resolution returns. The standard manager establishes this in its grain-bound constructor. Custom managers establish it in their constructor or registration factory by enrolling their for , or subscribing initialization and shutdown callbacks directly. This gives ordinary grains, application-owned bases, and `DurableGrain` the same recovery and shutdown lifecycle for their injected managers. -Managers created through with an explicit , and manually constructed managers, retain caller-owned initialization and disposal. This applies inside grain calls as well as outside the runtime. Register their states, await with a cancellation token, and dispose the manager when processing ends. A caller can deliberately assign lifecycle ownership by enrolling the manager in a grain-scoped registration factory. Creation through the explicit-journal factory keeps failure handling independent of the ambient activation, including after the caller enrolls that manager in a lifecycle. +Managers created through with an explicit , and managers constructed directly from storage without a grain context, retain caller-owned initialization and disposal. This applies inside grain calls as well as outside the runtime. Register their states, await with the operation's cancellation token, and dispose the manager when processing ends. A caller can deliberately assign lifecycle ownership by enrolling the manager in a grain-scoped registration factory. Creation through the explicit-journal factory keeps failure handling independent of the ambient activation, including after the caller enrolls that manager in a lifecycle. ## Mutation and write acknowledgement @@ -111,4 +111,4 @@ This behavior supports staged deployments and rollback. Keep the previous format ## Deactivation and shutdown -Grain deactivation stops the lifecycle-bound state manager's work loop, and activation-scope disposal releases its resources. Completed writes are the durability barrier. Await every required write during the grain call which made the mutation, and size host shutdown grace periods for writes already in progress. Owners of explicit-journal or manually constructed managers arrange their own shutdown and disposal. +Grain deactivation stops the lifecycle-bound state manager's work loop, and activation-scope disposal releases its resources. Completed writes are the durability barrier. Await every required write during the grain call which made the mutation, and size host shutdown grace periods for writes already in progress. Owners of explicit-journal or standalone managers arrange their own shutdown and disposal. diff --git a/docs/site/src/content/docs/grains/journaling/snippets/journaling/JournalingBasics.cs b/docs/site/src/content/docs/grains/journaling/snippets/journaling/JournalingBasics.cs index 97431c991e2..9db893973d3 100644 --- a/docs/site/src/content/docs/grains/journaling/snippets/journaling/JournalingBasics.cs +++ b/docs/site/src/content/docs/grains/journaling/snippets/journaling/JournalingBasics.cs @@ -5,9 +5,9 @@ namespace Orleans.Docs.Snippets.Journaling; public interface IShoppingCartGrain : IGrainWithStringKey { - ValueTask AddItem(string itemId, int quantity); + ValueTask AddItem(string itemId, int quantity, CancellationToken cancellationToken); - ValueTask> GetItems(); + ValueTask> GetItems(CancellationToken cancellationToken); } // @@ -16,14 +16,18 @@ public sealed class ShoppingCartGrain( Orleans.Journaling.IDurableDictionary items) : Orleans.Journaling.DurableGrain, IShoppingCartGrain { - public async ValueTask AddItem(string itemId, int quantity) + public async ValueTask AddItem(string itemId, int quantity, CancellationToken cancellationToken) { + cancellationToken.ThrowIfCancellationRequested(); items[itemId] = quantity; - await WriteStateAsync(); + await WriteStateAsync(cancellationToken); } - public ValueTask> GetItems() => - ValueTask.FromResult>( + public ValueTask> GetItems(CancellationToken cancellationToken) + { + cancellationToken.ThrowIfCancellationRequested(); + return ValueTask.FromResult>( new Dictionary(items)); + } } // diff --git a/docs/site/src/content/docs/grains/journaling/snippets/redis-journaling/Program.cs b/docs/site/src/content/docs/grains/journaling/snippets/redis-journaling/Program.cs index 9e77e6f2eac..6930ca434d1 100644 --- a/docs/site/src/content/docs/grains/journaling/snippets/redis-journaling/Program.cs +++ b/docs/site/src/content/docs/grains/journaling/snippets/redis-journaling/Program.cs @@ -1,6 +1,7 @@ using System.Text.Json.Serialization; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Options; using Orleans; using Orleans.Journaling; using Orleans.Journaling.Json; @@ -22,32 +23,41 @@ }); using var host = builder.Build(); -await host.StartAsync(); +var cancellationToken = host.Services.GetRequiredService().ApplicationStopping; +try +{ + await host.StartAsync(cancellationToken); + + var grain = host.Services.GetRequiredService() + .GetGrain("demo"); + var written = await grain.Increment(cancellationToken); -var grain = host.Services.GetRequiredService() - .GetGrain("demo"); -var written = await grain.Increment(); + await grain.Deactivate(cancellationToken); + await Task.Delay(TimeSpan.FromMilliseconds(500), cancellationToken); -await grain.Deactivate(); -await Task.Delay(TimeSpan.FromMilliseconds(500)); + var recovered = await grain.GetSnapshot(cancellationToken); + if (written.ActivationId == recovered.ActivationId || + written.Count != recovered.Count) + { + throw new InvalidOperationException("Redis journal recovery failed."); + } -var recovered = await grain.GetSnapshot(); -if (written.ActivationId == recovered.ActivationId || - written.Count != recovered.Count) + Console.WriteLine($"Recovered count: {recovered.Count}"); +} +finally { - throw new InvalidOperationException("Redis journal recovery failed."); + using var shutdown = new CancellationTokenSource( + host.Services.GetRequiredService>().Value.ShutdownTimeout); + await host.StopAsync(shutdown.Token); } -Console.WriteLine($"Recovered count: {recovered.Count}"); -await host.StopAsync(); - public interface IRedisJournalCounterGrain : IGrainWithStringKey { - ValueTask Increment(); + ValueTask Increment(CancellationToken cancellationToken); - ValueTask GetSnapshot(); + ValueTask GetSnapshot(CancellationToken cancellationToken); - ValueTask Deactivate(); + ValueTask Deactivate(CancellationToken cancellationToken); } // @@ -57,18 +67,23 @@ public sealed class RedisJournalCounterGrain( { private readonly Guid _activationId = Guid.NewGuid(); - public async ValueTask Increment() + public async ValueTask Increment(CancellationToken cancellationToken) { + cancellationToken.ThrowIfCancellationRequested(); count.Value++; - await WriteStateAsync(); + await WriteStateAsync(cancellationToken); return GetCurrentSnapshot(); } - public ValueTask GetSnapshot() => - ValueTask.FromResult(GetCurrentSnapshot()); + public ValueTask GetSnapshot(CancellationToken cancellationToken) + { + cancellationToken.ThrowIfCancellationRequested(); + return ValueTask.FromResult(GetCurrentSnapshot()); + } - public ValueTask Deactivate() + public ValueTask Deactivate(CancellationToken cancellationToken) { + cancellationToken.ThrowIfCancellationRequested(); DeactivateOnIdle(); return ValueTask.CompletedTask; } diff --git a/docs/site/src/content/docs/snippets/compiled/Grains/JournalingSnippets.cs b/docs/site/src/content/docs/snippets/compiled/Grains/JournalingSnippets.cs index 62c996c736a..da3e3edeafe 100644 --- a/docs/site/src/content/docs/snippets/compiled/Grains/JournalingSnippets.cs +++ b/docs/site/src/content/docs/snippets/compiled/Grains/JournalingSnippets.cs @@ -12,8 +12,8 @@ namespace Documentation.Grains.Journaling; // public interface IShoppingCartGrain : IGrainWithStringKey { - ValueTask AddItem(string itemId, int quantity); - ValueTask> GetItems(); + ValueTask AddItem(string itemId, int quantity, CancellationToken cancellationToken); + ValueTask> GetItems(CancellationToken cancellationToken); } public sealed class ShoppingCartGrain( @@ -21,14 +21,18 @@ public sealed class ShoppingCartGrain( [FromKeyedServices("cart")] IDurableDictionary cart) : Grain, IShoppingCartGrain { - public async ValueTask AddItem(string itemId, int quantity) + public async ValueTask AddItem(string itemId, int quantity, CancellationToken cancellationToken) { + cancellationToken.ThrowIfCancellationRequested(); cart[itemId] = quantity; - await stateManager.WriteStateAsync(CancellationToken.None); + await stateManager.WriteStateAsync(cancellationToken); } - public ValueTask> GetItems() => - new(cart.ToDictionary()); + public ValueTask> GetItems(CancellationToken cancellationToken) + { + cancellationToken.ThrowIfCancellationRequested(); + return new(cart.ToDictionary()); + } } // diff --git a/samples/JournalingAzureBlobJson/JournalingAzureBlobJson/Program.cs b/samples/JournalingAzureBlobJson/JournalingAzureBlobJson/Program.cs index 87259b00529..a9ba320ca2b 100644 --- a/samples/JournalingAzureBlobJson/JournalingAzureBlobJson/Program.cs +++ b/samples/JournalingAzureBlobJson/JournalingAzureBlobJson/Program.cs @@ -8,6 +8,7 @@ using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Hosting; using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Options; using Orleans.Configuration; using Orleans.Journaling; using Orleans.Journaling.Json; @@ -38,14 +39,9 @@ public static async Task Main(string[] args) var blobServiceClient = new BlobServiceClient(settings.ConnectionString); var container = blobServiceClient.GetBlobContainerClient(settings.ContainerName); - await container.CreateIfNotExistsAsync(); var walBlobName = $"{settings.BlobName}/wal"; var blob = container.GetAppendBlobClient(walBlobName); - if (settings.ResetBlob) - { - await blob.DeleteIfExistsAsync(); - } builder.UseOrleans(siloBuilder => { @@ -68,35 +64,48 @@ public static async Task Main(string[] args) }); using var host = builder.Build(); + var cancellationToken = host.Services.GetRequiredService().ApplicationStopping; - await host.StartAsync(); - - var client = host.Services.GetRequiredService(); - var grain = client.GetGrain("azure-json-codec"); + try + { + await host.StartAsync(cancellationToken); + await container.CreateIfNotExistsAsync(cancellationToken: cancellationToken); + if (settings.ResetBlob) + { + await blob.DeleteIfExistsAsync(cancellationToken: cancellationToken); + } - Console.WriteLine("Writing durable grain state..."); - var written = await grain.RunScenario(); - Console.WriteLine(JsonSerializer.Serialize(written, JournalingSampleJsonContext.Default.JournaledSampleSummary)); + var client = host.Services.GetRequiredService(); + var grain = client.GetGrain("azure-json-codec"); - Console.WriteLine("Deactivating and reactivating the grain to force recovery..."); - await grain.Deactivate(); - await Task.Delay(TimeSpan.FromMilliseconds(500)); + Console.WriteLine("Writing durable grain state..."); + var written = await grain.RunScenario(cancellationToken); + Console.WriteLine(JsonSerializer.Serialize(written, JournalingSampleJsonContext.Default.JournaledSampleSummary)); - var recovered = await grain.GetSummary(); - ValidateRecovery(written, recovered); - Console.WriteLine(JsonSerializer.Serialize(recovered, JournalingSampleJsonContext.Default.JournaledSampleSummary)); + Console.WriteLine("Deactivating and reactivating the grain to force recovery..."); + await grain.Deactivate(cancellationToken); + await Task.Delay(TimeSpan.FromMilliseconds(500), cancellationToken); - Console.WriteLine(); - Console.WriteLine($"Raw JSONL blob contents from {settings.ContainerName}/{walBlobName}:"); - Console.WriteLine(await DownloadBlobAsText(blob)); + var recovered = await grain.GetSummary(cancellationToken); + ValidateRecovery(written, recovered); + Console.WriteLine(JsonSerializer.Serialize(recovered, JournalingSampleJsonContext.Default.JournaledSampleSummary)); - await host.StopAsync(); - return 0; + Console.WriteLine(); + Console.WriteLine($"Raw JSONL blob contents from {settings.ContainerName}/{walBlobName}:"); + Console.WriteLine(await DownloadBlobAsText(blob, cancellationToken)); + return 0; + } + finally + { + using var shutdown = new CancellationTokenSource( + host.Services.GetRequiredService>().Value.ShutdownTimeout); + await host.StopAsync(shutdown.Token); + } } - private static async Task DownloadBlobAsText(AppendBlobClient blob) + private static async Task DownloadBlobAsText(AppendBlobClient blob, CancellationToken cancellationToken) { - var result = await blob.DownloadContentAsync(); + var result = await blob.DownloadContentAsync(cancellationToken); return Encoding.UTF8.GetString(result.Value.Content.ToArray()); } @@ -230,9 +239,9 @@ public static SampleSettings Parse(string[] args, string? aspireBlobConnectionSt public interface IJournaledSampleGrain : IGrainWithStringKey { - Task RunScenario(); - Task GetSummary(); - Task Deactivate(); + Task RunScenario(CancellationToken cancellationToken); + Task GetSummary(CancellationToken cancellationToken); + Task Deactivate(CancellationToken cancellationToken); } public sealed class JournaledSampleGrain( @@ -246,8 +255,9 @@ public sealed class JournaledSampleGrain( { private readonly Guid _activationId = Guid.NewGuid(); - public async Task RunScenario() + public async Task RunScenario(CancellationToken cancellationToken) { + cancellationToken.ThrowIfCancellationRequested(); inventory.Clear(); inventory["sku-apple"] = new InventoryItem("sku-apple", 12, 1.25m, ["fresh", "fruit"]); inventory["sku-orange"] = new InventoryItem("sku-orange", 9, 1.10m, ["citrus", "fruit"]); @@ -291,14 +301,19 @@ public async Task RunScenario() receipt.TrySetResult(new Receipt("receipt-001", OperationCount: 24, CompletedAt: DateTimeOffset.UtcNow)); - await WriteStateAsync(); + await WriteStateAsync(cancellationToken); return CreateSummary(); } - public Task GetSummary() => Task.FromResult(CreateSummary()); + public Task GetSummary(CancellationToken cancellationToken) + { + cancellationToken.ThrowIfCancellationRequested(); + return Task.FromResult(CreateSummary()); + } - public Task Deactivate() + public Task Deactivate(CancellationToken cancellationToken) { + cancellationToken.ThrowIfCancellationRequested(); DeactivateOnIdle(); return Task.CompletedTask; } diff --git a/samples/JournalingAzureBlobJson/README.md b/samples/JournalingAzureBlobJson/README.md index ad5c1ec71a9..1cc85b17ee4 100644 --- a/samples/JournalingAzureBlobJson/README.md +++ b/samples/JournalingAzureBlobJson/README.md @@ -5,6 +5,9 @@ Azure Blob Storage. It exercises durable dictionaries, lists, queues, sets, valu task completion sources, and regular persistent state, then deactivates and reactivates the grain to verify recovery. +The host's stopping token flows through grain calls, journal writes, and Azure +storage operations. Graceful shutdown uses the configured host shutdown deadline. + ## Run the sample Install the .NET 10 SDK, the Aspire CLI, and a Docker-compatible container runtime. diff --git a/src/Orleans.Journaling/README.md b/src/Orleans.Journaling/README.md index 29fedef9b49..a8514387d11 100644 --- a/src/Orleans.Journaling/README.md +++ b/src/Orleans.Journaling/README.md @@ -16,17 +16,12 @@ dotnet add package Microsoft.Orleans.Journaling ```csharp using System.Text.Json; using System.Text.Json.Serialization; +using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Hosting; using Orleans.Hosting; using Orleans.Journaling; using Orleans.Journaling.Json; -[JsonSerializable(typeof(DateTime))] -[JsonSerializable(typeof(int))] -[JsonSerializable(typeof(string))] -[JsonSerializable(typeof(ulong))] -internal partial class JournalJsonContext : JsonSerializerContext; - var builder = Host.CreateApplicationBuilder(args) .UseOrleans(siloBuilder => { @@ -36,7 +31,15 @@ var builder = Host.CreateApplicationBuilder(args) .UseJsonJournalFormat(JournalJsonContext.Default); }); -await builder.Build().RunAsync(); +using var host = builder.Build(); +var cancellationToken = host.Services.GetRequiredService().ApplicationStopping; +await host.RunAsync(cancellationToken); + +[JsonSerializable(typeof(DateTime))] +[JsonSerializable(typeof(int))] +[JsonSerializable(typeof(string))] +[JsonSerializable(typeof(ulong))] +internal partial class JournalJsonContext : JsonSerializerContext; ``` If you need to customize `JsonSerializerOptions`, configure `JsonJournalOptions` through the options pipeline: @@ -73,8 +76,8 @@ using Orleans.Journaling; public interface IShoppingCartGrain : IGrainWithStringKey { - ValueTask AddItem(string itemId, int quantity); - ValueTask> GetItems(); + ValueTask AddItem(string itemId, int quantity, CancellationToken cancellationToken); + ValueTask> GetItems(CancellationToken cancellationToken); } public sealed class ShoppingCartGrain( @@ -82,21 +85,26 @@ public sealed class ShoppingCartGrain( [FromKeyedServices("cart")] IDurableDictionary cart) : Grain, IShoppingCartGrain { - public async ValueTask AddItem(string itemId, int quantity) + public async ValueTask AddItem(string itemId, int quantity, CancellationToken cancellationToken) { + cancellationToken.ThrowIfCancellationRequested(); cart[itemId] = quantity; - await stateManager.WriteStateAsync(CancellationToken.None); + await stateManager.WriteStateAsync(cancellationToken); } - public ValueTask> GetItems() => new(cart.ToDictionary()); + public ValueTask> GetItems(CancellationToken cancellationToken) + { + cancellationToken.ThrowIfCancellationRequested(); + return new(cart.ToDictionary()); + } } ``` -The standard grain-scoped factory registered by `AddJournalStorage` enrolls the state manager in the grain lifecycle before returning it. Constructor-injected durable states register with that manager, and recovery completes before `OnActivateAsync` and grain requests. This works with `Grain` or an application-owned grain base class. Registering journal storage makes the services available; only activations which resolve the manager perform per-grain journal I/O. +The standard state manager registered by `AddJournalStorage` enrolls itself in the grain lifecycle when constructed with the activation's `IGrainContext`. Constructor-injected durable states register with that manager, and recovery completes before `OnActivateAsync` and grain requests. This works with `Grain` or an application-owned grain base class. Registering journal storage makes the services available; only activations which resolve the manager perform per-grain journal I/O. -`DurableGrain` remains an optional convenience base exposing `StateManager`, `GetOrCreateState`, and `WriteStateAsync`. Every grain-scoped manager registration owns lifecycle enrollment before returning the manager. Custom registrations follow the same contract, enrolling their `ILifecycleParticipant` or subscribing their initialization and shutdown callbacks directly. +`DurableGrain` remains an optional convenience base exposing `StateManager`, `GetOrCreateState`, and `WriteStateAsync`. Grain-scoped managers are enrolled before resolution returns. Custom managers establish this in their constructor or registration factory, enrolling their `ILifecycleParticipant` or subscribing their initialization and shutdown callbacks directly. -Managers created with an explicit `JournalId` through `IJournaledStateManagerFactory`, or constructed manually, retain caller-owned initialization and disposal even when created inside a grain call. Register their states and await `InitializeAsync(CancellationToken.None)` before use, or deliberately supply them through a registration which assigns lifecycle ownership. Creation through the explicit-journal factory keeps failure handling independent of the ambient grain context, including when the caller subsequently enrolls the manager in a lifecycle. The asynchronous `IJournaledStateManager` methods require a cancellation-token argument. +Managers created with an explicit `JournalId` through `IJournaledStateManagerFactory`, or constructed directly from storage without a grain context, retain caller-owned initialization and disposal even when created inside a grain call. Register their states and await `InitializeAsync(cancellationToken)` with the operation's token before use, or deliberately supply them through a registration which assigns lifecycle ownership. Creation through the explicit-journal factory keeps failure handling independent of the ambient grain context, including when the caller subsequently enrolls the manager in a lifecycle. Flow the caller's cancellation token through asynchronous grain and state-manager operations. All durable state types use the configured JSON codec automatically. Configure `JsonJournalOptions` to control the `JsonSerializerOptions` instance used for entry payloads. Journaling command names and record shape are fixed by the storage format, so serializer naming policies only affect user payload values.