Preserve strict native inspection and reject corrupt stores before provider writes

Author: KSemenenko

docs/ADR/ADR-060-native-internal-serialization.md

@@ -93,3 +93,44 @@ derived collection codecs fail closed before generated allocation, even below an
 object-typed field. Extending this closure requires a native shape specification,
 preflight/count/reference tests and homogeneous compatibility qualification.
 This is an internal concrete contract, not arbitrary Orleans codec compatibility.
+
+## Accepted qualification repair contract
+
+R11/R12 preserve AC-IS002/004/007 and the existing formats. Actual GitHub
+run37120641864 revealed nullable metadata construction, native reader exhaustion
+classification and stale fixtures. Native graph annotations must use .NET10's
+already-unwrapped Nullable<T> metadata; only genuine Orleans Reader buffer
+exhaustion joins coded corruption, while programming/session exceptions escape.
+
+R12F validates the complete recoverable journal/checkpoint prefix before opening
+ZoneTree's native provider. Under the already acquired node owner lock, use the
+same owned journal handle and bounded current codecs to verify complete frames,
+checksums, sequence, checkpoint metadata/footer and record semantics without apply,
+truncation or tree writes. Leave a permitted incomplete current tail untouched
+during preflight; ordinary ordered recovery alone applies/truncates it afterward.
+Unsupported complete legacy frames and complete corruption must fail without
+changing journal/identity/provider files. Reset the journal position before
+ordinary recovery; keep startup preflight distinct from acknowledged write gates.
+
+StorageRecovery owns the initializer and new preflight helper; UnitTests owns
+NativeStoreOpenPreflight regressions using real files plus unchanged historical
+file-preservation assertions. Root owns integration/docs; wire worker owns reader
+normalization; no shared runtime/phase-file overlap. Verify valid checkpoint/tail,
+torn-tail recovery, late complete corruption and legacy rejection through actual
+GitHub normal/scalar/recovery/RF3 suites. No migration or old-store conversion is
+introduced; rollback restores prior binaries for matching stores. Extra startup
+validation cost requires actual recovery/performance evidence and cannot count
+as a speed improvement. All fault assertions and numeric budgets remain.
+
+R12G preserves the existing strict cold-term read contract: the exact stored
+ReplicaEntry must pass the same bounded replica inspection as ordinary stored
+entry reads before its term is observed. Pass the owning configuration's
+MaxAppendEntries through the term reader; preserve its scoped storage identity,
+cut cache, index/term checks and lookup counts. ClusterReplication owns the two
+reader/caller files; the existing genuine-file malformed nested-operation
+recovery test is the acceptance oracle. Unknown nested authority fields remain
+corruption; ordinary bounded persistence evolution does not weaken this check.
+The borrowed storage span is copied for inspection only on a cold miss because
+the profile inspector requires owned ReadOnlyMemory for its borrowed proxies;
+no proxy or storage buffer escapes the read gate. Warm cut observations still
+perform no point lookup. Retain this ownership cost in performance accounting.

src/KeyLoad.Abstractions/Features/InternalSerialization/NativePayloadSyntax.cs

@@ -7,6 +7,21 @@ namespace KeyLoad.Features.InternalSerialization;
 // Official native headers/scalars are inspected before generated decoding allocates values.
 internal static class NativePayloadSyntax
 {
+    // Orleans 10.3.1 Reader uses this distinct throw site for exhausted/truncated buffers.
+    // Other InvalidOperationException failures, including session/codec invariants, must escape.
+    internal static bool IsReaderBufferFailure(Exception exception)
+    {
+        if (exception is not InvalidOperationException)
+        {
+            return false;
+        }
+        var method = exception.TargetSite;
+        return exception.Message == "Insufficient data present in buffer."
+            && method?.Name == "ThrowInsufficientData"
+            && method.DeclaringType is { IsGenericType: true } declaring
+            && declaring.GetGenericTypeDefinition() == typeof(Reader<>);
+    }
+
     internal static void Validate(ReadOnlySpan<byte> bytes, SerializerSession session, Type? expectedRootType = null, bool exactRoot = false)
     {
         var reader = Reader.Create(bytes, session);

src/KeyLoad.Abstractions/Features/InternalSerialization/NativeSerialization.cs

@@ -155,5 +155,6 @@ internal static long Measure<T>(T value, SerializerSession session)
     private static bool IsMalformed(Exception exception)
         => exception is SerializerException
             or ArgumentException or IndexOutOfRangeException or OverflowException or InvalidCastException
-            or FormatException or EndOfStreamException or TypeLoadException;
+            or FormatException or EndOfStreamException or TypeLoadException
+            || NativePayloadSyntax.IsReaderBufferFailure(exception);
 }

src/KeyLoad.Replication/Features/ClusterReplication/DurableReplicaLog.cs

@@ -58,7 +58,7 @@ private long Term(long index)
         {
             throw Errors.Fail(ErrorCode.NotFound, ReplicaPersistence.MissingEntry);
         }
-        var result = ReplicaTermObservationReader.Read(store, index, state.Term, termObservation);
+        var result = ReplicaTermObservationReader.Read(store, index, state.Term, configuration.MaxAppendEntries, termObservation);
         termObservation = result.Observation;
         return result.Term;
     }

src/KeyLoad.Replication/Features/ClusterReplication/ReplicaInspectionFields.cs

@@ -1,20 +1,55 @@
 using System.Buffers;
+using System.Collections.Immutable;
+using Microsoft.Extensions.DependencyInjection;
 using Orleans.Serialization.Buffers;
 using Orleans.Serialization.Codecs;
+using Orleans.Serialization.Session;
 using Orleans.Serialization.WireProtocol;
 
 namespace KeyLoad.Replication;
 
 internal static class ReplicaInspectionFields
 {
+    // CodecProvider searches generic definitions, so closed profile codecs require explicit resolution.
+    // This finite private dispatch shares the original session and never admits arbitrary codecs.
+    private static readonly Dictionary<Type, Type> ProfileCodecs = new()
+    {
+        [typeof(ReadOnlyMemory<byte>)] = typeof(ReplicaBorrowedBytesCodec),
+        [typeof(ImmutableArray<ReplicaEntry>)] = typeof(ReplicaEntriesInspectionCodec),
+        [typeof(ImmutableArray<string>)] = typeof(ReplicaVotersInspectionCodec),
+        [typeof(ReplicaEntry[])] = typeof(ReplicaEntryArrayInspectionCodec),
+        [typeof(string[])] = typeof(ReplicaVoterArrayInspectionCodec),
+        [typeof(string)] = typeof(ReplicaBorrowedStringCodec),
+        [typeof(ReplicaSnapshot)] = typeof(ReplicaSnapshotInspectionCodec),
+        [typeof(ReplicatedOperation)] = typeof(ReplicaOperationInspectionCodec)
+    };
+    private static readonly HashSet<Type> Scalars =
+    [
+        typeof(int), typeof(long), typeof(Guid), typeof(DateTimeOffset),
+        typeof(OperationKind), typeof(ErrorCode), typeof(ErrorCode?)
+    ];
+
     internal static string Sender<TInput>(ref Reader<TInput> reader)
         => ReplicaInspectionBuffers.For(reader.Session).Sender(Read<string, TInput>(ref reader, 0));
 
     internal static T Read<T, TInput>(ref Reader<TInput> reader, uint delta)
     {
         var field = reader.ReadFieldHeader();
         Require(field, delta, ExpectedFieldType<T>());
-        return reader.Session.CodecProvider.GetCodec<T>().ReadValue(ref reader, field)!;
+        return ProfileCodec<T>(reader.Session).ReadValue(ref reader, field)!;
+    }
+
+    private static IFieldCodec<T> ProfileCodec<T>(SerializerSession session)
+    {
+        if (ProfileCodecs.TryGetValue(typeof(T), out var codec))
+        {
+            return (IFieldCodec<T>)session.CodecProvider.Services.GetRequiredService(codec);
+        }
+        if (!Scalars.Contains(typeof(T)))
+        {
+            throw Errors.Fail(ErrorCode.Corruption, ReplicaPersistence.InvalidEncoding);
+        }
+        return session.CodecProvider.GetCodec<T>();
     }
 
     // These two current ungenerated contracts use Orleans' backing-integer codec.
@@ -63,5 +98,5 @@ internal static void EndBase<TInput>(ref Reader<TInput> reader)
 
     internal static void Write<T, TBufferWriter>(ref Writer<TBufferWriter> writer, uint delta, T value)
         where TBufferWriter : IBufferWriter<byte>
-        => writer.Session.CodecProvider.GetCodec<T>().WriteField(ref writer, delta, typeof(T), value);
+        => ProfileCodec<T>(writer.Session).WriteField(ref writer, delta, typeof(T), value);
 }

src/KeyLoad.Replication/Features/ClusterReplication/ReplicaNativeInspection.cs

@@ -52,8 +52,13 @@ private static void Configure(ISerializerBuilder builder)
         Register<ReplicaBorrowedBytesCodec>(builder);
         Register<ReplicaEntryInspectionCodec>(builder);
         Register<ReplicaOperationInspectionCodec>(builder);
-        Register<ReplicaEntryArrayInspectionCodec>(builder);
-        Register<ReplicaVoterArrayInspectionCodec>(builder);
+        builder.Services.AddSingleton(provider => new ReplicaEntryArrayInspectionCodec(provider.GetRequiredService<ReplicaEntryInspectionCodec>()));
+        builder.Services.AddSingleton(provider => new ReplicaVoterArrayInspectionCodec(provider.GetRequiredService<ReplicaBorrowedStringCodec>()));
+        builder.Configure(options =>
+        {
+            options.FieldCodecs.Add(typeof(ReplicaEntryArrayInspectionCodec));
+            options.FieldCodecs.Add(typeof(ReplicaVoterArrayInspectionCodec));
+        });
         Register<ReplicaEntriesInspectionCodec>(builder);
         Register<ReplicaVotersInspectionCodec>(builder);
         Register<ReplicaEntryBatchInspectionCodec>(builder);

src/KeyLoad.Replication/Features/ClusterReplication/ReplicaTermObservation.cs

@@ -32,7 +32,7 @@ internal readonly struct ReplicaTermObservationRead(long term, ReplicaTermObserv
 
 internal static class ReplicaTermObservationReader
 {
-    internal static ReplicaTermObservationRead Read(IAtomicStore store, long index, long hardStateTerm,
+    internal static ReplicaTermObservationRead Read(IAtomicStore store, long index, long hardStateTerm, int maximumEntries,
         ReplicaTermObservation? current)
     {
         return store.Read<ReplicaTermObservationRead>(view =>
@@ -55,7 +55,7 @@ internal static ReplicaTermObservationRead Read(IAtomicStore store, long index,
             long term = 0;
             var found = view.ReadValue(ReplicaProtocol.EntryStorageKey(index), bytes =>
             {
-                var entry = ReplicaProtocolCodec.Deserialize<ReplicaEntry>(bytes);
+                var entry = ReplicaProtocolCodec.DeserializeStored<ReplicaEntry>(bytes.ToArray(), maximumEntries);
                 if (entry.Index != index || entry.Term <= 0 || entry.Term > hardStateTerm)
                 {
                     throw Errors.Fail(ErrorCode.Corruption, ReplicaProtocol.CorruptLog);

src/KeyLoad.Storage.ZoneTree/Features/StorageRecovery/ZoneTreeJournalPreflight.cs

@@ -0,0 +1,69 @@
+using System.Buffers.Binary;
+using System.Security.Cryptography;
+using static KeyLoad.Storage.ZoneTree.ZoneTreePersistenceFormat;
+
+namespace KeyLoad.Storage.ZoneTree;
+
+// Verify the whole recoverable prefix before opening a provider which can rewrite its metadata.
+// The caller holds the owner lock and retains this same journal handle through ordered recovery.
+internal static class ZoneTreeJournalPreflight
+{
+    internal static void Validate(FileStream journal, ZoneTreeStoreOptions options, int identityVersion)
+    {
+        try
+        {
+            journal.Position = 0;
+            var header = new byte[HeaderLength];
+            var position = ReadCheckpoint(journal, options, header);
+            while (journal.Length - journal.Position >= HeaderLength)
+            {
+                if (!ReadFrame(journal, options, identityVersion, header, ref position))
+                {
+                    break;
+                }
+            }
+        }
+        finally
+        {
+            // Only ordinary recovery applies verified records and truncates an incomplete current tail.
+            journal.Position = 0;
+        }
+    }
+
+    private static long ReadCheckpoint(FileStream journal, ZoneTreeStoreOptions options, byte[] header)
+    {
+        if (journal.Length < HeaderLength)
+        {
+            return 0;
+        }
+        journal.ReadExactly(header);
+        journal.Position = 0;
+        if (BinaryPrimitives.ReadUInt64LittleEndian(header) != CheckpointMagic)
+        {
+            return 0;
+        }
+        // Normal recovery also supplies an apply callback. A no-op preserves those same bounds;
+        // the complete journal may exceed the standalone snapshot limit because it contains a tail.
+        return ZoneTreeCheckpointReader.Read(journal, options, static _ => { }).Position;
+    }
+
+    private static bool ReadFrame(FileStream journal, ZoneTreeStoreOptions options, int identityVersion,
+        byte[] header, ref long position)
+    {
+        journal.ReadExactly(header);
+        var (length, sequence) = ZoneTreeJournalRecovery.ValidateHeader(header, position, options.MaxFrameBytes, identityVersion);
+        if (journal.Length - journal.Position < length)
+        {
+            return false;
+        }
+        var payload = new byte[length];
+        journal.ReadExactly(payload);
+        if (!CryptographicOperations.FixedTimeEquals(SHA256.HashData(payload), header.AsSpan(ChecksumOffset, ChecksumLength)))
+        {
+            throw Errors.Fail(ErrorCode.Corruption, JournalChecksumInvalid);
+        }
+        _ = ZoneTreeJournalCodec.Deserialize(payload);
+        position = sequence;
+        return true;
+    }
+}

src/KeyLoad.Storage.ZoneTree/Features/StorageRecovery/ZoneTreeStoreInitializer.cs

@@ -12,8 +12,9 @@ internal static void Open(ZoneTreeStoreRuntime runtime)
             runtime.Ownership = new FileStream(Path.Combine(runtime.Options.Directory, OwnerLockFileName),
                 FileMode.OpenOrCreate, FileAccess.ReadWrite, FileShare.None);
             runtime.Identity = ZoneTreeIdentityFile.Open(runtime.Options, runtime.Ownership);
-            runtime.Tree = ZoneTreeTreeFactory.Open(runtime.Options);
             runtime.Journal = ZoneTreeStoreFiles.OpenJournal(runtime.Options);
+            ZoneTreeJournalPreflight.Validate(runtime.Journal, runtime.Options, runtime.Identity.FormatVersion);
+            runtime.Tree = ZoneTreeTreeFactory.Open(runtime.Options);
             ZoneTreeJournalRecovery.Recover(runtime);
             runtime.Maintainer = runtime.Tree.CreateMaintainer();
             ZoneTreeCheckpointReclaimer.Reclaim(runtime.Options.Directory);
@@ -31,6 +32,7 @@ internal static void OpenExisting(ZoneTreeStoreRuntime runtime, Guid expectedNod
             FileMode.Open, FileAccess.ReadWrite, FileShare.None);
         runtime.Identity = ZoneTreeIdentityFile.OpenExisting(runtime.Options, expectedNodeId);
         runtime.Journal = ZoneTreeStoreFiles.OpenJournal(runtime.Options, FileMode.Open);
+        ZoneTreeJournalPreflight.Validate(runtime.Journal, runtime.Options, runtime.Identity.FormatVersion);
         runtime.Tree = ZoneTreeTreeFactory.Open(runtime.Options, requireExisting: true);
         ZoneTreeJournalRecovery.Recover(runtime);
     }

tests/KeyLoad.RecoveryTests/Features/ClusterReplication/ReplicaSnapshotMetadataRecoveryTests.cs

@@ -6,8 +6,6 @@ namespace KeyLoad.RecoveryTests;
 /// <summary>AC-REP-004: real private-transfer metadata failures preserve the canonical cut and permit safe retry.</summary>
 internal sealed class ReplicaSnapshotMetadataRecoveryTests
 {
-    private static readonly byte[] TruncatedManifest = "{"u8.ToArray();
-
     /// <summary>A foreign-incarnation descriptor is rejected before private files or canonical state are changed.</summary>
     [Test]
     public async Task ForeignIncarnationSnapshotIsRejectedWithoutEffectsAndValidDescriptorCanRetry()
@@ -42,7 +40,9 @@ public async Task TruncatedPrivateManifestIsDiscardedAndVerifiedTransferCanRetry
         var receiver = trial.Receiver();
         await Assert.That(receiver.Begin(trial.Image)).IsEqualTo(0);
         var directory = Path.GetDirectoryName(trial.TargetImagePath(trial.Image))!;
-        await File.WriteAllBytesAsync(Path.Combine(directory, ReplicaProtocol.IncomingManifest), TruncatedManifest,
+        var manifestPath = Path.Combine(directory, ReplicaProtocol.IncomingManifest);
+        var manifest = await File.ReadAllBytesAsync(manifestPath, TestContext.Current!.Execution.CancellationToken);
+        await File.WriteAllBytesAsync(manifestPath, manifest[..^1],
             TestContext.Current!.Execution.CancellationToken);
 
         receiver.Recover();