Share one forkable hub between firehose and tier1

CHANGELOG.md

@@ -36,6 +36,10 @@ If you were at `firehose-core` version `1.0.0` and are bumping to `1.1.0`, you s
 
 ### Changed
 
+- `firehose` and `substreams-tier1` launched in the same process now share one forkable hub instead of each holding its own copy of the live blocks; on Solana that saves about 0.5 GiB per process. Firehose still streams complete blocks only. With `--firehose-discard-partial-blocks`, the two apps keep separate hubs, since tier1 needs the partial blocks.
+
+- The firehose hub keeps as many final blocks as the substreams-tier1 hub: two merged-blocks files worth, at least 200, instead of at least 500. Clients resuming between 200 and 500 blocks below the last irreversible block now read those blocks from merged-blocks files.
+
 - Apps running in the same process now share one session pool, created from `--common-session-plugin`, instead of each creating their own. With the `local://` plugin, firehose and substreams-tier1 running together now count against the same `max_sessions` and `max_sessions_per_organization` limits.
 
 - Bump `firehose-networks` to `v0.3.0`.

cmd/apps/firehose.go

@@ -99,6 +99,11 @@ func RegisterFirehoseApp[B firecore.Block](chain *firecore.Chain[B], rootLog *za
 				serverOptions = append(serverOptions, server.WithLeakyBucketLimiter(limiterSize, limiterRefillRate))
 			}
 
+			forkableHub, err := sharedLiveHub(runtime)
+			if err != nil {
+				return nil, fmt.Errorf("unable to set up the shared forkable hub: %w", err)
+			}
+
 			return firehose.New(appLogger, appTracer, &firehose.Config{
 				MergedBlocksStoreURL:    mergedBlocksStoreURL,
 				OneBlocksStoreURL:       oneBlocksStoreURL,
@@ -118,6 +123,7 @@ func RegisterFirehoseApp[B firecore.Block](chain *firecore.Chain[B], rootLog *za
 				TransformRegistry:          registry,
 				CheckPendingShutdown:       runtime.IsPendingShutdown,
 				InfoServer:                 runtime.InfoServer,
+				ForkableHub:                forkableHub,
 			}), nil
 		},
 	})

cmd/apps/shared_hub.go

@@ -0,0 +1,81 @@
+package apps
+
+import (
+	"fmt"
+	"slices"
+	"sync"
+
+	"github.com/spf13/viper"
+	"github.com/streamingfast/bstream"
+	"github.com/streamingfast/bstream/hub"
+	pbbstream "github.com/streamingfast/bstream/pb/sf/bstream/v1"
+	"github.com/streamingfast/dstore"
+	firecore "github.com/streamingfast/firehose-core"
+	"github.com/streamingfast/firehose-core/launcher"
+	"github.com/streamingfast/logging"
+	ssapp "github.com/streamingfast/substreams/app"
+)
+
+var liveHubLogger, _ = logging.PackageLogger("live-hub", "github.com/streamingfast/firehose-core/live-hub")
+
+var (
+	sharedLiveHubOnce sync.Once
+	sharedLiveHubHub  *hub.ForkableHub
+	sharedLiveHubErr  error
+)
+
+// sharedLiveHub returns the forkable hub the firehose and substreams-tier1 apps
+// share when both are launched in the process, so that live blocks are held in
+// memory once. It is built and started on the first call.
+//
+// It returns nil, each app then building its own hub, when one of the two apps
+// is not launched, when there is no live source, or when firehose drops partial
+// blocks (--firehose-discard-partial-blocks), which tier1 needs.
+func sharedLiveHub(runtime *launcher.Runtime) (*hub.ForkableHub, error) {
+	if !slices.Contains(runtime.Apps, "firehose") || !slices.Contains(runtime.Apps, "substreams-tier1") {
+		return nil, nil
+	}
+
+	blockStreamAddr := viper.GetString("common-live-blocks-addr")
+	if blockStreamAddr == "" {
+		return nil, nil
+	}
+
+	if viper.GetBool("firehose-discard-partial-blocks") {
+		liveHubLogger.Info("firehose and substreams-tier1 keep separate hubs, firehose discards partial blocks and tier1 needs them")
+		return nil, nil
+	}
+
+	sharedLiveHubOnce.Do(func() {
+		_, oneBlocksStoreURL, _, err := firecore.GetCommonStoresURLs(runtime.AbsDataDir)
+		if err != nil {
+			sharedLiveHubErr = err
+			return
+		}
+
+		oneBlocksStore, err := dstore.NewDBinStore(oneBlocksStoreURL)
+		if err != nil {
+			sharedLiveHubErr = fmt.Errorf("failed setting up one-block store from url %q: %w", oneBlocksStoreURL, err)
+			return
+		}
+
+		liveHubLogger.Info("firehose and substreams-tier1 share one forkable hub")
+		sharedLiveHubHub = ssapp.NewLiveHub(ssapp.LiveHubConfig{
+			BlockStreamAddr:        blockStreamAddr,
+			MergedBlocksBundleSize: bstream.DefaultMergedBlocksBundleSize,
+			OneBlocksStore:         oneBlocksStore,
+			Requester:              "firehose+substreams-tier1",
+			Logger:                 liveHubLogger,
+			OnBlock: func(blk *pbbstream.Block) {
+				headBlockNumMetric.SetUint64(blk.Number)
+				headTimeDriftmetric.SetBlockTime(blk.Time())
+				finalizedBlockNumMetric.SetUint64(blk.LibNum)
+				ss1HeadBlockNumMetric.SetUint64(blk.Number)
+				ss1HeadTimeDriftmetric.SetBlockTime(blk.Time())
+			},
+		})
+		go sharedLiveHubHub.Run()
+	})
+
+	return sharedLiveHubHub, sharedLiveHubErr
+}

cmd/apps/shared_hub_test.go

@@ -0,0 +1,39 @@
+package apps
+
+import (
+	"testing"
+
+	"github.com/spf13/viper"
+	"github.com/streamingfast/firehose-core/launcher"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestSharedLiveHubNotShared(t *testing.T) {
+	tests := []struct {
+		name                 string
+		apps                 []string
+		blockStreamAddr      string
+		discardPartialBlocks bool
+	}{
+		{name: "firehose alone", apps: []string{"firehose", "merger"}, blockStreamAddr: ":10014"},
+		{name: "tier1 alone", apps: []string{"substreams-tier1", "substreams-tier2"}, blockStreamAddr: ":10014"},
+		{name: "no live source", apps: []string{"firehose", "substreams-tier1"}},
+		{name: "firehose discards partial blocks", apps: []string{"firehose", "substreams-tier1"}, blockStreamAddr: ":10014", discardPartialBlocks: true},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			viper.Set("common-live-blocks-addr", tt.blockStreamAddr)
+			viper.Set("firehose-discard-partial-blocks", tt.discardPartialBlocks)
+			t.Cleanup(func() {
+				viper.Set("common-live-blocks-addr", nil)
+				viper.Set("firehose-discard-partial-blocks", nil)
+			})
+
+			h, err := sharedLiveHub(&launcher.Runtime{Apps: tt.apps})
+			require.NoError(t, err)
+			assert.Nil(t, h)
+		})
+	}
+}

cmd/apps/substreams_tier1.go

@@ -263,6 +263,11 @@ func RegisterSubstreamsTier1App[B firecore.Block](chain *firecore.Chain[B], root
 				NominalCapacity:    viper.GetFloat64("substreams-tier1-cpu-eviction-nominal-capacity"),
 			}
 
+			forkableHub, err := sharedLiveHub(runtime)
+			if err != nil {
+				return nil, fmt.Errorf("unable to set up the shared forkable hub: %w", err)
+			}
+
 			return app.NewTier1(appLogger,
 				config, &app.Tier1Modules{
 					Authenticator:         authenticator,
@@ -271,6 +276,7 @@ func RegisterSubstreamsTier1App[B firecore.Block](chain *firecore.Chain[B], root
 					CheckPendingShutDown:  runtime.IsPendingShutdown,
 					InfoServer:            runtime.InfoServer,
 					SessionPool:           runtime.SessionPool,
+					ForkableHub:           forkableHub,
 				}), nil
 		},
 	})

firehose/app/firehose/app.go

@@ -39,6 +39,8 @@ import (
 	"github.com/streamingfast/shutter"
 	"go.uber.org/atomic"
 	"go.uber.org/zap"
+
+	ssapp "github.com/streamingfast/substreams/app"
 )
 
 // Unlinkable live blocks in a row that mean the hub is wedged for good. Resets on
@@ -69,6 +71,11 @@ type Modules struct {
 
 	// Optional dependencies
 	FinalizedBlockNumberMetric *coremetrics.FinalizedBlockNum
+
+	// ForkableHub is a hub that another part of the process owns, runs and feeds
+	// metrics from, shared with substreams tier1. When nil, firehose builds and
+	// runs its own.
+	ForkableHub *hub.ForkableHub
 }
 
 type App struct {
@@ -119,11 +126,15 @@ func (a *App) Run() error {
 		}
 	}
 
-	withLive := a.config.BlockStreamAddr != ""
+	withLive := a.config.BlockStreamAddr != "" || a.modules.ForkableHub != nil
 
 	var forkableHub *hub.ForkableHub
 
-	if withLive {
+	if a.modules.ForkableHub != nil {
+		a.logger.Info("using the forkable hub shared by the process")
+		forkableHub = a.modules.ForkableHub
+		forkableHub.OnTerminated(a.Shutdown)
+	} else if withLive {
 		discardPartialBlocks := a.config.DiscardPartialBlocks
 		if discardPartialBlocks {
 			a.logger.Info("partial (flash) blocks will be discarded from the live source before reaching the forkable hub")
@@ -150,12 +161,9 @@ func (a *App) Run() error {
 			)
 		})
 
-		// the hub must hold at least two merged-blocks files worth of final
-		// blocks so the joining source can hand off from a file boundary
-		keepFinalBlocks := int(max(500, 2*bstream.DefaultMergedBlocksBundleSize))
 		forkableHub = hub.NewForkableHubWithOptions(
 			liveSourceFactory,
-			keepFinalBlocks,
+			ssapp.HubKeepFinalBlocks(bstream.DefaultMergedBlocksBundleSize),
 			oneBlocksStore,
 			[]hub.Option{
 				hub.WithLogger(a.logger),

go.mod

@@ -24,7 +24,7 @@ require (
 	github.com/spf13/cobra v1.10.2
 	github.com/spf13/pflag v1.0.10
 	github.com/spf13/viper v1.21.0
-	github.com/streamingfast/bstream v0.0.2-0.20261002142515-2d126b0b0009
+	github.com/streamingfast/bstream v0.0.2-0.20261008022448-30043dab66ef
 	github.com/streamingfast/cli v0.0.4-0.20250815192146-d8a233ec3d0b
 	github.com/streamingfast/dauth v0.0.0-20260318230957-4ab1e1d2ebc3
 	github.com/streamingfast/dbin v0.9.1-0.20231117225723-59790c798e2c
@@ -41,7 +41,7 @@ require (
 	github.com/streamingfast/payment-gateway v0.0.0-20260928153037-ec91d4feb67c
 	github.com/streamingfast/pbgo v0.0.6-0.20260206150405-2b95acf70437
 	github.com/streamingfast/snapshotter v0.0.0-20230316190750-5bcadfde44d0
-	github.com/streamingfast/substreams v1.24.1-0.20261002193310-24d98f5409d4
+	github.com/streamingfast/substreams v1.24.1-0.20261008022204-2409cca4e243
 	github.com/stretchr/testify v1.12.1
 	github.com/test-go/testify v1.1.4
 	github.com/testcontainers/testcontainers-go v0.44.0

go.sum

@@ -2292,6 +2292,8 @@ github.com/spiffe/go-spiffe/v2 v2.8.1/go.mod h1:47Q0Q9/AqGha8QLHp+kxpH4Wca7X7EnO
 github.com/stoewer/go-strcase v1.3.0/go.mod h1:fAH5hQ5pehh+j3nZfvwdk2RgEgQjAoM8wodgtPmh1xo=
 github.com/streamingfast/bstream v0.0.2-0.20261002142515-2d126b0b0009 h1:OD2RxdPwo4qtB6wtIBGp4DsqcQN1RDlQyZRMA0jnqo0=
 github.com/streamingfast/bstream v0.0.2-0.20261002142515-2d126b0b0009/go.mod h1:Asuul2Rnxln0Vk6eY869AZ2UybIQcB7eYOQAIhfQHzE=
+github.com/streamingfast/bstream v0.0.2-0.20261008022448-30043dab66ef h1:iaMJlAik2f1iEI22I2J31vpzOFDnHX+TVQutgVwKbFA=
+github.com/streamingfast/bstream v0.0.2-0.20261008022448-30043dab66ef/go.mod h1:duqjBtD6U7/Y2swB/VCwgzoLODC5L+x6fDmvlgZIAPM=
 github.com/streamingfast/cli v0.0.4-0.20250815192146-d8a233ec3d0b h1:ztYeX3/5rg2tV2EU7edcrcHzMz6wUbdJB+LqCrP5W8s=
 github.com/streamingfast/cli v0.0.4-0.20250815192146-d8a233ec3d0b/go.mod h1:o9R/tjNON01X2mgWL5qirl2MV6xQ4EZI5D504ST3K/M=
 github.com/streamingfast/dauth v0.0.0-20260318230957-4ab1e1d2ebc3 h1:Zo2daSGDUPoK32w9G1e9fZO3Q6b5Cvu5ZojdoD/Wgp8=
@@ -2351,6 +2353,8 @@ github.com/streamingfast/snapshotter v0.0.0-20230316190750-5bcadfde44d0 h1:Y15G1
 github.com/streamingfast/snapshotter v0.0.0-20230316190750-5bcadfde44d0/go.mod h1:/Rnz2TJvaShjUct0scZ9kKV2Jr9/+KBAoWy4UMYxgv4=
 github.com/streamingfast/substreams v1.24.1-0.20261002193310-24d98f5409d4 h1:BQg/21hD1y/Z87GZ44t8sk4AIztpD9+tJRG5cArP04E=
 github.com/streamingfast/substreams v1.24.1-0.20261002193310-24d98f5409d4/go.mod h1:PI+cREzoPJzPUgXboQeq0UvpI2roq3ye3NiGOkM44bY=
+github.com/streamingfast/substreams v1.24.1-0.20261008022204-2409cca4e243 h1:n921cWcgyf20p6HNzfSbz/D8vQQA7ZGsyyNBM/JTYz8=
+github.com/streamingfast/substreams v1.24.1-0.20261008022204-2409cca4e243/go.mod h1:WYbIWjEPskQX5cKQ/yOETQq1cWq5YFy1uCbXoYQvLMQ=
 github.com/streamingfast/wazero v0.0.0-20241202185309-91287c3640ed h1:LU6/c376zP1cMAo9L6rFLyjo0W7RU+hIh7BegH8Zo5M=
 github.com/streamingfast/wazero v0.0.0-20241202185309-91287c3640ed/go.mod h1:yAI0XTsMBhREkM/YDAK/zNou3GoiAce1P6+rp/wQhjs=
 github.com/streamingfast/worker-pool-protocol v0.0.0-20251029142144-b539534f3eb1 h1:aoPCoeTHwCQbzRwLBpT45GUvZgAzevIUBgPZjJzBbug=

launcher/launcher.go

@@ -77,6 +77,7 @@ func (l *Launcher) Launch(appNames []string) error {
 	if len(appNames) == 0 {
 		return fmt.Errorf("no apps specified")
 	}
+	l.runtime.Apps = appNames
 	// This is done first as a sanity check so we don't launch anything if something is misconfigured
 	for _, appID := range appNames {
 		appDef, found := AppRegistry[appID]

launcher/runtime.go

@@ -9,6 +9,9 @@ type Runtime struct {
 	AbsDataDir string
 	InfoServer *info.InfoServer
 
+	// Apps are the IDs of the apps launched in the process.
+	Apps []string
+
 	// SessionPool is the pool configured by `common-session-plugin`, shared by every app of the
 	// process.
 	SessionPool dsession.SessionPool

stream_factory.go

@@ -141,6 +141,10 @@ func (sf *StreamFactory) New(
 	// configuration explicit at the stream level
 	options = append(options, stream.WithMergedBlocksBundleSize(bstream.DefaultMergedBlocksBundleSize))
 
+	// Firehose serves complete blocks only. The hub holds partial blocks when it
+	// is shared with substreams tier1, which needs them.
+	options = append(options, stream.WithoutPartialBlocks())
+
 	str := stream.New(
 		forkedBlocksStore,
 		mergedBlocksStore,