diff --git a/internal/core/application/admin.go b/internal/core/application/admin.go index 1836c1ed6..cd245a2eb 100644 --- a/internal/core/application/admin.go +++ b/internal/core/application/admin.go @@ -859,50 +859,15 @@ func (a *adminService) saveBatchSweptEvents( } for _, leaf := range vtxosLeaves { - // The VTXO is the first non-anchor output; leaf txs can - // carry an anchor at vout 0, so the VTXO is not always at - // vout 0. extractVtxoOutpoint handles that. - vtxo, err := extractVtxoOutpoint(leaf) - if err != nil { - log.WithError(err).Errorf( - "failed to extract vtxo outpoint from leaf %s", - leaf.UnsignedTx.TxID(), - ) - continue - } - leafVtxos = append(leafVtxos, *vtxo) + // a leaf tx may pay multiple receivers, collect every vtxo outpoint it carries + leafVtxos = append(leafVtxos, leafVtxoOutpoints(leaf)...) } } } - // get preconfirmed vtxos - preconfirmedVtxos := make([]domain.Outpoint, 0) - if commitmentRootSwept { - var err error - preconfirmedVtxos, err = vtxoRepo.GetSweepableVtxosByCommitmentTxid( - ctx, - commitmentTxid, - ) - if err != nil { - log.WithError(err). - Error("error while getting sweepable vtxos by commitment txid") - } - } else { - seen := make(map[string]struct{}) - for _, leafVtxo := range leafVtxos { - children, err := vtxoRepo.GetAllChildrenVtxos(ctx, leafVtxo) - if err != nil { - log.WithError(err).Error("error while getting children vtxos") - continue - } - for _, child := range children { - if _, ok := seen[child.String()]; !ok { - preconfirmedVtxos = append(preconfirmedVtxos, child) - seen[child.String()] = struct{}{} - } - } - } - } + preconfirmedVtxos := collectPreconfirmedVtxos( + ctx, vtxoRepo, commitmentTxid, commitmentRootSwept, leafVtxos, + ) events, err := round.Sweep(leafVtxos, preconfirmedVtxos, txid, txhex) if err != nil { diff --git a/internal/core/application/indexer_test.go b/internal/core/application/indexer_test.go index 9e057d09b..a5fcf6ab5 100644 --- a/internal/core/application/indexer_test.go +++ b/internal/core/application/indexer_test.go @@ -110,7 +110,7 @@ func (m *mockVtxoRepoForIndexer) GetLeafVtxosForBatch( return nil, nil } -func (m *mockVtxoRepoForIndexer) GetSweepableVtxosByCommitmentTxid( +func (m *mockVtxoRepoForIndexer) GetSweepablePreconfirmedVtxosByCommitmentTxid( ctx context.Context, commitmentTxid string, ) ([]domain.Outpoint, error) { @@ -124,6 +124,13 @@ func (m *mockVtxoRepoForIndexer) GetAllChildrenVtxos( return nil, nil } +func (m *mockVtxoRepoForIndexer) GetDescendantVtxos( + ctx context.Context, + outpoint domain.Outpoint, +) ([]domain.Outpoint, error) { + return nil, nil +} + func (m *mockVtxoRepoForIndexer) GetCheckpointTxsByVtxoPubKeys( ctx context.Context, pubkeys []string, ) ([]domain.Tx, error) { diff --git a/internal/core/application/sweeper.go b/internal/core/application/sweeper.go index b513c0e4f..1b5a3e683 100644 --- a/internal/core/application/sweeper.go +++ b/internal/core/application/sweeper.go @@ -546,19 +546,8 @@ func (s *sweeper) createBatchSweepTask(commitmentTxid, vtxoTreeRootTxid string) } for _, leaf := range vtxosLeaves { - // The VTXO is the first non-anchor output; leaf txs can - // carry an anchor at vout 0, so the VTXO is not always - // at vout 0. extractVtxoOutpoint handles that. - vtxo, err := extractVtxoOutpoint(leaf) - if err != nil { - log.WithError(err).Errorf( - "failed to extract vtxo outpoint from leaf %s", - leaf.UnsignedTx.TxID(), - ) - continue - } - - sweepableVtxos = append(sweepableVtxos, *vtxo) + // a leaf tx may pay multiple receivers, collect every vtxo outpoint it carries + sweepableVtxos = append(sweepableVtxos, leafVtxoOutpoints(leaf)...) } if len(sweepableVtxos) <= 0 { @@ -686,9 +675,6 @@ func (s *sweeper) createBatchSweepTask(commitmentTxid, vtxoTreeRootTxid string) vtxoRepo := s.repoManager.Vtxos() eventRepo := s.repoManager.Events() - preconfirmedVtxos := make([]domain.Outpoint, 0) - var sweepErr error - commitmentRootSwept := false for _, output := range outputsToSweep { if output.Txid == commitmentTxid { @@ -697,32 +683,9 @@ func (s *sweeper) createBatchSweepTask(commitmentTxid, vtxoTreeRootTxid string) } } - if commitmentRootSwept { - // get all vtxos related to the batch commitment txid - preconfirmedVtxos, sweepErr = vtxoRepo.GetSweepableVtxosByCommitmentTxid( - ctx, - commitmentTxid, - ) - } else { - // get all vtxos related to the leaf swept - seen := make(map[string]struct{}) - for _, leafVtxo := range leafVtxoKeys { - children, childErr := vtxoRepo.GetAllChildrenVtxos(ctx, leafVtxo) - if childErr != nil { - log.WithError(childErr).Error("error while getting children vtxos") - continue - } - for _, child := range children { - if _, ok := seen[child.String()]; !ok { - preconfirmedVtxos = append(preconfirmedVtxos, child) - seen[child.String()] = struct{}{} - } - } - } - } - if sweepErr != nil { - log.WithError(sweepErr).Error("error while getting children vtxos") - } + preconfirmedVtxos := collectPreconfirmedVtxos( + ctx, vtxoRepo, commitmentTxid, commitmentRootSwept, leafVtxoKeys, + ) events, err := round.Sweep( leafVtxoKeys, diff --git a/internal/core/application/sweeper_test.go b/internal/core/application/sweeper_test.go index 5e6e1c86d..c8884706f 100644 --- a/internal/core/application/sweeper_test.go +++ b/internal/core/application/sweeper_test.go @@ -388,6 +388,16 @@ func (m *mockVtxoRepository) GetAllChildrenVtxos( return args.Get(0).([]domain.Outpoint), args.Error(1) } +func (m *mockVtxoRepository) GetDescendantVtxos( + ctx context.Context, outpoint domain.Outpoint, +) ([]domain.Outpoint, error) { + args := m.Called(ctx, outpoint) + if args.Get(0) == nil { + return nil, args.Error(1) + } + return args.Get(0).([]domain.Outpoint), args.Error(1) +} + func (m *mockVtxoRepository) GetVtxos( ctx context.Context, outpoints []domain.Outpoint, ) ([]domain.Vtxo, error) { @@ -466,7 +476,7 @@ func (m *mockVtxoRepository) GetLeafVtxosForBatch( return nil, nil } -func (m *mockVtxoRepository) GetSweepableVtxosByCommitmentTxid( +func (m *mockVtxoRepository) GetSweepablePreconfirmedVtxosByCommitmentTxid( ctx context.Context, commitmentTxid string, ) ([]domain.Outpoint, error) { return nil, nil diff --git a/internal/core/application/utils.go b/internal/core/application/utils.go index 44d842d97..2bcd3c61e 100644 --- a/internal/core/application/utils.go +++ b/internal/core/application/utils.go @@ -91,6 +91,60 @@ func findSweepableOutputs( return sweepableBatchOutputs, nil } +// leafVtxoOutpoints returns all vtxo outpoints of a leaf tx, skipping anchor and extension +func leafVtxoOutpoints(leaf *psbt.Packet) []domain.Outpoint { + txid := leaf.UnsignedTx.TxID() + outpoints := make([]domain.Outpoint, 0, len(leaf.UnsignedTx.TxOut)) + for i, out := range leaf.UnsignedTx.TxOut { + if bytes.Equal(out.PkScript, txutils.ANCHOR_PKSCRIPT) || + extension.IsExtension(out.PkScript) { + continue + } + outpoints = append(outpoints, domain.Outpoint{Txid: txid, VOut: uint32(i)}) + } + return outpoints +} + +// collectPreconfirmedVtxos returns the preconfirmed vtxos swept along with the given leaves, +// the whole batch is fetched if the commitment root itself is among the swept inputs, +// otherwise only the descendants of each swept leaf +func collectPreconfirmedVtxos( + ctx context.Context, + vtxoRepo domain.VtxoRepository, + commitmentTxid string, + commitmentRootSwept bool, + leafVtxos []domain.Outpoint, +) []domain.Outpoint { + preconfirmedVtxos := make([]domain.Outpoint, 0) + if commitmentRootSwept { + var err error + preconfirmedVtxos, err = vtxoRepo.GetSweepablePreconfirmedVtxosByCommitmentTxid( + ctx, commitmentTxid, + ) + if err != nil { + log.WithError(err). + Error("error while getting sweepable preconfirmed vtxos by commitment txid") + } + return preconfirmedVtxos + } + + seen := make(map[string]struct{}) + for _, leafVtxo := range leafVtxos { + descendants, err := vtxoRepo.GetDescendantVtxos(ctx, leafVtxo) + if err != nil { + log.WithError(err).Error("error while getting descendant vtxos") + continue + } + for _, descendant := range descendants { + if _, ok := seen[descendant.String()]; !ok { + preconfirmedVtxos = append(preconfirmedVtxos, descendant) + seen[descendant.String()] = struct{}{} + } + } + } + return preconfirmedVtxos +} + func getSpentVtxos(intents map[string]domain.Intent) []domain.Outpoint { vtxos := make([]domain.Outpoint, 0) for _, intent := range intents { diff --git a/internal/core/domain/round.go b/internal/core/domain/round.go index b56c67551..5e528fc32 100644 --- a/internal/core/domain/round.go +++ b/internal/core/domain/round.go @@ -212,9 +212,20 @@ func (r *Round) Sweep( return nil, nil } - sweptVtxosCount := countSweptLeafVtxos(r.Changes) + // count distinct leaf txids, a leaf tx may carry more than one vtxo output + sweptLeafTxids := make(map[string]struct{}) + for _, event := range r.Changes { + if e, ok := event.(BatchSwept); ok { + for _, leaf := range e.LeafVtxos { + sweptLeafTxids[leaf.Txid] = struct{}{} + } + } + } + for _, leaf := range leafVtxos { + sweptLeafTxids[leaf.Txid] = struct{}{} + } leavesCount := len(tree.FlatTxTree(r.VtxoTree).Leaves()) - fullySwept := len(leafVtxos)+sweptVtxosCount == leavesCount + fullySwept := len(sweptLeafTxids) == leavesCount event := BatchSwept{ RoundEvent: RoundEvent{ @@ -325,13 +336,3 @@ func (r *Round) raise(event Event) { r.Changes = append(r.Changes, event) r.on(event, false) } - -func countSweptLeafVtxos(events []Event) int { - count := 0 - for _, event := range events { - if e, ok := event.(BatchSwept); ok { - count += len(e.LeafVtxos) - } - } - return count -} diff --git a/internal/core/domain/round_test.go b/internal/core/domain/round_test.go index 9f0a66e42..a620e8e3d 100644 --- a/internal/core/domain/round_test.go +++ b/internal/core/domain/round_test.go @@ -83,6 +83,8 @@ var ( emptyPtx = "cHNldP8BAgQCAAAAAQQBAAEFAQABBgEDAfsEAgAAAAA=" emptyTx = "0200000000000000000000" txid = "0000000000000000000000000000000000000000000000000000000000000000" + leafTxid1 = "0000000000000000000000000000000000000000000000000000000000000001" + leafTxid2 = "0000000000000000000000000000000000000000000000000000000000000002" emptyForfeitTx = domain.ForfeitTx{ Txid: txid, Tx: emptyPtx, @@ -99,17 +101,17 @@ var ( Txid: txid, Tx: emptyPtx, Children: map[uint32]string{ - 0: txid, - 1: txid, + 0: leafTxid1, + 1: leafTxid2, }, }, { - Txid: txid, + Txid: leafTxid1, Tx: emptyPtx, Children: nil, }, { - Txid: txid, + Txid: leafTxid2, Tx: emptyPtx, Children: nil, }, diff --git a/internal/core/domain/vtxo_repo.go b/internal/core/domain/vtxo_repo.go index 638a35765..18f4c12cb 100644 --- a/internal/core/domain/vtxo_repo.go +++ b/internal/core/domain/vtxo_repo.go @@ -21,10 +21,14 @@ type VtxoRepository interface { UpdateVtxosExpiration(ctx context.Context, outpoints []Outpoint, expiresAt int64) error GetLeafVtxosForBatch(ctx context.Context, txid string) ([]Vtxo, error) GetCheckpointTxsByVtxoPubKeys(ctx context.Context, pubkeys []string) ([]Tx, error) - GetSweepableVtxosByCommitmentTxid( + // returns only the preconfirmed vtxos of the batch, leaves are excluded + GetSweepablePreconfirmedVtxosByCommitmentTxid( ctx context.Context, commitmentTxid string, ) ([]Outpoint, error) + // returns the vtxo of the given outpoint plus all its descendants GetAllChildrenVtxos(ctx context.Context, outpoint Outpoint) ([]Outpoint, error) + // returns only the descendants, the vtxo of the given outpoint is excluded + GetDescendantVtxos(ctx context.Context, outpoint Outpoint) ([]Outpoint, error) GetVtxoPubKeysByCommitmentTxid( ctx context.Context, commitmentTxid string, withMinimumAmount uint64, ) ( diff --git a/internal/infrastructure/db/badger/vtxo_repo.go b/internal/infrastructure/db/badger/vtxo_repo.go index 863cd180d..7189d9f4c 100644 --- a/internal/infrastructure/db/badger/vtxo_repo.go +++ b/internal/infrastructure/db/badger/vtxo_repo.go @@ -739,7 +739,7 @@ func (r *VtxoRepository) updateVtxo(ctx context.Context, vtxo *domain.Vtxo) erro return nil } -func (r *VtxoRepository) GetSweepableVtxosByCommitmentTxid( +func (r *VtxoRepository) GetSweepablePreconfirmedVtxosByCommitmentTxid( ctx context.Context, txid string, ) ([]domain.Outpoint, error) { @@ -767,8 +767,9 @@ func (r *VtxoRepository) GetSweepableVtxosByCommitmentTxid( for _, vtxo := range vtxos { outpointKey := vtxo.Outpoint.String() if !visited[outpointKey] { - if _, seen := visited[outpointKey]; !seen { - visited[outpointKey] = true + visited[outpointKey] = true + // only preconfirmed vtxos are returned, leaves are excluded + if vtxo.Preconfirmed { outpoints = append(outpoints, vtxo.Outpoint) } @@ -840,3 +841,62 @@ func (r *VtxoRepository) GetAllChildrenVtxos( return outpoints, nil } + +func (r *VtxoRepository) GetDescendantVtxos( + ctx context.Context, + outpoint domain.Outpoint, +) ([]domain.Outpoint, error) { + // Seed with the specific outpoint, not all vouts of the txid, so that + // sibling outputs (which belong to independent lineages) are not included. + seedQuery := badgerhold.Where("Txid").Eq(outpoint.Txid). + And("VOut").Eq(outpoint.VOut) + seedVtxos, err := r.findVtxos(ctx, seedQuery) + if err != nil { + return nil, fmt.Errorf("failed to find seed vtxo %s: %w", outpoint, err) + } + + visited := make(map[string]bool) + visitedTxids := make(map[string]bool) + var outpoints []domain.Outpoint + queue := make([]string, 0, len(seedVtxos)) + + // the seed vtxo is excluded from the result, only its spending tx is followed + for _, vtxo := range seedVtxos { + outpointKey := vtxo.Outpoint.String() + if !visited[outpointKey] { + visited[outpointKey] = true + if vtxo.ArkTxid != "" { + queue = append(queue, vtxo.ArkTxid) + } + } + } + + for len(queue) > 0 { + currentTxid := queue[0] + queue = queue[1:] + + if visitedTxids[currentTxid] { + continue + } + visitedTxids[currentTxid] = true + + query := badgerhold.Where("Txid").Eq(currentTxid) + vtxos, err := r.findVtxos(ctx, query) + if err != nil { + return nil, fmt.Errorf("failed to find vtxos for txid %s: %w", currentTxid, err) + } + + for _, vtxo := range vtxos { + outpointKey := vtxo.Outpoint.String() + if !visited[outpointKey] { + visited[outpointKey] = true + outpoints = append(outpoints, vtxo.Outpoint) + if vtxo.ArkTxid != "" { + queue = append(queue, vtxo.ArkTxid) + } + } + } + } + + return outpoints, nil +} diff --git a/internal/infrastructure/db/postgres/sqlc/queries/query.sql.go b/internal/infrastructure/db/postgres/sqlc/queries/query.sql.go index 784d219c1..31da7067c 100644 --- a/internal/infrastructure/db/postgres/sqlc/queries/query.sql.go +++ b/internal/infrastructure/db/postgres/sqlc/queries/query.sql.go @@ -525,6 +525,76 @@ func (q *Queries) SelectConvictionsInTimeRange(ctx context.Context, arg SelectCo return items, nil } +const selectDescendantVtxoOutpointsByArkTxid = `-- name: SelectDescendantVtxoOutpointsByArkTxid :many +WITH RECURSIVE descendants_chain AS ( + -- seed: only the specific outpoint, not all vouts of the txid + SELECT v.txid, v.vout, v.preconfirmed, v.ark_txid, v.spent_by, + 0 AS depth, + ARRAY[(v.txid||':'||v.vout)]::text[] AS visited + FROM vtxo v + WHERE v.txid = $1 AND v.vout = $2 + + UNION ALL + + -- children: next vtxo(s) are those whose txid == current.ark_txid + SELECT c.txid, c.vout, c.preconfirmed, c.ark_txid, c.spent_by, + w.depth + 1, + w.visited || (c.txid||':'||c.vout) + FROM descendants_chain w + JOIN vtxo c + ON c.txid = w.ark_txid + WHERE w.ark_txid IS NOT NULL + AND (c.txid||':'||c.vout) <> ALL (w.visited) -- cycle/visited guard +), +nodes AS ( + SELECT DISTINCT ON (txid, vout) + txid, vout, preconfirmed, depth + FROM descendants_chain + ORDER BY txid, vout, depth +) +SELECT txid, vout +FROM nodes +WHERE depth > 0 +ORDER BY depth, txid, vout +` + +type SelectDescendantVtxoOutpointsByArkTxidParams struct { + Txid string + Vout int32 +} + +type SelectDescendantVtxoOutpointsByArkTxidRow struct { + Txid string + Vout int32 +} + +// Same lineage walk as SelectVtxosOutpointsByArkTxidRecursive but the seed +// outpoint itself is excluded from the result (depth > 0), descendants only. +// keep one row per node at its MIN depth (layers) +// depth > 0 excludes the seed vtxo itself, descendants only +func (q *Queries) SelectDescendantVtxoOutpointsByArkTxid(ctx context.Context, arg SelectDescendantVtxoOutpointsByArkTxidParams) ([]SelectDescendantVtxoOutpointsByArkTxidRow, error) { + rows, err := q.db.QueryContext(ctx, selectDescendantVtxoOutpointsByArkTxid, arg.Txid, arg.Vout) + if err != nil { + return nil, err + } + defer rows.Close() + var items []SelectDescendantVtxoOutpointsByArkTxidRow + for rows.Next() { + var i SelectDescendantVtxoOutpointsByArkTxidRow + if err := rows.Scan(&i.Txid, &i.Vout); 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 selectExpiredRounds = `-- name: SelectExpiredRounds :many SELECT r.id, r.txid, CAST(r.ending_timestamp + r.vtxo_tree_expiration AS BIGINT) AS expired_at FROM round_with_commitment_tx_vw r @@ -1659,6 +1729,43 @@ func (q *Queries) SelectSettings(ctx context.Context) (Setting, error) { return i, err } +const selectSweepablePreconfirmedVtxoOutpointsByCommitmentTxid = `-- name: SelectSweepablePreconfirmedVtxoOutpointsByCommitmentTxid :many +SELECT DISTINCT v.txid AS vtxo_txid, v.vout AS vtxo_vout +FROM vtxo_vw v +WHERE v.swept = false + AND v.preconfirmed = true + AND (v.commitment_txid = $1 + OR (',' || COALESCE(v.commitments::text, '') || ',') LIKE '%,' || $1 || ',%') +` + +type SelectSweepablePreconfirmedVtxoOutpointsByCommitmentTxidRow struct { + VtxoTxid string + VtxoVout int32 +} + +func (q *Queries) SelectSweepablePreconfirmedVtxoOutpointsByCommitmentTxid(ctx context.Context, commitmentTxid string) ([]SelectSweepablePreconfirmedVtxoOutpointsByCommitmentTxidRow, error) { + rows, err := q.db.QueryContext(ctx, selectSweepablePreconfirmedVtxoOutpointsByCommitmentTxid, commitmentTxid) + if err != nil { + return nil, err + } + defer rows.Close() + var items []SelectSweepablePreconfirmedVtxoOutpointsByCommitmentTxidRow + for rows.Next() { + var i SelectSweepablePreconfirmedVtxoOutpointsByCommitmentTxidRow + if err := rows.Scan(&i.VtxoTxid, &i.VtxoVout); 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 selectSweepableRounds = `-- name: SelectSweepableRounds :many SELECT txid FROM round_with_commitment_tx_vw r WHERE r.swept = false AND r.ended = true AND r.failed = false @@ -1744,42 +1851,6 @@ func (q *Queries) SelectSweepableUnrolledVtxos(ctx context.Context) ([]SelectSwe return items, nil } -const selectSweepableVtxoOutpointsByCommitmentTxid = `-- name: SelectSweepableVtxoOutpointsByCommitmentTxid :many -SELECT DISTINCT v.txid AS vtxo_txid, v.vout AS vtxo_vout -FROM vtxo_vw v -WHERE v.swept = false - AND (v.commitment_txid = $1 - OR (',' || COALESCE(v.commitments::text, '') || ',') LIKE '%,' || $1 || ',%') -` - -type SelectSweepableVtxoOutpointsByCommitmentTxidRow struct { - VtxoTxid string - VtxoVout int32 -} - -func (q *Queries) SelectSweepableVtxoOutpointsByCommitmentTxid(ctx context.Context, commitmentTxid string) ([]SelectSweepableVtxoOutpointsByCommitmentTxidRow, error) { - rows, err := q.db.QueryContext(ctx, selectSweepableVtxoOutpointsByCommitmentTxid, commitmentTxid) - if err != nil { - return nil, err - } - defer rows.Close() - var items []SelectSweepableVtxoOutpointsByCommitmentTxidRow - for rows.Next() { - var i SelectSweepableVtxoOutpointsByCommitmentTxidRow - if err := rows.Scan(&i.VtxoTxid, &i.VtxoVout); 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 selectSweptMarkersByIds = `-- name: SelectSweptMarkersByIds :many SELECT marker_id, swept_at FROM swept_marker WHERE marker_id = ANY($1::text[]) ` diff --git a/internal/infrastructure/db/postgres/sqlc/query.sql b/internal/infrastructure/db/postgres/sqlc/query.sql index d8462c833..d6562dfc6 100644 --- a/internal/infrastructure/db/postgres/sqlc/query.sql +++ b/internal/infrastructure/db/postgres/sqlc/query.sql @@ -314,10 +314,11 @@ WHERE v.amount >= @min_amount ) ); --- name: SelectSweepableVtxoOutpointsByCommitmentTxid :many +-- name: SelectSweepablePreconfirmedVtxoOutpointsByCommitmentTxid :many SELECT DISTINCT v.txid AS vtxo_txid, v.vout AS vtxo_vout FROM vtxo_vw v WHERE v.swept = false + AND v.preconfirmed = true AND (v.commitment_txid = @commitment_txid OR (',' || COALESCE(v.commitments::text, '') || ',') LIKE '%,' || @commitment_txid || ',%'); @@ -357,6 +358,42 @@ SELECT txid, vout FROM nodes ORDER BY depth, txid, vout; +-- name: SelectDescendantVtxoOutpointsByArkTxid :many +-- Same lineage walk as SelectVtxosOutpointsByArkTxidRecursive but the seed +-- outpoint itself is excluded from the result (depth > 0), descendants only. +WITH RECURSIVE descendants_chain AS ( + -- seed: only the specific outpoint, not all vouts of the txid + SELECT v.txid, v.vout, v.preconfirmed, v.ark_txid, v.spent_by, + 0 AS depth, + ARRAY[(v.txid||':'||v.vout)]::text[] AS visited + FROM vtxo v + WHERE v.txid = @txid AND v.vout = @vout + + UNION ALL + + -- children: next vtxo(s) are those whose txid == current.ark_txid + SELECT c.txid, c.vout, c.preconfirmed, c.ark_txid, c.spent_by, + w.depth + 1, + w.visited || (c.txid||':'||c.vout) + FROM descendants_chain w + JOIN vtxo c + ON c.txid = w.ark_txid + WHERE w.ark_txid IS NOT NULL + AND (c.txid||':'||c.vout) <> ALL (w.visited) -- cycle/visited guard +), +-- keep one row per node at its MIN depth (layers) +nodes AS ( + SELECT DISTINCT ON (txid, vout) + txid, vout, preconfirmed, depth + FROM descendants_chain + ORDER BY txid, vout, depth +) +-- depth > 0 excludes the seed vtxo itself, descendants only +SELECT txid, vout +FROM nodes +WHERE depth > 0 +ORDER BY depth, txid, vout; + -- name: SelectSweepableUnrolledVtxos :many SELECT sqlc.embed(vtxo_vw) FROM vtxo_vw WHERE spent = true AND unrolled = true AND swept = false AND COALESCE(settled_by, '') = ''; diff --git a/internal/infrastructure/db/postgres/vtxo_repo.go b/internal/infrastructure/db/postgres/vtxo_repo.go index 019f9d615..d67cea49c 100644 --- a/internal/infrastructure/db/postgres/vtxo_repo.go +++ b/internal/infrastructure/db/postgres/vtxo_repo.go @@ -401,11 +401,13 @@ func (v *vtxoRepository) GetAllVtxosWithPubKeys( return vtxos, nil } -func (v *vtxoRepository) GetSweepableVtxosByCommitmentTxid( +func (v *vtxoRepository) GetSweepablePreconfirmedVtxosByCommitmentTxid( ctx context.Context, commitmentTxid string, ) ([]domain.Outpoint, error) { - res, err := v.querier.SelectSweepableVtxoOutpointsByCommitmentTxid(ctx, commitmentTxid) + res, err := v.querier.SelectSweepablePreconfirmedVtxoOutpointsByCommitmentTxid( + ctx, commitmentTxid, + ) if err != nil { return nil, err } @@ -446,6 +448,31 @@ func (v *vtxoRepository) GetAllChildrenVtxos( return outpoints, nil } +func (v *vtxoRepository) GetDescendantVtxos( + ctx context.Context, outpoint domain.Outpoint, +) ([]domain.Outpoint, error) { + res, err := v.querier.SelectDescendantVtxoOutpointsByArkTxid( + ctx, + queries.SelectDescendantVtxoOutpointsByArkTxidParams{ + Txid: outpoint.Txid, + Vout: int32(outpoint.VOut), + }, + ) + if err != nil { + return nil, err + } + + outpoints := make([]domain.Outpoint, 0, len(res)) + for _, row := range res { + outpoints = append(outpoints, domain.Outpoint{ + Txid: row.Txid, + VOut: uint32(row.Vout), + }) + } + + return outpoints, nil +} + func (v *vtxoRepository) GetVtxoPubKeysByCommitmentTxid( ctx context.Context, commitmentTxid string, withMinimumAmount uint64, ) ([]string, error) { diff --git a/internal/infrastructure/db/service_test.go b/internal/infrastructure/db/service_test.go index 7e7926b78..f41361a5e 100644 --- a/internal/infrastructure/db/service_test.go +++ b/internal/infrastructure/db/service_test.go @@ -943,6 +943,7 @@ func testVtxoRepository(t *testing.T, svc ports.RepoManager) { Amount: 2000, RootCommitmentTxid: commitmentTxid1, CommitmentTxids: []string{commitmentTxid1}, + Preconfirmed: true, ArkTxid: randomString(32), // Points to vtxo3 } @@ -955,6 +956,7 @@ func testVtxoRepository(t *testing.T, svc ports.RepoManager) { Amount: 3000, RootCommitmentTxid: commitmentTxid1, CommitmentTxids: []string{commitmentTxid1}, + Preconfirmed: true, ArkTxid: randomString(32), // Points to vtxo4 } @@ -967,6 +969,7 @@ func testVtxoRepository(t *testing.T, svc ports.RepoManager) { Amount: 4000, RootCommitmentTxid: commitmentTxid1, CommitmentTxids: []string{commitmentTxid1}, + Preconfirmed: true, ArkTxid: "", // End of chain - null ark_txid } @@ -976,12 +979,12 @@ func testVtxoRepository(t *testing.T, svc ports.RepoManager) { require.NoError(t, err) children, err := svc.Vtxos(). - GetSweepableVtxosByCommitmentTxid(ctx, vtxo1.RootCommitmentTxid) + GetSweepablePreconfirmedVtxosByCommitmentTxid(ctx, vtxo1.RootCommitmentTxid) require.NoError(t, err) - require.Len(t, children, 4) + require.Len(t, children, 3) + // vtxo1 is a leaf, only the preconfirmed descendants are returned expectedOutpoints := []domain.Outpoint{ - vtxo1.Outpoint, vtxo2.Outpoint, vtxo3.Outpoint, vtxo4.Outpoint, @@ -997,7 +1000,8 @@ func testVtxoRepository(t *testing.T, svc ports.RepoManager) { require.Equal(t, expectedOutpoints, children) // Test with non-existent txid - children, err = svc.Vtxos().GetSweepableVtxosByCommitmentTxid(ctx, randomString(32)) + children, err = svc.Vtxos(). + GetSweepablePreconfirmedVtxosByCommitmentTxid(ctx, randomString(32)) require.NoError(t, err) require.Empty(t, children) @@ -1006,11 +1010,36 @@ func testVtxoRepository(t *testing.T, svc ports.RepoManager) { require.NoError(t, err) require.Len(t, children, 4) // Should return all 4 vtxos in the chain + expectedChainOutpoints := []domain.Outpoint{ + vtxo1.Outpoint, + vtxo2.Outpoint, + vtxo3.Outpoint, + vtxo4.Outpoint, + } + sort.Slice(children, func(i, j int) bool { return children[i].Txid < children[j].Txid }) + sort.Slice(expectedChainOutpoints, func(i, j int) bool { + return expectedChainOutpoints[i].Txid < expectedChainOutpoints[j].Txid + }) - require.Equal(t, expectedOutpoints, children) + require.Equal(t, expectedChainOutpoints, children) + + // Test descendants only, the vtxo itself is excluded + descendants, err := svc.Vtxos().GetDescendantVtxos(ctx, vtxo1.Outpoint) + require.NoError(t, err) + require.Len(t, descendants, 3) + + sort.Slice(descendants, func(i, j int) bool { + return descendants[i].Txid < descendants[j].Txid + }) + require.Equal(t, expectedOutpoints, descendants) + + // Descendants of the end of the chain must be empty + descendants, err = svc.Vtxos().GetDescendantVtxos(ctx, vtxo4.Outpoint) + require.NoError(t, err) + require.Empty(t, descendants) // Test starting from middle of chain (vtxo2) children, err = svc.Vtxos().GetAllChildrenVtxos(ctx, vtxo2.Outpoint) diff --git a/internal/infrastructure/db/sqlite/sqlc/queries/query.sql.go b/internal/infrastructure/db/sqlite/sqlc/queries/query.sql.go index ddff072a8..f3e10dcbd 100644 --- a/internal/infrastructure/db/sqlite/sqlc/queries/query.sql.go +++ b/internal/infrastructure/db/sqlite/sqlc/queries/query.sql.go @@ -553,6 +553,76 @@ func (q *Queries) SelectConvictionsInTimeRange(ctx context.Context, arg SelectCo return items, nil } +const selectDescendantVtxoOutpointsByArkTxid = `-- name: SelectDescendantVtxoOutpointsByArkTxid :many +WITH RECURSIVE descendants_chain AS ( + -- seed: only the specific outpoint, not all vouts of the txid + SELECT v.txid, v.vout, v.preconfirmed, v.ark_txid, v.spent_by, + 0 AS depth, + v.txid||':'||v.vout AS visited + FROM vtxo v + WHERE v.txid = ?1 AND v.vout = ?2 + + UNION ALL + + -- children: next vtxo(s) are those whose txid == current.ark_txid + SELECT c.txid, c.vout, c.preconfirmed, c.ark_txid, c.spent_by, + w.depth + 1, + w.visited || ',' || (c.txid||':'||c.vout) + FROM descendants_chain w + JOIN vtxo c + ON c.txid = w.ark_txid + WHERE w.ark_txid IS NOT NULL + -- delimiter-bounded match so txid:1 cannot match inside txid:12 + AND ',' || w.visited || ',' NOT LIKE '%,' || (c.txid||':'||c.vout) || ',%' +), +nodes AS ( + SELECT txid, vout, preconfirmed, MIN(depth) as depth + FROM descendants_chain + GROUP BY txid, vout, preconfirmed +) +SELECT txid, vout +FROM nodes +WHERE depth > 0 +ORDER BY depth, txid, vout +` + +type SelectDescendantVtxoOutpointsByArkTxidParams struct { + Txid string + Vout int64 +} + +type SelectDescendantVtxoOutpointsByArkTxidRow struct { + Txid string + Vout int64 +} + +// Same lineage walk as SelectVtxosOutpointsByArkTxidRecursive but the seed +// outpoint itself is excluded from the result (depth > 0), descendants only. +// keep one row per node at its MIN depth (layers) +// depth > 0 excludes the seed vtxo itself, descendants only +func (q *Queries) SelectDescendantVtxoOutpointsByArkTxid(ctx context.Context, arg SelectDescendantVtxoOutpointsByArkTxidParams) ([]SelectDescendantVtxoOutpointsByArkTxidRow, error) { + rows, err := q.db.QueryContext(ctx, selectDescendantVtxoOutpointsByArkTxid, arg.Txid, arg.Vout) + if err != nil { + return nil, err + } + defer rows.Close() + var items []SelectDescendantVtxoOutpointsByArkTxidRow + for rows.Next() { + var i SelectDescendantVtxoOutpointsByArkTxidRow + if err := rows.Scan(&i.Txid, &i.Vout); 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 selectExpiredRounds = `-- name: SelectExpiredRounds :many SELECT r.id, r.txid, CAST(r.ending_timestamp + r.vtxo_tree_expiration AS BIGINT) AS expired_at FROM round_with_commitment_tx_vw r @@ -1695,6 +1765,43 @@ func (q *Queries) SelectSettings(ctx context.Context) (Setting, error) { return i, err } +const selectSweepablePreconfirmedVtxoOutpointsByCommitmentTxid = `-- name: SelectSweepablePreconfirmedVtxoOutpointsByCommitmentTxid :many +SELECT DISTINCT v.txid AS vtxo_txid, v.vout AS vtxo_vout +FROM vtxo_vw v +WHERE v.swept = false + AND v.preconfirmed = true + AND (v.commitment_txid = ?1 + OR (',' || COALESCE(v.commitments, '') || ',') LIKE '%,' || ?1 || ',%') +` + +type SelectSweepablePreconfirmedVtxoOutpointsByCommitmentTxidRow struct { + VtxoTxid string + VtxoVout int64 +} + +func (q *Queries) SelectSweepablePreconfirmedVtxoOutpointsByCommitmentTxid(ctx context.Context, commitmentTxid string) ([]SelectSweepablePreconfirmedVtxoOutpointsByCommitmentTxidRow, error) { + rows, err := q.db.QueryContext(ctx, selectSweepablePreconfirmedVtxoOutpointsByCommitmentTxid, commitmentTxid) + if err != nil { + return nil, err + } + defer rows.Close() + var items []SelectSweepablePreconfirmedVtxoOutpointsByCommitmentTxidRow + for rows.Next() { + var i SelectSweepablePreconfirmedVtxoOutpointsByCommitmentTxidRow + if err := rows.Scan(&i.VtxoTxid, &i.VtxoVout); 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 selectSweepableRounds = `-- name: SelectSweepableRounds :many SELECT txid FROM round_with_commitment_tx_vw r WHERE r.swept = false AND r.ended = true AND r.failed = false @@ -1780,42 +1887,6 @@ func (q *Queries) SelectSweepableUnrolledVtxos(ctx context.Context) ([]SelectSwe return items, nil } -const selectSweepableVtxoOutpointsByCommitmentTxid = `-- name: SelectSweepableVtxoOutpointsByCommitmentTxid :many -SELECT DISTINCT v.txid AS vtxo_txid, v.vout AS vtxo_vout -FROM vtxo_vw v -WHERE v.swept = false - AND (v.commitment_txid = ?1 - OR (',' || COALESCE(v.commitments, '') || ',') LIKE '%,' || ?1 || ',%') -` - -type SelectSweepableVtxoOutpointsByCommitmentTxidRow struct { - VtxoTxid string - VtxoVout int64 -} - -func (q *Queries) SelectSweepableVtxoOutpointsByCommitmentTxid(ctx context.Context, commitmentTxid string) ([]SelectSweepableVtxoOutpointsByCommitmentTxidRow, error) { - rows, err := q.db.QueryContext(ctx, selectSweepableVtxoOutpointsByCommitmentTxid, commitmentTxid) - if err != nil { - return nil, err - } - defer rows.Close() - var items []SelectSweepableVtxoOutpointsByCommitmentTxidRow - for rows.Next() { - var i SelectSweepableVtxoOutpointsByCommitmentTxidRow - if err := rows.Scan(&i.VtxoTxid, &i.VtxoVout); 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 selectSweptMarkersByIds = `-- name: SelectSweptMarkersByIds :many SELECT marker_id, swept_at FROM swept_marker WHERE marker_id IN (/*SLICE:marker_ids*/?) ` @@ -2425,7 +2496,8 @@ WITH RECURSIVE descendants_chain AS ( JOIN vtxo c ON c.txid = w.ark_txid WHERE w.ark_txid IS NOT NULL - AND w.visited NOT LIKE '%' || (c.txid||':'||c.vout) || '%' -- cycle/visited guard + -- delimiter-bounded match so txid:1 cannot match inside txid:12 + AND ',' || w.visited || ',' NOT LIKE '%,' || (c.txid||':'||c.vout) || ',%' ), nodes AS ( SELECT txid, vout, preconfirmed, MIN(depth) as depth diff --git a/internal/infrastructure/db/sqlite/sqlc/query.sql b/internal/infrastructure/db/sqlite/sqlc/query.sql index a1b80b8c4..3dd2f1d8c 100644 --- a/internal/infrastructure/db/sqlite/sqlc/query.sql +++ b/internal/infrastructure/db/sqlite/sqlc/query.sql @@ -325,10 +325,11 @@ WHERE v.amount >= sqlc.arg('min_amount') ) ); --- name: SelectSweepableVtxoOutpointsByCommitmentTxid :many +-- name: SelectSweepablePreconfirmedVtxoOutpointsByCommitmentTxid :many SELECT DISTINCT v.txid AS vtxo_txid, v.vout AS vtxo_vout FROM vtxo_vw v WHERE v.swept = false + AND v.preconfirmed = true AND (v.commitment_txid = @commitment_txid OR (',' || COALESCE(v.commitments, '') || ',') LIKE '%,' || @commitment_txid || ',%'); @@ -355,7 +356,8 @@ WITH RECURSIVE descendants_chain AS ( JOIN vtxo c ON c.txid = w.ark_txid WHERE w.ark_txid IS NOT NULL - AND w.visited NOT LIKE '%' || (c.txid||':'||c.vout) || '%' -- cycle/visited guard + -- delimiter-bounded match so txid:1 cannot match inside txid:12 + AND ',' || w.visited || ',' NOT LIKE '%,' || (c.txid||':'||c.vout) || ',%' ), -- keep one row per node at its MIN depth (layers) nodes AS ( @@ -367,6 +369,42 @@ SELECT txid, vout FROM nodes ORDER BY depth, txid, vout; +-- name: SelectDescendantVtxoOutpointsByArkTxid :many +-- Same lineage walk as SelectVtxosOutpointsByArkTxidRecursive but the seed +-- outpoint itself is excluded from the result (depth > 0), descendants only. +WITH RECURSIVE descendants_chain AS ( + -- seed: only the specific outpoint, not all vouts of the txid + SELECT v.txid, v.vout, v.preconfirmed, v.ark_txid, v.spent_by, + 0 AS depth, + v.txid||':'||v.vout AS visited + FROM vtxo v + WHERE v.txid = @txid AND v.vout = @vout + + UNION ALL + + -- children: next vtxo(s) are those whose txid == current.ark_txid + SELECT c.txid, c.vout, c.preconfirmed, c.ark_txid, c.spent_by, + w.depth + 1, + w.visited || ',' || (c.txid||':'||c.vout) + FROM descendants_chain w + JOIN vtxo c + ON c.txid = w.ark_txid + WHERE w.ark_txid IS NOT NULL + -- delimiter-bounded match so txid:1 cannot match inside txid:12 + AND ',' || w.visited || ',' NOT LIKE '%,' || (c.txid||':'||c.vout) || ',%' +), +-- keep one row per node at its MIN depth (layers) +nodes AS ( + SELECT txid, vout, preconfirmed, MIN(depth) as depth + FROM descendants_chain + GROUP BY txid, vout, preconfirmed +) +-- depth > 0 excludes the seed vtxo itself, descendants only +SELECT txid, vout +FROM nodes +WHERE depth > 0 +ORDER BY depth, txid, vout; + -- name: SelectSweepableUnrolledVtxos :many SELECT sqlc.embed(vtxo_vw) FROM vtxo_vw WHERE spent = true AND unrolled = true AND swept = false AND (COALESCE(settled_by, '') = ''); diff --git a/internal/infrastructure/db/sqlite/vtxo_repo.go b/internal/infrastructure/db/sqlite/vtxo_repo.go index 8b9ae98ab..bbd03a3b1 100644 --- a/internal/infrastructure/db/sqlite/vtxo_repo.go +++ b/internal/infrastructure/db/sqlite/vtxo_repo.go @@ -451,16 +451,18 @@ func (v *vtxoRepository) GetAllVtxosWithPubKeys( return vtxos, nil } -func (v *vtxoRepository) GetSweepableVtxosByCommitmentTxid( +func (v *vtxoRepository) GetSweepablePreconfirmedVtxosByCommitmentTxid( ctx context.Context, commitmentTxid string, ) ( []domain.Outpoint, error, ) { - var res []queries.SelectSweepableVtxoOutpointsByCommitmentTxidRow + var res []queries.SelectSweepablePreconfirmedVtxoOutpointsByCommitmentTxidRow if err := withReadQuerier(ctx, v.db, func(q *queries.Queries) error { var err error - res, err = q.SelectSweepableVtxoOutpointsByCommitmentTxid(ctx, commitmentTxid) + res, err = q.SelectSweepablePreconfirmedVtxoOutpointsByCommitmentTxid( + ctx, commitmentTxid, + ) return err }); err != nil { return nil, err @@ -506,6 +508,35 @@ func (v *vtxoRepository) GetAllChildrenVtxos( return outpoints, nil } +func (v *vtxoRepository) GetDescendantVtxos( + ctx context.Context, outpoint domain.Outpoint, +) ([]domain.Outpoint, error) { + var res []queries.SelectDescendantVtxoOutpointsByArkTxidRow + if err := withReadQuerier(ctx, v.db, func(q *queries.Queries) error { + var err error + res, err = q.SelectDescendantVtxoOutpointsByArkTxid( + ctx, + queries.SelectDescendantVtxoOutpointsByArkTxidParams{ + Txid: outpoint.Txid, + Vout: int64(outpoint.VOut), + }, + ) + return err + }); err != nil { + return nil, err + } + + outpoints := make([]domain.Outpoint, 0, len(res)) + for _, row := range res { + outpoints = append(outpoints, domain.Outpoint{ + Txid: row.Txid, + VOut: uint32(row.Vout), + }) + } + + return outpoints, nil +} + func (v *vtxoRepository) GetVtxoPubKeysByCommitmentTxid( ctx context.Context, commitmentTxid string, withMinimumAmount uint64, ) ([]string, error) { diff --git a/internal/test/e2e/e2e_test.go b/internal/test/e2e/e2e_test.go index 634054af0..8d153a096 100644 --- a/internal/test/e2e/e2e_test.go +++ b/internal/test/e2e/e2e_test.go @@ -3422,6 +3422,15 @@ func TestSweep(t *testing.T) { require.NotEmpty(t, sweepEvent.Tx) require.NotEmpty(t, sweepEvent.SweptVtxos) + // the swept vtxos list must not contain duplicated outpoints + seen := make(map[string]struct{}) + for _, swept := range sweepEvent.SweptVtxos { + key := fmt.Sprintf("%s:%d", swept.Txid, swept.VOut) + _, duplicated := seen[key] + require.False(t, duplicated, "duplicated swept vtxo %s", key) + seen[key] = struct{}{} + } + // give time to indexer to update its state time.Sleep(5 * time.Second)