test(journaling): pin binary journal compatibility

docs/site/src/content/docs/grains/journaling/configuration.md

@@ -152,6 +152,8 @@ Serializer naming policies affect application payload values. Journal command na
 
 Providers expose the persisted format key as <xref:Orleans.Journaling.IJournalMetadata.FormatKey> and <xref:Orleans.Journaling.JournalMetadata.FormatKey>. Recovery selects the stored reader independently of the configured write format. When they differ, the next write creates a full snapshot using the configured format and updates the metadata. <xref:Orleans.Journaling.Json.JsonLinesJournalFormat.JournalFormatKey> supplies the JSON Lines format key.
 
+When the stored format key is absent, recovery uses the configured format key. Configure `orleans-binary` to recover binary journals with missing format metadata. Before selecting a different write format, establish the stored key as `orleans-binary` using the provider's format-metadata mechanism and verify recovery. Back up the bytes and metadata together.
+
 Use this deployment sequence:
 
 1. Back up the journal data and provider metadata as one recoverable unit.
@@ -164,6 +166,8 @@ The Orleans binary format key is `orleans-binary`. Configure it explicitly while
 
 :::code language="csharp" source="./snippets/journaling/JournalingConfiguration.cs" id="configure_binary_format":::
 
+The binary reader recovers both legacy variable-length and V1 fixed-width entries. Current appends and snapshots use V1, preserving recovery of mixed journals through upgrades. See [OrleansBinary persisted frames](runtime-behavior.md#orleansbinary-persisted-frames) for the framing and command-codec contract.
+
 An unknown stored format key or incompatible payload fails recovery and leaves the journal unchanged.
 
 ## Configure state retirement

docs/site/src/content/docs/grains/journaling/runtime-behavior.md

@@ -35,6 +35,23 @@ State creation belongs to setup before initialization. After initialization begi
 
 Storage providers can split reads at arbitrary byte boundaries. The journal format buffers incomplete entries and only applies complete ordered records.
 
+### OrleansBinary persisted frames
+
+The `orleans-binary` format reads both legacy variable-length frames (V0) and fixed-width frames (V1). It selects the framing for each entry, so replay preserves physical command order when V1 appends follow a legacy journal. Current appends, preserved-entry writes, and snapshots emit V1.
+
+| Framing | Entry layout | Body length includes |
+| --- | --- | --- |
+| V0 | Orleans varuint32 body length, Orleans varuint64 stream ID, command-format byte `0`, command payload | Encoded stream ID, command-format byte, and command payload |
+| V1 | Framing byte `1`, little-endian uint32 body length, little-endian uint32 stream ID, command payload | Four-byte stream ID and command payload |
+
+The V0 length prefix uses Orleans' variable-length integer encoding. The V1 marker occupies the first byte of each entry and identifies the fixed-width layout. These framing identities remain stable across upgrades.
+
+The journal reader consumes the frame header and the V0 command-format byte before passing exactly one command payload to the state machine. Dictionary, list, queue, set, value, and persistent-state codecs read an Orleans varuint32 command discriminator; the durable task completion source codec reads a single status byte. Both framings use the same command payload encoding, command identities, and Orleans field codecs for operands.
+
+During streaming recovery, incomplete headers and bodies remain buffered until their full entry arrives. Completed input with a truncated or malformed entry fails recovery with the entry's byte offset. Unsupported legacy command-format versions fail before that entry is applied. Retain the readers and command codecs for persisted entries and retired streams throughout the upgrade and rollback window.
+
+Storage format metadata selects the journal reader independently of these per-entry framing versions. When the stored format key is absent, the configured format key selects the reader. Configure `orleans-binary` while recovering binary journals with missing format metadata; see [Migrate a journal format](configuration.md#migrate-a-journal-format) for changing the write format.
+
 ### Custom and caller-owned managers
 
 Grain-scoped managers are enrolled before resolution returns. The standard manager establishes this in its grain-bound constructor and implements both the grain-facing `IDurableStateManager` and the independent journal-owner <xref:Orleans.Journaling.IJournaledStateManager>. Custom grain manager replacements provide both contracts and their registrations as appropriate. They establish enrollment in their constructor or registration factory by enrolling their <xref:Orleans.ILifecycleParticipant`1> for <xref:Orleans.Runtime.IGrainLifecycle>, or subscribing initialization and shutdown callbacks directly. This gives ordinary grains, application-owned bases, and `DurableGrain` the same recovery and shutdown lifecycle for their injected managers.

src/Orleans.Journaling/README.md

@@ -74,7 +74,7 @@ Recovery reads the selected provider's physical namespace using the existing
 journal identity. Changing the selection for a grain type with existing journals
 requires a deliberate data migration or cutover strategy, including rollback.
 
-JSON Lines is the default `JournaledStateManagerOptions.JournalFormatKey`. Storage providers expose the stored journal format key through `IJournalMetadata.FormatKey` and `JournalMetadata.FormatKey`. During recovery, Orleans uses that stored key to select the matching journal format and durable operation codecs. If a non-empty journal has no stored format metadata, Orleans treats it as legacy OrleansBinary data for compatibility.
+JSON Lines is the default `JournaledStateManagerOptions.JournalFormatKey`. Storage providers expose the stored journal format key through `IJournalMetadata.FormatKey` and `JournalMetadata.FormatKey`. During recovery, Orleans uses that stored key to select the matching journal format and durable operation codecs. When the stored key is absent, recovery uses the configured format key. Configure `orleans-binary` to recover binary journals whose metadata has no format key.
 
 If you already have data written with the OrleansBinary format, you can keep using it while you plan a migration:
 
@@ -86,7 +86,9 @@ siloBuilder
             options.JournalFormatKey = "orleans-binary"));
 ```
 
-To migrate to JSON, configure `JournaledStateManagerOptions.JournalFormatKey` to `JsonLinesJournalFormat.JournalFormatKey` and call `UseJsonJournalFormat(...)`, provided by `JsonJournalHostingExtensions`. When a grain recovers data written with a different format than the configured write format, the next write is forced to a full snapshot so the journal is rewritten using JSON and the storage format metadata is updated.
+The OrleansBinary reader accepts legacy variable-length frames and V1 fixed-width frames in physical order, including a legacy journal followed by newly appended V1 entries. Current appends and snapshots use framing marker `1`. Legacy frames carry a command-format byte `0` before the command payload; the reader consumes it before dispatching to the same durable command codecs.
+
+To migrate to JSON, first ensure the binary journal's stored format key is `orleans-binary`, then configure `JournaledStateManagerOptions.JournalFormatKey` to `JsonLinesJournalFormat.JournalFormatKey` and call `UseJsonJournalFormat(...)`, provided by `JsonJournalHostingExtensions`. When a grain recovers data written with a different format than the configured write format, the next write is forced to a full snapshot so the journal is rewritten using JSON and the storage format metadata is updated.
 
 ## Example - Using durable state
 ```csharp
@@ -294,7 +296,7 @@ Each record contains the state id as element 0 and the durable operation payload
 
 Inside the operation payload array, element 0 is the command name, followed by command-specific operands such as keys, values, item arrays, or versions. Storage write batches append one or more complete JSON Lines records without adding a separate extent envelope or final container-close step.
 
-Existing data is read using its stored format metadata, or as legacy OrleansBinary data when metadata is absent, and migrated to the configured write format by the next snapshot write.
+Existing data is read using its stored format key, or the configured format key when the stored key is absent. When stored and configured formats differ, the next write replaces the journal with a snapshot in the configured format.
 
 ## Catalog enumeration
 

test/Orleans.Journaling.Tests/UpstreamMainCompatibilityTests.cs

@@ -1,4 +1,5 @@
 using System.Buffers;
+using System.Buffers.Binary;
 using System.Runtime.CompilerServices;
 using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Logging;
@@ -12,24 +13,23 @@
 namespace Orleans.Journaling.Tests;
 
 /// <summary>
-/// Cross-version recovery test: feeds a pre-canned journal blob captured from <c>upstream/main</c>
-/// (using the legacy <c>StateMachineManager</c> + inlined <c>Durable*</c> writers) through the PR's
-/// <see cref="JournaledStateManager"/> wired with the OrleansBinary codecs, and asserts the recovered
-/// in-memory state matches what main wrote.
+/// Recovers pinned legacy and released OrleansBinary journals, then verifies current append and
+/// snapshot writes recover the same durable state.
 /// </summary>
 /// <remarks>
 /// The binary fixture under <c>fixtures</c> was produced by the
 /// <c>CaptureCrossVersionRecoveryFixture.EmitMainJournalHex</c> test running against
-/// upstream/main (<c>5989958561</c>). The capture exercises every <c>Durable*</c> type and pins the
-/// stream-registration order so the journal stream IDs (8..14) line up with the registrations below.
-/// To regenerate after an intentional wire-format change to the OrleansBinary codecs, re-run that
-/// fixture in a worktree pinned to upstream/main, replace the fixture bytes, and accept the updated
-/// Verify snapshot under <c>snapshots</c>.
+/// upstream/main (<c>5989958561</c>) using the legacy <c>StateMachineManager</c> and inlined writers.
+/// The V1 hexadecimal fixtures were emitted by the codecs in the released
+/// <c>Microsoft.Orleans.Journaling 10.3.1-alpha.1</c> package, built from
+/// <c>137d9acc17830f15b13a4eb0058d6cee633cad5e</c>. Both journal fixtures exercise all seven durable
+/// types and pin the registration order to stream IDs 8..14. Keep these inputs pinned to their
+/// producing versions and add fixtures for future formats alongside them.
 /// </remarks>
 [TestSuite("BVT")]
 [TestProvider("None")]
 [TestCategory("BVT")]
-public sealed class UpstreamMainCompatibilityTests : JournalingTestBase
+public sealed class UpstreamMainCompatibilityTests : JournalingTestBase, IDisposable
 {
     [Fact]
     public async Task OrleansBinary_RecoversUpstreamMainWrittenJournal()
@@ -39,30 +39,141 @@ public async Task OrleansBinary_RecoversUpstreamMainWrittenJournal()
         await storage.AppendAsync(new ReadOnlySequence<byte>(journalBytes), TestContext.Current.CancellationToken);
 
         var states = CreateStates(storage);
+        await using var manager = states.Manager;
         using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(30));
-        await states.Manager.InitializeAsync(cts.Token);
+        await manager.InitializeAsync(cts.Token);
 
+        await AssertRecoveredStates(states, appended: false);
+        await Verify(JournalSnapshotFormatting.FormatBinary(journalBytes), extension: "txt").UseDirectory("snapshots");
+    }
+
+    [Theory]
+    [InlineData(false)]
+    [InlineData(true)]
+    public async Task OrleansBinary_RecoversReleasedBaselineWrittenJournal(bool hasFormatMetadata)
+    {
+        var journalBytes = LoadReleasedFixture("journal");
+        Assert.Equal(19, AssertV1Framing(journalBytes));
+        var storage = await CreateStorage(journalBytes, hasFormatMetadata);
+        var states = CreateStates(new CompatibilityStorage(storage));
+        await using var manager = states.Manager;
+
+        await manager.InitializeAsync(TestContext.Current.CancellationToken);
+
+        await AssertRecoveredStates(states, appended: false);
+        Assert.Equal(journalBytes, Assert.Single(storage.Segments));
+    }
+
+    [Theory]
+    [InlineData(false, false)]
+    [InlineData(false, true)]
+    [InlineData(true, false)]
+    [InlineData(true, true)]
+    public async Task OrleansBinary_AppendsAndCompactsPersistedJournalWithRecoverableV1(
+        bool legacyFraming,
+        bool hasFormatMetadata)
+    {
+        var journalBytes = legacyFraming ? LoadUpstreamMainJournal() : LoadReleasedFixture("journal");
+        var storage = await CreateStorage(journalBytes, hasFormatMetadata);
+        var streamingStorage = new CompatibilityStorage(storage);
+        var states = CreateStates(streamingStorage);
+        await using (var manager = states.Manager)
+        {
+            await manager.InitializeAsync(TestContext.Current.CancellationToken);
+            await AssertRecoveredStates(states, appended: false);
+
+            states.Dictionary["alpha"] = 42;
+            states.List.Add("fourth");
+            states.Queue.Enqueue(30);
+            states.Set.Add("baz");
+            states.Value.Value = 43;
+            ((IStorage<string>)states.State).State = "goodbye";
+            await manager.WriteStateAsync(TestContext.Current.CancellationToken);
+        }
+
+        Assert.Collection(storage.Segments,
+            original => Assert.Equal(journalBytes, original),
+            appended => Assert.Equal(LoadReleasedFixture("append"), appended));
+        Assert.Equal(OrleansBinaryJournalFormat.JournalFormatKey, storage.StoredJournalFormatKey);
+
+        var recovered = CreateStates(streamingStorage);
+        await using (var manager = recovered.Manager)
+        {
+            await manager.InitializeAsync(TestContext.Current.CancellationToken);
+            await AssertRecoveredStates(recovered, appended: true);
+
+            streamingStorage.IsCompactionRequested = true;
+            await manager.WriteStateAsync(TestContext.Current.CancellationToken);
+        }
+
+        Assert.Equal(9, AssertV1Framing(Assert.Single(storage.Segments)));
+        Assert.Equal(OrleansBinaryJournalFormat.JournalFormatKey, storage.StoredJournalFormatKey);
+        var compacted = CreateStates(streamingStorage);
+        await using (var manager = compacted.Manager)
+        {
+            await manager.InitializeAsync(TestContext.Current.CancellationToken);
+            await AssertRecoveredStates(compacted, appended: true);
+        }
+    }
+
+    private static async Task AssertRecoveredStates(DurableStates states, bool appended)
+    {
         Assert.Equal(2, states.Dictionary.Count);
-        Assert.Equal(1, states.Dictionary["alpha"]);
+        Assert.Equal(appended ? 42 : 1, states.Dictionary["alpha"]);
         Assert.Equal(2, states.Dictionary["beta"]);
 
-        Assert.Equal(["first", "second", "third"], states.List);
+        Assert.Equal(appended ? ["first", "second", "third", "fourth"] : ["first", "second", "third"], states.List);
 
-        Assert.Equal(2, states.Queue.Count);
-        Assert.Equal(10, states.Queue.Dequeue());
-        Assert.Equal(20, states.Queue.Dequeue());
+        Assert.Equal(appended ? 3 : 2, states.Queue.Count);
+        Assert.Equal(appended ? [10, 20, 30] : [10, 20], states.Queue);
 
-        Assert.True(states.Set.SetEquals(["foo", "bar"]));
+        Assert.Equal(appended ? 3 : 2, states.Set.Count);
+        Assert.True(states.Set.SetEquals(appended ? ["foo", "bar", "baz"] : ["foo", "bar"]));
 
-        Assert.Equal(42, states.Value.Value);
+        Assert.Equal(appended ? 43 : 42, states.Value.Value);
 
-        Assert.Equal("hello", ((IStorage<string>)states.State).State);
+        Assert.Equal(appended ? "goodbye" : "hello", ((IStorage<string>)states.State).State);
 
         Assert.Equal(DurableTaskCompletionSourceStatus.Completed, states.Tcs.State.Status);
         Assert.Equal(99, states.Tcs.State.Value);
-        Assert.Equal(99, await states.Tcs.Task);
+        Assert.Equal(99, await states.Tcs.Task.WaitAsync(TestContext.Current.CancellationToken));
+    }
 
-        await Verify(JournalSnapshotFormatting.FormatBinary(journalBytes), extension: "txt").UseDirectory("snapshots");
+    private static int AssertV1Framing(byte[] bytes)
+    {
+        var entryCount = 0;
+        var offset = 0;
+        while (offset < bytes.Length)
+        {
+            Assert.True(bytes.Length - offset >= 9);
+            Assert.Equal(1, bytes[offset]);
+            var bodyLength = BinaryPrimitives.ReadUInt32LittleEndian(bytes.AsSpan(offset + 1, sizeof(uint)));
+            Assert.True(bodyLength >= sizeof(uint));
+            Assert.True(bodyLength <= bytes.Length - offset - 5);
+            offset += checked(5 + (int)bodyLength);
+            entryCount++;
+        }
+
+        Assert.Equal(bytes.Length, offset);
+        return entryCount;
+    }
+
+    private static async Task<VolatileJournalStorage> CreateStorage(byte[] bytes, bool hasFormatMetadata)
+    {
+        var storage = new VolatileJournalStorage(hasFormatMetadata ? OrleansBinaryJournalFormat.JournalFormatKey : null);
+        await storage.AppendAsync(new ReadOnlySequence<byte>(bytes), TestContext.Current.CancellationToken);
+        Assert.Equal(hasFormatMetadata ? OrleansBinaryJournalFormat.JournalFormatKey : null, storage.StoredJournalFormatKey);
+        storage.SetConfiguredJournalFormatKey(OrleansBinaryJournalFormat.JournalFormatKey);
+        return storage;
+    }
+
+    private static byte[] LoadReleasedFixture(string kind, [CallerFilePath] string sourceFile = "")
+    {
+        var fixturePath = Path.Combine(
+            Path.GetDirectoryName(sourceFile)!,
+            "fixtures",
+            $"UpstreamMainCompatibilityTests.released-10.3.1-v1-{kind}.hex");
+        return Convert.FromHexString(File.ReadAllText(fixturePath).Trim());
     }
 
     private static byte[] LoadUpstreamMainJournal([CallerFilePath] string sourceFile = "")
@@ -117,6 +228,39 @@ private JournaledStateManager CreateManager(IJournalStorage storage)
 
     private DeepCopier<T> Copier<T>() => ServiceProvider.GetRequiredService<DeepCopier<T>>();
 
+    public void Dispose() => ServiceProvider.Dispose();
+
+    private sealed class CompatibilityStorage(VolatileJournalStorage storage) : IJournalStorage
+    {
+        public bool IsCompactionRequested { get; set; }
+
+        public async ValueTask ReadAsync(IJournalStorageConsumer consumer, CancellationToken cancellationToken)
+        {
+            var metadata = await storage.GetMetadataAsync(cancellationToken);
+            consumer.Read(ReadBytes(), metadata, complete: true);
+
+            IEnumerable<ReadOnlyMemory<byte>> ReadBytes()
+            {
+                foreach (var segment in storage.Segments)
+                {
+                    for (var i = 0; i < segment.Length; i++)
+                    {
+                        cancellationToken.ThrowIfCancellationRequested();
+                        yield return segment.AsMemory(i, 1);
+                    }
+                }
+            }
+        }
+
+        public ValueTask AppendAsync(ReadOnlySequence<byte> value, CancellationToken cancellationToken) =>
+            storage.AppendAsync(value, cancellationToken);
+
+        public ValueTask ReplaceAsync(ReadOnlySequence<byte> value, CancellationToken cancellationToken) =>
+            storage.ReplaceAsync(value, cancellationToken);
+
+        public ValueTask DeleteAsync(CancellationToken cancellationToken) => storage.DeleteAsync(cancellationToken);
+    }
+
     private sealed record DurableStates(
         JournaledStateManager Manager,
         DurableDictionary<string, int> Dictionary,

test/Orleans.Journaling.Tests/fixtures/UpstreamMainCompatibilityTests.released-10.3.1-v1-append.hex

@@ -0,0 +1 @@
+010E0000000800000001400B616C70686101A9010D0000000900000001400D666F7572746801070000000A000000010079010A0000000B00000001400762617A01070000000C0000000100AD010F0000000D00000001400F676F6F646279650301080000000E00000001001A03