Preserve messaging JSON and repair authorization operation checks

Keep original queue and topic payload/header text through existing bounded JSON validation. Consume the published Communication 10.3.11 failure-serialization repair.

Keep complete lifetime-bound record checks, current policy CAS, native MCP 401 boundaries, original issuer replay, explicit scoped lifecycle data grants and verified cursor continuations. Strengthen existing exact-byte quota and cold-reopen controls without adding test cases.

Validation: full Debug build with warnaserror (0 warnings/errors), canonical format verification, governance, 8 focused topic/queue units and 4 whole sensitive-data cluster cases passed. Full module qualification remains open; the original two-of-three silo loss case conflicts with the selected membership quorum and awaits owner topology direction.

Author: KSemenenko

Directory.Packages.props

@@ -10,9 +10,9 @@
     <PackageVersion Include="ZoneTree" Version="2.0.2" />
     <PackageVersion Include="ZoneTree.Analytics" Version="1.0.2" />
     <PackageVersion Include="ZoneTree.FullTextSearch" Version="1.1.0" />
-    <PackageVersion Include="ManagedCode.Communication" Version="10.3.10" />
-    <PackageVersion Include="ManagedCode.Communication.AspNetCore" Version="10.3.10" />
-    <PackageVersion Include="ManagedCode.Communication.Orleans" Version="10.3.10" />
+    <PackageVersion Include="ManagedCode.Communication" Version="10.3.11" />
+    <PackageVersion Include="ManagedCode.Communication.AspNetCore" Version="10.3.11" />
+    <PackageVersion Include="ManagedCode.Communication.Orleans" Version="10.3.11" />
     <PackageVersion Include="ManagedCode.TimeSeries" Version="10.1.2" />
     <PackageVersion Include="ManagedCode.Storage.FileSystem" Version="10.1.5" />
     <PackageVersion Include="ManagedCode.Storage.Server" Version="10.1.5" />

src/KeyLoad.Core/Messaging/Commands/MessagingCommandsTopicPublishReads.cs

@@ -70,8 +70,8 @@ internal static TopicPublicationProgress AppendTopicEvents(this DatabaseEngine e
             var idKey = KeySpace.Partition(TopicEventIdKeySpace, source.Partition, source.Resource, source.Generation, item.EventId);
             var data = item with
             {
-                PayloadJson = JsonData.Validate(item.PayloadJson, engine.Limits),
-                HeadersJson = JsonData.Validate(item.HeadersJson, engine.Limits)
+                PayloadJson = JsonData.ValidateContent(item.PayloadJson, engine.Limits),
+                HeadersJson = JsonData.ValidateContent(item.HeadersJson, engine.Limits)
             };
             if (tx.ReadOwnedValue(EventStreamsCommandsTopicRetentionCommands.PurgedTopicIdentityKey(source, item.EventId)) is { } purgedBytes)
             {

src/KeyLoad.Core/Messaging/Execution/MessagingExecutionMessaging.cs

@@ -66,7 +66,7 @@ internal static MutationReceipt Enqueue(this DatabaseEngine engine, IAtomicTrans
             engine.Authorization.RequireFieldWrite(principal, resource with { FieldPolicies = resource.HeaderPolicies }, policy.Path);
         }
 
-        var body = new MessageBody(message.MessageId, JsonData.Validate(message.PayloadJson, engine.Limits), JsonData.Validate(message.HeadersJson, engine.Limits), message.OrderingKey, JsonData.Fingerprint(message));
+        var body = new MessageBody(message.MessageId, JsonData.ValidateContent(message.PayloadJson, engine.Limits), JsonData.ValidateContent(message.HeadersJson, engine.Limits), message.OrderingKey, JsonData.Fingerprint(message));
         PersistEnqueuedMessage(tx, lane, message, resource, body, now);
         return new(MessagingCommandsQueueSweepTransitions.EnqueueReceiptKind, message.Queue, message.MessageId, MessagingSingleElementCount);
     }

tests/KeyLoad.IntegrationTests/Authorization/Assertions/Catalog/DiscoveryAuthorityValues.cs

@@ -8,7 +8,7 @@ namespace KeyLoad.IntegrationTests.Features.Authorization;
 internal static class DiscoveryAuthorityValues
 {
     internal static DocumentResult Expected(DiscoveryAuthorityState state)
-        => new(state.Scenario.Reference, McpCallerProtocol.FirstRevision, McpDocumentProtocol.InitialJson, false, []);
+        => new(state.Scenario.StoredReference, McpCallerProtocol.FirstRevision, McpDocumentProtocol.InitialJson, false, []);
 
     internal static async Task<Guid> FirstAsync(McpOfficialClient session, DiscoveryAuthorityState state, CancellationToken token)
     {

tests/KeyLoad.IntegrationTests/Authorization/Assertions/Credentials/Bootstrap/ConcurrentPrincipalAuthorityAssertions.cs

@@ -41,5 +41,5 @@ internal static async Task FollowingAsync(McpOfficialClient firstClient, McpOffi
     }
 
     private static DocumentResult Expected(McpDocumentScenario scenario)
-        => new(scenario.Reference, McpCallerProtocol.FirstRevision, McpDocumentProtocol.InitialJson, false, []);
+        => new(scenario.StoredReference, McpCallerProtocol.FirstRevision, McpDocumentProtocol.InitialJson, false, []);
 }

tests/KeyLoad.IntegrationTests/Authorization/Assertions/Credentials/Bootstrap/InitialCredentialAuthorityValues.cs

@@ -7,7 +7,7 @@ namespace KeyLoad.IntegrationTests.Features.Authorization;
 internal static class InitialCredentialAuthorityValues
 {
     private static DocumentResult Expected(InitialCredentialAuthorityState state)
-        => new(state.Seed.Scenario.Reference, McpCallerProtocol.FirstRevision, McpDocumentProtocol.InitialJson, false, []);
+        => new(state.Seed.Scenario.StoredReference, McpCallerProtocol.FirstRevision, McpDocumentProtocol.InitialJson, false, []);
 
     internal static async Task HealthyAsync(global::KeyLoad.Client.KeyLoadClient root, global::KeyLoad.Client.KeyLoadClient reader, McpOfficialClient session,
         InitialCredentialAuthorityState state, CancellationToken token)

tests/KeyLoad.IntegrationTests/Authorization/Assertions/Credentials/Bootstrap/InitialPrincipalEpochClusterValues.cs

@@ -26,7 +26,7 @@ internal static async Task<string> ImageAsync(RequestCqrsClusterCallers root, In
         await ServerFailureObserver.ObserveAsync(async () =>
         {
             document = await McpCallerAssertions.SdkSuccessAsync(await root.Sdk.GetAsync(state.Reference, token));
-            await SqlClusterProtocol.EqualAsync(new DocumentReadResult(new(state.Reference, revision,
+            await SqlClusterProtocol.EqualAsync(new DocumentReadResult(new(state.StoredReference, revision,
                 revision == 1L ? "{\"value\":0}" : "{\"value\":7}", false, [])), document);
         }, failures);
         await ServerFailureObserver.ObserveAsync(async () =>
@@ -58,7 +58,7 @@ await ServerFailureObserver.ObserveAsync(async () =>
     internal static async Task DocumentAsync(RequestCqrsClusterCallers reader, InitialPrincipalEpochClusterState state,
         long revision, CancellationToken token)
     {
-        var expected = new DocumentReadResult(new(state.Reference, revision, revision == 1L ? "{\"value\":0}" : "{\"value\":7}", false, []));
+        var expected = new DocumentReadResult(new(state.StoredReference, revision, revision == 1L ? "{\"value\":0}" : "{\"value\":7}", false, []));
         await SqlClusterProtocol.EqualAsync(expected, await McpCallerAssertions.SdkSuccessAsync(await reader.Sdk.GetAsync(state.Reference, token)));
         await SqlClusterProtocol.EqualAsync(expected, (await McpCallerAssertions.SuccessAsync<DocumentReadResult>(
             await reader.Mcp.CallAsync(McpCallerTools.DocumentsGet, new GetDocumentRequest(state.Reference), token))).Value);

tests/KeyLoad.IntegrationTests/Authorization/Assertions/Credentials/Verification/CanceledCredentialAuthorityAssertions.cs

@@ -30,7 +30,7 @@ internal static async Task HealthyAsync(KeyLoadClient reader, McpOfficialClient
     }
 
     private static DocumentResult Expected(McpDocumentScenario scenario)
-        => new(scenario.Reference with { Lifetime = scenario.Lifetime }, McpCallerProtocol.FirstRevision,
+        => new(scenario.StoredReference, McpCallerProtocol.FirstRevision,
             McpDocumentProtocol.InitialJson, false, []);
 }
 

tests/KeyLoad.IntegrationTests/Authorization/Assertions/Eventing/EventingRequiredClaimClusterCurrent.cs

@@ -11,13 +11,14 @@ internal static class EventingRequiredClaimClusterCurrent
     internal static async Task<string> ImageAsync(RequestCqrsClusterCallers root, EventingRequiredClaimClusterState state, CancellationToken token)
     {
         var input = state.Input;
-        var expected = state.Completed ? RequiredProcessingPolicyClusterValues.Expected(input, "first", 1, false)
-            : new MessageInspection(new("first", MessageState.Ready, 0, 1L, 1L, null, null), input.Payload, input.Headers);
+        var expected = state.Claimed is null
+            ? new MessageInspection(new("first", MessageState.Ready, 0, 1L, 1L, null, null), input.Payload, input.Headers)
+            : RequiredProcessingPolicyClusterValues.Expected(input, "first", state.Completed ? 1 : 0, false);
         var message = await McpCallerAssertions.SdkSuccessAsync(await root.Sdk.InspectAsync(new(input.Lane, "first"), token));
         await SqlClusterProtocol.EqualAsync(new MessageInspectionResult(expected), message);
         var document = await McpCallerAssertions.SdkSuccessAsync(await root.Sdk.GetAsync(input.Reference("first"), token));
         await SqlClusterProtocol.EqualAsync(new DocumentReadResult(state.Completed
-            ? new(input.Reference("first"), 1L, RequiredProcessingPolicyClusterState.EffectJson("first"), false, []) : null), document);
+            ? new(input.StoredReference("first"), 1L, RequiredProcessingPolicyClusterState.EffectJson("first"), false, []) : null), document);
         var outbox = await McpCallerAssertions.SdkSuccessAsync(await root.Sdk.OutboxStatusAsync(input.Partition, token));
         var count = state.Completed ? 2L : 1L;
         await Assert.That(outbox.Head.Tail).IsEqualTo(count);

tests/KeyLoad.IntegrationTests/Authorization/Assertions/Eventing/EventingRequiredClaimClusterReceive.cs

@@ -0,0 +1,33 @@
+using KeyLoad.IntegrationTests.Features.ClientApi;
+using KeyLoad.IntegrationTests.Features.Messaging;
+using KeyLoad.IntegrationTests.Features.OrleansExecution;
+using KeyLoad.IntegrationTests.Features.QueryExecution;
+
+namespace KeyLoad.IntegrationTests.Features.Authorization;
+
+internal static class EventingRequiredClaimClusterReceive
+{
+    internal static Task AcceptProbeAsync(RequestCqrsClusterCallers root, RequestCqrsClusterCallers reader,
+        EventingRequiredClaimClusterState state, CancellationToken token)
+        => EventingRequiredClaimClusterCurrent.ObserveAsync(root, state, async () =>
+        {
+            var actual = await McpCallerAssertions.SdkSuccessAsync(await reader.Sdk.ReceiveAsync(state.EpochProbe, token));
+            await Assert.That(actual.RequestId).IsEqualTo(state.EpochProbe.RequestId);
+            await Assert.That(actual.Deliveries).IsEmpty();
+            await Assert.That(actual.Token.Incarnation).IsEqualTo(state.Input.Producers[0].Receipt.Token.Incarnation);
+            await Assert.That(actual.Token.AtomicPartitionId).IsEqualTo(state.Input.Partition.AtomicPartitionId);
+            await Assert.That(actual.Token.Position).IsGreaterThan(state.Input.Producers[0].Receipt.Token.Position);
+            state.EpochReceipt = actual;
+        }, token);
+
+    internal static async Task ReplayAsync(RequestCqrsClusterCallers root, RequestCqrsClusterCallers reader,
+        EventingRequiredClaimClusterState state, ReceiveRequest request, ReceiveResult expected, CancellationToken token)
+    {
+        for (var route = 0; route < 4; route++)
+        {
+            await EventingRequiredClaimClusterCurrent.ObserveAsync(root, state, async () =>
+                await SqlClusterProtocol.EqualAsync(expected, await QueueLifecyclePublicRoutes.CallAsync(reader, route,
+                    state.Input.Partition, McpCallerTools.MessagesReceive, request, () => reader.Sdk.ReceiveAsync(request, token), token)), token);
+        }
+    }
+}