electrum: proper log output of tx watcher

To optimise the log output of the transaction watcher,
only output to the required debug log.
Also, swap.ErrDoesNotExist is now handled in the 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 {
+		if err != nil && !errors.Is(err, swap.ErrSwapDoesNotExist) {
 			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,11 @@ 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",
-			o.txID.String(), getHeight(hs, o.txID).Height(), currentHeight.Height())
+		return false, nil
 	}
 	return true, o.cb(o.swapID.String(), rawTx, nil)
 }
@@ -139,11 +140,11 @@ 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",
-			o.txID.String(), getHeight(hs, o.txID).Height(), currentHeight.Height())
+		return false, nil
 	}
 	return true, o.cb(o.swapID.String())
 }

lwk/electrumtxwatcher.go

@@ -64,7 +64,7 @@ func (r *electrumTxWatcher) StartWatchingTxs() error {
 				r.mu.Lock()
 				r.blockHeight = electrum.BlocKHeight(blockHeader.Height)
 				r.mu.Unlock()
-				log.Infof("New block received. block height:%d", r.blockHeight)
+				log.Debugf("New block received. block height:%d", r.blockHeight)
 				err = r.subscriber.Update(ctx, r.blockHeight)
 				if err != nil {
 					log.Infof("Error notifying tx observers: %v", err)

lwk/lwkwallet.go

@@ -35,7 +35,8 @@ type SatPerKVByte float64
 func SatPerKVByteFromFeeBTCPerKb(feeBTCPerKb float64) SatPerKVByte {
 	s := SatPerKVByte(feeBTCPerKb * math.Pow10(btcToSatoshiExp) / kb)
 	if s < minimumSatPerByte {
-		log.Infof("using minimum fee: %v.", minimumSatPerByte)
+		log.Debugf("Using minimum fee rate of %v sat/kw",
+			minimumSatPerByte)
 		return minimumSatPerByte
 	}
 	return s