From 0f5524a3aaf0bec5b69c356180adaa9129f342a9 Mon Sep 17 00:00:00 2001 From: Reuben Bond Date: Wed, 16 Sep 2026 12:31:44 -0700 Subject: [PATCH 1/4] fix(journaling): harden recovery boundaries --- .../JournaledStateManager.cs | 44 ++++++++- .../StateManagerTests.cs | 99 ++++++++++++++++++- 2 files changed, 136 insertions(+), 7 deletions(-) diff --git a/src/Orleans.Journaling/JournaledStateManager.cs b/src/Orleans.Journaling/JournaledStateManager.cs index 658e3594695..05170daa152 100644 --- a/src/Orleans.Journaling/JournaledStateManager.cs +++ b/src/Orleans.Journaling/JournaledStateManager.cs @@ -29,6 +29,7 @@ internal sealed partial class JournaledStateManager : IJournaledStateManager, IJ private readonly RetiredStateTracker _retirementTracker; private Task? _workLoop; private ManagerState _state; + private long _recoveryGeneration; private bool _migrationSnapshotRequired; private int _disposed; @@ -138,6 +139,7 @@ public async ValueTask InitializeAsync(CancellationToken cancellationToken) bool didEnqueue; lock (_lock) { + ThrowIfInitializationFenced(); if (_workLoop is null) { _workLoop = Start(); @@ -196,6 +198,8 @@ private async Task WorkLoop() { _state = ManagerState.Ready; } + + _recoveryGeneration++; } needsRecovery = false; @@ -207,7 +211,7 @@ private async Task WorkLoop() { if (fenceOnFailure) { - _state = ManagerState.Ready; + _state = ManagerState.Fenced; } } @@ -248,6 +252,13 @@ private async Task WorkLoop() try { + if (workItem is AppendJournalWorkItem or WriteSnapshotWorkItem + && workItem.RecoveryGeneration != Volatile.Read(ref _recoveryGeneration)) + { + throw new InvalidOperationException( + "The journaled state operation was queued before recovery discarded its mutations. Retry the operation."); + } + // Note that the implementation of each command is inlined to avoid allocating unnecessary async states. // We are ok sacrificing some code organization for performance in the inner loop. switch (workItem) @@ -464,6 +475,8 @@ private async Task WorkLoop() var id = _journalStreamDirectory.GetNextJournalStreamId(); _journalStreamDirectory.Set(name, id); } + + _recoveryGeneration++; } break; } @@ -474,6 +487,7 @@ private async Task WorkLoop() lock (_lock) { _state = ManagerState.Ready; + _recoveryGeneration++; } break; @@ -483,6 +497,7 @@ private async Task WorkLoop() { lock (_lock) { + ThrowIfInitializationFenced(); _state = ManagerState.Ready; } break; @@ -541,7 +556,7 @@ private async Task WorkLoop() } } - if (IsRecoverySignal(exception)) + if (workItem is not RevertPendingChangesWorkItem && IsRecoverySignal(exception)) { Debug.Assert(recoveryTrigger is null); recoveryTrigger = workItem; @@ -573,7 +588,10 @@ private async Task WorkLoop() } catch (Exception exception) { - needsRecovery = true; + lock (_lock) + { + needsRecovery = _state is not ManagerState.Fenced; + } if (_shutdownCancellation.Token.IsCancellationRequested) { CompleteRecoveryTrigger(); @@ -742,6 +760,14 @@ private async Task RecoverAsync(CancellationToken cancellationToken) lock (_lock) { + foreach (var (name, state) in _states) + { + if (state is not RetiredState && !_journalStreamDirectory.ContainsKey(name)) + { + _journalStreamDirectory.Set(name, _journalStreamDirectory.GetNextJournalStreamId()); + } + } + foreach ((var name, var state) in _states) { state.OnRecoveryCompleted(); @@ -943,6 +969,15 @@ private void ThrowIfStateOperationsUnavailable() } } + private void ThrowIfInitializationFenced() + { + if (_state is ManagerState.Fenced) + { + throw new InvalidOperationException( + "The journaled state manager is fenced because recovery failed. Call RevertPendingChangesAsync to retry recovery."); + } + } + private async ValueTask ReadStorageAsync(IJournalStorageConsumer consumer, CancellationToken cancellationToken) { var startTimestamp = _shared.TimeProvider.GetTimestamp(); @@ -1021,6 +1056,7 @@ private Task EnqueueOrGetPendingWorkItem(out bool didEnqueue) var newWorkItem = new TWorkItem(); newWorkItem.EnqueuedTimestamp = _shared.TimeProvider.GetTimestamp(); + newWorkItem.RecoveryGeneration = _recoveryGeneration; newWorkItem.RecordTraceContext(); _workQueue.Enqueue(newWorkItem); didEnqueue = true; @@ -1114,6 +1150,8 @@ private abstract class WorkItem : TaskCompletionSource public long EnqueuedTimestamp { get; set; } + public long RecoveryGeneration { get; set; } + public ActivityContext TraceParent { get; private set; } public IReadOnlyList? TraceLinks => _links; diff --git a/test/Orleans.Journaling.Tests/StateManagerTests.cs b/test/Orleans.Journaling.Tests/StateManagerTests.cs index 10ff9aa5c58..7a54c92a232 100644 --- a/test/Orleans.Journaling.Tests/StateManagerTests.cs +++ b/test/Orleans.Journaling.Tests/StateManagerTests.cs @@ -628,7 +628,7 @@ await sut.Manager.WriteStateAsync(TestContext.Current.CancellationToken).AsTask( } [Fact] - public async Task StateManager_WriteStateAsync_RetriesRecoveryAfterRepeatedFailures() + public async Task StateManager_WriteStateAsync_FencesWritesAfterRepeatedRecoveryFailures() { var storage = new CapturingStorage(); var sut = CreateTestSystem(storage: storage); @@ -652,13 +652,22 @@ public async Task StateManager_WriteStateAsync_RetriesRecoveryAfterRepeatedFailu var secondRecoveryFailure = new IOException("Expected second recovery failure."); storage.NextReadException = secondRecoveryFailure; var recoveryException = await Assert.ThrowsAsync( - () => sut.Manager.WriteStateAsync(TestContext.Current.CancellationToken).AsTask() + () => sut.Manager.RevertPendingChangesAsync(TestContext.Current.CancellationToken).AsTask() .WaitAsync(TimeSpan.FromSeconds(10), TestContext.Current.CancellationToken)); Assert.Same(secondRecoveryFailure, recoveryException); - await sut.Manager.WriteStateAsync(TestContext.Current.CancellationToken).AsTask() - .WaitAsync(TimeSpan.FromSeconds(10), TestContext.Current.CancellationToken); + var writeException = await Assert.ThrowsAsync( + () => sut.Manager.WriteStateAsync(TestContext.Current.CancellationToken).AsTask() + .WaitAsync(TimeSpan.FromSeconds(10), TestContext.Current.CancellationToken)); + Assert.Contains("fenced", writeException.Message, StringComparison.OrdinalIgnoreCase); + + var initializeException = await Assert.ThrowsAsync( + () => sut.Manager.InitializeAsync(TestContext.Current.CancellationToken).AsTask() + .WaitAsync(TimeSpan.FromSeconds(10), TestContext.Current.CancellationToken)); + Assert.Contains("fenced", initializeException.Message, StringComparison.OrdinalIgnoreCase); + await sut.Manager.RevertPendingChangesAsync(TestContext.Current.CancellationToken).AsTask() + .WaitAsync(TimeSpan.FromSeconds(10), TestContext.Current.CancellationToken); Assert.True(dictionary.ContainsKey("first")); Assert.False(dictionary.ContainsKey("second")); } @@ -688,6 +697,82 @@ public async Task StateManager_RevertPendingChanges_RestoresLastDurableState() Assert.Equal(1, value.Value); } + [Fact] + public async Task StateManager_RecoveryRejectsWritesQueuedBeforeMutationsWereDiscarded() + { + var storage = new BlockingAppendStorage + { + FirstAppendException = new InconsistentStateException("Expected write conflict.") + }; + var sut = CreateTestSystem(storage: storage); + var value = new DurableValue("value", sut.Manager, CreateValueCodec()); + await sut.Lifecycle.OnStart(TestContext.Current.CancellationToken); + value.Value = 1; + + var conflictingWrite = sut.Manager.WriteStateAsync(TestContext.Current.CancellationToken).AsTask(); + await storage.FirstAppendStarted.Task.WaitAsync( + TimeSpan.FromSeconds(10), + TestContext.Current.CancellationToken); + value.Value = 2; + var queuedWrite = sut.Manager.WriteStateAsync(TestContext.Current.CancellationToken).AsTask(); + + storage.AllowFirstAppend.TrySetResult(); + + await Assert.ThrowsAsync( + () => conflictingWrite.WaitAsync( + TimeSpan.FromSeconds(10), + TestContext.Current.CancellationToken)); + var exception = await Assert.ThrowsAsync( + () => queuedWrite.WaitAsync( + TimeSpan.FromSeconds(10), + TestContext.Current.CancellationToken)); + + Assert.Contains("queued before recovery discarded its mutations", exception.Message, StringComparison.Ordinal); + Assert.Empty(storage.Appends); + } + + [Fact] + public async Task StateManager_InconsistentStateDuringExplicitRevertRemainsFenced() + { + var storage = new CapturingStorage(); + var sut = CreateTestSystem(storage: storage); + var value = new DurableValue("value", sut.Manager, CreateValueCodec()); + + await sut.Lifecycle.OnStart(TestContext.Current.CancellationToken); + value.Value = 1; + await sut.Manager.WriteStateAsync(TestContext.Current.CancellationToken); + value.Value = 2; + storage.NextReadException = new InconsistentStateException("Expected explicit recovery conflict."); + + await Assert.ThrowsAsync( + () => sut.Manager.RevertPendingChangesAsync(TestContext.Current.CancellationToken).AsTask()); + var writeException = await Assert.ThrowsAsync( + () => sut.Manager.WriteStateAsync(TestContext.Current.CancellationToken).AsTask()); + Assert.Contains("fenced", writeException.Message, StringComparison.OrdinalIgnoreCase); + + await sut.Manager.RevertPendingChangesAsync(TestContext.Current.CancellationToken); + Assert.Equal(1, value.Value); + } + + [Fact] + public async Task StateManager_RevertAgainstEmptyJournalResetsUncommittedState() + { + var storage = new CapturingStorage(); + var sut = CreateTestSystem(storage: storage); + var value = new DurableValue("value", sut.Manager, CreateValueCodec()); + await sut.Lifecycle.OnStart(TestContext.Current.CancellationToken); + value.Value = 42; + + await sut.Manager.RevertPendingChangesAsync(TestContext.Current.CancellationToken); + + Assert.Equal(0, value.Value); + await sut.Manager.WriteStateAsync(TestContext.Current.CancellationToken); + var recovered = CreateTestSystem(storage: storage); + var recoveredValue = new DurableValue("value", recovered.Manager, CreateValueCodec()); + await recovered.Lifecycle.OnStart(TestContext.Current.CancellationToken); + Assert.Equal(0, recoveredValue.Value); + } + [Fact] public async Task StateManager_FencesStateOperationsUntilFailedRevertIsRetriedSuccessfully() { @@ -1905,6 +1990,8 @@ private sealed class BlockingAppendStorage : IJournalStorage public TaskCompletionSource AllowFirstAppend { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously); + public Exception? FirstAppendException { get; init; } + public List Appends { get; } = []; public bool IsCompactionRequested => false; @@ -1924,6 +2011,10 @@ public async ValueTask AppendAsync(ReadOnlySequence value, CancellationTok { FirstAppendStarted.SetResult(); await AllowFirstAppend.Task.WaitAsync(cancellationToken); + if (FirstAppendException is { } exception) + { + throw exception; + } } Appends.Add(value.ToArray()); From 24a60b46fa5367850665e2793a894f4623fd1909 Mon Sep 17 00:00:00 2001 From: Reuben Bond Date: Wed, 16 Sep 2026 14:14:35 -0700 Subject: [PATCH 2/4] fix(journaling): close recovery queue races --- .../JournaledStateManager.cs | 24 ++++++++++++------- .../StateManagerTests.cs | 5 ++++ 2 files changed, 20 insertions(+), 9 deletions(-) diff --git a/src/Orleans.Journaling/JournaledStateManager.cs b/src/Orleans.Journaling/JournaledStateManager.cs index 05170daa152..7fd77185757 100644 --- a/src/Orleans.Journaling/JournaledStateManager.cs +++ b/src/Orleans.Journaling/JournaledStateManager.cs @@ -205,13 +205,14 @@ private async Task WorkLoop() needsRecovery = false; CompleteRecoveryTrigger(); } - catch + catch (Exception exception) { lock (_lock) { if (fenceOnFailure) { _state = ManagerState.Fenced; + FaultQueuedWorkItemsUnderLock(exception); } } @@ -591,6 +592,10 @@ private async Task WorkLoop() lock (_lock) { needsRecovery = _state is not ManagerState.Fenced; + if (needsRecovery) + { + FaultQueuedWorkItemsUnderLock(exception); + } } if (_shutdownCancellation.Token.IsCancellationRequested) { @@ -600,7 +605,6 @@ private async Task WorkLoop() try { - FaultQueuedWorkItems(exception); LogErrorProcessingWorkItems(_shared.Logger, exception); } finally @@ -623,15 +627,11 @@ void CompleteRecoveryTrigger() } } - private void FaultQueuedWorkItems(Exception exception) + private void FaultQueuedWorkItemsUnderLock(Exception exception) { - lock (_lock) + while (_workQueue.TryDequeue(out var workItem)) { - while (_workQueue.TryDequeue(out var workItem)) - { - workItem.TrySetException(exception); - } - + workItem.TrySetException(exception); } } @@ -1048,6 +1048,12 @@ private Task EnqueueOrGetPendingWorkItem(out bool didEnqueue) continue; } + if (workItem is AppendJournalWorkItem or WriteSnapshotWorkItem + && workItem.RecoveryGeneration != _recoveryGeneration) + { + continue; + } + workItem.RecordTraceContext(); workItem.AddCaller(); didEnqueue = false; diff --git a/test/Orleans.Journaling.Tests/StateManagerTests.cs b/test/Orleans.Journaling.Tests/StateManagerTests.cs index 7a54c92a232..3f1e82edbe6 100644 --- a/test/Orleans.Journaling.Tests/StateManagerTests.cs +++ b/test/Orleans.Journaling.Tests/StateManagerTests.cs @@ -729,6 +729,11 @@ await Assert.ThrowsAsync( Assert.Contains("queued before recovery discarded its mutations", exception.Message, StringComparison.Ordinal); Assert.Empty(storage.Appends); + + value.Value = 3; + await sut.Manager.WriteStateAsync(TestContext.Current.CancellationToken); + + Assert.Single(storage.Appends); } [Fact] From 893088b927aceda48d7c0e933e555c673cd3347d Mon Sep 17 00:00:00 2001 From: Reuben Bond Date: Wed, 16 Sep 2026 14:38:54 -0700 Subject: [PATCH 3/4] fix(journaling): preserve fenced recovery queue --- .../JournaledStateManager.cs | 25 +++++--- .../StateManagerTests.cs | 64 +++++++++++++++++++ 2 files changed, 81 insertions(+), 8 deletions(-) diff --git a/src/Orleans.Journaling/JournaledStateManager.cs b/src/Orleans.Journaling/JournaledStateManager.cs index 7fd77185757..9b08bee8332 100644 --- a/src/Orleans.Journaling/JournaledStateManager.cs +++ b/src/Orleans.Journaling/JournaledStateManager.cs @@ -569,16 +569,19 @@ private async Task WorkLoop() needsRecovery = true; } - else + else if (workItem is RevertPendingChangesWorkItem) { - workItem.SetException(exception); - if (workItem is RevertPendingChangesWorkItem) + lock (_lock) { - lock (_lock) - { - _state = ManagerState.Fenced; - } + _state = ManagerState.Fenced; + FaultQueuedWorkItemsUnderLock(exception); } + + workItem.SetException(exception); + } + else + { + workItem.SetException(exception); } } finally @@ -764,7 +767,13 @@ private async Task RecoverAsync(CancellationToken cancellationToken) { if (state is not RetiredState && !_journalStreamDirectory.ContainsKey(name)) { - _journalStreamDirectory.Set(name, _journalStreamDirectory.GetNextJournalStreamId()); + var id = _journalStreamDirectory.GetNextJournalStreamId(); + while (_statesMap.ContainsKey(id)) + { + id = checked(id + 1); + } + + _journalStreamDirectory.Set(name, id); } } diff --git a/test/Orleans.Journaling.Tests/StateManagerTests.cs b/test/Orleans.Journaling.Tests/StateManagerTests.cs index 3f1e82edbe6..3e70fe19cc7 100644 --- a/test/Orleans.Journaling.Tests/StateManagerTests.cs +++ b/test/Orleans.Journaling.Tests/StateManagerTests.cs @@ -759,6 +759,34 @@ await Assert.ThrowsAsync( Assert.Equal(1, value.Value); } + [Fact] + public async Task StateManager_FailedExplicitRevertFaultsQueuedInitialization() + { + var expected = new IOException("Expected explicit recovery failure."); + var storage = new BlockingRecoveryStorage + { + RecoveryReadException = expected + }; + var sut = CreateTestSystem(storage: storage); + await sut.Lifecycle.OnStart(TestContext.Current.CancellationToken); + + var revert = sut.Manager.RevertPendingChangesAsync(TestContext.Current.CancellationToken).AsTask(); + await storage.RecoveryReadStarted.Task.WaitAsync( + TimeSpan.FromSeconds(10), + TestContext.Current.CancellationToken); + var initialize = sut.Manager.InitializeAsync(TestContext.Current.CancellationToken).AsTask(); + + storage.AllowRecoveryRead.SetResult(); + + var revertException = await Assert.ThrowsAsync(() => revert); + Assert.Same(expected, revertException); + var initializeException = await Assert.ThrowsAsync(() => initialize); + Assert.Same(expected, initializeException); + var writeException = await Assert.ThrowsAsync( + () => sut.Manager.WriteStateAsync(TestContext.Current.CancellationToken).AsTask()); + Assert.Contains("fenced", writeException.Message, StringComparison.OrdinalIgnoreCase); + } + [Fact] public async Task StateManager_RevertAgainstEmptyJournalResetsUncommittedState() { @@ -1062,6 +1090,36 @@ public async Task StateManager_RegisterState_AfterRecovery_AllocatesAboveRecover Assert.Contains(entries, entry => entry.StreamId.Value == recoveredStreamId.Value + 1); } + [Fact] + public async Task StateManager_Recovery_RebindsStateAboveUnknownStreamIds() + { + var storage = new CapturingStorage { IsCompactionRequested = true }; + var unknownStreamId = new JournalStreamId(8); + var unknownPayload = new byte[] { 1, 2, 3 }; + using (var segment = new OrleansBinaryJournalBufferWriter()) + { + using (var entry = segment.CreateJournalStreamWriter(unknownStreamId).BeginEntry()) + { + entry.Writer.Write(unknownPayload); + entry.Commit(); + } + + using var committed = segment.GetBuffer(); + await storage.AppendAsync(committed.AsReadOnlySequence(), CancellationToken.None); + } + + var sut = CreateTestSystem(storage: storage); + var value = new DurableValue("value", sut.Manager, CreateValueCodec()); + await sut.Lifecycle.OnStart(TestContext.Current.CancellationToken); + value.Value = 42; + + await sut.Manager.WriteStateAsync(TestContext.Current.CancellationToken); + + var entries = ReadBinaryEntries(Assert.Single(storage.Replaces)); + Assert.Contains(entries, entry => entry.StreamId == unknownStreamId && entry.Payload.SequenceEqual(unknownPayload)); + Assert.Contains(entries, entry => entry.StreamId.Value == unknownStreamId.Value + 1); + } + [Fact] public async Task StateManager_Recovery_ReadsConcatenatedJournalData() { @@ -2076,11 +2134,17 @@ public async ValueTask ReadAsync(IJournalStorageConsumer consumer, CancellationT { RecoveryReadStarted.SetResult(); await AllowRecoveryRead.Task.WaitAsync(cancellationToken); + if (RecoveryReadException is { } exception) + { + throw exception; + } } consumer.Complete(metadata: null); } + public Exception? RecoveryReadException { get; init; } + public ValueTask ReplaceAsync(ReadOnlySequence value, CancellationToken cancellationToken) => default; public ValueTask AppendAsync(ReadOnlySequence value, CancellationToken cancellationToken) => default; From 7c7f8f735e627147f5a037559b6257d7bb73ffa1 Mon Sep 17 00:00:00 2001 From: Reuben Bond Date: Wed, 16 Sep 2026 16:00:07 -0700 Subject: [PATCH 4/4] fix(journaling)!: recover failed operations in new activations Stage only safe-to-commit mutations. Permanently fence failed managers and deactivate grain owners instead of resetting live state beneath interleaved methods. BREAKING CHANGE: Remove IJournaledStateManager.RevertPendingChangesAsync. Recovery after failure requires a fresh activation or a new standalone manager with new state instances. --- .../docs/grains/journaling/azure-storage.md | 2 +- .../content/docs/grains/journaling/index.md | 4 + .../docs/grains/journaling/operations.md | 4 +- .../journaling/redis-journal-storage.md | 2 +- .../grains/journaling/runtime-behavior.md | 35 +- .../CompatibilitySuppressions.xml | 14 + src/Orleans.Journaling/IJournaledState.cs | 16 +- .../IJournaledStateManager.cs | 25 +- .../JournaledStateManager.cs | 229 ++------ src/Orleans.Journaling/README.md | 16 + .../Orleans.Journaling/Orleans.Journaling.cs | 1 - .../DurableListJournalBenchmarks.cs | 1 - .../JournaledJobShardDiscoveryTests.cs | 1 - .../DurableCollectionDirectWriteTests.cs | 1 - .../DurableListDirectWriteTests.cs | 1 - .../StateManagerTests.cs | 540 ++++++------------ 16 files changed, 336 insertions(+), 556 deletions(-) diff --git a/docs/site/src/content/docs/grains/journaling/azure-storage.md b/docs/site/src/content/docs/grains/journaling/azure-storage.md index 02c0f17dce1..1c206e72697 100644 --- a/docs/site/src/content/docs/grains/journaling/azure-storage.md +++ b/docs/site/src/content/docs/grains/journaling/azure-storage.md @@ -60,7 +60,7 @@ Metadata-enabled catalog queries select the header's format and caller-owned met ## Optimistic concurrency -Both providers condition append, replace, and delete operations on the last observed ETag. Metadata-only conflicts receive a bounded in-place refresh and retry. A journal-content conflict raises , which causes the Journaling state manager to recover before later work. +Both providers condition append, replace, and delete operations on the last observed ETag. Metadata-only conflicts receive a bounded in-place refresh and retry. A journal-content conflict raises , which permanently fences the Journaling state manager and requests grain deactivation. A fresh activation replays the durable journal. Configure the metadata-only retry cap and backoff with the corresponding `MaxMetadataOnlyConflictRetries`, `MetadataOnlyConflictInitialBackoff`, and `MetadataOnlyConflictMaxBackoff` options. diff --git a/docs/site/src/content/docs/grains/journaling/index.md b/docs/site/src/content/docs/grains/journaling/index.md index 54fe1369d48..f91cc12b6a2 100644 --- a/docs/site/src/content/docs/grains/journaling/index.md +++ b/docs/site/src/content/docs/grains/journaling/index.md @@ -26,6 +26,10 @@ A journaling grain derives from and recei 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; +a fresh activation reconstructs the durable outcome before processing resumes. + Each named state has a stable stream identity within the grain journal. Keep those names stable across deployments so recovery can bind stored operations to the intended state. ## Journal lifecycle diff --git a/docs/site/src/content/docs/grains/journaling/operations.md b/docs/site/src/content/docs/grains/journaling/operations.md index a43bedc4f28..b7d322bcbd6 100644 --- a/docs/site/src/content/docs/grains/journaling/operations.md +++ b/docs/site/src/content/docs/grains/journaling/operations.md @@ -103,8 +103,8 @@ Exercise restore procedures regularly: | --- | --- | --- | | Activation fails with a journal format key | Recovery leaves storage unchanged | Deploy the required format and command codecs, or restore metadata which names the actual stored format | | Activation reports malformed or truncated journal data | Recovery stops before activation completes | Quarantine writes, restore a consistent backup, and preserve the failed data for diagnosis | -| during a write | The write faults and the manager recovers before later work | Find stale or competing writers, confirm grain identity and cluster configuration, then retry the application command according to its idempotency contract | -| Snapshot replacement fails after its pending append succeeds | The write faults while the mutation remains durable in the append history | Reconcile using the operation identifier and retry the command only through its idempotency protocol | +| during a write | The manager fences operations and requests deactivation; a fresh activation recovers durable state | Find stale or competing writers, confirm grain identity and cluster configuration, then retry the application command according to its idempotency contract | +| Storage commits a write but its acknowledgement is lost | The manager fences operations and requests deactivation; the new activation observes the committed mutation | Reconcile using the operation identifier and retry the command only through its idempotency protocol | | Write latency rises with storage queue duration | Writes for the journal are waiting behind earlier operations | Inspect provider latency and throttling, hot-grain traffic, write frequency, and snapshot size | | Recovery time and journal bytes grow | The current append history is large | Confirm compaction thresholds are enabled and that snapshot replacements succeed | | Format migration reports an unregistered state | A retired stream still requires its previous codec | Re-register the state for migration or retain the previous write format until retirement and compaction remove it | diff --git a/docs/site/src/content/docs/grains/journaling/redis-journal-storage.md b/docs/site/src/content/docs/grains/journaling/redis-journal-storage.md index e41b330b554..49587289afa 100644 --- a/docs/site/src/content/docs/grains/journaling/redis-journal-storage.md +++ b/docs/site/src/content/docs/grains/journaling/redis-journal-storage.md @@ -23,7 +23,7 @@ The [runnable Redis Journaling sample](samples.md#run-the-redis-sample) uses the The provider stores journal data in Redis strings and journal metadata in Redis hashes. Per-journal reads and mutations use atomic Lua scripts. Catalog operations discover journals by scanning metadata keys on each connected primary Redis server. -Optimistic concurrency protects append, replace, and delete operations. A stale writer receives , and the Journaling state manager recovers the stored journal before processing later work. Concurrent reads observe either the journal before a replacement or the complete replacement. +Optimistic concurrency protects append, replace, and delete operations. A stale writer receives , and the Journaling state manager permanently fences operations and requests grain deactivation. A fresh activation recovers the stored journal. Concurrent reads observe either the journal before a replacement or the complete replacement. supports: 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 063b8f85c61..09194a43383 100644 --- a/docs/site/src/content/docs/grains/journaling/runtime-behavior.md +++ b/docs/site/src/content/docs/grains/journaling/runtime-behavior.md @@ -35,29 +35,52 @@ 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 can rewind changes that were never acknowledged by storage. +> 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 + +All interleaved callers share the manager's pending journal. Prepare fallible work, external acknowledgements, +and proposed output in operation-local data. After establishing that an outcome is safe to commit, apply its +mutations to durable state and initiate a write. Coordinate that transition with other interleaved operations +which can affect the same decision. Any caller's write can include staged mutations from other calls. + +If an application error occurs after staging and makes those mutations unsafe to commit, end the activation's +use of the manager and request deactivation. In-flight methods can retain local decisions and references +across awaits; a fresh activation reconstructs both application and durable state together. ## Consistency and competing writers Orleans grain placement normally supplies a single active writer for a grain identity. Journal storage providers also use optimistic concurrency to protect the journal when a stale or competing writer reaches storage. -An append, replacement, or delete with stale storage metadata throws . The state manager marks the journal for recovery, resets its in-memory durable states, and replays the current stored journal before processing later work. The failed write task remains faulted so the grain call reports an uncertain outcome. +An append, replacement, or delete with stale storage metadata throws . The state manager permanently fences further operations, faults queued work, and requests grain deactivation. The failed operation reports its original exception. A new activation replays the stored journal to determine the durable outcome. Design commands to tolerate retries at the application boundary. Use operation identifiers when a caller can repeat a command after an uncertain network outcome. ## Storage failures -A normal append failure leaves encoded pending entries available for a later write attempt. +A failed append, snapshot replacement, delete, or initialization permanently fences that manager instance. +Queued operations fault, and later write, delete, registration, and initialization requests fail explicitly. +Existing in-memory state remains available to in-flight calls until deactivation completes. The grain runtime +starts deactivation as part of handling the failure. + +A storage operation can commit before its acknowledgement is lost. Treat a failed write as an uncertain +application outcome and reconcile in a new activation using an operation identifier or another idempotency +mechanism. + +Owners of standalone managers created through +dispose the failed manager and create a new one for the same . Register +new durable state instances and initialize them before resuming processing. -A compaction write has two storage stages: it first appends committed pending entries, then publishes a snapshot replacement. The append can succeed before the replacement fails. In that outcome, faults even though the state mutation is durable in the append history. Treat every failed write as an uncertain application outcome and retry commands using an operation identifier or another idempotency mechanism. +Cancelling a write's cancellation token stops the caller's wait. An already queued write continues to its +storage outcome, so the caller reconciles that outcome before retrying the command. -Recovery exceptions fault activation or queued work rather than replacing or truncating stored data. Restore the required format/codec registration or repair the backing data before retrying activation. +An initialization failure preserves stored data for diagnosis. Restore the required format/codec registration +or repair the backing data before creating a fresh manager or retrying activation. ## Compaction Each provider reports when its journal crosses a configured storage threshold. The next : -1. Persists any already-buffered append data. 1. Builds a snapshot containing the state directory and every active durable state. 1. Atomically replaces the published journal with the snapshot. 1. Clears the compaction request after storage acknowledges the replacement. diff --git a/src/Orleans.Journaling/CompatibilitySuppressions.xml b/src/Orleans.Journaling/CompatibilitySuppressions.xml index cf125a42760..940e1aff1b4 100644 --- a/src/Orleans.Journaling/CompatibilitySuppressions.xml +++ b/src/Orleans.Journaling/CompatibilitySuppressions.xml @@ -1,6 +1,13 @@  + + CP0002 + M:Orleans.Journaling.IJournaledStateManager.RevertPendingChangesAsync(System.Threading.CancellationToken) + lib/net10.0/Orleans.Journaling.dll + lib/net10.0/Orleans.Journaling.dll + true + CP0002 M:Orleans.Journaling.IJournalStorageCatalog.ListAsync(Orleans.Journaling.JournalId,System.Threading.CancellationToken) @@ -15,6 +22,13 @@ lib/net10.0/Orleans.Journaling.dll true + + CP0002 + M:Orleans.Journaling.IJournaledStateManager.RevertPendingChangesAsync(System.Threading.CancellationToken) + lib/net8.0/Orleans.Journaling.dll + lib/net8.0/Orleans.Journaling.dll + true + CP0002 M:Orleans.Journaling.IJournalStorageCatalog.ListAsync(Orleans.Journaling.JournalId,System.Threading.CancellationToken) diff --git a/src/Orleans.Journaling/IJournaledState.cs b/src/Orleans.Journaling/IJournaledState.cs index 79d9230d0a4..e69f61e9c83 100644 --- a/src/Orleans.Journaling/IJournaledState.cs +++ b/src/Orleans.Journaling/IJournaledState.cs @@ -14,7 +14,7 @@ namespace Orleans.Journaling; /// both apply the mutation locally and emit the corresponding command to the journal). /// /// -/// At the end of the grain turn, the journaled state manager calls +/// When the application requests a write, the journaled state manager calls /// (and occasionally ) to materialize /// the pending changes, then flushes the journal to durable storage. /// @@ -25,16 +25,14 @@ namespace Orleans.Journaling; /// rather than treating in-memory mutations as durable. /// /// -/// On failure, the journaled state manager triggers recovery by calling -/// followed by replaying snapshots and entries from durable storage. Implementations must -/// therefore make any volatile in-memory bookkeeping fully recoverable from the journal. +/// A failed journal operation permanently fences the manager and requests grain deactivation. +/// A new manager initializes new state instances by calling and replaying durable entries. /// /// /// -/// In other words, in-memory mutations are journaled within the same grain turn and become -/// durable when the journal flushes. Implementations are free to apply mutations eagerly -/// (before the durable write completes); recovery rebuilds in-memory state from the journal, -/// so a turn-failure-and-recovery cycle observably rewinds any unflushed changes. +/// Application code prepares fallible work in operation-local data and stages only mutations which are +/// safe to commit. Staged mutations are shared by all interleaved callers using the same manager. +/// Storage acknowledgement establishes durability; recovery takes place in a fresh manager and state instances. /// /// public interface IJournaledState @@ -55,7 +53,7 @@ public interface IJournaledState /// /// /// If the state has any volatile state, it must be cleared by this method. - /// This method can be called at any point in the state's lifetime, including during recovery. + /// The manager calls this method when binding a stream, deleting the journal, or retiring a state. /// void Reset(JournalStreamWriter writer); diff --git a/src/Orleans.Journaling/IJournaledStateManager.cs b/src/Orleans.Journaling/IJournaledStateManager.cs index 2861d6e3deb..fb1493799d6 100644 --- a/src/Orleans.Journaling/IJournaledStateManager.cs +++ b/src/Orleans.Journaling/IJournaledStateManager.cs @@ -3,7 +3,7 @@ namespace Orleans.Journaling; /// -/// Manages the states for a given grain. +/// Manages the durable states associated with a journal. /// public interface IJournaledStateManager : IAsyncDisposable { @@ -11,8 +11,11 @@ public interface IJournaledStateManager : IAsyncDisposable ValueTask IAsyncDisposable.DisposeAsync() => default; /// - /// Initializes the state manager. + /// Initializes the state manager by replaying its journal. /// + /// + /// A failed initialization permanently fences this instance. Recover by creating a new manager and new state instances. + /// /// The cancellation token. /// A which represents the operation. ValueTask InitializeAsync(CancellationToken cancellationToken); @@ -34,20 +37,24 @@ public interface IJournaledStateManager : IAsyncDisposable /// /// Prepares and persists an update to the journal. /// + /// + /// Stage mutations only after the operation has established that they are safe to commit. Pending changes + /// are shared by all callers using this manager. A write failure permanently fences the manager and requests + /// deactivation of its owning grain. Owners of standalone managers must dispose the failed instance and + /// create a new manager with new state instances to recover durable state. + /// Cancellation stops the caller's wait; an already queued write continues to completion. + /// /// The cancellation token. /// A which represents the operation. ValueTask WriteStateAsync(CancellationToken cancellationToken); - /// - /// Discards uncommitted mutations and reloads the last durable state. - /// - /// The cancellation token. - /// A which represents the operation. - ValueTask RevertPendingChangesAsync(CancellationToken cancellationToken); - /// /// Resets this instance, removing any persistent state. /// + /// + /// Quiesce other operations before deleting state: deletion resets every registered durable state. + /// A failed deletion permanently fences the manager and requests deactivation of its owning grain. + /// /// The cancellation token. /// A which represents the operation. ValueTask DeleteStateAsync(CancellationToken cancellationToken); diff --git a/src/Orleans.Journaling/JournaledStateManager.cs b/src/Orleans.Journaling/JournaledStateManager.cs index 9b08bee8332..8c30a7811e3 100644 --- a/src/Orleans.Journaling/JournaledStateManager.cs +++ b/src/Orleans.Journaling/JournaledStateManager.cs @@ -5,7 +5,6 @@ using Orleans.Diagnostics; using Orleans.Serialization.Buffers; using Orleans.Runtime.Internal; -using Orleans.Storage; namespace Orleans.Journaling; @@ -21,6 +20,7 @@ internal sealed partial class JournaledStateManager : IJournaledStateManager, IJ private readonly Dictionary _statesMap = []; private readonly JournaledStateManagerShared _shared; private readonly IJournalStorage _storage; + private readonly IGrainContext? _grainContext; private readonly JournalBufferWriter _journalWriter; private readonly SingleWaiterAutoResetEvent _workSignal = new() { RunContinuationsAsynchronously = true }; private readonly Queue _workQueue = new(); @@ -29,13 +29,15 @@ internal sealed partial class JournaledStateManager : IJournaledStateManager, IJ private readonly RetiredStateTracker _retirementTracker; private Task? _workLoop; private ManagerState _state; - private long _recoveryGeneration; + private long _deletionGeneration; + private Exception? _failure; private bool _migrationSnapshotRequired; private int _disposed; public JournaledStateManager(JournaledStateManagerShared shared, IJournalStorageProvider storageProvider, IGrainContext grainContext) : this(shared, CreateStorage(storageProvider, CreateJournalId(grainContext))) { + _grainContext = grainContext; } public JournaledStateManager(JournaledStateManagerShared shared, IJournalStorageProvider storageProvider, JournalId journalId) @@ -93,6 +95,7 @@ public void RegisterState(string name, IJournaledState state) lock (_lock) { + ThrowIfFenced(); if (_states.TryGetValue(name, out var existing)) { if (existing is RetiredState vessel) @@ -139,7 +142,7 @@ public async ValueTask InitializeAsync(CancellationToken cancellationToken) bool didEnqueue; lock (_lock) { - ThrowIfInitializationFenced(); + ThrowIfFenced(); if (_workLoop is null) { _workLoop = Start(); @@ -165,9 +168,16 @@ private Task Start() private async Task WorkLoop() { await Task.CompletedTask.ConfigureAwait(ConfigureAwaitOptions.ContinueOnCapturedContext | ConfigureAwaitOptions.ForceYielding); - var needsRecovery = true; - WorkItem? recoveryTrigger = null; - Exception? recoveryTriggerException = null; + try + { + await RecoverAsync(_shutdownCancellation.Token).ConfigureAwait(true); + } + catch (Exception exception) + { + Fence(exception); + return; + } + while (!_shutdownCancellation.Token.IsCancellationRequested) { try @@ -177,49 +187,6 @@ private async Task WorkLoop() while (true) { - if (needsRecovery) - { - bool fenceOnFailure; - lock (_lock) - { - fenceOnFailure = _state is not ManagerState.Unknown; - if (fenceOnFailure) - { - _state = ManagerState.Recovering; - } - } - - try - { - await RecoverAsync(_shutdownCancellation.Token).ConfigureAwait(true); - lock (_lock) - { - if (fenceOnFailure) - { - _state = ManagerState.Ready; - } - - _recoveryGeneration++; - } - - needsRecovery = false; - CompleteRecoveryTrigger(); - } - catch (Exception exception) - { - lock (_lock) - { - if (fenceOnFailure) - { - _state = ManagerState.Fenced; - FaultQueuedWorkItemsUnderLock(exception); - } - } - - throw; - } - } - WorkItem workItem; lock (_lock) { @@ -254,10 +221,10 @@ private async Task WorkLoop() try { if (workItem is AppendJournalWorkItem or WriteSnapshotWorkItem - && workItem.RecoveryGeneration != Volatile.Read(ref _recoveryGeneration)) + && workItem.DeletionGeneration != Volatile.Read(ref _deletionGeneration)) { throw new InvalidOperationException( - "The journaled state operation was queued before recovery discarded its mutations. Retry the operation."); + "The journaled state operation was queued before deletion reset its state."); } // Note that the implementation of each command is inlined to avoid allocating unnecessary async states. @@ -477,28 +444,16 @@ private async Task WorkLoop() _journalStreamDirectory.Set(name, id); } - _recoveryGeneration++; + _deletionGeneration++; } break; } - case RevertPendingChangesWorkItem: - { - await RecoverAsync(_shutdownCancellation.Token).ConfigureAwait(true); - lock (_lock) - { - _state = ManagerState.Ready; - _recoveryGeneration++; - } - - break; - } - case InitializeWorkItem: { lock (_lock) { - ThrowIfInitializationFenced(); + ThrowIfFenced(); _state = ManagerState.Ready; } break; @@ -557,32 +512,16 @@ private async Task WorkLoop() } } - if (workItem is not RevertPendingChangesWorkItem && IsRecoverySignal(exception)) + try { - Debug.Assert(recoveryTrigger is null); - recoveryTrigger = workItem; - recoveryTriggerException = exception; - lock (_lock) - { - _state = ManagerState.Recovering; - } - - needsRecovery = true; + Fence(exception); } - else if (workItem is RevertPendingChangesWorkItem) - { - lock (_lock) - { - _state = ManagerState.Fenced; - FaultQueuedWorkItemsUnderLock(exception); - } - - workItem.SetException(exception); - } - else + finally { workItem.SetException(exception); } + + return; } finally { @@ -592,41 +531,35 @@ private async Task WorkLoop() } catch (Exception exception) { - lock (_lock) - { - needsRecovery = _state is not ManagerState.Fenced; - if (needsRecovery) - { - FaultQueuedWorkItemsUnderLock(exception); - } - } - if (_shutdownCancellation.Token.IsCancellationRequested) - { - CompleteRecoveryTrigger(); - return; - } - - try - { - LogErrorProcessingWorkItems(_shared.Logger, exception); - } - finally - { - CompleteRecoveryTrigger(); - } + Fence(exception); + return; } } + } + + private void Fence(Exception exception) + { + lock (_lock) + { + _state = ManagerState.Fenced; + _failure = exception; + } - void CompleteRecoveryTrigger() + try { - if (recoveryTrigger is not { } trigger) + if (!_shutdownCancellation.IsCancellationRequested) { - return; + LogErrorProcessingWorkItems(_shared.Logger, exception); + _grainContext?.Deactivate(new DeactivationReason( + DeactivationReasonCode.ApplicationError, exception, "Journaled state operation failed.")); + } + } + finally + { + lock (_lock) + { + FaultQueuedWorkItemsUnderLock(exception); } - - trigger.SetException(recoveryTriggerException!); - recoveryTrigger = null; - recoveryTriggerException = null; } } @@ -638,19 +571,6 @@ private void FaultQueuedWorkItemsUnderLock(Exception exception) } } - private static bool IsRecoverySignal(Exception exception) - { - for (Exception? current = exception; current is not null; current = current.InnerException) - { - if (current is InconsistentStateException) - { - return true; - } - } - - return false; - } - private void RetireOrResurectStates() { foreach (var (name, timestamp) in _retirementTracker) @@ -935,55 +855,23 @@ public async ValueTask WriteStateAsync(CancellationToken cancellationToken) } } - public async ValueTask RevertPendingChangesAsync(CancellationToken cancellationToken) - { - cancellationToken.ThrowIfCancellationRequested(); - Task pendingRecovery; - bool didEnqueue; - lock (_lock) - { - _shutdownCancellation.Token.ThrowIfCancellationRequested(); - if (_state is ManagerState.Unknown) - { - throw new InvalidOperationException("The journaled state manager has not been initialized."); - } - - pendingRecovery = EnqueueOrGetPendingWorkItem(out didEnqueue); - _state = ManagerState.Recovering; - } - - if (didEnqueue) - { - _workSignal.Signal(); - } - - await pendingRecovery.WaitAsync(cancellationToken); - } - private void ThrowIfStateOperationsUnavailable() { _shutdownCancellation.Token.ThrowIfCancellationRequested(); - switch (_state) + ThrowIfFenced(); + if (_state is not ManagerState.Ready) { - case ManagerState.Unknown: - throw new InvalidOperationException("The journaled state manager has not been initialized."); - case ManagerState.Ready: - return; - case ManagerState.Recovering: - throw new InvalidOperationException("Journaled state operations are unavailable while recovery is in progress."); - case ManagerState.Fenced: - throw new InvalidOperationException("Journaled state operations are fenced because recovery failed. Call RevertPendingChangesAsync to retry recovery."); - default: - throw new UnreachableException(); + throw new InvalidOperationException("The journaled state manager has not been initialized."); } } - private void ThrowIfInitializationFenced() + private void ThrowIfFenced() { if (_state is ManagerState.Fenced) { throw new InvalidOperationException( - "The journaled state manager is fenced because recovery failed. Call RevertPendingChangesAsync to retry recovery."); + "Journaled state operations are fenced because an operation failed. Recover using a new grain activation or dispose this manager and create a new instance.", + _failure); } } @@ -1058,7 +946,7 @@ private Task EnqueueOrGetPendingWorkItem(out bool didEnqueue) } if (workItem is AppendJournalWorkItem or WriteSnapshotWorkItem - && workItem.RecoveryGeneration != _recoveryGeneration) + && workItem.DeletionGeneration != _deletionGeneration) { continue; } @@ -1071,7 +959,7 @@ private Task EnqueueOrGetPendingWorkItem(out bool didEnqueue) var newWorkItem = new TWorkItem(); newWorkItem.EnqueuedTimestamp = _shared.TimeProvider.GetTimestamp(); - newWorkItem.RecoveryGeneration = _recoveryGeneration; + newWorkItem.DeletionGeneration = _deletionGeneration; newWorkItem.RecordTraceContext(); _workQueue.Enqueue(newWorkItem); didEnqueue = true; @@ -1165,7 +1053,7 @@ private abstract class WorkItem : TaskCompletionSource public long EnqueuedTimestamp { get; set; } - public long RecoveryGeneration { get; set; } + public long DeletionGeneration { get; set; } public ActivityContext TraceParent { get; private set; } @@ -1254,8 +1142,6 @@ private sealed class WriteSnapshotWorkItem : WorkItem; private sealed class DeleteStateWorkItem : WorkItem; - private sealed class RevertPendingChangesWorkItem : WorkItem; - private sealed class RegisterStateWorkItem(string name) : WorkItem(name) { public string Name => (string)Task.AsyncState!; @@ -1265,7 +1151,6 @@ private enum ManagerState : byte { Unknown, Ready, - Recovering, Fenced } diff --git a/src/Orleans.Journaling/README.md b/src/Orleans.Journaling/README.md index 497d42c1981..d70daff3ab5 100644 --- a/src/Orleans.Journaling/README.md +++ b/src/Orleans.Journaling/README.md @@ -94,6 +94,22 @@ All durable state types use the configured JSON codec automatically. Configure ` 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. +## Staging and failure boundaries + +The pending journal is shared by every caller using a state manager. Prepare fallible work and external +acknowledgements in operation-local data. Once an outcome is safe to commit, apply its mutations to the +durable states and await `WriteStateAsync`. Applications are responsible for sequencing that transition +with other interleaved operations and for making uncertain-outcome retries idempotent. + +A failed journal operation permanently fences the manager, faults queued operations, and requests +deactivation of the associated grain. In-flight calls retain their existing in-memory state while subsequent +state-manager operations fail explicitly. A new activation recovers the actual durable outcome. +For a manager created through `IJournaledStateManagerFactory`, dispose the failed instance and create +another manager for the same `JournalId`, registering new state instances before initialization. + +Cancelling a caller's wait leaves an already queued write running. Observe durability through write +acknowledgement or a fresh activation before deciding whether to retry an application command. + ## Storage format The JSON journaling format stores journal entries as true JSON Lines: UTF-8 text, no byte order mark, and one JSON array per journal entry line. Each line is terminated by `\n`. Recovery accepts both LF and CRLF line endings. Storage providers which use format metadata should store `JsonJournalExtensions.JournalFormatKey` as the format key and may use `application/jsonl` as the MIME type. diff --git a/src/api/Orleans.Journaling/Orleans.Journaling.cs b/src/api/Orleans.Journaling/Orleans.Journaling.cs index 237a4df87e2..c41fca5e2f4 100644 --- a/src/api/Orleans.Journaling/Orleans.Journaling.cs +++ b/src/api/Orleans.Journaling/Orleans.Journaling.cs @@ -226,7 +226,6 @@ public partial interface IJournaledStateManager : System.IAsyncDisposable System.Threading.Tasks.ValueTask DeleteStateAsync(System.Threading.CancellationToken cancellationToken); System.Threading.Tasks.ValueTask InitializeAsync(System.Threading.CancellationToken cancellationToken); void RegisterState(string name, IJournaledState state); - System.Threading.Tasks.ValueTask RevertPendingChangesAsync(System.Threading.CancellationToken cancellationToken); System.Threading.Tasks.ValueTask System.IAsyncDisposable.DisposeAsync(); bool TryGetState(string name, out IJournaledState? state); System.Threading.Tasks.ValueTask WriteStateAsync(System.Threading.CancellationToken cancellationToken); diff --git a/test/Benchmarks/Journaling/DurableListJournalBenchmarks.cs b/test/Benchmarks/Journaling/DurableListJournalBenchmarks.cs index 5a7a602c620..df650ca42c4 100644 --- a/test/Benchmarks/Journaling/DurableListJournalBenchmarks.cs +++ b/test/Benchmarks/Journaling/DurableListJournalBenchmarks.cs @@ -138,7 +138,6 @@ public bool TryGetState(string name, [NotNullWhen(true)] out IJournaledState sta public ValueTask WriteStateAsync(CancellationToken cancellationToken) => default; - public ValueTask RevertPendingChangesAsync(CancellationToken cancellationToken) => throw new NotSupportedException(); public ValueTask DeleteStateAsync(CancellationToken cancellationToken) => default; } diff --git a/test/Orleans.DurableJobs.Tests/DurableJobs/JournaledJobShardDiscoveryTests.cs b/test/Orleans.DurableJobs.Tests/DurableJobs/JournaledJobShardDiscoveryTests.cs index 074495abc97..72125dec0d5 100644 --- a/test/Orleans.DurableJobs.Tests/DurableJobs/JournaledJobShardDiscoveryTests.cs +++ b/test/Orleans.DurableJobs.Tests/DurableJobs/JournaledJobShardDiscoveryTests.cs @@ -734,7 +734,6 @@ private sealed class TrackingJournaledStateManager(IJournaledStateManager inner) public void RegisterState(string name, IJournaledState state) => inner.RegisterState(name, state); public bool TryGetState(string name, [NotNullWhen(true)] out IJournaledState? state) => inner.TryGetState(name, out state); public ValueTask WriteStateAsync(CancellationToken cancellationToken) => inner.WriteStateAsync(cancellationToken); - public ValueTask RevertPendingChangesAsync(CancellationToken cancellationToken) => inner.RevertPendingChangesAsync(cancellationToken); public ValueTask DeleteStateAsync(CancellationToken cancellationToken) => inner.DeleteStateAsync(cancellationToken); public async ValueTask DisposeAsync() diff --git a/test/Orleans.Journaling.Tests/DurableCollectionDirectWriteTests.cs b/test/Orleans.Journaling.Tests/DurableCollectionDirectWriteTests.cs index bdd53841f81..a28ff003da8 100644 --- a/test/Orleans.Journaling.Tests/DurableCollectionDirectWriteTests.cs +++ b/test/Orleans.Journaling.Tests/DurableCollectionDirectWriteTests.cs @@ -205,7 +205,6 @@ public bool TryGetState(string name, [NotNullWhen(true)] out IJournaledState? st public ValueTask WriteStateAsync(CancellationToken cancellationToken) => default; - public ValueTask RevertPendingChangesAsync(CancellationToken cancellationToken) => throw new NotSupportedException(); public ValueTask DeleteStateAsync(CancellationToken cancellationToken) => default; } diff --git a/test/Orleans.Journaling.Tests/DurableListDirectWriteTests.cs b/test/Orleans.Journaling.Tests/DurableListDirectWriteTests.cs index 5dff6e54cd0..d213f7229c1 100644 --- a/test/Orleans.Journaling.Tests/DurableListDirectWriteTests.cs +++ b/test/Orleans.Journaling.Tests/DurableListDirectWriteTests.cs @@ -104,7 +104,6 @@ public bool TryGetState(string name, [NotNullWhen(true)] out IJournaledState? st public ValueTask WriteStateAsync(CancellationToken cancellationToken) => default; - public ValueTask RevertPendingChangesAsync(CancellationToken cancellationToken) => throw new NotSupportedException(); public ValueTask DeleteStateAsync(CancellationToken cancellationToken) => default; } diff --git a/test/Orleans.Journaling.Tests/StateManagerTests.cs b/test/Orleans.Journaling.Tests/StateManagerTests.cs index 3e70fe19cc7..99f37710436 100644 --- a/test/Orleans.Journaling.Tests/StateManagerTests.cs +++ b/test/Orleans.Journaling.Tests/StateManagerTests.cs @@ -2,12 +2,15 @@ using System.Collections.Concurrent; using System.Diagnostics.Metrics; using System.Runtime.ExceptionServices; +using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Options; using Microsoft.Extensions.Time.Testing; using Orleans.Serialization.Buffers; using Orleans.Serialization.Buffers.Adaptors; using Orleans.Serialization.Session; using Orleans.Storage; +using NSubstitute; using Xunit; namespace Orleans.Journaling.Tests; @@ -118,8 +121,8 @@ public async Task StateManager_WriteOperations_RequireSuccessfulInitialization() var deleteException = await Assert.ThrowsAsync( () => sut.Manager.DeleteStateAsync(TestContext.Current.CancellationToken).AsTask()); - Assert.Contains("not been initialized", writeException.Message, StringComparison.Ordinal); - Assert.Contains("not been initialized", deleteException.Message, StringComparison.Ordinal); + Assert.Contains("fenced", writeException.Message, StringComparison.Ordinal); + Assert.Contains("fenced", deleteException.Message, StringComparison.Ordinal); } [Fact] @@ -133,8 +136,6 @@ await Assert.ThrowsAnyAsync( () => sut.Manager.WriteStateAsync(TestContext.Current.CancellationToken).AsTask()); await Assert.ThrowsAnyAsync( () => sut.Manager.DeleteStateAsync(TestContext.Current.CancellationToken).AsTask()); - await Assert.ThrowsAnyAsync( - () => sut.Manager.RevertPendingChangesAsync(TestContext.Current.CancellationToken).AsTask()); } [Fact] @@ -446,7 +447,7 @@ public async Task StateManager_SnapshotFlush_ResetsPendingAppendData() } [Fact] - public async Task StateManager_SnapshotFailure_PreservesMixedPendingMutationsForOneCommitRetry() + public async Task StateManager_SnapshotFailure_FencesMixedPendingMutations() { var storage = new CapturingStorage(); var sut = CreateTestSystem(storage: storage); @@ -487,17 +488,12 @@ public async Task StateManager_SnapshotFailure_PreservesMixedPendingMutationsFor Assert.Equal(2, dictionary["pending"]); Assert.Equal(2, value.Value); - await sut.Manager.WriteStateAsync(TestContext.Current.CancellationToken); - - Assert.Single(storage.Appends); - Assert.Single(storage.Replaces); - Assert.Equal(["replace-failed", "replace"], storage.OperationLog); - var successfulRetryBytes = storage.Replaces[0]; - Assert.Equal(failedAttemptBytes, successfulRetryBytes); - Assert.Equal(2, storage.ReplaceAttemptCount); - Assert.Equal(2, notifications.WriteCompletedCount); - Assert.Equal(notificationCountBeforeFailure + 1, notifications.WriteCompletedCount); - Assert.Equal(0, sut.Manager.PendingWriteByteCount); + var fenced = await Assert.ThrowsAsync( + () => sut.Manager.WriteStateAsync(TestContext.Current.CancellationToken).AsTask()); + Assert.Same(expected, fenced.InnerException); + Assert.Empty(storage.Replaces); + Assert.Equal(1, storage.ReplaceAttemptCount); + Assert.Equal(notificationCountBeforeFailure, notifications.WriteCompletedCount); var recovered = CreateTestSystem(storage: storage); var recoveredDictionary = new DurableDictionary("dict", recovered.Manager, CreateDictionaryCodec()); @@ -505,8 +501,8 @@ public async Task StateManager_SnapshotFailure_PreservesMixedPendingMutationsFor await recovered.Lifecycle.OnStart(TestContext.Current.CancellationToken); Assert.Equal(1, recoveredDictionary["persisted"]); - Assert.Equal(2, recoveredDictionary["pending"]); - Assert.Equal(2, recoveredValue.Value); + Assert.False(recoveredDictionary.ContainsKey("pending")); + Assert.Equal(1, recoveredValue.Value); } [Fact] @@ -554,318 +550,180 @@ public async Task StateManager_DirectWriteFailure_DoesNotPersistPartialEntry() Assert.Equal(1, recoveredDictionary[1]); } - [Fact] - public async Task StateManager_WriteStateAsync_RecoversAfterInconsistentStateException() - { - var storage = new CapturingStorage(); - var sut = CreateTestSystem(storage: storage); - var dictionary = new DurableDictionary("dict", sut.Manager, CreateDictionaryCodec()); - - await sut.Lifecycle.OnStart(TestContext.Current.CancellationToken); - dictionary.Add("first", 1); - await sut.Manager.WriteStateAsync(CancellationToken.None); - - var expected = new InconsistentStateException("Expected storage write conflict."); - storage.NextAppendException = expected; - storage.BlockNextRead = true; - dictionary.Add("second", 2); - - var failedWrite = sut.Manager.WriteStateAsync(CancellationToken.None).AsTask(); - await storage.BlockedReadStarted.Task.WaitAsync(TimeSpan.FromSeconds(10), TestContext.Current.CancellationToken); - Assert.False(failedWrite.IsCompleted); - - storage.AllowBlockedRead.SetResult(); - var exception = await Assert.ThrowsAsync( - () => failedWrite.WaitAsync(TimeSpan.FromSeconds(10), TestContext.Current.CancellationToken)); - Assert.Same(expected, exception); - - await sut.Manager.WriteStateAsync(TestContext.Current.CancellationToken).AsTask() - .WaitAsync(TimeSpan.FromSeconds(10), TestContext.Current.CancellationToken); - - Assert.True(dictionary.ContainsKey("first")); - Assert.False(dictionary.ContainsKey("second")); - - var recovered = CreateTestSystem(storage: storage); - var recoveredDictionary = new DurableDictionary("dict", recovered.Manager, CreateDictionaryCodec()); - await recovered.Lifecycle.OnStart(TestContext.Current.CancellationToken); - - Assert.Equal(1, recoveredDictionary["first"]); - Assert.False(recoveredDictionary.ContainsKey("second")); - } - - [Fact] - public async Task StateManager_WriteStateAsync_PreservesPendingStateAfterTransientStorageWriteFailure() + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task StateManager_WriteFailurePreservesInterleavedStateAndFencesAllCallers(bool conflict) { var storage = new CapturingStorage(); var sut = CreateTestSystem(storage: storage); var dictionary = new DurableDictionary("dict", sut.Manager, CreateDictionaryCodec()); - + var value = new DurableValue("value", sut.Manager, CreateValueCodec()); await sut.Lifecycle.OnStart(TestContext.Current.CancellationToken); - dictionary.Add("first", 1); - await sut.Manager.WriteStateAsync(CancellationToken.None); + dictionary.Add("persisted", 1); + value.Value = 1; + await sut.Manager.WriteStateAsync(TestContext.Current.CancellationToken); storage.ResetReadConsumeCount(); - var expected = new IOException("Expected storage write failure."); + Exception expected = conflict + ? new InconsistentStateException("Expected storage conflict.") + : new IOException("Expected storage failure."); storage.NextAppendException = expected; - dictionary.Add("second", 2); + storage.BlockNextAppend = true; + value.Value = 2; + var firstWrite = sut.Manager.WriteStateAsync(TestContext.Current.CancellationToken).AsTask(); + await storage.BlockedAppendStarted.Task.WaitAsync(TimeSpan.FromSeconds(10), TestContext.Current.CancellationToken); - var exception = await Assert.ThrowsAsync( - () => sut.Manager.WriteStateAsync(CancellationToken.None).AsTask()); - Assert.Same(expected, exception); + dictionary.Add("interleaved", 3); + var queuedWrite = sut.Manager.WriteStateAsync(TestContext.Current.CancellationToken).AsTask(); + var queuedDelete = sut.Manager.DeleteStateAsync(TestContext.Current.CancellationToken).AsTask(); + var queuedInitialize = sut.Manager.InitializeAsync(TestContext.Current.CancellationToken).AsTask(); + storage.ReleaseAppend.SetResult(); - await sut.Manager.WriteStateAsync(TestContext.Current.CancellationToken).AsTask() - .WaitAsync(TimeSpan.FromSeconds(10), TestContext.Current.CancellationToken); + foreach (var task in new[] { firstWrite, queuedWrite, queuedDelete, queuedInitialize }) + { + var exception = await Record.ExceptionAsync( + () => task.WaitAsync(TimeSpan.FromSeconds(10), TestContext.Current.CancellationToken)); + Assert.Same(expected, exception); + } + Assert.Equal(2, value.Value); + Assert.Equal(3, dictionary["interleaved"]); Assert.Equal(0, storage.ReadConsumeCount); - Assert.Equal(2, dictionary["second"]); + Assert.Single(storage.Appends); + Assert.Equal(0, storage.DeleteCount); + var lateWrite = await Assert.ThrowsAsync( + () => sut.Manager.WriteStateAsync(TestContext.Current.CancellationToken).AsTask()); + Assert.Same(expected, lateWrite.InnerException); + await Assert.ThrowsAsync( + () => sut.Manager.DeleteStateAsync(TestContext.Current.CancellationToken).AsTask()); + await Assert.ThrowsAsync( + () => sut.Manager.InitializeAsync(TestContext.Current.CancellationToken).AsTask()); + Assert.Throws(() => sut.Manager.RegisterState("late", new AlwaysWritingState())); + + await sut.Manager.DisposeAsync(); var recovered = CreateTestSystem(storage: storage); + var recoveredValue = new DurableValue("value", recovered.Manager, CreateValueCodec()); var recoveredDictionary = new DurableDictionary("dict", recovered.Manager, CreateDictionaryCodec()); await recovered.Lifecycle.OnStart(TestContext.Current.CancellationToken); - - Assert.Equal(1, recoveredDictionary["first"]); - Assert.Equal(2, recoveredDictionary["second"]); - } - - [Fact] - public async Task StateManager_WriteStateAsync_FencesWritesAfterRepeatedRecoveryFailures() - { - var storage = new CapturingStorage(); - var sut = CreateTestSystem(storage: storage); - var dictionary = new DurableDictionary("dict", sut.Manager, CreateDictionaryCodec()); - - await sut.Lifecycle.OnStart(TestContext.Current.CancellationToken); - dictionary.Add("first", 1); - await sut.Manager.WriteStateAsync(CancellationToken.None); - - var conflict = new InconsistentStateException("Expected storage write conflict."); - var firstRecoveryFailure = new IOException("Expected first recovery failure."); - storage.NextAppendException = conflict; - storage.NextReadException = firstRecoveryFailure; - dictionary.Add("second", 2); - - var conflictException = await Assert.ThrowsAsync( - () => sut.Manager.WriteStateAsync(TestContext.Current.CancellationToken).AsTask() - .WaitAsync(TimeSpan.FromSeconds(10), TestContext.Current.CancellationToken)); - Assert.Same(conflict, conflictException); - - var secondRecoveryFailure = new IOException("Expected second recovery failure."); - storage.NextReadException = secondRecoveryFailure; - var recoveryException = await Assert.ThrowsAsync( - () => sut.Manager.RevertPendingChangesAsync(TestContext.Current.CancellationToken).AsTask() - .WaitAsync(TimeSpan.FromSeconds(10), TestContext.Current.CancellationToken)); - Assert.Same(secondRecoveryFailure, recoveryException); - - var writeException = await Assert.ThrowsAsync( - () => sut.Manager.WriteStateAsync(TestContext.Current.CancellationToken).AsTask() - .WaitAsync(TimeSpan.FromSeconds(10), TestContext.Current.CancellationToken)); - Assert.Contains("fenced", writeException.Message, StringComparison.OrdinalIgnoreCase); - - var initializeException = await Assert.ThrowsAsync( - () => sut.Manager.InitializeAsync(TestContext.Current.CancellationToken).AsTask() - .WaitAsync(TimeSpan.FromSeconds(10), TestContext.Current.CancellationToken)); - Assert.Contains("fenced", initializeException.Message, StringComparison.OrdinalIgnoreCase); - - await sut.Manager.RevertPendingChangesAsync(TestContext.Current.CancellationToken).AsTask() - .WaitAsync(TimeSpan.FromSeconds(10), TestContext.Current.CancellationToken); - Assert.True(dictionary.ContainsKey("first")); - Assert.False(dictionary.ContainsKey("second")); + Assert.Equal(1, recoveredValue.Value); + Assert.Single(recoveredDictionary); + Assert.Equal(1, recoveredDictionary["persisted"]); } - [Fact] - public async Task StateManager_RevertPendingChanges_RestoresLastDurableState() + [Theory] + [InlineData("initialize")] + [InlineData("append")] + [InlineData("replace")] + [InlineData("delete")] + public async Task StateManager_FailureDeactivatesOwningGrain(string operation) { + var expected = new IOException("Expected journal operation failure."); var storage = new CapturingStorage(); - var sut = CreateTestSystem(storage: storage); - var dictionary = new DurableDictionary("dict", sut.Manager, CreateDictionaryCodec()); - var value = new DurableValue("value", sut.Manager, CreateValueCodec()); - - await sut.Lifecycle.OnStart(TestContext.Current.CancellationToken); - dictionary.Add("persisted", 1); - value.Value = 1; - await sut.Manager.WriteStateAsync(TestContext.Current.CancellationToken); - - dictionary.Add("pending", 2); - value.Value = 2; - Assert.True(sut.Manager.PendingWriteByteCount > 0); - - await sut.Manager.RevertPendingChangesAsync(TestContext.Current.CancellationToken); - - Assert.Equal(0, sut.Manager.PendingWriteByteCount); - Assert.Equal(1, dictionary["persisted"]); - Assert.False(dictionary.ContainsKey("pending")); - Assert.Equal(1, value.Value); - } + var storageProvider = Substitute.For(); + var context = Substitute.For(); + var grainId = GrainId.Create("test-grain", "failing-journal"); + context.GrainId.Returns(grainId); + storageProvider.CreateStorage(JournalId.FromGrainId(grainId)).Returns(storage); + var shared = new JournaledStateManagerShared( + ServiceProvider.GetRequiredService>(), + Options.Create(ManagerOptions), TimeProvider.System, ServiceProvider); + await using var manager = new JournaledStateManager(shared, storageProvider, context); + var value = new DurableValue("value", manager, CreateValueCodec()); + if (operation == "initialize") + { + storage.NextReadException = expected; + } + else + { + await manager.InitializeAsync(TestContext.Current.CancellationToken); + value.Value = 42; + storage.NextAppendException = expected; + storage.NextReplaceException = expected; + storage.NextDeleteException = expected; + storage.IsCompactionRequested = operation == "replace"; + } - [Fact] - public async Task StateManager_RecoveryRejectsWritesQueuedBeforeMutationsWereDiscarded() - { - var storage = new BlockingAppendStorage + var failedOperation = operation switch { - FirstAppendException = new InconsistentStateException("Expected write conflict.") + "initialize" => manager.InitializeAsync(TestContext.Current.CancellationToken), + "delete" => manager.DeleteStateAsync(TestContext.Current.CancellationToken), + _ => manager.WriteStateAsync(TestContext.Current.CancellationToken) }; - var sut = CreateTestSystem(storage: storage); - var value = new DurableValue("value", sut.Manager, CreateValueCodec()); - await sut.Lifecycle.OnStart(TestContext.Current.CancellationToken); - value.Value = 1; - - var conflictingWrite = sut.Manager.WriteStateAsync(TestContext.Current.CancellationToken).AsTask(); - await storage.FirstAppendStarted.Task.WaitAsync( - TimeSpan.FromSeconds(10), - TestContext.Current.CancellationToken); - value.Value = 2; - var queuedWrite = sut.Manager.WriteStateAsync(TestContext.Current.CancellationToken).AsTask(); - - storage.AllowFirstAppend.TrySetResult(); - - await Assert.ThrowsAsync( - () => conflictingWrite.WaitAsync( - TimeSpan.FromSeconds(10), - TestContext.Current.CancellationToken)); - var exception = await Assert.ThrowsAsync( - () => queuedWrite.WaitAsync( - TimeSpan.FromSeconds(10), - TestContext.Current.CancellationToken)); - - Assert.Contains("queued before recovery discarded its mutations", exception.Message, StringComparison.Ordinal); - Assert.Empty(storage.Appends); - - value.Value = 3; - await sut.Manager.WriteStateAsync(TestContext.Current.CancellationToken); - - Assert.Single(storage.Appends); + var exception = await Assert.ThrowsAsync(() => failedOperation.AsTask()); + Assert.Same(expected, exception); + context.Received(1).Deactivate( + Arg.Is(reason => reason.ReasonCode == DeactivationReasonCode.ApplicationError && ReferenceEquals(reason.Exception, expected)), + Arg.Any()); + await Assert.ThrowsAsync( + () => manager.InitializeAsync(TestContext.Current.CancellationToken).AsTask()); } - [Fact] - public async Task StateManager_InconsistentStateDuringExplicitRevertRemainsFenced() + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task StateManager_FactoryRecreationRecoversActualCommitOutcome(bool committed) { var storage = new CapturingStorage(); - var sut = CreateTestSystem(storage: storage); - var value = new DurableValue("value", sut.Manager, CreateValueCodec()); - - await sut.Lifecycle.OnStart(TestContext.Current.CancellationToken); + var storageProvider = Substitute.For(); + var journalId = new JournalId("standalone/fail-closed"); + storageProvider.CreateStorage(journalId).Returns(storage); + var shared = new JournaledStateManagerShared( + ServiceProvider.GetRequiredService>(), + Options.Create(ManagerOptions), TimeProvider.System, ServiceProvider); + var factory = new JournaledStateManagerFactory(shared, storageProvider); + var manager = factory.Create(journalId); + var value = new DurableValue("value", manager, CreateValueCodec()); + await manager.InitializeAsync(TestContext.Current.CancellationToken); value.Value = 1; - await sut.Manager.WriteStateAsync(TestContext.Current.CancellationToken); + await manager.WriteStateAsync(TestContext.Current.CancellationToken); value.Value = 2; - storage.NextReadException = new InconsistentStateException("Expected explicit recovery conflict."); - - await Assert.ThrowsAsync( - () => sut.Manager.RevertPendingChangesAsync(TestContext.Current.CancellationToken).AsTask()); - var writeException = await Assert.ThrowsAsync( - () => sut.Manager.WriteStateAsync(TestContext.Current.CancellationToken).AsTask()); - Assert.Contains("fenced", writeException.Message, StringComparison.OrdinalIgnoreCase); - - await sut.Manager.RevertPendingChangesAsync(TestContext.Current.CancellationToken); - Assert.Equal(1, value.Value); - } - - [Fact] - public async Task StateManager_FailedExplicitRevertFaultsQueuedInitialization() - { - var expected = new IOException("Expected explicit recovery failure."); - var storage = new BlockingRecoveryStorage + var expected = new IOException("Commit acknowledgement failed."); + if (committed) { - RecoveryReadException = expected - }; - var sut = CreateTestSystem(storage: storage); - await sut.Lifecycle.OnStart(TestContext.Current.CancellationToken); - - var revert = sut.Manager.RevertPendingChangesAsync(TestContext.Current.CancellationToken).AsTask(); - await storage.RecoveryReadStarted.Task.WaitAsync( - TimeSpan.FromSeconds(10), - TestContext.Current.CancellationToken); - var initialize = sut.Manager.InitializeAsync(TestContext.Current.CancellationToken).AsTask(); + storage.NextPostAppendException = expected; + } + else + { + storage.NextAppendException = expected; + } - storage.AllowRecoveryRead.SetResult(); + var exception = await Assert.ThrowsAsync( + () => manager.WriteStateAsync(TestContext.Current.CancellationToken).AsTask()); + Assert.Same(expected, exception); + Assert.Equal(2, value.Value); + await manager.DisposeAsync(); - var revertException = await Assert.ThrowsAsync(() => revert); - Assert.Same(expected, revertException); - var initializeException = await Assert.ThrowsAsync(() => initialize); - Assert.Same(expected, initializeException); - var writeException = await Assert.ThrowsAsync( - () => sut.Manager.WriteStateAsync(TestContext.Current.CancellationToken).AsTask()); - Assert.Contains("fenced", writeException.Message, StringComparison.OrdinalIgnoreCase); + await using var recovered = factory.Create(journalId); + var recoveredValue = new DurableValue("value", recovered, CreateValueCodec()); + await recovered.InitializeAsync(TestContext.Current.CancellationToken); + Assert.Equal(committed ? 2 : 1, recoveredValue.Value); + recoveredValue.Value = 3; + await recovered.WriteStateAsync(TestContext.Current.CancellationToken); + storageProvider.Received(2).CreateStorage(journalId); } [Fact] - public async Task StateManager_RevertAgainstEmptyJournalResetsUncommittedState() + public async Task StateManager_CancelledWriteWaitPreservesCommit() { - var storage = new CapturingStorage(); + var storage = new CapturingStorage { BlockNextAppend = true }; var sut = CreateTestSystem(storage: storage); var value = new DurableValue("value", sut.Manager, CreateValueCodec()); await sut.Lifecycle.OnStart(TestContext.Current.CancellationToken); value.Value = 42; - - await sut.Manager.RevertPendingChangesAsync(TestContext.Current.CancellationToken); - - Assert.Equal(0, value.Value); + using var cancellation = new CancellationTokenSource(); + var write = sut.Manager.WriteStateAsync(cancellation.Token).AsTask(); + await storage.AppendEntered.Task.WaitAsync(TimeSpan.FromSeconds(10), TestContext.Current.CancellationToken); + cancellation.Cancel(); + await Assert.ThrowsAnyAsync(() => write); + storage.ReleaseAppend.SetResult(); await sut.Manager.WriteStateAsync(TestContext.Current.CancellationToken); + var recovered = CreateTestSystem(storage: storage); var recoveredValue = new DurableValue("value", recovered.Manager, CreateValueCodec()); await recovered.Lifecycle.OnStart(TestContext.Current.CancellationToken); - Assert.Equal(0, recoveredValue.Value); - } - - [Fact] - public async Task StateManager_FencesStateOperationsUntilFailedRevertIsRetriedSuccessfully() - { - var storage = new CapturingStorage(); - var sut = CreateTestSystem(storage: storage); - var value = new DurableValue("value", sut.Manager, CreateValueCodec()); - - await sut.Lifecycle.OnStart(TestContext.Current.CancellationToken); - value.Value = 1; - await sut.Manager.WriteStateAsync(TestContext.Current.CancellationToken); - value.Value = 2; - - var expected = new IOException("Expected recovery failure."); - storage.NextReadException = expected; - var exception = await Assert.ThrowsAsync( - () => sut.Manager.RevertPendingChangesAsync(TestContext.Current.CancellationToken).AsTask()); - Assert.Same(expected, exception); - - var writeException = await Assert.ThrowsAsync( - () => sut.Manager.WriteStateAsync(TestContext.Current.CancellationToken).AsTask()); - Assert.Contains("state operations are fenced", writeException.Message, StringComparison.Ordinal); - Assert.Contains("fenced", writeException.Message, StringComparison.OrdinalIgnoreCase); - Assert.Contains("Call RevertPendingChangesAsync", writeException.Message, StringComparison.Ordinal); - var deleteException = await Assert.ThrowsAsync( - () => sut.Manager.DeleteStateAsync(TestContext.Current.CancellationToken).AsTask()); - Assert.Contains("state operations are fenced", deleteException.Message, StringComparison.Ordinal); - Assert.Contains("Call RevertPendingChangesAsync", deleteException.Message, StringComparison.Ordinal); - - await sut.Manager.RevertPendingChangesAsync(TestContext.Current.CancellationToken); - Assert.Equal(1, value.Value); - - value.Value = 3; - await sut.Manager.WriteStateAsync(TestContext.Current.CancellationToken); - Assert.Equal(3, value.Value); - } - - [Fact] - public async Task StateManager_StateOperations_ReportRecoveryInProgress() - { - var storage = new BlockingRecoveryStorage(); - var sut = CreateTestSystem(storage: storage); - await sut.Lifecycle.OnStart(TestContext.Current.CancellationToken); - - var recovery = sut.Manager.RevertPendingChangesAsync(TestContext.Current.CancellationToken).AsTask(); - await storage.RecoveryReadStarted.Task.WaitAsync( - TimeSpan.FromSeconds(10), - TestContext.Current.CancellationToken); - - var writeException = await Assert.ThrowsAsync( - () => sut.Manager.WriteStateAsync(TestContext.Current.CancellationToken).AsTask()); - var deleteException = await Assert.ThrowsAsync( - () => sut.Manager.DeleteStateAsync(TestContext.Current.CancellationToken).AsTask()); - - Assert.Contains("state operations are unavailable while recovery is in progress", writeException.Message, StringComparison.Ordinal); - Assert.Contains("state operations are unavailable while recovery is in progress", deleteException.Message, StringComparison.Ordinal); - Assert.DoesNotContain("RevertPendingChangesAsync", writeException.Message, StringComparison.Ordinal); - Assert.DoesNotContain("RevertPendingChangesAsync", deleteException.Message, StringComparison.Ordinal); - - storage.AllowRecoveryRead.SetResult(); - await recovery.WaitAsync(TimeSpan.FromSeconds(10), TestContext.Current.CancellationToken); + Assert.Equal(42, recoveredValue.Value); } [Fact] @@ -1205,11 +1063,9 @@ public async Task StateManager_Recovery_RejectsMalformedTrailingData() } [Fact] - public async Task StateManager_RecoveryRetry_ReplaysFixedStorage() + public async Task StateManager_FreshRecovery_ReplaysFixedStorage() { var validBytes = CreatePersistedValueBytes("value", 42); - // The manager retries recovery on its background work loop, so the repaired - // storage state must be available on the retry without depending on test-thread timing. var storage = new MutableReadStorage([.. validBytes, 1, 2, 3], validBytes); var sut = CreateTestSystem(storage: storage); var value = new DurableValue("value", sut.Manager, CreateValueCodec()); @@ -1218,6 +1074,11 @@ await Assert.ThrowsAsync( () => sut.Lifecycle.OnStart(TestContext.Current.CancellationToken) .WaitAsync(TimeSpan.FromSeconds(10), TestContext.Current.CancellationToken)); + await Assert.ThrowsAsync( + () => sut.Manager.InitializeAsync(TestContext.Current.CancellationToken).AsTask()); + await sut.Manager.DisposeAsync(); + sut = CreateTestSystem(storage: storage); + value = new DurableValue("value", sut.Manager, CreateValueCodec()); await sut.Manager.InitializeAsync(TestContext.Current.CancellationToken).AsTask() .WaitAsync(TimeSpan.FromSeconds(10), TestContext.Current.CancellationToken); await sut.Manager.WriteStateAsync(TestContext.Current.CancellationToken).AsTask() @@ -1228,7 +1089,7 @@ await sut.Manager.WriteStateAsync(TestContext.Current.CancellationToken).AsTask( } [Fact] - public async Task StateManager_RecoveryRetry_PreservesUnknownStreamOnce() + public async Task StateManager_FreshRecovery_PreservesUnknownStreamOnce() { var validBytes = CreateUnknownStreamBytes(new JournalStreamId(99), [1, 2, 3]); var storage = new MutableReadStorage([.. validBytes, 1, 2, 3], validBytes) { IsCompactionRequested = true }; @@ -1238,6 +1099,10 @@ await Assert.ThrowsAsync( () => sut.Lifecycle.OnStart(TestContext.Current.CancellationToken) .WaitAsync(TimeSpan.FromSeconds(10), TestContext.Current.CancellationToken)); + await Assert.ThrowsAsync( + () => sut.Manager.InitializeAsync(TestContext.Current.CancellationToken).AsTask()); + await sut.Manager.DisposeAsync(); + sut = CreateTestSystem(storage: storage); await sut.Manager.InitializeAsync(TestContext.Current.CancellationToken).AsTask() .WaitAsync(TimeSpan.FromSeconds(10), TestContext.Current.CancellationToken); await sut.Manager.WriteStateAsync(TestContext.Current.CancellationToken).AsTask() @@ -1250,7 +1115,7 @@ await sut.Manager.WriteStateAsync(TestContext.Current.CancellationToken).AsTask( } [Fact] - public async Task StateManager_RecoveryRetry_RemovesStaleRetiredPlaceholder() + public async Task StateManager_FreshRecovery_RemovesStaleRetiredPlaceholder() { var storage = new MutableReadStorage([.. CreateNamedUnknownStreamBytes("stale", new JournalStreamId(8), [1, 2, 3]), 1, 2, 3], []); var sut = CreateTestSystem(storage: storage); @@ -1259,6 +1124,10 @@ await Assert.ThrowsAsync( () => sut.Lifecycle.OnStart(TestContext.Current.CancellationToken) .WaitAsync(TimeSpan.FromSeconds(10), TestContext.Current.CancellationToken)); + await Assert.ThrowsAsync( + () => sut.Manager.InitializeAsync(TestContext.Current.CancellationToken).AsTask()); + await sut.Manager.DisposeAsync(); + sut = CreateTestSystem(storage: storage); await sut.Manager.InitializeAsync(TestContext.Current.CancellationToken).AsTask() .WaitAsync(TimeSpan.FromSeconds(10), TestContext.Current.CancellationToken); await sut.Manager.WriteStateAsync(TestContext.Current.CancellationToken).AsTask() @@ -1657,7 +1526,7 @@ public async Task WorkLoop_AppendDeleteReplacePersistsPostDeleteStateAndFreshMan } [Fact] - public async Task RecoverAsync_UnsupportedLegacyRecordCanRetryAfterRepairWithoutPartialApplication() + public async Task RecoverAsync_UnsupportedLegacyRecordRecoversInNewManagerAfterRepair() { var unsupportedBytes = CreateUnsupportedLegacyCommandVersionRecord(streamId: 128, commandVersion: 1); var validBytes = CreatePersistedStringValueBytes("value", "recovered"); @@ -1669,8 +1538,6 @@ public async Task RecoverAsync_UnsupportedLegacyRecordCanRetryAfterRepairWithout var exception = await Assert.ThrowsAsync( () => sut.Lifecycle.OnStart(TestContext.Current.CancellationToken) .WaitAsync(TimeSpan.FromSeconds(10), TestContext.Current.CancellationToken)); - var initialization = sut.Manager.InitializeAsync(TestContext.Current.CancellationToken).AsTask(); - await storage.BlockedReadStarted.Task.WaitAsync(TimeSpan.FromSeconds(10), TestContext.Current.CancellationToken); Assert.Equal( "Failed to recover journaling state using journal format key 'orleans-binary'. " + @@ -1684,6 +1551,11 @@ public async Task RecoverAsync_UnsupportedLegacyRecordCanRetryAfterRepairWithout Assert.Empty(storage.Replaces); Assert.Empty(storage.OperationLog); + await sut.Manager.DisposeAsync(); + sut = CreateTestSystem(storage: storage); + value = new DurableValue("value", sut.Manager, codec); + var initialization = sut.Manager.InitializeAsync(TestContext.Current.CancellationToken).AsTask(); + await storage.BlockedReadStarted.Task.WaitAsync(TimeSpan.FromSeconds(10), TestContext.Current.CancellationToken); storage.AllowBlockedRead.SetResult(); await initialization.WaitAsync(TimeSpan.FromSeconds(10), TestContext.Current.CancellationToken); @@ -2053,8 +1925,6 @@ private sealed class BlockingAppendStorage : IJournalStorage public TaskCompletionSource AllowFirstAppend { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously); - public Exception? FirstAppendException { get; init; } - public List Appends { get; } = []; public bool IsCompactionRequested => false; @@ -2074,10 +1944,6 @@ public async ValueTask AppendAsync(ReadOnlySequence value, CancellationTok { FirstAppendStarted.SetResult(); await AllowFirstAppend.Task.WaitAsync(cancellationToken); - if (FirstAppendException is { } exception) - { - throw exception; - } } Appends.Add(value.ToArray()); @@ -2118,40 +1984,6 @@ public async ValueTask DeleteAsync(CancellationToken cancellationToken) } } - private sealed class BlockingRecoveryStorage : IJournalStorage - { - private int _readCount; - - public TaskCompletionSource RecoveryReadStarted { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously); - - public TaskCompletionSource AllowRecoveryRead { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously); - - public bool IsCompactionRequested => false; - - public async ValueTask ReadAsync(IJournalStorageConsumer consumer, CancellationToken cancellationToken) - { - if (Interlocked.Increment(ref _readCount) == 2) - { - RecoveryReadStarted.SetResult(); - await AllowRecoveryRead.Task.WaitAsync(cancellationToken); - if (RecoveryReadException is { } exception) - { - throw exception; - } - } - - consumer.Complete(metadata: null); - } - - public Exception? RecoveryReadException { get; init; } - - public ValueTask ReplaceAsync(ReadOnlySequence value, CancellationToken cancellationToken) => default; - - public ValueTask AppendAsync(ReadOnlySequence value, CancellationToken cancellationToken) => default; - - public ValueTask DeleteAsync(CancellationToken cancellationToken) => default; - } - private sealed class CapturingStorage : IJournalStorage { private readonly object _lock = new(); @@ -2181,6 +2013,8 @@ public byte[] RecoverableBytes public TaskCompletionSource AppendEntered { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously); + public TaskCompletionSource BlockedAppendStarted { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously); + public TaskCompletionSource ReleaseAppend { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously); public TaskCompletionSource DeleteEntered { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously); @@ -2199,13 +2033,11 @@ public byte[] RecoverableBytes public Exception? NextAppendException { get; set; } - public Exception? NextReadException { get; set; } - - public bool BlockNextRead { get; set; } + public Exception? NextPostAppendException { get; set; } - public TaskCompletionSource BlockedReadStarted { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously); + public Exception? NextDeleteException { get; set; } - public TaskCompletionSource AllowBlockedRead { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously); + public Exception? NextReadException { get; set; } public Exception? NextReplaceException { get; set; } @@ -2238,13 +2070,6 @@ public async ValueTask ReadAsync(IJournalStorageConsumer consumer, CancellationT throw exception; } - if (BlockNextRead) - { - BlockNextRead = false; - BlockedReadStarted.SetResult(); - await AllowBlockedRead.Task.WaitAsync(cancellationToken); - } - byte[][] segments; lock (_lock) { @@ -2370,24 +2195,31 @@ public async ValueTask AppendAsync(ReadOnlySequence value, CancellationTok } } - if (exceptionToThrow is not null) - { - ExceptionDispatchInfo.Throw(exceptionToThrow); - } - AppendEntered.TrySetResult(); if (BlockNextAppend) { BlockNextAppend = false; + BlockedAppendStarted.SetResult(); await ReleaseAppend.Task.WaitAsync(cancellationToken); } + if (exceptionToThrow is not null) + { + ExceptionDispatchInfo.Throw(exceptionToThrow); + } + var bytes = value.ToArray(); lock (_lock) { Appends.Add(bytes); _segments.Add(bytes); } + + if (NextPostAppendException is { } postAppendException) + { + NextPostAppendException = null; + throw postAppendException; + } } finally { @@ -2398,6 +2230,12 @@ public async ValueTask AppendAsync(ReadOnlySequence value, CancellationTok public ValueTask DeleteAsync(CancellationToken cancellationToken) { cancellationToken.ThrowIfCancellationRequested(); + if (NextDeleteException is { } exception) + { + NextDeleteException = null; + throw exception; + } + lock (_lock) { DeleteEnteredWhileAppendInProgress = Volatile.Read(ref _activeAppends) > 0;