Skip to content

Commit 3d857e4

Browse files
KSemenenkoKSemenenko
committed
Reauthorize receive results before publication
Recheck current field and header policy for known applied queue and subscription receives after the last awaited publication barrier, returning a closed refusal without changing stored command outcomes. Remove the obsolete document outcome admission branch and retain fresh authorization before retry/outcome reads. Align existing credential, ANN, canonical-cut and restore assertions with their original persisted commands, resource identities and exact lease timestamps. Verified: complete Debug solution build with -warnaserror (0 warnings/errors), full dotnet format verification, repository governance (22 projects, 6 modules), and native current-contract selection (296/296, no skips/timeouts; unchanged DLL/PDB). Whole-suite, recovery, Aspire and numeric coverage qualification remain open; original CI failures are retained for follow-up.
1 parent c6ecfcb commit 3d857e4

29 files changed

Lines changed: 197 additions & 267 deletions

File tree

‎src/KeyLoad.Clustering/ClusterRouting/Admission/NativeCommandWorkAdmission.cs‎

Lines changed: 4 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -6,19 +6,19 @@ namespace KeyLoad.Clustering;
66
/// <summary>Reuses Core admission at the actual command owner without another physical catalog.</summary>
77
internal static class NativeCommandWorkAdmission
88
{
9-
internal static void Bind(DatabaseEngine database, DecodedGrainRequest request, PrincipalRecord principal,
9+
internal static Command<DatabaseCommand>? Bind(DatabaseEngine database, DecodedGrainRequest request, PrincipalRecord principal,
1010
TimeProvider clock, NativeCapabilityWorkLifetime? work, Command<DatabaseCommand>? originalIntent = null)
1111
{
1212
if (work is null)
1313
{
14-
return; // Direct component execution has no hosted resource-lifetime binding.
14+
return null; // Direct component execution has no hosted resource-lifetime binding.
1515
}
1616
var kind = request.Envelope.CommandKind
1717
?? throw Errors.Fail(ErrorCode.TokenInvalidated, GrainRoutingProtocol.InvalidRequest);
1818
if (kind == OperationKind.DropTable || DatabaseEngine.IsResourceFreeCommandWork(kind))
1919
{
2020
work.Bind([]); // Dropping a resource cannot retain its closing lifetime.
21-
return;
21+
return null;
2222
}
2323
var operation = originalIntent ?? (CarriesVerifiedOperation(kind)
2424
? database.OwnCommand(GrainNativePayload.Read<Command<DatabaseCommand>>(request.Payload))
@@ -28,6 +28,7 @@ internal static void Bind(DatabaseEngine database, DecodedGrainRequest request,
2828
throw Errors.Fail(ErrorCode.TokenInvalidated, GrainRoutingProtocol.InvalidRequest);
2929
}
3030
database.ValidateResourceEffectSubmission(operation, clock.GetUtcNow(), work);
31+
return operation;
3132
}
3233

3334
internal static NativeCommandWorkProofV1 Capture(DecodedGrainRequest request,

‎src/KeyLoad.Clustering/ClusterRouting/Commands/GrainCommandExecutor.cs‎

Lines changed: 10 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -49,7 +49,7 @@ private async Task<GrainOperationReply> ExecuteCoreAsync(DecodedGrainRequest req
4949
?? throw Errors.Fail(ErrorCode.TokenInvalidated, GrainRoutingProtocol.InvalidRequest);
5050
await NativeCommandCapabilityAuthorization.ObserveAndValidateAsync(database, clock, settings,
5151
codec, request, GrainRequestPhase.BeforeSubmit, context, operationToken).ConfigureAwait(true);
52-
NativeCommandWorkAdmission.Bind(database, request, principal, clock, work);
52+
var original = NativeCommandWorkAdmission.Bind(database, request, principal, clock, work);
5353
var result = await SubmitCommandAsync(kind, request, principal, operationToken).ConfigureAwait(true);
5454
stage = GrainFailureStage.ReplyEncoding;
5555
await ObservePhaseAsync(request, GrainRequestPhase.SubmitReturned, context, operationToken).ConfigureAwait(true);
@@ -64,6 +64,15 @@ await NativeCommandCapabilityAuthorization.ObserveAndValidateAsync(database, clo
6464
};
6565
await NativeCommandResultWorkBinding.BindBeforePublicationAsync(database, coordinator, clock,
6666
request, reply, work, operationToken).ConfigureAwait(true);
67+
if (original is not null && kind is OperationKind.Receive or OperationKind.ReceiveSubscription)
68+
{
69+
var budget = new ReadExecutionBudget(database.OperationLimitsOptions, clock, operationToken);
70+
budget.ConstrainLifetime(request.Envelope.ExpiresAt);
71+
budget.AttachResourceLifetimeBinding(work);
72+
result = database.ResolveAppliedSubmissionOutcome(original, budget);
73+
var current = GrainReplyFactory.Operation(result: result, cancellationToken: operationToken, options: options);
74+
return current.Error is null ? current with { WorkProof = reply.WorkProof } : current;
75+
}
6776
return reply;
6877
}
6978
catch (Exception error) when (NativeCqrsBoundaryErrors.IsNonFatal(error))

‎src/KeyLoad.Core/ClusterRouting/Execution/NativeDocumentSubmissionAdmission.cs‎

Lines changed: 0 additions & 154 deletions
This file was deleted.

‎src/KeyLoad.Core/CommandOutcomes.cs‎

Lines changed: 5 additions & 33 deletions
Original file line numberDiff line numberDiff line change
@@ -38,47 +38,19 @@ internal Result<object> ResolveSubmissionOutcome(Command<DatabaseCommand> operat
3838
{
3939
ArgumentNullException.ThrowIfNull(operation);
4040
var normalized = OwnCommand(operation);
41-
return Store.Read(view => ReadValidatedSubmissionOutcome(view, normalized, allowStoredDocumentDenial: true));
41+
return Store.Read(view => ReadValidatedSubmissionOutcome(view, normalized));
4242
}
4343

4444
private Result<object> ReadValidatedSubmissionOutcome(IKeyValueView view, Command<DatabaseCommand> operation,
45-
bool allowStoredDocumentDenial = false, CommandOutcomeSelection? verifiedSelection = null)
45+
CommandOutcomeSelection? verifiedSelection = null)
4646
{
4747
var principal = Principal(view, operation.UserId!, Clock.GetUtcNow());
48-
KeyLoadException? denied = null;
49-
try
50-
{
51-
AuthorizeOperation(view, principal, operation);
52-
}
53-
catch (KeyLoadException refusal) when (allowStoredDocumentDenial && refusal.Code == ErrorCode.PermissionDenied)
54-
{
55-
if (ReadOrdinaryDocumentSubmission(operation) is null)
56-
{
57-
throw;
58-
}
59-
denied = refusal;
60-
}
48+
AuthorizeOperation(view, principal, operation);
6149
var scope = CommandOutcomePartitionIdentity.Resolve(operation);
6250
var selection = verifiedSelection ?? CommandOutcomeKeyResolver.Select(view, operation.UserId!, operation.CommandId, scope);
63-
var outcome = selection.Outcome;
64-
if (outcome is null)
65-
{
66-
if (denied is not null)
67-
{
68-
System.Runtime.ExceptionServices.ExceptionDispatchInfo.Capture(denied).Throw();
69-
}
70-
throw Errors.Fail(ErrorCode.RecoveryRequired, MissingDurableOutcomeMessage);
71-
}
51+
var outcome = selection.Outcome
52+
?? throw Errors.Fail(ErrorCode.RecoveryRequired, MissingDurableOutcomeMessage);
7253
ValidateSubmissionOutcomeIdentity(view, principal, operation, selection, outcome, verifiedSelection.HasValue);
73-
if (denied is not null && (Errors.Code(outcome.Result.Problem) != denied.Code
74-
|| !string.Equals(outcome.Result.Problem?.Detail, denied.Message, StringComparison.Ordinal)
75-
|| outcome.Result.Value is not null
76-
|| outcome.BlobAuthority is not null || outcome.CompositionAuthority is not null
77-
|| outcome.OnlineTextAuthority is not null || outcome.RemoteTransferFailureAuthority is not null
78-
))
79-
{
80-
System.Runtime.ExceptionServices.ExceptionDispatchInfo.Capture(denied).Throw();
81-
}
8254
ValidateCachedResult(view, principal, EvaluateCommand(operation, Clock.GetUtcNow()), outcome);
8355
return outcome.Result;
8456
}

‎src/KeyLoad.Core/Messaging/Commands/QueueRetries/QueueRetryAuthorization.cs‎

Lines changed: 26 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,26 @@
1+
using KeyLoad.Storage;
2+
using ManagedCode.Communication.Commands;
3+
4+
namespace KeyLoad.Core;
5+
6+
public sealed partial class DatabaseEngine
7+
{
8+
private bool TryAuthorizeQueueRetry(IKeyValueView view, Command<DatabaseCommand> operation,
9+
out PrincipalRecord? principal, bool preserveBudgetFailure)
10+
{
11+
principal = null;
12+
try
13+
{
14+
principal = Principal(view, operation.UserId!, Clock.GetUtcNow());
15+
AuthorizeOperation(view, principal, operation);
16+
return true;
17+
}
18+
catch (KeyLoadException failure) when (failure.Code is not (ErrorCode.Corruption or ErrorCode.FormatUnsupported
19+
or ErrorCode.RecoveryRequired or ErrorCode.UnknownWriteOutcome)
20+
&& (!preserveBudgetFailure || failure.Code != ErrorCode.BudgetExceeded))
21+
{
22+
principal = null;
23+
return false;
24+
}
25+
}
26+
}

‎src/KeyLoad.Core/Messaging/Commands/QueueRetries/QueueRetryPreparation.cs‎

Lines changed: 5 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -41,18 +41,20 @@ private Command<DatabaseCommand> PrepareQueueRetryInView(IKeyValueView view, Com
4141
return operation;
4242
}
4343
}
44+
if (!TryAuthorizeQueueRetry(view, operation, out var principal, preserveBudgetFailure) || principal is null)
45+
{
46+
return operation;
47+
}
4448
var scope = CommandOutcomePartitionIdentity.Resolve(operation);
4549
var selection = CommandOutcomeKeyResolver.Select(view, operation.UserId!, operation.CommandId, scope);
4650
if (selection.Outcome is { } outcome)
4751
{
48-
ValidateRetainedQueueRetryAdmission(view, operation, selection, outcome, preserveBudgetFailure);
52+
ValidateRetainedQueueRetryAdmission(view, principal, operation, selection, outcome);
4953
return operation;
5054
}
5155
List<QueueRetryInput> inputs;
5256
try
5357
{
54-
var principal = Principal(view, operation.UserId!, Clock.GetUtcNow());
55-
_ = AuthorizeOperation(view, principal, operation);
5658
inputs = ReadQueueRetryInputs(view, principal, operation);
5759
}
5860
catch (KeyLoadException failure) when (failure.Code is not (ErrorCode.Corruption or ErrorCode.FormatUnsupported

‎src/KeyLoad.Core/Messaging/Execution/DatabaseEngine.cs‎

Lines changed: 3 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -6,16 +6,11 @@ namespace KeyLoad.Core;
66

77
public sealed partial class DatabaseEngine
88
{
9-
private void ValidateRetainedQueueRetryAdmission(IKeyValueView view, Command<DatabaseCommand> operation,
10-
CommandOutcomeSelection selection, StoredOutcome outcome, bool preserveBudgetFailure = false)
9+
private void ValidateRetainedQueueRetryAdmission(IKeyValueView view, PrincipalRecord principal,
10+
Command<DatabaseCommand> operation, CommandOutcomeSelection selection, StoredOutcome outcome)
1111
{
1212
if (outcome.Incarnation != Store.Identity.Incarnation
13-
|| outcome.Fingerprint == CommandFingerprint(operation))
14-
{
15-
return;
16-
}
17-
if (!TryAuthorizeOrdinaryDocumentSubmission(view, operation, out var principal, out _, preserveBudgetFailure)
18-
|| principal is null || outcome.PolicyEpoch != principal.PolicyEpoch)
13+
|| outcome.Fingerprint == CommandFingerprint(operation) || outcome.PolicyEpoch != principal.PolicyEpoch)
1914
{
2015
return;
2116
}

‎src/KeyLoad.Core/ResourceExecution/Execution/DatabaseEngine.cs‎

Lines changed: 15 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -9,29 +9,35 @@ namespace KeyLoad.Core;
99

1010
public sealed partial class DatabaseEngine
1111
{
12-
internal Result<object> ResolveAppliedSubmissionOutcome(Command<DatabaseCommand> operation)
12+
internal Result<object> ResolveAppliedSubmissionOutcome(Command<DatabaseCommand> operation, ReadExecutionBudget? work = null)
1313
{
1414
ArgumentNullException.ThrowIfNull(operation);
15+
work?.Check();
1516
var normalized = OwnCommand(operation);
16-
return Store.Read(view =>
17+
var result = Store.Read(raw =>
1718
{
18-
var scope = CommandOutcomePartitionIdentity.Resolve(normalized);
19-
var selection = CommandOutcomeKeyResolver.Select(view, normalized.UserId!, normalized.CommandId, scope);
20-
var outcome = selection.Outcome
21-
?? throw Errors.Fail(ErrorCode.RecoveryRequired, MissingDurableOutcomeMessage);
19+
var view = work is null ? raw : work.CreateView(raw);
2220
try
2321
{
22+
var principal = Principal(view, normalized.UserId!, Clock.GetUtcNow());
23+
AuthorizeOperation(view, principal, normalized);
24+
var scope = CommandOutcomePartitionIdentity.Resolve(normalized);
25+
var selection = CommandOutcomeKeyResolver.Select(view, normalized.UserId!, normalized.CommandId, scope);
26+
var outcome = selection.Outcome
27+
?? throw Errors.Fail(ErrorCode.RecoveryRequired, MissingDurableOutcomeMessage);
2428
ValidateStoredSubmissionIdentity(view, normalized, selection, outcome);
25-
return ReadValidatedSubmissionOutcome(view, normalized,
26-
allowStoredDocumentDenial: true, verifiedSelection: selection);
29+
return ReadValidatedSubmissionOutcome(view, normalized, verifiedSelection: selection);
2730
}
2831
catch (KeyLoadException refusal) when (CqrsRuntimeFailures.FindFatal(refusal) is null
2932
&& refusal.Code is ErrorCode.PermissionDenied or ErrorCode.TokenInvalidated
30-
or ErrorCode.StaleLease or ErrorCode.Conflict)
33+
or ErrorCode.StaleLease or ErrorCode.LeaseExpired or ErrorCode.Conflict)
3134
{
3235
return ManagedCode.Communication.Result<object>.Fail(Errors.Problem(refusal.Code, refusal.Message));
3336
}
3437
});
38+
work?.CheckResult(result.Value ?? result);
39+
work?.Check();
40+
return result;
3541
}
3642

3743
private void ValidateStoredSubmissionIdentity(IKeyValueView view, Command<DatabaseCommand> operation,

‎src/KeyLoad.Core/ResourceExecution/Queries/DatabaseEngine/DatabaseEngine.RetainedSubmissionOutcome.cs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -33,7 +33,7 @@ public sealed partial class DatabaseEngine
3333
var view = work is null ? raw : work.CreateView(raw);
3434
try
3535
{
36-
return ReadValidatedSubmissionOutcome(view, normalized, allowStoredDocumentDenial: true);
36+
return ReadValidatedSubmissionOutcome(view, normalized);
3737
}
3838
catch (KeyLoadException missing) when (missing.Code == ErrorCode.RecoveryRequired
3939
&& string.Equals(missing.Message, MissingDurableOutcomeMessage, StringComparison.Ordinal))

‎src/KeyLoad.Core/ResourceExecution/Queries/DatabaseEngine/DatabaseEngine.cs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -17,7 +17,7 @@ internal Result<object> ResolveSubmissionOutcome(Command<DatabaseCommand> operat
1717
return Store.Read(raw =>
1818
{
1919
var view = work.CreateView(raw);
20-
var result = ReadValidatedSubmissionOutcome(view, normalized, allowStoredDocumentDenial: true);
20+
var result = ReadValidatedSubmissionOutcome(view, normalized);
2121
ObserveConfiguredResourceOutcome(view, normalized, result, work);
2222
work.CheckResult(result.Value ?? result);
2323
work.Check();

0 commit comments

Comments
 (0)