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