sweepbatcher: unify storing batch status with mock

SQL database doesn't distinguish between Open and Closed states, interpreting
both as Open. This commit changes the mock to behave the same.

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

@@ -35,7 +35,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)
 		}
 	}
@@ -90,7 +90,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
@@ -189,7 +189,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

@@ -778,8 +778,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.
@@ -1935,7 +1935,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
@@ -1979,7 +1979,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)
@@ -2080,18 +2080,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
@@ -669,7 +657,7 @@ func (b *Batcher) handleSweeps(ctx context.Context, sweeps []*sweep,
 
 		// 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,
 			)
@@ -916,15 +904,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