From 2307ec786456c0644abfc148dea5faa6b2bfc725 Mon Sep 17 00:00:00 2001 From: raagan-u Date: Mon, 23 Mar 2026 11:36:49 +0530 Subject: [PATCH 01/13] update timeout and exponential retry --- btc/batcher.go | 2 +- btc/client_test.go | 56 +++++++++++++++++++++++++++------------------ btc/indexer.go | 19 +++++++++++---- btc/indexer_test.go | 43 ++++++++++++++++++++++++++++++++++ 4 files changed, 93 insertions(+), 27 deletions(-) diff --git a/btc/batcher.go b/btc/batcher.go index 37b707a..83adf2a 100644 --- a/btc/batcher.go +++ b/btc/batcher.go @@ -20,7 +20,7 @@ import ( var ( GuardianSpendWeight = 1 + 1 + 1 + 73 + 1 + 73 + 1 + 1 + 1 + 200 // TODO: update Guardian Script Size to the right number SegwitSpendWeight = txsizes.RedeemP2WPKHInputWitnessWeight - DefaultAPITimeout = 5 * time.Second + DefaultAPITimeout = 60 * time.Second ) var ( diff --git a/btc/client_test.go b/btc/client_test.go index 7f65fc4..ddae055 100644 --- a/btc/client_test.go +++ b/btc/client_test.go @@ -273,32 +273,44 @@ var _ = Describe("bitcoin client", func() { }) Context("Test BitcoinRPCClient", func() { - bitcoinClient := btc.NewBitcoinClient("admin1", "123", "http://0.0.0.0:18443") - chainParams := chaincfg.RegressionNetParams - indexer := localnet.BTCIndexer() - feeEstimator := btc.NewFixFeeEstimator(16) + var ( + bitcoinClient btc.BitcoinClient + chainParams chaincfg.Params + indexer btc.IndexerClient + feeEstimator btc.FeeEstimator + alice, bob btc.Wallet + send func(wallet btc.Wallet, to btcutil.Address, amount int64) (string, btc.Transaction) + ) + + BeforeEach(func() { + bitcoinClient = btc.NewBitcoinClient("admin1", "123", "http://0.0.0.0:18443") + chainParams = chaincfg.RegressionNetParams + indexer = localnet.BTCIndexer() + feeEstimator = btc.NewFixFeeEstimator(16) + + newWallet := func() btc.Wallet { + privKey, err := btcec.NewPrivateKey() + Expect(err).To(BeNil()) + wallet, err := btc.NewSimpleWallet(privKey, &chainParams, indexer, feeEstimator, btc.HighFee) + Expect(err).To(BeNil()) + return wallet + } - newWallet := func() btc.Wallet { - privKey, err := btcec.NewPrivateKey() - Expect(err).To(BeNil()) - wallet, err := btc.NewSimpleWallet(privKey, &chainParams, indexer, feeEstimator, btc.HighFee) - Expect(err).To(BeNil()) - return wallet - } + alice = newWallet() + bob = newWallet() - alice := newWallet() - bob := newWallet() + _, err := localnet.FundBitcoin(alice.Address().EncodeAddress(), indexer) + Expect(err).To(BeNil()) - _, err := localnet.FundBitcoin(alice.Address().EncodeAddress(), indexer) - Expect(err).To(BeNil()) + send = func(wallet btc.Wallet, to btcutil.Address, amount int64) (string, btc.Transaction) { + txid, err := wallet.Send(context.Background(), []btc.SendRequest{{Amount: amount, To: to}}, nil, nil) + Expect(err).To(BeNil()) + tx, _, err := wallet.Status(context.Background(), txid) + Expect(err).To(BeNil()) + return txid, tx + } + }) - send := func(wallet btc.Wallet, to btcutil.Address, amount int64) (string, btc.Transaction) { - txid, err := wallet.Send(context.Background(), []btc.SendRequest{{Amount: amount, To: to}}, nil, nil) - Expect(err).To(BeNil()) - tx, _, err := wallet.Status(context.Background(), txid) - Expect(err).To(BeNil()) - return txid, tx - } It("should return mempool entry", func(ctx context.Context) { // Helper to check mempool entry fee diff --git a/btc/indexer.go b/btc/indexer.go index dd891a9..376f5d6 100644 --- a/btc/indexer.go +++ b/btc/indexer.go @@ -421,9 +421,11 @@ func (client *electrsIndexerClient) FeeEstimate(ctx context.Context) (FeeSuggest return fees, err } +// retry runs f until it succeeds, ctx is done, or f returns NoRetryError. +// Waits use exponential backoff from dur; any ctx deadline cancels the current +// wait early via select on ctx.Done(). func retry(logger *zap.Logger, ctx context.Context, dur time.Duration, f func() error) error { - ticker := time.NewTicker(dur) - defer ticker.Stop() + backoff := dur err := f() for err != nil { @@ -433,13 +435,22 @@ func retry(logger *zap.Logger, ctx context.Context, dur time.Duration, f func() return err } - logger.Debug("retrying", zap.Any("error", err.Error())) + logger.Debug("retrying", zap.Any("error", err.Error()), zap.Duration("backoff", backoff)) + timer := time.NewTimer(backoff) select { case <-ctx.Done(): + if !timer.Stop() { + <-timer.C + } return fmt.Errorf("%v : %v", ctx.Err(), err) - case <-ticker.C: + case <-timer.C: err = f() } + if err != nil { + if next := backoff * 2; next > backoff { + backoff = next + } + } } return nil } diff --git a/btc/indexer_test.go b/btc/indexer_test.go index 341e103..b275136 100644 --- a/btc/indexer_test.go +++ b/btc/indexer_test.go @@ -5,6 +5,9 @@ import ( "encoding/hex" "errors" "fmt" + "net/http" + "net/http/httptest" + "sync/atomic" "time" "github.com/btcsuite/btcd/btcec/v2" @@ -13,12 +16,52 @@ import ( "github.com/catalogfi/blockchain/btc" "github.com/catalogfi/blockchain/localnet" "github.com/fatih/color" + "go.uber.org/zap" . "github.com/onsi/ginkgo/v2" . "github.com/onsi/gomega" ) var _ = Describe("Indexer client", func() { + Context("electrs HTTP retry", func() { + It("retries with backoff until the indexer responds successfully", func() { + var attempts atomic.Int64 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + n := attempts.Add(1) + if r.URL.Path != "/blocks/tip/height" { + http.NotFound(w, r) + return + } + if n < 3 { + w.WriteHeader(http.StatusInternalServerError) + return + } + _, _ = w.Write([]byte("42")) + })) + defer srv.Close() + + c := btc.NewElectrsIndexerClient(zap.NewNop(), srv.URL, 5*time.Millisecond) + tip, err := c.GetTipBlockHeight(context.Background()) + Expect(err).To(BeNil()) + Expect(tip).To(Equal(uint64(42))) + Expect(attempts.Load()).To(BeNumerically(">=", 3)) + }) + + It("stops waiting and returns when the context is cancelled during backoff", func() { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusInternalServerError) + })) + defer srv.Close() + + c := btc.NewElectrsIndexerClient(zap.NewNop(), srv.URL, time.Second) + ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond) + defer cancel() + _, err := c.GetTipBlockHeight(ctx) + Expect(err).To(HaveOccurred()) + Expect(err.Error()).To(ContainSubstring(context.DeadlineExceeded.Error())) + }) + }) + Context("When using electrs API", func() { It("should be able to fetch the utxos of an address ", func() { By("GetUTXOs()") From 77ed11e7c4139ee4a1ec46801ec38714e017368c Mon Sep 17 00:00:00 2001 From: raagan-u Date: Mon, 23 Mar 2026 16:46:25 +0530 Subject: [PATCH 02/13] revert to ticks --- btc/indexer.go | 47 +++++++++++++++++++++++++++++++++++------------ 1 file changed, 35 insertions(+), 12 deletions(-) diff --git a/btc/indexer.go b/btc/indexer.go index 376f5d6..1f19a81 100644 --- a/btc/indexer.go +++ b/btc/indexer.go @@ -424,8 +424,40 @@ func (client *electrsIndexerClient) FeeEstimate(ctx context.Context) (FeeSuggest // retry runs f until it succeeds, ctx is done, or f returns NoRetryError. // Waits use exponential backoff from dur; any ctx deadline cancels the current // wait early via select on ctx.Done(). +// func retry(logger *zap.Logger, ctx context.Context, dur time.Duration, f func() error) error { +// backoff := dur + +// err := f() +// for err != nil { +// // Skip retrying if it's a `NoRetryError` +// var e *NoRetryError +// if errors.As(err, &e) { +// return err +// } + +// logger.Debug("retrying", zap.Any("error", err.Error()), zap.Duration("backoff", backoff)) +// timer := time.NewTimer(backoff) +// select { +// case <-ctx.Done(): +// if !timer.Stop() { +// <-timer.C +// } +// return fmt.Errorf("%v : %v", ctx.Err(), err) +// case <-timer.C: +// err = f() +// } +// if err != nil { +// if next := backoff * 2; next > backoff { +// backoff = next +// } +// } +// } +// return nil +// } + func retry(logger *zap.Logger, ctx context.Context, dur time.Duration, f func() error) error { - backoff := dur + ticker := time.NewTicker(dur) + defer ticker.Stop() err := f() for err != nil { @@ -435,22 +467,13 @@ func retry(logger *zap.Logger, ctx context.Context, dur time.Duration, f func() return err } - logger.Debug("retrying", zap.Any("error", err.Error()), zap.Duration("backoff", backoff)) - timer := time.NewTimer(backoff) + logger.Debug("retrying", zap.Any("error", err.Error())) select { case <-ctx.Done(): - if !timer.Stop() { - <-timer.C - } return fmt.Errorf("%v : %v", ctx.Err(), err) - case <-timer.C: + case <-ticker.C: err = f() } - if err != nil { - if next := backoff * 2; next > backoff { - backoff = next - } - } } return nil } From 74f19e2b60cbad2b6d2ae424bb40dd7cdc32740f Mon Sep 17 00:00:00 2001 From: raagan-u Date: Mon, 23 Mar 2026 16:46:44 +0530 Subject: [PATCH 03/13] improve batch submission logic --- btc/rbf.go | 51 +++++++++++++++++++++++++++++++++++++++++++++------ 1 file changed, 45 insertions(+), 6 deletions(-) diff --git a/btc/rbf.go b/btc/rbf.go index 9d85664..8f887f2 100644 --- a/btc/rbf.go +++ b/btc/rbf.go @@ -263,15 +263,54 @@ func (w *batcherWallet) createNewRBFBatch(c context.Context, previousUTXOs UTXOs return err } - w.logger.Info("submitted rbf tx", zap.String("txid", tx.TxHash().String())) + txID := tx.TxHash().String() + w.logger.Info("submitted rbf tx", zap.String("txid", txID)) + + // Retrieve the full transaction from the indexer. + // This is critical without it the batch won't be persisted and + // we'll hit "no confirmed batch found" errors. We use dedicated + // contexts (detached from the caller) so that upstream cancellation cannot prevent us from + // fetching and persisting the batch after a successful submission. + // + // Each attempt gives the indexer DefaultAPITimeout (60s) where its internal tick-based + // retry will keep polling. If it fails, we check the Bitcoin node's mempool: + // - Node has the tx → indexer is lagging, retry indexer.GetTx. + // - Node doesn't have it → re-submit the tx, then retry. + const maxRetrievalAttempts = 5 var transaction Transaction - err = withContextTimeout(c, DefaultAPITimeout, func(ctx context.Context) error { - transaction, err = w.indexer.GetTx(ctx, tx.TxHash().String()) - return err - }) + for attempt := 0; attempt < maxRetrievalAttempts; attempt++ { + retrievalCtx, retrievalCancel := context.WithTimeout(context.Background(), DefaultAPITimeout) + transaction, err = w.indexer.GetTx(retrievalCtx, txID) + retrievalCancel() + if err == nil { + break + } + w.logger.Error("failed to get tx from indexer", zap.Error(err), zap.String("txid", txID), zap.Int("attempt", attempt+1)) + + // Check if the node has the tx in its mempool before deciding to re-submit + mempoolCtx, mempoolCancel := context.WithTimeout(context.Background(), DefaultAPITimeout) + _, mempoolErr := w.rpc.GetMempoolEntry(mempoolCtx, txID) + mempoolCancel() + + if mempoolErr != nil { + // Node doesn't have it — re-submit and retry + w.logger.Warn("tx not found in node mempool, re-submitting", zap.Error(mempoolErr), zap.String("txid", txID)) + resubmitCtx, resubmitCancel := context.WithTimeout(context.Background(), DefaultAPITimeout) + if resubmitErr := w.indexer.SubmitTx(resubmitCtx, tx); resubmitErr != nil { + w.logger.Warn("failed to re-submit tx", zap.Error(resubmitErr), zap.String("txid", txID)) + } else { + w.logger.Info("re-submitted tx", zap.String("txid", txID)) + } + resubmitCancel() + } else { + // Node has the tx but indexer hasn't caught up — next attempt will retry + w.logger.Info("tx found in node mempool, indexer likely lagging behind", zap.String("txid", txID)) + } + } if err != nil { - return err + w.logger.Error("exhausted all attempts to retrieve tx", zap.String("txid", txID)) + return fmt.Errorf("failed to retrieve tx %s after submission: %w", txID, err) } // Create a new batch with the transaction details and save it to the cache From 9f6b7165512a04c450f8c4e380aa58cc508a1b37 Mon Sep 17 00:00:00 2001 From: raagan-u Date: Mon, 23 Mar 2026 16:51:44 +0530 Subject: [PATCH 04/13] add final attempt and submission check --- btc/rbf.go | 17 ++++++++++++++--- 1 file changed, 14 insertions(+), 3 deletions(-) diff --git a/btc/rbf.go b/btc/rbf.go index 8f887f2..4e747a8 100644 --- a/btc/rbf.go +++ b/btc/rbf.go @@ -276,9 +276,10 @@ func (w *batcherWallet) createNewRBFBatch(c context.Context, previousUTXOs UTXOs // retry will keep polling. If it fails, we check the Bitcoin node's mempool: // - Node has the tx → indexer is lagging, retry indexer.GetTx. // - Node doesn't have it → re-submit the tx, then retry. - const maxRetrievalAttempts = 5 + const maxRetrievalAttempts = 3 var transaction Transaction + extendedRetry := false for attempt := 0; attempt < maxRetrievalAttempts; attempt++ { retrievalCtx, retrievalCancel := context.WithTimeout(context.Background(), DefaultAPITimeout) transaction, err = w.indexer.GetTx(retrievalCtx, txID) @@ -298,14 +299,24 @@ func (w *batcherWallet) createNewRBFBatch(c context.Context, previousUTXOs UTXOs w.logger.Warn("tx not found in node mempool, re-submitting", zap.Error(mempoolErr), zap.String("txid", txID)) resubmitCtx, resubmitCancel := context.WithTimeout(context.Background(), DefaultAPITimeout) if resubmitErr := w.indexer.SubmitTx(resubmitCtx, tx); resubmitErr != nil { + // If the previous submission already went through (e.g. "already in mempool" + // or "already in chain"), this is harmless — we just need to retry GetTx. w.logger.Warn("failed to re-submit tx", zap.Error(resubmitErr), zap.String("txid", txID)) } else { w.logger.Info("re-submitted tx", zap.String("txid", txID)) } resubmitCancel() } else { - // Node has the tx but indexer hasn't caught up — next attempt will retry - w.logger.Info("tx found in node mempool, indexer likely lagging behind", zap.String("txid", txID)) + // Node has the tx but indexer hasn't caught up — if this is the last attempt, + // grant one extra retry since we know the tx exists. Only extend once to + // avoid looping indefinitely. + if attempt == maxRetrievalAttempts-1 && !extendedRetry { + attempt-- + extendedRetry = true + w.logger.Info("tx confirmed in node mempool on last attempt, retrying indexer once more", zap.String("txid", txID)) + } else { + w.logger.Info("tx found in node mempool, indexer likely lagging behind", zap.String("txid", txID)) + } } } if err != nil { From 73da593625cbc0a54ba4aca758ce887542fd9348 Mon Sep 17 00:00:00 2001 From: raagan-u Date: Mon, 23 Mar 2026 18:07:04 +0530 Subject: [PATCH 05/13] update save logic --- btc/indexer.go | 45 ++++++++++----------------------------------- btc/rbf.go | 45 +++++++++++++++++++++++++++++++-------------- 2 files changed, 41 insertions(+), 49 deletions(-) diff --git a/btc/indexer.go b/btc/indexer.go index 1f19a81..a4ce806 100644 --- a/btc/indexer.go +++ b/btc/indexer.go @@ -306,7 +306,16 @@ func (client *electrsIndexerClient) GetTx(ctx context.Context, txid string) (Tra // Send the request var tx Transaction if err := retry(client.logger, ctx, client.retryInterval, func() error { - resp, err := http.Get(endpoint) + // Per-request deadline so a single slow/hanging response doesn't block + // for the entire parent context timeout. + reqCtx, reqCancel := context.WithTimeout(ctx, client.retryInterval) + defer reqCancel() + + req, err := http.NewRequestWithContext(reqCtx, http.MethodGet, endpoint, nil) + if err != nil { + return err + } + resp, err := http.DefaultClient.Do(req) if err != nil { return err } @@ -421,40 +430,6 @@ func (client *electrsIndexerClient) FeeEstimate(ctx context.Context) (FeeSuggest return fees, err } -// retry runs f until it succeeds, ctx is done, or f returns NoRetryError. -// Waits use exponential backoff from dur; any ctx deadline cancels the current -// wait early via select on ctx.Done(). -// func retry(logger *zap.Logger, ctx context.Context, dur time.Duration, f func() error) error { -// backoff := dur - -// err := f() -// for err != nil { -// // Skip retrying if it's a `NoRetryError` -// var e *NoRetryError -// if errors.As(err, &e) { -// return err -// } - -// logger.Debug("retrying", zap.Any("error", err.Error()), zap.Duration("backoff", backoff)) -// timer := time.NewTimer(backoff) -// select { -// case <-ctx.Done(): -// if !timer.Stop() { -// <-timer.C -// } -// return fmt.Errorf("%v : %v", ctx.Err(), err) -// case <-timer.C: -// err = f() -// } -// if err != nil { -// if next := backoff * 2; next > backoff { -// backoff = next -// } -// } -// } -// return nil -// } - func retry(logger *zap.Logger, ctx context.Context, dur time.Duration, f func() error) error { ticker := time.NewTicker(dur) defer ticker.Stop() diff --git a/btc/rbf.go b/btc/rbf.go index 4e747a8..56cafbd 100644 --- a/btc/rbf.go +++ b/btc/rbf.go @@ -279,8 +279,9 @@ func (w *batcherWallet) createNewRBFBatch(c context.Context, previousUTXOs UTXOs const maxRetrievalAttempts = 3 var transaction Transaction - extendedRetry := false - for attempt := 0; attempt < maxRetrievalAttempts; attempt++ { + hasAppearedInMemppool := false + attempt := 0 + for attempt < maxRetrievalAttempts { retrievalCtx, retrievalCancel := context.WithTimeout(context.Background(), DefaultAPITimeout) transaction, err = w.indexer.GetTx(retrievalCtx, txID) retrievalCancel() @@ -295,6 +296,20 @@ func (w *batcherWallet) createNewRBFBatch(c context.Context, previousUTXOs UTXOs mempoolCancel() if mempoolErr != nil { + // if the tx has already appeared in mempool before, we return an error + if hasAppearedInMemppool { + // tx could have been confirmed, so we check one last time if it is confirmed + ctx, cancel := context.WithTimeout(context.Background(), DefaultAPITimeout) + confirmedTx, err := w.indexer.GetTx(ctx, txID) + cancel() + if err == nil && confirmedTx.Status.Confirmed { + w.logger.Info("tx confirmed in chain", zap.String("txid", txID)) + transaction = confirmedTx + break + } + w.logger.Error("tx not found in node mempool, but has already appeared in mempool", zap.Error(mempoolErr), zap.String("txid", txID)) + return fmt.Errorf("tx not found in node mempool, but has already appeared in mempool before : %w", mempoolErr) + } // Node doesn't have it — re-submit and retry w.logger.Warn("tx not found in node mempool, re-submitting", zap.Error(mempoolErr), zap.String("txid", txID)) resubmitCtx, resubmitCancel := context.WithTimeout(context.Background(), DefaultAPITimeout) @@ -307,21 +322,23 @@ func (w *batcherWallet) createNewRBFBatch(c context.Context, previousUTXOs UTXOs } resubmitCancel() } else { - // Node has the tx but indexer hasn't caught up — if this is the last attempt, - // grant one extra retry since we know the tx exists. Only extend once to - // avoid looping indefinitely. - if attempt == maxRetrievalAttempts-1 && !extendedRetry { - attempt-- - extendedRetry = true - w.logger.Info("tx confirmed in node mempool on last attempt, retrying indexer once more", zap.String("txid", txID)) - } else { - w.logger.Info("tx found in node mempool, indexer likely lagging behind", zap.String("txid", txID)) - } + // it is in the node, but not in indexer yet, so we wait indefinitely for the indexer to catch up + hasAppearedInMemppool = true + w.logger.Info("tx found in node mempool, waiting for indexer to catch up", zap.String("txid", txID)) + continue; } + attempt++ } if err != nil { - w.logger.Error("exhausted all attempts to retrieve tx", zap.String("txid", txID)) - return fmt.Errorf("failed to retrieve tx %s after submission: %w", txID, err) + w.logger.Error("exhausted all attempts to retrieve tx, storing local tx details", zap.String("txid", txID)) + transaction = Transaction{ + TxID: txID, + Version: int(tx.Version), + LockTime: int(tx.LockTime), + Status: Status{ + Confirmed: false, + }, + } } // Create a new batch with the transaction details and save it to the cache From 8fde309aa09e2a892bae387146179cf6fdb61402 Mon Sep 17 00:00:00 2001 From: raagan-u Date: Mon, 23 Mar 2026 18:38:07 +0530 Subject: [PATCH 06/13] review fixes --- btc/batcher.go | 2 +- btc/indexer.go | 39 +++++++++++++++++++++++++++++++++++++-- btc/rbf.go | 2 +- 3 files changed, 39 insertions(+), 4 deletions(-) diff --git a/btc/batcher.go b/btc/batcher.go index 83adf2a..820c904 100644 --- a/btc/batcher.go +++ b/btc/batcher.go @@ -20,7 +20,7 @@ import ( var ( GuardianSpendWeight = 1 + 1 + 1 + 73 + 1 + 73 + 1 + 1 + 1 + 200 // TODO: update Guardian Script Size to the right number SegwitSpendWeight = txsizes.RedeemP2WPKHInputWitnessWeight - DefaultAPITimeout = 60 * time.Second + DefaultAPITimeout = 300 * time.Second ) var ( diff --git a/btc/indexer.go b/btc/indexer.go index a4ce806..59acbe4 100644 --- a/btc/indexer.go +++ b/btc/indexer.go @@ -305,10 +305,11 @@ func (client *electrsIndexerClient) GetTx(ctx context.Context, txid string) (Tra // Send the request var tx Transaction - if err := retry(client.logger, ctx, client.retryInterval, func() error { + baseBackoff := 2 * time.Second + if err := exponentialBackoffRetry(client.logger, ctx, baseBackoff, func(t time.Duration) error { // Per-request deadline so a single slow/hanging response doesn't block // for the entire parent context timeout. - reqCtx, reqCancel := context.WithTimeout(ctx, client.retryInterval) + reqCtx, reqCancel := context.WithTimeout(ctx, t) defer reqCancel() req, err := http.NewRequestWithContext(reqCtx, http.MethodGet, endpoint, nil) @@ -453,6 +454,40 @@ func retry(logger *zap.Logger, ctx context.Context, dur time.Duration, f func() return nil } +// retry runs f until it succeeds, ctx is done, or f returns NoRetryError. +// Waits use exponential backoff from dur; any ctx deadline cancels the current +// wait early via select on ctx.Done(). +func exponentialBackoffRetry(logger *zap.Logger, ctx context.Context, dur time.Duration, f func(t time.Duration) error) error { + backoff := dur + + err := f(backoff) + for err != nil { + // Skip retrying if it's a `NoRetryError` + var e *NoRetryError + if errors.As(err, &e) { + return err + } + + logger.Debug("retrying", zap.Any("error", err.Error()), zap.Duration("backoff", backoff)) + timer := time.NewTimer(backoff) + select { + case <-ctx.Done(): + if !timer.Stop() { + <-timer.C + } + return fmt.Errorf("%v : %v", ctx.Err(), err) + case <-timer.C: + err = f(backoff) + } + if err != nil { + if next := backoff * 2; next > backoff { + backoff = next + } + } + } + return nil +} + type NoRetryError struct { err error } diff --git a/btc/rbf.go b/btc/rbf.go index 56cafbd..84e210b 100644 --- a/btc/rbf.go +++ b/btc/rbf.go @@ -276,7 +276,7 @@ func (w *batcherWallet) createNewRBFBatch(c context.Context, previousUTXOs UTXOs // retry will keep polling. If it fails, we check the Bitcoin node's mempool: // - Node has the tx → indexer is lagging, retry indexer.GetTx. // - Node doesn't have it → re-submit the tx, then retry. - const maxRetrievalAttempts = 3 + const maxRetrievalAttempts = 5 var transaction Transaction hasAppearedInMemppool := false From 69bff7c86a5df9fd0592a96ec527856f12127c4a Mon Sep 17 00:00:00 2001 From: raagan-u Date: Tue, 24 Mar 2026 12:02:43 +0530 Subject: [PATCH 07/13] add tests --- btc/rbf_proxy_test.go | 365 ++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 365 insertions(+) create mode 100644 btc/rbf_proxy_test.go diff --git a/btc/rbf_proxy_test.go b/btc/rbf_proxy_test.go new file mode 100644 index 0000000..ba58bac --- /dev/null +++ b/btc/rbf_proxy_test.go @@ -0,0 +1,365 @@ +package btc_test + +import ( + "bytes" + "context" + "fmt" + "io" + "net/http" + "net/http/httptest" + "net/url" + "os" + "os/exec" + "strings" + "sync" + "sync/atomic" + "time" + + "github.com/btcsuite/btcd/btcec/v2" + "github.com/btcsuite/btcd/chaincfg" + "github.com/catalogfi/blockchain/btc" + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + "github.com/syndtr/goleveldb/leveldb" + "go.uber.org/zap" +) + +// delayProxy is a reverse proxy that adds configurable delays to matching requests. +type delayProxy struct { + target *url.URL + + mu sync.RWMutex + rules []delayRule + reqCount atomic.Int64 +} + +type delayRule struct { + // match returns true if this rule should apply to the request. + match func(r *http.Request, body []byte) bool + delay time.Duration +} + +func newDelayProxy(target string) *delayProxy { + u, err := url.Parse(target) + Expect(err).To(BeNil()) + return &delayProxy{target: u} +} + +func (p *delayProxy) addRule(match func(r *http.Request, body []byte) bool, delay time.Duration) { + p.mu.Lock() + defer p.mu.Unlock() + p.rules = append(p.rules, delayRule{match: match, delay: delay}) +} + +func (p *delayProxy) clearRules() { + p.mu.Lock() + defer p.mu.Unlock() + p.rules = nil +} + +func (p *delayProxy) ServeHTTP(w http.ResponseWriter, r *http.Request) { + p.reqCount.Add(1) + + // Read the body so we can inspect it for JSON-RPC method matching, + // then replay it to the upstream. + bodyBytes, _ := io.ReadAll(r.Body) + r.Body.Close() + + // Check delay rules + p.mu.RLock() + for _, rule := range p.rules { + if rule.match(r, bodyBytes) { + p.mu.RUnlock() + time.Sleep(rule.delay) + p.mu.RLock() + break + } + } + p.mu.RUnlock() + + // Build upstream request + upstreamURL := *p.target + upstreamURL.Path = r.URL.Path + upstreamURL.RawQuery = r.URL.RawQuery + + bodyReader := bytes.NewReader(bodyBytes) + upReq, err := http.NewRequestWithContext(r.Context(), r.Method, upstreamURL.String(), bodyReader) + if err != nil { + http.Error(w, err.Error(), http.StatusBadGateway) + return + } + upReq.Header = r.Header.Clone() + upReq.ContentLength = int64(len(bodyBytes)) + + resp, err := http.DefaultClient.Do(upReq) + if err != nil { + http.Error(w, err.Error(), http.StatusBadGateway) + return + } + defer resp.Body.Close() + + for k, vs := range resp.Header { + for _, v := range vs { + w.Header().Add(k, v) + } + } + w.WriteHeader(resp.StatusCode) + io.Copy(w, resp.Body) +} + +// mineAndWait mines n blocks and waits for the indexer to index them. +func mineAndWait(proxyIndexer btc.IndexerClient, n int) { + heightBefore, err := proxyIndexer.GetTipBlockHeight(context.Background()) + Expect(err).To(BeNil()) + + out, cmdErr := exec.Command("merry", "rpc", "--generate", fmt.Sprintf("%d", n)).CombinedOutput() + Expect(cmdErr).To(BeNil(), "merry rpc --generate failed: %s", string(out)) + + Eventually(func() bool { + h, err := proxyIndexer.GetTipBlockHeight(context.Background()) + if err != nil { + return false + } + return h >= heightBefore+uint64(n) + }, 30*time.Second, 1*time.Second).Should(BeTrue(), "indexer should catch up to new block height") +} + +var _ = Describe("BatchWallet:RBF:DelayProxy", Ordered, func() { + chainParams := &chaincfg.RegressionNetParams + logger, err := zap.NewDevelopment() + Expect(err).To(BeNil()) + + privateKey, err := btcec.NewPrivateKey() + Expect(err).To(BeNil()) + + mockFeeEstimator := NewMockFeeEstimator(1) + dbPath := "./test_rbf_proxy" + + // Real backends + const realIndexerURL = "http://0.0.0.0:30000" + const realNodeURL = "http://0.0.0.0:18443" + const nodeUser = "admin1" + const nodePass = "123" + + // Proxies + indexerProxy := newDelayProxy(realIndexerURL) + nodeProxy := newDelayProxy(realNodeURL) + + var indexerServer *httptest.Server + var nodeServer *httptest.Server + var wallet btc.BatcherWallet + var cache btc.Cache + var proxyIndexer btc.IndexerClient + + BeforeAll(func() { + // Start proxy servers + indexerServer = httptest.NewServer(indexerProxy) + nodeServer = httptest.NewServer(nodeProxy) + + // Create indexer client pointing at the proxy + proxyIndexer = btc.NewElectrsIndexerClient(logger, indexerServer.URL, 2*time.Second) + + // Create RPC client pointing at the node proxy + nodeProxyURL := nodeServer.URL + bitcoinRPC := btc.NewBitcoinClient(nodeUser, nodePass, nodeProxyURL) + + db, err := leveldb.OpenFile(dbPath, nil) + Expect(err).To(BeNil()) + + cache = btc.NewBatcherCache(db, "proxy_test", btc.RBF) + wallet, err = btc.NewBatcherWallet( + privateKey, + proxyIndexer, + mockFeeEstimator, + chainParams, + cache, + logger, + &bitcoinRPC, + btc.WithPTI(5*time.Second), + btc.WithStrategy(btc.RBF), + ) + Expect(err).To(BeNil()) + + // Fund the wallet via merry faucet directly + out, cmdErr := exec.Command("merry", "faucet", "--to", wallet.Address().EncodeAddress()).CombinedOutput() + Expect(cmdErr).To(BeNil(), "merry faucet failed: %s", string(out)) + fmt.Printf("[DelayProxy] Funded wallet: %s\n", strings.TrimSpace(string(out))) + + // Mine a block so the funding tx confirms + mineAndWait(proxyIndexer, 1) + + // Wait for the indexer to have confirmed UTXOs for the wallet. + Eventually(func() bool { + utxos, err := proxyIndexer.GetUTXOs(context.Background(), wallet.Address()) + if err != nil { + return false + } + return len(utxos) > 0 + }, 60*time.Second, 2*time.Second).Should(BeTrue(), "wallet should have UTXOs after funding+mining") + + // Start the wallet once — it stays running across all tests + err = wallet.Start(context.Background()) + Expect(err).To(BeNil()) + }) + + AfterAll(func() { + if wallet != nil { + wallet.Stop() + } + indexerServer.Close() + nodeServer.Close() + os.RemoveAll(dbPath) + }) + + // AfterEach: mine blocks to confirm the current batch so the batcher + // doesn't get stuck on updateRBF/getmempooldescendants for the next test. + AfterEach(func() { + indexerProxy.clearRules() + nodeProxy.clearRules() + mineAndWait(proxyIndexer, 1) + // Give the batcher a couple of ticks to recognize the confirmation + time.Sleep(10 * time.Second) + }) + + Context("when GetTx is delayed beyond a single retry interval", func() { + It("should still persist the batch after retrying", func() { + // Delay GetTx responses by 5s — longer than the 2s retry interval, + // but within the 300s DefaultAPITimeout. The per-request context + // timeout (retryInterval=2s) should cause it to retry, and eventually + // the request will succeed when we drop the delay. + getTxCallCount := atomic.Int64{} + indexerProxy.addRule(func(r *http.Request, body []byte) bool { + // Match GET /tx/ (not POST /tx which is SubmitTx) + if r.Method == http.MethodGet && strings.Contains(r.URL.Path, "/tx/") { + count := getTxCallCount.Add(1) + // Delay the first 3 calls, then let subsequent ones pass fast. + // This simulates the indexer being slow to index initially. + if count <= 3 { + return true + } + } + return false + }, 5*time.Second) + + // Send a transaction + req := []btc.SendRequest{ + { + Amount: 10000, + To: wallet.Address(), + }, + } + + id, err := wallet.Send(context.Background(), req, nil, nil) + Expect(err).To(BeNil()) + + // Wait for the batch to be processed + var tx btc.Transaction + var ok bool + Eventually(func() bool { + tx, ok, err = wallet.Status(context.Background(), id) + Expect(err).To(BeNil()) + return ok + }, 120*time.Second, 3*time.Second).Should(BeTrue()) + Expect(tx.TxID).ShouldNot(BeEmpty()) + + // Verify the batch was persisted in cache + batch, err := cache.ReadLatestBatch(context.Background()) + Expect(err).To(BeNil()) + Expect(batch.Tx.TxID).ShouldNot(BeEmpty()) + + fmt.Printf("[DelayProxy] Test1: Batch persisted with txid=%s after %d GetTx calls\n", batch.Tx.TxID, getTxCallCount.Load()) + }) + }) + + Context("when GetTx is completely unreachable but node has the tx", func() { + It("should find the tx via node mempool and keep retrying indexer", func() { + // Block ALL GetTx responses with a huge delay (simulating indexer down) + // but let SubmitTx pass through. The node will have it, so the retry + // loop should detect it via GetMempoolEntry and keep trying. + indexerProxy.addRule(func(r *http.Request, body []byte) bool { + return r.Method == http.MethodGet && strings.Contains(r.URL.Path, "/tx/") + }, 300*time.Second) // effectively blocks the call — will hit the per-request 2s timeout + + req := []btc.SendRequest{ + { + Amount: 10000, + To: wallet.Address(), + }, + } + + id, err := wallet.Send(context.Background(), req, nil, nil) + Expect(err).To(BeNil()) + + // Give it enough time for the first indexer attempt to fail and hit the node check + time.Sleep(15 * time.Second) + + // Now remove the delay — simulating indexer recovery + indexerProxy.clearRules() + + // The retry loop should eventually succeed + var tx btc.Transaction + var ok bool + Eventually(func() bool { + tx, ok, err = wallet.Status(context.Background(), id) + Expect(err).To(BeNil()) + return ok + }, 120*time.Second, 3*time.Second).Should(BeTrue()) + Expect(tx.TxID).ShouldNot(BeEmpty()) + + batch, err := cache.ReadLatestBatch(context.Background()) + Expect(err).To(BeNil()) + Expect(batch.Tx.TxID).ShouldNot(BeEmpty()) + + fmt.Printf("[DelayProxy] Test2: Batch persisted after indexer recovery, txid=%s\n", batch.Tx.TxID) + }) + }) + + Context("when both indexer and node are slow", func() { + It("should re-submit and eventually persist the batch", func() { + // Delay both GetTx and getmempoolentry initially + callCount := atomic.Int64{} + + indexerProxy.addRule(func(r *http.Request, body []byte) bool { + if r.Method == http.MethodGet && strings.Contains(r.URL.Path, "/tx/") { + count := callCount.Add(1) + return count <= 2 // delay first 2 GetTx calls + } + return false + }, 5*time.Second) + + nodeProxy.addRule(func(r *http.Request, body []byte) bool { + return strings.Contains(string(body), "getmempoolentry") + }, 4*time.Second) + + req := []btc.SendRequest{ + { + Amount: 10000, + To: wallet.Address(), + }, + } + + id, err := wallet.Send(context.Background(), req, nil, nil) + Expect(err).To(BeNil()) + + // Wait and then drop delays + time.Sleep(10 * time.Second) + indexerProxy.clearRules() + nodeProxy.clearRules() + + var tx btc.Transaction + var ok bool + Eventually(func() bool { + tx, ok, err = wallet.Status(context.Background(), id) + Expect(err).To(BeNil()) + return ok + }, 120*time.Second, 3*time.Second).Should(BeTrue()) + Expect(tx.TxID).ShouldNot(BeEmpty()) + + batch, err := cache.ReadLatestBatch(context.Background()) + Expect(err).To(BeNil()) + Expect(batch.Tx.TxID).ShouldNot(BeEmpty()) + + fmt.Printf("[DelayProxy] Test3: Batch persisted despite dual delays, txid=%s\n", batch.Tx.TxID) + }) + }) +}) From 995376d33dbc94fdd423bb5a7e5947cd89870e7b Mon Sep 17 00:00:00 2001 From: raagan-u Date: Fri, 27 Mar 2026 20:00:49 +0530 Subject: [PATCH 08/13] add warn log and fix context --- .DS_Store | Bin 0 -> 6148 bytes btc/indexer.go | 4 +++- btc/rbf.go | 8 ++------ 3 files changed, 5 insertions(+), 7 deletions(-) create mode 100644 .DS_Store diff --git a/.DS_Store b/.DS_Store new file mode 100644 index 0000000000000000000000000000000000000000..3ec3c971ed4a7fcd99d11f59214c2d401f132d34 GIT binary patch literal 6148 zcmeHKy-ve05I)mVDp(K$LckDKg!mnREtK*EeE~>|_)$v`e`1Wpz7iI)25@O&n zc!ADLd}m)9TS*ZM2qB+jf9c)X-{nhU*F>Z$=Zz(zQ6eg%F&2l>+(!`(UdnEb@)mhOo~}|>4I7g z-}uyA8nO9?K29p4zl^hW&qf`-iACls;~R*f_73m}-0u5Fll+cna_kzS({}eA$jM9Z z{;NEvE(*A=pEB&(Y~>MyzV`}v1-t@11$cc3&=^C(#Gqa}(CH%pun4y{T{y?Xb3Jv<+E8rDy6)2dW zqrCrb^*{f+gZwYAfLGvODWHngTD5{(lD#!^bG+9EXxnIPoR=8XL(s`~tS@*go Date: Fri, 27 Mar 2026 20:01:08 +0530 Subject: [PATCH 09/13] remove DS_Store --- .DS_Store | Bin 6148 -> 0 bytes 1 file changed, 0 insertions(+), 0 deletions(-) delete mode 100644 .DS_Store diff --git a/.DS_Store b/.DS_Store deleted file mode 100644 index 3ec3c971ed4a7fcd99d11f59214c2d401f132d34..0000000000000000000000000000000000000000 GIT binary patch literal 0 HcmV?d00001 literal 6148 zcmeHKy-ve05I)mVDp(K$LckDKg!mnREtK*EeE~>|_)$v`e`1Wpz7iI)25@O&n zc!ADLd}m)9TS*ZM2qB+jf9c)X-{nhU*F>Z$=Zz(zQ6eg%F&2l>+(!`(UdnEb@)mhOo~}|>4I7g z-}uyA8nO9?K29p4zl^hW&qf`-iACls;~R*f_73m}-0u5Fll+cna_kzS({}eA$jM9Z z{;NEvE(*A=pEB&(Y~>MyzV`}v1-t@11$cc3&=^C(#Gqa}(CH%pun4y{T{y?Xb3Jv<+E8rDy6)2dW zqrCrb^*{f+gZwYAfLGvODWHngTD5{(lD#!^bG+9EXxnIPoR=8XL(s`~tS@*go Date: Fri, 27 Mar 2026 20:04:13 +0530 Subject: [PATCH 10/13] add with default api timeout to other call sites --- btc/rbf.go | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/btc/rbf.go b/btc/rbf.go index a537326..596396a 100644 --- a/btc/rbf.go +++ b/btc/rbf.go @@ -117,8 +117,10 @@ func (w *batcherWallet) reSubmitBatchWithNewRequests(c context.Context, batch Ba return fmt.Errorf("transaction %s in the batch has no weight", batch.Tx.TxID) } + ctx, cancel := context.WithTimeout(c, DefaultAPITimeout) + defer cancel() // Calculate the current fee rate for the batch transaction. - rbfFeeInfo, err := w.rpc.GetRBFTxFeeInfo(c, batch.Tx.TxID) + rbfFeeInfo, err := w.rpc.GetRBFTxFeeInfo(ctx, batch.Tx.TxID) if err != nil { if !errors.Is(err, ErrTxNotFound) { w.logger.Error("failed to get RBF fee info", zap.Error(err), zap.String("txid", batch.Tx.TxID)) @@ -433,8 +435,10 @@ func (w *batcherWallet) updateRBF(c context.Context, requiredFeeRate int) error return nil } + ctx, cancel := context.WithTimeout(c, DefaultAPITimeout) + defer cancel() // get current tx fee info - feeInfo, err := w.rpc.GetRBFTxFeeInfo(c, tx.TxID) + feeInfo, err := w.rpc.GetRBFTxFeeInfo(ctx, tx.TxID) if err != nil { if !errors.Is(err, ErrTxNotFound) { w.logger.Error("failed to get RBF fee info", zap.Error(err), zap.String("txid", tx.TxID)) From cc76cec976c0b22b310347747e48f7a59338b9a7 Mon Sep 17 00:00:00 2001 From: raagan-u Date: Fri, 27 Mar 2026 20:25:46 +0530 Subject: [PATCH 11/13] add tests --- btc/indexer.go | 10 ++++++++-- btc/indexer_test.go | 39 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 47 insertions(+), 2 deletions(-) diff --git a/btc/indexer.go b/btc/indexer.go index e5fd41b..5835e77 100644 --- a/btc/indexer.go +++ b/btc/indexer.go @@ -298,8 +298,14 @@ func (client *electrsIndexerClient) GetTxHex(ctx context.Context, txid string) ( } func (client *electrsIndexerClient) GetTx(ctx context.Context, txid string) (Transaction, error) { - ctx, cancel := context.WithTimeout(ctx, DefaultAPITimeout) - defer cancel() + _, ok := ctx.Deadline() + // Always set a per-call timeout if one isn't already present, but avoid redundant context wrapping. + if !ok { + var cancel context.CancelFunc + ctx, cancel = context.WithTimeout(ctx, DefaultAPITimeout) + defer cancel() + } + endpoint, err := url.JoinPath(client.url, "tx", txid) if err != nil { return Transaction{}, err diff --git a/btc/indexer_test.go b/btc/indexer_test.go index b275136..e45bfb3 100644 --- a/btc/indexer_test.go +++ b/btc/indexer_test.go @@ -62,6 +62,45 @@ var _ = Describe("Indexer client", func() { }) }) + Context("GetTx deadline guard", func() { + It("applies DefaultAPITimeout when context has no deadline", func() { + // Server returns a valid tx JSON immediately, so the test completes fast + // even though GetTx internally adds a 300s timeout. + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + fmt.Fprint(w, `{"txid":"deadbeef","version":2,"locktime":0,"vin":[],"vout":[],"size":0,"weight":0,"fee":0,"status":{"confirmed":false}}`) + })) + defer srv.Close() + + c := btc.NewElectrsIndexerClient(zap.NewNop(), srv.URL, 5*time.Millisecond) + + // Pass context.Background() directly — no deadline. + // GetTx should add its own DefaultAPITimeout and succeed. + tx, err := c.GetTx(context.Background(), "deadbeef") + Expect(err).To(BeNil()) + Expect(tx.TxID).To(Equal("deadbeef")) + }) + + It("does not double-wrap when context already carries a deadline", func() { + // Server returns a valid tx JSON on the first attempt. + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + fmt.Fprint(w, `{"txid":"abc123","version":2,"locktime":0,"vin":[],"vout":[],"size":0,"weight":0,"fee":0,"status":{"confirmed":false}}`) + })) + defer srv.Close() + + c := btc.NewElectrsIndexerClient(zap.NewNop(), srv.URL, 5*time.Millisecond) + + // Caller already sets a tight deadline — GetTx should skip wrapping. + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + tx, err := c.GetTx(ctx, "abc123") + Expect(err).To(BeNil()) + Expect(tx.TxID).To(Equal("abc123")) + }) + }) + Context("When using electrs API", func() { It("should be able to fetch the utxos of an address ", func() { By("GetUTXOs()") From 220557d239967779aa2c6795bab0c4f831711b7e Mon Sep 17 00:00:00 2001 From: raagan-u Date: Fri, 27 Mar 2026 20:39:54 +0530 Subject: [PATCH 12/13] update deadline and default timeouts --- btc/indexer.go | 2 +- btc/rbf.go | 8 ++++++-- 2 files changed, 7 insertions(+), 3 deletions(-) diff --git a/btc/indexer.go b/btc/indexer.go index 5835e77..519bf22 100644 --- a/btc/indexer.go +++ b/btc/indexer.go @@ -302,7 +302,7 @@ func (client *electrsIndexerClient) GetTx(ctx context.Context, txid string) (Tra // Always set a per-call timeout if one isn't already present, but avoid redundant context wrapping. if !ok { var cancel context.CancelFunc - ctx, cancel = context.WithTimeout(ctx, DefaultAPITimeout) + ctx, cancel = context.WithTimeout(ctx, 60*time.Second) defer cancel() } diff --git a/btc/rbf.go b/btc/rbf.go index 596396a..4e8c016 100644 --- a/btc/rbf.go +++ b/btc/rbf.go @@ -284,7 +284,9 @@ func (w *batcherWallet) createNewRBFBatch(c context.Context, previousUTXOs UTXOs hasAppearedInMemppool := false attempt := 0 for attempt < maxRetrievalAttempts { - transaction, err = w.indexer.GetTx(c, txID) + ctx, cancel := context.WithTimeout(c, DefaultAPITimeout) + transaction, err = w.indexer.GetTx(ctx, txID) + cancel() if err == nil { break } @@ -298,8 +300,10 @@ func (w *batcherWallet) createNewRBFBatch(c context.Context, previousUTXOs UTXOs if mempoolErr != nil { // if the tx has already appeared in mempool before, we return an error if hasAppearedInMemppool { + ctx, cancel := context.WithTimeout(c, DefaultAPITimeout) // tx could have been confirmed, so we check one last time if it is confirmed - confirmedTx, err := w.indexer.GetTx(c, txID) + confirmedTx, err := w.indexer.GetTx(ctx, txID) + cancel() if err == nil && confirmedTx.Status.Confirmed { w.logger.Info("tx confirmed in chain", zap.String("txid", txID)) transaction = confirmedTx From f025cf9f0cf80e14a9bc57f1072416ea65562c15 Mon Sep 17 00:00:00 2001 From: raagan-u <73926723+raagan-u@users.noreply.github.com> Date: Mon, 27 Apr 2026 10:53:24 +0530 Subject: [PATCH 13/13] fix: update GetTx and add waitForTx() (#63) * update GetTx with NoRetryError returns and clean exponential backoff * add waitForTx * update tests * add per request timeout from backoff * add context with timeout for WaitTx * add TxIns and TxOuts for persisting on failure * skip status on fallback save * identify fallback save --- btc/indexer.go | 49 +++---- btc/rbf.go | 264 +++++++++++++++++++++++++++-------- btc/rbf_proxy_test.go | 316 ++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 544 insertions(+), 85 deletions(-) diff --git a/btc/indexer.go b/btc/indexer.go index 519bf22..50752aa 100644 --- a/btc/indexer.go +++ b/btc/indexer.go @@ -298,9 +298,7 @@ func (client *electrsIndexerClient) GetTxHex(ctx context.Context, txid string) ( } func (client *electrsIndexerClient) GetTx(ctx context.Context, txid string) (Transaction, error) { - _, ok := ctx.Deadline() - // Always set a per-call timeout if one isn't already present, but avoid redundant context wrapping. - if !ok { + if _, ok := ctx.Deadline(); !ok { var cancel context.CancelFunc ctx, cancel = context.WithTimeout(ctx, 60*time.Second) defer cancel() @@ -311,12 +309,9 @@ func (client *electrsIndexerClient) GetTx(ctx context.Context, txid string) (Tra return Transaction{}, err } - // Send the request var tx Transaction baseBackoff := 2 * time.Second if err := exponentialBackoffRetry(client.logger, ctx, baseBackoff, func(t time.Duration) error { - // Per-request deadline so a single slow/hanging response doesn't block - // for the entire parent context timeout. reqCtx, reqCancel := context.WithTimeout(ctx, t) defer reqCancel() @@ -333,14 +328,16 @@ func (client *electrsIndexerClient) GetTx(ctx context.Context, txid string) (Tra if resp.StatusCode != http.StatusOK { errMsg, err := io.ReadAll(resp.Body) if err != nil { - return fmt.Errorf("fail to read response from %s: %w", endpoint, err) + return NewNoRetryError(fmt.Errorf("failed to read response from %s: %w", endpoint, err)) } - return fmt.Errorf("GetTx : %v", string(errMsg)) + if resp.StatusCode == http.StatusNotFound || resp.StatusCode == http.StatusBadRequest { + return NewNoRetryError(fmt.Errorf("GetTx: %s", errMsg)) + } + return fmt.Errorf("GetTx: status %d: %s", resp.StatusCode, errMsg) } - // Decode response if err := json.NewDecoder(resp.Body).Decode(&tx); err != nil { - return fmt.Errorf("failed to decode UTXOs: %w", err) + return NewNoRetryError(fmt.Errorf("failed to decode transaction: %w", err)) } return nil }); err != nil { @@ -451,7 +448,7 @@ func retry(logger *zap.Logger, ctx context.Context, dur time.Duration, f func() return err } - logger.Debug("retrying", zap.Any("error", err.Error())) + logger.Warn("retrying", zap.Any("error", err.Error())) select { case <-ctx.Done(): return fmt.Errorf("%v : %v", ctx.Err(), err) @@ -462,38 +459,34 @@ func retry(logger *zap.Logger, ctx context.Context, dur time.Duration, f func() return nil } -// retry runs f until it succeeds, ctx is done, or f returns NoRetryError. -// Waits use exponential backoff from dur; any ctx deadline cancels the current -// wait early via select on ctx.Done(). -func exponentialBackoffRetry(logger *zap.Logger, ctx context.Context, dur time.Duration, f func(t time.Duration) error) error { - backoff := dur +// exponentialBackoffRetry runs f until it succeeds, ctx is done, or f returns +// a NoRetryError. Backoff doubles on each failure starting from base. +func exponentialBackoffRetry(logger *zap.Logger, ctx context.Context, base time.Duration, f func(t time.Duration) error) error { + backoff := base + + for { + err := f(backoff) + if err == nil { + return nil + } - err := f(backoff) - for err != nil { - // Skip retrying if it's a `NoRetryError` var e *NoRetryError if errors.As(err, &e) { return err } - logger.Warn("retrying", zap.Any("error", err.Error()), zap.Duration("backoff", backoff)) + logger.Warn("retrying", zap.Error(err), zap.Duration("backoff", backoff)) timer := time.NewTimer(backoff) select { case <-ctx.Done(): if !timer.Stop() { <-timer.C } - return fmt.Errorf("%v : %v", ctx.Err(), err) + return fmt.Errorf("%v : %w", ctx.Err(), err) case <-timer.C: - err = f(backoff) - } - if err != nil { - if next := backoff * 2; next > backoff { - backoff = next - } + backoff *= 2 } } - return nil } type NoRetryError struct { diff --git a/btc/rbf.go b/btc/rbf.go index 4e8c016..2ed27d5 100644 --- a/btc/rbf.go +++ b/btc/rbf.go @@ -7,6 +7,7 @@ import ( "fmt" "math" "strconv" + "time" "strings" "github.com/btcsuite/btcd/blockchain" @@ -281,65 +282,28 @@ func (w *batcherWallet) createNewRBFBatch(c context.Context, previousUTXOs UTXOs const maxRetrievalAttempts = 5 var transaction Transaction - hasAppearedInMemppool := false - attempt := 0 - for attempt < maxRetrievalAttempts { - ctx, cancel := context.WithTimeout(c, DefaultAPITimeout) - transaction, err = w.indexer.GetTx(ctx, txID) - cancel() - if err == nil { - break - } - w.logger.Error("failed to get tx from indexer", zap.Error(err), zap.String("txid", txID), zap.Int("attempt", attempt+1)) - - // Check if the node has the tx in its mempool before deciding to re-submit - mempoolCtx, mempoolCancel := context.WithTimeout(context.Background(), DefaultAPITimeout) - _, mempoolErr := w.rpc.GetMempoolEntry(mempoolCtx, txID) - mempoolCancel() - - if mempoolErr != nil { - // if the tx has already appeared in mempool before, we return an error - if hasAppearedInMemppool { - ctx, cancel := context.WithTimeout(c, DefaultAPITimeout) - // tx could have been confirmed, so we check one last time if it is confirmed - confirmedTx, err := w.indexer.GetTx(ctx, txID) - cancel() - if err == nil && confirmedTx.Status.Confirmed { - w.logger.Info("tx confirmed in chain", zap.String("txid", txID)) - transaction = confirmedTx - break - } - w.logger.Error("tx not found in node mempool, but has already appeared in mempool", zap.Error(mempoolErr), zap.String("txid", txID)) - return fmt.Errorf("tx not found in node mempool, but has already appeared in mempool before : %w", mempoolErr) - } - // Node doesn't have it — re-submit and retry - w.logger.Warn("tx not found in node mempool, re-submitting", zap.Error(mempoolErr), zap.String("txid", txID)) - resubmitCtx, resubmitCancel := context.WithTimeout(context.Background(), DefaultAPITimeout) - if resubmitErr := w.indexer.SubmitTx(resubmitCtx, tx); resubmitErr != nil { - // If the previous submission already went through (e.g. "already in mempool" - // or "already in chain"), this is harmless — we just need to retry GetTx. - w.logger.Warn("failed to re-submit tx", zap.Error(resubmitErr), zap.String("txid", txID)) - } else { - w.logger.Info("re-submitted tx", zap.String("txid", txID)) - } - resubmitCancel() - } else { - // it is in the node, but not in indexer yet, so we wait indefinitely for the indexer to catch up - hasAppearedInMemppool = true - w.logger.Info("tx found in node mempool, waiting for indexer to catch up", zap.String("txid", txID)) - continue; - } - attempt++ - } - if err != nil { - w.logger.Error("exhausted all attempts to retrieve tx, storing local tx details", zap.String("txid", txID)) + ctx, cancel := context.WithTimeout(c, DefaultAPITimeout) + indexerTx, err := w.waitForTx(ctx, tx, maxRetrievalAttempts) + cancel() + switch { + case err != nil: + // Something dangerous happened (vanished from mempool, resubmit + // failed to produce a fetchable tx). Do not populate the batch. + w.logger.Error("dangerous failure waiting for tx", zap.String("txid", txID), zap.Error(err)) + return err + case indexerTx != nil: + transaction = *indexerTx + default: + // (nil, nil): indexer never returned the tx but nothing went wrong + // (pure indexer lag). Safe to fall back to local tx details. + w.logger.Warn("indexer did not return tx, using local tx details", zap.String("txid", txID)) transaction = Transaction{ - TxID: txID, - Version: int(tx.Version), + TxID: txID, + Version: int(tx.Version), LockTime: int(tx.LockTime), - Status: Status{ - Confirmed: false, - }, + Weight: -1, // updating with -1 to indicate that the batch was not a successful save to the cache + VINs: wireTxInsToVINs(tx.TxIn), + VOUTs: wireTxOutsToPrevouts(tx.TxOut), } } @@ -858,6 +822,192 @@ func (w *batcherWallet) getUnconfirmedUtxos(ctx context.Context) (map[string]boo return avoidUtxos, nil } +// waitForTx attempts to retrieve a transaction from the indexer after it has +// been submitted. It is the caller's responsibility to ensure the transaction +// has already been broadcast before calling this function. +// +// Return values: +// - (*Transaction, nil) — indexer returned the tx, use it directly. +// - (nil, nil) — the tx is in the node mempool but the indexer +// never caught up within maxMempoolWaits (~60 s). Nothing dangerous; +// the caller may safely fall back to local tx details. +// - (nil, error) — something dangerous happened and the caller must +// NOT populate a batch. This includes: tx vanished from the node mempool +// after previously being seen, tx still not found after a resubmission, +// or the parent context was cancelled. +// +// Decision tree per iteration: +// 1. Indexer has the tx → return (*tx, nil). +// 2. Indexer doesn't have it, node mempool has it → indexer is lagging. +// Wait retryDelay and retry without burning an attempt. Capped at +// maxMempoolWaits iterations; if exceeded → return (nil, nil). +// 3. Indexer doesn't have it, node mempool doesn't have it: +// a. If previously seen in the mempool → it may have been confirmed +// and evicted. One final indexer check; if confirmed return (*tx, nil), +// otherwise return (nil, error). +// b. If we already resubmitted → return (nil, error). We submitted but +// still can't verify the tx; populating a batch would be dangerous. +// c. Otherwise → re-submit the tx once, mark hasResubmitted, increment +// attempt, wait retryDelay, and retry. +// +// In practice the attempt counter only increments on resubmission (step 3c), +// which can happen at most once. Subsequent iterations where neither the +// indexer nor the mempool has the tx exit immediately via step 3b. The for +// loop's maxAttempts exit is therefore a defensive fallback. +func (w *batcherWallet) waitForTx(c context.Context, tx *wire.MsgTx, maxAttempts int) (*Transaction, error) { + const retryDelay = 2 * time.Second + + txID := tx.TxHash().String() + seenInMempool := false + hasResubmitted := false + mempoolWaits := 0 + // maxMempoolWaits caps how long we wait for the indexer to catch up when + // the tx is confirmed in the node mempool. With retryDelay=2s this gives + // ~60s before we give up and let the caller use local tx details. + const maxMempoolWaits = 30 + + for attempt := 0; attempt < maxAttempts; { + // 1. Ask the indexer. + ctx, cancel := context.WithTimeout(c, DefaultAPITimeout) + transaction, err := w.indexer.GetTx(ctx, txID) + cancel() + if err == nil { + return &transaction, nil + } + + w.logger.Error("failed to get tx from indexer", + zap.Error(err), + zap.String("txid", txID), + zap.Int("attempt", attempt+1), + zap.Int("maxAttempts", maxAttempts), + ) + + // 2. Ask the node mempool. + mempoolCtx, mempoolCancel := context.WithTimeout(c, DefaultAPITimeout) + _, mempoolErr := w.rpc.GetMempoolEntry(mempoolCtx, txID) + mempoolCancel() + + if mempoolErr == nil { + // Node has it, indexer is just lagging — wait and retry without + // burning an attempt. + seenInMempool = true + mempoolWaits++ + if mempoolWaits > maxMempoolWaits { + w.logger.Warn("indexer still hasn't caught up after max mempool waits, giving up", + zap.String("txid", txID), + zap.Int("mempoolWaits", mempoolWaits), + ) + return nil, nil + } + w.logger.Info("tx found in node mempool, waiting for indexer to catch up", + zap.String("txid", txID), + zap.Int("mempoolWait", mempoolWaits), + zap.Int("maxMempoolWaits", maxMempoolWaits), + ) + select { + case <-time.After(retryDelay): + case <-c.Done(): + return nil, fmt.Errorf("context cancelled waiting for indexer to catch up for tx %s: %w", txID, c.Err()) + } + continue + } + + // 3. Neither indexer nor node has the tx. + if seenInMempool { + // It was in the mempool before but has now vanished — it likely + // got confirmed and evicted. Do one final indexer check before + // giving up. + finalCtx, finalCancel := context.WithTimeout(c, DefaultAPITimeout) + confirmedTx, finalErr := w.indexer.GetTx(finalCtx, txID) + finalCancel() + if finalErr == nil && confirmedTx.Status.Confirmed { + w.logger.Info("tx confirmed in chain", zap.String("txid", txID)) + return &confirmedTx, nil + } + return nil, fmt.Errorf("tx %s vanished from node mempool after being seen: %w", txID, mempoolErr) + } + + if hasResubmitted { + // We already resubmitted but still can't find the tx anywhere. + // Dangerous to proceed — the tx state is unknown. + return nil, fmt.Errorf("tx %s not found after resubmission, refusing to populate batch", txID) + } + + // Never seen in mempool and haven't resubmitted yet — submit once. + w.logger.Warn("tx not found in node mempool, re-submitting", + zap.Error(mempoolErr), + zap.String("txid", txID), + zap.Int("attempt", attempt+1), + ) + resubmitCtx, resubmitCancel := context.WithTimeout(c, DefaultAPITimeout) + if resubmitErr := w.indexer.SubmitTx(resubmitCtx, tx); resubmitErr != nil { + w.logger.Warn("failed to re-submit tx", zap.Error(resubmitErr), zap.String("txid", txID)) + } else { + w.logger.Info("re-submitted tx", zap.String("txid", txID)) + } + resubmitCancel() + hasResubmitted = true + attempt++ + + // Give the resubmitted tx time to propagate before retrying. + select { + case <-time.After(retryDelay): + case <-c.Done(): + return nil, fmt.Errorf("context cancelled after resubmitting tx %s: %w", txID, c.Err()) + } + } + + // This is unreachable in practice: attempt only increments on resubmit + // (once), and subsequent loops with hasResubmitted=true exit via path D. + // Kept as a defensive fallback. + return nil, fmt.Errorf("exhausted %d attempts to retrieve tx %s", maxAttempts, txID) +} + +// wireTxInsToVINs converts a slice of *wire.TxIn into our []VIN shape for +// local batch persistence when the indexer has not yet returned the tx. +// +// Prevout cannot be derived from a wire.TxIn (it describes the output being +// spent, which lives in a prior transaction) and is left zero-valued. When the +// indexer catches up on the next RBF cycle, the batch is replaced with a +// fully-populated record. +func wireTxInsToVINs(txIns []*wire.TxIn) []VIN { + vins := make([]VIN, len(txIns)) + for i, in := range txIns { + var witness *[]string + if len(in.Witness) > 0 { + items := make([]string, len(in.Witness)) + for j, item := range in.Witness { + items[j] = hex.EncodeToString(item) + } + witness = &items + } + vins[i] = VIN{ + TxID: in.PreviousOutPoint.Hash.String(), + Vout: int(in.PreviousOutPoint.Index), + ScriptSig: hex.EncodeToString(in.SignatureScript), + Witness: witness, + Sequence: int(in.Sequence), + } + } + return vins +} + +// wireTxOutsToPrevouts converts a slice of *wire.TxOut into our []Prevout shape +// for local batch persistence. ScriptPubKeyType / ScriptPubKeyAddress are left +// empty: btcd's ScriptClass.String() format does not match the electrs wire +// format (e.g. "witness_v1_taproot" vs "v1_p2tr") used elsewhere in the +// codebase, so populating them here would produce subtly-wrong data. +func wireTxOutsToPrevouts(txOuts []*wire.TxOut) []Prevout { + prevouts := make([]Prevout, len(txOuts)) + for i, out := range txOuts { + prevouts[i] = Prevout{ + ScriptPubKey: hex.EncodeToString(out.PkScript), + Value: int(out.Value), + } + } + return prevouts +} + // buildRBFTransaction builds an unsigned transaction with the given UTXOs, recipients, change address, and fee // // checkValidity is used to determine if the transaction should be validated while building diff --git a/btc/rbf_proxy_test.go b/btc/rbf_proxy_test.go index ba58bac..53156f0 100644 --- a/btc/rbf_proxy_test.go +++ b/btc/rbf_proxy_test.go @@ -362,4 +362,320 @@ var _ = Describe("BatchWallet:RBF:DelayProxy", Ordered, func() { fmt.Printf("[DelayProxy] Test3: Batch persisted despite dual delays, txid=%s\n", batch.Tx.TxID) }) }) + + // --------------------------------------------------------------- + // waitForTx: indexer permanently 404, node always has it in mempool. + // After maxMempoolWaits (30 × 2s ≈ 60s) the function should return + // (nil, nil) and the batch should be persisted with local tx details + // (i.e. no VOUTs, no Weight, no Fee). + // --------------------------------------------------------------- + Context("when indexer permanently returns 404 but node has the tx (nil,nil path)", func() { + It("should persist the batch with local tx details after mempool-wait cap", func() { + // Return 404 instantly for GetTx — triggers NoRetryError so the + // indexer client returns immediately without backoff. + indexerProxy.addRule(func(r *http.Request, body []byte) bool { + return false // no delay needed — see intercept below + }, 0) + + // We need to intercept (not just delay) GetTx so it returns 404 + // during the submission window. Once waitForTx completes and the + // batch is persisted, we unblock GetTx so Status() can work. + var blockGetTx atomic.Bool + blockGetTx.Store(true) + interceptServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if blockGetTx.Load() && r.Method == http.MethodGet && strings.Contains(r.URL.Path, "/tx/") && !strings.Contains(r.URL.Path, "/txs") { + w.WriteHeader(http.StatusNotFound) + w.Write([]byte("Transaction not found")) + return + } + // Proxy everything else (and unblocked GetTx) to the real indexer + indexerProxy.ServeHTTP(w, r) + })) + defer interceptServer.Close() + + // Create a wallet that uses the intercept indexer + interceptIndexer := btc.NewElectrsIndexerClient(logger, interceptServer.URL, 2*time.Second) + + pk, err := btcec.NewPrivateKey() + Expect(err).To(BeNil()) + + interceptDBPath := "./test_rbf_proxy_nilnil" + db2, err := leveldb.OpenFile(interceptDBPath, nil) + Expect(err).To(BeNil()) + defer os.RemoveAll(interceptDBPath) + defer db2.Close() + + interceptCache := btc.NewBatcherCache(db2, "nilnil_test", btc.RBF) + nodeProxyURL := nodeServer.URL + bitcoinRPC2 := btc.NewBitcoinClient("admin1", "123", nodeProxyURL) + + wallet2, err := btc.NewBatcherWallet( + pk, + interceptIndexer, + NewMockFeeEstimator(1), + chainParams, + interceptCache, + logger, + &bitcoinRPC2, + btc.WithPTI(5*time.Second), + btc.WithStrategy(btc.RBF), + ) + Expect(err).To(BeNil()) + + // Fund via merry + out, cmdErr := exec.Command("merry", "faucet", "--to", wallet2.Address().EncodeAddress()).CombinedOutput() + Expect(cmdErr).To(BeNil(), "merry faucet failed: %s", string(out)) + mineAndWait(proxyIndexer, 1) + + // Wait for UTXOs — use the REAL indexer to confirm funding, because + // interceptIndexer blocks GetTx but UTXOs endpoint is fine. + Eventually(func() bool { + utxos, err := interceptIndexer.GetUTXOs(context.Background(), wallet2.Address()) + if err != nil { + return false + } + return len(utxos) > 0 + }, 60*time.Second, 2*time.Second).Should(BeTrue()) + + err = wallet2.Start(context.Background()) + Expect(err).To(BeNil()) + defer wallet2.Stop() + + req := []btc.SendRequest{ + {Amount: 10000, To: wallet2.Address()}, + } + + id, err := wallet2.Send(context.Background(), req, nil, nil) + Expect(err).To(BeNil()) + + // Wait for the batch to be persisted via the nil,nil path. + // waitForTx loops through ~30 mempool waits (≈60s), returns + // (nil,nil), and the caller saves the batch with local tx details. + var batch btc.Batch + Eventually(func() error { + b, bErr := interceptCache.ReadLatestBatch(context.Background()) + if bErr != nil { + return bErr + } + batch = b + return nil + }, 180*time.Second, 3*time.Second).Should(Succeed()) + Expect(batch.Tx.TxID).ShouldNot(BeEmpty()) + + // The batch should have NO VOUTs since it was built from local tx + // details (the nil,nil path), not from the indexer response. + Expect(batch.Tx.VOUTs).Should(BeEmpty()) + Expect(batch.Tx.Weight).Should(Equal(0)) + + // Now unblock GetTx so Status() can fetch from the real indexer. + blockGetTx.Store(false) + + // Mine a block so the indexer picks up the tx. + mineAndWait(proxyIndexer, 1) + + // Status should now succeed via the real indexer. + var tx btc.Transaction + var ok bool + Eventually(func() bool { + tx, ok, err = wallet2.Status(context.Background(), id) + if err != nil { + return false + } + return ok + }, 60*time.Second, 3*time.Second).Should(BeTrue()) + Expect(tx.TxID).ShouldNot(BeEmpty()) + + fmt.Printf("[DelayProxy] Test4 (nil,nil path): Batch persisted with local details, txid=%s\n", batch.Tx.TxID) + }) + }) + + // --------------------------------------------------------------- + // waitForTx: both indexer and node return errors. After resubmit, + // still nothing found → (nil, error). The batch should NOT be + // persisted and the batcher should error. + // --------------------------------------------------------------- + Context("when both indexer and node permanently reject the tx (nil,error path)", func() { + It("should not persist a batch when tx is unfindable after resubmit", func() { + // Intercept: indexer returns 404 for GetTx, AND node returns + // "Transaction not in mempool" for getmempoolentry. + interceptIndexerServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method == http.MethodGet && strings.Contains(r.URL.Path, "/tx/") && !strings.Contains(r.URL.Path, "/txs") { + w.WriteHeader(http.StatusNotFound) + w.Write([]byte("Transaction not found")) + return + } + indexerProxy.ServeHTTP(w, r) + })) + defer interceptIndexerServer.Close() + + // For the node, we need getmempoolentry to fail. We intercept the + // RPC and return a JSON-RPC error for getmempoolentry. + interceptNodeServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + bodyBytes, _ := io.ReadAll(r.Body) + r.Body.Close() + if strings.Contains(string(bodyBytes), "getmempoolentry") { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusOK) + w.Write([]byte(`{"result":null,"error":{"code":-5,"message":"Transaction not in mempool"},"id":1}`)) + return + } + // Replay the body for upstream + r.Body = io.NopCloser(bytes.NewReader(bodyBytes)) + nodeProxy.ServeHTTP(w, r) + })) + defer interceptNodeServer.Close() + + interceptIndexer := btc.NewElectrsIndexerClient(logger, interceptIndexerServer.URL, 2*time.Second) + bitcoinRPC3 := btc.NewBitcoinClient("admin1", "123", interceptNodeServer.URL) + + pk, err := btcec.NewPrivateKey() + Expect(err).To(BeNil()) + + interceptDBPath := "./test_rbf_proxy_nilerr" + db3, err := leveldb.OpenFile(interceptDBPath, nil) + Expect(err).To(BeNil()) + defer os.RemoveAll(interceptDBPath) + defer db3.Close() + + interceptCache := btc.NewBatcherCache(db3, "nilerr_test", btc.RBF) + + wallet3, err := btc.NewBatcherWallet( + pk, + interceptIndexer, + NewMockFeeEstimator(1), + chainParams, + interceptCache, + logger, + &bitcoinRPC3, + btc.WithPTI(5*time.Second), + btc.WithStrategy(btc.RBF), + ) + Expect(err).To(BeNil()) + + // Fund via merry + out, cmdErr := exec.Command("merry", "faucet", "--to", wallet3.Address().EncodeAddress()).CombinedOutput() + Expect(cmdErr).To(BeNil(), "merry faucet failed: %s", string(out)) + mineAndWait(proxyIndexer, 1) + + Eventually(func() bool { + utxos, err := interceptIndexer.GetUTXOs(context.Background(), wallet3.Address()) + if err != nil { + return false + } + return len(utxos) > 0 + }, 60*time.Second, 2*time.Second).Should(BeTrue()) + + err = wallet3.Start(context.Background()) + Expect(err).To(BeNil()) + defer wallet3.Stop() + + req := []btc.SendRequest{ + {Amount: 10000, To: wallet3.Address()}, + } + + _, err = wallet3.Send(context.Background(), req, nil, nil) + Expect(err).To(BeNil()) + + // Give the batcher enough time to attempt createNewRBFBatch. + // waitForTx should hit: GetTx fail → GetMempoolEntry fail → + // resubmit → loop → GetTx fail → GetMempoolEntry fail → + // hasResubmitted=true → return (nil, error). Total ≈ 6s. + time.Sleep(20 * time.Second) + + // The batch should NOT have been persisted because waitForTx + // returned (nil, error). + _, err = interceptCache.ReadLatestBatch(context.Background()) + Expect(err).ShouldNot(BeNil()) + + fmt.Println("[DelayProxy] Test5 (nil,error path): Batch correctly NOT persisted after resubmit failure") + }) + }) + + // --------------------------------------------------------------- + // waitForTx: context cancellation while waiting for indexer to + // catch up (tx is in node mempool). Should return (nil, error). + // --------------------------------------------------------------- + Context("when context is cancelled during mempool-wait loop", func() { + It("should return context cancellation error and not persist batch", func() { + // Block GetTx permanently so waitForTx enters the mempool-wait loop. + interceptServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method == http.MethodGet && strings.Contains(r.URL.Path, "/tx/") && !strings.Contains(r.URL.Path, "/txs") { + w.WriteHeader(http.StatusNotFound) + w.Write([]byte("Transaction not found")) + return + } + indexerProxy.ServeHTTP(w, r) + })) + defer interceptServer.Close() + + interceptIndexer := btc.NewElectrsIndexerClient(logger, interceptServer.URL, 2*time.Second) + + pk, err := btcec.NewPrivateKey() + Expect(err).To(BeNil()) + + interceptDBPath := "./test_rbf_proxy_ctxcancel" + db4, err := leveldb.OpenFile(interceptDBPath, nil) + Expect(err).To(BeNil()) + defer os.RemoveAll(interceptDBPath) + defer db4.Close() + + interceptCache := btc.NewBatcherCache(db4, "ctxcancel_test", btc.RBF) + nodeProxyURL := nodeServer.URL + bitcoinRPC4 := btc.NewBitcoinClient("admin1", "123", nodeProxyURL) + + wallet4, err := btc.NewBatcherWallet( + pk, + interceptIndexer, + NewMockFeeEstimator(1), + chainParams, + interceptCache, + logger, + &bitcoinRPC4, + btc.WithPTI(5*time.Second), + btc.WithStrategy(btc.RBF), + ) + Expect(err).To(BeNil()) + + // Fund + out, cmdErr := exec.Command("merry", "faucet", "--to", wallet4.Address().EncodeAddress()).CombinedOutput() + Expect(cmdErr).To(BeNil(), "merry faucet failed: %s", string(out)) + mineAndWait(proxyIndexer, 1) + + Eventually(func() bool { + utxos, err := interceptIndexer.GetUTXOs(context.Background(), wallet4.Address()) + if err != nil { + return false + } + return len(utxos) > 0 + }, 60*time.Second, 2*time.Second).Should(BeTrue()) + + // Use a cancellable context for the wallet lifecycle + ctx, cancel := context.WithCancel(context.Background()) + err = wallet4.Start(ctx) + Expect(err).To(BeNil()) + defer wallet4.Stop() + + req := []btc.SendRequest{ + {Amount: 10000, To: wallet4.Address()}, + } + + _, err = wallet4.Send(ctx, req, nil, nil) + Expect(err).To(BeNil()) + + // Let the batcher enter the mempool-wait loop (node has the tx, + // indexer permanently 404). After a few iterations, cancel. + time.Sleep(10 * time.Second) + cancel() + + // Give it time to react to cancellation + time.Sleep(5 * time.Second) + + // The batch should NOT be persisted because context was cancelled + // before the mempool-wait cap was reached. + _, err = interceptCache.ReadLatestBatch(context.Background()) + Expect(err).ShouldNot(BeNil()) + + fmt.Println("[DelayProxy] Test6 (context cancel): Batch correctly NOT persisted after cancellation") + }) + }) })