Merge branch 'main' into feature/fixCheckOldBlockIds

Author: oskarszoon

compose/docker-compose-3blasters.yml

@@ -147,6 +147,15 @@ services:
       JAEGER_AGENT_HOST: localhost
       JAEGER_AGENT_PORT: 6831
       logLevel: "DEBUG"
+      KAFKA_BLOCKS: blocks-teranode1
+      KAFKA_BLOCKS_FINAL: blocks-final-teranode1
+      KAFKA_INVALID_BLOCKS: invalid-blocks-teranode1
+      KAFKA_INVALID_SUBTREES: invalid-subtrees-teranode1
+      KAFKA_LEGACY_INV: legacy-inv-teranode1
+      KAFKA_REJECTEDTX: rejectedtx-teranode1
+      KAFKA_SUBTREES: subtrees-teranode1
+      KAFKA_TXMETA: txmeta-teranode1
+      KAFKA_VALIDATORTXS: validatortxs-teranode1
     volumes:
       - ../settings.conf:/app/settings.conf
       - ../settings_local.conf:/app/settings_local.conf
@@ -209,6 +218,15 @@ services:
       JAEGER_AGENT_HOST: localhost
       JAEGER_AGENT_PORT: 6831
       logLevel: "DEBUG"
+      KAFKA_BLOCKS: blocks-teranode2
+      KAFKA_BLOCKS_FINAL: blocks-final-teranode2
+      KAFKA_INVALID_BLOCKS: invalid-blocks-teranode2
+      KAFKA_INVALID_SUBTREES: invalid-subtrees-teranode2
+      KAFKA_LEGACY_INV: legacy-inv-teranode2
+      KAFKA_REJECTEDTX: rejectedtx-teranode2
+      KAFKA_SUBTREES: subtrees-teranode2
+      KAFKA_TXMETA: txmeta-teranode2
+      KAFKA_VALIDATORTXS: validatortxs-teranode2
     volumes:
       - ../settings.conf:/app/settings.conf
       - ../settings_local.conf:/app/settings_local.conf
@@ -271,6 +289,15 @@ services:
       JAEGER_AGENT_HOST: localhost
       JAEGER_AGENT_PORT: 6831
       logLevel: "DEBUG"
+      KAFKA_BLOCKS: blocks-teranode3
+      KAFKA_BLOCKS_FINAL: blocks-final-teranode3
+      KAFKA_INVALID_BLOCKS: invalid-blocks-teranode3
+      KAFKA_INVALID_SUBTREES: invalid-subtrees-teranode3
+      KAFKA_LEGACY_INV: legacy-inv-teranode3
+      KAFKA_REJECTEDTX: rejectedtx-teranode3
+      KAFKA_SUBTREES: subtrees-teranode3
+      KAFKA_TXMETA: txmeta-teranode3
+      KAFKA_VALIDATORTXS: validatortxs-teranode3
     volumes:
       - ../settings.conf:/app/settings.conf
       - ../settings_local.conf:/app/settings_local.conf
@@ -378,6 +405,8 @@ services:
     restart: no
 
   coinbase2:
+    profiles:
+      - skip
     <<: *teranode-coinbase-base
     container_name: coinbase2
     depends_on:
@@ -430,6 +459,8 @@ services:
 
   coinbase3:
     <<: *teranode-coinbase-base
+    profiles:
+      - skip
     container_name: coinbase3
     depends_on:
       - teranode3
@@ -488,6 +519,8 @@ services:
 
   tx-blaster-2:
     <<: *teranode-coinbase-base
+    profiles:
+      - skip
     container_name: tx-blaster-2
     depends_on:
       - coinbase2
@@ -513,6 +546,8 @@ services:
       - ../data/txblaster2:/app/data
 
   tx-blaster-3:
+    profiles:
+      - skip
     <<: *teranode-coinbase-base
     container_name: tx-blaster-3
     depends_on:

settings.conf

@@ -624,6 +624,11 @@ http_sign_response = true
 
 http_timeout = 30000
 
+# HTTP streaming timeout in milliseconds for large file downloads (subtree data, blocks during catchup)
+# This is longer than http_timeout to accommodate large subtree data files which can be 100+ MB
+# Default: 300000ms (5 minutes)
+http_streaming_timeout = 300000
+
 # IPV6 Addresses
 # --------------
 # ipv6_addresses = ff02::1234

util/http.go

@@ -5,6 +5,7 @@ import (
 	"context"
 	"io"
 	"net/http"
+	"strings"
 	"time"
 
 	"github.com/bsv-blockchain/teranode/errors"
@@ -14,7 +15,12 @@ import (
 var (
 	// httpRequestTimeout defines the default HTTP request timeout in milliseconds
 	// when no deadline is set on the context.
-	httpRequestTimeout, _ = gocore.Config().GetInt("http_timeout", 60)
+	httpRequestTimeout, _ = gocore.Config().GetInt("http_timeout", 60000)
+
+	// httpStreamingTimeout defines the default HTTP streaming timeout in milliseconds
+	// for operations that stream large responses. This is longer than httpRequestTimeout
+	// to accommodate large block/subtree downloads during catchup.
+	httpStreamingTimeout, _ = gocore.Config().GetInt("http_streaming_timeout", 300000) // 5 minutes default
 )
 
 // DoHTTPRequest performs an HTTP GET or POST request and returns the response body as bytes.
@@ -57,17 +63,34 @@ func DoHTTPRequest(ctx context.Context, url string, requestBody ...[]byte) ([]by
 	}
 }
 
+// readCloserWithCancel wraps an io.ReadCloser and calls a cancel function when closed.
+type readCloserWithCancel struct {
+	io.ReadCloser
+	cancelFn context.CancelFunc
+}
+
+func (r *readCloserWithCancel) Close() error {
+	defer r.cancelFn()
+	return r.ReadCloser.Close()
+}
+
 // DoHTTPRequestBodyReader performs an HTTP request and returns the response body as a ReadCloser.
 // This is more memory-efficient for large responses as it streams the data.
 // Caller is responsible for closing the returned ReadCloser.
+// Applies a default timeout of 5 minutes (configurable via http_streaming_timeout) when no
+// deadline is set on the context. This timeout is longer than the standard HTTP timeout
+// to accommodate large file downloads during operations like P2P catchup.
 func DoHTTPRequestBodyReader(ctx context.Context, url string, requestBody ...[]byte) (io.ReadCloser, error) {
-	bodyReaderCloser, cancelFn, err := doHTTPRequest(ctx, url, requestBody...)
+	bodyReaderCloser, cancelFn, err := doHTTPRequestForStreaming(ctx, url, requestBody...)
 	if err != nil {
 		cancelFn()
 		return nil, err
 	}
 
-	return bodyReaderCloser, nil
+	return &readCloserWithCancel{
+		ReadCloser: bodyReaderCloser,
+		cancelFn:   cancelFn,
+	}, nil
 }
 
 func doHTTPRequest(ctx context.Context, url string, requestBody ...[]byte) (io.ReadCloser, context.CancelFunc, error) {
@@ -79,6 +102,25 @@ func doHTTPRequest(ctx context.Context, url string, requestBody ...[]byte) (io.R
 		ctx, cancelFn = context.WithTimeout(ctx, time.Duration(httpRequestTimeout)*time.Millisecond)
 	}
 
+	return executeHTTPRequest(ctx, cancelFn, url, requestBody...)
+}
+
+// doHTTPRequestForStreaming performs an HTTP request with a longer timeout suitable for streaming.
+// Applies httpStreamingTimeout (default 5 minutes) when no deadline exists on the context.
+func doHTTPRequestForStreaming(ctx context.Context, url string, requestBody ...[]byte) (io.ReadCloser, context.CancelFunc, error) {
+	cancelFn := func() {
+		// noop
+	}
+
+	if _, ok := ctx.Deadline(); !ok {
+		ctx, cancelFn = context.WithTimeout(ctx, time.Duration(httpStreamingTimeout)*time.Millisecond)
+	}
+
+	return executeHTTPRequest(ctx, cancelFn, url, requestBody...)
+}
+
+// executeHTTPRequest performs the actual HTTP request with the given context.
+func executeHTTPRequest(ctx context.Context, cancelFn context.CancelFunc, url string, requestBody ...[]byte) (io.ReadCloser, context.CancelFunc, error) {
 	httpClient := http.DefaultClient
 
 	req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
@@ -125,7 +167,8 @@ func doHTTPRequest(ctx context.Context, url string, requestBody ...[]byte) (io.R
 		return nil, cancelFn, errFn("http request [%s] returned status code [%d]", url, resp.StatusCode)
 	}
 
-	isHTML := resp.Header.Get("content-type") == "text/html"
+	ct := strings.ToLower(resp.Header.Get("content-type"))
+	isHTML := strings.HasPrefix(ct, "text/html")
 	if isHTML {
 		return nil, cancelFn, errors.NewServiceError("http request [%s] returned HTML - assume bad URL", url)
 	}

util/http_test.go

@@ -281,6 +281,110 @@ func TestDoHTTPRequestBodyReaderServerError(t *testing.T) {
 	assert.Contains(t, err.Error(), "500")
 }
 
+func TestDoHTTPRequestBodyReaderNoTimeoutOnSlowStream(t *testing.T) {
+	// This test verifies that DoHTTPRequestBodyReader successfully completes
+	// for quick responses, proving the streaming timeout is long enough
+	responseData := []byte("streaming response data")
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Content-Type", "application/octet-stream")
+		w.WriteHeader(http.StatusOK)
+		_, err := w.Write(responseData)
+		require.NoError(t, err)
+	}))
+	defer server.Close()
+
+	// Context without timeout - function should apply streaming timeout (5 min default)
+	ctx := context.Background()
+	reader, err := DoHTTPRequestBodyReader(ctx, server.URL)
+	require.NoError(t, err)
+	require.NotNil(t, reader)
+	defer func(reader io.ReadCloser) {
+		_ = reader.Close()
+	}(reader)
+
+	body, err := io.ReadAll(reader)
+	require.NoError(t, err)
+	assert.Equal(t, responseData, body)
+}
+
+func TestDoHTTPRequestBodyReaderWithShortTimeout(t *testing.T) {
+	// This test uses a custom short streaming timeout to actually verify timeout behavior
+	// Save and restore the original timeout
+	originalTimeout := httpStreamingTimeout
+	httpStreamingTimeout = 500 // 500ms for testing - enough for connection but not for slow read
+	defer func() {
+		httpStreamingTimeout = originalTimeout
+	}()
+
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Content-Type", "application/octet-stream")
+		w.WriteHeader(http.StatusOK)
+		// Write some data immediately
+		flusher, ok := w.(http.Flusher)
+		require.True(t, ok, "Expected http.ResponseWriter to be an http.Flusher")
+		_, _ = w.Write([]byte("starting..."))
+		flusher.Flush()
+
+		// Sleep longer than timeout - this should cause the read to fail
+		time.Sleep(1 * time.Second)
+		_, _ = w.Write([]byte("this should not be received"))
+	}))
+	defer server.Close()
+
+	ctx := context.Background()
+	reader, err := DoHTTPRequestBodyReader(ctx, server.URL)
+	require.NoError(t, err)
+	require.NotNil(t, reader)
+	defer func(reader io.ReadCloser) {
+		_ = reader.Close()
+	}(reader)
+
+	// Reading should fail due to timeout while reading the body
+	_, err = io.ReadAll(reader)
+	// Should get an error because the server is too slow
+	require.Error(t, err, "Should get timeout error on slow stream")
+}
+
+func TestDoHTTPRequestBodyReaderRespectsExistingDeadline(t *testing.T) {
+	// This test verifies that DoHTTPRequestBodyReader respects an existing context deadline
+	// We do this by setting a very short deadline and ensuring it times out quickly
+	// (rather than waiting for the 5-minute streaming timeout)
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Content-Type", "application/octet-stream")
+		w.WriteHeader(http.StatusOK)
+		// Write initial data
+		_, _ = w.Write([]byte("start"))
+		if flusher, ok := w.(http.Flusher); ok {
+			flusher.Flush()
+		}
+		// Sleep for 2 seconds - longer than our 100ms deadline
+		time.Sleep(2 * time.Second)
+		_, _ = w.Write([]byte("should timeout before this"))
+	}))
+	defer server.Close()
+
+	// Context with very short 100ms timeout (much shorter than 5-minute streaming default)
+	ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
+	defer cancel()
+
+	start := time.Now()
+	reader, err := DoHTTPRequestBodyReader(ctx, server.URL)
+	require.NoError(t, err)
+	require.NotNil(t, reader)
+	defer func(reader io.ReadCloser) {
+		_ = reader.Close()
+	}(reader)
+
+	// This should timeout due to our 100ms deadline, not the 5-minute streaming timeout
+	_, err = io.ReadAll(reader)
+	elapsed := time.Since(start)
+
+	// Should fail with timeout
+	require.Error(t, err, "Should timeout with custom deadline")
+	// Should have timed out quickly (within 1 second), not after 5 minutes
+	assert.Less(t, elapsed, 2*time.Second, "Should respect short custom deadline, not wait for streaming timeout")
+}
+
 func TestDoHTTPRequestEmptyRequestBody(t *testing.T) {
 	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		// Empty slice still triggers POST because len(requestBody) > 0

util/kafka/kafka_producer_async.go

@@ -21,8 +21,16 @@ import (
 	"github.com/bsv-blockchain/teranode/util"
 	inmemorykafka "github.com/bsv-blockchain/teranode/util/kafka/in_memory_kafka"
 	"github.com/bsv-blockchain/teranode/util/retry"
+	"github.com/rcrowley/go-metrics"
 )
 
+// init disables go-metrics globally to prevent memory leak from exponential decay sample heap.
+// This must be set before any Sarama clients are created.
+// See: https://github.com/IBM/sarama/issues/1321
+func init() {
+	metrics.UseNilMetrics = true
+}
+
 // KafkaAsyncProducerI defines the interface for asynchronous Kafka producer operations.
 type KafkaAsyncProducerI interface {
 	// Start begins the async producer operation with the given message channel