kjh2064/KArtSell.Aegis
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: DEBT-018 - make outbox writes co-transactional with entity writes

Browse Source 
TradeExecution: TradeOutboxPublisher.PublishAsync replaced with
UpdateAndPublishAsync, which opens one connection/transaction, updates
trade status and writes the outbox message on it, then commits once.
Used by the 3 call sites that publish an event after a status update
(SubmitTradeHandler, PollTradeStatusHandler's FullyFilled branch,
ConfirmSettlementHandler).

PortfolioReconciliation: ReconcileTradeHandler now injects the
request-scoped IDbConnection (the same instance ReconciliationSql
already uses) instead of opening a second separate connection via
IDbConnectionFactory, begins one transaction shared by
ReconciliationEngine.ReconcileTradeAsync and the outbox writes, and
commits once. Required adding IDbTransaction-aware overloads of
GetHoldingAsync/UpsertHoldingAsync/InsertReconciliationLogAsync - the
read needed one too, since Npgsql throws if a command on a connection
with a pending transaction doesn't have it attached.

dotnet build KArtSell.sln -c Release: clean. Integration tests: 17
pure-logic tests pass, 13 DB-backed tests fail with the pre-existing
connection-refused error (no SSH tunnel in this environment) - the
transactional changes themselves are not yet verified against a live
database. Also corrected two WBS_PROGRESS_TRACKER.csv rows that
inaccurately said frontend UI work was "in progress" when the
background agents building VS-28/VS-29 UI had actually failed
(hit the session's spend limit) before committing anything.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
김재현
2026-08-08 23:04:44 +09:00
parent 3c56c0926a
commit 8ed232e224
7 changed files with 164 additions and 88 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/KArtSell.Modules.ModelOperations/TradeExecution/TradeHandlers.cs
+29 -16
View File
@@ -57,9 +57,13 @@ public class SubmitTradeHandler
);
trade.MarkSubmitted(orderId, response);
await _sql.UpdateTradeStatusAsync(trade, response, null, ct);
await PublishEventAsync(
await TradeOutboxPublisher.UpdateAndPublishAsync(
_connectionFactory,
_sql,
_outbox,
_clock,
trade,
response,
"TradeSubmitted",
new TradeSubmittedEvent
{
@@ -87,9 +91,6 @@ public class SubmitTradeHandler
throw;
}
}
private async Task PublishEventAsync<T>(string eventType, T @event, Guid correlationId, CancellationToken ct) where T : class
=> await TradeOutboxPublisher.PublishAsync(_connectionFactory, _outbox, _clock, eventType, @event, correlationId, ct);
}
public class PollTradeStatusCommand
@@ -149,14 +150,15 @@ public class PollTradeStatusHandler
trade.MarkFilled(executedQty, unitPrice, response, _clock.UtcNow.UtcDateTime);
}
await _sql.UpdateTradeStatusAsync(trade, response, null, ct);
if (trade.Status is TradeStatus.FullyFilled)
{
await TradeOutboxPublisher.PublishAsync(
await TradeOutboxPublisher.UpdateAndPublishAsync(
_connectionFactory,
_sql,
_outbox,
_clock,
trade,
response,
"TradeFilled",
new TradeFilledEvent
{
@@ -168,6 +170,10 @@ public class PollTradeStatusHandler
command.CorrelationId,
ct);
}
else
{
await _sql.UpdateTradeStatusAsync(trade, response, null, ct);
}
_logger.LogInformation("Trade status updated: {TradeId} -> {Status}", trade.Id, status);
}
@@ -234,12 +240,13 @@ public class ConfirmSettlementHandler
if (success)
{
trade.MarkConfirmed(_clock.UtcNow.UtcDateTime, command.Commission);
await _sql.UpdateTradeStatusAsync(trade, response, null, ct);
await TradeOutboxPublisher.PublishAsync(
await TradeOutboxPublisher.UpdateAndPublishAsync(
_connectionFactory,
_sql,
_outbox,
_clock,
trade,
response,
"TradeSettled",
new TradeSettledEvent
{
@@ -287,16 +294,19 @@ public class TradeSettledEvent
}
/// <summary>
/// DEBT-TRADE-001: outbox write happens in its own transaction, separate from the
/// preceding trade status update (which owns its own connection in TradeSql). Not yet
/// atomic with the state transition. See TECH_DEBT_REGISTER.md.
/// DEBT-018 (fixed): the trade status update and the outbox event write now share one
/// connection/transaction, committed together, instead of the status update committing on
/// TradeSql's own connection and the outbox write committing separately afterward.
/// </summary>
internal static class TradeOutboxPublisher
{
public static async Task PublishAsync<T>(
public static async Task UpdateAndPublishAsync<T>(
IDbConnectionFactory connectionFactory,
ITradeSql sql,
IOutboxWriter outbox,
IClock clock,
Trade trade,
System.Text.Json.JsonElement? kisResponse,
string eventType,
T @event,
Guid correlationId,
@@ -314,7 +324,10 @@ internal static class TradeOutboxPublisher
await using var connection = await connectionFactory.OpenAsync(ct);
await using var transaction = await connection.BeginTransactionAsync(ct);
await sql.UpdateTradeStatusAsync(connection, transaction, trade, kisResponse, errorMessage: null, ct);
await outbox.AddAsync(connection, transaction, message, ct);
await transaction.CommitAsync(ct);
}
}
src/KArtSell.Modules.ModelOperations/TradeExecution/TradeSql.cs
+67 -41
View File
@@ -1,3 +1,4 @@
using System.Data.Common;
using System.Text.Json;
using Dapper;
using Microsoft.Extensions.Logging;
@@ -13,6 +14,13 @@ public interface ITradeSql
Task<IEnumerable<Trade>> GetTradesByDecisionIdAsync(Guid sellDecisionId, Guid correlationId, CancellationToken ct = default);
Task InsertTradeAsync(Trade trade, CancellationToken ct = default);
Task UpdateTradeStatusAsync(Trade trade, JsonElement? kisResponse, string? errorMessage, CancellationToken ct = default);
/// <summary>
/// DEBT-018: transaction-aware overload so the caller can write the outbox event on the same
/// connection/transaction as this status update, committing both atomically. See
/// TradeOutboxPublisher/SubmitTradeHandler.
/// </summary>
Task UpdateTradeStatusAsync(DbConnection connection, DbTransaction transaction, Trade trade, JsonElement? kisResponse, string? errorMessage, CancellationToken ct = default);
Task<int> CountTradesByStatusAsync(TradeStatus status, CancellationToken ct = default);
}
@@ -154,6 +162,47 @@ public class TradeSql : ITradeSql
_logger.LogInformation("Inserted trade {TradeId}", trade.Id);
}
private const string UpdateTradeStatusSql = """
INSERT INTO model_operations.trade_status_history
(id, trade_id, old_status, new_status, transitioned_at, kis_response, error_message, published_at, correlation_id)
SELECT @id, id, status, @newStatus, NOW(), @kisResponse::jsonb, @errorMessage, NOW(), @correlationId
FROM model_operations.trades
WHERE id = @tradeId;
UPDATE model_operations.trades
SET status = @newStatus,
kis_order_id = COALESCE(@kisOrderId, kis_order_id),
executed_quantity = COALESCE(@executedQuantity, executed_quantity),
unit_price = COALESCE(@unitPrice, unit_price),
total_amount = COALESCE(@totalAmount, total_amount),
commission = COALESCE(@commission, commission),
net_proceeds = COALESCE(@netProceeds, net_proceeds),
execution_timestamp = COALESCE(@executionTimestamp, execution_timestamp),
settlement_timestamp = COALESCE(@settlementTimestamp, settlement_timestamp),
kis_response = COALESCE(@kisResponse::jsonb, kis_response),
error_message = COALESCE(@errorMessage, error_message),
revision = revision + 1
WHERE id = @tradeId
""";
private static object BuildUpdateTradeStatusParams(Trade trade, JsonElement? kisResponse, string? errorMessage) => new
{
id = Guid.NewGuid(),
tradeId = trade.Id,
newStatus = trade.Status.ToString(),
kisOrderId = trade.KisOrderId,
executedQuantity = trade.ExecutedQuantity,
unitPrice = trade.UnitPrice,
totalAmount = trade.TotalAmount,
commission = trade.Commission,
netProceeds = trade.NetProceeds,
executionTimestamp = trade.ExecutionTimestamp,
settlementTimestamp = trade.SettlementTimestamp,
kisResponse = kisResponse?.ToString(),
errorMessage,
correlationId = trade.CorrelationId
};
public async Task UpdateTradeStatusAsync(
Trade trade,
JsonElement? kisResponse,
@@ -161,51 +210,28 @@ public class TradeSql : ITradeSql
CancellationToken ct = default)
{
using var connection = await _dataSource.OpenConnectionAsync(ct);
const string sql = """
INSERT INTO model_operations.trade_status_history
(id, trade_id, old_status, new_status, transitioned_at, kis_response, error_message, published_at, correlation_id)
SELECT @id, id, status, @newStatus, NOW(), @kisResponse::jsonb, @errorMessage, NOW(), @correlationId
FROM model_operations.trades
WHERE id = @tradeId;
UPDATE model_operations.trades
SET status = @newStatus,
kis_order_id = COALESCE(@kisOrderId, kis_order_id),
executed_quantity = COALESCE(@executedQuantity, executed_quantity),
unit_price = COALESCE(@unitPrice, unit_price),
total_amount = COALESCE(@totalAmount, total_amount),
commission = COALESCE(@commission, commission),
net_proceeds = COALESCE(@netProceeds, net_proceeds),
execution_timestamp = COALESCE(@executionTimestamp, execution_timestamp),
settlement_timestamp = COALESCE(@settlementTimestamp, settlement_timestamp),
kis_response = COALESCE(@kisResponse::jsonb, kis_response),
error_message = COALESCE(@errorMessage, error_message),
revision = revision + 1
WHERE id = @tradeId
""";
await connection.ExecuteAsync(sql, new
{
id = Guid.NewGuid(),
tradeId = trade.Id,
newStatus = trade.Status.ToString(),
kisOrderId = trade.KisOrderId,
executedQuantity = trade.ExecutedQuantity,
unitPrice = trade.UnitPrice,
totalAmount = trade.TotalAmount,
commission = trade.Commission,
netProceeds = trade.NetProceeds,
executionTimestamp = trade.ExecutionTimestamp,
settlementTimestamp = trade.SettlementTimestamp,
kisResponse = kisResponse?.ToString(),
errorMessage,
correlationId = trade.CorrelationId
});
await connection.ExecuteAsync(UpdateTradeStatusSql, BuildUpdateTradeStatusParams(trade, kisResponse, errorMessage));
_logger.LogInformation("Updated trade {TradeId} status to {Status}", trade.Id, trade.Status);
}
public async Task UpdateTradeStatusAsync(
DbConnection connection,
DbTransaction transaction,
Trade trade,
JsonElement? kisResponse,
string? errorMessage,
CancellationToken ct = default)
{
await connection.ExecuteAsync(new CommandDefinition(
UpdateTradeStatusSql,
BuildUpdateTradeStatusParams(trade, kisResponse, errorMessage),
transaction,
cancellationToken: ct));
_logger.LogInformation("Updated trade {TradeId} status to {Status} (transactional)", trade.Id, trade.Status);
}
public async Task<int> CountTradesByStatusAsync(TradeStatus status, CancellationToken ct = default)
{
using var connection = await _dataSource.OpenConnectionAsync(ct);