[RelayMiner] Use a customized HTTP client in RelayMiner (#1764)

## Summary

Use a customized HTTP client in RelayMiner

## Issue

The Default HTTP client, used by Relay Miner, is not optimized for high
throughput.

- Issue_or_PR: #{ISSUE_OR_PR_NUMBER}

## Type of change

Select one or more from the following:

- [x] New feature, functionality or library
- [ ] Bug fix
- [ ] Code health or cleanup
- [ ] Documentation
- [ ] Other (specify)

## Sanity Checklist

- [x] I have updated the GitHub Issue Metadata: `assignees`,
`reviewers`, `labels`, `project`, `iteration` and `milestone`
- [ ] For docs: `make docusaurus_start`
- [ ] For small changes: `make go_develop_and_test` and `make test_e2e`
- [ ] For major changes: `devnet-test-e2e` label to run E2E tests in CI
- [ ] For migration changes: `make test_e2e_oneshot`
- [ ] 'TODO's, configurations and other docs

---------

Co-authored-by: Redouane Lakrache <[email protected]>
Co-authored-by: Daniel Olshansky <[email protected]>

Author: 3 people

pkg/network/concurrency/buffer_pool.go

@@ -0,0 +1,90 @@
+// Buffer Pool for High-Concurrency HTTP Processing
+// ================================================
+//
+// This buffer pool manages reusable byte buffers to optimize memory allocation
+// for high-throughput HTTP response processing. When handling thousands of
+// concurrent HTTP requests with large response bodies (blockchain data often
+// exceeds 1MB), naive allocation patterns create significant performance issues.
+//
+// Memory Allocation Patterns:
+//   - Without pooling: Each request allocates new []byte buffers
+//   - With pooling: Buffers are reused across requests via sync.Pool
+//
+// Benefits:
+//   - Reduces garbage collection pressure
+//   - Provides predictable memory usage under load
+//   - Maintains consistent performance during traffic spikes
+//   - Size limits prevent memory bloat
+//
+// The pool automatically grows buffer capacity as needed while preventing
+// oversized buffers from being returned to avoid memory waste.
+package concurrency
+
+import (
+	"bytes"
+	"io"
+	"sync"
+)
+
+const (
+	// DefaultInitialBufferSize is the initial size of the buffer pool.
+	// Start with 256KB buffers - can grow as needed
+	DefaultInitialBufferSize = 256 * 1024
+
+	// TODO_IMPROVE: Make this configurable via YAML settings
+	// DefaultMaxBufferSize is the maximum size of the buffer pool.
+	// Set the max buffer size to 4MB to avoid memory bloat.
+	DefaultMaxBufferSize = 4 * 1024 * 1024
+)
+
+// BufferPool manages reusable byte buffers to reduce GC pressure.
+// Uses sync.Pool for efficient buffer recycling with size limits.
+type BufferPool struct {
+	pool          sync.Pool
+	maxReaderSize int64
+}
+
+func NewBufferPool(maxReaderSize int64) *BufferPool {
+	return &BufferPool{
+		pool: sync.Pool{
+			New: func() interface{} {
+				return bytes.NewBuffer(make([]byte, 0, DefaultInitialBufferSize))
+			},
+		},
+		maxReaderSize: maxReaderSize,
+	}
+}
+
+// getBuffer retrieves a buffer from the pool.
+func (bp *BufferPool) getBuffer() *bytes.Buffer {
+	buf := bp.pool.Get().(*bytes.Buffer)
+	buf.Reset() // Always reset to ensure clean state
+	return buf
+}
+
+// putBuffer returns a buffer to the pool.
+// Buffers larger than maxBufferSize are not returned to avoid memory bloat.
+func (bp *BufferPool) putBuffer(buf *bytes.Buffer) {
+	// Skip pooling oversized buffers to prevent memory bloat
+	if buf.Cap() > DefaultMaxBufferSize {
+		return
+	}
+	bp.pool.Put(buf)
+}
+
+// ReadWithBuffer reads from an io.Reader using a pooled buffer.
+func (bp *BufferPool) ReadWithBuffer(r io.Reader) ([]byte, error) {
+	buf := bp.getBuffer()
+	defer bp.putBuffer(buf)
+
+	limitedReader := io.LimitReader(r, bp.maxReaderSize)
+	_, err := buf.ReadFrom(limitedReader)
+	if err != nil {
+		return nil, err
+	}
+
+	// Return independent copy to avoid data races
+	result := make([]byte, buf.Len())
+	copy(result, buf.Bytes())
+	return result, nil
+}

pkg/network/concurrency/concurrency_limiter.go

@@ -0,0 +1,83 @@
+// Concurrency Limiter for Resource Management
+// ===========================================
+//
+// This concurrency limiter implements a semaphore pattern to bound the number
+// of concurrent HTTP operations, preventing resource exhaustion under high load.
+//
+// When processing thousands of simultaneous HTTP requests, unlimited concurrency
+// can overwhelm system resources (memory, file descriptors, network connections).
+//
+// Resource Protection Mechanisms:
+//   - Semaphore-based admission control using buffered channels
+//   - Context-aware blocking with cancellation support
+//   - Real-time tracking of active request counts
+//   - Graceful degradation when limits are exceeded
+//
+// Operational Characteristics:
+//   - Blocks new requests when limit is reached
+//   - Respects context cancellation for timeout handling
+//   - Integrates with metrics for observability
+//   - Thread-safe for concurrent access
+//
+// The limiter prevents cascading failures by ensuring system resources remain
+// available even during traffic spikes or slow downstream services.
+package concurrency
+
+import (
+	"context"
+	"sync"
+)
+
+// TODO_IMPROVE: Make this configurable via settings
+const (
+	defaultMaxConcurrentRequests = 1_000_000
+)
+
+// ConcurrencyLimiter bounds concurrent operations via semaphore pattern.
+// Prevents resource exhaustion and tracks active request counts.
+type ConcurrencyLimiter struct {
+	semaphore      chan struct{}
+	maxConcurrent  int
+	activeRequests int64
+	mu             sync.RWMutex
+}
+
+// NewConcurrencyLimiter creates a limiter that bounds concurrent operations.
+func NewConcurrencyLimiter(maxConcurrent int) *ConcurrencyLimiter {
+	if maxConcurrent <= 0 {
+		maxConcurrent = defaultMaxConcurrentRequests // Default reasonable limit
+	}
+
+	return &ConcurrencyLimiter{
+		semaphore:     make(chan struct{}, maxConcurrent),
+		maxConcurrent: maxConcurrent,
+	}
+}
+
+// TODO_TECHDEBT(@adshmh): Track active relays for observability
+//
+// Acquire blocks until a slot is available or context is canceled.
+// Returns true if acquired, false if context was canceled.
+func (cl *ConcurrencyLimiter) Acquire(ctx context.Context) bool {
+	select {
+	case cl.semaphore <- struct{}{}:
+		cl.mu.Lock()
+		cl.activeRequests++
+		cl.mu.Unlock()
+		return true
+	case <-ctx.Done():
+		return false
+	}
+}
+
+// Release returns a slot to the pool.
+func (cl *ConcurrencyLimiter) Release() {
+	select {
+	case <-cl.semaphore:
+		cl.mu.Lock()
+		cl.activeRequests--
+		cl.mu.Unlock()
+	default:
+		// TODO_TECHDEBT: Log acquire/release mismatch for debugging
+	}
+}