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..cf143fcb --- /dev/null +++ b/docs/operations/payments.md @@ -0,0 +1,84 @@ +# 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` | + +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. + +## 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. 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 + 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..73a5933a 100644 --- a/internal/dispatcher/database/queries/settlement.sql +++ b/internal/dispatcher/database/queries/settlement.sql @@ -15,3 +15,46 @@ 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); + +-- 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/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..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 = ? @@ -40,6 +54,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 +171,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 +241,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..54553bc3 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,87 @@ 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. 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 + 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 +460,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 +476,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 @@ -426,99 +489,146 @@ 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 - } - 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 RefundLocal, err } - 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) + return RefundLocal, 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 RefundLocal, 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 RefundLocal, 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 { + 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), + }) + 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 RefundLocal, 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 refundOutcome(state), err } func (p *PaymentHandler) CreateEarningsIfNotExists(execID string, currency string, wallet string, queries *database.Queries, ctx context.Context) error { @@ -556,11 +666,29 @@ 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, + } + _, 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, + }) + if err != nil { + return fmt.Errorf("reserve payout: %w", err) + } + if reserved != 1 { + return errPayoutNotReserved + } + return nil + }) + return err } 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..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 { @@ -88,13 +88,13 @@ 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") } }) 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} { @@ -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") } }) @@ -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") { @@ -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/transfers.go b/internal/dispatcher/payments/transfers.go new file mode 100644 index 00000000..ecbf028c --- /dev/null +++ b/internal/dispatcher/payments/transfers.go @@ -0,0 +1,267 @@ +// 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 + +// 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. + 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, 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 p.executeReserved(ctx, reserved), 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 { + readCtx, cancel := context.WithTimeout(ctx, chainReadTimeout) + var err error + 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) + } + } + + 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. 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. +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 +} + +// 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) { + 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(record, t, transferUnknown, t.Digest, "last check: lookup inconclusive: "+err.Error()) + case outcome == sui.TransferConfirmed && verified: + p.recordTransfer(record, t, transferConfirmed, t.Digest, "") + case outcome == sui.TransferConfirmed: + 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(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(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(record, t, state, prepared.Digest, detail) + case outcome == sui.TransferNotFound: + p.recordTransfer(record, t, transferUnknown, t.Digest, "last check: not found after the chain accepted it") + default: + 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 new file mode 100644 index 00000000..a4e95fea --- /dev/null +++ b/internal/dispatcher/payments/transfers_test.go @@ -0,0 +1,881 @@ +// 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) + } +} + +// 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) + 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, 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 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 { + 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) + } + 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) + } + } + 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/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) + } + } +} 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))