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.

Author: starius

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
 }
 

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
 	}
 

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

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