From 4672277ddc783876264c5a4501aacf80dd09a75b Mon Sep 17 00:00:00 2001 From: TheodorAdrienIsaak Mattli Date: Mon, 5 Oct 2026 11:50:40 +0200 Subject: [PATCH 1/2] Reserve payouts exactly and record chain transfers before broadcast Executor payouts zeroed the balance after the transfer, so earnings credited in between were lost; refunds and payouts held a SQL transaction across the chain call, and a transaction-wide refund wrote order states unconditionally and refunded credited orders. Give payouts and refunds one lifecycle: prepare the transfer (a chain read), commit the local decision together with a reserved chain transfer row carrying the digest, the signed transaction and its signature, broadcast the signed transaction under a two-minute bound, and record the outcome by a conditional write: sent, failed (a failed payout releases its amount in the same write) or unknown. Reconcile sent, unknown and stale reserved rows on the payout loop's ticker with one lookup: confirmed with the expected effects, failed, or resubmitted once from the identical stored bytes when the chain does not know the transaction; anything unverified stays unknown. A payout reserves exactly the balance it read and is refused while another is open; refunds change the transaction status and each outstanding order conditionally, with a refund settlement row, before the chain call. The maintenance pass settles chain refunds of failed runs through the same lifecycle and never selects an order with a refund transfer again. GitHub issues #80 and #81. --- docs/operations/configuration.md | 8 +- docs/operations/payments.md | 79 ++ .../database/queries/settlement.sql | 37 + .../database/queries/transactions.sql | 10 +- .../dispatcher/database/settlement.sql.go | 187 +++++ .../dispatcher/database/transactions.sql.go | 10 +- internal/dispatcher/payments/payments.go | 236 ++++-- internal/dispatcher/payments/payments_test.go | 79 +- internal/dispatcher/payments/payout.go | 67 +- internal/dispatcher/payments/payout_test.go | 135 ++- .../payments/refund_admission_test.go | 8 +- .../dispatcher/payments/settlement_test.go | 75 +- .../dispatcher/payments/sui/listener_test.go | 34 +- internal/dispatcher/payments/transfers.go | 250 ++++++ .../dispatcher/payments/transfers_test.go | 787 ++++++++++++++++++ internal/dispatcher/settlement_sweep.go | 5 +- .../transport/api/policy_bounds_test.go | 7 - 17 files changed, 1705 insertions(+), 309 deletions(-) create mode 100644 docs/operations/payments.md create mode 100644 internal/dispatcher/payments/transfers.go create mode 100644 internal/dispatcher/payments/transfers_test.go diff --git a/docs/operations/configuration.md b/docs/operations/configuration.md index 0e2c6b0d..6d5d4d6e 100644 --- a/docs/operations/configuration.md +++ b/docs/operations/configuration.md @@ -132,8 +132,12 @@ whether it was applied, a duplicate, a mismatch, expired or for an unknown intent), and `chain_transfers`, the outbound payout and refund transfers and their states, including each signed transaction and its signature, stored before broadcast. -Both are kept indefinitely for reconciliation; they stay empty while chain -payments are disabled. +A transfer row is `reserved` before the chain is called, then `sent`, +`confirmed`, `failed` (nothing moved; a failed payout's amount is back in the +balance) or `unknown` (the outcome is not known yet; reconciliation keeps +checking it). Both are kept indefinitely for reconciliation; they stay empty +while chain payments are disabled. See [chain payments](payments.md) for the +transfer lifecycle and what each state means for an operator. ## State and upgrades diff --git a/docs/operations/payments.md b/docs/operations/payments.md new file mode 100644 index 00000000..4867afb9 --- /dev/null +++ b/docs/operations/payments.md @@ -0,0 +1,79 @@ +# Chain payments + +This page describes how a dispatcher with chain payments enabled sends money: +executor payouts and buyer refunds. TEST payments move no funds and create none +of the records below. While `[sui] disabled = true`, nothing here runs. + +## Transfer lifecycle + +Every outbound transfer, payout or refund, follows the same steps and leaves +one row in the dispatcher's `chain_transfers` table. + +1. **Prepare.** The dispatcher selects coins and gas, builds and signs the + transaction and computes its digest. This only reads from the chain. If it + fails, nothing is recorded and nothing is sent. +2. **Reserve.** One database transaction records the local decision and the + transfer row in state `reserved`, with the digest, the signed transaction + and its signature. For a payout the decision is subtracting exactly the + amount being paid from the executor's balance; earnings credited later stay + in the balance for the next payout. For a refund it is marking the orders + (and, for a transaction refund, the transaction) refunded with one refund + settlement record per order. The signed transaction is stored before it is + ever submitted, and no database transaction is held open while the chain is + called. +3. **Submit** the stored transaction. +4. **Record** the outcome: + - `sent`: the chain accepted the transaction and reported success. + - `failed`: the transaction was not submitted, or the chain executed it + with a failure status. No coins moved. A failed payout returns its amount + to the executor's balance in the same database transaction. + - `unknown`: the transaction may have been submitted, but the response was + lost or incomplete. The row keeps the amount reserved until it is + resolved. + +A payout is not started while the same executor and currency has a `reserved`, +`sent` or `unknown` payout. A balance without a payout wallet is not paid out; +each payout pass logs how many such balances it skipped. + +A failed run's refund is attempted when the run ends. If that attempt never +reserved its transfer, for example because the dispatcher stopped first, the +periodic settlement pass sends it. The pass never takes an order that already +has a refund transfer, whatever that transfer's state, so one order is +refunded by at most one transfer; an uncertain outcome is resolved by +reconciliation below. While chain payments are disabled, the pass leaves +chain-currency orders as they are. + +## Reconciliation + +Once a minute the dispatcher looks up, by digest, every `sent` and `unknown` +transfer and every `reserved` transfer older than ten minutes (left by a stop +between the reservation and the submission): + +| Lookup result | New state | +| --- | --- | +| Executed successfully, and the receiver's credit matches the amount and coin | `confirmed` | +| Executed successfully, but the node returned no balance changes to check the credit against | `unknown` | +| Executed with failure status | `failed` (a payout's amount returns to the balance) | +| Not found, row `reserved` or `unknown` | the stored signed transaction is submitted again and the outcome recorded as above | +| Not found, row `sent` | `unknown` | +| Lookup error, or a credit that does not match | `unknown` | + +A resubmission sends the identical signed bytes, which have the same digest, +so the chain executes the transfer at most once. A transfer is never rebuilt +with new bytes because a response was lost. + +## What an operator sees + +The `detail` column of a `failed` or `unknown` row says why, and `updated_at` +is the time of the last check. + +- **`failed` payout**: nothing was paid and the amount is back in the + executor's balance; the next payout pass tries again with a new transfer. +- **`failed` refund**: the orders stay refunded and the amount is owed to the + buyer's refund address. It is not retried automatically; the row is the + record of what is owed. +- **`unknown`**: reconciliation keeps checking it every minute. A payout in + this state blocks further payouts of that executor and currency, so the + amount cannot be paid twice. If it stays `unknown`, for example because the + node never returns balance changes or the stored transaction can no longer + be executed, check the digest on a chain explorer before changing anything. diff --git a/internal/dispatcher/database/queries/settlement.sql b/internal/dispatcher/database/queries/settlement.sql index f50a1b6c..074a4399 100644 --- a/internal/dispatcher/database/queries/settlement.sql +++ b/internal/dispatcher/database/queries/settlement.sql @@ -15,3 +15,40 @@ ORDER BY observed_at, tx_digest, event_seq; UPDATE transactions SET status = sqlc.arg(paid) WHERE id = sqlc.arg(id) AND status = sqlc.arg(outstanding); + +-- name: ReservePayout :execrows +UPDATE earnings +SET current_balance = current_balance - sqlc.arg(amount) +WHERE earnings.executor_id = sqlc.arg(executor_id) AND earnings.currency = sqlc.arg(currency) + AND earnings.current_balance >= sqlc.arg(amount) + AND NOT EXISTS ( + SELECT 1 FROM chain_transfers t + WHERE t.kind = 'payout' AND t.executor_id = sqlc.arg(executor_id) AND t.currency = sqlc.arg(currency) + AND t.state IN ('reserved', 'sent', 'unknown') + ); + +-- name: ReleasePayout :exec +UPDATE earnings +SET current_balance = current_balance + sqlc.arg(amount) +WHERE executor_id = sqlc.arg(executor_id) AND currency = sqlc.arg(currency); + +-- name: InsertChainTransfer :one +INSERT INTO chain_transfers (kind, executor_id, transaction_id, order_id, amount, currency, receiver, state, digest, signed_transaction, signature, created_at, updated_at) +VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) +RETURNING id; + +-- name: UpdateChainTransfer :execrows +UPDATE chain_transfers +SET state = sqlc.arg(state), digest = sqlc.arg(digest), detail = sqlc.arg(detail), updated_at = sqlc.arg(updated_at) +WHERE id = sqlc.arg(id) AND state = sqlc.arg(from_state); + +-- name: ListUnresolvedChainTransfers :many +SELECT * FROM chain_transfers +WHERE state IN ('sent', 'unknown') OR (state = 'reserved' AND updated_at < sqlc.arg(reserved_before)) +ORDER BY updated_at, id +LIMIT sqlc.arg(row_limit); + +-- name: RefundPaidTransaction :execrows +UPDATE transactions +SET status = sqlc.arg(refunded) +WHERE id = sqlc.arg(id) AND status = sqlc.arg(paid); diff --git a/internal/dispatcher/database/queries/transactions.sql b/internal/dispatcher/database/queries/transactions.sql index 9b00e1ad..ea7374dd 100644 --- a/internal/dispatcher/database/queries/transactions.sql +++ b/internal/dispatcher/database/queries/transactions.sql @@ -75,13 +75,19 @@ SELECT * FROM order_settlements WHERE transaction_id = ? AND order_id = ?; -- name: ListPendingSettlements :many --- Only wholly local settlements: credits of any currency and TEST refunds. +-- An order covered by a refund transfer in any state is never listed: the +-- transfer's own reconciliation resolves it, so a pass cannot send it twice. SELECT sqlc.embed(d), e.exit_code, o.currency FROM debuglet_order o JOIN debuglets d ON d.id = o.debuglet_id AND d.transaction_id = o.transaction_id AND d.order_id = o.order_id JOIN measurement_execution e ON e.debuglet_id = d.id WHERE o.state = sqlc.arg(outstanding_state) AND d.state = sqlc.arg(exited_state) - AND e.exit_code IS NOT NULL AND (e.exit_code = 0 OR o.currency = 'TEST') + AND e.exit_code IS NOT NULL + AND NOT EXISTS ( + SELECT 1 FROM chain_transfers t + WHERE t.kind = 'refund' AND t.transaction_id = o.transaction_id + AND (t.order_id IS NULL OR t.order_id = o.order_id) + ) AND d.id > sqlc.arg(after_id) ORDER BY d.id LIMIT sqlc.arg(row_limit); diff --git a/internal/dispatcher/database/settlement.sql.go b/internal/dispatcher/database/settlement.sql.go index 9b101752..834b4cd0 100644 --- a/internal/dispatcher/database/settlement.sql.go +++ b/internal/dispatcher/database/settlement.sql.go @@ -40,6 +40,49 @@ func (q *Queries) GetPaymentReceipt(ctx context.Context, arg GetPaymentReceiptPa return i, err } +const insertChainTransfer = `-- name: InsertChainTransfer :one +INSERT INTO chain_transfers (kind, executor_id, transaction_id, order_id, amount, currency, receiver, state, digest, signed_transaction, signature, created_at, updated_at) +VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) +RETURNING id +` + +type InsertChainTransferParams struct { + Kind string + ExecutorID string + TransactionID string + OrderID sql.NullInt64 + Amount int64 + Currency string + Receiver string + State string + Digest string + SignedTransaction []byte + Signature string + CreatedAt models.UTCTime + UpdatedAt models.UTCTime +} + +func (q *Queries) InsertChainTransfer(ctx context.Context, arg InsertChainTransferParams) (int64, error) { + row := q.db.QueryRowContext(ctx, insertChainTransfer, + arg.Kind, + arg.ExecutorID, + arg.TransactionID, + arg.OrderID, + arg.Amount, + arg.Currency, + arg.Receiver, + arg.State, + arg.Digest, + arg.SignedTransaction, + arg.Signature, + arg.CreatedAt, + arg.UpdatedAt, + ) + var id int64 + err := row.Scan(&id) + return id, err +} + const insertPaymentReceipt = `-- name: InsertPaymentReceipt :exec INSERT INTO payment_receipts (tx_digest, event_seq, nonce, disposition, amount, coin_type, receiver, checkpoint, observed_at, detail) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) @@ -114,6 +157,57 @@ func (q *Queries) ListPaymentReceiptsByNonce(ctx context.Context, nonce string) return items, nil } +const listUnresolvedChainTransfers = `-- name: ListUnresolvedChainTransfers :many +SELECT id, kind, executor_id, transaction_id, order_id, amount, currency, receiver, state, digest, signed_transaction, signature, detail, created_at, updated_at FROM chain_transfers +WHERE state IN ('sent', 'unknown') OR (state = 'reserved' AND updated_at < ?1) +ORDER BY updated_at, id +LIMIT ?2 +` + +type ListUnresolvedChainTransfersParams struct { + ReservedBefore models.UTCTime + RowLimit int64 +} + +func (q *Queries) ListUnresolvedChainTransfers(ctx context.Context, arg ListUnresolvedChainTransfersParams) ([]ChainTransfer, error) { + rows, err := q.db.QueryContext(ctx, listUnresolvedChainTransfers, arg.ReservedBefore, arg.RowLimit) + if err != nil { + return nil, err + } + defer rows.Close() + var items []ChainTransfer + for rows.Next() { + var i ChainTransfer + if err := rows.Scan( + &i.ID, + &i.Kind, + &i.ExecutorID, + &i.TransactionID, + &i.OrderID, + &i.Amount, + &i.Currency, + &i.Receiver, + &i.State, + &i.Digest, + &i.SignedTransaction, + &i.Signature, + &i.Detail, + &i.CreatedAt, + &i.UpdatedAt, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Close(); err != nil { + return nil, err + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + const markTransactionPaid = `-- name: MarkTransactionPaid :execrows UPDATE transactions SET status = ?1 @@ -133,3 +227,96 @@ func (q *Queries) MarkTransactionPaid(ctx context.Context, arg MarkTransactionPa } return result.RowsAffected() } + +const refundPaidTransaction = `-- name: RefundPaidTransaction :execrows +UPDATE transactions +SET status = ?1 +WHERE id = ?2 AND status = ?3 +` + +type RefundPaidTransactionParams struct { + Refunded int64 + ID string + Paid int64 +} + +func (q *Queries) RefundPaidTransaction(ctx context.Context, arg RefundPaidTransactionParams) (int64, error) { + result, err := q.db.ExecContext(ctx, refundPaidTransaction, arg.Refunded, arg.ID, arg.Paid) + if err != nil { + return 0, err + } + return result.RowsAffected() +} + +const releasePayout = `-- name: ReleasePayout :exec +UPDATE earnings +SET current_balance = current_balance + ?1 +WHERE executor_id = ?2 AND currency = ?3 +` + +type ReleasePayoutParams struct { + Amount int64 + ExecutorID string + Currency string +} + +func (q *Queries) ReleasePayout(ctx context.Context, arg ReleasePayoutParams) error { + _, err := q.db.ExecContext(ctx, releasePayout, arg.Amount, arg.ExecutorID, arg.Currency) + return err +} + +const reservePayout = `-- name: ReservePayout :execrows +UPDATE earnings +SET current_balance = current_balance - ?1 +WHERE earnings.executor_id = ?2 AND earnings.currency = ?3 + AND earnings.current_balance >= ?1 + AND NOT EXISTS ( + SELECT 1 FROM chain_transfers t + WHERE t.kind = 'payout' AND t.executor_id = ?2 AND t.currency = ?3 + AND t.state IN ('reserved', 'sent', 'unknown') + ) +` + +type ReservePayoutParams struct { + Amount int64 + ExecutorID string + Currency string +} + +func (q *Queries) ReservePayout(ctx context.Context, arg ReservePayoutParams) (int64, error) { + result, err := q.db.ExecContext(ctx, reservePayout, arg.Amount, arg.ExecutorID, arg.Currency) + if err != nil { + return 0, err + } + return result.RowsAffected() +} + +const updateChainTransfer = `-- name: UpdateChainTransfer :execrows +UPDATE chain_transfers +SET state = ?1, digest = ?2, detail = ?3, updated_at = ?4 +WHERE id = ?5 AND state = ?6 +` + +type UpdateChainTransferParams struct { + State string + Digest string + Detail string + UpdatedAt models.UTCTime + ID int64 + FromState string +} + +func (q *Queries) UpdateChainTransfer(ctx context.Context, arg UpdateChainTransferParams) (int64, error) { + result, err := q.db.ExecContext(ctx, updateChainTransfer, + arg.State, + arg.Digest, + arg.Detail, + arg.UpdatedAt, + arg.ID, + arg.FromState, + ) + if err != nil { + return 0, err + } + return result.RowsAffected() +} diff --git a/internal/dispatcher/database/transactions.sql.go b/internal/dispatcher/database/transactions.sql.go index ad5fef9e..8f13f290 100644 --- a/internal/dispatcher/database/transactions.sql.go +++ b/internal/dispatcher/database/transactions.sql.go @@ -469,7 +469,12 @@ JOIN debuglets d ON d.id = o.debuglet_id AND d.transaction_id = o.transaction_id AND d.order_id = o.order_id JOIN measurement_execution e ON e.debuglet_id = d.id WHERE o.state = ?1 AND d.state = ?2 - AND e.exit_code IS NOT NULL AND (e.exit_code = 0 OR o.currency = 'TEST') + AND e.exit_code IS NOT NULL + AND NOT EXISTS ( + SELECT 1 FROM chain_transfers t + WHERE t.kind = 'refund' AND t.transaction_id = o.transaction_id + AND (t.order_id IS NULL OR t.order_id = o.order_id) + ) AND d.id > ?3 ORDER BY d.id LIMIT ?4 @@ -488,7 +493,8 @@ type ListPendingSettlementsRow struct { Currency string } -// Only wholly local settlements: credits of any currency and TEST refunds. +// An order covered by a refund transfer in any state is never listed: the +// transfer's own reconciliation resolves it, so a pass cannot send it twice. func (q *Queries) ListPendingSettlements(ctx context.Context, arg ListPendingSettlementsParams) ([]ListPendingSettlementsRow, error) { rows, err := q.db.QueryContext(ctx, listPendingSettlements, arg.OutstandingState, diff --git a/internal/dispatcher/payments/payments.go b/internal/dispatcher/payments/payments.go index 70166eeb..bb3d0733 100644 --- a/internal/dispatcher/payments/payments.go +++ b/internal/dispatcher/payments/payments.go @@ -283,11 +283,10 @@ func (p *PaymentHandler) RefundDebugletOrder(debuglet *database.Debuglet, refund // the decision in order_settlements in the same SQL transaction, together with // the earnings credit or the chain refund. The conditional state write is the // cheap guard against a second settlement, the settlement row's primary key the -// durable one; any failure rolls back all of it. A USDC refund is sent inside -// the SQL transaction, so a failure after the transfer may leave the order -// Outstanding although the money moved: chain refunds are therefore attempted -// once, from the terminal path, and are not retried automatically until a -// durable transfer record takes them over. +// durable one; any failure rolls back all of it. A USDC refund instead goes +// through the transfer lifecycle, which marks the order Refunded in the SQL +// transaction that reserves its transfer, before the chain is called; an +// uncertain outcome is then the transfer's and is resolved by reconciliation. func (p *PaymentHandler) settleOrder(ctx context.Context, debuglet *database.Debuglet, kind string, refundAddress string, chainOnly bool) error { tx, err := p.db.BeginTx(ctx, nil) if err != nil { @@ -315,14 +314,25 @@ func (p *PaymentHandler) settleOrder(ctx context.Context, debuglet *database.Deb } else if err := p.requireChain(order.Currency, "credit order"); err != nil { return err } + if kind == settlementRefund && order.Currency == "USDC" && order.State == int64(models.Outstanding) { + // The chain refund is not sent inside this SQL transaction: the + // transfer lifecycle commits the decision with the reserved transfer + // before it calls the chain. + tx.Rollback() + order.ExecutorID = debuglet.ExecutorID + if err := p.refundOnChain(ctx, order.TransactionID, []database.DebugletOrder{order}, nil); !errors.Is(err, errAlreadySettled) { + return err + } + // Settled concurrently: an order leaves Outstanding only as Credited + // or Refunded. + return nil + } - changed, err := queries.TransitionDebugletOrder(ctx, database.TransitionDebugletOrderParams{ - ToState: int64(to), TransactionID: key.TransactionID, OrderID: key.OrderID, FromState: int64(models.Outstanding), - }) + changed, err := recordSettlement(ctx, queries, order, debuglet.ExecutorID, kind) if err != nil { - return fmt.Errorf("failed to update state of order of %s: %w", debuglet.Uuid.String(), err) + return fmt.Errorf("failed to settle order of %s: %w", debuglet.Uuid.String(), err) } - if changed == 0 { + if !changed { // Only an outstanding order is settled. A credited order is already // included in the executor's earnings; a refunded order is not credited. current, err := queries.GetDebugletOrder(ctx, key) @@ -337,21 +347,8 @@ func (p *PaymentHandler) settleOrder(ctx context.Context, debuglet *database.Deb } } order.State = int64(to) - if err := queries.InsertOrderSettlement(ctx, database.InsertOrderSettlementParams{ - TransactionID: order.TransactionID, - OrderID: order.OrderID, - Kind: kind, - Amount: order.Price, - Currency: order.Currency, - ExecutorID: debuglet.ExecutorID, - DebugletID: order.DebugletID, - RecordedAt: models.NewUTCTime(time.Now()), - }); err != nil { - return fmt.Errorf("failed to record settlement of order of %s: %w", debuglet.Uuid.String(), err) - } - switch { - case kind == settlementCredit: + if kind == settlementCredit { if err := p.CreateEarningsIfNotExists(debuglet.ExecutorID, order.Currency, "", queries, ctx); err != nil { return fmt.Errorf("failed to create earnings of %s: %w", debuglet.ExecutorID, err) } @@ -363,21 +360,86 @@ func (p *PaymentHandler) settleOrder(ctx context.Context, debuglet *database.Deb return fmt.Errorf("failed to credit earnings of %s: %w", debuglet.ExecutorID, err) } p.logger.Debug("Credited Executor", zap.String("ID", debuglet.ExecutorID), zap.String("currency", order.Currency), zap.Int64("amount", order.Price)) - case order.Currency == "USDC": - if err := p.sui.RefundDebuglet(&order, refundAddress, ctx); err != nil { - return err - } } return tx.Commit() } +// recordSettlement moves an Outstanding order to Credited or Refunded and +// records the decision in order_settlements, through queries. It reports false +// and writes nothing when the order is no longer Outstanding. +func recordSettlement(ctx context.Context, queries *database.Queries, order database.DebugletOrder, executorID string, kind string) (bool, error) { + to := models.Credited + if kind == settlementRefund { + to = models.Refunded + } + changed, err := queries.TransitionDebugletOrder(ctx, database.TransitionDebugletOrderParams{ + ToState: int64(to), TransactionID: order.TransactionID, OrderID: order.OrderID, FromState: int64(models.Outstanding), + }) + if err != nil { + return false, fmt.Errorf("update order state: %w", err) + } + if changed == 0 { + return false, nil + } + if err := queries.InsertOrderSettlement(ctx, database.InsertOrderSettlementParams{ + TransactionID: order.TransactionID, + OrderID: order.OrderID, + Kind: kind, + Amount: order.Price, + Currency: order.Currency, + ExecutorID: executorID, + DebugletID: order.DebugletID, + RecordedAt: models.NewUTCTime(time.Now()), + }); err != nil { + return false, fmt.Errorf("record settlement: %w", err) + } + return true, nil +} + +// refundOnChain refunds orders, Outstanding orders of one transaction with one +// currency and refund address, in one transfer of their total. Each order +// becomes Refunded with its refund settlement row, together with the writes +// of extra, in the SQL transaction that reserves the transfer. If any order +// is no longer Outstanding, nothing is written or sent and errAlreadySettled is +// returned. A transfer that then fails or stays unknown leaves the refund +// decided: the amount is owed and visible on the transfer row. +func (p *PaymentHandler) refundOnChain(ctx context.Context, transactionID string, orders []database.DebugletOrder, extra func(*database.Queries) error) error { + transfer := database.ChainTransfer{Kind: transferRefund, TransactionID: transactionID} + for _, order := range orders { + transfer.Amount += order.Price + transfer.Currency, transfer.Receiver = order.Currency, order.RefundAddress + } + if len(orders) == 1 { + transfer.OrderID = sql.NullInt64{Int64: orders[0].OrderID, Valid: true} + } + return p.sendTransfer(ctx, transfer, func(queries *database.Queries) error { + if extra != nil { + if err := extra(queries); err != nil { + return err + } + } + for _, order := range orders { + changed, err := recordSettlement(ctx, queries, order, order.ExecutorID, settlementRefund) + if err != nil { + return fmt.Errorf("refund order %d of transaction %s: %w", order.OrderID, transactionID, err) + } + if !changed { + return errAlreadySettled + } + } + return nil + }) +} + // SettlePendingOrders applies SettleTerminalOrder to at most limit Outstanding // orders whose run is terminal with a recorded exit code and whose run id is // above after, in run order. A run without a recorded exit code is not settled: -// nothing infers its outcome. Only wholly local settlements are retried, credits -// and TEST refunds; a chain refund is never sent again from here and is counted -// as deferred. An order that fails stays Outstanding for a later pass and the -// others are still settled. next is the last run id attempted, or 0 once the +// nothing infers its outcome. A chain refund is taken only while no refund +// transfer covers the order: one reserved by an earlier attempt, in any state, +// is left to ReconcileTransfers, so a pass never sends a refund twice. While +// chain payments are disabled, chain-currency orders are counted as deferred +// and not attempted. An order that fails stays Outstanding for a later pass and +// the others are still settled. next is the last run id attempted, or 0 once the // listing is exhausted, so that the following pass continues after it and a // failing order never keeps later ones from being attempted. The first failure // is returned with the counts. @@ -397,7 +459,7 @@ func (p *PaymentHandler) SettlePendingOrders(ctx context.Context, after int64, l } run := row.Debuglet next = run.ID - if row.ExitCode.Int64 != 0 && isChainCurrency(row.Currency) { + if isChainCurrency(row.Currency) && p.chainDisabled() { deferred++ continue } @@ -413,7 +475,7 @@ func (p *PaymentHandler) SettlePendingOrders(ctx context.Context, after int64, l p.logger.Info("Settled order of terminal run", zap.String("debugletID", run.Uuid.String()), zap.Int64("exitCode", row.ExitCode.Int64)) } if deferred > 0 { - p.logger.Debug("Chain refunds are not retried automatically", zap.Int("deferred", deferred)) + p.logger.Debug("Chain settlements wait while chain payments are disabled", zap.Int("deferred", deferred)) } if exhausted { next = 0 @@ -437,6 +499,8 @@ func (p *PaymentHandler) RefundUnadmittedTransaction(transactionId string, ctx c } func (p *PaymentHandler) refundTransaction(transactionId string, ctx context.Context, unadmittedOnly bool) error { + // The transaction and its orders are read from one snapshot; the writes + // are conditional and decide again when the refund is reserved. tx, err := p.db.BeginTx(ctx, nil) if err != nil { return fmt.Errorf("Failed to begin transaction: %s", err.Error()) @@ -465,60 +529,59 @@ func (p *PaymentHandler) refundTransaction(transactionId string, ctx context.Con if err := p.requireChain(currency, "refund transaction"); err != nil { return err } - if unadmittedOnly { - changed, err := queries.RefundUnadmittedTransaction(ctx, database.RefundUnadmittedTransactionParams{ - RefundedStatus: int64(models.Refunded), TransactionID: transactionId, PaidStatus: int64(models.Paid), - }) - if err != nil { - return err - } - if changed != 1 { - return ErrTransactionClaimed - } - } - totalRefundValue := int64(0) + var outstanding []database.DebugletOrder for _, order := range orders { if order.Currency != currency || order.RefundAddress != refundAddress { // should never happen because these values are always set together but let's guard anyway return fmt.Errorf("Inconsistent orders in transaction %s", transactionId) } - if order.State == int64(models.Refunded) { - p.logger.Warn("Order has already been refunded", zap.String("transactionID", transactionId), zap.Int64("orderID", order.OrderID)) - continue + if unadmittedOnly && order.DebugletID.Valid { + return ErrTransactionClaimed } - totalRefundValue += order.Price - if _, err := queries.UpdateDebugletOrderState(ctx, database.UpdateDebugletOrderStateParams{ - State: int64(models.Refunded), - TransactionID: transactionId, - OrderID: order.OrderID, - }); err != nil { - return err + if order.State != int64(models.Outstanding) { + // A credited order was paid for with work; a refunded one is owed already. + p.logger.Warn("Order has already been settled", zap.String("transactionID", transactionId), zap.Int64("orderID", order.OrderID)) + continue } + outstanding = append(outstanding, order) } + if currency != "USDC" { + return fmt.Errorf("Refunds are not supported for currency %s", currency) + } + tx.Rollback() + // The transaction is refunded, not only its order rows. A submission is // admitted on the transaction's status, so a transaction left Paid after // its money went back would admit the same batch again and have the work // done a second time for a payment that no longer exists. - if !unadmittedOnly { - if err := queries.UpdateTransactionStatus(ctx, database.UpdateTransactionStatusParams{ - Status: int64(models.Refunded), ID: transactionId, - }); err != nil { + err = p.refundOnChain(ctx, transactionId, outstanding, func(queries *database.Queries) error { + if unadmittedOnly { + changed, err := queries.RefundUnadmittedTransaction(ctx, database.RefundUnadmittedTransactionParams{ + RefundedStatus: int64(models.Refunded), TransactionID: transactionId, PaidStatus: int64(models.Paid), + }) + if err != nil { + return err + } + if changed != 1 { + return ErrTransactionClaimed + } + return nil + } + changed, err := queries.RefundPaidTransaction(ctx, database.RefundPaidTransactionParams{ + Refunded: int64(models.Refunded), ID: transactionId, Paid: int64(models.Paid), + }) + if err != nil { return err } + if changed != 1 { + return fmt.Errorf("Tried to refund transaction that has not been payed: %s", transactionId) + } + return nil + }) + if errors.Is(err, errAlreadySettled) { + return fmt.Errorf("orders of transaction %s were settled during the refund; nothing was refunded", transactionId) } - - switch currency { - case "USDC": - _, err = p.sui.TransferCoins(uint64(totalRefundValue), sui.GetCoinType("USDC", p.cfg.Sui.Network), refundAddress, ctx) - default: - err = fmt.Errorf("Refunds are not supported for currency %s", orders[0].Currency) - } - - if err != nil { - return fmt.Errorf("Failed to execute refund transaction: %s", err.Error()) - } - - return tx.Commit() + return err } func (p *PaymentHandler) CreateEarningsIfNotExists(execID string, currency string, wallet string, queries *database.Queries, ctx context.Context) error { @@ -556,11 +619,28 @@ func (p *PaymentHandler) PayoutExecutor(earning database.Earning, ctx context.Co if err := p.requireChain(earning.Currency, "pay out earnings"); err != nil { return err } - switch earning.Currency { - case "USDC": - _, err := p.sui.TransferCoins(uint64(earning.CurrentBalance), sui.GetCoinType("USDC", p.cfg.Sui.Network), earning.SuiWalletAddress, ctx) - return err - default: + if earning.Currency != "USDC" { return fmt.Errorf("Unknown/unallowed currency %s", earning.Currency) } + if earning.CurrentBalance <= 0 { + return nil + } + // Exactly the balance read is reserved, so earnings credited meanwhile + // stay in the balance for the next payout. + payout := database.ChainTransfer{ + Kind: transferPayout, ExecutorID: earning.ExecutorID, Amount: earning.CurrentBalance, + Currency: earning.Currency, Receiver: earning.SuiWalletAddress, + } + return p.sendTransfer(ctx, payout, func(queries *database.Queries) error { + reserved, err := queries.ReservePayout(ctx, database.ReservePayoutParams{ + Amount: earning.CurrentBalance, ExecutorID: earning.ExecutorID, Currency: earning.Currency, + }) + if err != nil { + return fmt.Errorf("reserve payout: %w", err) + } + if reserved != 1 { + return errPayoutNotReserved + } + return nil + }) } diff --git a/internal/dispatcher/payments/payments_test.go b/internal/dispatcher/payments/payments_test.go index db70998b..0d4487fe 100644 --- a/internal/dispatcher/payments/payments_test.go +++ b/internal/dispatcher/payments/payments_test.go @@ -868,12 +868,6 @@ func TestDisabledTESTPaths(t *testing.T) { mock.ExpectBegin() expectTransactionRead(mock, "TEST", models.Paid) expectTransactionOrdersRead(mock, "TEST", 1) - mock.ExpectQuery(updateDebugletOrderStateQuery). - WithArgs(int64(models.Refunded), testTxID, testOrderID). - WillReturnRows(orderRow("TEST", models.Refunded)) - mock.ExpectExec(updateTransactionStatusQuery). - WithArgs(int64(models.Refunded), testTxID). - WillReturnResult(sqlmock.NewResult(0, 1)) mock.ExpectRollback() err := h.RefundTransaction(testTxID, context.Background()) @@ -1005,77 +999,6 @@ func TestUnrelatedReadErrorsArePreserved(t *testing.T) { func TestEnabledChainMethodsForwardToBackend(t *testing.T) { const testReceiver = "0x00000000000000000000000000000000000000000000000000000000000000aa" - t.Run("RefundDebugletOrder USDC", func(t *testing.T) { - db, mock := newMockDB(t) - h, rec := newEnabledHandler(t, db, enabledConfig(t)) - mock.ExpectBegin() - expectOrderRead(mock, "USDC", models.Outstanding) - mock.ExpectExec(transitionDebugletOrderQuery). - WithArgs(int64(models.Refunded), testTxID, testOrderID, int64(models.Outstanding)). - WillReturnResult(sqlmock.NewResult(0, 1)) - mock.ExpectExec("INSERT INTO order_settlements"). - WithArgs(testTxID, testOrderID, "refund", testPrice, "USDC", testExecutor, nil, sqlmock.AnyArg()). - WillReturnResult(sqlmock.NewResult(0, 1)) - mock.ExpectCommit() - - if err := h.RefundDebugletOrder(testDebuglet(), testRefund, context.Background()); err != nil { - t.Fatalf("RefundDebugletOrder: %v", err) - } - if calls := rec.chain.Calls(); len(calls) != 1 || calls[0] != "RefundDebuglet" { - t.Fatalf("backend calls %v", calls) - } - if rec.chain.refundAddress != testRefund || rec.chain.refundOrder == nil || rec.chain.refundOrder.State != int64(models.Refunded) { - t.Fatalf("forwarded refund (%v, %q)", rec.chain.refundOrder, rec.chain.refundAddress) - } - assertMet(t, mock) - }) - - t.Run("RefundTransaction USDC", func(t *testing.T) { - db, mock := newMockDB(t) - h, rec := newEnabledHandler(t, db, enabledConfig(t)) - mock.ExpectBegin() - expectTransactionRead(mock, "USDC", models.Paid) - expectTransactionOrdersRead(mock, "USDC", 2) - for i := int64(1); i <= 2; i++ { - mock.ExpectQuery(updateDebugletOrderStateQuery). - WithArgs(int64(models.Refunded), testTxID, i). - WillReturnRows(orderRow("USDC", models.Refunded)) - } - mock.ExpectExec(updateTransactionStatusQuery). - WithArgs(int64(models.Refunded), testTxID). - WillReturnResult(sqlmock.NewResult(0, 1)) - mock.ExpectCommit() - - if err := h.RefundTransaction(testTxID, context.Background()); err != nil { - t.Fatalf("RefundTransaction: %v", err) - } - if calls := rec.chain.Calls(); len(calls) != 1 || calls[0] != "TransferCoins" { - t.Fatalf("backend calls %v", calls) - } - if rec.chain.transferAmount != uint64(2*testPrice) || rec.chain.transferAddress != testRefund || - rec.chain.transferCoinType != sui.GetCoinType("USDC", "testnet") { - t.Fatalf("forwarded transfer (%d, %q, %q)", rec.chain.transferAmount, rec.chain.transferCoinType, rec.chain.transferAddress) - } - assertMet(t, mock) - }) - - t.Run("PayoutExecutor USDC", func(t *testing.T) { - db, mock := newMockDB(t) - h, rec := newEnabledHandler(t, db, enabledConfig(t)) - err := h.PayoutExecutor(database.Earning{ExecutorID: testExecutor, Currency: "USDC", CurrentBalance: 42, SuiWalletAddress: "0xwallet"}, context.Background()) - if err != nil { - t.Fatalf("PayoutExecutor: %v", err) - } - if calls := rec.chain.Calls(); len(calls) != 1 || calls[0] != "TransferCoins" { - t.Fatalf("backend calls %v", calls) - } - if rec.chain.transferAmount != 42 || rec.chain.transferAddress != "0xwallet" || - rec.chain.transferCoinType != sui.GetCoinType("USDC", "testnet") { - t.Fatalf("forwarded transfer (%d, %q, %q)", rec.chain.transferAmount, rec.chain.transferCoinType, rec.chain.transferAddress) - } - assertMet(t, mock) - }) - t.Run("TransferUSDC reaches the backend once", func(t *testing.T) { db, mock := newMockDB(t) h, rec := newEnabledHandler(t, db, enabledConfig(t)) @@ -1261,7 +1184,7 @@ func TestRefundTransactionSpendsTheTransactionItself(t *testing.T) { if err := h.RefundTransaction(id, ctx); err != nil { t.Fatalf("refund: %v", err) } - if !strings.Contains(strings.Join(rec.chain.Calls(), " "), "TransferCoins") { + if !strings.Contains(strings.Join(rec.chain.Calls(), " "), "ExecuteTransfer") { t.Fatalf("the money was not moved: %v", rec.chain.Calls()) } refunded, err := queries.GetTransactionByID(ctx, id) diff --git a/internal/dispatcher/payments/payout.go b/internal/dispatcher/payments/payout.go index 3f0dfc84..2eaf7abc 100644 --- a/internal/dispatcher/payments/payout.go +++ b/internal/dispatcher/payments/payout.go @@ -12,16 +12,27 @@ import ( "go.uber.org/zap" ) +// Handler pays out one executor's earnings and reconciles outbound transfers +// whose outcome is not known yet. type Handler interface { PayoutExecutor(database.Earning, context.Context) error + ReconcileTransfers(ctx context.Context, limit int) error } type PayoutTicker struct { - ticker *time.Ticker - database *sql.DB - handler Handler - logger *zap.Logger + ticker *time.Ticker + reconcile *time.Ticker + database *sql.DB + handler Handler + logger *zap.Logger } +// Reconciliation looks up at most reconcileLimit open transfers every +// reconcileInterval. +const ( + reconcileInterval = time.Minute + reconcileLimit = 100 +) + func NextTickDuration() time.Duration { now := time.Now() //Ticks every day at midnight @@ -34,19 +45,22 @@ func NextTickDuration() time.Duration { func NewPayoutTicker(db *sql.DB, h Handler, logger *zap.Logger) *PayoutTicker { return &PayoutTicker{ - ticker: time.NewTicker(NextTickDuration()), - database: db, - handler: h, - logger: logger, + ticker: time.NewTicker(NextTickDuration()), + reconcile: time.NewTicker(reconcileInterval), + database: db, + handler: h, + logger: logger, } } -// StartPayoutLoop pays executors on every tick until ctx is cancelled. On -// cancellation it stops the ticker and returns nil, mirroring the Sui listener, -// so PaymentHandler.Start can join both loops promptly instead of waiting for -// the next midnight. +// StartPayoutLoop pays executors on every payout tick and reconciles open +// transfers on every reconciliation tick until ctx is cancelled. On +// cancellation it stops both tickers and returns nil, mirroring the Sui +// listener, so PaymentHandler.Start can join both loops promptly instead of +// waiting for the next midnight. func (pt *PayoutTicker) StartPayoutLoop(ctx context.Context) error { defer pt.ticker.Stop() + defer pt.reconcile.Stop() for { select { case <-ctx.Done(): @@ -55,10 +69,16 @@ func (pt *PayoutTicker) StartPayoutLoop(ctx context.Context) error { pt.logger.Info("Starting payout") pt.PayExecutors(ctx) pt.ticker.Reset(NextTickDuration()) + case <-pt.reconcile.C: + if err := pt.handler.ReconcileTransfers(ctx, reconcileLimit); err != nil { + pt.logger.Error("Failed to reconcile transfers", zap.String("error", err.Error())) + } } } } +// PayExecutors pays out every positive chain balance. A balance without a +// payout wallet is not paid out; how many there are is logged once per pass. func (pt *PayoutTicker) PayExecutors(ctx context.Context) { queries := database.New(pt.database) earnings, err := queries.GetEarnings(ctx) @@ -66,15 +86,24 @@ func (pt *PayoutTicker) PayExecutors(ctx context.Context) { pt.logger.Error("Failed to query earnings", zap.String("error", err.Error())) return } + withoutWallet := 0 for _, earning := range earnings { - err := pt.handler.PayoutExecutor(earning, ctx) - if err == nil { - queries.SettleEarning(ctx, database.SettleEarningParams{ - ExecutorID: earning.ExecutorID, - Currency: earning.Currency, - }) - } else { + if ctx.Err() != nil { + return + } + // TEST earnings are bookkeeping only and are never paid out. + if earning.CurrentBalance <= 0 || !isChainCurrency(earning.Currency) { + continue + } + if earning.SuiWalletAddress == "" { + withoutWallet++ + continue + } + if err := pt.handler.PayoutExecutor(earning, ctx); err != nil { pt.logger.Error("Failed to Settle Earnings", zap.String("execID", earning.ExecutorID), zap.String("error", err.Error())) } } + if withoutWallet > 0 { + pt.logger.Warn("Earnings without a payout wallet were not paid out", zap.Int("balances", withoutWallet)) + } } diff --git a/internal/dispatcher/payments/payout_test.go b/internal/dispatcher/payments/payout_test.go index 86070a36..0344a5c9 100644 --- a/internal/dispatcher/payments/payout_test.go +++ b/internal/dispatcher/payments/payout_test.go @@ -2,71 +2,60 @@ package payments import ( "context" - "database/sql/driver" "errors" + "fmt" "regexp" "sync" "testing" "time" "github.com/netsec-ethz/debuglet/internal/dispatcher/database" + "github.com/netsec-ethz/debuglet/internal/dispatcher/payments/sui" "github.com/DATA-DOG/go-sqlmock" "go.uber.org/zap" + "go.uber.org/zap/zapcore" + "go.uber.org/zap/zaptest/observer" ) var ( getEarningsQuery = regexp.QuoteMeta( "SELECT executor_id, currency, total_income, current_balance, sui_wallet_address FROM earnings", ) + `\s*$` - settleEarningQuery = regexp.QuoteMeta( - "UPDATE earnings \nSET current_balance = 0\nWHERE executor_id = ? AND currency = ?", - ) ) // recordingHandler is a payout Handler that records every earning it is asked -// to pay and fails for the executors listed in failFor. +// to pay and counts reconciliation passes. type recordingHandler struct { - mu sync.Mutex - paid []database.Earning - failFor map[string]bool + mu sync.Mutex + paid []database.Earning + reconciled int } func (r *recordingHandler) PayoutExecutor(e database.Earning, ctx context.Context) error { r.mu.Lock() defer r.mu.Unlock() r.paid = append(r.paid, e) - if r.failFor[e.ExecutorID] { - return errScripted - } return nil } -func (r *recordingHandler) Paid() []database.Earning { +func (r *recordingHandler) ReconcileTransfers(ctx context.Context, limit int) error { r.mu.Lock() defer r.mu.Unlock() - return append([]database.Earning(nil), r.paid...) -} - -// argRecorder is a sqlmock argument matcher that accepts any value and keeps -// what it saw, so a test can assert which rows a statement was run for even -// when the production code discards the statement's error. -type argRecorder struct { - mu sync.Mutex - seen []driver.Value + r.reconciled++ + return nil } -func (r *argRecorder) Match(v driver.Value) bool { +func (r *recordingHandler) Paid() []database.Earning { r.mu.Lock() defer r.mu.Unlock() - r.seen = append(r.seen, v) - return true + return append([]database.Earning(nil), r.paid...) } -func (r *argRecorder) Seen() []driver.Value { +func (r *recordingHandler) Reconciled() int { r.mu.Lock() defer r.mu.Unlock() - return append([]driver.Value(nil), r.seen...) + return r.reconciled } func newTestTicker(t *testing.T, period time.Duration, handler Handler) (*PayoutTicker, sqlmock.Sqlmock) { @@ -74,7 +63,9 @@ func newTestTicker(t *testing.T, period time.Duration, handler Handler) (*Payout db, mock := newMockDB(t) ticker := time.NewTicker(period) t.Cleanup(ticker.Stop) - return &PayoutTicker{ticker: ticker, database: db, handler: handler, logger: zap.NewNop()}, mock + reconcile := time.NewTicker(period) + t.Cleanup(reconcile.Stop) + return &PayoutTicker{ticker: ticker, reconcile: reconcile, database: db, handler: handler, logger: zap.NewNop()}, mock } // TestStartPayoutLoopReturnsOnCancel runs the loop with a ticker that would @@ -123,6 +114,8 @@ func TestStartPayoutLoopStopsTicker(t *testing.T) { select { case tick := <-pt.ticker.C: t.Fatalf("ticker delivered a tick at %v after the loop returned; ticker not stopped", tick) + case tick := <-pt.reconcile.C: + t.Fatalf("reconciliation ticker delivered a tick at %v after the loop returned; ticker not stopped", tick) case <-time.After(10 * period): } if paid := handler.Paid(); len(paid) != 0 { @@ -131,33 +124,83 @@ func TestStartPayoutLoopStopsTicker(t *testing.T) { assertMet(t, mock) } -// TestPayExecutorsSettlesOnlySuccessfulPayouts drives PayExecutors directly: -// a successful payout settles the earning row, a failed one leaves it alone. -// The failing executor is listed first so a wrong settle would be recorded by -// the argument recorder before the expected one. -func TestPayExecutorsSettlesOnlySuccessfulPayouts(t *testing.T) { - handler := &recordingHandler{failFor: map[string]bool{"exec-fail": true}} +// TestStartPayoutLoopReconciles runs the loop with a payout tick an hour away +// and a short reconciliation tick: reconciliation runs without a payout. +func TestStartPayoutLoopReconciles(t *testing.T) { + handler := &recordingHandler{} pt, mock := newTestTicker(t, time.Hour, handler) + pt.reconcile.Reset(time.Millisecond) - mock.ExpectQuery(getEarningsQuery).WillReturnRows(sqlmock.NewRows(earningsColumns). - AddRow("exec-fail", "USDC", int64(10), int64(10), "0xfail"). - AddRow("exec-ok", "USDC", int64(20), int64(20), "0xok")) - settled := &argRecorder{} - mock.ExpectExec(settleEarningQuery).WithArgs(settled, "USDC").WillReturnResult(sqlmock.NewResult(0, 1)) + ctx, cancel := context.WithCancel(context.Background()) + errCh := make(chan error, 1) + go func() { errCh <- pt.StartPayoutLoop(ctx) }() + deadline := time.Now().Add(waitBound) + for handler.Reconciled() < 2 && time.Now().Before(deadline) { + time.Sleep(time.Millisecond) + } + cancel() + if err := waitErr(t, errCh, "StartPayoutLoop after cancellation"); err != nil { + t.Fatalf("StartPayoutLoop returned %v", err) + } + if n := handler.Reconciled(); n < 2 { + t.Fatalf("%d reconciliation passes within %v", n, waitBound) + } + if paid := handler.Paid(); len(paid) != 0 { + t.Fatalf("reconciliation paid executors: %v", paid) + } + assertMet(t, mock) +} + +// TestPayExecutorsSettlesOnlySuccessfulPayouts drives PayExecutors over a +// real database: a successful payout leaves the balance reserved in a sent +// transfer, a failed one restores the balance and records the failure. A +// zero balance is skipped, and a balance without a wallet is not paid out and +// counted once. +func TestPayExecutorsSettlesOnlySuccessfulPayouts(t *testing.T) { + db, h, chain := transferFixture(t) + const failWallet, okWallet = "0xfa11", "0x0c" + chain.executeFor = map[string]error{failWallet: fmt.Errorf("%w: refused", sui.ErrNotBroadcast)} + seedEarning(t, db, "exec-fail", 10, failWallet) + seedEarning(t, db, "exec-ok", 20, okWallet) + seedEarning(t, db, "exec-empty", 0, "0x0e") + seedEarning(t, db, "exec-nowallet", 30, "") + if _, err := database.New(db).CreateEarnings(t.Context(), database.CreateEarningsParams{ExecutorID: "exec-test", Currency: "TEST"}); err != nil { + t.Fatal(err) + } + if err := database.New(db).AddEarnings(t.Context(), database.AddEarningsParams{Amount: 40, ExecutorID: "exec-test", Currency: "TEST"}); err != nil { + t.Fatal(err) + } + core, logs := observer.New(zapcore.DebugLevel) + pt := &PayoutTicker{database: db, handler: h, logger: zap.New(core)} pt.PayExecutors(context.Background()) - paid := handler.Paid() - if len(paid) != 2 || paid[0].ExecutorID != "exec-fail" || paid[1].ExecutorID != "exec-ok" { - t.Fatalf("payout attempts %v", paid) + if n := count(chain.Calls(), "ExecuteTransfer"); n != 2 || count(chain.Calls(), "PrepareTransfer") != 2 { + t.Fatalf("%d executions, want 2", n) } - if paid[1].CurrentBalance != 20 || paid[1].SuiWalletAddress != "0xok" { - t.Fatalf("earning forwarded to handler %+v", paid[1]) + for executor, want := range map[string]int64{"exec-fail": 10, "exec-ok": 0, "exec-empty": 0, "exec-nowallet": 30} { + if got := earningOf(t, db, executor).CurrentBalance; got != want { + t.Fatalf("balance of %s %d, want %d", executor, got, want) + } } - if seen := settled.Seen(); len(seen) != 1 || seen[0] != "exec-ok" { - t.Fatalf("settled executors %v, want [exec-ok]", seen) + states := map[string]string{} + for _, row := range transferRows(t, db) { + states[row.ExecutorID] = row.State + } + if fmt.Sprint(states) != fmt.Sprint(map[string]string{"exec-fail": transferFailed, "exec-ok": transferSent}) { + t.Fatalf("transfer states %v", states) + } + test, err := database.New(db).GetEarningsIn(t.Context(), database.GetEarningsInParams{ExecutorID: "exec-test", Currency: "TEST"}) + if err != nil || test.CurrentBalance != 40 { + t.Fatalf("TEST earning %+v, %v; want it untouched", test, err) + } + if n := logs.FilterField(zap.String("execID", "exec-test")).Len(); n != 0 { + t.Fatalf("TEST earning logged %d times", n) + } + warned := logs.FilterMessage("Earnings without a payout wallet were not paid out").All() + if len(warned) != 1 || warned[0].ContextMap()["balances"] != int64(1) { + t.Fatalf("wallet warnings %v", warned) } - assertMet(t, mock) } func TestPayExecutorsQueryError(t *testing.T) { diff --git a/internal/dispatcher/payments/refund_admission_test.go b/internal/dispatcher/payments/refund_admission_test.go index 5e9be0e0..19434365 100644 --- a/internal/dispatcher/payments/refund_admission_test.go +++ b/internal/dispatcher/payments/refund_admission_test.go @@ -88,7 +88,7 @@ func TestUnadmittedRefundAndAdmissionHaveOneWinner(t *testing.T) { if err := h.RefundTransaction(testTxID, t.Context()); err != nil { t.Fatalf("transaction-wide fallback: %v", err) } - if orderState(t, db) != models.Refunded || strings.Join(chain.Calls(), ",") != "TransferCoins" { + if orderState(t, db) != models.Refunded || strings.Join(chain.Calls(), ",") != "PrepareTransfer,ExecuteTransfer" { t.Fatal("transaction-wide fallback changed") } }) @@ -112,7 +112,7 @@ func TestUnadmittedRefundAndAdmissionHaveOneWinner(t *testing.T) { if err := db.QueryRow("SELECT COUNT(*) FROM debuglets").Scan(&runs); err != nil || runs != 0 { t.Fatalf("refused admission left %d runs: %v", runs, err) } - if orderState(t, db) != models.Refunded || strings.Join(chain.Calls(), ",") != "TransferCoins" { + if orderState(t, db) != models.Refunded || strings.Join(chain.Calls(), ",") != "PrepareTransfer,ExecuteTransfer" { t.Fatal("refund or spent order changed") } }) @@ -177,10 +177,10 @@ func TestUnadmittedRefundRacesAdmissionOnTwoHandles(t *testing.T) { t.Fatal(err) } if refunded { - if runs != 0 || orderState(t, db) != models.Refunded || strings.Join(chain.Calls(), ",") != "TransferCoins" { + if runs != 0 || orderState(t, db) != models.Refunded || strings.Join(chain.Calls(), ",") != "PrepareTransfer,ExecuteTransfer" { t.Fatal("refund winner also admitted work or transferred twice") } - } else if runs != 1 || orderState(t, db) != models.Outstanding || len(chain.Calls()) != 0 { + } else if runs != 1 || orderState(t, db) != models.Outstanding || strings.Contains(strings.Join(chain.Calls(), ","), "ExecuteTransfer") { t.Fatal("admission winner was refunded or duplicated") } }) diff --git a/internal/dispatcher/payments/settlement_test.go b/internal/dispatcher/payments/settlement_test.go index adbc24f3..47ab1072 100644 --- a/internal/dispatcher/payments/settlement_test.go +++ b/internal/dispatcher/payments/settlement_test.go @@ -5,12 +5,10 @@ package payments import ( "database/sql" - "errors" "strings" "testing" "time" - "github.com/DATA-DOG/go-sqlmock" "github.com/google/uuid" "github.com/netsec-ethz/debuglet/internal/dispatcher/database" "github.com/netsec-ethz/debuglet/internal/dispatcher/models" @@ -217,73 +215,20 @@ func seedFailedUSDCRun(t *testing.T, db *sql.DB) database.Debuglet { return run } -// A chain refund whose outcome is uncertain is not sent again: after a failed -// inline refund the order stays Outstanding with no settlement row, and a -// later settlement pass does not select it, so the chain backend is reached -// exactly once. -func TestSettlementPassNeverResendsAChainRefund(t *testing.T) { +// While chain payments are disabled a settlement pass defers a chain order +// without reaching the chain backend and leaves it Outstanding. +func TestSettlementPassDefersChainOrdersWhileDisabled(t *testing.T) { db := newRefundDatabase(t) - h, rec := newEnabledHandler(t, db, enabledConfig(t)) - rec.chain.refundErr = errors.New("transfer response lost") - run := seedFailedUSDCRun(t, db) - - if err := h.SettleTerminalOrder(t.Context(), &run, 3); !errors.Is(err, rec.chain.refundErr) { - t.Fatalf("SettleTerminalOrder = %v, want the chain failure", err) - } - if got := orderState(t, db); got != models.Outstanding { - t.Fatalf("order state %v after a failed chain refund, want %v", got, models.Outstanding) - } - if n, _ := settlementOf(t, db); n != 0 { - t.Fatalf("%d settlement rows after a failed chain refund", n) - } - for range 2 { - settled, failed, deferred, next, err := h.SettlePendingOrders(t.Context(), 0, 32) - if settled != 0 || failed != 0 || deferred != 0 || next != 0 || err != nil { - t.Fatalf("pass: settled %d, failed %d, deferred %d, next %d, error %v", settled, failed, deferred, next, err) - } - } - if calls := rec.chain.Calls(); len(calls) != 1 || calls[0] != "RefundDebuglet" { - t.Fatalf("chain backend calls %v, want one RefundDebuglet", calls) - } - if got := orderState(t, db); got != models.Outstanding { - t.Fatalf("order state %v after the passes, want %v", got, models.Outstanding) - } -} - -// A chain refund that was sent but whose local commit failed leaves the order -// Outstanding. A settlement pass that is handed such an order anyway defers it -// without reaching the chain backend again. -func TestSettlementPassDefersAChainRefundAfterAFailedCommit(t *testing.T) { - db, mock := newMockDB(t) - h, rec := newEnabledHandler(t, db, enabledConfig(t)) - run := database.Debuglet{ID: 9, Uuid: uuid.New(), TransactionID: testTxID, OrderID: testOrderID, ExecutorID: testExecutor, State: models.RunStateExited} - - mock.ExpectBegin() - expectOrderRead(mock, "USDC", models.Outstanding) - mock.ExpectExec(transitionDebugletOrderQuery). - WithArgs(int64(models.Refunded), testTxID, testOrderID, int64(models.Outstanding)). - WillReturnResult(sqlmock.NewResult(0, 1)) - mock.ExpectExec("INSERT INTO order_settlements"). - WithArgs(testTxID, testOrderID, "refund", testPrice, "USDC", testExecutor, nil, sqlmock.AnyArg()). - WillReturnResult(sqlmock.NewResult(0, 1)) - mock.ExpectCommit().WillReturnError(errScripted) - if err := h.SettleTerminalOrder(t.Context(), &run, 3); !errors.Is(err, errScripted) { - t.Fatalf("SettleTerminalOrder = %v, want the failed commit", err) - } - - now := time.Now().UTC() - mock.ExpectQuery("-- name: ListPendingSettlements"). - WithArgs(int64(models.Outstanding), int64(models.RunStateExited), int64(0), int64(32)). - WillReturnRows(sqlmock.NewRows([]string{"id", "uuid", "start_time", "end_time", "usage", "ceil_bw", "executor_id", - "addresses", "state", "error", "transaction_id", "order_id", "dispatcher_incarnation", "session_id", "exit_code", "currency"}). - AddRow(run.ID, run.Uuid.String(), now, now, 0, 0, testExecutor, "", int64(models.RunStateExited), nil, - testTxID, testOrderID, "", "", 3, "USDC")) + h, rec := newDisabledHandler(t, db, true, true) + seedFailedUSDCRun(t, db) settled, failed, deferred, next, err := h.SettlePendingOrders(t.Context(), 0, 32) if settled != 0 || failed != 0 || deferred != 1 || next != 0 || err != nil { t.Fatalf("pass: settled %d, failed %d, deferred %d, next %d, error %v", settled, failed, deferred, next, err) } - if calls := rec.chain.Calls(); len(calls) != 1 || calls[0] != "RefundDebuglet" { - t.Fatalf("chain backend calls %v, want one RefundDebuglet", calls) + if calls := rec.chain.Calls(); len(calls) != 0 { + t.Fatalf("chain backend calls %v while disabled", calls) + } + if got := orderState(t, db); got != models.Outstanding { + t.Fatalf("order state %v, want %v", got, models.Outstanding) } - assertMet(t, mock) } diff --git a/internal/dispatcher/payments/sui/listener_test.go b/internal/dispatcher/payments/sui/listener_test.go index 75ca5225..ce615021 100644 --- a/internal/dispatcher/payments/sui/listener_test.go +++ b/internal/dispatcher/payments/sui/listener_test.go @@ -4,6 +4,7 @@ package sui_test import ( + "bytes" "context" "database/sql" "encoding/base64" @@ -13,6 +14,7 @@ import ( "math" "net/http" "net/http/httptest" + "os" "path/filepath" "strconv" "strings" @@ -30,6 +32,7 @@ import ( suiModels "github.com/block-vision/sui-go-sdk/models" "github.com/block-vision/sui-go-sdk/mystenbcs" v2 "github.com/block-vision/sui-go-sdk/pb/sui/rpc/v2" + "github.com/block-vision/sui-go-sdk/signer" "go.uber.org/zap" "go.uber.org/zap/zapcore" "go.uber.org/zap/zaptest/observer" @@ -45,7 +48,10 @@ const ( ) var ( - ourAddress = strings.Repeat("ab", 32) + // ourSeed is the Ed25519 seed of the dispatcher key in the fixture's + // keystore; ourAddress is its address without the 0x prefix. + ourSeed = bytes.Repeat([]byte{0x5e}, 32) + ourAddress = strings.TrimPrefix(signer.NewSigner(ourSeed).Address, "0x") otherAddress = strings.Repeat("cd", 32) // The chain reports coin types without the 0x prefix. usdcType = strings.TrimPrefix(sui.GetCoinType("USDC", "testnet"), "0x") @@ -222,9 +228,24 @@ type fixture struct { ph *payments.PaymentHandler } +// writeKeystore writes a keystore file holding the dispatcher key: a JSON +// array with one base64 entry of the Ed25519 flag byte and the seed. +func writeKeystore(t *testing.T) string { + t.Helper() + raw, err := json.Marshal([]string{base64.StdEncoding.EncodeToString(append([]byte{0}, ourSeed...))}) + if err != nil { + t.Fatal(err) + } + path := filepath.Join(t.TempDir(), "sui.keystore") + if err := os.WriteFile(path, raw, 0o600); err != nil { + t.Fatal(err) + } + return path +} + // newFixture is a real dispatcher database, an enabled payment handler as the // listener's fulfiller and a local GraphQL server. No chain endpoint is -// contacted: the gRPC client is never used and the keystore does not exist. +// contacted: the gRPC client is never used. func newFixture(t *testing.T) *fixture { t.Helper() db, err := sqlitedb.Open(filepath.Join(t.TempDir(), "dispatcher.sqlite"), sqlitedb.Create()) @@ -244,10 +265,15 @@ func newFixture(t *testing.T) *fixture { GraphQLURL: httpServer.URL, Address: "0x" + strings.ToUpper(ourAddress), PaymentKitPackage: testPackage, - KeystorePath: filepath.Join(t.TempDir(), "missing.keystore"), + PaymentRegistryId: "0x1", + KeystorePath: writeKeystore(t), }} core, logs := observer.New(zapcore.InfoLevel) - return &fixture{db: db, cfg: cfg, server: server, core: core, logs: logs, ph: payments.NewPaymentHandler(db, cfg, zap.New(core))} + ph, err := payments.NewPaymentHandler(db, cfg, zap.New(core)) + if err != nil { + t.Fatalf("NewPaymentHandler: %v", err) + } + return &fixture{db: db, cfg: cfg, server: server, core: core, logs: logs, ph: ph} } // listener is a fresh listener, as after a dispatcher restart. diff --git a/internal/dispatcher/payments/transfers.go b/internal/dispatcher/payments/transfers.go new file mode 100644 index 00000000..1f7e29e0 --- /dev/null +++ b/internal/dispatcher/payments/transfers.go @@ -0,0 +1,250 @@ +// SPDX-License-Identifier: Apache-2.0 +// Copyright 2026 ETH Zurich + +package payments + +import ( + "context" + "errors" + "fmt" + "time" + + "github.com/netsec-ethz/debuglet/internal/dispatcher/database" + "github.com/netsec-ethz/debuglet/internal/dispatcher/models" + "github.com/netsec-ethz/debuglet/internal/dispatcher/payments/sui" + + "go.uber.org/zap" +) + +// Kinds and states of outbound transfers (chain_transfers). A transfer is +// reserved before the chain is called, sent once the chain accepted it and +// confirmed once a lookup found it executed with the intended credit. failed +// means no coins moved; unknown means the outcome is not known and only +// reconciliation resolves it. +const ( + transferPayout = "payout" + transferRefund = "refund" + + transferReserved = "reserved" + transferSent = "sent" + transferConfirmed = "confirmed" + transferFailed = "failed" + transferUnknown = "unknown" +) + +// staleReservation is how long a reserved transfer may wait for its chain +// call. An older reserved row was left by a stop between the reservation and +// the call, and reconciliation looks it up. +const staleReservation = 10 * time.Minute + +// executeTimeout bounds one submission of a transfer. A submission that does +// not answer in time has an unknown outcome, which is how its error is +// recorded. +const executeTimeout = 2 * time.Minute + +var ( + // errAlreadySettled is returned by a transfer decision that finds the + // decision already taken. Nothing is reserved and nothing is sent. + errAlreadySettled = errors.New("already settled") + // errPayoutNotReserved means the balance could not be reserved: another + // payout of the same executor and currency is still open, or the balance + // changed since it was read. + errPayoutNotReserved = errors.New("payout not reserved: an earlier payout is still open or the balance changed") +) + +// reservedTransfer is a transfer row in state reserved with its signed +// transaction, ready to be submitted. +type reservedTransfer struct { + row database.ChainTransfer + prepared *sui.PreparedTransfer +} + +// sendTransfer runs the transfer lifecycle: the chain backend builds and signs +// the transfer; one SQL transaction applies decide (the local decision the +// transfer pays for) and reserves the transfer row with the signed +// transaction; then the transfer is submitted and its outcome recorded. An +// error means the decision was not taken and nothing was sent. Once the +// decision is committed, a failed or unknown transfer is recorded on its row +// and logged, and nil is returned. +func (p *PaymentHandler) sendTransfer(ctx context.Context, transfer database.ChainTransfer, decide func(*database.Queries) error) error { + reserved, err := p.prepareAndReserve(ctx, transfer, decide) + if err != nil || reserved == nil { + return err + } + p.executeReserved(ctx, reserved) + return nil +} + +// prepareAndReserve prepares the transfer, then commits decide together with +// the reserved transfer row. A transfer of amount zero moves nothing: decide is +// committed alone and nil is returned without a reservation. +func (p *PaymentHandler) prepareAndReserve(ctx context.Context, transfer database.ChainTransfer, decide func(*database.Queries) error) (*reservedTransfer, error) { + var prepared *sui.PreparedTransfer + if transfer.Amount > 0 { + var err error + prepared, err = p.sui.PrepareTransfer(ctx, uint64(transfer.Amount), sui.GetCoinType(transfer.Currency, p.cfg.Sui.Network), transfer.Receiver) + if err != nil { + return nil, fmt.Errorf("prepare %s transfer: %w", transfer.Kind, err) + } + } + + tx, err := p.db.BeginTx(ctx, nil) + if err != nil { + return nil, fmt.Errorf("failed to begin transaction: %w", err) + } + defer tx.Rollback() + queries := database.New(p.db).WithTx(tx) + if err := decide(queries); err != nil { + return nil, err + } + if prepared != nil { + now := models.NewUTCTime(time.Now()) + transfer.State, transfer.Digest, transfer.CreatedAt, transfer.UpdatedAt = transferReserved, prepared.Digest, now, now + transfer.SignedTransaction, transfer.Signature = append([]byte{}, prepared.Bytes()...), prepared.Signature() + transfer.ID, err = queries.InsertChainTransfer(ctx, database.InsertChainTransferParams{ + Kind: transfer.Kind, ExecutorID: transfer.ExecutorID, TransactionID: transfer.TransactionID, OrderID: transfer.OrderID, + Amount: transfer.Amount, Currency: transfer.Currency, Receiver: transfer.Receiver, State: transfer.State, + Digest: transfer.Digest, SignedTransaction: transfer.SignedTransaction, Signature: transfer.Signature, + CreatedAt: transfer.CreatedAt, UpdatedAt: transfer.UpdatedAt, + }) + if err != nil { + return nil, fmt.Errorf("reserve %s transfer: %w", transfer.Kind, err) + } + } + if err := tx.Commit(); err != nil { + return nil, fmt.Errorf("commit %s decision: %w", transfer.Kind, err) + } + if prepared == nil { + return nil, nil + } + return &reservedTransfer{row: transfer, prepared: prepared}, nil +} + +// executeReserved submits a reserved transfer and records the outcome. The +// outcome is recorded even when ctx ended during the call; if it cannot be +// recorded, the row stays reserved and reconciliation resolves it. +func (p *PaymentHandler) executeReserved(ctx context.Context, reserved *reservedTransfer) { + state, detail := executionOutcome(p.submit(ctx, reserved.prepared)) + p.recordTransfer(context.WithoutCancel(ctx), reserved.row, state, reserved.prepared.Digest, detail) +} + +// submit runs ExecuteTransfer bounded by executeTimeout. +func (p *PaymentHandler) submit(ctx context.Context, prepared *sui.PreparedTransfer) error { + ctx, cancel := context.WithTimeout(ctx, executeTimeout) + defer cancel() + return p.sui.ExecuteTransfer(ctx, prepared) +} + +// executionOutcome maps the result of ExecuteTransfer to the state it leaves: +// accepted is sent; not broadcast or executed with failure status is failed; +// anything else may have been submitted and is unknown. +func executionOutcome(err error) (state string, detail string) { + switch { + case err == nil: + return transferSent, "" + case errors.Is(err, sui.ErrNotBroadcast), errors.Is(err, sui.ErrTransferFailed): + return transferFailed, err.Error() + default: + return transferUnknown, err.Error() + } +} + +// recordTransfer moves the transfer from its current state to state with the +// digest the chain backend reported. A payout that becomes failed returns its +// amount to the executor's balance in the same SQL transaction. The write is +// conditional on the state the caller read, so an outcome recorded +// concurrently is not overwritten and a balance is released once. +func (p *PaymentHandler) recordTransfer(ctx context.Context, t database.ChainTransfer, state string, digest string, detail string) { + logger := p.logger.With(zap.Int64("transfer", t.ID), zap.String("kind", t.Kind), zap.String("digest", digest)) + if digest != t.Digest { + logger.Warn("Transfer digest differs from the reserved digest", zap.String("reserved", t.Digest)) + } + err := func() error { + tx, err := p.db.BeginTx(ctx, nil) + if err != nil { + return err + } + defer tx.Rollback() + queries := database.New(p.db).WithTx(tx) + changed, err := queries.UpdateChainTransfer(ctx, database.UpdateChainTransferParams{ + State: state, Digest: digest, Detail: detail, UpdatedAt: models.NewUTCTime(time.Now()), ID: t.ID, FromState: t.State, + }) + if err != nil { + return err + } + if changed == 0 { + logger.Debug("Transfer outcome was recorded concurrently", zap.String("from", t.State)) + return nil + } + if state == transferFailed && t.Kind == transferPayout { + if err := queries.ReleasePayout(ctx, database.ReleasePayoutParams{Amount: t.Amount, ExecutorID: t.ExecutorID, Currency: t.Currency}); err != nil { + return err + } + } + return tx.Commit() + }() + switch { + case err != nil: + logger.Error("Failed to record transfer outcome", zap.String("state", state), zap.String("error", err.Error())) + case state == t.State: + logger.Debug("Transfer is still open", zap.String("state", state), zap.String("detail", detail)) + case state == transferFailed || state == transferUnknown: + logger.Warn("Transfer did not complete", zap.String("state", state), zap.String("detail", detail)) + default: + logger.Info("Recorded transfer", zap.String("state", state)) + } +} + +// ReconcileTransfers looks up at most limit open transfers by digest: sent and +// unknown transfers and reserved transfers older than staleReservation. A +// transfer is confirmed only when the chain executed it successfully and the +// receiver's credit was verified; executed with failure status it is failed; +// a transfer the chain does not know is submitted again with its stored signed +// transaction, which has the same digest and cannot execute twice. Anything +// inconclusive leaves the transfer unknown with the last check in its detail. +func (p *PaymentHandler) ReconcileTransfers(ctx context.Context, limit int) error { + if p.chainDisabled() { + return fmt.Errorf("reconcile transfers: %w", ErrPaymentsDisabled) + } + open, err := database.New(p.db).ListUnresolvedChainTransfers(ctx, database.ListUnresolvedChainTransfersParams{ + ReservedBefore: models.NewUTCTime(time.Now().Add(-staleReservation)), RowLimit: int64(limit), + }) + if err != nil { + return fmt.Errorf("list open transfers: %w", err) + } + for _, t := range open { + if ctx.Err() != nil { + break + } + p.reconcileTransfer(ctx, t) + } + return nil +} + +func (p *PaymentHandler) reconcileTransfer(ctx context.Context, t database.ChainTransfer) { + outcome, verified, err := p.sui.LookupTransfer(ctx, t.Digest, &sui.TransferExpectation{ + Receiver: t.Receiver, Amount: uint64(t.Amount), CoinType: sui.GetCoinType(t.Currency, p.cfg.Sui.Network), + }) + switch { + case err != nil: + p.recordTransfer(ctx, t, transferUnknown, t.Digest, "last check: lookup inconclusive: "+err.Error()) + case outcome == sui.TransferConfirmed && verified: + p.recordTransfer(ctx, t, transferConfirmed, t.Digest, "") + case outcome == sui.TransferConfirmed: + p.recordTransfer(ctx, t, transferUnknown, t.Digest, "last check: executed successfully, but the node returned no balance changes, so the receiver's credit could not be verified") + case outcome == sui.TransferFailed: + p.recordTransfer(ctx, t, transferFailed, t.Digest, "executed with failure status") + case outcome == sui.TransferNotFound && t.State != transferSent: + prepared, err := sui.RestorePreparedTransfer(t.Digest, t.SignedTransaction, t.Signature) + if err != nil { + p.recordTransfer(ctx, t, transferUnknown, t.Digest, "last check: not found, and the stored transaction cannot be submitted again: "+err.Error()) + return + } + state, detail := executionOutcome(p.submit(ctx, prepared)) + p.recordTransfer(context.WithoutCancel(ctx), t, state, prepared.Digest, detail) + case outcome == sui.TransferNotFound: + p.recordTransfer(ctx, t, transferUnknown, t.Digest, "last check: not found after the chain accepted it") + default: + p.recordTransfer(ctx, t, transferUnknown, t.Digest, fmt.Sprintf("last check: unexpected lookup outcome %q", outcome)) + } +} diff --git a/internal/dispatcher/payments/transfers_test.go b/internal/dispatcher/payments/transfers_test.go new file mode 100644 index 00000000..0e6786b8 --- /dev/null +++ b/internal/dispatcher/payments/transfers_test.go @@ -0,0 +1,787 @@ +// SPDX-License-Identifier: Apache-2.0 +// Copyright 2026 ETH Zurich + +package payments + +import ( + "bytes" + "context" + "database/sql" + "errors" + "fmt" + "strings" + "sync" + "testing" + "time" + + "github.com/netsec-ethz/debuglet/internal/dispatcher/database" + "github.com/netsec-ethz/debuglet/internal/dispatcher/models" + "github.com/netsec-ethz/debuglet/internal/dispatcher/payments/sui" + + "github.com/block-vision/sui-go-sdk/utils" +) + +const ( + testWallet = "0x00000000000000000000000000000000000000000000000000000000000000aa" + testBalance = int64(100) +) + +var errLostResponse = errors.New("connection reset while waiting for the response") + +// lookupResult is one scripted LookupTransfer answer. +type lookupResult struct { + outcome sui.TransferOutcome + verified bool + err error +} + +// transferChain scripts the transfer methods of fakeChain. A prepared +// transfer carries signed bytes with their real digest, so it can be stored +// and restored; ExecuteTransfer and LookupTransfer answer from per-call +// scripts. onPrepare and onExecute run during the chain call. +type transferChain struct { + *fakeChain + + smu sync.Mutex + prepares int + receiverOf map[string]string // digest -> receiver + executeErrs []error // consumed per call; nil once exhausted + executeFor map[string]error // by receiver, before executeErrs + executed [][]byte + lookups []lookupResult // consumed per call + expected []sui.TransferExpectation + onPrepare func() + onExecute func() + // blockExecute makes ExecuteTransfer wait until its context ends; + // executeDeadline records that context's deadline. + blockExecute bool + executeDeadline time.Time +} + +func (c *transferChain) PrepareTransfer(ctx context.Context, amount uint64, coinType string, receiver string) (*sui.PreparedTransfer, error) { + c.record("PrepareTransfer") + if c.onPrepare != nil { + c.onPrepare() + } + c.fakeChain.mu.Lock() + c.transferAmount, c.transferCoinType, c.transferAddress = amount, coinType, receiver + c.fakeChain.mu.Unlock() + c.smu.Lock() + defer c.smu.Unlock() + c.prepares++ + if c.transferErr != nil { + return nil, c.transferErr + } + txBytes := []byte(fmt.Sprintf("transfer %d: %d %s to %s", c.prepares, amount, coinType, receiver)) + digest, err := utils.GetTxDigestFromBytes(txBytes) + if err != nil { + return nil, err + } + if c.receiverOf == nil { + c.receiverOf = map[string]string{} + } + c.receiverOf[digest] = receiver + return sui.RestorePreparedTransfer(digest, txBytes, fmt.Sprintf("signature %d", c.prepares)) +} + +func (c *transferChain) ExecuteTransfer(ctx context.Context, prepared *sui.PreparedTransfer) error { + c.record("ExecuteTransfer") + if c.onExecute != nil { + c.onExecute() + } + c.smu.Lock() + defer c.smu.Unlock() + c.executeDeadline, _ = ctx.Deadline() + if c.blockExecute { + c.smu.Unlock() + <-ctx.Done() + c.smu.Lock() + return fmt.Errorf("execute transfer %s: %w", prepared.Digest, ctx.Err()) + } + c.executed = append(c.executed, bytes.Clone(prepared.Bytes())) + if err, ok := c.executeFor[c.receiverOf[prepared.Digest]]; ok { + return err + } + if len(c.executeErrs) == 0 { + return nil + } + err := c.executeErrs[0] + c.executeErrs = c.executeErrs[1:] + return err +} + +func (c *transferChain) LookupTransfer(ctx context.Context, digest string, expect *sui.TransferExpectation) (sui.TransferOutcome, bool, error) { + c.record("LookupTransfer") + c.smu.Lock() + defer c.smu.Unlock() + c.expected = append(c.expected, *expect) + if len(c.lookups) == 0 { + return "", false, errors.New("no scripted lookup") + } + r := c.lookups[0] + c.lookups = c.lookups[1:] + return r.outcome, r.verified, r.err +} + +func (c *transferChain) Executed() [][]byte { + c.smu.Lock() + defer c.smu.Unlock() + return append([][]byte(nil), c.executed...) +} + +// transferFixture is an enabled handler on a real dispatcher database with a +// scripted chain. +func transferFixture(t *testing.T) (*sql.DB, *PaymentHandler, *transferChain) { + t.Helper() + db := newRefundDatabase(t) + h, rec := newEnabledHandler(t, db, enabledConfig(t)) + chain := &transferChain{fakeChain: rec.chain} + h.sui = chain + return db, h, chain +} + +// seedEarning credits executor with amount USDC and sets its payout wallet. +func seedEarning(t *testing.T, db *sql.DB, executor string, amount int64, wallet string) database.Earning { + t.Helper() + q := database.New(db) + if _, err := q.CreateEarnings(t.Context(), database.CreateEarningsParams{ExecutorID: executor, Currency: "USDC", SuiWalletAddress: wallet}); err != nil { + t.Fatal(err) + } + creditEarning(t, db, executor, amount) + return earningOf(t, db, executor) +} + +func creditEarning(t *testing.T, db *sql.DB, executor string, amount int64) { + t.Helper() + ctx, cancel := context.WithTimeout(context.Background(), waitBound) + defer cancel() + if err := database.New(db).AddEarnings(ctx, database.AddEarningsParams{Amount: amount, ExecutorID: executor, Currency: "USDC"}); err != nil { + t.Errorf("credit during the chain call (is a SQL transaction held across it?): %v", err) + } +} + +func earningOf(t *testing.T, db *sql.DB, executor string) database.Earning { + t.Helper() + e, err := database.New(db).GetEarningsIn(t.Context(), database.GetEarningsInParams{ExecutorID: executor, Currency: "USDC"}) + if err != nil { + t.Fatal(err) + } + return e +} + +// transferRows returns every chain_transfers row in insertion order. +func transferRows(t *testing.T, db *sql.DB) []database.ChainTransfer { + t.Helper() + rows, err := db.QueryContext(t.Context(), `SELECT id, kind, executor_id, transaction_id, order_id, amount, currency, receiver, +state, digest, signed_transaction, signature, detail FROM chain_transfers ORDER BY id`) + if err != nil { + t.Fatal(err) + } + defer rows.Close() + var out []database.ChainTransfer + for rows.Next() { + var r database.ChainTransfer + if err := rows.Scan(&r.ID, &r.Kind, &r.ExecutorID, &r.TransactionID, &r.OrderID, &r.Amount, &r.Currency, &r.Receiver, + &r.State, &r.Digest, &r.SignedTransaction, &r.Signature, &r.Detail); err != nil { + t.Fatal(err) + } + out = append(out, r) + } + if err := rows.Err(); err != nil { + t.Fatal(err) + } + return out +} + +// onlyTransfer requires exactly one transfer row and returns it. +func onlyTransfer(t *testing.T, db *sql.DB) database.ChainTransfer { + t.Helper() + rows := transferRows(t, db) + if len(rows) != 1 { + t.Fatalf("%d transfer rows, want 1: %+v", len(rows), rows) + } + return rows[0] +} + +func count(calls []string, name string) int { + n := 0 + for _, c := range calls { + if c == name { + n++ + } + } + return n +} + +// ---- payouts ---- + +// The payout reserves the balance it read. Earnings credited while the chain +// is called, which needs the database to be free, stay in the balance. +func TestPayoutReservesTheBalanceReadAndKeepsLaterEarnings(t *testing.T) { + db, h, chain := transferFixture(t) + earning := seedEarning(t, db, testExecutor, testBalance, testWallet) + chain.onPrepare = func() { creditEarning(t, db, testExecutor, 30) } + chain.onExecute = func() { creditEarning(t, db, testExecutor, 5) } + + if err := h.PayoutExecutor(earning, t.Context()); err != nil { + t.Fatalf("PayoutExecutor: %v", err) + } + if got := earningOf(t, db, testExecutor); got.CurrentBalance != 35 || got.TotalIncome != testBalance+35 { + t.Fatalf("earning after the payout %+v, want balance 35 of income %d", got, testBalance+35) + } + row := onlyTransfer(t, db) + if row.Kind != transferPayout || row.State != transferSent || row.Amount != testBalance || row.Receiver != testWallet || + row.ExecutorID != testExecutor || row.Currency != "USDC" || row.Digest == "" || row.Signature == "" { + t.Fatalf("payout row %+v", row) + } + if executed := chain.Executed(); len(executed) != 1 || !bytes.Equal(executed[0], row.SignedTransaction) { + t.Fatalf("the submitted transaction is not the stored one") + } + if chain.transferAmount != uint64(testBalance) || chain.transferCoinType != sui.GetCoinType("USDC", "testnet") || chain.transferAddress != testWallet { + t.Fatalf("prepared %d %s to %s", chain.transferAmount, chain.transferCoinType, chain.transferAddress) + } +} + +// An open payout blocks the next one for the same executor and currency. +func TestSecondPayoutIsRefusedWhileOneIsOpen(t *testing.T) { + db, h, chain := transferFixture(t) + earning := seedEarning(t, db, testExecutor, testBalance, testWallet) + chain.executeErrs = []error{errLostResponse} + if err := h.PayoutExecutor(earning, t.Context()); err != nil { + t.Fatalf("PayoutExecutor: %v", err) + } + if row := onlyTransfer(t, db); row.State != transferUnknown || !strings.Contains(row.Detail, errLostResponse.Error()) { + t.Fatalf("payout with a lost response: %+v", row) + } + + creditEarning(t, db, testExecutor, 20) + if err := h.PayoutExecutor(earningOf(t, db, testExecutor), t.Context()); !errors.Is(err, errPayoutNotReserved) { + t.Fatalf("second payout = %v, want %v", err, errPayoutNotReserved) + } + if got := earningOf(t, db, testExecutor).CurrentBalance; got != 20 { + t.Fatalf("balance %d after the refused payout, want 20", got) + } + if n := count(chain.Calls(), "ExecuteTransfer"); n != 1 || len(transferRows(t, db)) != 1 { + t.Fatalf("%d executions, %d rows after the refused payout", n, len(transferRows(t, db))) + } +} + +// A transfer that was not broadcast, or executed with failure status, moved +// nothing: the row is failed and the balance is restored. +func TestFailedPayoutRestoresTheBalance(t *testing.T) { + for name, executeErr := range map[string]error{ + "not broadcast": fmt.Errorf("%w: gas coins changed", sui.ErrNotBroadcast), + "failure status": fmt.Errorf("%w: InsufficientCoinBalance", sui.ErrTransferFailed), + } { + t.Run(name, func(t *testing.T) { + db, h, chain := transferFixture(t) + earning := seedEarning(t, db, testExecutor, testBalance, testWallet) + chain.executeErrs = []error{executeErr} + if err := h.PayoutExecutor(earning, t.Context()); err != nil { + t.Fatalf("PayoutExecutor: %v", err) + } + if row := onlyTransfer(t, db); row.State != transferFailed || row.Detail != executeErr.Error() { + t.Fatalf("failed payout row %+v", row) + } + if got := earningOf(t, db, testExecutor).CurrentBalance; got != testBalance { + t.Fatalf("balance %d after a failed payout, want %d", got, testBalance) + } + // A failed payout does not block the next one. + chain.executeErrs = nil + if err := h.PayoutExecutor(earningOf(t, db, testExecutor), t.Context()); err != nil { + t.Fatalf("payout after a failed one: %v", err) + } + if rows := transferRows(t, db); len(rows) != 2 || rows[1].State != transferSent { + t.Fatalf("rows after the second payout %+v", rows) + } + }) + } +} + +// A submission is bounded by executeTimeout; one that does not answer before +// its context ends is recorded as unknown, and the amount stays reserved. +func TestUnansweredSubmissionIsUnknown(t *testing.T) { + db, h, chain := transferFixture(t) + earning := seedEarning(t, db, testExecutor, testBalance, testWallet) + chain.blockExecute = true + ctx, cancel := context.WithTimeout(t.Context(), waitBound/2) + defer cancel() + start := time.Now() + if err := h.PayoutExecutor(earning, ctx); err != nil { + t.Fatalf("PayoutExecutor: %v", err) + } + if chain.executeDeadline.IsZero() || chain.executeDeadline.After(start.Add(executeTimeout)) { + t.Fatalf("submission deadline %v, want one within %v of %v", chain.executeDeadline, executeTimeout, start) + } + if row := onlyTransfer(t, db); row.State != transferUnknown || !strings.Contains(row.Detail, context.DeadlineExceeded.Error()) { + t.Fatalf("unanswered submission recorded as %+v", row) + } + if got := earningOf(t, db, testExecutor).CurrentBalance; got != 0 { + t.Fatalf("balance %d while the payout is unknown, want 0", got) + } +} + +// A preparation failure reserves nothing. +func TestUnpreparedPayoutReservesNothing(t *testing.T) { + db, h, chain := transferFixture(t) + earning := seedEarning(t, db, testExecutor, testBalance, testWallet) + chain.transferErr = fmt.Errorf("%w: insufficient USDC balance", sui.ErrNotBroadcast) + if err := h.PayoutExecutor(earning, t.Context()); !errors.Is(err, sui.ErrNotBroadcast) { + t.Fatalf("PayoutExecutor = %v", err) + } + if rows := transferRows(t, db); len(rows) != 0 || earningOf(t, db, testExecutor).CurrentBalance != testBalance { + t.Fatalf("an unprepared payout changed state: %+v", rows) + } +} + +// ---- reconciliation ---- + +// unknownPayout leaves one payout in state unknown. +func unknownPayout(t *testing.T) (*sql.DB, *PaymentHandler, *transferChain) { + t.Helper() + db, h, chain := transferFixture(t) + chain.executeErrs = []error{errLostResponse} + if err := h.PayoutExecutor(seedEarning(t, db, testExecutor, testBalance, testWallet), t.Context()); err != nil { + t.Fatalf("PayoutExecutor: %v", err) + } + if row := onlyTransfer(t, db); row.State != transferUnknown { + t.Fatalf("payout with a lost response: %+v", row) + } + return db, h, chain +} + +func reconcile(t *testing.T, h *PaymentHandler) { + t.Helper() + if err := h.ReconcileTransfers(t.Context(), 10); err != nil { + t.Fatalf("ReconcileTransfers: %v", err) + } +} + +// A lost response resolved as confirmed by lookup completes the transfer +// without sending it again; the lookup checks the intended credit. +func TestUnknownTransferConfirmedByLookupIsNotSentAgain(t *testing.T) { + db, h, chain := unknownPayout(t) + digest := onlyTransfer(t, db).Digest + chain.lookups = []lookupResult{{outcome: sui.TransferConfirmed, verified: true}} + reconcile(t, h) + + if row := onlyTransfer(t, db); row.State != transferConfirmed || row.Digest != digest { + t.Fatalf("reconciled row %+v", row) + } + if n := count(chain.Calls(), "ExecuteTransfer"); n != 1 { + t.Fatalf("%d executions, want the original one only", n) + } + want := sui.TransferExpectation{Receiver: testWallet, Amount: uint64(testBalance), CoinType: sui.GetCoinType("USDC", "testnet")} + if len(chain.expected) != 1 || chain.expected[0] != want { + t.Fatalf("lookup expectation %+v, want %+v", chain.expected, want) + } + if got := earningOf(t, db, testExecutor).CurrentBalance; got != 0 { + t.Fatalf("balance %d after a confirmed payout", got) + } + // A confirmed transfer is not looked up again. + reconcile(t, h) + if n := count(chain.Calls(), "LookupTransfer"); n != 1 { + t.Fatalf("%d lookups, want 1", n) + } +} + +// Inconclusive lookups keep the transfer unknown with the last check noted +// and the amount still reserved. +func TestInconclusiveLookupKeepsTheTransferUnknown(t *testing.T) { + for name, tc := range map[string]struct { + result lookupResult + detail string + }{ + "lookup error": {lookupResult{err: errors.New("node unavailable")}, "lookup inconclusive: node unavailable"}, + "unverified": {lookupResult{outcome: sui.TransferConfirmed}, "credit could not be verified"}, + } { + t.Run(name, func(t *testing.T) { + db, h, chain := unknownPayout(t) + chain.lookups = []lookupResult{tc.result} + reconcile(t, h) + if row := onlyTransfer(t, db); row.State != transferUnknown || !strings.Contains(row.Detail, tc.detail) { + t.Fatalf("row after an inconclusive lookup %+v, want detail %q", row, tc.detail) + } + if n := count(chain.Calls(), "ExecuteTransfer"); n != 1 { + t.Fatalf("%d executions after an inconclusive lookup", n) + } + if got := earningOf(t, db, testExecutor).CurrentBalance; got != 0 { + t.Fatalf("balance %d while the payout is unknown, want 0", got) + } + }) + } +} + +// A transfer the chain executed with failure status releases the payout. +func TestLookupFailedReleasesThePayout(t *testing.T) { + db, h, chain := unknownPayout(t) + chain.lookups = []lookupResult{{outcome: sui.TransferFailed}} + reconcile(t, h) + if row := onlyTransfer(t, db); row.State != transferFailed { + t.Fatalf("row %+v", row) + } + if got := earningOf(t, db, testExecutor).CurrentBalance; got != testBalance { + t.Fatalf("balance %d, want %d", got, testBalance) + } +} + +// A transfer the chain does not know is submitted again with the stored +// bytes, once, and the outcome is recorded as for the first submission. +func TestNotFoundResubmitsTheStoredTransactionOnce(t *testing.T) { + db, h, chain := unknownPayout(t) + stored := onlyTransfer(t, db) + chain.lookups = []lookupResult{{outcome: sui.TransferNotFound}, {outcome: sui.TransferConfirmed, verified: true}} + reconcile(t, h) + + executed := chain.Executed() + if len(executed) != 2 || !bytes.Equal(executed[1], stored.SignedTransaction) || !bytes.Equal(executed[0], executed[1]) { + t.Fatalf("resubmission did not send the stored transaction once: %d executions", len(executed)) + } + if row := onlyTransfer(t, db); row.State != transferSent || row.Digest != stored.Digest { + t.Fatalf("row after the resubmission %+v", row) + } + reconcile(t, h) + if row := onlyTransfer(t, db); row.State != transferConfirmed { + t.Fatalf("row after the second check %+v", row) + } + if n := len(chain.Executed()); n != 2 || chain.prepares != 1 { + t.Fatalf("%d executions and %d preparations, want 2 and 1", n, chain.prepares) + } +} + +// A stop between the reservation and the chain call leaves the transfer +// reserved. Reconciliation leaves a fresh reservation alone, and resolves an +// old one by its digest. +func TestReservedTransferLeftByAStopIsReconciled(t *testing.T) { + db, h, chain := transferFixture(t) + earning := seedEarning(t, db, testExecutor, testBalance, testWallet) + reserved, err := h.prepareAndReserve(t.Context(), database.ChainTransfer{ + Kind: transferPayout, ExecutorID: testExecutor, Amount: earning.CurrentBalance, Currency: "USDC", Receiver: testWallet, + }, func(q *database.Queries) error { + _, err := q.ReservePayout(t.Context(), database.ReservePayoutParams{Amount: earning.CurrentBalance, ExecutorID: testExecutor, Currency: "USDC"}) + return err + }) + if err != nil || reserved == nil { + t.Fatalf("prepareAndReserve: %v", err) + } + if row := onlyTransfer(t, db); row.State != transferReserved || earningOf(t, db, testExecutor).CurrentBalance != 0 { + t.Fatalf("reserved row %+v", row) + } + + reconcile(t, h) + if n := count(chain.Calls(), "LookupTransfer"); n != 0 { + t.Fatalf("a fresh reservation was looked up %d times", n) + } + + old := models.NewUTCTime(time.Now().Add(-2 * staleReservation)) + if _, err := db.ExecContext(t.Context(), "UPDATE chain_transfers SET updated_at = ?", old); err != nil { + t.Fatal(err) + } + chain.lookups = []lookupResult{{outcome: sui.TransferNotFound}} + reconcile(t, h) + row := onlyTransfer(t, db) + if row.State != transferSent || row.Digest != reserved.prepared.Digest { + t.Fatalf("row after reconciliation %+v", row) + } + if executed := chain.Executed(); len(executed) != 1 || !bytes.Equal(executed[0], row.SignedTransaction) { + t.Fatalf("the stored transaction was not submitted once") + } +} + +func TestReconcileTransfersDisabled(t *testing.T) { + h, rec := newDisabledHandler(t, newRefundDatabase(t), true, true) + if err := h.ReconcileTransfers(t.Context(), 10); !errors.Is(err, ErrPaymentsDisabled) { + t.Fatalf("ReconcileTransfers = %v, want %v", err, ErrPaymentsDisabled) + } + if calls := rec.chain.Calls(); len(calls) != 0 { + t.Fatalf("disabled reconciliation reached the chain: %v", calls) + } +} + +// ---- refunds ---- + +// seedUSDCTransaction stores a paid USDC transaction with one Outstanding +// order per price. +func seedUSDCTransaction(t *testing.T, db *sql.DB, prices ...int64) { + t.Helper() + q := database.New(db) + var total int64 + for _, p := range prices { + total += p + } + if _, err := q.CreateTransaction(t.Context(), database.CreateTransactionParams{ + ID: testTxID, AuthKey: "key", Price: total, Currency: "USDC", Method: "SUI", + ExpiresAt: models.NewUTCTime(time.Now().Add(time.Hour)), Status: int64(models.Paid), Hash: testHash, + }); err != nil { + t.Fatal(err) + } + for i, p := range prices { + if _, err := q.CreateDebugletOrder(t.Context(), database.CreateDebugletOrderParams{ + TransactionID: testTxID, OrderID: int64(i + 1), ExecutorID: testExecutor, Price: p, + Currency: "USDC", RefundAddress: testWallet, State: int64(models.Outstanding), + }); err != nil { + t.Fatal(err) + } + } +} + +// refundState returns the transaction status, the order states and the +// number of refund settlement rows. +func refundState(t *testing.T, db *sql.DB) (models.TransactionState, []models.TransactionState, int) { + t.Helper() + q := database.New(db) + tr, err := q.GetTransactionByID(t.Context(), testTxID) + if err != nil { + t.Fatal(err) + } + orders, err := q.GetTransactionOrders(t.Context(), testTxID) + if err != nil { + t.Fatal(err) + } + var states []models.TransactionState + for _, o := range orders { + states = append(states, models.TransactionState(o.State)) + } + var settlements int + if err := db.QueryRow("SELECT COUNT(*) FROM order_settlements WHERE kind = 'refund'").Scan(&settlements); err != nil { + t.Fatal(err) + } + return models.TransactionState(tr.Status), states, settlements +} + +// A transaction refund is one transfer of the total and one refund +// settlement per order. +func TestRefundTransactionIsOneTransferWithOneSettlementPerOrder(t *testing.T) { + db, h, chain := transferFixture(t) + seedUSDCTransaction(t, db, 7, 8) + if err := h.RefundTransaction(testTxID, t.Context()); err != nil { + t.Fatalf("RefundTransaction: %v", err) + } + status, orders, settlements := refundState(t, db) + if status != models.Refunded || fmt.Sprint(orders) != fmt.Sprint([]models.TransactionState{models.Refunded, models.Refunded}) || settlements != 2 { + t.Fatalf("after the refund: transaction %v, orders %v, %d settlements", status, orders, settlements) + } + row := onlyTransfer(t, db) + if row.Kind != transferRefund || row.Amount != 15 || row.Receiver != testWallet || row.TransactionID != testTxID || + row.OrderID.Valid || row.State != transferSent { + t.Fatalf("refund row %+v", row) + } + if n := count(chain.Calls(), "ExecuteTransfer"); n != 1 { + t.Fatalf("%d executions", n) + } +} + +// A refund whose transfer fails stays decided: the orders are Refunded and +// the owed amount is visible as a failed transfer. +func TestFailedRefundTransferKeepsTheDecision(t *testing.T) { + db, h, chain := transferFixture(t) + seedUSDCTransaction(t, db, 7, 8) + chain.executeErrs = []error{fmt.Errorf("%w: refused", sui.ErrTransferFailed)} + if err := h.RefundTransaction(testTxID, t.Context()); err != nil { + t.Fatalf("RefundTransaction: %v", err) + } + status, orders, settlements := refundState(t, db) + if status != models.Refunded || orders[0] != models.Refunded || orders[1] != models.Refunded || settlements != 2 { + t.Fatalf("after a failed transfer: transaction %v, orders %v, %d settlements", status, orders, settlements) + } + if row := onlyTransfer(t, db); row.State != transferFailed || row.Amount != 15 { + t.Fatalf("refund row %+v", row) + } +} + +// A repeated refund request finds the transaction refunded and sends nothing. +func TestDuplicateRefundRequestSendsNothing(t *testing.T) { + db, h, chain := transferFixture(t) + seedUSDCTransaction(t, db, 7) + if err := h.RefundTransaction(testTxID, t.Context()); err != nil { + t.Fatalf("RefundTransaction: %v", err) + } + for _, refund := range []func(string, context.Context) error{h.RefundTransaction, h.RefundUnadmittedTransaction} { + if err := refund(testTxID, t.Context()); err == nil || !strings.Contains(err.Error(), "has not been payed") { + t.Fatalf("repeated refund = %v", err) + } + } + if n := count(chain.Calls(), "ExecuteTransfer"); n != 1 || len(transferRows(t, db)) != 1 { + t.Fatalf("%d executions after repeated requests", n) + } +} + +// Only Outstanding orders are refunded: a credited order was paid with work. +func TestRefundTransactionSkipsSettledOrders(t *testing.T) { + db, h, _ := transferFixture(t) + seedUSDCTransaction(t, db, 7, 8) + if err := h.SetDebugletOrderComplete(&database.Debuglet{TransactionID: testTxID, OrderID: 2, ExecutorID: testExecutor}, t.Context()); err != nil { + t.Fatal(err) + } + if err := h.RefundTransaction(testTxID, t.Context()); err != nil { + t.Fatalf("RefundTransaction: %v", err) + } + if _, orders, settlements := refundState(t, db); orders[0] != models.Refunded || orders[1] != models.Credited || settlements != 1 { + t.Fatalf("orders %v, %d refund settlements", orders, settlements) + } + if row := onlyTransfer(t, db); row.Amount != 7 || !row.OrderID.Valid || row.OrderID.Int64 != 1 { + t.Fatalf("refund row %+v", row) + } +} + +// A failed USDC run is refunded through the same lifecycle. +func TestSettleTerminalOrderRefundsAFailedUSDCRun(t *testing.T) { + db, h, chain := transferFixture(t) + seedUSDCTransaction(t, db, 7) + run := &database.Debuglet{TransactionID: testTxID, OrderID: testOrderID, ExecutorID: testExecutor} + for range 2 { + if err := h.SettleTerminalOrder(t.Context(), run, 1); err != nil { + t.Fatalf("SettleTerminalOrder: %v", err) + } + } + if _, orders, settlements := refundState(t, db); orders[0] != models.Refunded || settlements != 1 { + t.Fatalf("orders %v, %d settlements", orders, settlements) + } + if row := onlyTransfer(t, db); row.Amount != 7 || row.Kind != transferRefund || row.State != transferSent { + t.Fatalf("refund row %+v", row) + } + if n := count(chain.Calls(), "ExecuteTransfer"); n != 1 { + t.Fatalf("%d executions", n) + } +} + +// RefundDebugletOrder refunds one USDC order to its refund address through +// the same lifecycle; the order's transfer row names the order. +func TestRefundDebugletOrderUSDC(t *testing.T) { + db, h, chain := transferFixture(t) + seedUSDCTransaction(t, db, 7) + if err := h.RefundDebugletOrder(&database.Debuglet{TransactionID: testTxID, OrderID: testOrderID, ExecutorID: testExecutor}, testWallet, t.Context()); err != nil { + t.Fatalf("RefundDebugletOrder: %v", err) + } + status, orders, settlements := refundState(t, db) + if status != models.Paid || orders[0] != models.Refunded || settlements != 1 { + t.Fatalf("transaction %v, orders %v, %d settlements", status, orders, settlements) + } + if row := onlyTransfer(t, db); row.Amount != 7 || row.Receiver != testWallet || row.OrderID.Int64 != testOrderID || row.State != transferSent { + t.Fatalf("refund row %+v", row) + } + if chain.transferCoinType != sui.GetCoinType("USDC", "testnet") || count(chain.Calls(), "ExecuteTransfer") != 1 { + t.Fatalf("prepared %s, calls %v", chain.transferCoinType, chain.Calls()) + } +} + +// A reservation that cannot be written leaves the decision untaken and +// submits nothing. +func TestRefusedReservationLeavesStateUntouched(t *testing.T) { + refuse := func(t *testing.T, db *sql.DB) { + t.Helper() + if _, err := db.ExecContext(t.Context(), `CREATE TRIGGER refuse_transfer BEFORE INSERT ON chain_transfers +BEGIN SELECT RAISE(ABORT, 'transfer refused'); END;`); err != nil { + t.Fatal(err) + } + } + t.Run("payout", func(t *testing.T) { + db, h, chain := transferFixture(t) + earning := seedEarning(t, db, testExecutor, testBalance, testWallet) + refuse(t, db) + if err := h.PayoutExecutor(earning, t.Context()); err == nil || !strings.Contains(err.Error(), "transfer refused") { + t.Fatalf("PayoutExecutor = %v", err) + } + if got := earningOf(t, db, testExecutor).CurrentBalance; got != testBalance { + t.Fatalf("balance %d after a refused reservation", got) + } + if n := count(chain.Calls(), "ExecuteTransfer"); n != 0 { + t.Fatalf("%d executions after a refused reservation", n) + } + }) + t.Run("refund", func(t *testing.T) { + db, h, chain := transferFixture(t) + seedUSDCTransaction(t, db, 7, 8) + refuse(t, db) + if err := h.RefundTransaction(testTxID, t.Context()); err == nil || !strings.Contains(err.Error(), "transfer refused") { + t.Fatalf("RefundTransaction = %v", err) + } + status, orders, settlements := refundState(t, db) + if status != models.Paid || orders[0] != models.Outstanding || orders[1] != models.Outstanding || settlements != 0 { + t.Fatalf("after a refused reservation: transaction %v, orders %v, %d settlements", status, orders, settlements) + } + if n := count(chain.Calls(), "ExecuteTransfer"); n != 0 { + t.Fatalf("%d executions after a refused reservation", n) + } + }) +} + +// ---- settlement passes ---- + +// A failed USDC run whose inline refund lost its response is decided: the +// order is Refunded and the transfer unknown with its signed transaction +// kept. A settlement pass then selects nothing for the order and reaches the +// chain backend zero times; reconciliation resolves the transfer. +func TestSettlementPassLeavesAnUnknownRefundToReconciliation(t *testing.T) { + for name, block := range map[string]bool{"no answer": true, "lost response": false} { + t.Run(name, func(t *testing.T) { + db, h, chain := transferFixture(t) + run := seedFailedUSDCRun(t, db) + ctx := t.Context() + if block { + chain.blockExecute = true + var cancel context.CancelFunc + ctx, cancel = context.WithTimeout(ctx, waitBound/2) + defer cancel() + } else { + chain.executeErrs = []error{errLostResponse} + } + if err := h.SettleTerminalOrder(ctx, &run, 3); err != nil { + t.Fatalf("SettleTerminalOrder: %v", err) + } + if got := orderState(t, db); got != models.Refunded { + t.Fatalf("order state %v after the refund was reserved, want %v", got, models.Refunded) + } + if row := onlyTransfer(t, db); row.State != transferUnknown || len(row.SignedTransaction) == 0 || row.Signature == "" { + t.Fatalf("refund row %+v", row) + } + calls := len(chain.Calls()) + for range 2 { + settled, failed, deferred, next, err := h.SettlePendingOrders(t.Context(), 0, 32) + if settled != 0 || failed != 0 || deferred != 0 || next != 0 || err != nil { + t.Fatalf("pass: settled %d, failed %d, deferred %d, next %d, error %v", settled, failed, deferred, next, err) + } + } + if got := chain.Calls()[calls:]; len(got) != 0 { + t.Fatalf("settlement passes reached the chain backend: %v", got) + } + chain.lookups = []lookupResult{{outcome: sui.TransferConfirmed, verified: true}} + reconcile(t, h) + if row := onlyTransfer(t, db); row.State != transferConfirmed { + t.Fatalf("reconciled refund row %+v", row) + } + if n, _ := settlementOf(t, db); n != 1 || count(chain.Calls(), "ExecuteTransfer") != 1 { + t.Fatalf("%d settlement rows, calls %v", n, chain.Calls()) + } + }) + } +} + +// A refund owed but never attempted (the process stopped before the inline +// attempt reserved anything) is sent by a settlement pass exactly once, with +// the signed transaction reserved before it is submitted. +func TestSettlementPassSendsAnOwedRefundOnce(t *testing.T) { + db, h, chain := transferFixture(t) + seedFailedUSDCRun(t, db) + chain.onExecute = func() { + rows := transferRows(t, db) + if len(rows) != 1 || rows[0].State != transferReserved || len(rows[0].SignedTransaction) == 0 || orderState(t, db) != models.Refunded { + t.Errorf("submitted before the refund was reserved: %+v", rows) + } + } + if settled, failed, deferred, _, err := h.SettlePendingOrders(t.Context(), 0, 32); settled != 1 || failed != 0 || deferred != 0 || err != nil { + t.Fatalf("first pass: settled %d, failed %d, deferred %d, error %v", settled, failed, deferred, err) + } + if settled, failed, deferred, _, err := h.SettlePendingOrders(t.Context(), 0, 32); settled != 0 || failed != 0 || deferred != 0 || err != nil { + t.Fatalf("second pass: settled %d, failed %d, deferred %d, error %v", settled, failed, deferred, err) + } + if row := onlyTransfer(t, db); row.State != transferSent || row.Amount != testPrice || row.Receiver != testRefund { + t.Fatalf("refund row %+v", row) + } + if n := count(chain.Calls(), "ExecuteTransfer"); n != 1 { + t.Fatalf("%d executions, want 1", n) + } + if n, row := settlementOf(t, db); n != 1 || row.Kind != settlementRefund { + t.Fatalf("%d settlement rows, first %+v", n, row) + } +} diff --git a/internal/dispatcher/settlement_sweep.go b/internal/dispatcher/settlement_sweep.go index f9c2dfba..2e47ca69 100644 --- a/internal/dispatcher/settlement_sweep.go +++ b/internal/dispatcher/settlement_sweep.go @@ -23,8 +23,9 @@ const ( // request and settlement is idempotent per order, so an order settled by the // terminal path meanwhile is left as it is. Each pass attempts at most // settlementSweepLimit orders, continuing after the last run the previous pass -// attempted and starting over once the listing is exhausted. Chain refunds are -// not retried here; a durable transfer record is to take them over. +// attempted and starting over once the listing is exhausted. A chain refund is +// sent from here only while no refund transfer covers the order; one that +// exists, whatever its outcome, is resolved by transfer reconciliation. func (d *Dispatcher) sweepPendingSettlements(last time.Time) time.Time { now := d.now() if !last.IsZero() && now.Sub(last) < settlementSweepInterval { diff --git a/internal/dispatcher/transport/api/policy_bounds_test.go b/internal/dispatcher/transport/api/policy_bounds_test.go index b87561f7..ce6336ad 100644 --- a/internal/dispatcher/transport/api/policy_bounds_test.go +++ b/internal/dispatcher/transport/api/policy_bounds_test.go @@ -9,7 +9,6 @@ import ( "testing" "time" - "github.com/DATA-DOG/go-sqlmock" "github.com/netsec-ethz/debuglet/internal/bitrate" "github.com/netsec-ethz/debuglet/internal/dispatcher/models" ) @@ -284,12 +283,6 @@ func TestAdmissionRejectsAWindowThatDoesNotFit(t *testing.T) { f.mock.ExpectQuery(modeGetTransactionOrdersQuery). WithArgs(modeChainTxID). WillReturnRows(modeOrderRows(modeChainTxID, "TEST", models.Outstanding)) - f.mock.ExpectExec("(?s)UPDATE transactions.*NOT EXISTS"). - WithArgs(int64(models.Refunded), modeChainTxID, int64(models.Paid)). - WillReturnResult(sqlmock.NewResult(0, 1)) - f.mock.ExpectQuery(modeUpdateOrderStateQuery). - WithArgs(int64(models.Refunded), modeChainTxID, modeOrderID). - WillReturnRows(modeOrderRows(modeChainTxID, "TEST", models.Refunded)) f.mock.ExpectRollback() rec := f.do(http.MethodPut, "/debuglet", modeSubmitBody(modeChainTxID, "", debuglets)) From 66fe62874eabf3652dd93d867a1c7e83b601ebb3 Mon Sep 17 00:00:00 2001 From: TheodorAdrienIsaak Mattli Date: Mon, 5 Oct 2026 14:08:58 +0200 Subject: [PATCH 2/2] Report the refund's outcome and bound reconciliation lookups After a refund decision was committed, the transfer lifecycle returned no outcome to its caller, so a buyer refused during maintenance was told the payment had been refunded even when the transfer failed or its outcome was unknown, and later requests repeated it. Return the recorded outcome and word the refusal as sent, pending confirmation, or recorded but not sent and owed; a repeated request reads the refund transfer's state. The order stays unspendable after the decision. A reconciliation lookup ran under the daemon's lifetime context on an SDK path with no deadline, so one lookup that never completed stopped every later row and all payouts. Bound each lookup; a timeout is recorded as an unknown outcome with the reservation kept, and the pass continues with the next row. GitHub issues #80 and #81. --- docs/operations/payments.md | 7 +- .../database/queries/settlement.sql | 6 ++ .../dispatcher/database/settlement.sql.go | 14 +++ internal/dispatcher/payments/payments.go | 88 +++++++++++---- .../payments/refund_admission_test.go | 8 +- internal/dispatcher/payments/transfers.go | 53 +++++---- .../dispatcher/payments/transfers_test.go | 102 +++++++++++++++++- .../transport/api/handlers_debuglet.go | 62 +++++++++-- .../transport/api/maintenance_test.go | 52 +++++++++ 9 files changed, 334 insertions(+), 58 deletions(-) diff --git a/docs/operations/payments.md b/docs/operations/payments.md index 4867afb9..cf143fcb 100644 --- a/docs/operations/payments.md +++ b/docs/operations/payments.md @@ -58,6 +58,9 @@ between the reservation and the submission): | Not found, row `sent` | `unknown` | | Lookup error, or a credit that does not match | `unknown` | +Each lookup waits at most 30 seconds; one that does not answer leaves the +transfer `unknown` and the pass continues with the next transfer. + A resubmission sends the identical signed bytes, which have the same digest, so the chain executes the transfer at most once. A transfer is never rebuilt with new bytes because a response was lost. @@ -71,7 +74,9 @@ is the time of the last check. executor's balance; the next payout pass tries again with a new transfer. - **`failed` refund**: the orders stay refunded and the amount is owed to the buyer's refund address. It is not retried automatically; the row is the - record of what is owed. + record of what is owed. A buyer whose paid order is refunded because + submissions are paused is told whether the refund was sent, is pending + confirmation, or could not be sent and is owed. - **`unknown`**: reconciliation keeps checking it every minute. A payout in this state blocks further payouts of that executor and currency, so the amount cannot be paid twice. If it stays `unknown`, for example because the diff --git a/internal/dispatcher/database/queries/settlement.sql b/internal/dispatcher/database/queries/settlement.sql index 074a4399..73a5933a 100644 --- a/internal/dispatcher/database/queries/settlement.sql +++ b/internal/dispatcher/database/queries/settlement.sql @@ -52,3 +52,9 @@ LIMIT sqlc.arg(row_limit); UPDATE transactions SET status = sqlc.arg(refunded) WHERE id = sqlc.arg(id) AND status = sqlc.arg(paid); + +-- name: GetLatestRefundTransferState :one +SELECT state FROM chain_transfers +WHERE kind = 'refund' AND transaction_id = ? +ORDER BY id DESC +LIMIT 1; diff --git a/internal/dispatcher/database/settlement.sql.go b/internal/dispatcher/database/settlement.sql.go index 834b4cd0..f4998c17 100644 --- a/internal/dispatcher/database/settlement.sql.go +++ b/internal/dispatcher/database/settlement.sql.go @@ -12,6 +12,20 @@ import ( "github.com/netsec-ethz/debuglet/internal/dispatcher/models" ) +const getLatestRefundTransferState = `-- name: GetLatestRefundTransferState :one +SELECT state FROM chain_transfers +WHERE kind = 'refund' AND transaction_id = ? +ORDER BY id DESC +LIMIT 1 +` + +func (q *Queries) GetLatestRefundTransferState(ctx context.Context, transactionID string) (string, error) { + row := q.db.QueryRowContext(ctx, getLatestRefundTransferState, transactionID) + var state string + err := row.Scan(&state) + return state, err +} + const getPaymentReceipt = `-- name: GetPaymentReceipt :one SELECT tx_digest, event_seq, nonce, disposition, amount, coin_type, receiver, checkpoint, observed_at, detail FROM payment_receipts WHERE tx_digest = ? AND event_seq = ? diff --git a/internal/dispatcher/payments/payments.go b/internal/dispatcher/payments/payments.go index bb3d0733..54553bc3 100644 --- a/internal/dispatcher/payments/payments.go +++ b/internal/dispatcher/payments/payments.go @@ -320,7 +320,7 @@ func (p *PaymentHandler) settleOrder(ctx context.Context, debuglet *database.Deb // before it calls the chain. tx.Rollback() order.ExecutorID = debuglet.ExecutorID - if err := p.refundOnChain(ctx, order.TransactionID, []database.DebugletOrder{order}, nil); !errors.Is(err, errAlreadySettled) { + if _, err := p.refundOnChain(ctx, order.TransactionID, []database.DebugletOrder{order}, nil); !errors.Is(err, errAlreadySettled) { return err } // Settled concurrently: an order leaves Outstanding only as Credited @@ -402,8 +402,9 @@ func recordSettlement(ctx context.Context, queries *database.Queries, order data // of extra, in the SQL transaction that reserves the transfer. If any order // is no longer Outstanding, nothing is written or sent and errAlreadySettled is // returned. A transfer that then fails or stays unknown leaves the refund -// decided: the amount is owed and visible on the transfer row. -func (p *PaymentHandler) refundOnChain(ctx context.Context, transactionID string, orders []database.DebugletOrder, extra func(*database.Queries) error) error { +// decided: the amount is owed and visible on the transfer row. The state +// recorded on that row is returned, empty when nothing was transferred. +func (p *PaymentHandler) refundOnChain(ctx context.Context, transactionID string, orders []database.DebugletOrder, extra func(*database.Queries) error) (string, error) { transfer := database.ChainTransfer{Kind: transferRefund, TransactionID: transactionID} for _, order := range orders { transfer.Amount += order.Price @@ -488,55 +489,101 @@ func (p *PaymentHandler) SettlePendingOrders(ctx context.Context, after int64, l // RefundTransaction retains transaction-wide refunds, including claimed orders. // A submission caller uses it only after its own admission committed. func (p *PaymentHandler) RefundTransaction(transactionId string, ctx context.Context) error { - return p.refundTransaction(transactionId, ctx, false) + _, err := p.refundTransaction(transactionId, ctx, false) + return err } // RefundUnadmittedTransaction refunds a paid intent only while none of its // orders has been claimed. Its SQL write excludes a concurrent admission before -// any refund is sent to the chain backend. -func (p *PaymentHandler) RefundUnadmittedTransaction(transactionId string, ctx context.Context) error { +// any refund is sent to the chain backend. An error means nothing was decided +// and the intent stays spendable; otherwise the outcome says what became of +// the money. +func (p *PaymentHandler) RefundUnadmittedTransaction(transactionId string, ctx context.Context) (RefundOutcome, error) { return p.refundTransaction(transactionId, ctx, true) } -func (p *PaymentHandler) refundTransaction(transactionId string, ctx context.Context, unadmittedOnly bool) error { +// RefundOutcome is what a decided refund did with the money. +type RefundOutcome int + +const ( + // RefundLocal: nothing was transferred on chain, as for a TEST refund. + RefundLocal RefundOutcome = iota + // RefundSent: the chain accepted the refund transfer. + RefundSent + // RefundPending: the refund transfer is recorded and its outcome is not + // known yet; reconciliation resolves it. + RefundPending + // RefundFailed: the refund transfer is recorded but could not be sent; the + // amount is owed to the refund address. + RefundFailed +) + +// refundOutcome maps the state of a refund transfer row; "" is no row. +func refundOutcome(state string) RefundOutcome { + switch state { + case transferSent, transferConfirmed: + return RefundSent + case transferReserved, transferUnknown: + return RefundPending + case transferFailed: + return RefundFailed + default: + return RefundLocal + } +} + +// RefundOutcomeOf reports the outcome of the newest refund transfer of a +// transaction, RefundLocal when it has none. +func (p *PaymentHandler) RefundOutcomeOf(ctx context.Context, transactionId string) (RefundOutcome, error) { + state, err := database.New(p.db).GetLatestRefundTransferState(ctx, transactionId) + if errors.Is(err, sql.ErrNoRows) { + return RefundLocal, nil + } + if err != nil { + return RefundLocal, err + } + return refundOutcome(state), nil +} + +func (p *PaymentHandler) refundTransaction(transactionId string, ctx context.Context, unadmittedOnly bool) (RefundOutcome, error) { // The transaction and its orders are read from one snapshot; the writes // are conditional and decide again when the refund is reserved. tx, err := p.db.BeginTx(ctx, nil) if err != nil { - return fmt.Errorf("Failed to begin transaction: %s", err.Error()) + return RefundLocal, fmt.Errorf("Failed to begin transaction: %s", err.Error()) } defer tx.Rollback() queries := database.New(p.db).WithTx(tx) transaction, err := queries.GetTransactionByID(ctx, transactionId) if err != nil { - return err + return RefundLocal, err } if transaction.Status != int64(models.Paid) { - return fmt.Errorf("Tried to refund transaction that has not been payed: %s", transactionId) + return RefundLocal, fmt.Errorf("Tried to refund transaction that has not been payed: %s", transactionId) } orders, err := queries.GetTransactionOrders(ctx, transactionId) if err != nil { - return err + return RefundLocal, err } // The currency and the refund address live on the order rows, so a paid // transaction that has none says nothing about where the money would go. // It is reported rather than indexed into. if len(orders) == 0 { - return fmt.Errorf("Transaction %s has no orders to refund", transactionId) + return RefundLocal, fmt.Errorf("Transaction %s has no orders to refund", transactionId) } currency := orders[0].Currency refundAddress := orders[0].RefundAddress if err := p.requireChain(currency, "refund transaction"); err != nil { - return err + return RefundLocal, err } var outstanding []database.DebugletOrder for _, order := range orders { if order.Currency != currency || order.RefundAddress != refundAddress { // should never happen because these values are always set together but let's guard anyway - return fmt.Errorf("Inconsistent orders in transaction %s", transactionId) + return RefundLocal, fmt.Errorf("Inconsistent orders in transaction %s", transactionId) } if unadmittedOnly && order.DebugletID.Valid { - return ErrTransactionClaimed + return RefundLocal, ErrTransactionClaimed } if order.State != int64(models.Outstanding) { // A credited order was paid for with work; a refunded one is owed already. @@ -546,7 +593,7 @@ func (p *PaymentHandler) refundTransaction(transactionId string, ctx context.Con outstanding = append(outstanding, order) } if currency != "USDC" { - return fmt.Errorf("Refunds are not supported for currency %s", currency) + return RefundLocal, fmt.Errorf("Refunds are not supported for currency %s", currency) } tx.Rollback() @@ -554,7 +601,7 @@ func (p *PaymentHandler) refundTransaction(transactionId string, ctx context.Con // admitted on the transaction's status, so a transaction left Paid after // its money went back would admit the same batch again and have the work // done a second time for a payment that no longer exists. - err = p.refundOnChain(ctx, transactionId, outstanding, func(queries *database.Queries) error { + state, err := p.refundOnChain(ctx, transactionId, outstanding, func(queries *database.Queries) error { if unadmittedOnly { changed, err := queries.RefundUnadmittedTransaction(ctx, database.RefundUnadmittedTransactionParams{ RefundedStatus: int64(models.Refunded), TransactionID: transactionId, PaidStatus: int64(models.Paid), @@ -579,9 +626,9 @@ func (p *PaymentHandler) refundTransaction(transactionId string, ctx context.Con return nil }) if errors.Is(err, errAlreadySettled) { - return fmt.Errorf("orders of transaction %s were settled during the refund; nothing was refunded", transactionId) + return RefundLocal, fmt.Errorf("orders of transaction %s were settled during the refund; nothing was refunded", transactionId) } - return err + return refundOutcome(state), err } func (p *PaymentHandler) CreateEarningsIfNotExists(execID string, currency string, wallet string, queries *database.Queries, ctx context.Context) error { @@ -631,7 +678,7 @@ func (p *PaymentHandler) PayoutExecutor(earning database.Earning, ctx context.Co Kind: transferPayout, ExecutorID: earning.ExecutorID, Amount: earning.CurrentBalance, Currency: earning.Currency, Receiver: earning.SuiWalletAddress, } - return p.sendTransfer(ctx, payout, func(queries *database.Queries) error { + _, err := p.sendTransfer(ctx, payout, func(queries *database.Queries) error { reserved, err := queries.ReservePayout(ctx, database.ReservePayoutParams{ Amount: earning.CurrentBalance, ExecutorID: earning.ExecutorID, Currency: earning.Currency, }) @@ -643,4 +690,5 @@ func (p *PaymentHandler) PayoutExecutor(earning database.Earning, ctx context.Co } return nil }) + return err } diff --git a/internal/dispatcher/payments/refund_admission_test.go b/internal/dispatcher/payments/refund_admission_test.go index 19434365..4ecb99c7 100644 --- a/internal/dispatcher/payments/refund_admission_test.go +++ b/internal/dispatcher/payments/refund_admission_test.go @@ -78,7 +78,7 @@ func TestUnadmittedRefundAndAdmissionHaveOneWinner(t *testing.T) { if claimed, err := claimRefundOrder(t.Context(), db); err != nil || !claimed { t.Fatalf("claim: %t, %v", claimed, err) } - if err := h.RefundUnadmittedTransaction(testTxID, t.Context()); !errors.Is(err, ErrTransactionClaimed) { + if _, err := h.RefundUnadmittedTransaction(testTxID, t.Context()); !errors.Is(err, ErrTransactionClaimed) { t.Fatalf("refund admitted work: %v", err) } if got := orderState(t, db); got != models.Outstanding || len(chain.Calls()) != 0 { @@ -94,7 +94,7 @@ func TestUnadmittedRefundAndAdmissionHaveOneWinner(t *testing.T) { }) t.Run("refund before stale paid preflight", func(t *testing.T) { db, h, chain := admissionRefundFixture(t, "USDC") - if err := h.RefundUnadmittedTransaction(testTxID, t.Context()); err != nil { + if _, err := h.RefundUnadmittedTransaction(testTxID, t.Context()); err != nil { t.Fatal(err) } for _, latePayment := range []bool{false, true} { @@ -123,7 +123,7 @@ func TestUnadmittedRefundFailureLeavesIntentSpendable(t *testing.T) { t.Run(currency, func(t *testing.T) { db, h, chain := admissionRefundFixture(t, currency) chain.transferErr = errors.New("fixture refund rejected") - if err := h.RefundUnadmittedTransaction(testTxID, t.Context()); err == nil { + if _, err := h.RefundUnadmittedTransaction(testTxID, t.Context()); err == nil { t.Fatal("unsupported or rejected refund succeeded") } transaction, err := database.New(db).GetTransactionByID(t.Context(), testTxID) @@ -161,7 +161,7 @@ func TestUnadmittedRefundRacesAdmissionOnTwoHandles(t *testing.T) { claimed, err = claimRefundOrder(t.Context(), other) return err }, - func() error { return h.RefundUnadmittedTransaction(testTxID, t.Context()) }, + func() error { _, err := h.RefundUnadmittedTransaction(testTxID, t.Context()); return err }, }) for _, err := range errs { if err != nil && !errors.Is(err, ErrTransactionClaimed) && !strings.Contains(err.Error(), "database is locked") { diff --git a/internal/dispatcher/payments/transfers.go b/internal/dispatcher/payments/transfers.go index 1f7e29e0..ecbf028c 100644 --- a/internal/dispatcher/payments/transfers.go +++ b/internal/dispatcher/payments/transfers.go @@ -42,6 +42,11 @@ const staleReservation = 10 * time.Minute // recorded. const executeTimeout = 2 * time.Minute +// chainReadTimeout bounds one chain read: the preparation of a transfer and +// each reconciliation lookup. A lookup that does not answer in time leaves the +// transfer unknown; it never means that the money was not sent. +var chainReadTimeout = 30 * time.Second + var ( // errAlreadySettled is returned by a transfer decision that finds the // decision already taken. Nothing is reserved and nothing is sent. @@ -64,15 +69,15 @@ type reservedTransfer struct { // transfer pays for) and reserves the transfer row with the signed // transaction; then the transfer is submitted and its outcome recorded. An // error means the decision was not taken and nothing was sent. Once the -// decision is committed, a failed or unknown transfer is recorded on its row -// and logged, and nil is returned. -func (p *PaymentHandler) sendTransfer(ctx context.Context, transfer database.ChainTransfer, decide func(*database.Queries) error) error { +// decision is committed, the error is nil and state is the outcome recorded on +// the transfer row (sent, failed or unknown), or empty when nothing was to be +// transferred. +func (p *PaymentHandler) sendTransfer(ctx context.Context, transfer database.ChainTransfer, decide func(*database.Queries) error) (state string, err error) { reserved, err := p.prepareAndReserve(ctx, transfer, decide) if err != nil || reserved == nil { - return err + return "", err } - p.executeReserved(ctx, reserved) - return nil + return p.executeReserved(ctx, reserved), nil } // prepareAndReserve prepares the transfer, then commits decide together with @@ -81,8 +86,10 @@ func (p *PaymentHandler) sendTransfer(ctx context.Context, transfer database.Cha func (p *PaymentHandler) prepareAndReserve(ctx context.Context, transfer database.ChainTransfer, decide func(*database.Queries) error) (*reservedTransfer, error) { var prepared *sui.PreparedTransfer if transfer.Amount > 0 { + readCtx, cancel := context.WithTimeout(ctx, chainReadTimeout) var err error - prepared, err = p.sui.PrepareTransfer(ctx, uint64(transfer.Amount), sui.GetCoinType(transfer.Currency, p.cfg.Sui.Network), transfer.Receiver) + prepared, err = p.sui.PrepareTransfer(readCtx, uint64(transfer.Amount), sui.GetCoinType(transfer.Currency, p.cfg.Sui.Network), transfer.Receiver) + cancel() if err != nil { return nil, fmt.Errorf("prepare %s transfer: %w", transfer.Kind, err) } @@ -122,10 +129,12 @@ func (p *PaymentHandler) prepareAndReserve(ctx context.Context, transfer databas // executeReserved submits a reserved transfer and records the outcome. The // outcome is recorded even when ctx ended during the call; if it cannot be -// recorded, the row stays reserved and reconciliation resolves it. -func (p *PaymentHandler) executeReserved(ctx context.Context, reserved *reservedTransfer) { +// recorded, the row stays reserved and reconciliation resolves it. It returns +// the outcome's state. +func (p *PaymentHandler) executeReserved(ctx context.Context, reserved *reservedTransfer) string { state, detail := executionOutcome(p.submit(ctx, reserved.prepared)) p.recordTransfer(context.WithoutCancel(ctx), reserved.row, state, reserved.prepared.Digest, detail) + return state } // submit runs ExecuteTransfer bounded by executeTimeout. @@ -221,30 +230,38 @@ func (p *PaymentHandler) ReconcileTransfers(ctx context.Context, limit int) erro return nil } +// reconcileTransfer looks one transfer up within chainReadTimeout and records +// what the lookup found. The outcome is recorded even when ctx ended. func (p *PaymentHandler) reconcileTransfer(ctx context.Context, t database.ChainTransfer) { - outcome, verified, err := p.sui.LookupTransfer(ctx, t.Digest, &sui.TransferExpectation{ + lookupCtx, cancel := context.WithTimeout(ctx, chainReadTimeout) + outcome, verified, err := p.sui.LookupTransfer(lookupCtx, t.Digest, &sui.TransferExpectation{ Receiver: t.Receiver, Amount: uint64(t.Amount), CoinType: sui.GetCoinType(t.Currency, p.cfg.Sui.Network), }) + timedOut := errors.Is(lookupCtx.Err(), context.DeadlineExceeded) && ctx.Err() == nil + cancel() + record := context.WithoutCancel(ctx) switch { + case err != nil && timedOut: + p.recordTransfer(record, t, transferUnknown, t.Digest, "last check: lookup timed out") case err != nil: - p.recordTransfer(ctx, t, transferUnknown, t.Digest, "last check: lookup inconclusive: "+err.Error()) + p.recordTransfer(record, t, transferUnknown, t.Digest, "last check: lookup inconclusive: "+err.Error()) case outcome == sui.TransferConfirmed && verified: - p.recordTransfer(ctx, t, transferConfirmed, t.Digest, "") + p.recordTransfer(record, t, transferConfirmed, t.Digest, "") case outcome == sui.TransferConfirmed: - p.recordTransfer(ctx, t, transferUnknown, t.Digest, "last check: executed successfully, but the node returned no balance changes, so the receiver's credit could not be verified") + p.recordTransfer(record, t, transferUnknown, t.Digest, "last check: executed successfully, but the node returned no balance changes, so the receiver's credit could not be verified") case outcome == sui.TransferFailed: - p.recordTransfer(ctx, t, transferFailed, t.Digest, "executed with failure status") + p.recordTransfer(record, t, transferFailed, t.Digest, "executed with failure status") case outcome == sui.TransferNotFound && t.State != transferSent: prepared, err := sui.RestorePreparedTransfer(t.Digest, t.SignedTransaction, t.Signature) if err != nil { - p.recordTransfer(ctx, t, transferUnknown, t.Digest, "last check: not found, and the stored transaction cannot be submitted again: "+err.Error()) + p.recordTransfer(record, t, transferUnknown, t.Digest, "last check: not found, and the stored transaction cannot be submitted again: "+err.Error()) return } state, detail := executionOutcome(p.submit(ctx, prepared)) - p.recordTransfer(context.WithoutCancel(ctx), t, state, prepared.Digest, detail) + p.recordTransfer(record, t, state, prepared.Digest, detail) case outcome == sui.TransferNotFound: - p.recordTransfer(ctx, t, transferUnknown, t.Digest, "last check: not found after the chain accepted it") + p.recordTransfer(record, t, transferUnknown, t.Digest, "last check: not found after the chain accepted it") default: - p.recordTransfer(ctx, t, transferUnknown, t.Digest, fmt.Sprintf("last check: unexpected lookup outcome %q", outcome)) + p.recordTransfer(record, t, transferUnknown, t.Digest, fmt.Sprintf("last check: unexpected lookup outcome %q", outcome)) } } diff --git a/internal/dispatcher/payments/transfers_test.go b/internal/dispatcher/payments/transfers_test.go index 0e6786b8..a4e95fea 100644 --- a/internal/dispatcher/payments/transfers_test.go +++ b/internal/dispatcher/payments/transfers_test.go @@ -321,6 +321,93 @@ func TestUnansweredSubmissionIsUnknown(t *testing.T) { } } +// The outcome a decided refund returns, and the one later read from its +// transfer row, follow the recorded state; a TEST refund has no row. +func TestRefundOutcomeFollowsTheRecordedTransfer(t *testing.T) { + for name, tc := range map[string]struct { + executeErr error + want RefundOutcome + }{ + "sent": {nil, RefundSent}, + "unknown": {errLostResponse, RefundPending}, + "failed": {fmt.Errorf("%w: refused", sui.ErrNotBroadcast), RefundFailed}, + } { + t.Run(name, func(t *testing.T) { + db, h, chain := transferFixture(t) + seedUSDCTransaction(t, db, 7) + chain.executeErrs = []error{tc.executeErr} + if outcome, err := h.RefundUnadmittedTransaction(testTxID, t.Context()); outcome != tc.want || err != nil { + t.Fatalf("RefundUnadmittedTransaction = (%v, %v), want (%v, nil)", outcome, err, tc.want) + } + if outcome, err := h.RefundOutcomeOf(t.Context(), testTxID); outcome != tc.want || err != nil { + t.Fatalf("RefundOutcomeOf = (%v, %v), want (%v, nil)", outcome, err, tc.want) + } + }) + } + _, h, _ := transferFixture(t) + if outcome, err := h.RefundOutcomeOf(t.Context(), "no-transfer"); outcome != RefundLocal || err != nil { + t.Fatalf("RefundOutcomeOf without a transfer = (%v, %v)", outcome, err) + } +} + +// hangingLookup answers LookupTransfer for the digest hang only when its +// context ends, and confirms every other digest. +type hangingLookup struct { + *transferChain + hang string +} + +func (c hangingLookup) LookupTransfer(ctx context.Context, digest string, expect *sui.TransferExpectation) (sui.TransferOutcome, bool, error) { + c.record("LookupTransfer") + if digest == c.hang { + <-ctx.Done() + return "", false, ctx.Err() + } + return sui.TransferConfirmed, true, nil +} + +// A lookup that never answers is cut off at chainReadTimeout: its transfer is +// recorded unknown with the reservation kept, and later transfers are still +// reconciled in the same pass. +func TestHungLookupDoesNotStopReconciliation(t *testing.T) { + const bound = 200 * time.Millisecond + saved := chainReadTimeout + chainReadTimeout = bound + t.Cleanup(func() { chainReadTimeout = saved }) + + db, h, chain := transferFixture(t) + chain.executeErrs = []error{errLostResponse, errLostResponse} + for _, executor := range []string{"exec-a", "exec-b"} { + if err := h.PayoutExecutor(seedEarning(t, db, executor, testBalance, testWallet), t.Context()); err != nil { + t.Fatalf("PayoutExecutor(%s): %v", executor, err) + } + } + rows := transferRows(t, db) + if len(rows) != 2 || rows[0].State != transferUnknown || rows[1].State != transferUnknown { + t.Fatalf("payouts with lost responses: %+v", rows) + } + h.sui = hangingLookup{transferChain: chain, hang: rows[0].Digest} + + start := time.Now() + reconcile(t, h) + if elapsed := time.Since(start); elapsed < bound || elapsed > bound+waitBound { + t.Fatalf("pass took %v with a lookup bound of %v", elapsed, bound) + } + byExecutor := map[string]database.ChainTransfer{} + for _, row := range transferRows(t, db) { + byExecutor[row.ExecutorID] = row + } + if row := byExecutor["exec-a"]; row.State != transferUnknown || row.Detail != "last check: lookup timed out" { + t.Fatalf("hung lookup recorded %+v", row) + } + if row := byExecutor["exec-b"]; row.State != transferConfirmed { + t.Fatalf("later transfer recorded %+v", row) + } + if got := earningOf(t, db, "exec-a").CurrentBalance; got != 0 { + t.Fatalf("balance %d after a timed-out lookup, want the reservation kept", got) + } +} + // A preparation failure reserves nothing. func TestUnpreparedPayoutReservesNothing(t *testing.T) { db, h, chain := transferFixture(t) @@ -572,13 +659,16 @@ func TestRefundTransactionIsOneTransferWithOneSettlementPerOrder(t *testing.T) { } // A refund whose transfer fails stays decided: the orders are Refunded and -// the owed amount is visible as a failed transfer. +// the owed amount is visible as a failed transfer, and the caller is told so. func TestFailedRefundTransferKeepsTheDecision(t *testing.T) { db, h, chain := transferFixture(t) seedUSDCTransaction(t, db, 7, 8) chain.executeErrs = []error{fmt.Errorf("%w: refused", sui.ErrTransferFailed)} - if err := h.RefundTransaction(testTxID, t.Context()); err != nil { - t.Fatalf("RefundTransaction: %v", err) + if outcome, err := h.RefundUnadmittedTransaction(testTxID, t.Context()); outcome != RefundFailed || err != nil { + t.Fatalf("RefundUnadmittedTransaction = (%v, %v), want (%v, nil)", outcome, err, RefundFailed) + } + if outcome, err := h.RefundOutcomeOf(t.Context(), testTxID); outcome != RefundFailed || err != nil { + t.Fatalf("RefundOutcomeOf = (%v, %v), want (%v, nil)", outcome, err, RefundFailed) } status, orders, settlements := refundState(t, db) if status != models.Refunded || orders[0] != models.Refunded || orders[1] != models.Refunded || settlements != 2 { @@ -596,7 +686,11 @@ func TestDuplicateRefundRequestSendsNothing(t *testing.T) { if err := h.RefundTransaction(testTxID, t.Context()); err != nil { t.Fatalf("RefundTransaction: %v", err) } - for _, refund := range []func(string, context.Context) error{h.RefundTransaction, h.RefundUnadmittedTransaction} { + unadmitted := func(id string, ctx context.Context) error { + _, err := h.RefundUnadmittedTransaction(id, ctx) + return err + } + for _, refund := range []func(string, context.Context) error{h.RefundTransaction, unadmitted} { if err := refund(testTxID, t.Context()); err == nil || !strings.Contains(err.Error(), "has not been payed") { t.Fatalf("repeated refund = %v", err) } diff --git a/internal/dispatcher/transport/api/handlers_debuglet.go b/internal/dispatcher/transport/api/handlers_debuglet.go index 713fdac4..f9d7a621 100644 --- a/internal/dispatcher/transport/api/handlers_debuglet.go +++ b/internal/dispatcher/transport/api/handlers_debuglet.go @@ -4,6 +4,7 @@ package api import ( + "context" "database/sql" "encoding/base64" "errors" @@ -142,7 +143,10 @@ func (h *Handler) PutDebuglets(c echo.Context) error { // those runs may be executing or credited, and refunding would pay // for the same work twice. if !errors.Is(err, dispatcher.ErrPaymentInUse) { - refund := h.dispatcher.Payment.RefundUnadmittedTransaction + refund := func(id string, ctx context.Context) error { + _, err := h.dispatcher.Payment.RefundUnadmittedTransaction(id, ctx) + return err + } if errors.Is(err, dispatcher.ErrAdmissionCommitted) { refund = h.dispatcher.Payment.RefundTransaction } @@ -196,10 +200,12 @@ func (h *Handler) PutDebuglets(c echo.Context) error { // what it decides is what happens to a payment order that was already paid // when admission stopped. Such an order is refunded here, because the batch it // paid for is not being accepted, and a refunded order is spent: the same -// batch is refused on the transaction's status from then on. A refund that -// cannot be performed is not silently dropped either: the refusal then says -// the order is still paid, so the same batch can be submitted again once -// admission resumes. +// batch is refused on the transaction's status from then on. The refusal says +// what became of the money: sent, recorded and pending confirmation, or +// recorded but not sent and owed, on this request and on any later one. A +// refund that cannot be decided is not silently dropped either: the refusal +// then says the order is still paid, so the same batch can be submitted again +// once admission resumes. // // The order is read only to answer that question, and only an order the caller // has proved it may spend is reported on at all. An unknown transaction, @@ -215,19 +221,53 @@ func (h *Handler) refuseForMaintenance(c echo.Context, established *caller, req switch { case !spendable: case tx.Status == int64(models.Refunded): - message += "; this payment order was refunded and cannot be spent again" + message += h.recordedRefund(ctx, req.TransactionId) case tx.Status == int64(models.Paid): - if refundErr := h.dispatcher.Payment.RefundUnadmittedTransaction(req.TransactionId, ctx); refundErr != nil { - h.logger.Warn("Failed to refund transaction", zap.String("ID", daemonlog.Identifier(req.TransactionId))) - h.logger.Debug("Private request diagnostic", zap.String("ID", daemonlog.Identifier(req.TransactionId)), zap.String("error", requestDiagnostic(c, refundErr, req.AuthKey))) - message += "; this payment order is paid and was not refunded, so it stays paid and the same batch can be submitted again once admission resumes" + outcome, refundErr := h.dispatcher.Payment.RefundUnadmittedTransaction(req.TransactionId, ctx) + if refundErr == nil { + message += refundStatement(outcome, true) + break + } + h.logger.Warn("Failed to refund transaction", zap.String("ID", daemonlog.Identifier(req.TransactionId))) + h.logger.Debug("Private request diagnostic", zap.String("ID", daemonlog.Identifier(req.TransactionId)), zap.String("error", requestDiagnostic(c, refundErr, req.AuthKey))) + // This request decided nothing, but a concurrent one may have. + if now, err := h.dispatcher.Payment.GetTransaction(ctx, req.TransactionId); err == nil && now.Status == int64(models.Refunded) { + message += h.recordedRefund(ctx, req.TransactionId) } else { - message += "; this payment order was paid and has been refunded, so it cannot be spent again" + message += "; this payment order is paid and was not refunded, so it stays paid and the same batch can be submitted again once admission resumes" } } return apiError(http.StatusServiceUnavailable, CodeUnavailable, message) } +// recordedRefund words the refund of an already refunded transaction from its +// recorded refund transfer. +func (h *Handler) recordedRefund(ctx context.Context, transactionID string) string { + outcome, err := h.dispatcher.Payment.RefundOutcomeOf(ctx, transactionID) + if err != nil { + h.logger.Debug("Failed to read the refund outcome", zap.String("ID", daemonlog.Identifier(transactionID)), zap.String("error", daemonlog.Diagnostic(err))) + return "; this payment order has a recorded refund and cannot be spent again" + } + return refundStatement(outcome, false) +} + +// refundStatement says what a decided refund did with the buyer's money. +// initial marks the request that decided it. +func refundStatement(outcome payments.RefundOutcome, initial bool) string { + switch outcome { + case payments.RefundSent: + return "; this payment order was paid and its refund has been sent, so it cannot be spent again" + case payments.RefundPending: + return "; this payment order was paid; its refund was recorded and is pending confirmation, so it cannot be spent again" + case payments.RefundFailed: + return "; this payment order was paid; its refund was recorded but could not be sent and is owed to the refund address, so it cannot be spent again" + } + if initial { + return "; this payment order was paid and has been refunded, so it cannot be spent again" + } + return "; this payment order was refunded and cannot be spent again" +} + // GET /debuglet/:id/logs func (h *Handler) GetDebugletLogs(c echo.Context) error { debugletID := c.Param("id") diff --git a/internal/dispatcher/transport/api/maintenance_test.go b/internal/dispatcher/transport/api/maintenance_test.go index bb494439..ad1abe7a 100644 --- a/internal/dispatcher/transport/api/maintenance_test.go +++ b/internal/dispatcher/transport/api/maintenance_test.go @@ -344,3 +344,55 @@ func TestAnUnauthenticatedRequestIsRefusedBeforeMaintenanceIsConsulted(t *testin }) } } + +// A refunded order is described by what its refund did with the money: a +// refund transfer that was sent, is pending, or could not be sent and is owed +// is said to be so on every later request; a TEST refund, which has no +// transfer, keeps its wording. +func TestARefundedOrderIsDescribedByItsRefundTransfer(t *testing.T) { + const prefix = "dispatcher is in maintenance: submission admission is stopped (planned upgrade)" + for state, want := range map[string]string{ + "sent": "; this payment order was paid and its refund has been sent, so it cannot be spent again", + "confirmed": "; this payment order was paid and its refund has been sent, so it cannot be spent again", + "reserved": "; this payment order was paid; its refund was recorded and is pending confirmation, so it cannot be spent again", + "unknown": "; this payment order was paid; its refund was recorded and is pending confirmation, so it cannot be spent again", + "failed": "; this payment order was paid; its refund was recorded but could not be sent and is owed to the refund address, so it cannot be spent again", + "": "; this payment order was refunded and cannot be spent again", + } { + t.Run(state, func(t *testing.T) { + f := newMaintenanceFixture(t) + const id, key = "refunded-order", "auth-key" + batch := maintenanceBatch() + f.seedOrder(id, key, batch, models.Refunded) + if state != "" { + now := models.NewUTCTime(time.Now()) + if _, err := f.db.Exec(`INSERT INTO chain_transfers (kind, transaction_id, amount, currency, receiver, state, created_at, updated_at) +VALUES ('refund', ?, 1000, 'USDC', '0xaa', ?, ?, ?)`, id, state, now, now); err != nil { + t.Fatal(err) + } + } + f.pause("planned upgrade") + for range 2 { + status, refusal := f.put("/debuglet", SubmitDebugletsRequest{TransactionId: id, AuthKey: key, Debuglets: batch}) + if status != http.StatusServiceUnavailable || refusal.Message != prefix+want { + t.Fatalf("status %d, message %q; want %q", status, refusal.Message, prefix+want) + } + } + }) + } +} + +// The request that decides a refund words its outcome the same way; a refund +// that moved nothing on chain keeps the wording of a TEST refund. +func TestRefundStatementOfTheDecidingRequest(t *testing.T) { + for outcome, want := range map[payments.RefundOutcome]string{ + payments.RefundSent: "; this payment order was paid and its refund has been sent, so it cannot be spent again", + payments.RefundPending: "; this payment order was paid; its refund was recorded and is pending confirmation, so it cannot be spent again", + payments.RefundFailed: "; this payment order was paid; its refund was recorded but could not be sent and is owed to the refund address, so it cannot be spent again", + payments.RefundLocal: "; this payment order was paid and has been refunded, so it cannot be spent again", + } { + if got := refundStatement(outcome, true); got != want { + t.Fatalf("refundStatement(%v) = %q, want %q", outcome, got, want) + } + } +}