Skip to content

Commit 902e004

Browse files
bitprecipicebitprecipice
committed
Merge remote-tracking branch 'origin/main'
# Conflicts: # db_migrations.go
2 parents b27aacd + af8117e commit 902e004

11 files changed

Lines changed: 1007 additions & 102 deletions

‎controller/account_payment_controller.go‎

Lines changed: 60 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -305,36 +305,84 @@ func advancePayment(
305305
txResponseBodyBytes = txResult.ResponseBodyBytes
306306
status = tx.State
307307

308-
// Check the Circle Status of the payment
309-
// INITIATED, PENDING_RISK_SCREENING, DENIED, QUEUED, SENT, CONFIRMED, COMPLETE, FAILED, CANCELLED
308+
// Check the Circle status of the payment. Every non-terminal state stays
309+
// in retry; age is not a cancellation condition.
310310
switch strings.ToUpper(status) {
311-
case "INITIATED", "PENDING_RISK_SCREENING", "QUEUED", "SENT", "CONFIRMED":
311+
case "INITIATED", "PENDING_RISK_SCREENING", "CLEARED", "QUEUED":
312312
// check later
313313
return
314314

315+
case "SENT", "STUCK", "CONFIRMED":
316+
// Circle has assigned a chain transaction hash by SENT. Persist it
317+
// before terminal completion so a long-running retry remains
318+
// externally reconcilable and visible to the account holder.
319+
if tx.TxHash != "" {
320+
if err := model.UpdatePaymentProgress(
321+
clientSession.Ctx,
322+
payment.PaymentId,
323+
string(txResponseBodyBytes),
324+
tx.TxHash,
325+
); err != nil {
326+
returnErr = fmt.Errorf("[%s]Payment progress error = %s", payment.PaymentId, err)
327+
}
328+
}
329+
return
330+
315331
case "DENIED", "FAILED":
316332
returnErr = fmt.Errorf("[%s]error = %s", payment.PaymentId, status)
317333
// remove the payment record so it can be recreated
318-
model.RemovePaymentRecord(
334+
if err := model.RemovePaymentRecord(
319335
clientSession.Ctx,
320336
payment.PaymentId,
321-
)
337+
); err != nil {
338+
returnErr = fmt.Errorf("[%s]error = %s; payment reset error = %s", payment.PaymentId, status, err)
339+
}
322340
return
323341

324342
case "CANCELLED":
325-
model.CancelPayment(clientSession.Ctx, payment.PaymentId)
343+
// A chain hash and CANCELLED are contradictory. Preserve both pieces
344+
// of evidence and keep reconciling; releasing the sweeps here could
345+
// pay an already-broadcast transaction twice.
346+
if tx.TxHash != "" || payment.TxHash != nil {
347+
txHash := tx.TxHash
348+
if txHash == "" {
349+
txHash = *payment.TxHash
350+
}
351+
if err := model.UpdatePaymentProgress(
352+
clientSession.Ctx,
353+
payment.PaymentId,
354+
string(txResponseBodyBytes),
355+
txHash,
356+
); err != nil {
357+
returnErr = fmt.Errorf("[%s]Payment progress error = %s", payment.PaymentId, err)
358+
return
359+
}
360+
returnErr = fmt.Errorf("[%s]Circle returned CANCELLED for a payment with transaction hash %s", payment.PaymentId, txHash)
361+
return
362+
}
363+
if err := model.CancelPaymentAfterProcessorCancellation(
364+
clientSession.Ctx,
365+
payment.PaymentId,
366+
string(txResponseBodyBytes),
367+
); err != nil {
368+
returnErr = fmt.Errorf("[%s]Payment cancellation error = %s", payment.PaymentId, err)
369+
return
370+
}
326371
canceled = true
327372
return
328373

329374
case "COMPLETE":
330375

331376
// mark the payment complete in our DB
332-
model.CompletePayment(
377+
if err := model.CompletePayment(
333378
clientSession.Ctx,
334379
payment.PaymentId,
335380
string(txResponseBodyBytes),
336381
tx.TxHash,
337-
)
382+
); err != nil {
383+
returnErr = fmt.Errorf("[%s]Payment completion error = %s", payment.PaymentId, err)
384+
return
385+
}
338386
complete = true
339387

340388
userAuth, err := model.GetUserAuth(clientSession.Ctx, payment.NetworkId)
@@ -434,7 +482,10 @@ func advancePayment(
434482
// to the payment not being large enough to cover the transfer fee.
435483
glog.Info("[payout][%s]payout - fee is negative\n", payment.PaymentId)
436484

437-
model.CancelPayment(clientSession.Ctx, payment.PaymentId)
485+
if err := model.CancelPayment(clientSession.Ctx, payment.PaymentId); err != nil {
486+
returnErr = fmt.Errorf("[%s]Payment cancellation error = %s", payment.PaymentId, err)
487+
return
488+
}
438489
canceled = true
439490
return
440491
}

‎controller/account_payment_controller_test.go‎

Lines changed: 191 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -584,6 +584,197 @@ func TestAdvancePaymentWalletSafetyAndIdempotency(t *testing.T) {
584584
})
585585
}
586586

587+
// Circle retry states must remain pending regardless of age. CONFIRMED already
588+
// has an on-chain hash, so persisting that hash makes the retry reconcilable;
589+
// only a non-contradictory terminal CANCELLED state may clear retry markers and
590+
// release a payment for replacement.
591+
func TestAdvancePaymentRetryAndTerminalCancellationState(t *testing.T) {
592+
server.DefaultTestEnv().Run(t, func(t testing.TB) {
593+
ctx := context.Background()
594+
networkId := server.NewId()
595+
clientSession := session.Testing_CreateClientSession(ctx, &jwt.ByJwt{
596+
NetworkId: networkId,
597+
})
598+
defer clientSession.Cancel()
599+
600+
insertRetryPayment := func(record string, txHash *string) *model.AccountPayment {
601+
paymentId := server.NewId()
602+
server.Tx(ctx, func(tx server.PgTx) {
603+
server.RaisePgResult(tx.Exec(
604+
ctx,
605+
`
606+
INSERT INTO account_payment (
607+
payment_id,
608+
payment_plan_id,
609+
network_id,
610+
wallet_id,
611+
payout_byte_count,
612+
payout_nano_cents,
613+
min_sweep_time,
614+
payment_record,
615+
circle_idempotency_key,
616+
tx_hash
617+
) VALUES ($1, $2, $3, NULL, 100, 100, $4, $5, $6, $7)
618+
`,
619+
paymentId,
620+
server.NewId(),
621+
networkId,
622+
server.NowUtc(),
623+
record,
624+
server.NewId(),
625+
txHash,
626+
))
627+
})
628+
payment, err := model.GetPayment(ctx, paymentId)
629+
connect.AssertEqual(t, err, nil)
630+
connect.AssertNotEqual(t, payment, nil)
631+
return payment
632+
}
633+
paymentHasIdempotencyKey := func(paymentId server.Id) bool {
634+
var hasIdempotencyKey bool
635+
server.Db(ctx, func(conn server.PgConn) {
636+
result, queryErr := conn.Query(
637+
ctx,
638+
`SELECT circle_idempotency_key IS NOT NULL FROM account_payment WHERE payment_id = $1`,
639+
paymentId,
640+
)
641+
server.WithPgResult(result, queryErr, func() {
642+
if result.Next() {
643+
server.Raise(result.Scan(&hasIdempotencyKey))
644+
}
645+
})
646+
})
647+
return hasIdempotencyKey
648+
}
649+
650+
{
651+
payment := insertRetryPayment("circle-confirmed", nil)
652+
const txHash = "confirmed-chain-hash"
653+
const receipt = `{"state":"CONFIRMED","txHash":"confirmed-chain-hash"}`
654+
SetCircleClient(&mockCircleApiClient{
655+
GetTransactionFunc: func(context.Context, string) (*GetTransactionResult, error) {
656+
return &GetTransactionResult{
657+
Transaction: CircleTransaction{
658+
State: "CONFIRMED",
659+
TxHash: txHash,
660+
},
661+
ResponseBodyBytes: []byte(receipt),
662+
}, nil
663+
},
664+
})
665+
666+
complete, canceled, err := advancePayment(payment, clientSession)
667+
connect.AssertEqual(t, err, nil)
668+
connect.AssertEqual(t, complete, false)
669+
connect.AssertEqual(t, canceled, false)
670+
671+
updated, err := model.GetPayment(ctx, payment.PaymentId)
672+
connect.AssertEqual(t, err, nil)
673+
connect.AssertEqual(t, updated.Completed, false)
674+
connect.AssertEqual(t, updated.Canceled, false)
675+
connect.AssertNotEqual(t, updated.PaymentRecord, nil)
676+
connect.AssertNotEqual(t, updated.TxHash, nil)
677+
connect.AssertNotEqual(t, updated.PaymentReceipt, nil)
678+
connect.AssertEqual(t, *updated.PaymentRecord, "circle-confirmed")
679+
connect.AssertEqual(t, *updated.TxHash, txHash)
680+
connect.AssertEqual(t, *updated.PaymentReceipt, receipt)
681+
}
682+
683+
{
684+
// Reproduce the old cancel/complete race with the stale payment object
685+
// already held by the worker. A rejected DB completion must not be
686+
// reported as complete, or this worker would silently stop retrying.
687+
payment := insertRetryPayment("circle-complete-race", nil)
688+
server.Tx(ctx, func(tx server.PgTx) {
689+
server.RaisePgResult(tx.Exec(
690+
ctx,
691+
`UPDATE account_payment SET canceled = true, cancel_time = now() WHERE payment_id = $1`,
692+
payment.PaymentId,
693+
))
694+
})
695+
SetCircleClient(&mockCircleApiClient{
696+
GetTransactionFunc: func(context.Context, string) (*GetTransactionResult, error) {
697+
return &GetTransactionResult{
698+
Transaction: CircleTransaction{
699+
State: "COMPLETE",
700+
TxHash: "complete-chain-hash",
701+
},
702+
ResponseBodyBytes: []byte(`{"state":"COMPLETE"}`),
703+
}, nil
704+
},
705+
})
706+
707+
complete, canceled, err := advancePayment(payment, clientSession)
708+
connect.AssertNotEqual(t, err, nil)
709+
connect.AssertEqual(t, complete, false)
710+
connect.AssertEqual(t, canceled, false)
711+
712+
updated, err := model.GetPayment(ctx, payment.PaymentId)
713+
connect.AssertEqual(t, err, nil)
714+
connect.AssertEqual(t, updated.Completed, false)
715+
connect.AssertEqual(t, updated.Canceled, true)
716+
connect.AssertNotEqual(t, updated.PaymentRecord, nil)
717+
}
718+
719+
{
720+
oldHash := "mempool-hash"
721+
payment := insertRetryPayment("circle-cancelled-after-sent", &oldHash)
722+
const receipt = `{"state":"CANCELLED"}`
723+
SetCircleClient(&mockCircleApiClient{
724+
GetTransactionFunc: func(context.Context, string) (*GetTransactionResult, error) {
725+
return &GetTransactionResult{
726+
Transaction: CircleTransaction{State: "CANCELLED"},
727+
ResponseBodyBytes: []byte(receipt),
728+
}, nil
729+
},
730+
})
731+
732+
complete, canceled, err := advancePayment(payment, clientSession)
733+
connect.AssertNotEqual(t, err, nil)
734+
connect.AssertEqual(t, complete, false)
735+
connect.AssertEqual(t, canceled, false)
736+
737+
updated, err := model.GetPayment(ctx, payment.PaymentId)
738+
connect.AssertEqual(t, err, nil)
739+
connect.AssertEqual(t, updated.Canceled, false)
740+
connect.AssertNotEqual(t, updated.PaymentRecord, nil)
741+
connect.AssertNotEqual(t, updated.TxHash, nil)
742+
connect.AssertNotEqual(t, updated.PaymentReceipt, nil)
743+
connect.AssertEqual(t, *updated.PaymentRecord, "circle-cancelled-after-sent")
744+
connect.AssertEqual(t, *updated.TxHash, oldHash)
745+
connect.AssertEqual(t, *updated.PaymentReceipt, receipt)
746+
connect.AssertEqual(t, paymentHasIdempotencyKey(payment.PaymentId), true)
747+
}
748+
749+
{
750+
payment := insertRetryPayment("circle-cancelled", nil)
751+
const receipt = `{"state":"CANCELLED"}`
752+
SetCircleClient(&mockCircleApiClient{
753+
GetTransactionFunc: func(context.Context, string) (*GetTransactionResult, error) {
754+
return &GetTransactionResult{
755+
Transaction: CircleTransaction{State: "CANCELLED"},
756+
ResponseBodyBytes: []byte(receipt),
757+
}, nil
758+
},
759+
})
760+
761+
complete, canceled, err := advancePayment(payment, clientSession)
762+
connect.AssertEqual(t, err, nil)
763+
connect.AssertEqual(t, complete, false)
764+
connect.AssertEqual(t, canceled, true)
765+
766+
updated, err := model.GetPayment(ctx, payment.PaymentId)
767+
connect.AssertEqual(t, err, nil)
768+
connect.AssertEqual(t, updated.Canceled, true)
769+
connect.AssertEqual(t, updated.PaymentRecord, nil)
770+
connect.AssertEqual(t, updated.TxHash, nil)
771+
connect.AssertNotEqual(t, updated.PaymentReceipt, nil)
772+
connect.AssertEqual(t, *updated.PaymentReceipt, receipt)
773+
connect.AssertEqual(t, paymentHasIdempotencyKey(payment.PaymentId), false)
774+
}
775+
})
776+
}
777+
587778
func TestFeeToUsd(t *testing.T) {
588779
server.DefaultTestEnv().Run(t, func(t testing.TB) {
589780
coinbaseClient := &mockCoinbaseClient{

‎db_migrations.go‎

Lines changed: 39 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -3760,13 +3760,12 @@ var migrations = []any{
37603760
ON verify_provider_stats (period_end)
37613761
`),
37623762

3763-
// CancelHungAccountPayments (model/account_payment_model.go, daily): `UPDATE
3764-
// account_payment SET canceled = true ... WHERE NOT completed AND NOT
3765-
// canceled AND create_time < $1`. No index served it; it scanned the whole
3766-
// non-completed band, which grows monotonically (canceled rows stay
3767-
// completed = false and account_payment is never deleted). This partial
3768-
// indexes exactly the truly-pending set, ordered by create_time so the scan
3769-
// is a tight range that stops early.
3763+
// CancelHungAccountPayments (model/account_payment_model.go, daily) selects
3764+
// old pending rows and then filters out all Circle retry markers. No index
3765+
// served its create_time bound; it scanned the whole non-completed band,
3766+
// which grows monotonically (canceled rows stay completed = false and
3767+
// account_payment is never deleted). This partial index bounds that scan to
3768+
// pending rows and orders it by create_time so the range stops early.
37703769
newSqlMigration(`
37713770
CREATE INDEX IF NOT EXISTS account_payment_pending_create_time
37723771
ON account_payment (create_time) WHERE (NOT completed AND NOT canceled)
@@ -4169,13 +4168,12 @@ var migrations = []any{
41694168
)
41704169
`),
41714170

4172-
// The payout planner (planPayments) re-picks sweeps whose payment was
4173-
// canceled (CancelHungAccountPayments sets canceled=true but does not null the
4174-
// sweep's payment_id). The payout query's canceled UNION arm drives from the
4175-
// small canceled set joined to sweeps by payment_id; this partial index over
4176-
// just the canceled rows makes finding those payment_ids an index-only scan
4177-
// instead of a seq scan of account_payment. account_payment is large: pre-
4178-
// create manually with CREATE INDEX CONCURRENTLY out of band; the IF NOT
4171+
// The payout planner (planPayments) re-picks sweeps whose payment was safely
4172+
// canceled and has no Circle retry markers. Its canceled UNION arm drives
4173+
// from the small canceled set joined to sweeps by payment_id; this partial
4174+
// index over canceled rows makes finding those payment_ids an indexed scan
4175+
// instead of a seq scan of account_payment. account_payment is large:
4176+
// pre-create manually with CREATE INDEX CONCURRENTLY out of band; the IF NOT
41794177
// EXISTS gate makes this migration a no-op once it is pre-created.
41804178
newSqlMigration(`
41814179
CREATE INDEX IF NOT EXISTS account_payment_canceled_payment_id
@@ -5653,6 +5651,33 @@ var migrations = []any{
56535651
`DROP INDEX IF EXISTS transfer_contract_open_source_id_companion_contract_id`,
56545652
),
56555653

5654+
// A 30-day hung-payment sweep briefly canceled Circle retries and made their
5655+
// sweeps eligible for another payout. The age relationship fingerprints that
5656+
// sweep, and an attached sweep proves the planner has not already reassigned
5657+
// it. Restore only rows satisfying both; ambiguous rows remain canceled for
5658+
// manual reconciliation rather than risking two live payments.
5659+
newSqlMigration(`
5660+
/* restore_canceled_circle_retries */
5661+
UPDATE account_payment AS payment
5662+
SET
5663+
canceled = false,
5664+
cancel_time = NULL
5665+
WHERE
5666+
payment.canceled AND
5667+
NOT payment.completed AND
5668+
payment.cancel_time >= payment.create_time + INTERVAL '30 days' AND
5669+
(
5670+
payment.circle_idempotency_key IS NOT NULL OR
5671+
payment.payment_record IS NOT NULL OR
5672+
payment.tx_hash IS NOT NULL
5673+
) AND
5674+
EXISTS (
5675+
SELECT 1
5676+
FROM transfer_escrow_sweep AS sweep
5677+
WHERE sweep.payment_id = payment.payment_id
5678+
)
5679+
`),
5680+
56565681
// Durable sim-latency competition control plane. The queue is deliberately
56575682
// independent of pending_task: untrusted submissions are claimed by a
56585683
// dedicated evaluator identity, and the singleton slot below makes the FIFO

0 commit comments

Comments
 (0)