fix: duplicated swept vtxos in sweep tx event

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 {

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) {

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,

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

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 {

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
-}

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,
 		},

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,
 	) (