diff --git a/btc/batcher.go b/btc/batcher.go index 37b707a..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 = 5 * time.Second + DefaultAPITimeout = 300 * 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..50752aa 100644 --- a/btc/indexer.go +++ b/btc/indexer.go @@ -298,15 +298,28 @@ func (client *electrsIndexerClient) GetTxHex(ctx context.Context, txid string) ( } func (client *electrsIndexerClient) GetTx(ctx context.Context, txid string) (Transaction, error) { + if _, ok := ctx.Deadline(); !ok { + var cancel context.CancelFunc + ctx, cancel = context.WithTimeout(ctx, 60*time.Second) + defer cancel() + } + endpoint, err := url.JoinPath(client.url, "tx", txid) if err != nil { return Transaction{}, err } - // Send the request var tx Transaction - if err := retry(client.logger, ctx, client.retryInterval, func() error { - resp, err := http.Get(endpoint) + baseBackoff := 2 * time.Second + if err := exponentialBackoffRetry(client.logger, ctx, baseBackoff, func(t time.Duration) error { + reqCtx, reqCancel := context.WithTimeout(ctx, t) + 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 } @@ -315,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)) + } + if resp.StatusCode == http.StatusNotFound || resp.StatusCode == http.StatusBadRequest { + return NewNoRetryError(fmt.Errorf("GetTx: %s", errMsg)) } - return fmt.Errorf("GetTx : %v", string(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 { @@ -433,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) @@ -444,6 +459,36 @@ func retry(logger *zap.Logger, ctx context.Context, dur time.Duration, f func() return nil } +// 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 + } + + var e *NoRetryError + if errors.As(err, &e) { + return err + } + + 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 : %w", ctx.Err(), err) + case <-timer.C: + backoff *= 2 + } + } +} + type NoRetryError struct { err error } diff --git a/btc/indexer_test.go b/btc/indexer_test.go index 341e103..e45bfb3 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,91 @@ 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("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()") diff --git a/btc/rbf.go b/btc/rbf.go index 9d85664..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" @@ -117,8 +118,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)) @@ -263,15 +266,45 @@ 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 - }) - if err != nil { + 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), + LockTime: int(tx.LockTime), + 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), + } } // Create a new batch with the transaction details and save it to the cache @@ -370,8 +403,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)) @@ -787,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 new file mode 100644 index 0000000..53156f0 --- /dev/null +++ b/btc/rbf_proxy_test.go @@ -0,0 +1,681 @@ +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) + }) + }) + + // --------------------------------------------------------------- + // 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") + }) + }) +})