From 980c7d4e00ea9ec5bfcd2b4edbf91deb406f2174 Mon Sep 17 00:00:00 2001 From: Boris Nagaev Date: Tue, 8 Apr 2025 20:44:30 -0300 Subject: [PATCH] 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