feat(bitcoinspv/client): non-blocking SendBlocks

bitcoinspv/bootstrap.go

@@ -274,6 +274,8 @@ func (r *Relayer) backfillIndexer(ctx context.Context, startHeight, endHeight in
 			}
 		}
 	}
+
+	r.btcIndexer.Flush()
 	r.logger.Info().Msg("Indexer backfill completed successfully.")
 	return nil
 }

bitcoinspv/clients/btcindexer/client.go

@@ -19,61 +19,82 @@
 	maxRetries     = 4
 	initialBackoff = 500 * time.Millisecond
 	maxBackoff     = 8 * time.Second
+	queueSize      = 100
 )
 
 // Client is a client for communicating with the nBTC indexer worker.
-// It wraps the btcindexer API client to add retry logic.
+// It wraps the btcindexer API client to add retry logic and async sending.
 type Client struct {
 	logger    zerolog.Logger
 	apiClient btcindexer.Client
 	network   string
+
+	blocksChan chan []*types.IndexedBlock
+	done       chan struct{}
 }
 
 // NewClient creates a new client for the indexer.
 func NewClient(url string, network string, authToken string, parentLogger zerolog.Logger) *Client {
-	return &Client{
-		logger:    parentLogger.With().Str("module", "btcindexer_client").Logger(),
-		apiClient: btcindexer.NewClient(url, authToken),
-		network:   network,
+	c := &Client{
+		logger:     parentLogger.With().Str("module", "btcindexer_client").Logger(),
+		apiClient:  btcindexer.NewClient(url, authToken),
+		network:    network,
+		blocksChan: make(chan []*types.IndexedBlock, queueSize),
+		done:       make(chan struct{}),
+	}
+	go c.worker()
+	return c
+}
+
+// worker processes blocks from the queue in a background goroutine.
+func (c *Client) worker() {
+	defer close(c.done)
+	for blocks := range c.blocksChan {
+		if err := c.sendBlocksWithRetry(blocks); err != nil {
+			c.logger.Error().Err(err).Msg("Failed to send blocks to indexer")
+		}
 	}
 }
 
-// SendBlocks sends a batch of blocks to the indexer with a retry mechanism.
-// TODO: this should not block the main process
-// probably we should use CF queues
-func (c *Client) SendBlocks(ctx context.Context, blocks []*types.IndexedBlock) error {
+// SendBlocks enqueues blocks for async sending to the indexer.
+// It returns an error only if the queue is full.
+func (c *Client) SendBlocks(_ context.Context, blocks []*types.IndexedBlock) error {
 	if c == nil {
 		return errors.New("btcindexer.Client is not initialized")
 	}
 	if len(blocks) == 0 {
 		return nil
 	}
 
+	select {
+	case c.blocksChan <- blocks:
+		return nil
+	default:
+		return errors.New("indexer queue is full, dropping blocks")
+	}
+}
+
+// sendBlocksWithRetry sends a batch of blocks to the indexer with a retry mechanism.
+func (c *Client) sendBlocksWithRetry(blocks []*types.IndexedBlock) error {
 	payload, err := c.preparePayload(blocks)
 	if err != nil {
 		return err
 	}
 
 	var lastErr error
 	for attempt := 0; attempt <= maxRetries; attempt++ {
-		if ctx.Err() != nil {
-			return ctx.Err()
-		}
-
 		shouldRetry, err := c.sendAndHandleResponse(payload)
 		if err != nil {
-			// Non-retryable
 			return err
 		}
 
 		if !shouldRetry {
-			// Success
 			return nil
 		}
 
 		lastErr = fmt.Errorf("attempt %d failed, retrying", attempt+1)
 		c.logger.Warn().Err(err).Msg("Retrying indexer call...")
-		c.backoff(ctx, attempt)
+		c.backoff(attempt)
 	}
 
 	return fmt.Errorf("failed to send blocks to indexer after %d attempts: %w", maxRetries+1, lastErr)
@@ -123,24 +144,19 @@
 	return putBlocksReq, nil
 }
 
-func (c *Client) backoff(ctx context.Context, attempt int) {
+func (c *Client) backoff(attempt int) {
 	if attempt >= maxRetries {
 		return
 	}
 	backoff := time.Duration(1<<attempt) * initialBackoff
 	if backoff > maxBackoff {
 		backoff = maxBackoff
 	}
-	// NOTE: we dont need secure random generation here, its just retry
 	jitter := time.Duration(rand.Intn(1000)) * time.Millisecond //nolint:gosec
 	totalBackoff := backoff + jitter
 
 	c.logger.Info().Dur("wait_duration", totalBackoff).Msgf("Waiting before next attempt..")
-
-	select {
-	case <-time.After(totalBackoff):
-	case <-ctx.Done():
-	}
+	time.Sleep(totalBackoff)
 }
 
 // GetLatestHeight returns the latest block height known to the indexer
@@ -153,3 +169,16 @@
 
 	return height, nil
 }
+
+// Close stops the background worker and waits for it to finish.
+func (c *Client) Close() {
+	close(c.blocksChan)
+	<-c.done
+}
+
+// Flush waits for all queued blocks to be sent.
+func (c *Client) Flush() {
+	for len(c.blocksChan) > 0 {
+		time.Sleep(100 * time.Millisecond)
+	}
+}

bitcoinspv/clients/btcindexer/interface.go

@@ -10,4 +10,6 @@ import (
 type Indexer interface {
 	SendBlocks(ctx context.Context, blocks []*types.IndexedBlock) error
 	GetLatestHeight() (int64, error)
+	Flush()
+	Close()
 }

bitcoinspv/relayer.go

@@ -144,6 +144,9 @@ func (r *Relayer) Stop() {
 // WaitForShutdown waits for all relayer goroutines to complete before returning
 func (r *Relayer) WaitForShutdown() {
 	r.wg.Wait()
+	if r.btcIndexer != nil {
+		r.btcIndexer.Close()
+	}
 }
 
 // UploadToWalrus upload full BTC block to Walrus