Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 9 additions & 6 deletions backend/src/B3.Trading.Api/AlgoEndpoints.cs
Original file line number Diff line number Diff line change
Expand Up @@ -59,12 +59,14 @@ public static IEndpointRouteBuilder MapAlgo(this IEndpointRouteBuilder app)
EventDispatcher dispatcher,
DrainState drain,
IOutboundRecoveryGate recovery,
IOutboundCommandProtector protector,
SymbolDirectory symbols,
IAlgoSignalQueue signals,
ITickSizeProvider tickSizes) =>
{
var firm = ResolveFirm(ctx);
if (!recovery.IsBusinessIngressOpen(firm))
var owner = ResolveOwner(ctx, registry);
if (!recovery.IsBusinessIngressOpen(firm, protector.CreateStableEndClientRefCandidates(firm, owner.Value)))
return RecoveryUnavailable();
if (drain.IsDraining)
{
Expand Down Expand Up @@ -304,7 +306,6 @@ public static IEndpointRouteBuilder MapAlgo(this IEndpointRouteBuilder app)
default:
return Results.BadRequest(new { error = $"unsupported algo type '{type}'" });
}
var owner = ResolveOwner(ctx, registry);
var algoId = algoIds.Generate(firm);
var createdAt = DateTimeOffset.UtcNow;
var algo = new Algo(algoId, owner, firm, req.Symbol, securityId,
Expand Down Expand Up @@ -396,10 +397,12 @@ public static IEndpointRouteBuilder MapAlgo(this IEndpointRouteBuilder app)
AlgoBook algos,
DrainState drain,
IOutboundRecoveryGate recovery,
IOutboundCommandProtector protector,
IAlgoSignalQueue signals) =>
{
var firm = ResolveFirm(ctx);
if (!recovery.IsBusinessIngressOpen(firm))
var owner = ResolveOwner(ctx, registry);
if (!recovery.IsBusinessIngressOpen(firm, protector.CreateStableEndClientRefCandidates(firm, owner.Value)))
return RecoveryUnavailable();
if (drain.IsDraining)
{
Expand All @@ -418,7 +421,6 @@ public static IEndpointRouteBuilder MapAlgo(this IEndpointRouteBuilder app)
return Results.BadRequest(new { error = "at least one of newQuantity or newPrice must be set" });
if (req.NewQuantity is { } q && q <= 0)
return Results.BadRequest(new { error = "newQuantity must be positive" });
var owner = ResolveOwner(ctx, registry);
if (!algos.TryGet(firm, id, out var algo) || algo is null || algo.Owner != owner)
return Results.NotFound();
if (algo.IsTerminal)
Expand Down Expand Up @@ -479,10 +481,12 @@ public static IEndpointRouteBuilder MapAlgo(this IEndpointRouteBuilder app)
EventDispatcher dispatcher,
DrainState drain,
IOutboundRecoveryGate recovery,
IOutboundCommandProtector protector,
IAlgoSignalQueue signals) =>
{
var firm = ResolveFirm(ctx);
if (!recovery.IsBusinessIngressOpen(firm))
var owner = ResolveOwner(ctx, registry);
if (!recovery.IsBusinessIngressOpen(firm, protector.CreateStableEndClientRefCandidates(firm, owner.Value)))
return RecoveryUnavailable();
if (drain.IsDraining)
{
Expand All @@ -495,7 +499,6 @@ public static IEndpointRouteBuilder MapAlgo(this IEndpointRouteBuilder app)

if (!ulong.TryParse(algoId, out var id) || id == 0)
return Results.NotFound();
var owner = ResolveOwner(ctx, registry);
if (!algos.TryGet(firm, id, out var algo) || algo is null || algo.Owner != owner)
return Results.NotFound();

Expand Down
25 changes: 24 additions & 1 deletion backend/src/B3.Trading.Api/Lifecycle/DrainState.cs
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,27 @@ public sealed class DrainState : IDrainController
private string? _reason;

public bool IsDraining => Interlocked.Read(ref _draining) == 1;

/// <summary>
/// Narrower than <see cref="IsDraining"/>: true only when at least one
/// *non*-outbound-reconciliation drain reason is active (e.g.
/// <c>host_stopping</c>, a WAL/persistence fault). Outbound-reconciliation
/// reasons (<see cref="IsOutboundReconciliationReason"/>) mean specific
/// order mutations/end-clients require attention — not that the process
/// itself is unfit to receive traffic (see #780). Order-submission paths
/// should keep using <see cref="IsDraining"/> (a pending reconciliation
/// must still block new order mutations); only pod-routing probes
/// (<c>/ready</c>) should use this narrower signal.
/// </summary>
public bool BlocksPodRouting
{
get
{
lock (_gate)
return _reasons.Any(reason => !IsOutboundReconciliationReason(reason));
}
}

public string? Reason => Volatile.Read(ref _reason);
public TimeSpan Uptime => _uptime.Elapsed;
public DateTimeOffset StartedAt { get; } = DateTimeOffset.UtcNow;
Expand All @@ -52,7 +73,8 @@ public bool TryEndOutboundReconciliationDrain()
{
var removed =
_reasons.Remove("outbound_new_order_reconciliation_required") |
_reasons.Remove("outbound_cancel_replace_reconciliation_required");
_reasons.Remove("outbound_cancel_replace_reconciliation_required") |
_reasons.Remove("outbound_reconciliation_marker_recovered");
return TryEndDrainUnsafe(removed);
}
}
Expand Down Expand Up @@ -88,6 +110,7 @@ private static bool IsOutboundReconciliationReason(string reason) =>
reason is
"outbound_new_order_reconciliation_required" or
"outbound_cancel_replace_reconciliation_required" or
"outbound_reconciliation_marker_recovered" or
"cold_start_unresolved_lifecycle_intents";
}

Expand Down
15 changes: 9 additions & 6 deletions backend/src/B3.Trading.Api/OrdersEndpoints.cs
Original file line number Diff line number Diff line change
Expand Up @@ -41,12 +41,14 @@ public static IEndpointRouteBuilder MapOrders(this IEndpointRouteBuilder app)
SymbolDirectory symbols,
SubAccountsRegistry subAccounts,
IOutboundRecoveryGate recovery,
IOutboundCommandProtector protector,
ILoggerFactory loggerFactory,
CancellationToken ct) =>
{
var logger = loggerFactory.CreateLogger("B3.Trading.Api.OrdersEndpoints");
var firm = ResolveFirm(ctx);
if (!recovery.IsBusinessIngressOpen(firm))
var owner = ResolveOwner(ctx, registry);
if (!recovery.IsBusinessIngressOpen(firm, protector.CreateStableEndClientRefCandidates(firm, owner.Value)))
return RecoveryUnavailable();
if (!Enum.TryParse<OrderSide>(req.Side, ignoreCase: true, out var side))
return Results.BadRequest(new { error = $"invalid side '{req.Side}'" });
Expand Down Expand Up @@ -104,7 +106,6 @@ public static IEndpointRouteBuilder MapOrders(this IEndpointRouteBuilder app)
}
}

var owner = ResolveOwner(ctx, registry);
var idempotencyValues = ctx.Request.Headers["Idempotency-Key"];
if (idempotencyValues.Count > 1)
return Results.BadRequest(new { error = "multiple Idempotency-Key values are not allowed" });
Expand Down Expand Up @@ -300,17 +301,18 @@ await submitter.SubmitAsync(
OutboundMutationLedger outboundLedger,
WorkingOrderBook book,
IOutboundRecoveryGate recovery,
IOutboundCommandProtector protector,
ILoggerFactory loggerFactory,
CancellationToken ct) =>
{
var logger = loggerFactory.CreateLogger("B3.Trading.Api.OrdersEndpoints");
var firm = ResolveFirm(ctx);
if (!recovery.IsBusinessIngressOpen(firm))
var owner = ResolveOwner(ctx, registry);
if (!recovery.IsBusinessIngressOpen(firm, protector.CreateStableEndClientRefCandidates(firm, owner.Value)))
return RecoveryUnavailable();
if (!ulong.TryParse(clOrdId, out var clOrdIdU))
return Results.NotFound();

var owner = ResolveOwner(ctx, registry);
var idempotencyValues = ctx.Request.Headers["Idempotency-Key"];
if (idempotencyValues.Count > 1)
return Results.BadRequest(new { error = "multiple Idempotency-Key values are not allowed" });
Expand Down Expand Up @@ -422,17 +424,18 @@ await modifier.ModifyAsync(
OutboundMutationLedger outboundLedger,
WorkingOrderBook book,
IOutboundRecoveryGate recovery,
IOutboundCommandProtector protector,
ILoggerFactory loggerFactory,
CancellationToken ct) =>
{
var logger = loggerFactory.CreateLogger("B3.Trading.Api.OrdersEndpoints");
var firm = ResolveFirm(ctx);
if (!recovery.IsBusinessIngressOpen(firm))
var owner = ResolveOwner(ctx, registry);
if (!recovery.IsBusinessIngressOpen(firm, protector.CreateStableEndClientRefCandidates(firm, owner.Value)))
return RecoveryUnavailable();
if (!ulong.TryParse(clOrdId, out var clOrdIdU))
return Results.NotFound();

var owner = ResolveOwner(ctx, registry);
var idempotencyValues = ctx.Request.Headers["Idempotency-Key"];
if (idempotencyValues.Count > 1)
return Results.BadRequest(new { error = "multiple Idempotency-Key values are not allowed" });
Expand Down
9 changes: 7 additions & 2 deletions backend/src/B3.Trading.Application/OrderCancelService.cs
Original file line number Diff line number Diff line change
Expand Up @@ -41,6 +41,7 @@ public sealed class OrderCancelService
private readonly RestOrderIdempotencyStore? _restIdempotency;
private readonly TimeProvider _clock;
private readonly IOutboundRecoveryGate _outboundRecovery;
private readonly Outbound.IOutboundCommandProtector? _commandProtector;

public OrderCancelService(
ClOrdIdPrefixRegistry clOrdIds,
Expand All @@ -58,7 +59,8 @@ public OrderCancelService(
CancelReplaceOutboundCoordinator? outboundCoordinator = null,
RestOrderIdempotencyStore? restIdempotency = null,
TimeProvider? clock = null,
IOutboundRecoveryGate? outboundRecovery = null)
IOutboundRecoveryGate? outboundRecovery = null,
Outbound.IOutboundCommandProtector? commandProtector = null)
{
_clOrdIds = clOrdIds;
_ownership = ownership;
Expand All @@ -80,6 +82,7 @@ public OrderCancelService(
_restIdempotency = restIdempotency;
_clock = clock ?? TimeProvider.System;
_outboundRecovery = outboundRecovery ?? ImmediateOutboundRecoveryGate.Instance;
_commandProtector = commandProtector;
}

/// <summary>
Expand All @@ -101,7 +104,9 @@ public async Task<OrderCancelResult> CancelAsync(
AlgoOutboundOriginIdentity? algoOriginIdentity = null)
{
if (firmId is { } recoveryFirm
? !_outboundRecovery.IsBusinessIngressOpen(recoveryFirm)
? !_outboundRecovery.IsBusinessIngressOpen(
recoveryFirm,
_commandProtector?.CreateStableEndClientRefCandidates(recoveryFirm, owner.Value))
: !_outboundRecovery.IsReady)
{
return OrderCancelResult.ReconciliationRequired(
Expand Down
9 changes: 7 additions & 2 deletions backend/src/B3.Trading.Application/OrderModifyService.cs
Original file line number Diff line number Diff line change
Expand Up @@ -47,6 +47,7 @@ public sealed class OrderModifyService
private readonly RestOrderIdempotencyStore? _restIdempotency;
private readonly TimeProvider _clock;
private readonly IOutboundRecoveryGate _outboundRecovery;
private readonly IOutboundCommandProtector? _commandProtector;

public OrderModifyService(
ClOrdIdPrefixRegistry clOrdIds,
Expand All @@ -69,7 +70,8 @@ public OrderModifyService(
CancelReplaceOutboundCoordinator? outboundCoordinator = null,
RestOrderIdempotencyStore? restIdempotency = null,
TimeProvider? clock = null,
IOutboundRecoveryGate? outboundRecovery = null)
IOutboundRecoveryGate? outboundRecovery = null,
IOutboundCommandProtector? commandProtector = null)
{
_clOrdIds = clOrdIds;
_ownership = ownership;
Expand All @@ -96,6 +98,7 @@ public OrderModifyService(
_restIdempotency = restIdempotency;
_clock = clock ?? TimeProvider.System;
_outboundRecovery = outboundRecovery ?? ImmediateOutboundRecoveryGate.Instance;
_commandProtector = commandProtector;
}

/// <summary>
Expand Down Expand Up @@ -126,7 +129,9 @@ public OrderModifyService(
public async Task<OrderModifyResult> ModifyAsync(OrderModifyRequest req, CancellationToken ct)
{
if (req?.FirmId is { } recoveryFirm
? !_outboundRecovery.IsBusinessIngressOpen(recoveryFirm)
? !_outboundRecovery.IsBusinessIngressOpen(
recoveryFirm,
_commandProtector?.CreateStableEndClientRefCandidates(recoveryFirm, req.Owner.Value))
: !_outboundRecovery.IsReady)
{
return OrderModifyResult.Drained;
Expand Down
9 changes: 7 additions & 2 deletions backend/src/B3.Trading.Application/OrderSubmissionService.cs
Original file line number Diff line number Diff line change
Expand Up @@ -62,6 +62,7 @@ public sealed class OrderSubmissionService
private readonly TimeProvider _clock;
private readonly IOutboundRecoveryGate _outboundRecovery;
private readonly IOutboundSubmissionFaultInjector? _faultInjector;
private readonly IOutboundCommandProtector? _commandProtector;

public OrderSubmissionService(
ClOrdIdPrefixRegistry clOrdIds,
Expand All @@ -86,7 +87,8 @@ public OrderSubmissionService(
RestOrderIdempotencyStore? restIdempotency = null,
TimeProvider? clock = null,
IOutboundRecoveryGate? outboundRecovery = null,
IOutboundSubmissionFaultInjector? faultInjector = null)
IOutboundSubmissionFaultInjector? faultInjector = null,
IOutboundCommandProtector? commandProtector = null)
{
_clOrdIds = clOrdIds;
_ownership = ownership;
Expand All @@ -111,6 +113,7 @@ public OrderSubmissionService(
_clock = clock ?? TimeProvider.System;
_outboundRecovery = outboundRecovery ?? ImmediateOutboundRecoveryGate.Instance;
_faultInjector = faultInjector;
_commandProtector = commandProtector;
}

/// <summary>
Expand All @@ -123,7 +126,9 @@ public async Task<OrderSubmissionResult> SubmitAsync(OrderSubmissionRequest req,
{
if (req is null
? !_outboundRecovery.IsReady
: !_outboundRecovery.IsBusinessIngressOpen(req.FirmId))
: !_outboundRecovery.IsBusinessIngressOpen(
req.FirmId,
_commandProtector?.CreateStableEndClientRefCandidates(req.FirmId, req.Owner.Value)))
{
return OrderSubmissionResult.Drained;
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -92,6 +92,49 @@ static void Increment(Dictionary<string, int> counts, string firmId) =>
counts[firmId] = counts.GetValueOrDefault(firmId) + 1;
}

/// <summary>
/// Readiness-blocking counts scoped to <paramref name="firmId"/>, keyed
/// by end-client ref (#781 layer 2). Unmatched inbound evidence for the
/// firm cannot be attributed to a specific end-client, so it is bucketed
/// under <see cref="UnknownEndClientRef"/> — that bucket is intentionally
/// treated as blocking *every* end-client of the firm (we cannot rule any
/// of them out), unlike the cross-firm <see cref="UnknownFirmId"/> bucket
/// (#781 layer 3), which must never broadcast beyond its own
/// unattributed scope.
/// </summary>
public const string UnknownEndClientRef = "<unknown>";

public IReadOnlyDictionary<string, int> GetReadinessBlockingCountsByEndClient(string firmId)
{
ArgumentException.ThrowIfNullOrWhiteSpace(firmId);
lock (_gate)
{
var counts = new Dictionary<string, int>(StringComparer.Ordinal);
foreach (var mutation in _mutations.Values.Where(m =>
IsReadinessBlocking(m) && string.Equals(m.FirmId, firmId, StringComparison.Ordinal)))
{
Increment(
counts,
string.IsNullOrWhiteSpace(mutation.EndClientRef) ? UnknownEndClientRef : mutation.EndClientRef);
}
foreach (var evidence in _inboundEvidence.Values.Where(e =>
string.Equals(e.FirmId, firmId, StringComparison.Ordinal)
&& e.Disposition is not InboundVenueEvidenceDisposition.Matched
and not InboundVenueEvidenceDisposition.SupersededSession
&& (e.MatchedMutationIds.Count == 0
|| !e.MatchedMutationIds.Any(_mutations.ContainsKey))))
{
// Inbound evidence carries no end-client attribution — it can
// only be scoped to the firm it arrived for.
Increment(counts, UnknownEndClientRef);
}
return counts;
}

static void Increment(Dictionary<string, int> counts, string endClientRef) =>
counts[endClientRef] = counts.GetValueOrDefault(endClientRef) + 1;
}

public int InboundEvidenceCount
{
get { lock (_gate) return _inboundEvidence.Count; }
Expand Down
Loading
Loading