Skip to content
Closed
Show file tree
Hide file tree
Changes from 17 commits
Commits
Show all changes
22 commits
Select commit Hold shift + click to select a range
8d98f37
feat(journaling): add operation prerequisite and completion hooks
ReubenBond Oct 10, 2026
2c30156
fix(journaling): close work admission during hook shutdown
ReubenBond Oct 10, 2026
8b5b3e9
test(journaling): initialize storage callback barriers before registr…
ReubenBond Oct 10, 2026
2f21229
refactor(journaling): trim operation hook API and preparation
ReubenBond Oct 11, 2026
33749ac
fix(journaling): preserve hooks for queued persistence
ReubenBond Oct 11, 2026
43b25a0
feat(journaling)!: require journal operation hook support
ReubenBond Oct 11, 2026
659766b
refactor(journaling): simplify hook execution and failure state
ReubenBond Oct 11, 2026
104fdab
refactor(journaling): fix hook registration at initialization
ReubenBond Oct 11, 2026
16843f0
refactor(journaling): name the persistence operation type
ReubenBond Oct 11, 2026
a3523e7
refactor(journaling): select explicit snapshot work directly
ReubenBond Oct 11, 2026
363c048
refactor(journaling): guard snapshot selection with when
ReubenBond Oct 11, 2026
07be377
refactor(journaling): centralize snapshot requirement
ReubenBond Oct 11, 2026
46f1e99
refactor(journaling): coalesce pending journal writes
ReubenBond Oct 11, 2026
6b77e55
refactor(journaling): consume pending snapshot requirements
ReubenBond Oct 11, 2026
158fcf3
refactor(journaling): order hook preparation by registration
ReubenBond Oct 11, 2026
470ec86
refactor(journaling): decouple hook registration freeze
ReubenBond Oct 11, 2026
065c3da
feat(journaling)!: serialize dynamic hook registration
ReubenBond Oct 11, 2026
5ab481c
refactor(journaling)!: keep operation hooks private
ReubenBond Oct 12, 2026
8d58d85
fix(journaling): capture directly after synchronous final hooks
ReubenBond Oct 12, 2026
2da269a
refactor(journaling): extract capture-safe operation helpers
ReubenBond Oct 12, 2026
ded01ef
refactor(journaling): simplify recovery and before-hook traversal
ReubenBond Oct 12, 2026
d391273
refactor(journaling): derive disposed state from shared completion
ReubenBond Oct 12, 2026
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -54,7 +54,12 @@ Durable collections encode their operation before applying it to the in-memory c
- **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.

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.
Concurrent calls made while a write is queued share that queued operation. A snapshot requirement
from any caller upgrades pending work to a snapshot. The manager consumes the requirement when
selecting a write and restores it if a prerequisite hook fails. Storage acknowledgement leaves
newer pending requirements untouched.
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 `WriteStateAsync` completes. Recovery reconstructs durable state in a new activation.
Expand All @@ -68,6 +73,50 @@ Orleans executes that synchronous block on a single activation thread. Another g
the operation awaits, so keep shared state safe to commit at each await. Any caller's write can include
staged mutations from other calls.

## Journal operation hooks

Activation-scoped features coordinate prerequisites and completion through
<xref:Orleans.Journaling.IJournaledStateManager.Hooks>. The owner exposes a stable, lazily
allocated read-only view of <xref:Orleans.Journaling.IJournaledStateHook> registrations.
Use <xref:Orleans.Journaling.IJournaledStateManager.AddHookAsync*> and
<xref:Orleans.Journaling.IJournaledStateManager.RemoveHookAsync*> to change the set. Idle changes
apply synchronously; busy changes queue between complete operations, after current after callbacks.
The hook set stays stable throughout each operation. Queued changes also separate coalesced writes,
so writes admitted after a change observe that change.
Cancellation ends only the caller's wait while the manager owns admitted changes.
Every owner implements the registration methods and callback boundaries; delegating owners forward
them to their inner owner. Synchronous state callbacks can request a queued change and observe its
completion outside the callback. Standalone callers coordinate view inspection with changes.

Each actual append, snapshot, or deletion runs before callbacks in reverse list order, outside
the manager lock. Add the final prerequisite at index zero so its callback runs last before
synchronous capture or storage deletion. Prerequisites remain valid through capture or deletion, covering changes staged
during asynchronous preparation and while the work loop resumes. Preserve operation-local bookkeeping for
the captured batch separately from changes staged later.

Storage acknowledgement and registered-state acknowledgement or reset precede after callbacks.
All after callbacks run in list order, including for successful zero-byte writes. Coalesced callers
share callbacks for the actual operation. Features implement the before and after callbacks on
their own hook, retaining feature identity and operation-local bookkeeping there.

A failed prerequisite reports <xref:Orleans.Journaling.JournaledStatePreCommitException> with pending
state retained for an explicit persistence retry after the prerequisite is restored. A failed after
callback reports <xref:Orleans.Journaling.JournaledStatePostCommitException> with persistence completed.
Remaining after callbacks run, multiple failures are aggregated, and the manager stays usable.
The feature's durable recovery protocol resumes interrupted post-persistence work. Storage and
state-processing failures retain the manager's fencing and fresh-recovery behavior.

Hook callbacks receive the owner's shutdown token. Cancelling a caller's wait leaves the owned
operation running through its actual outcome. Disposal drains owned hooks and storage before releasing
journal resources, including when cancellation callbacks or cleanup fail. Concurrent disposal callers
share this completion. Hook implementations complete without calling initialization, persistence,
or disposal on their own owner. Awaiting an operation serialized behind the current callback creates
a circular dependency. Dependencies between journal owners also remain acyclic.
Shutdown closes work admission and cancels queued operations while the current
operation drains to its actual storage and hook outcome. For deletion, the feature owner stops
admission and drains feature operations before
queuing the whole-journal reset.

## 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.
Expand Down
42 changes: 42 additions & 0 deletions src/Orleans.Journaling/CompatibilitySuppressions.xml
Original file line number Diff line number Diff line change
Expand Up @@ -239,13 +239,27 @@
<Right>lib/net8.0/Orleans.Journaling.dll</Right>
<IsBaselineSuppression>true</IsBaselineSuppression>
</Suppression>
<Suppression>
<DiagnosticId>CP0006</DiagnosticId>
<Target>M:Orleans.Journaling.IJournaledStateManager.AddHookAsync(Orleans.Journaling.IJournaledStateHook,System.Nullable{System.Int32},System.Threading.CancellationToken)</Target>
<Left>lib/net10.0/Orleans.Journaling.dll</Left>
<Right>lib/net10.0/Orleans.Journaling.dll</Right>
<IsBaselineSuppression>true</IsBaselineSuppression>
</Suppression>
<Suppression>
<DiagnosticId>CP0006</DiagnosticId>
<Target>M:Orleans.Journaling.IJournaledStateManager.RegisterStateMachine(System.String,Orleans.Journaling.IStateMachine)</Target>
<Left>lib/net10.0/Orleans.Journaling.dll</Left>
<Right>lib/net10.0/Orleans.Journaling.dll</Right>
<IsBaselineSuppression>true</IsBaselineSuppression>
</Suppression>
<Suppression>
<DiagnosticId>CP0006</DiagnosticId>
<Target>M:Orleans.Journaling.IJournaledStateManager.RemoveHookAsync(Orleans.Journaling.IJournaledStateHook,System.Threading.CancellationToken)</Target>
<Left>lib/net10.0/Orleans.Journaling.dll</Left>
<Right>lib/net10.0/Orleans.Journaling.dll</Right>
<IsBaselineSuppression>true</IsBaselineSuppression>
</Suppression>
<Suppression>
<DiagnosticId>CP0006</DiagnosticId>
<Target>M:Orleans.Journaling.IJournaledStateManager.TryGetStateMachine(System.String,Orleans.Journaling.IStateMachine@)</Target>
Expand All @@ -267,20 +281,41 @@
<Right>lib/net10.0/Orleans.Journaling.dll</Right>
<IsBaselineSuppression>true</IsBaselineSuppression>
</Suppression>
<Suppression>
<DiagnosticId>CP0006</DiagnosticId>
<Target>P:Orleans.Journaling.IJournaledStateManager.Hooks</Target>
<Left>lib/net10.0/Orleans.Journaling.dll</Left>
<Right>lib/net10.0/Orleans.Journaling.dll</Right>
<IsBaselineSuppression>true</IsBaselineSuppression>
</Suppression>
<Suppression>
<DiagnosticId>CP0006</DiagnosticId>
<Target>P:Orleans.Journaling.IJournalMetadata.FormatKey</Target>
<Left>lib/net10.0/Orleans.Journaling.dll</Left>
<Right>lib/net10.0/Orleans.Journaling.dll</Right>
<IsBaselineSuppression>true</IsBaselineSuppression>
</Suppression>
<Suppression>
<DiagnosticId>CP0006</DiagnosticId>
<Target>M:Orleans.Journaling.IJournaledStateManager.AddHookAsync(Orleans.Journaling.IJournaledStateHook,System.Nullable{System.Int32},System.Threading.CancellationToken)</Target>
<Left>lib/net8.0/Orleans.Journaling.dll</Left>
<Right>lib/net8.0/Orleans.Journaling.dll</Right>
<IsBaselineSuppression>true</IsBaselineSuppression>
</Suppression>
<Suppression>
<DiagnosticId>CP0006</DiagnosticId>
<Target>M:Orleans.Journaling.IJournaledStateManager.RegisterStateMachine(System.String,Orleans.Journaling.IStateMachine)</Target>
<Left>lib/net8.0/Orleans.Journaling.dll</Left>
<Right>lib/net8.0/Orleans.Journaling.dll</Right>
<IsBaselineSuppression>true</IsBaselineSuppression>
</Suppression>
<Suppression>
<DiagnosticId>CP0006</DiagnosticId>
<Target>M:Orleans.Journaling.IJournaledStateManager.RemoveHookAsync(Orleans.Journaling.IJournaledStateHook,System.Threading.CancellationToken)</Target>
<Left>lib/net8.0/Orleans.Journaling.dll</Left>
<Right>lib/net8.0/Orleans.Journaling.dll</Right>
<IsBaselineSuppression>true</IsBaselineSuppression>
</Suppression>
<Suppression>
<DiagnosticId>CP0006</DiagnosticId>
<Target>M:Orleans.Journaling.IJournaledStateManager.TryGetStateMachine(System.String,Orleans.Journaling.IStateMachine@)</Target>
Expand All @@ -302,6 +337,13 @@
<Right>lib/net8.0/Orleans.Journaling.dll</Right>
<IsBaselineSuppression>true</IsBaselineSuppression>
</Suppression>
<Suppression>
<DiagnosticId>CP0006</DiagnosticId>
<Target>P:Orleans.Journaling.IJournaledStateManager.Hooks</Target>
<Left>lib/net8.0/Orleans.Journaling.dll</Left>
<Right>lib/net8.0/Orleans.Journaling.dll</Right>
<IsBaselineSuppression>true</IsBaselineSuppression>
</Suppression>
<Suppression>
<DiagnosticId>CP0006</DiagnosticId>
<Target>P:Orleans.Journaling.IJournalMetadata.FormatKey</Target>
Expand Down
62 changes: 62 additions & 0 deletions src/Orleans.Journaling/IJournaledStateHook.cs
Original file line number Diff line number Diff line change
@@ -0,0 +1,62 @@
namespace Orleans.Journaling;

/// <summary>
/// Identifies a journal persistence operation.
/// </summary>
public enum JournaledStateOperation
{
/// <summary>Appends pending journal entries.</summary>
Write,

/// <summary>Replaces storage with a snapshot.</summary>
Snapshot,

/// <summary>Deletes storage and resets registered states.</summary>
Delete
}

/// <summary>
/// Participates in the prerequisites and completion of actual journal operations.
/// </summary>
/// <remarks>
/// Before hooks run in reverse list order on the owner's logical execution context, outside its lock.
/// The hook at index zero supplies the final prerequisite before capture or deletion.
/// Prerequisites remain valid until capture or deletion, covering changes staged during asynchronous
/// preparation and while the journal work loop resumes. After hooks run in forward list order after
/// storage acknowledgement and state acknowledgement or reset, including successful writes which
/// produce no storage bytes. Hooks retain operation-local data across these boundaries and keep later
/// pending changes separate. Hook implementations must complete without calling initialization,
/// persistence, or disposal on their own journal owner. Those operations are serialized behind the
/// current operation, so awaiting them from a hook would create a circular dependency.
/// Dependencies between journal owners must also remain acyclic.
/// Callbacks can request a hook registration change, observing its queued completion outside the
/// callback rather than awaiting work which follows the current operation.
/// </remarks>
public interface IJournaledStateHook
{
/// <summary>
/// Establishes prerequisites before capture or storage deletion.
/// </summary>
/// <param name="operation">The operation about to execute.</param>
/// <param name="cancellationToken">The token for the owned operation's lifetime.</param>
/// <returns>A completion representing the prerequisite work.</returns>
/// <remarks>
/// Failure reports <see cref="JournaledStatePreCommitException"/> and retains pending changes
/// for an explicit retry. All staged changes remain safe to commit. Full deletion requires the
/// owner to stop admission and drain feature operations before queuing deletion.
/// </remarks>
ValueTask BeforeOperationAsync(JournaledStateOperation operation, CancellationToken cancellationToken) => default;

/// <summary>
/// Completes post-persistence work after the operation succeeds.
/// </summary>
/// <param name="operation">The completed operation.</param>
/// <param name="cancellationToken">The token for the owned operation's lifetime.</param>
/// <returns>A completion representing post-persistence work.</returns>
/// <remarks>
/// Every after hook is invoked even when an earlier after hook fails. Failures are surfaced as
/// <see cref="JournaledStatePostCommitException"/> and leave the successfully persisted owner usable.
/// Durable feature state supplies recovery for interrupted post-persistence work.
/// </remarks>
ValueTask AfterOperationAsync(JournaledStateOperation operation, CancellationToken cancellationToken) => default;
}
54 changes: 51 additions & 3 deletions src/Orleans.Journaling/IJournaledStateManager.cs
Original file line number Diff line number Diff line change
Expand Up @@ -8,10 +8,52 @@ namespace Orleans.Journaling;
/// <remarks>
/// The owner registers state machines and initializes the journal before using recovered state.
/// State machine instances and their dependencies retain the lifetime assigned by their caller.
/// Disposing this manager stops journal processing and releases its journal resources.
/// Disposal drains owned storage and hook operations before releasing resources.
/// </remarks>
public interface IJournaledStateManager : IAsyncDisposable
{
/// <summary>
/// Gets the read-only view of registered journal operation hooks.
/// </summary>
/// <remarks>
/// Inspect hooks on the owner's logical execution context, coordinating standalone inspection
/// with registration changes. Use <see cref="AddHookAsync"/> and <see cref="RemoveHookAsync"/>
/// to change registrations. Before hooks execute in
/// reverse list order; after hooks execute in forward list order. Register the final capture
/// prerequisite at index zero so it runs last before capture or deletion.
/// Registration is independent of state-machine registration and persists through recovery and deletion.
/// Every implementation provides this list and invokes its hooks at the documented operation boundaries.
/// </remarks>
IReadOnlyList<IJournaledStateHook> Hooks { get; }

/// <summary>
/// Adds a hook before the next persistence operation which follows this registration.
/// </summary>
/// <param name="hook">The hook to register.</param>
/// <param name="index">The insertion index, or <see langword="null"/> to append.</param>
/// <param name="cancellationToken">The token used to cancel the caller's wait.</param>
/// <returns>A completion representing application of the registration.</returns>
/// <remarks>
/// Idle registration applies synchronously. Otherwise it is queued between owned operations,
/// after the current operation's after hooks finish. Queued registration is a write-coalescing barrier.
/// Cancellation ends the caller's wait while admitted registration remains owned by the manager.
/// Synchronous state callbacks can request registration and observe its completion outside the callback.
/// Await registration before staging changes whose safety depends on the added hook.
/// </remarks>
ValueTask AddHookAsync(IJournaledStateHook hook, int? index = null, CancellationToken cancellationToken = default);

/// <summary>
/// Removes a hook between persistence operations.
/// </summary>
/// <param name="hook">The registered hook to remove.</param>
/// <param name="cancellationToken">The token used to cancel the caller's wait.</param>
/// <returns>A completion representing application of the removal.</returns>
/// <exception cref="InvalidOperationException">The hook is not registered when the removal is applied.</exception>
/// <remarks>
/// Removal has the same ownership and ordering guarantees as <see cref="AddHookAsync"/>.
/// </remarks>
ValueTask RemoveHookAsync(IJournaledStateHook hook, CancellationToken cancellationToken = default);

/// <inheritdoc/>
ValueTask IAsyncDisposable.DisposeAsync() => default;

Expand Down Expand Up @@ -51,8 +93,11 @@ public interface IJournaledStateManager : IAsyncDisposable
/// <remarks>
/// Stage mutations only after establishing that they are safe to commit. Pending changes are shared
/// by all callers using this manager. Storage acknowledgement establishes durability.
/// A failed journal operation fences the manager; recovery requires a new manager and state machine instances.
/// A storage or state-processing failure fences the manager; recovery requires a new manager and state machine instances.
/// Cancellation stops the caller's wait; an already queued write continues to its storage outcome.
/// Before-hook failure reports <see cref="JournaledStatePreCommitException"/> and retains pending
/// changes for an explicit retry. After-hook failure reports
/// <see cref="JournaledStatePostCommitException"/> after successful persistence.
/// </remarks>
/// <param name="cancellationToken">The token used to cancel the caller's wait.</param>
/// <returns>A task representing the write acknowledgement.</returns>
Expand All @@ -64,7 +109,10 @@ public interface IJournaledStateManager : IAsyncDisposable
/// <remarks>
/// The caller keeps other operations quiescent through completion: deletion resets every registered state machine.
/// Cancellation ends the caller's wait; an already queued deletion continues to its storage and reset outcome.
/// A failed deletion permanently fences the manager and requests deactivation of its owning grain.
/// A storage or state-reset failure permanently fences the manager and requests deactivation of its owning grain.
/// Before-hook failure reports <see cref="JournaledStatePreCommitException"/> with state retained.
/// After-hook failure reports <see cref="JournaledStatePostCommitException"/> after storage deletion
/// and state reset succeed.
/// </remarks>
/// <param name="cancellationToken">The cancellation token.</param>
/// <returns>A <see cref="ValueTask"/> which represents the operation.</returns>
Expand Down
Loading
Loading