Fix: "tree signing session not found" error (#323)

* failing test

* fix duplicate input register

* fix btc-embedded coin selection

* rename test

* add checks in failing test case

* fixes GetEventStream

* add TODO comment in createPoolTx

* update with master changes

* fix server unit test

* increase liquidity of testing ASP

* simplify AliceSeveralPaymentsBob test

pkg/client-sdk/client.go

@@ -18,6 +18,7 @@ import (
 	filestore "github.com/ark-network/ark/pkg/client-sdk/wallet/singlekey/store/file"
 	inmemorystore "github.com/ark-network/ark/pkg/client-sdk/wallet/singlekey/store/inmemory"
 	"github.com/decred/dcrd/dcrec/secp256k1/v4"
+	"github.com/sirupsen/logrus"
 )
 
 const (
@@ -236,11 +237,13 @@ func (a *arkClient) ping(
 	ticker := time.NewTicker(5 * time.Second)
 
 	go func(t *time.Ticker) {
-		// nolint
+		if _, err := a.client.Ping(ctx, paymentID); err != nil {
-		a.client.Ping(ctx, paymentID)
+			logrus.Warnf("failed to ping asp: %s", err)
+		}
 		for range t.C {
-			// nolint
+			if _, err := a.client.Ping(ctx, paymentID); err != nil {
-			a.client.Ping(ctx, paymentID)
+				logrus.Warnf("failed to ping asp: %s", err)
+			}
 		}
 	}(ticker)
 

pkg/client-sdk/client/client.go

@@ -32,7 +32,7 @@ type ASPClient interface {
 	) error
 	GetEventStream(
 		ctx context.Context, paymentID string,
-	) (<-chan RoundEventChannel, error)
+	) (<-chan RoundEventChannel, func(), error)
 	Ping(ctx context.Context, paymentID string) (RoundEvent, error)
 	FinalizePayment(
 		ctx context.Context, signedForfeitTxs []string, signedRoundTx string,

pkg/client-sdk/client/grpc/client.go

@@ -15,6 +15,7 @@ import (
 	"github.com/ark-network/ark/pkg/client-sdk/internal/utils"
 	"github.com/decred/dcrd/dcrec/secp256k1/v4"
 	"github.com/lightningnetwork/lnd/lnwallet/chainfee"
+	"github.com/sirupsen/logrus"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials"
 	"google.golang.org/grpc/credentials/insecure"
@@ -23,7 +24,6 @@ import (
 type grpcClient struct {
 	conn      *grpc.ClientConn
 	svc       arkv1.ArkServiceClient
-	eventsCh  chan client.RoundEventChannel
 	treeCache *utils.Cache[tree.CongestionTree]
 }
 
@@ -48,10 +48,9 @@ func NewClient(aspUrl string) (client.ASPClient, error) {
 	}
 
 	svc := arkv1.NewArkServiceClient(conn)
-	eventsCh := make(chan client.RoundEventChannel)
 	treeCache := utils.NewCache[tree.CongestionTree]()
 
-	return &grpcClient{conn, svc, eventsCh, treeCache}, nil
+	return &grpcClient{conn, svc, treeCache}, nil
 }
 
 func (c *grpcClient) Close() {
@@ -61,34 +60,47 @@ func (c *grpcClient) Close() {
 
 func (a *grpcClient) GetEventStream(
 	ctx context.Context, paymentID string,
-) (<-chan client.RoundEventChannel, error) {
+) (<-chan client.RoundEventChannel, func(), error) {
 	req := &arkv1.GetEventStreamRequest{}
 	stream, err := a.svc.GetEventStream(ctx, req)
 	if err != nil {
-		return nil, err
+		return nil, nil, err
 	}
 
+	eventsCh := make(chan client.RoundEventChannel)
+
 	go func() {
-		defer close(a.eventsCh)
+		defer close(eventsCh)
 
 		for {
-			resp, err := stream.Recv()
+			select {
-			if err != nil {
+			case <-stream.Context().Done():
-				a.eventsCh <- client.RoundEventChannel{Err: err}
 				return
-			}
+			default:
+				resp, err := stream.Recv()
+				if err != nil {
+					eventsCh <- client.RoundEventChannel{Err: err}
+					return
+				}
 
-			ev, err := event{resp}.toRoundEvent()
+				ev, err := event{resp}.toRoundEvent()
-			if err != nil {
+				if err != nil {
-				a.eventsCh <- client.RoundEventChannel{Err: err}
+					eventsCh <- client.RoundEventChannel{Err: err}
-				return
+					return
-			}
+				}
 
-			a.eventsCh <- client.RoundEventChannel{Event: ev}
+				eventsCh <- client.RoundEventChannel{Event: ev}
+			}
 		}
 	}()
 
-	return a.eventsCh, nil
+	closeFn := func() {
+		if err := stream.CloseSend(); err != nil {
+			logrus.Warnf("failed to close stream: %v", err)
+		}
+	}
+
+	return eventsCh, closeFn, nil
 }
 
 func (a *grpcClient) GetInfo(ctx context.Context) (*client.Info, error) {
@@ -184,6 +196,10 @@ func (a *grpcClient) Ping(
 		return nil, err
 	}
 
+	if resp.GetEvent() == nil {
+		return nil, nil
+	}
+
 	return event{resp}.toRoundEvent()
 }
 

pkg/client-sdk/client/rest/client.go

@@ -26,7 +26,6 @@ import (
 
 type restClient struct {
 	svc            ark_service.ClientService
-	eventsCh       chan client.RoundEventChannel
 	requestTimeout time.Duration
 	treeCache      *utils.Cache[tree.CongestionTree]
 }
@@ -39,41 +38,46 @@ func NewClient(aspUrl string) (client.ASPClient, error) {
 	if err != nil {
 		return nil, err
 	}
-	eventsCh := make(chan client.RoundEventChannel)
 	reqTimeout := 15 * time.Second
 	treeCache := utils.NewCache[tree.CongestionTree]()
 
-	return &restClient{svc, eventsCh, reqTimeout, treeCache}, nil
+	return &restClient{svc, reqTimeout, treeCache}, nil
 }
 
 func (c *restClient) Close() {}
 
 func (a *restClient) GetEventStream(
 	ctx context.Context, paymentID string,
-) (<-chan client.RoundEventChannel, error) {
+) (<-chan client.RoundEventChannel, func(), error) {
+	eventsCh := make(chan client.RoundEventChannel)
+	stopCh := make(chan struct{})
+
 	go func(payID string) {
-		defer close(a.eventsCh)
+		defer close(eventsCh)
+		defer close(stopCh)
 
 		timeout := time.After(a.requestTimeout)
 
 		for {
 			select {
+			case <-stopCh:
+				return
 			case <-timeout:
-				a.eventsCh <- client.RoundEventChannel{
+				eventsCh <- client.RoundEventChannel{
 					Err: fmt.Errorf("timeout reached"),
 				}
 				return
 			default:
 				event, err := a.Ping(ctx, payID)
 				if err != nil {
-					a.eventsCh <- client.RoundEventChannel{
+					eventsCh <- client.RoundEventChannel{
 						Err: err,
 					}
 					return
 				}
 
 				if event != nil {
-					a.eventsCh <- client.RoundEventChannel{
+					eventsCh <- client.RoundEventChannel{
 						Event: event,
 					}
 				}
@@ -83,7 +87,11 @@ func (a *restClient) GetEventStream(
 		}
 	}(paymentID)
 
-	return a.eventsCh, nil
+	close := func() {
+		stopCh <- struct{}{}
+	}
+
+	return eventsCh, close, nil
 }
 
 func (a *restClient) GetInfo(

pkg/client-sdk/covenant_client.go

@@ -1011,7 +1011,7 @@ func (a *covenantArkClient) handleRoundStream(
 	boardingDescriptor string,
 	receivers []client.Output,
 ) (string, error) {
-	eventsCh, err := a.client.GetEventStream(ctx, paymentID)
+	eventsCh, close, err := a.client.GetEventStream(ctx, paymentID)
 	if err != nil {
 		return "", err
 	}
@@ -1021,7 +1021,10 @@ func (a *covenantArkClient) handleRoundStream(
 		pingStop = a.ping(ctx, paymentID)
 	}
 
-	defer pingStop()
+	defer func() {
+		pingStop()
+		close()
+	}()
 
 	for {
 		select {

pkg/client-sdk/covenantless_client.go

@@ -1094,7 +1094,7 @@ func (a *covenantlessArkClient) handleRoundStream(
 	receivers []client.Output,
 	roundEphemeralKey *secp256k1.PrivateKey,
 ) (string, error) {
-	eventsCh, err := a.client.GetEventStream(ctx, paymentID)
+	eventsCh, close, err := a.client.GetEventStream(ctx, paymentID)
 	if err != nil {
 		return "", err
 	}
@@ -1104,7 +1104,10 @@ func (a *covenantlessArkClient) handleRoundStream(
 		pingStop = a.ping(ctx, paymentID)
 	}
 
-	defer pingStop()
+	defer func() {
+		pingStop()
+		close()
+	}()
 
 	var signerSession bitcointree.SignerSession
 
@@ -1120,14 +1123,16 @@ func (a *covenantlessArkClient) handleRoundStream(
 	for {
 		select {
 		case <-ctx.Done():
-			return "", ctx.Err()
+			return "", fmt.Errorf("context done %s", ctx.Err())
 		case notify := <-eventsCh:
 			if notify.Err != nil {
-				return "", err
+				return "", notify.Err
 			}
-
 			switch event := notify.Event; event.(type) {
 			case client.RoundFinalizedEvent:
+				if step != roundFinalization {
+					continue
+				}
 				return event.(client.RoundFinalizedEvent).Txid, nil
 			case client.RoundFailedEvent:
 				return "", fmt.Errorf("round failed: %s", event.(client.RoundFailedEvent).Reason)

server/internal/core/application/covenantless.go

@@ -643,7 +643,6 @@ func (s *covenantlessService) RegisterCosignerNonces(
 	if err != nil {
 		return fmt.Errorf("failed to decode nonces: %s", err)
 	}
-
 	session.lock.Lock()
 	defer session.lock.Unlock()
 
@@ -654,7 +653,9 @@ func (s *covenantlessService) RegisterCosignerNonces(
 	session.nonces[pubkey] = nonces
 
 	if len(session.nonces) == session.nbCosigners-1 { // exclude the ASP
-		session.nonceDoneC <- struct{}{}
+		go func() {
+			session.nonceDoneC <- struct{}{}
+		}()
 	}
 
 	return nil
@@ -683,7 +684,9 @@ func (s *covenantlessService) RegisterCosignerSignatures(
 	session.signatures[pubkey] = signatures
 
 	if len(session.signatures) == session.nbCosigners-1 { // exclude the ASP
-		session.sigDoneC <- struct{}{}
+		go func() {
+			session.sigDoneC <- struct{}{}
+		}()
 	}
 
 	return nil
@@ -1078,7 +1081,6 @@ func (s *covenantlessService) finalizeRound() {
 	txid, err := s.wallet.BroadcastTransaction(ctx, signedRoundTx)
 	if err != nil {
 		changes = round.Fail(fmt.Errorf("failed to broadcast pool tx: %s", err))
-		log.WithError(err).Warn("failed to broadcast pool tx")
 		return
 	}
 

server/internal/core/application/utils.go

@@ -63,7 +63,27 @@ func (m *paymentsMap) push(payment domain.Payment, boardingInputs []ports.Boardi
 	defer m.lock.Unlock()
 
 	if _, ok := m.payments[payment.Id]; ok {
-		return fmt.Errorf("duplicated inputs")
+		return fmt.Errorf("duplicated payment %s", payment.Id)
+	}
+
+	for _, input := range payment.Inputs {
+		for _, pay := range m.payments {
+			for _, pInput := range pay.Inputs {
+				if input.VtxoKey.Txid == pInput.VtxoKey.Txid && input.VtxoKey.VOut == pInput.VtxoKey.VOut {
+					return fmt.Errorf("duplicated input, %s:%d already used by payment %s", input.VtxoKey.Txid, input.VtxoKey.VOut, pay.Id)
+				}
+			}
+		}
+	}
+
+	for _, input := range boardingInputs {
+		for _, pay := range m.payments {
+			for _, pBoardingInput := range pay.boardingInputs {
+				if input.Txid == pBoardingInput.Txid && input.VOut == pBoardingInput.VOut {
+					return fmt.Errorf("duplicated boarding input, %s:%d already used by payment %s", input.Txid, input.VOut, pay.Id)
+				}
+			}
+		}
 	}
 
 	m.payments[payment.Id] = &timedPayment{payment, boardingInputs, time.Now(), time.Time{}}

server/internal/infrastructure/tx-builder/covenantless/builder.go

@@ -613,6 +613,7 @@ func (b *txBuilder) BuildAsyncPaymentTransactions(
 	}, nil
 }
 
+// TODO use lnd CoinSelect to craft the pool tx
 func (b *txBuilder) createPoolTx(
 	sharedOutputAmount int64,
 	sharedOutputScript []byte,
@@ -857,27 +858,33 @@ func (b *txBuilder) createPoolTx(
 				return nil, err
 			}
 
+			dust = 0
 			if change > 0 {
-				address, err := b.wallet.DeriveAddresses(ctx, 1)
+				if change < dustLimit {
-				if err != nil {
+					dust = change
-					return nil, err
+					change = 0
-				}
+				} else {
+					address, err := b.wallet.DeriveAddresses(ctx, 1)
+					if err != nil {
+						return nil, err
+					}
 
-				addr, err := btcutil.DecodeAddress(address[0], b.onchainNetwork())
+					addr, err := btcutil.DecodeAddress(address[0], b.onchainNetwork())
-				if err != nil {
+					if err != nil {
-					return nil, err
+						return nil, err
-				}
+					}
 
-				aspScript, err := txscript.PayToAddrScript(addr)
+					aspScript, err := txscript.PayToAddrScript(addr)
-				if err != nil {
+					if err != nil {
-					return nil, err
+						return nil, err
-				}
+					}
 
-				ptx.UnsignedTx.AddTxOut(&wire.TxOut{
+					ptx.UnsignedTx.AddTxOut(&wire.TxOut{
-					Value:    int64(change),
+						Value:    int64(change),
-					PkScript: aspScript,
+						PkScript: aspScript,
-				})
+					})
-				ptx.Outputs = append(ptx.Outputs, psbt.POutput{})
+					ptx.Outputs = append(ptx.Outputs, psbt.POutput{})
+				}
 			}
 
 			for _, utxo := range newUtxos {
@@ -910,6 +917,34 @@ func (b *txBuilder) createPoolTx(
 				}
 			}
 
+			b64, err = ptx.B64Encode()
+			if err != nil {
+				return nil, err
+			}
+
+			feeAmount, err = b.wallet.EstimateFees(ctx, b64)
+			if err != nil {
+				return nil, err
+			}
+
+			if dust > feeAmount {
+				feeAmount = dust
+			} else {
+				feeAmount += dust
+			}
+
+			if dust == 0 {
+				if feeAmount == change {
+					// fees = change, remove change output
+					ptx.UnsignedTx.TxOut = ptx.UnsignedTx.TxOut[:len(ptx.UnsignedTx.TxOut)-1]
+					ptx.Outputs = ptx.Outputs[:len(ptx.Outputs)-1]
+				} else if feeAmount < change {
+					// change covers the fees, reduce change amount
+					ptx.UnsignedTx.TxOut[len(ptx.Outputs)-1].Value = int64(change - feeAmount)
+				} else {
+					return nil, fmt.Errorf("change is not enough to cover fees")
+				}
+			}
 		}
 	} else if feeAmount-dust > 0 {
 		newUtxos, change, err := b.selectUtxos(ctx, sweptRounds, feeAmount-dust)
@@ -917,8 +952,12 @@ func (b *txBuilder) createPoolTx(
 			return nil, err
 		}
 
+		dust = 0
 		if change > 0 {
-			if change > dustLimit {
+			if change < dustLimit {
+				dust = change
+				change = 0
+			} else {
 				address, err := b.wallet.DeriveAddresses(ctx, 1)
 				if err != nil {
 					return nil, err
@@ -971,6 +1010,35 @@ func (b *txBuilder) createPoolTx(
 				return nil, err
 			}
 		}
+
+		b64, err = ptx.B64Encode()
+		if err != nil {
+			return nil, err
+		}
+
+		feeAmount, err = b.wallet.EstimateFees(ctx, b64)
+		if err != nil {
+			return nil, err
+		}
+
+		if dust > feeAmount {
+			feeAmount = dust
+		} else {
+			feeAmount += dust
+		}
+
+		if dust == 0 {
+			if feeAmount == change {
+				// fees = change, remove change output
+				ptx.UnsignedTx.TxOut = ptx.UnsignedTx.TxOut[:len(ptx.UnsignedTx.TxOut)-1]
+				ptx.Outputs = ptx.Outputs[:len(ptx.Outputs)-1]
+			} else if feeAmount < change {
+				// change covers the fees, reduce change amount
+				ptx.UnsignedTx.TxOut[len(ptx.Outputs)-1].Value = int64(change - feeAmount)
+			} else {
+				return nil, fmt.Errorf("change is not enough to cover fees")
+			}
+		}
 	}
 
 	// remove input taproot leaf script

server/internal/infrastructure/tx-builder/covenantless/builder_test.go

@@ -22,6 +22,7 @@ import (
 const (
 	testingKey        = "020000000000000000000000000000000000000000000000000000000000000001"
 	connectorAddress  = "bc1py00yhcjpcj0k0sqra0etq0u3yy0purmspppsw0shyzyfe8c83tmq5h6kc2"
+	changeAddress     = "bcrt1qhhq55mut9easvrncy4se8q6vg3crlug7yj4j56"
 	roundLifetime     = int64(1209344)
 	boardingExitDelay = int64(512)
 	minRelayFeeRate   = 3
@@ -37,7 +38,9 @@ func TestMain(m *testing.M) {
 	wallet.On("EstimateFees", mock.Anything, mock.Anything).
 		Return(uint64(100), nil)
 	wallet.On("SelectUtxos", mock.Anything, mock.Anything, mock.Anything).
-		Return(randomInput, uint64(0), nil)
+		Return(randomInput, uint64(1000), nil)
+	wallet.On("DeriveAddresses", mock.Anything, mock.Anything).
+		Return([]string{changeAddress}, nil)
 	wallet.On("DeriveConnectorAddress", mock.Anything).
 		Return(connectorAddress, nil)
 	wallet.On("MinRelayFee", mock.Anything, mock.Anything).

server/internal/infrastructure/wallet/btc-embedded/wallet.go

@@ -556,6 +556,18 @@ func (s *service) SelectUtxos(ctx context.Context, _ string, amount uint64) ([]p
 		return nil, 0, fmt.Errorf("insufficient funds to select %d, only %d available", amount, selectedAmount)
 	}
 
+	for _, utxo := range selectedUtxos {
+		if _, err := w.LeaseOutput(
+			wtxmgr.LockID(utxo.(coinTxInput).Hash),
+			wire.OutPoint{
+				Hash:  utxo.(coinTxInput).Hash,
+				Index: utxo.(coinTxInput).Index,
+			},
+			outputLockDuration,
+		); err != nil {
+			return nil, 0, err
+		}
+	}
 	return selectedUtxos, selectedAmount - amount, nil
 }
 

server/internal/interface/grpc/handlers/arkservice.go

@@ -16,8 +16,9 @@ import (
 )
 
 type listener struct {
-	id string
+	id   string
-	ch chan *arkv1.GetEventStreamResponse
+	done chan struct{}
+	ch   chan *arkv1.GetEventStreamResponse
 }
 
 type handler struct {
@@ -300,21 +301,25 @@ func (h *handler) GetRoundById(
 }
 
 func (h *handler) GetEventStream(_ *arkv1.GetEventStreamRequest, stream arkv1.ArkService_GetEventStreamServer) error {
+	doneCh := make(chan struct{})
+
 	listener := &listener{
-		id: uuid.NewString(),
+		id:   uuid.NewString(),
-		ch: make(chan *arkv1.GetEventStreamResponse),
+		done: doneCh,
+		ch:   make(chan *arkv1.GetEventStreamResponse),
 	}
 
+	h.pushListener(listener)
 	defer h.removeListener(listener.id)
 	defer close(listener.ch)
-
+	defer close(doneCh)
-	h.pushListener(listener)
 
 	for {
 		select {
 		case <-stream.Context().Done():
 			return nil
-
+		case <-doneCh:
+			return nil
 		case ev := <-listener.ch:
 			if err := stream.Send(ev); err != nil {
 				return err
@@ -476,6 +481,7 @@ func (h *handler) listenToEvents() {
 	channel := h.svc.GetEventsChannel(context.Background())
 	for event := range channel {
 		var ev *arkv1.GetEventStreamResponse
+		shouldClose := false
 
 		switch e := event.(type) {
 		case domain.RoundFinalizationStarted:
@@ -491,6 +497,7 @@ func (h *handler) listenToEvents() {
 				},
 			}
 		case domain.RoundFinalized:
+			shouldClose = true
 			ev = &arkv1.GetEventStreamResponse{
 				Event: &arkv1.GetEventStreamResponse_RoundFinalized{
 					RoundFinalized: &arkv1.RoundFinalizedEvent{
@@ -500,6 +507,7 @@ func (h *handler) listenToEvents() {
 				},
 			}
 		case domain.RoundFailed:
+			shouldClose = true
 			ev = &arkv1.GetEventStreamResponse{
 				Event: &arkv1.GetEventStreamResponse_RoundFailed{
 					RoundFailed: &arkv1.RoundFailed{
@@ -542,8 +550,14 @@ func (h *handler) listenToEvents() {
 		}
 
 		if ev != nil {
-			for _, listener := range h.listeners {
+			logrus.Debugf("forwarding event to %d listeners", len(h.listeners))
-				listener.ch <- ev
+			for _, l := range h.listeners {
+				go func(l *listener) {
+					l.ch <- ev
+					if shouldClose {
+						l.done <- struct{}{}
+					}
+				}(l)
 			}
 		}
 	}

server/test/e2e/covenantless/e2e_test.go

@@ -283,6 +283,85 @@ func TestReactToAsyncSpentVtxosRedemption(t *testing.T) {
 	})
 }
 
+func TestAliceSeveralPaymentsToBob(t *testing.T) {
+	ctx := context.Background()
+	alice, grpcAlice := setupArkSDK(t)
+	defer grpcAlice.Close()
+
+	bob, grpcBob := setupArkSDK(t)
+	defer grpcBob.Close()
+
+	_, boardingAddress, err := alice.Receive(ctx)
+	require.NoError(t, err)
+
+	_, err = utils.RunCommand("nigiri", "faucet", boardingAddress)
+	require.NoError(t, err)
+
+	time.Sleep(5 * time.Second)
+
+	_, err = alice.Claim(ctx)
+	require.NoError(t, err)
+
+	bobAddress, _, err := bob.Receive(ctx)
+	require.NoError(t, err)
+
+	_, err = alice.SendOffChain(ctx, false, []arksdk.Receiver{arksdk.NewBitcoinReceiver(bobAddress, 1000)})
+	require.NoError(t, err)
+
+	time.Sleep(2 * time.Second)
+
+	bobVtxos, _, err := bob.ListVtxos(ctx)
+	require.NoError(t, err)
+	require.Len(t, bobVtxos, 1)
+
+	_, err = bob.Claim(ctx)
+	require.NoError(t, err)
+
+	_, err = alice.Claim(ctx)
+	require.NoError(t, err)
+
+	_, err = alice.SendOffChain(ctx, false, []arksdk.Receiver{arksdk.NewBitcoinReceiver(bobAddress, 10000)})
+	require.NoError(t, err)
+
+	time.Sleep(2 * time.Second)
+
+	bobVtxos, _, err = bob.ListVtxos(ctx)
+	require.NoError(t, err)
+	require.Len(t, bobVtxos, 2)
+
+	_, err = alice.SendOffChain(ctx, false, []arksdk.Receiver{arksdk.NewBitcoinReceiver(bobAddress, 10000)})
+	require.NoError(t, err)
+
+	time.Sleep(2 * time.Second)
+
+	bobVtxos, _, err = bob.ListVtxos(ctx)
+	require.NoError(t, err)
+	require.Len(t, bobVtxos, 3)
+
+	_, err = alice.SendAsync(ctx, false, []arksdk.Receiver{arksdk.NewBitcoinReceiver(bobAddress, 10000)})
+	require.NoError(t, err)
+
+	time.Sleep(2 * time.Second)
+
+	bobVtxos, _, err = bob.ListVtxos(ctx)
+	require.NoError(t, err)
+	require.Len(t, bobVtxos, 4)
+
+	_, err = alice.Claim(ctx)
+	require.NoError(t, err)
+
+	// bobVtxos should be unique
+	uniqueVtxos := make(map[string]struct{})
+	for _, v := range bobVtxos {
+		uniqueVtxos[fmt.Sprintf("%s:%d", v.Txid, v.VOut)] = struct{}{}
+	}
+	require.Len(t, uniqueVtxos, 4)
+
+	_, err = bob.Claim(ctx)
+	require.NoError(t, err)
+
+}
+
 func runClarkCommand(arg ...string) (string, error) {
 	args := append([]string{"exec", "-t", "clarkd", "ark"}, arg...)
 	return utils.RunCommand("docker", args...)
@@ -357,43 +436,16 @@ func setupAspWallet() error {
 		return fmt.Errorf("failed to parse response: %s", err)
 	}
 
-	_, err = utils.RunCommand("nigiri", "faucet", addr.Address)
+	const numberOfFaucet = 15 // must cover the liquidity needed for all tests
-	if err != nil {
-		return fmt.Errorf("failed to fund wallet: %s", err)
-	}
 
-	_, err = utils.RunCommand("nigiri", "faucet", addr.Address)
+	for i := 0; i < numberOfFaucet; i++ {
-	if err != nil {
+		_, err = utils.RunCommand("nigiri", "faucet", addr.Address)
-		return fmt.Errorf("failed to fund wallet: %s", err)
+		if err != nil {
-	}
+			return fmt.Errorf("failed to fund wallet: %s", err)
-
+		}
-	_, err = utils.RunCommand("nigiri", "faucet", addr.Address)
-	if err != nil {
-		return fmt.Errorf("failed to fund wallet: %s", err)
-	}
-
-	_, err = utils.RunCommand("nigiri", "faucet", addr.Address)
-	if err != nil {
-		return fmt.Errorf("failed to fund wallet: %s", err)
-	}
-
-	_, err = utils.RunCommand("nigiri", "faucet", addr.Address)
-	if err != nil {
-		return fmt.Errorf("failed to fund wallet: %s", err)
-	}
-
-	_, err = utils.RunCommand("nigiri", "faucet", addr.Address)
-	if err != nil {
-		return fmt.Errorf("failed to fund wallet: %s", err)
-	}
-
-	_, err = utils.RunCommand("nigiri", "faucet", addr.Address)
-	if err != nil {
-		return fmt.Errorf("failed to fund wallet: %s", err)
 	}
 
 	time.Sleep(5 * time.Second)
-
 	return nil
 }