From 6a32b4613234358ccd3cb4d0a5dc362e0bb7f6a8 Mon Sep 17 00:00:00 2001
From: Boris Nagaev <bnagaev@gmail.com>
Date: Tue, 8 Apr 2025 12:02:23 -0300
Subject: [PATCH 01/13] sweepbatcher: mark channels receive- or send-only

---
 sweepbatcher/sweep_batcher.go | 6 +++---
 1 file changed, 3 insertions(+), 3 deletions(-)

diff --git a/sweepbatcher/sweep_batcher.go b/sweepbatcher/sweep_batcher.go
index 03b697b1..6ab67035 100644
--- a/sweepbatcher/sweep_batcher.go
+++ b/sweepbatcher/sweep_batcher.go
@@ -307,13 +307,13 @@ type SpendDetail struct {
 // that the sweep was successful.
 type SpendNotifier struct {
 	// SpendChan is a channel where the spend details are received.
-	SpendChan chan *SpendDetail
+	SpendChan chan<- *SpendDetail
 
 	// SpendErrChan is a channel where spend errors are received.
-	SpendErrChan chan error
+	SpendErrChan chan<- error
 
 	// QuitChan is a channel that can be closed to stop the notifier.
-	QuitChan chan bool
+	QuitChan <-chan bool
 }
 
 var (

From 501faa83756a14ea27c46173e66156810986d817 Mon Sep 17 00:00:00 2001
From: Boris Nagaev <bnagaev@gmail.com>
Date: Tue, 8 Apr 2025 12:34:42 -0300
Subject: [PATCH 02/13] sweepbatcher: unblock sending if notifier quits

---
 sweepbatcher/sweep_batcher.go | 18 ++++++++++++++++++
 1 file changed, 18 insertions(+)

diff --git a/sweepbatcher/sweep_batcher.go b/sweepbatcher/sweep_batcher.go
index 6ab67035..c5498c46 100644
--- a/sweepbatcher/sweep_batcher.go
+++ b/sweepbatcher/sweep_batcher.go
@@ -1176,7 +1176,15 @@ func (b *Batcher) monitorSpendAndNotify(ctx context.Context, sweep *sweep,
 				}
 
 				select {
+				// Try to write the update to the notification
+				// channel.
 				case notifier.SpendChan <- spendDetail:
+
+				// If a quit signal was provided by the swap,
+				// continue.
+				case <-notifier.QuitChan:
+
+				// If the context was canceled, stop.
 				case <-ctx.Done():
 				}
 
@@ -1184,7 +1192,15 @@ func (b *Batcher) monitorSpendAndNotify(ctx context.Context, sweep *sweep,
 
 			case err := <-spendErr:
 				select {
+				// Try to write the error to the notification
+				// channel.
 				case notifier.SpendErrChan <- err:
+
+				// If a quit signal was provided by the swap,
+				// continue.
+				case <-notifier.QuitChan:
+
+				// If the context was canceled, stop.
 				case <-ctx.Done():
 				}
 
@@ -1194,9 +1210,11 @@ func (b *Batcher) monitorSpendAndNotify(ctx context.Context, sweep *sweep,
 
 				return
 
+			// If a quit signal was provided by the swap, continue.
 			case <-notifier.QuitChan:
 				return
 
+			// If the context was canceled, stop.
 			case <-ctx.Done():
 				return
 			}

From 39f859a04655c1c541fcea22ccec299416b33ce5 Mon Sep 17 00:00:00 2001
From: Boris Nagaev <bnagaev@gmail.com>
Date: Tue, 8 Apr 2025 12:36:03 -0300
Subject: [PATCH 03/13] sweepbatcher: send spend error to notifier

---
 sweepbatcher/sweep_batch.go | 39 +++++++++++++++++++++++++++++++++++++
 1 file changed, 39 insertions(+)

diff --git a/sweepbatcher/sweep_batch.go b/sweepbatcher/sweep_batch.go
index a437461a..15e7d006 100644
--- a/sweepbatcher/sweep_batch.go
+++ b/sweepbatcher/sweep_batch.go
@@ -1834,6 +1834,8 @@ func (b *batch) monitorSpend(ctx context.Context, primarySweep sweep) error {
 				return
 
 			case err := <-spendErr:
+				b.writeToSpendErrChan(ctx, err)
+
 				b.writeToErrChan(
 					fmt.Errorf("spend error: %w", err),
 				)
@@ -2285,6 +2287,43 @@ func (b *batch) writeToErrChan(err error) {
 	}
 }
 
+// writeToSpendErrChan sends an error to spend error channels of all the sweeps.
+func (b *batch) writeToSpendErrChan(ctx context.Context, spendErr error) {
+	done, err := b.scheduleNextCall()
+	if err != nil {
+		done()
+
+		return
+	}
+	notifiers := make([]*SpendNotifier, 0, len(b.sweeps))
+	for _, s := range b.sweeps {
+		// If the sweep's notifier is empty then this means that a swap
+		// is not waiting to read an update from it, so we can skip
+		// the notification part.
+		if s.notifier == nil || s.notifier.SpendErrChan == nil {
+			continue
+		}
+
+		notifiers = append(notifiers, s.notifier)
+	}
+	done()
+
+	for _, notifier := range notifiers {
+		select {
+		// Try to write the error to the notification
+		// channel.
+		case notifier.SpendErrChan <- spendErr:
+
+		// If a quit signal was provided by the swap,
+		// continue.
+		case <-notifier.QuitChan:
+
+		// If the context was canceled, stop.
+		case <-ctx.Done():
+		}
+	}
+}
+
 func (b *batch) persistSweep(ctx context.Context, sweep sweep,
 	completed bool) error {
 

From 7636ffdc6477163001fa90fc0e4b489eb12fa595 Mon Sep 17 00:00:00 2001
From: Boris Nagaev <bnagaev@gmail.com>
Date: Tue, 8 Apr 2025 15:33:05 -0300
Subject: [PATCH 04/13] test: fix error messages

---
 test/context.go | 10 +++++-----
 1 file changed, 5 insertions(+), 5 deletions(-)

diff --git a/test/context.go b/test/context.go
index 5fae5935..943a1dee 100644
--- a/test/context.go
+++ b/test/context.go
@@ -61,7 +61,7 @@ func (ctx *Context) NotifySpend(tx *wire.MsgTx, inputIndex uint32) {
 		SpenderInputIndex: inputIndex,
 	}:
 	case <-time.After(Timeout):
-		ctx.T.Fatalf("htlc spend not consumed")
+		ctx.T.Fatalf("spend not consumed")
 	}
 }
 
@@ -74,7 +74,7 @@ func (ctx *Context) NotifyConf(tx *wire.MsgTx) {
 		Tx: tx,
 	}:
 	case <-time.After(Timeout):
-		ctx.T.Fatalf("htlc spend not consumed")
+		ctx.T.Fatalf("confirmation not consumed")
 	}
 }
 
@@ -86,7 +86,7 @@ func (ctx *Context) AssertRegisterSpendNtfn(script []byte) {
 	case spendIntent := <-ctx.Lnd.RegisterSpendChannel:
 		require.Equal(
 			ctx.T, script, spendIntent.PkScript,
-			"server not listening for published htlc script",
+			"server not listening for published script",
 		)
 
 	case <-time.After(Timeout):
@@ -134,7 +134,7 @@ func (ctx *Context) AssertRegisterConf(expectTxHash bool, confs int32) *ConfRegi
 		require.Equal(ctx.T, confs, confIntent.NumConfs)
 
 	case <-time.After(Timeout):
-		ctx.T.Fatalf("htlc confirmed not subscribed to")
+		ctx.T.Fatalf("tx confirmed not subscribed to")
 	}
 
 	return confIntent
@@ -249,7 +249,7 @@ func (ctx *Context) GetOutputIndex(tx *wire.MsgTx,
 		}
 	}
 
-	ctx.T.Fatal("htlc not present in tx")
+	ctx.T.Fatal("the output not present in tx")
 	return 0
 }
 

From c036c2adbdbe04e951032c1ebadf5b67a895186b Mon Sep 17 00:00:00 2001
From: Boris Nagaev <bnagaev@gmail.com>
Date: Tue, 8 Apr 2025 18:07:13 -0300
Subject: [PATCH 05/13] sweepbatcher: fix mock TotalSweptAmount

Forgot to return the calculated total value.
---
 sweepbatcher/store_mock.go | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/sweepbatcher/store_mock.go b/sweepbatcher/store_mock.go
index 90896aac..0f6c5e3c 100644
--- a/sweepbatcher/store_mock.go
+++ b/sweepbatcher/store_mock.go
@@ -212,5 +212,5 @@ func (s *StoreMock) TotalSweptAmount(ctx context.Context, batchID int32) (
 		}
 	}
 
-	return 0, nil
+	return total, nil
 }

From cb7c31c1b94488769cc83a8794f930031411f52d Mon Sep 17 00:00:00 2001
From: Boris Nagaev <bnagaev@gmail.com>
Date: Tue, 8 Apr 2025 18:30:35 -0300
Subject: [PATCH 06/13] sweepbatcher: fix mistake in batch reading from DB

---
 sweepbatcher/store.go | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/sweepbatcher/store.go b/sweepbatcher/store.go
index 01b9e74a..bb245e5f 100644
--- a/sweepbatcher/store.go
+++ b/sweepbatcher/store.go
@@ -259,7 +259,7 @@ func convertBatchRow(row sqlc.SweepBatch) *dbBatch {
 	}
 
 	if row.Confirmed {
-		batch.State = batchOpen
+		batch.State = batchConfirmed
 	}
 
 	if row.BatchTxID.Valid {

From 2f05934dbb7482fac4cad474363f944927c3cf66 Mon Sep 17 00:00:00 2001
From: Boris Nagaev <bnagaev@gmail.com>
Date: Tue, 8 Apr 2025 18:40:36 -0300
Subject: [PATCH 07/13] test/chainnotifier_mock: support errors

Support sending errors to error channels returned by RegisterSpendNtfn and
RegisterConfirmationsNtfn.
---
 test/chainnotifier_mock.go | 26 +++++++++++++++++++++++++-
 1 file changed, 25 insertions(+), 1 deletion(-)

diff --git a/test/chainnotifier_mock.go b/test/chainnotifier_mock.go
index 3889ce06..96e72f23 100644
--- a/test/chainnotifier_mock.go
+++ b/test/chainnotifier_mock.go
@@ -36,6 +36,7 @@ type SpendRegistration struct {
 	Outpoint   *wire.OutPoint
 	PkScript   []byte
 	HeightHint int32
+	ErrChan    chan<- error
 }
 
 // ConfRegistration contains registration details.
@@ -45,18 +46,24 @@ type ConfRegistration struct {
 	HeightHint int32
 	NumConfs   int32
 	ConfChan   chan *chainntnfs.TxConfirmation
+	ErrChan    chan<- error
 }
 
 func (c *mockChainNotifier) RegisterSpendNtfn(ctx context.Context,
 	outpoint *wire.OutPoint, pkScript []byte, heightHint int32) (
 	chan *chainntnfs.SpendDetail, chan error, error) {
 
-	c.lnd.RegisterSpendChannel <- &SpendRegistration{
+	spendErrChan := make(chan error, 1)
+
+	reg := &SpendRegistration{
 		HeightHint: heightHint,
 		Outpoint:   outpoint,
 		PkScript:   pkScript,
+		ErrChan:    spendErrChan,
 	}
 
+	c.lnd.RegisterSpendChannel <- reg
+
 	spendChan := make(chan *chainntnfs.SpendDetail, 1)
 	errChan := make(chan error, 1)
 
@@ -70,6 +77,13 @@ func (c *mockChainNotifier) RegisterSpendNtfn(ctx context.Context,
 			case spendChan <- m:
 			case <-ctx.Done():
 			}
+
+		case err := <-spendErrChan:
+			select {
+			case errChan <- err:
+			case <-ctx.Done():
+			}
+
 		case <-ctx.Done():
 		}
 	}()
@@ -129,12 +143,15 @@ func (c *mockChainNotifier) RegisterConfirmationsNtfn(ctx context.Context,
 	opts ...lndclient.NotifierOption) (chan *chainntnfs.TxConfirmation,
 	chan error, error) {
 
+	confErrChan := make(chan error, 1)
+
 	reg := &ConfRegistration{
 		PkScript:   pkScript,
 		TxID:       txid,
 		HeightHint: heightHint,
 		NumConfs:   numConfs,
 		ConfChan:   make(chan *chainntnfs.TxConfirmation, 1),
+		ErrChan:    confErrChan,
 	}
 
 	c.Lock()
@@ -169,6 +186,13 @@ func (c *mockChainNotifier) RegisterConfirmationsNtfn(ctx context.Context,
 				}
 			}
 			c.Unlock()
+
+		case err := <-confErrChan:
+			select {
+			case errChan <- err:
+			case <-ctx.Done():
+			}
+
 		case <-ctx.Done():
 		}
 	}()

From 01664ad6358a9929e56220677f83febbd2fad0c2 Mon Sep 17 00:00:00 2001
From: Boris Nagaev <bnagaev@gmail.com>
Date: Tue, 8 Apr 2025 18:43:21 -0300
Subject: [PATCH 08/13] sweepbatcher: cancel spendCtx after processing

Function monitorSpendAndNotify used to cancel the context passed to
RegisterSpendNtfn right after starting the goroutine processing results.
Spend notifications were missed.

Now the context is canceled when the goroutine finishes.
---
 sweepbatcher/sweep_batcher.go | 6 +++++-
 1 file changed, 5 insertions(+), 1 deletion(-)

diff --git a/sweepbatcher/sweep_batcher.go b/sweepbatcher/sweep_batcher.go
index c5498c46..c8ee03b7 100644
--- a/sweepbatcher/sweep_batcher.go
+++ b/sweepbatcher/sweep_batcher.go
@@ -1128,11 +1128,12 @@ func (b *Batcher) monitorSpendAndNotify(ctx context.Context, sweep *sweep,
 	parentBatchID int32, notifier *SpendNotifier) error {
 
 	spendCtx, cancel := context.WithCancel(ctx)
-	defer cancel()
 
 	// Then we get the total amount that was swept by the batch.
 	totalSwept, err := b.store.TotalSweptAmount(ctx, parentBatchID)
 	if err != nil {
+		cancel()
+
 		return err
 	}
 
@@ -1141,11 +1142,14 @@ func (b *Batcher) monitorSpendAndNotify(ctx context.Context, sweep *sweep,
 		sweep.initiationHeight,
 	)
 	if err != nil {
+		cancel()
+
 		return err
 	}
 
 	b.wg.Add(1)
 	go func() {
+		defer cancel()
 		defer b.wg.Done()
 		infof("Batcher monitoring spend for swap %x",
 			sweep.swapHash[:6])

From fa1bc48a686b070e5e78a2c0a9d6ad83804628ff Mon Sep 17 00:00:00 2001
From: Boris Nagaev <bnagaev@gmail.com>
Date: Tue, 8 Apr 2025 18:48:34 -0300
Subject: [PATCH 09/13] sweepbatcher: remove unneeded for loops

The loop always had exactly one iteration.
---
 sweepbatcher/sweep_batch.go   |  81 ++++++++++--------------
 sweepbatcher/sweep_batcher.go | 116 ++++++++++++++++------------------
 2 files changed, 89 insertions(+), 108 deletions(-)

diff --git a/sweepbatcher/sweep_batch.go b/sweepbatcher/sweep_batch.go
index 15e7d006..ee623659 100644
--- a/sweepbatcher/sweep_batch.go
+++ b/sweepbatcher/sweep_batch.go
@@ -1822,29 +1822,22 @@ func (b *batch) monitorSpend(ctx context.Context, primarySweep sweep) error {
 		b.Infof("monitoring spend for outpoint %s",
 			primarySweep.outpoint.String())
 
-		for {
+		select {
+		case spend := <-spendChan:
 			select {
-			case spend := <-spendChan:
-				select {
-				case b.spendChan <- spend:
-
-				case <-ctx.Done():
-				}
-
-				return
-
-			case err := <-spendErr:
-				b.writeToSpendErrChan(ctx, err)
-
-				b.writeToErrChan(
-					fmt.Errorf("spend error: %w", err),
-				)
-
-				return
+			case b.spendChan <- spend:
 
 			case <-ctx.Done():
-				return
 			}
+
+		case err := <-spendErr:
+			b.writeToSpendErrChan(ctx, err)
+
+			b.writeToErrChan(
+				fmt.Errorf("spend error: %w", err),
+			)
+
+		case <-ctx.Done():
 		}
 	}()
 
@@ -1878,39 +1871,31 @@ func (b *batch) monitorConfirmations(ctx context.Context) error {
 		defer cancel()
 		defer b.wg.Done()
 
-		for {
+		select {
+		case conf := <-confChan:
 			select {
-			case conf := <-confChan:
-				select {
-				case b.confChan <- conf:
-
-				case <-ctx.Done():
-				}
-
-				return
-
-			case err := <-errChan:
-				b.writeToErrChan(fmt.Errorf("confirmations "+
-					"monitoring error: %w", err))
-
-				return
-
-			case <-reorgChan:
-				// A re-org has been detected. We set the batch
-				// state back to open since our batch
-				// transaction is no longer present in any
-				// block. We can accept more sweeps and try to
-				// publish new transactions, at this point we
-				// need to monitor again for a new spend.
-				select {
-				case b.reorgChan <- struct{}{}:
-				case <-ctx.Done():
-				}
-				return
+			case b.confChan <- conf:
 
 			case <-ctx.Done():
-				return
 			}
+
+		case err := <-errChan:
+			b.writeToErrChan(fmt.Errorf("confirmations "+
+				"monitoring error: %w", err))
+
+		case <-reorgChan:
+			// A re-org has been detected. We set the batch
+			// state back to open since our batch
+			// transaction is no longer present in any
+			// block. We can accept more sweeps and try to
+			// publish new transactions, at this point we
+			// need to monitor again for a new spend.
+			select {
+			case b.reorgChan <- struct{}{}:
+			case <-ctx.Done():
+			}
+
+		case <-ctx.Done():
 		}
 	}()
 
diff --git a/sweepbatcher/sweep_batcher.go b/sweepbatcher/sweep_batcher.go
index c8ee03b7..f87abcaf 100644
--- a/sweepbatcher/sweep_batcher.go
+++ b/sweepbatcher/sweep_batcher.go
@@ -1154,74 +1154,70 @@ func (b *Batcher) monitorSpendAndNotify(ctx context.Context, sweep *sweep,
 		infof("Batcher monitoring spend for swap %x",
 			sweep.swapHash[:6])
 
-		for {
+		select {
+		case spend := <-spendChan:
+			spendTx := spend.SpendingTx
+			// Calculate the fee portion that each sweep should pay
+			// for the batch.
+			feePortionPerSweep, roundingDifference :=
+				getFeePortionForSweep(
+					spendTx, len(spendTx.TxIn),
+					totalSwept,
+				)
+
+			onChainFeePortion := getFeePortionPaidBySweep(
+				spendTx, feePortionPerSweep,
+				roundingDifference, sweep,
+			)
+
+			// Notify the requester of the spend with the spend
+			// details, including the fee portion for this
+			// particular sweep.
+			spendDetail := &SpendDetail{
+				Tx:                spendTx,
+				OnChainFeePortion: onChainFeePortion,
+			}
+
 			select {
-			case spend := <-spendChan:
-				spendTx := spend.SpendingTx
-				// Calculate the fee portion that each sweep
-				// should pay for the batch.
-				feePortionPerSweep, roundingDifference :=
-					getFeePortionForSweep(
-						spendTx, len(spendTx.TxIn),
-						totalSwept,
-					)
-
-				onChainFeePortion := getFeePortionPaidBySweep(
-					spendTx, feePortionPerSweep,
-					roundingDifference, sweep,
-				)
-
-				// Notify the requester of the spend
-				// with the spend details, including the fee
-				// portion for this particular sweep.
-				spendDetail := &SpendDetail{
-					Tx:                spendTx,
-					OnChainFeePortion: onChainFeePortion,
-				}
-
-				select {
-				// Try to write the update to the notification
-				// channel.
-				case notifier.SpendChan <- spendDetail:
-
-				// If a quit signal was provided by the swap,
-				// continue.
-				case <-notifier.QuitChan:
-
-				// If the context was canceled, stop.
-				case <-ctx.Done():
-				}
-
-				return
-
-			case err := <-spendErr:
-				select {
-				// Try to write the error to the notification
-				// channel.
-				case notifier.SpendErrChan <- err:
-
-				// If a quit signal was provided by the swap,
-				// continue.
-				case <-notifier.QuitChan:
-
-				// If the context was canceled, stop.
-				case <-ctx.Done():
-				}
-
-				b.writeToErrChan(
-					ctx, fmt.Errorf("spend error: %w", err),
-				)
-
-				return
+			// Try to write the update to the notification channel.
+			case notifier.SpendChan <- spendDetail:
 
 			// If a quit signal was provided by the swap, continue.
 			case <-notifier.QuitChan:
-				return
 
 			// If the context was canceled, stop.
 			case <-ctx.Done():
-				return
 			}
+
+			return
+
+		case err := <-spendErr:
+			select {
+			// Try to write the error to the notification
+			// channel.
+			case notifier.SpendErrChan <- err:
+
+			// If a quit signal was provided by the swap,
+			// continue.
+			case <-notifier.QuitChan:
+
+			// If the context was canceled, stop.
+			case <-ctx.Done():
+			}
+
+			b.writeToErrChan(
+				ctx, fmt.Errorf("spend error: %w", err),
+			)
+
+			return
+
+		// If a quit signal was provided by the swap, continue.
+		case <-notifier.QuitChan:
+			return
+
+		// If the context was canceled, stop.
+		case <-ctx.Done():
+			return
 		}
 	}()
 

From 7edbfeadd8a31ebe070ccca2468b708ea9b78356 Mon Sep 17 00:00:00 2001
From: Boris Nagaev <bnagaev@gmail.com>
Date: Tue, 8 Apr 2025 20:42:14 -0300
Subject: [PATCH 10/13] sweepbatcher: store batch status before monitoring

If monitorConfirmations fails, we still want to persist the state to DB.
---
 sweepbatcher/sweep_batch.go | 15 +++++++++------
 1 file changed, 9 insertions(+), 6 deletions(-)

diff --git a/sweepbatcher/sweep_batch.go b/sweepbatcher/sweep_batch.go
index ee623659..25c58ba3 100644
--- a/sweepbatcher/sweep_batch.go
+++ b/sweepbatcher/sweep_batch.go
@@ -2099,16 +2099,19 @@ func (b *batch) handleSpend(ctx context.Context, spendTx *wire.MsgTx) error {
 		"purged swaps: %v, purged groups: %v", confirmedSweeps,
 		purgedSweeps, purgedSwaps, len(purgeList))
 
-	err = b.monitorConfirmations(ctx)
-	if err != nil {
-		return err
-	}
-
 	// We are no longer able to accept new sweeps, so we mark the batch as
 	// closed and persist on storage.
 	b.state = Closed
 
-	return b.persist(ctx)
+	if err = b.persist(ctx); err != nil {
+		return fmt.Errorf("saving batch failed: %w", err)
+	}
+
+	if err = b.monitorConfirmations(ctx); err != nil {
+		return fmt.Errorf("monitorConfirmations failed: %w", err)
+	}
+
+	return nil
 }
 
 // handleConf handles a confirmation notification. This is the final step of the

From 980c7d4e00ea9ec5bfcd2b4edbf91deb406f2174 Mon Sep 17 00:00:00 2001
From: Boris Nagaev <bnagaev@gmail.com>
Date: Tue, 8 Apr 2025 20:44:30 -0300
Subject: [PATCH 11/13] sweepbatcher: align dbBatch type with DB schema

Previously, dbBatch had a State field (enum: Open, Closed, Confirmed), but in
the database it is represented as a boolean Confirmed. The Closed state was
stored the same way as Open. This wasn't an issue in practice, since an Open
batch is quickly transitioned to Closed after startup.

However, the in-memory mock stores plain dbBatch instances, leading to
inconsistent behavior between the mock and the real DB-backed store. This
commit updates dbBatch to match the database representation by replacing
the State field with a Confirmed boolean.
---
 sweepbatcher/store.go         | 23 ++++++-----------------
 sweepbatcher/store_mock.go    |  6 +++---
 sweepbatcher/sweep_batch.go   | 27 +++++++--------------------
 sweepbatcher/sweep_batcher.go | 35 +++++++++++++----------------------
 4 files changed, 29 insertions(+), 62 deletions(-)

diff --git a/sweepbatcher/store.go b/sweepbatcher/store.go
index bb245e5f..1b87cde8 100644
--- a/sweepbatcher/store.go
+++ b/sweepbatcher/store.go
@@ -213,8 +213,8 @@ type dbBatch struct {
 	// ID is the unique identifier of the batch.
 	ID int32
 
-	// State is the current state of the batch.
-	State string
+	// Confirmed is set when the batch is fully confirmed.
+	Confirmed bool
 
 	// BatchTxid is the txid of the batch transaction.
 	BatchTxid chainhash.Hash
@@ -255,11 +255,8 @@ type dbSweep struct {
 // convertBatchRow converts a batch row from db to a sweepbatcher.Batch struct.
 func convertBatchRow(row sqlc.SweepBatch) *dbBatch {
 	batch := dbBatch{
-		ID: row.ID,
-	}
-
-	if row.Confirmed {
-		batch.State = batchConfirmed
+		ID:        row.ID,
+		Confirmed: row.Confirmed,
 	}
 
 	if row.BatchTxID.Valid {
@@ -288,7 +285,7 @@ func convertBatchRow(row sqlc.SweepBatch) *dbBatch {
 // it into the database.
 func batchToInsertArgs(batch dbBatch) sqlc.InsertBatchParams {
 	args := sqlc.InsertBatchParams{
-		Confirmed: false,
+		Confirmed: batch.Confirmed,
 		BatchTxID: sql.NullString{
 			Valid:  true,
 			String: batch.BatchTxid.String(),
@@ -305,10 +302,6 @@ func batchToInsertArgs(batch dbBatch) sqlc.InsertBatchParams {
 		MaxTimeoutDistance: batch.MaxTimeoutDistance,
 	}
 
-	if batch.State == batchConfirmed {
-		args.Confirmed = true
-	}
-
 	return args
 }
 
@@ -317,7 +310,7 @@ func batchToInsertArgs(batch dbBatch) sqlc.InsertBatchParams {
 func batchToUpdateArgs(batch dbBatch) sqlc.UpdateBatchParams {
 	args := sqlc.UpdateBatchParams{
 		ID:        batch.ID,
-		Confirmed: false,
+		Confirmed: batch.Confirmed,
 		BatchTxID: sql.NullString{
 			Valid:  true,
 			String: batch.BatchTxid.String(),
@@ -333,10 +326,6 @@ func batchToUpdateArgs(batch dbBatch) sqlc.UpdateBatchParams {
 		},
 	}
 
-	if batch.State == batchConfirmed {
-		args.Confirmed = true
-	}
-
 	return args
 }
 
diff --git a/sweepbatcher/store_mock.go b/sweepbatcher/store_mock.go
index 0f6c5e3c..d5a3ffbc 100644
--- a/sweepbatcher/store_mock.go
+++ b/sweepbatcher/store_mock.go
@@ -36,7 +36,7 @@ func (s *StoreMock) FetchUnconfirmedSweepBatches(ctx context.Context) (
 
 	result := []*dbBatch{}
 	for _, batch := range s.batches {
-		if batch.State != "confirmed" {
+		if !batch.Confirmed {
 			result = append(result, &batch)
 		}
 	}
@@ -91,7 +91,7 @@ func (s *StoreMock) ConfirmBatch(ctx context.Context, id int32) error {
 		return errors.New("batch not found")
 	}
 
-	batch.State = "confirmed"
+	batch.Confirmed = true
 	s.batches[batch.ID] = batch
 
 	return nil
@@ -201,7 +201,7 @@ func (s *StoreMock) TotalSweptAmount(ctx context.Context, batchID int32) (
 		return 0, errors.New("batch not found")
 	}
 
-	if batch.State != batchConfirmed && batch.State != batchClosed {
+	if !batch.Confirmed {
 		return 0, nil
 	}
 
diff --git a/sweepbatcher/sweep_batch.go b/sweepbatcher/sweep_batch.go
index 25c58ba3..dc57b0f9 100644
--- a/sweepbatcher/sweep_batch.go
+++ b/sweepbatcher/sweep_batch.go
@@ -135,7 +135,9 @@ const (
 	Open batchState = 0
 
 	// Closed is the state in which the batch is no longer able to accept
-	// new sweeps.
+	// new sweeps. NOTE: this state exists only in-memory. In the database
+	// it is stored as Open and converted to Closed after a spend
+	// notification arrives (quickly after start of Batch.Run).
 	Closed batchState = 1
 
 	// Confirmed is the state in which the batch transaction has reached the
@@ -870,8 +872,8 @@ func (b *batch) Run(ctx context.Context) error {
 	// completes.
 	timerChan := clock.TickAfter(b.cfg.batchPublishDelay)
 
-	b.Infof("started, primary %s, total sweeps %d",
-		b.primarySweepID, len(b.sweeps))
+	b.Infof("started, primary %s, total sweeps %d, state: %d",
+		b.primarySweepID, len(b.sweeps), b.state)
 
 	for {
 		// If the batch is not empty, find earliest initialDelay.
@@ -2179,7 +2181,7 @@ func (b *batch) persist(ctx context.Context) error {
 	bch := &dbBatch{}
 
 	bch.ID = b.id
-	bch.State = stateEnumToString(b.state)
+	bch.Confirmed = b.state == Confirmed
 
 	if b.batchTxid != nil {
 		bch.BatchTxid = *b.batchTxid
@@ -2238,7 +2240,7 @@ func (b *batch) getBatchDestAddr(ctx context.Context) (btcutil.Address, error) {
 
 func (b *batch) insertAndAcquireID(ctx context.Context) (int32, error) {
 	bch := &dbBatch{}
-	bch.State = stateEnumToString(b.state)
+	bch.Confirmed = b.state == Confirmed
 	bch.MaxTimeoutDistance = b.cfg.maxTimeoutDistance
 
 	id, err := b.store.InsertSweepBatch(ctx, bch)
@@ -2339,18 +2341,3 @@ func clampBatchFee(fee btcutil.Amount,
 
 	return fee
 }
-
-func stateEnumToString(state batchState) string {
-	switch state {
-	case Open:
-		return batchOpen
-
-	case Closed:
-		return batchClosed
-
-	case Confirmed:
-		return batchConfirmed
-	}
-
-	return ""
-}
diff --git a/sweepbatcher/sweep_batcher.go b/sweepbatcher/sweep_batcher.go
index f87abcaf..9e3a7ab6 100644
--- a/sweepbatcher/sweep_batcher.go
+++ b/sweepbatcher/sweep_batcher.go
@@ -31,18 +31,6 @@ const (
 	// of sweeps that can appear in the same batch.
 	defaultMaxTimeoutDistance = 288
 
-	// batchOpen is the string representation of the state of a batch that
-	// is open.
-	batchOpen = "open"
-
-	// batchClosed is the string representation of the state of a batch
-	// that is closed.
-	batchClosed = "closed"
-
-	// batchConfirmed is the string representation of the state of a batch
-	// that is confirmed.
-	batchConfirmed = "confirmed"
-
 	// defaultMainnetPublishDelay is the default publish delay that is used
 	// for mainnet.
 	defaultMainnetPublishDelay = 5 * time.Second
@@ -760,7 +748,7 @@ func (b *Batcher) AddSweep(ctx context.Context, sweepReq *SweepRequest) error {
 				"sweep %x: %w", sweep.swapHash[:6], err)
 		}
 
-		if parentBatch.State == batchConfirmed {
+		if parentBatch.Confirmed {
 			fullyConfirmed = true
 		}
 	}
@@ -844,7 +832,7 @@ func (b *Batcher) handleSweeps(ctx context.Context, sweeps []*sweep,
 	if completed && *notifier != (SpendNotifier{}) {
 		// The parent batch is indeed confirmed, meaning it is complete
 		// and we won't be able to attach this sweep to it.
-		if parentBatch.State == batchConfirmed {
+		if parentBatch.Confirmed {
 			return b.monitorSpendAndNotify(
 				ctx, sweep, parentBatch.ID, notifier,
 			)
@@ -1093,15 +1081,18 @@ func (b *Batcher) FetchUnconfirmedBatches(ctx context.Context) ([]*batch,
 		batch := batch{}
 		batch.id = bch.ID
 
-		switch bch.State {
-		case batchOpen:
-			batch.state = Open
-
-		case batchClosed:
-			batch.state = Closed
-
-		case batchConfirmed:
+		if bch.Confirmed {
 			batch.state = Confirmed
+		} else {
+			// We don't store Closed state separately in DB.
+			// If the batch is closed (included into a block, but
+			// not fully confirmed), it is now considered Open
+			// again. It will receive a spending notification as
+			// soon as it starts, so it is not an issue. If a sweep
+			// manages to be added during this time, it will be
+			// detected as missing when analyzing the spend
+			// notification and will be added to new batch.
+			batch.state = Open
 		}
 
 		batch.batchTxid = &bch.BatchTxid

From 8466e0f8eb76d98ef0872fe9a3a75d0cce99d618 Mon Sep 17 00:00:00 2001
From: Boris Nagaev <bnagaev@gmail.com>
Date: Tue, 8 Apr 2025 18:46:35 -0300
Subject: [PATCH 12/13] sweepbatcher: test spending notification and error

---
 sweepbatcher/sweep_batcher_test.go | 244 ++++++++++++++++++++++++++---
 1 file changed, 220 insertions(+), 24 deletions(-)

diff --git a/sweepbatcher/sweep_batcher_test.go b/sweepbatcher/sweep_batcher_test.go
index cd5bab01..13188ff5 100644
--- a/sweepbatcher/sweep_batcher_test.go
+++ b/sweepbatcher/sweep_batcher_test.go
@@ -762,9 +762,9 @@ func testSweepBatcherSimpleLifecycle(t *testing.T, store testStore,
 	batcher := NewBatcher(lnd.WalletKit, lnd.ChainNotifier, lnd.Signer,
 		testMuSig2SignSweep, testVerifySchnorrSig, lnd.ChainParams,
 		batcherStore, sweepStore)
+	runErrChan := make(chan error)
 	go func() {
-		err := batcher.Run(ctx)
-		checkBatcherError(t, err)
+		runErrChan <- batcher.Run(ctx)
 	}()
 
 	// Create a sweep request.
@@ -772,13 +772,24 @@ func testSweepBatcherSimpleLifecycle(t *testing.T, store testStore,
 		Hash:  chainhash.Hash{1, 1},
 		Index: 1,
 	}
+	const (
+		inputValue  = 111
+		outputValue = 50
+		fee         = inputValue - outputValue
+	)
+	spendErrChan := make(chan error, 1)
+	notifier := &SpendNotifier{
+		SpendChan:    make(chan *SpendDetail, 1),
+		SpendErrChan: spendErrChan,
+		QuitChan:     make(chan bool, 1),
+	}
 	sweepReq1 := SweepRequest{
 		SwapHash: lntypes.Hash{1, 1, 1},
 		Inputs: []Input{{
-			Value:    111,
+			Value:    inputValue,
 			Outpoint: op1,
 		}},
-		Notifier: &dummyNotifier,
+		Notifier: notifier,
 	}
 
 	const initiationHeight = 550
@@ -786,7 +797,7 @@ func testSweepBatcherSimpleLifecycle(t *testing.T, store testStore,
 	swap1 := &loopdb.LoopOutContract{
 		SwapContract: loopdb.SwapContract{
 			CltvExpiry:       111,
-			AmountRequested:  111,
+			AmountRequested:  inputValue,
 			ProtocolVersion:  loopdb.ProtocolVersionMuSig2,
 			HtlcKeys:         htlcKeys,
 			InitiationHeight: initiationHeight,
@@ -806,33 +817,27 @@ func testSweepBatcherSimpleLifecycle(t *testing.T, store testStore,
 
 	// When batch is successfully created it will execute it's first step,
 	// which leads to a spend monitor of the primary sweep.
-	<-lnd.RegisterSpendChannel
+	spendReg := <-lnd.RegisterSpendChannel
+
+	// Wait for tx to be published.
+	<-lnd.TxPublishChannel
 
 	// Eventually request will be consumed and a new batch will spin up.
+	var primarySweepID wire.OutPoint
 	require.Eventually(t, func() bool {
-		return batcher.numBatches(ctx) == 1
-	}, test.Timeout, eventuallyCheckFrequency)
+		batch := tryGetOnlyBatch(ctx, batcher)
+		if batch == nil {
+			return false
+		}
 
-	// Find the batch and assign it to a local variable for easier access.
-	batch := &batch{}
-	for _, btch := range getBatches(ctx, batcher) {
-		btch.testRunInEventLoop(ctx, func() {
-			if btch.primarySweepID == op1 {
-				batch = btch
-			}
-		})
-	}
+		primarySweepID = batch.snapshot(ctx).primarySweepID
 
-	require.Eventually(t, func() bool {
 		// Batch should have the sweep stored.
 		return batch.numSweeps(ctx) == 1
 	}, test.Timeout, eventuallyCheckFrequency)
 
 	// The primary sweep id should be that of the first inserted sweep.
-	require.Equal(t, batch.primarySweepID, op1)
-
-	// Wait for tx to be published.
-	<-lnd.TxPublishChannel
+	require.Equal(t, primarySweepID, op1)
 
 	err = lnd.NotifyHeight(601)
 	require.NoError(t, err)
@@ -840,7 +845,11 @@ func testSweepBatcherSimpleLifecycle(t *testing.T, store testStore,
 	// After receiving a height notification the batch will step again,
 	// leading to a new spend monitoring.
 	require.Eventually(t, func() bool {
-		batch := batch.snapshot(ctx)
+		batch := tryGetOnlyBatch(ctx, batcher)
+		if batch == nil {
+			return false
+		}
+		batch = batch.snapshot(ctx)
 
 		return batch.currentHeight == 601
 	}, test.Timeout, eventuallyCheckFrequency)
@@ -848,6 +857,58 @@ func testSweepBatcherSimpleLifecycle(t *testing.T, store testStore,
 	// Wait for tx to be published.
 	<-lnd.TxPublishChannel
 
+	// Emulate spend error.
+	testError := errors.New("test error")
+	spendReg.ErrChan <- testError
+
+	// Make sure the caller of AddSweep got the spending error.
+	notifierErr := <-spendErrChan
+	require.Error(t, notifierErr)
+	require.ErrorIs(t, notifierErr, testError)
+
+	// Wait for the batcher to crash because of the spending error.
+	runErr := <-runErrChan
+	require.ErrorIs(t, runErr, testError)
+
+	// Now launch the batcher again.
+	batcher = NewBatcher(lnd.WalletKit, lnd.ChainNotifier, lnd.Signer,
+		testMuSig2SignSweep, testVerifySchnorrSig, lnd.ChainParams,
+		batcherStore, sweepStore)
+	go func() {
+		runErrChan <- batcher.Run(ctx)
+	}()
+
+	// When batch is successfully created it will execute it's first step,
+	// which leads to a spend monitor of the primary sweep.
+	<-lnd.RegisterSpendChannel
+
+	// Wait for tx to be published.
+	<-lnd.TxPublishChannel
+
+	// Deliver sweep request to batcher.
+	spendChan := make(chan *SpendDetail, 1)
+	notifier = &SpendNotifier{
+		SpendChan:    spendChan,
+		SpendErrChan: make(chan error, 1),
+		QuitChan:     make(chan bool, 1),
+	}
+	sweepReq1.Notifier = notifier
+	require.NoError(t, batcher.AddSweep(ctx, &sweepReq1))
+
+	// Wait for the notifier to be installed.
+	require.Eventually(t, func() bool {
+		batch := tryGetOnlyBatch(ctx, batcher)
+		if batch == nil {
+			return false
+		}
+		batch = batch.snapshot(ctx)
+
+		sweep := batch.sweeps[batch.primarySweepID]
+
+		return sweep.notifier != nil &&
+			sweep.notifier.SpendChan == spendChan
+	}, test.Timeout, eventuallyCheckFrequency)
+
 	// Create the spending tx that will trigger the spend monitor of the
 	// batch.
 	spendingTx := &wire.MsgTx{
@@ -861,6 +922,7 @@ func testSweepBatcherSimpleLifecycle(t *testing.T, store testStore,
 		},
 		TxOut: []*wire.TxOut{
 			{
+				Value:    outputValue,
 				PkScript: []byte{3, 2, 1},
 			},
 		},
@@ -879,6 +941,11 @@ func testSweepBatcherSimpleLifecycle(t *testing.T, store testStore,
 	// We notify the spend.
 	lnd.SpendChannel <- spendDetail
 
+	// Make sure the notifier got a proper spending notification.
+	spending := <-spendChan
+	require.Equal(t, spendingTxHash, spending.Tx.TxHash())
+	require.Equal(t, btcutil.Amount(fee), spending.OnChainFeePortion)
+
 	// After receiving the spend, the batch is now monitoring for confs.
 	confReg := <-lnd.RegisterConfChannel
 
@@ -889,7 +956,84 @@ func testSweepBatcherSimpleLifecycle(t *testing.T, store testStore,
 	// The batch should eventually read the spend notification and progress
 	// its state to closed.
 	require.Eventually(t, func() bool {
-		batch := batch.snapshot(ctx)
+		batch := tryGetOnlyBatch(ctx, batcher)
+		if batch == nil {
+			return false
+		}
+		batch = batch.snapshot(ctx)
+
+		return batch.state == Closed
+	}, test.Timeout, eventuallyCheckFrequency)
+
+	// Emulate a confirmation error.
+	confReg.ErrChan <- testError
+
+	// Wait for the batcher to crash because of the confirmation error.
+	runErr = <-runErrChan
+	require.ErrorIs(t, runErr, testError)
+
+	// Now launch the batcher again.
+	batcher = NewBatcher(lnd.WalletKit, lnd.ChainNotifier, lnd.Signer,
+		testMuSig2SignSweep, testVerifySchnorrSig, lnd.ChainParams,
+		batcherStore, sweepStore)
+	go func() {
+		runErrChan <- batcher.Run(ctx)
+	}()
+
+	// When batch is successfully created it will execute it's first step,
+	// which leads to a spend monitor of the primary sweep.
+	<-lnd.RegisterSpendChannel
+
+	// Deliver sweep request to batcher.
+	spendChan = make(chan *SpendDetail, 1)
+	notifier = &SpendNotifier{
+		SpendChan:    spendChan,
+		SpendErrChan: make(chan error, 1),
+		QuitChan:     make(chan bool, 1),
+	}
+	sweepReq1.Notifier = notifier
+	require.NoError(t, batcher.AddSweep(ctx, &sweepReq1))
+
+	// Wait for tx to be published. A closed batch is stored in DB as Open.
+	<-lnd.TxPublishChannel
+
+	// Wait for the notifier to be installed.
+	require.Eventually(t, func() bool {
+		batch := tryGetOnlyBatch(ctx, batcher)
+		if batch == nil {
+			return false
+		}
+		batch = batch.snapshot(ctx)
+
+		sweep := batch.sweeps[batch.primarySweepID]
+
+		return sweep.notifier != nil &&
+			sweep.notifier.SpendChan == spendChan
+	}, test.Timeout, eventuallyCheckFrequency)
+
+	// We notify the spend.
+	lnd.SpendChannel <- spendDetail
+
+	// Make sure the notifier got a proper spending notification.
+	spending = <-spendChan
+	require.Equal(t, spendingTxHash, spending.Tx.TxHash())
+	require.Equal(t, btcutil.Amount(fee), spending.OnChainFeePortion)
+
+	// After receiving the spend, the batch is now monitoring for confs.
+	confReg = <-lnd.RegisterConfChannel
+
+	// Make sure the confirmation has proper height hint. It should pass
+	// the swap initiation height, not the current height.
+	require.Equal(t, int32(initiationHeight), confReg.HeightHint)
+
+	// The batch should eventually read the spend notification and progress
+	// its state to closed.
+	require.Eventually(t, func() bool {
+		batch := tryGetOnlyBatch(ctx, batcher)
+		if batch == nil {
+			return false
+		}
+		batch = batch.snapshot(ctx)
 
 		return batch.state == Closed
 	}, test.Timeout, eventuallyCheckFrequency)
@@ -905,8 +1049,60 @@ func testSweepBatcherSimpleLifecycle(t *testing.T, store testStore,
 	// Eventually the batch receives the confirmation notification and
 	// confirms itself.
 	require.Eventually(t, func() bool {
+		batch := tryGetOnlyBatch(ctx, batcher)
+		if batch == nil {
+			return false
+		}
+
 		return batch.isComplete()
 	}, test.Timeout, eventuallyCheckFrequency)
+
+	// Now emulate adding the sweep again after it was fully confirmed.
+	// This triggers another code path (monitorSpendAndNotify).
+	spendChan = make(chan *SpendDetail, 1)
+	notifier = &SpendNotifier{
+		SpendChan:    spendChan,
+		SpendErrChan: make(chan error, 1),
+		QuitChan:     make(chan bool, 1),
+	}
+	sweepReq1.Notifier = notifier
+	require.NoError(t, batcher.AddSweep(ctx, &sweepReq1))
+
+	// Expect a spending registration.
+	<-lnd.RegisterSpendChannel
+
+	// We notify the spend.
+	lnd.SpendChannel <- spendDetail
+
+	// Now expect the notifier to produce the spending details.
+	spending = <-spendChan
+	require.Equal(t, spendingTxHash, spending.Tx.TxHash())
+	require.Equal(t, btcutil.Amount(fee), spending.OnChainFeePortion)
+
+	// Now check what happens in case of a spending error.
+	spendErrChan = make(chan error, 1)
+	notifier = &SpendNotifier{
+		SpendChan:    make(chan *SpendDetail, 1),
+		SpendErrChan: spendErrChan,
+		QuitChan:     make(chan bool, 1),
+	}
+	sweepReq1.Notifier = notifier
+	require.NoError(t, batcher.AddSweep(ctx, &sweepReq1))
+
+	// Expect a spending registration.
+	spendReg = <-lnd.RegisterSpendChannel
+
+	// Emulate spend error.
+	spendReg.ErrChan <- testError
+
+	// Make sure the caller of AddSweep got the spending error.
+	notifierErr = <-spendErrChan
+	require.Error(t, notifierErr)
+	require.ErrorIs(t, notifierErr, testError)
+
+	// Wait for the batcher to crash because of the spending error.
+	runErr = <-runErrChan
+	require.ErrorIs(t, runErr, testError)
 }
 
 // wrappedLogger implements btclog.Logger, recording last debug message format.

From c70257a4a537d9b43469cb1312ed719e8da5783b Mon Sep 17 00:00:00 2001
From: Boris Nagaev <bnagaev@gmail.com>
Date: Tue, 8 Apr 2025 23:15:23 -0300
Subject: [PATCH 13/13] sweepbatcher: notify caller about confirmations

Add fields ConfChan and ConfErrChan to SpendNotifier type which is a part of
SweepRequest passed to AddSweep method.

This is needed to reuse confirmation notifications on the calling side the same
way it is done for spending notifications.
---
 sweepbatcher/sweep_batch.go        |  90 +++++++++++++++++++++-
 sweepbatcher/sweep_batcher.go      | 116 ++++++++++++++++++++++++++++-
 sweepbatcher/sweep_batcher_test.go |  69 ++++++++++++++++-
 3 files changed, 272 insertions(+), 3 deletions(-)

diff --git a/sweepbatcher/sweep_batch.go b/sweepbatcher/sweep_batch.go
index dc57b0f9..3084a26a 100644
--- a/sweepbatcher/sweep_batch.go
+++ b/sweepbatcher/sweep_batch.go
@@ -1882,6 +1882,8 @@ func (b *batch) monitorConfirmations(ctx context.Context) error {
 			}
 
 		case err := <-errChan:
+			b.writeToConfErrChan(ctx, err)
+
 			b.writeToErrChan(fmt.Errorf("confirmations "+
 				"monitoring error: %w", err))
 
@@ -2158,7 +2160,55 @@ func (b *batch) handleConf(ctx context.Context,
 	b.Infof("confirmed in txid %s", b.batchTxid)
 	b.state = Confirmed
 
-	return b.store.ConfirmBatch(ctx, b.id)
+	if err := b.store.ConfirmBatch(ctx, b.id); err != nil {
+		return fmt.Errorf("failed to store confirmed state: %w", err)
+	}
+
+	// Calculate the fee portion that each sweep should pay for the batch.
+	// TODO: make sure spendTx matches b.sweeps.
+	var totalSweptAmt btcutil.Amount
+	for _, s := range b.sweeps {
+		totalSweptAmt += s.value
+	}
+	feePortionPaidPerSweep, roundingDifference := getFeePortionForSweep(
+		spendTx, len(b.sweeps), totalSweptAmt,
+	)
+
+	// Send the confirmation to all the notifiers.
+	for _, s := range b.sweeps {
+		// If the sweep's notifier is empty then this means that
+		// a swap is not waiting to read an update from it, so
+		// we can skip the notification part.
+		if s.notifier == nil || s.notifier.ConfChan == nil {
+			continue
+		}
+
+		confDetail := &ConfDetail{
+			TxConfirmation: conf,
+			OnChainFeePortion: getFeePortionPaidBySweep(
+				spendTx, feePortionPaidPerSweep,
+				roundingDifference, &s,
+			),
+		}
+
+		// Notify the caller in a goroutine to avoid possible dead-lock.
+		go func(notifier *SpendNotifier) {
+			// Note that we don't unblock on ctx, because it will
+			// expire soon, when batch.Run completes. The caller is
+			// responsible to consume ConfChan or close QuitChan.
+			select {
+			// Try to write the confirmation to the notification
+			// channel.
+			case notifier.ConfChan <- confDetail:
+
+			// If a quit signal was provided by the swap,
+			// continue.
+			case <-notifier.QuitChan:
+			}
+		}(s.notifier)
+	}
+
+	return nil
 }
 
 // isComplete returns true if the batch is completed. This method is used by the
@@ -2314,6 +2364,44 @@ func (b *batch) writeToSpendErrChan(ctx context.Context, spendErr error) {
 	}
 }
 
+// writeToConfErrChan sends an error to confirmation error channels of all the
+// sweeps.
+func (b *batch) writeToConfErrChan(ctx context.Context, confErr error) {
+	done, err := b.scheduleNextCall()
+	if err != nil {
+		done()
+
+		return
+	}
+	notifiers := make([]*SpendNotifier, 0, len(b.sweeps))
+	for _, s := range b.sweeps {
+		// If the sweep's notifier is empty then this means that a swap
+		// is not waiting to read an update from it, so we can skip
+		// the notification part.
+		if s.notifier == nil || s.notifier.ConfErrChan == nil {
+			continue
+		}
+
+		notifiers = append(notifiers, s.notifier)
+	}
+	done()
+
+	for _, notifier := range notifiers {
+		select {
+		// Try to write the error to the notification
+		// channel.
+		case notifier.ConfErrChan <- confErr:
+
+		// If a quit signal was provided by the swap,
+		// continue.
+		case <-notifier.QuitChan:
+
+		// If the context was canceled, stop.
+		case <-ctx.Done():
+		}
+	}
+}
+
 func (b *batch) persistSweep(ctx context.Context, sweep sweep,
 	completed bool) error {
 
diff --git a/sweepbatcher/sweep_batcher.go b/sweepbatcher/sweep_batcher.go
index 9e3a7ab6..b89ba5f6 100644
--- a/sweepbatcher/sweep_batcher.go
+++ b/sweepbatcher/sweep_batcher.go
@@ -20,6 +20,7 @@ import (
 	"github.com/lightninglabs/loop/loopdb"
 	"github.com/lightninglabs/loop/swap"
 	"github.com/lightninglabs/loop/utils"
+	"github.com/lightningnetwork/lnd/chainntnfs"
 	"github.com/lightningnetwork/lnd/clock"
 	"github.com/lightningnetwork/lnd/input"
 	"github.com/lightningnetwork/lnd/lntypes"
@@ -280,6 +281,8 @@ type addSweepsRequest struct {
 	parentBatch *dbBatch
 }
 
+// SpendDetail is a notification that is send to the user of sweepbatcher when
+// a batch gets the first confirmation.
 type SpendDetail struct {
 	// Tx is the transaction that spent the outpoint.
 	Tx *wire.MsgTx
@@ -291,6 +294,19 @@ type SpendDetail struct {
 	OnChainFeePortion btcutil.Amount
 }
 
+// ConfDetail is a notification that is send to the user of sweepbatcher when
+// a batch is fully confirmed, i.e. gets batchConfHeight confirmations.
+type ConfDetail struct {
+	// TxConfirmation has data about the confirmation of the transaction.
+	*chainntnfs.TxConfirmation
+
+	// OnChainFeePortion is the fee portion that was paid to get this sweep
+	// confirmed on chain. This is the difference between the value of the
+	// outpoint and the value of all sweeps that were included in the batch
+	// divided by the number of sweeps.
+	OnChainFeePortion btcutil.Amount
+}
+
 // SpendNotifier is a notifier that is used to notify the requester of a sweep
 // that the sweep was successful.
 type SpendNotifier struct {
@@ -300,6 +316,14 @@ type SpendNotifier struct {
 	// SpendErrChan is a channel where spend errors are received.
 	SpendErrChan chan<- error
 
+	// ConfChan is a channel where the confirmation details are received.
+	// This channel is optional.
+	ConfChan chan<- *ConfDetail
+
+	// ConfErrChan is a channel where confirmation errors are received.
+	// This channel is optional.
+	ConfErrChan chan<- error
+
 	// QuitChan is a channel that can be closed to stop the notifier.
 	QuitChan <-chan bool
 }
@@ -1114,7 +1138,9 @@ func (b *Batcher) FetchUnconfirmedBatches(ctx context.Context) ([]*batch,
 }
 
 // monitorSpendAndNotify monitors the spend of a specific outpoint and writes
-// the response back to the response channel.
+// the response back to the response channel. It is called if the batch is fully
+// confirmed and we just need to deliver the data back to the caller though
+// SpendNotifier.
 func (b *Batcher) monitorSpendAndNotify(ctx context.Context, sweep *sweep,
 	parentBatchID int32, notifier *SpendNotifier) error {
 
@@ -1172,6 +1198,16 @@ func (b *Batcher) monitorSpendAndNotify(ctx context.Context, sweep *sweep,
 			select {
 			// Try to write the update to the notification channel.
 			case notifier.SpendChan <- spendDetail:
+				err := b.monitorConfAndNotify(
+					ctx, sweep, notifier, spendTx,
+					onChainFeePortion,
+				)
+				if err != nil {
+					b.writeToErrChan(
+						ctx, fmt.Errorf("monitor conf "+
+							"failed: %w", err),
+					)
+				}
 
 			// If a quit signal was provided by the swap, continue.
 			case <-notifier.QuitChan:
@@ -1215,6 +1251,84 @@ func (b *Batcher) monitorSpendAndNotify(ctx context.Context, sweep *sweep,
 	return nil
 }
 
+// monitorConfAndNotify monitors the confirmation of a specific transaction and
+// writes the response back to the response channel. It is called if the batch
+// is fully confirmed and we just need to deliver the data back to the caller
+// though SpendNotifier.
+func (b *Batcher) monitorConfAndNotify(ctx context.Context, sweep *sweep,
+	notifier *SpendNotifier, spendTx *wire.MsgTx,
+	onChainFeePortion btcutil.Amount) error {
+
+	// If confirmation notifications were not requested, stop.
+	if notifier.ConfChan == nil && notifier.ConfErrChan == nil {
+		return nil
+	}
+
+	batchTxid := spendTx.TxHash()
+
+	if len(spendTx.TxOut) != 1 {
+		return fmt.Errorf("unexpected number of outputs in batch: %d, "+
+			"want %d", len(spendTx.TxOut), 1)
+	}
+	batchPkScript := spendTx.TxOut[0].PkScript
+
+	reorgChan := make(chan struct{})
+
+	confCtx, cancel := context.WithCancel(ctx)
+
+	confChan, errChan, err := b.chainNotifier.RegisterConfirmationsNtfn(
+		confCtx, &batchTxid, batchPkScript, batchConfHeight,
+		sweep.initiationHeight, lndclient.WithReOrgChan(reorgChan),
+	)
+	if err != nil {
+		cancel()
+		return err
+	}
+
+	b.wg.Add(1)
+	go func() {
+		defer cancel()
+		defer b.wg.Done()
+
+		select {
+		case conf := <-confChan:
+			if notifier.ConfChan != nil {
+				confDetail := &ConfDetail{
+					TxConfirmation:    conf,
+					OnChainFeePortion: onChainFeePortion,
+				}
+
+				select {
+				case notifier.ConfChan <- confDetail:
+				case <-notifier.QuitChan:
+				case <-ctx.Done():
+				}
+			}
+
+		case err := <-errChan:
+			if notifier.ConfErrChan != nil {
+				select {
+				case notifier.ConfErrChan <- err:
+				case <-notifier.QuitChan:
+				case <-ctx.Done():
+				}
+			}
+
+			b.writeToErrChan(ctx, fmt.Errorf("confirmations "+
+				"monitoring error: %w", err))
+
+		case <-reorgChan:
+			// A re-org has been detected, but the batch is fully
+			// confirmed and this is unexpected. Crash the batcher.
+			b.writeToErrChan(ctx, fmt.Errorf("unexpected reorg"))
+
+		case <-ctx.Done():
+		}
+	}()
+
+	return nil
+}
+
 func (b *Batcher) writeToErrChan(ctx context.Context, err error) {
 	select {
 	case b.errChan <- err:
diff --git a/sweepbatcher/sweep_batcher_test.go b/sweepbatcher/sweep_batcher_test.go
index 13188ff5..fa871829 100644
--- a/sweepbatcher/sweep_batcher_test.go
+++ b/sweepbatcher/sweep_batcher_test.go
@@ -887,9 +887,11 @@ func testSweepBatcherSimpleLifecycle(t *testing.T, store testStore,
 
 	// Deliver sweep request to batcher.
 	spendChan := make(chan *SpendDetail, 1)
+	confErrChan := make(chan error)
 	notifier = &SpendNotifier{
 		SpendChan:    spendChan,
 		SpendErrChan: make(chan error, 1),
+		ConfErrChan:  confErrChan,
 		QuitChan:     make(chan bool, 1),
 	}
 	sweepReq1.Notifier = notifier
@@ -968,6 +970,10 @@ func testSweepBatcherSimpleLifecycle(t *testing.T, store testStore,
 	// Emulate a confirmation error.
 	confReg.ErrChan <- testError
 
+	// Make sure the notifier gets the confirmation error.
+	confErr := <-confErrChan
+	require.ErrorIs(t, confErr, testError)
+
 	// Wait for the batcher to crash because of the confirmation error.
 	runErr = <-runErrChan
 	require.ErrorIs(t, runErr, testError)
@@ -986,9 +992,11 @@ func testSweepBatcherSimpleLifecycle(t *testing.T, store testStore,
 
 	// Deliver sweep request to batcher.
 	spendChan = make(chan *SpendDetail, 1)
+	confChan := make(chan *ConfDetail)
 	notifier = &SpendNotifier{
 		SpendChan:    spendChan,
 		SpendErrChan: make(chan error, 1),
+		ConfChan:     confChan,
 		QuitChan:     make(chan bool, 1),
 	}
 	sweepReq1.Notifier = notifier
@@ -1043,9 +1051,16 @@ func testSweepBatcherSimpleLifecycle(t *testing.T, store testStore,
 
 	// We mock the tx confirmation notification.
 	lnd.ConfChannel <- &chainntnfs.TxConfirmation{
-		Tx: spendingTx,
+		BlockHeight: 604,
+		Tx:          spendingTx,
 	}
 
+	// Make sure the notifier gets a confirmation notification.
+	conf := <-confChan
+	require.Equal(t, uint32(604), conf.BlockHeight)
+	require.Equal(t, spendingTx.TxHash(), conf.Tx.TxHash())
+	require.Equal(t, btcutil.Amount(fee), conf.OnChainFeePortion)
+
 	// Eventually the batch receives the confirmation notification and
 	// confirms itself.
 	require.Eventually(t, func() bool {
@@ -1060,9 +1075,11 @@ func testSweepBatcherSimpleLifecycle(t *testing.T, store testStore,
 	// Now emulate adding the sweep again after it was fully confirmed.
 	// This triggers another code path (monitorSpendAndNotify).
 	spendChan = make(chan *SpendDetail, 1)
+	confChan = make(chan *ConfDetail)
 	notifier = &SpendNotifier{
 		SpendChan:    spendChan,
 		SpendErrChan: make(chan error, 1),
+		ConfChan:     confChan,
 		QuitChan:     make(chan bool, 1),
 	}
 	sweepReq1.Notifier = notifier
@@ -1079,6 +1096,19 @@ func testSweepBatcherSimpleLifecycle(t *testing.T, store testStore,
 	require.Equal(t, spendingTxHash, spending.Tx.TxHash())
 	require.Equal(t, btcutil.Amount(fee), spending.OnChainFeePortion)
 
+	// We mock the tx confirmation notification.
+	<-lnd.RegisterConfChannel
+	lnd.ConfChannel <- &chainntnfs.TxConfirmation{
+		BlockHeight: 604,
+		Tx:          spendingTx,
+	}
+
+	// Make sure the notifier gets a confirmation notification.
+	conf = <-confChan
+	require.Equal(t, uint32(604), conf.BlockHeight)
+	require.Equal(t, spendingTx.TxHash(), conf.Tx.TxHash())
+	require.Equal(t, btcutil.Amount(fee), conf.OnChainFeePortion)
+
 	// Now check what happens in case of a spending error.
 	spendErrChan = make(chan error, 1)
 	notifier = &SpendNotifier{
@@ -1103,6 +1133,43 @@ func testSweepBatcherSimpleLifecycle(t *testing.T, store testStore,
 	// Wait for the batcher to crash because of the spending error.
 	runErr = <-runErrChan
 	require.ErrorIs(t, runErr, testError)
+
+	// Now launch the batcher again.
+	batcher = NewBatcher(lnd.WalletKit, lnd.ChainNotifier, lnd.Signer,
+		testMuSig2SignSweep, testVerifySchnorrSig, lnd.ChainParams,
+		batcherStore, sweepStore)
+	go func() {
+		runErrChan <- batcher.Run(ctx)
+	}()
+
+	// Now check what happens in case of a confirmation error.
+	confErrChan = make(chan error, 1)
+	notifier = &SpendNotifier{
+		SpendChan:    make(chan *SpendDetail, 1),
+		SpendErrChan: make(chan error, 1),
+		ConfErrChan:  confErrChan,
+		QuitChan:     make(chan bool, 1),
+	}
+	sweepReq1.Notifier = notifier
+	require.NoError(t, batcher.AddSweep(ctx, &sweepReq1))
+
+	// Expect a spending registration.
+	<-lnd.RegisterSpendChannel
+
+	// We notify the spend.
+	lnd.SpendChannel <- spendDetail
+
+	// We mock the tx confirmation error notification.
+	confReg = <-lnd.RegisterConfChannel
+	confReg.ErrChan <- testError
+
+	// Make sure the notifier gets the confirmation error.
+	confErr = <-confErrChan
+	require.ErrorIs(t, confErr, testError)
+
+	// Wait for the batcher to crash because of the confirmation error.
+	runErr = <-runErrChan
+	require.ErrorIs(t, runErr, testError)
 }
 
 // wrappedLogger implements btclog.Logger, recording last debug message format.