feat(event-sourcing): add journaled state log consistency provider

[ReubenBond]

Problem

Orleans.EventSourcing currently has storage-backed log consistency providers, but no provider that stores the relevant grain's event log in its own Orleans.Journaling state machine.

Solution

Add a journaled-state-backed log consistency provider which registers durable state for the event log and write-vector metadata. The provider wires journaling registration through silo builder extensions, uses Orleans binary journaling for arbitrary event payloads, performs explicit durable rereads for recovery, and clears only provider-owned journaled state so unrelated grain state is preserved.

Rationale

Using the grain's journaled state keeps event-sourcing events colocated with the activation's journaling infrastructure while preserving the existing log consistency protocol semantics. The explicit read API avoids relying on initialization for ambiguous write recovery, and the clear behavior avoids deleting unrelated durable state owned by the same grain.

Notes

The new provider depends on the experimental Orleans.Journaling APIs, so warning suppressions are kept local to the new integration points.

Microsoft Reviewers: Open in CodeFlow

[Copilot]

Pull request overview

Introduces a new log-consistency provider, Orleans.EventSourcing.JournaledState, that stores the event log and write-vector for log-consistent grains inside the grain's own Orleans.Journaling state machine. To support the explicit-reread recovery flow used by the provider, IJournaledStateManager gains a new ReadStateAsync method backed by a ReadStateWorkItem in the journaling work loop, and a related migration-snapshot fix avoids double-appending during format migration. New silo builder extensions, public-API surface updates, and end-to-end tests (including a grain variant with auxiliary durable state) round out the change.

Changes:

• Add IJournaledStateManager.ReadStateAsync and the matching work-loop handler; suppress an append during the migration-snapshot path; honor cancellation on Initialize/Delete/Read via WaitAsync.
• Add the JournaledState log-consistency provider (LogConsistencyProvider, LogViewAdaptor<TLogView, TLogEntry>) and AddJournaledStateBasedLogConsistencyProvider[AsDefault] silo-builder extensions, registering durable list/value for the event log and write-vector.
• Add tests for the new provider (persistence across activation, Clear preserving auxiliary journaled state), a journaling read-state test, a JSON-to-binary migration "no-append" test, plus interface plumbing in test grains and fixtures; update generated API surfaces; swap storedEtag/currentEtag arguments in AzureBlobJournalStorage.CreateInconsistentWalStateException.

Show a summary per file

File	Description
src/Orleans.EventSourcing/JournaledState/LogConsistencyProvider.cs	New ILogViewAdaptorFactory that constructs the journaled-state log view adaptor.
src/Orleans.EventSourcing/JournaledState/LogViewAdaptor.cs	Core adaptor that persists event log and write-vector via IJournaledStateManager, with read/write/recovery loops.
src/Orleans.EventSourcing/Hosting/JournaledStateSiloBuilderExtensions.cs	Silo builder extensions to register the provider; sets JournalFormatKey to orleans-binary.
src/Orleans.EventSourcing/Orleans.EventSourcing.csproj	Adds project reference to Orleans.Journaling.
src/Orleans.Journaling/IJournaledStateManager.cs	Adds ReadStateAsync interface method.
src/Orleans.Journaling/JournaledStateManager.cs	Implements ReadStateAsync/ReadStateWorkItem, applies WaitAsync for cancellation, and avoids appending during migration snapshots.
src/Azure/Orleans.Journaling.AzureStorage/AzureBlobJournalStorage.cs	Swaps the storedEtag/currentEtag arguments when creating InconsistentStateException.
src/api/Orleans.EventSourcing/Orleans.EventSourcing.cs	Public-API surface updates for new provider and extensions.
src/api/Orleans.Journaling/Orleans.Journaling.cs	Public-API surface update for ReadStateAsync.
test/Orleans.EventSourcing.Tests/EventSourcingTests/EventSourcingClusterFixture.cs	Registers the new provider and a volatile journal storage in the test cluster.
test/Orleans.EventSourcing.Tests/EventSourcingTests/LogTestGrainClearTests.cs	Adds tests for Clear, persistence, and preservation of auxiliary durable state.
test/Grains/TestGrainInterfaces/ILogTestGrain.cs	Adds ILogTestGrainWithAuxiliaryState interface.
test/Grains/TestGrains/LogTestGrainVariations.cs	Adds journaled-state grain variants, including one with auxiliary durable state.
test/Orleans.Journaling.Tests/JournalBatchTests.cs	Adds test verifying ReadStateAsync discards unflushed changes.
test/Orleans.Journaling.Tests/DurableListDirectWriteTests.cs	Implements new ReadStateAsync on test fake.
test/Orleans.Journaling.Tests/DurableCollectionDirectWriteTests.cs	Implements new ReadStateAsync on test fake.
test/Orleans.Journaling.Json.Tests/CodecRecoveryTests.cs	Adds test that JSON→binary migration replaces without appending, plus a CountingStorage helper.

Copilot's findings

• Files reviewed: 17/17 changed files
• Comments generated: 2

[Copilot]

Review details

• Files reviewed: 17/17 changed files
• Comments generated: 1
• Review effort level: Lite

[Copilot]

Review details

Suppressed comments (1)

Previously missed (1) — in code that hasn't changed since the last review.

src/Orleans.EventSourcing/JournaledState/LogViewAdaptor.cs:68

• RetrieveLogSegment allocates enumerators and an intermediate array via LINQ (Skip/Take/ToArray). This is on a protocol hot path and differs from other adaptors (eg, LogStorage/LogViewAdaptor.cs uses List.GetRange). Consider copying via indexing to avoid LINQ allocations.

            return Task.FromResult<IReadOnlyList<TLogEntry>>(eventLog.Skip(fromVersion).Take(toVersion - fromVersion).ToArray());

• Files reviewed: 17/17 changed files
• Comments generated: 1
• Review effort level: Lite

[Copilot]

Review details

Suppressed comments (3)

Previously missed (2) — in code that hasn't changed since the last review.

src/Orleans.EventSourcing/JournaledState/LogViewAdaptor.cs:67

• RetrieveLogSegment currently uses Skip/Take/ToArray over IDurableList, which adds iterator/alloc overhead on a potentially hot path. Consider using an index-based copy (similar to src/Orleans.EventSourcing/LogStorage/LogViewAdaptor.cs:91-98 using GetRange).

            return Task.FromResult<IReadOnlyList<TLogEntry>>(eventLog.Skip(fromVersion).Take(toVersion - fromVersion).ToArray());

test/Orleans.EventSourcing.Tests/EventSourcingTests/LogTestGrainClearTests.cs:88

• Using Task.Delay after grain.Deactivate() can be flaky and slows the test suite. Prefer the deterministic test helper TestCluster.DeactivateAsync(grain) (see test/Orleans.DefaultCluster.Tests/GrainActivateDeactivateTests.cs:72) to ensure the activation is actually torn down before re-reading state.

This issue also appears on line 116 of the same file.

            await grain.Deactivate();
            await Task.Delay(TimeSpan.FromMilliseconds(100));

test/Orleans.EventSourcing.Tests/EventSourcingTests/LogTestGrainClearTests.cs:118

• Using Task.Delay after grain.Deactivate() can be flaky and slows the test suite. Prefer the deterministic test helper TestCluster.DeactivateAsync(grain) (see test/Orleans.DefaultCluster.Tests/GrainActivateDeactivateTests.cs:72) to ensure the activation is actually torn down before re-reading state.

            await grain.Deactivate();
            await Task.Delay(TimeSpan.FromMilliseconds(100));

• Files reviewed: 17/17 changed files
• Comments generated: 0 new
• Review effort level: Lite

[Copilot]

Review details

• Files reviewed: 31/31 changed files
• Comments generated: 1
• Review effort level: Lite

[Copilot]

Review details

Suppressed comments (2)

Previously missed (2) — in code that hasn't changed since the last review.

src/Orleans.Journaling/IJournaledStateManager.cs:72

• The default implementation of HasPendingWrites treats PendingWriteByteCount == -1 ("unsupported") as having pending writes because it only checks "!= 0". That makes HasPendingWrites always true for implementations which leave PendingWriteByteCount at its default -1 (e.g., lightweight test/benchmark managers), which is incorrect and can cause spurious flush decisions.

    /// <summary>
    /// Gets a value indicating whether any registered state has changes which have not been written to storage.
    /// </summary>
    bool HasPendingWrites => PendingWriteByteCount != 0;

test/Orleans.EventSourcing.Tests/EventSourcingTests/EventSourcingClusterFixture.cs:36

• JournalStorageProvider is static and wraps VolatileJournalStorageProvider (which keeps an in-memory store in an instance field). This makes journal storage state and injected failures shared across all EventSourcingClusterFixture instances (and therefore across parallel test classes), risking cross-test contamination/flakiness and leaking persisted journal data between clusters.

    public void FailNextJournalAppend(IAddressable grain, Exception exception, bool afterWrite = false)
    {
        JournalStorageProvider.FailNextAppend(JournalId.FromGrainId(grain.GetGrainId()), exception, afterWrite);
    }

    private static readonly FaultInjectingJournalStorageProvider JournalStorageProvider = new();

• Files reviewed: 31/31 changed files
• Comments generated: 0 new
• Review effort level: Lite

[Copilot]

Review details

• Files reviewed: 33/33 changed files
• Comments generated: 2
• Review effort level: Lite

[Copilot]

Copilot review overview

Review tier: Lite
Findings: 1

New issues introduced by this change (1)

Severity	Finding
	src/​Orleans.EventSourcing/​JournaledState/​LogViewAdaptor.cs — RetrieveLogSegment uses LINQ (Skip/Take/ToArray) on every call, which adds iterator/alloc…

[Copilot]

Copilot review overview

Review tier: Lite
Findings: 1

New issues introduced by this change (1)

Severity	Finding
	src/​Orleans.EventSourcing/​JournaledState/​LogViewAdaptor.cs — RetrieveLogSegment validates toVersion against _eventLog.Count, but WriteAsync appends…

Issues resolved since last review (1)

Severity	Finding
	src/​Orleans.EventSourcing/​JournaledState/​LogViewAdaptor.cs — RetrieveLogSegment uses LINQ (Skip/Take/ToArray) on every call, which adds iterator/alloc… View resolved comment

Suppressed comments (2)

Previously missed (2) — in code that hasn't changed since the last review.

test/Orleans.Journaling.Tests/StateManagerTests.cs:8

• using Orleans.Core; is unused in this test file. With code-style analyzers enabled during build, this can fail the build (IDE0005/unused using). Remove the directive.
  src/Orleans.Journaling/IJournaledStateManager.cs:82
• The default implementation of HasPendingWrites treats the sentinel value PendingWriteByteCount == -1 ("sampling not supported") as "has pending writes" because it checks != 0. That makes HasPendingWrites unreliable for implementations which don't override these members.

    long PendingWriteByteCount => -1;

    /// <summary>
    /// Gets a value indicating whether any registered state has changes which have not been written to storage.
    /// </summary>
    bool HasPendingWrites => PendingWriteByteCount != 0;

[Copilot]

Copilot review overview

Review tier: Lite
Findings: 1

New issues introduced by this change (1)

Severity	Finding
	src/​Orleans.Journaling/​IJournaledState.cs — HasPendingChanges defaults to true, which makes any IJournaledState implementation that…

Issues resolved since last review (1)

Severity	Finding
	src/​Orleans.EventSourcing/​JournaledState/​LogViewAdaptor.cs — RetrieveLogSegment validates toVersion against _eventLog.Count, but WriteAsync appends… View resolved comment

Suppressed comments (1)

Previously missed (1) — in code that hasn't changed since the last review.

src/Orleans.Journaling/IJournaledStateManager.cs:83

• HasPendingWrites is implemented as PendingWriteByteCount != 0, but PendingWriteByteCount explicitly returns a negative value (default -1) when an implementation doesn't support sampling. With the current default, those implementations will always report HasPendingWrites == true, which defeats the purpose of the property and can cause callers to flush unnecessarily. Consider treating negative values as "unknown" and defaulting to false unless the implementation can positively report pending bytes.

    /// <summary>
    /// Gets an approximate count of bytes accumulated in the in-memory journal buffer that have
    /// not yet been flushed to storage. Returns a negative value when the implementation does not
    /// support sampling pending bytes.
    /// </summary>
    /// <remarks>
    /// This is intended for diagnostics and instrumentation; the returned value may race with
    /// concurrent writers and should not be used for correctness decisions.
    /// </remarks>
    long PendingWriteByteCount => -1;

    /// <summary>
    /// Gets a value indicating whether any registered state has changes which have not been written to storage.
    /// </summary>
    bool HasPendingWrites => PendingWriteByteCount != 0;
}

[Copilot]

Copilot review overview

Review tier: Lite
Findings: None

Issues resolved since last review (2)

Severity	Finding
	src/​Orleans.Journaling/​IJournaledState.cs — HasPendingChanges defaults to true, which makes any IJournaledState implementation that… View resolved comment
	src/​Orleans.EventSourcing/​JournaledState/​LogViewAdaptor.cs — RetrieveLogSegment validates toVersion against _eventLog.Count, but WriteAsync appends… View resolved comment

Suppressed comments (1)

Previously missed (1) — in code that hasn't changed since the last review.

src/Orleans.Journaling/IJournaledStateManager.cs:82

• HasPendingWrites defaults to PendingWriteByteCount != 0, but PendingWriteByteCount itself defaults to -1 for implementations which don't support sampling. That makes HasPendingWrites permanently true for those implementations, which can cause unnecessary flush/recovery loops in callers which use this signal. Consider treating negative (unsupported) as "no reliable signal" by defaulting to > 0 instead of != 0.

    /// <summary>
    /// Gets a value indicating whether any registered state has changes which have not been written to storage.
    /// </summary>
    bool HasPendingWrites => PendingWriteByteCount != 0;