electrum: log tx confirmation progress

Changed so that transaction confirmations are
not output as errors but added to the debug log.
Handle swap.ErrDoesNotExist in callback.

electrum/block_subscriber.go

@@ -2,9 +2,11 @@ package electrum
 
 import (
 	"context"
+	"errors"
 	"sync"
 
 	"github.com/elementsproject/peerswap/log"
+	"github.com/elementsproject/peerswap/swap"
 )
 
 type BlocKHeight uint32
@@ -55,13 +57,21 @@ func (h *liquidBlockHeaderSubscriber) Update(ctx context.Context, blockHeight Bl
 	defer h.mu.Unlock()
 	for _, observer := range h.txObservers {
 		callbacked, err := observer.Callback(ctx, blockHeight)
-		if callbacked && err == nil {
-			// callbacked and no error, remove observer
-			h.Deregister(observer)
+		if callbacked {
+			if err == nil || errors.Is(err, swap.ErrSwapDoesNotExist) {
+				// callbacked and no error, remove observer
+				h.Deregister(observer)
+			}
 		}
 		if err != nil {
 			log.Infof("Error in callback: %v", err)
 		}
 	}
 	return nil
 }
+
+func (h *liquidBlockHeaderSubscriber) Count() int {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	return len(h.txObservers)
+}

electrum/block_subscriber_test.go

@@ -9,14 +9,22 @@ import (
 )
 
 type testtxo struct {
-	swapid swap.SwapId
+	swapid    swap.SwapId
+	getSwapID func() swap.SwapId
+	callback  func(context.Context, BlocKHeight) (bool, error)
 }
 
 func (t *testtxo) GetSwapID() swap.SwapId {
+	if t.getSwapID != nil {
+		return t.getSwapID()
+	}
 	return t.swapid
 }
 
-func (t *testtxo) Callback(context.Context, BlocKHeight) (bool, error) {
+func (t *testtxo) Callback(ctx context.Context, b BlocKHeight) (bool, error) {
+	if t.callback != nil {
+		return t.callback(ctx, b)
+	}
 	return true, nil
 }
 
@@ -51,3 +59,57 @@ func TestConcurrentUpdate(t *testing.T) {
 		t.Errorf("Expected length %d, but got %d", 0, len(h.txObservers))
 	}
 }
+
+func Test_liquidBlockHeaderSubscriber_Update(t *testing.T) {
+	t.Parallel()
+	var blockHeight BlocKHeight = 10
+	tests := map[string]struct {
+		txObservers []TXObserver
+		count       int
+		wantErr     bool
+	}{
+		"no observers": {
+			txObservers: nil,
+		},
+		"observers": {
+			txObservers: []TXObserver{
+				&testtxo{},
+				&testtxo{},
+			},
+		},
+		"swap does not exists": {
+			txObservers: []TXObserver{
+				&testtxo{
+					callback: func(context.Context, BlocKHeight) (bool, error) {
+						return true, swap.ErrSwapDoesNotExist
+					},
+				},
+			},
+		},
+		"error in callback": {
+			txObservers: []TXObserver{
+				&testtxo{
+					callback: func(context.Context, BlocKHeight) (bool, error) {
+						return true, swap.ErrEventRejected
+					},
+				},
+			},
+			count: 1,
+		},
+	}
+
+	for name, tt := range tests {
+		t.Run(name, func(t *testing.T) {
+			t.Parallel()
+			h := &liquidBlockHeaderSubscriber{
+				txObservers: tt.txObservers,
+			}
+			if err := h.Update(context.Background(), blockHeight); (err != nil) != tt.wantErr {
+				t.Errorf("liquidBlockHeaderSubscriber.Update() error = %v, wantErr %v", err, tt.wantErr)
+			}
+			if len(h.txObservers) != tt.count {
+				t.Errorf("Expected length %d, but got %d", tt.count, len(h.txObservers))
+			}
+		})
+	}
+}

electrum/tx_observer.go

@@ -8,6 +8,7 @@ import (
 	"github.com/btcsuite/btcd/chaincfg/chainhash"
 	"github.com/btcsuite/btcd/txscript"
 	"github.com/checksum0/go-electrum/electrum"
+	"github.com/elementsproject/peerswap/log"
 	"github.com/elementsproject/peerswap/onchain"
 	"github.com/elementsproject/peerswap/swap"
 )
@@ -93,11 +94,13 @@ func (o *observeOpeningTX) Callback(ctx context.Context, currentHeight BlocKHeig
 	}
 	rawTx, err := o.electrumClient.GetRawTransaction(ctx, o.txID.String())
 	if err != nil {
-		return false, fmt.Errorf("failed to get raw transaction: %w", err)
+		log.Debugf("failed to get raw transaction: %s", o.txID.String())
+		return false, nil
 	}
 	if !(currentHeight.Height() >= getHeight(hs, o.txID).Height()+uint32(onchain.LiquidConfs)-1) {
-		return false, fmt.Errorf("not enough confirmations for opening transaction. txhash: %s.height: %d, current: %d",
+		log.Debugf("not enough confirmations for opening transaction. txhash: %s.height: %d, current: %d",
 			o.txID.String(), getHeight(hs, o.txID).Height(), currentHeight.Height())
+		return false, nil
 	}
 	return true, o.cb(o.swapID.String(), rawTx, nil)
 }
@@ -139,11 +142,13 @@ func (o *observeCSVTX) Callback(ctx context.Context, currentHeight BlocKHeight)
 		return false, fmt.Errorf("failed to get history: %w", err)
 	}
 	if !(getHeight(hs, o.txID).Confirmed()) {
-		return false, fmt.Errorf("the transaction is unconfirmed")
+		log.Debugf("the transaction is unconfirmed. txhash: %s", o.txID.String())
+		return false, nil
 	}
 	if !(currentHeight.Height() >= getHeight(hs, o.txID).Height()+uint32(onchain.LiquidCsv-1)) {
-		return false, fmt.Errorf("not enough confirmations for csv transaction. txhash: %s.height: %d, current: %d",
+		log.Debugf("not enough confirmations for csv transaction. txhash: %s.height: %d, current: %d",
 			o.txID.String(), getHeight(hs, o.txID).Height(), currentHeight.Height())
+		return false, nil
 	}
 	return true, o.cb(o.swapID.String())
 }