Feature/mmap store backend

app/tier1.go

@@ -81,6 +81,8 @@ type Tier1Config struct {
 
 	TmpDir                  string
 	StateStoreURL           string
+	StoresScratchSpace      string
+	StoresBackend           string
 	QuickSaveStoreURL       string
 	StateStoreDefaultTag    string
 	BlockType               string
@@ -313,6 +315,8 @@ func (a *Tier1App) Run() error {
 		a.modules.SessionPool,
 		foundationalStoreEndpoints,
 		a.config.HostedStoreRegistryAddress,
+		a.config.StoresScratchSpace,
+		a.config.StoresBackend,
 		opts...,
 	)
 	if err != nil {

app/tier2.go

@@ -32,12 +32,14 @@ type Tier2Config struct {
 
 	PipelineOptions []pipeline.Option
 
-	MaximumConcurrentRequests  uint64
-	WASMExtensions             wasm.WASMExtensioner
-	BlockExecutionTimeout      time.Duration
-	SegmentExecutionTimeout    time.Duration
-	TmpDir                     string
-	HostedStoreRegistryAddress string // for hosted stores; legacy use JSON
+	MaximumConcurrentRequests uint64
+	WASMExtensions            wasm.WASMExtensioner
+	BlockExecutionTimeout     time.Duration
+	SegmentExecutionTimeout   time.Duration
+	TmpDir                    string
+	StoresScratchSpace             string
+	StoresBackend                  string
+	HostedStoreRegistryAddress     string
 
 	Tracing bool
 }
@@ -107,6 +109,13 @@ func (a *Tier2App) Run() error {
 		opts = append(opts, service.WithWASMExtensioner(a.config.WASMExtensions))
 	}
 
+	if a.config.StoresScratchSpace != "" {
+		opts = append(opts, service.WithStoresScratchSpace(a.config.StoresScratchSpace))
+	}
+	if a.config.StoresBackend != "" {
+		opts = append(opts, service.WithStoresBackend(a.config.StoresBackend))
+	}
+
 	svc, err := service.NewTier2(
 		ctx,
 		a.logger,

docs/release-notes/change-log.md

@@ -29,6 +29,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 
 ### Fixed
 
+- Server: canceled store loads now abort promptly instead of reading the entire (multi-GB) store file into the heap before returning. A tier2 store `Load` now checks the request context while streaming entries, so a canceled/disconnected request stops hydrating immediately. Also `memoryKVImpl.Close()` now drops its backing map (was a no-op), and the post-load `SetMetadata` goroutine no longer captures the whole loaded store (only the object store, filename and metadata) and runs under a bounded 30s timeout — together these stop a finished/canceled request from transiently pinning gigabytes of store data.
 - Server: bumped `dstore`, which now sends S3 request checksums only when the target requires them, fixing uploads/downloads against S3-compatible stores that reject the newer default checksum headers.
 - Server: the quicksave block count now counts settled blocks (normal or last-partial) instead of skipping partials entirely, so quicksave arms correctly on flash-block chains. The minimum sent-block threshold before a quicksave triggers was raised from 25 to 50.
 - Server: `tier1` calls to hosted foundational stores now forward the `x-organization-id` identity header (alongside the existing trusted headers), so a store's internal trust-based listener can authorize the request without an end-user JWT. This fixes `Unauthenticated: required authorization token not found` errors when reading from hosted foundational stores resolved via the control-plane registry.

go.mod

@@ -77,6 +77,7 @@ require (
 	github.com/test-go/testify v1.1.4
 	github.com/tetratelabs/wazero v1.8.0
 	github.com/tidwall/pretty v1.2.1
+	go.etcd.io/bbolt v1.4.3
 	go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.65.0
 	go.opentelemetry.io/otel v1.43.0
 	go.opentelemetry.io/otel/trace v1.43.0

go.sum

@@ -580,6 +580,8 @@ github.com/yuin/goldmark v1.5.4 h1:2uY/xC0roWy8IBEGLgB1ywIoEJFGmRrX21YQcvGZzjU=
 github.com/yuin/goldmark v1.5.4/go.mod h1:6yULJ656Px+3vBD8DxQVa3kxgyrAnzto9xy5taEt/CY=
 github.com/yuin/goldmark-emoji v1.0.2 h1:c/RgTShNgHTtc6xdz2KKI74jJr6rWi7FPgnP9GAsO5s=
 github.com/yuin/goldmark-emoji v1.0.2/go.mod h1:RhP/RWpexdp+KHs7ghKnifRoIs/Bq4nDS7tRbCkOwKY=
+go.etcd.io/bbolt v1.4.3 h1:dEadXpI6G79deX5prL3QRNP6JB8UxVkqo4UPnHaNXJo=
+go.etcd.io/bbolt v1.4.3/go.mod h1:tKQlpPaYCVFctUIgFKFnAlvbmB3tpy1vkTnDWohtc0E=
 go.opencensus.io v0.21.0/go.mod h1:mSImk1erAIZhrmZN+AvHh14ztQfjbGwt4TtuofqLduU=
 go.opencensus.io v0.22.0/go.mod h1:+kGneAE2xo2IficOXnaByMWTGM9T73dGwxeWcUqIpI8=
 go.opencensus.io v0.22.1/go.mod h1:Ap50jQcDJrx6rB6VgeeFPtuPIf3wMRvRfrfYDO6+BmA=

go.work.sum

@@ -1318,6 +1318,7 @@ github.com/streamingfast/bstream v0.0.2-0.20260219185151-b5ba3f4bc777 h1:odiGZ0Z
 github.com/streamingfast/bstream v0.0.2-0.20260219185151-b5ba3f4bc777/go.mod h1:6IkpMw6g2z+E4gH2DJkm6vsQfZ8Y/WdYdYWjsLJIYr4=
 github.com/streamingfast/bstream v0.0.2-0.20260219205021-a73f004bfdea h1:dshasnf6SEnOvhmbsMxw6SA+qm+eiuNp45IQX47Ivac=
 github.com/streamingfast/bstream v0.0.2-0.20260219205021-a73f004bfdea/go.mod h1:6IkpMw6g2z+E4gH2DJkm6vsQfZ8Y/WdYdYWjsLJIYr4=
+github.com/streamingfast/bstream v0.0.2-0.20260611155534-4edda1cff251/go.mod h1:Asuul2Rnxln0Vk6eY869AZ2UybIQcB7eYOQAIhfQHzE=
 github.com/streamingfast/derr v0.0.0-20210811180100-9138d738bcec/go.mod h1:ulVfui/yGXmPBbt9aAqCWdAjM7YxnZkYHzvQktLfw3M=
 github.com/streamingfast/dgrpc v0.0.0-20250227145723-9bc2e4941b4e/go.mod h1:qdcskj9WmO+nDjgxyafc5oMCftTZolOk693p2ZFUdO8=
 github.com/streamingfast/dgrpc v0.0.0-20251218133127-15b36e02a74f/go.mod h1:B+RAf6/idk6Kz41wEAEeQH1PW3Bk6WQkmvuRhdTEgKg=
@@ -1413,6 +1414,7 @@ go.etcd.io/etcd/pkg/v3 v3.6.8 h1:Xe+LIL974spy8b4nEx3H0KMr1ofq3r0kh6FbU3aw4es=
 go.etcd.io/etcd/pkg/v3 v3.6.8/go.mod h1:TRibVNe+FqJIe1abOAA1PsuQ4wqO87ZaOoprg09Tn8c=
 go.etcd.io/etcd/server/v3 v3.6.8 h1:U2strdSEy1U8qcSzRIdkYpvOPtBy/9i/IfaaCI9flZ4=
 go.etcd.io/etcd/server/v3 v3.6.8/go.mod h1:88dCtwUnSirkUoJbflQxxWXqtBSZa6lSG0Kuej+dois=
+go.etcd.io/gofail v0.2.0/go.mod h1:nL3ILMGfkXTekKI3clMBNazKnjUZjYLKmBHzsVAnC1o=
 go.etcd.io/raft/v3 v3.6.0 h1:5NtvbDVYpnfZWcIHgGRk9DyzkBIXOi8j+DDp1IcnUWQ=
 go.etcd.io/raft/v3 v3.6.0/go.mod h1:nLvLevg6+xrVtHUmVaTcTz603gQPHfh7kUAwV6YpfGo=
 go.opencensus.io v0.23.0/go.mod h1:XItmlyltB5F7CS4xOC1DcqMoFqwtC6OG2xF7mCv7P7E=

metrics/metrics.go

@@ -32,6 +32,10 @@ var SkippedCachedWasmModules = MetricSet.NewCounter("substreams_skipped_cached_w
 
 var Tier1OutputHeadBlockRelativeTime *dmetrics.HeadBlockRelativeTime
 
+// Store backend metrics
+var StoreBackendType *dmetrics.GaugeVec
+var StoreMmapFileSizeBytes *dmetrics.GaugeVec
+var StoreMmapOperationsTotal *dmetrics.CounterVec
 var Tier1ActiveRequestsHardLimit *dmetrics.Gauge
 var Tier2MaxConcurrentRequests *dmetrics.Gauge
 
@@ -55,6 +59,22 @@ func DeclareTier1Metrics(zlog *zap.Logger) {
 	Tier1OutputHeadBlockRelativeTime = MetricSet.NewHeadBlockRelativeTime("substreams_output")
 	Tier1ActiveRequestsHardLimit = MetricSet.NewGauge("substreams_tier1_active_requests_hard_limit", "Hard limit of concurrent active requests on tier1 (0 means unlimited)")
 
+	StoreBackendType = MetricSet.NewGaugeVec(
+		"substreams_store_backend_type",
+		[]string{"backend", "store_name"},
+		"Store backend type in use (1=mmap, 2=memory)",
+	)
+	StoreMmapFileSizeBytes = MetricSet.NewGaugeVec(
+		"substreams_store_mmap_file_size_bytes",
+		[]string{"store_name"},
+		"Size of mmap database files in bytes",
+	)
+	StoreMmapOperationsTotal = MetricSet.NewCounterVec(
+		"substreams_store_mmap_operations_total",
+		[]string{"operation", "store_name"},
+		"Total number of mmap operations (get, set, delete, scan)",
+	)
+
 	zlog.Info("registering substreams tier1 metrics")
 }
 

orchestrator/parallelprocessor.go

@@ -103,6 +103,7 @@ func (b *ParallelProcessor) Stages() *stage.Stages {
 func (b *ParallelProcessor) Run(ctx context.Context, checkPendingShutdown func() bool) (storeMap store.Map, err error) {
 	ctx, cancel := context.WithCancel(ctx)
 	defer cancel()
+	defer b.scheduler.Stages.Close()
 
 	go func() {
 		for {

orchestrator/stage/fetchstorage_test.go

@@ -569,7 +569,7 @@ func testStoreConfig(
 	valueType string,
 	objStore dstore.Store,
 ) *store.Config {
-	conf, err := store.NewConfig(name, moduleInitialBlock, moduleHash, updatePolicy, valueType, objStore, nil, 0)
+	conf, err := store.NewConfig(name, moduleInitialBlock, moduleHash, updatePolicy, valueType, objStore, nil, 0, "", "")
 	require.NoError(t, err)
 	return conf
 }

orchestrator/stage/modstate.go

@@ -43,6 +43,7 @@ func (s *StoreModuleState) estimateStoreSizeBytes(ctx context.Context, exclusive
 	}
 
 	fullKV := s.storeConfig.NewFullKV(s.logger)
+	defer fullKV.Close() // only used for GetSize; releases the backing mmap file
 
 	moduleInitBlock := s.storeConfig.ModuleInitialBlock()
 	if moduleInitBlock < exclusiveEndBlock {
@@ -70,24 +71,63 @@ func (s *StoreModuleState) getStore(ctx context.Context, exclusiveEndBlock uint6
 		fullKVFile := store.NewCompleteFileInfo(s.name, moduleInitBlock, exclusiveEndBlock)
 		err := loadStore.Load(ctx, fullKVFile)
 		if err != nil {
-			if errors.Is(err, store.ErrInvalidFullKVFile) {
+			// Only a live context means real corruption: a canceled load also
+			// surfaces as ErrInvalidFullKVFile and must not delete a valid file.
+			if errors.Is(err, store.ErrInvalidFullKVFile) && ctx.Err() == nil {
 				s.logger.Warn("found a corrupted fullKV store, deleting it", zap.Error(err), zap.String("store_name", loadStore.Name()), zap.String("store_hash", loadStore.ModuleHash()), zap.String("filename", fullKVFile.Filename))
 				if err := loadStore.Delete(ctx, fullKVFile); err != nil {
 					s.logger.Error("cannot delete corrupted fullKV store", zap.Error(err), zap.String("store_name", loadStore.Name()), zap.String("store_hash", loadStore.ModuleHash()), zap.String("filename", fullKVFile.Filename))
 				}
 			}
+			loadStore.Close()
 			return nil, fmt.Errorf("load full store %s (%s): %w", loadStore.Name(), loadStore.ModuleHash(), err)
 		}
 	}
+	if s.cachedStore != nil {
+		s.cachedStore.Close()
+	}
 	s.cachedStore = loadStore
 	s.lastBlockInStore = exclusiveEndBlock
 	return loadStore, nil
 }
 
+func (s *StoreModuleState) Close() {
+	if s.cachedStore != nil {
+		s.cachedStore.Close()
+		s.cachedStore = nil
+	}
+}
+
 func (s *StoreModuleState) derivePartialKV(initialBlock uint64) *store.PartialKV {
 	return s.storeConfig.NewPartialKV(initialBlock, s.logger)
 }
 
+// tryLoadFullKV attempts to load a FullKV from storage for the given block range.
+// It does NOT touch cachedStore — use getStore for cache-aware loading.
+// Returns (nil, nil) if moduleInitBlock >= exclusiveEndBlock.
+// Returns (nil, err) if the file doesn't exist or is corrupt; callers treat this as "no fullKV available".
+func (s *StoreModuleState) tryLoadFullKV(ctx context.Context, exclusiveEndBlock uint64) (*store.FullKV, error) {
+	moduleInitBlock := s.storeConfig.ModuleInitialBlock()
+	if moduleInitBlock >= exclusiveEndBlock {
+		return nil, nil
+	}
+	kv := s.storeConfig.NewFullKV(s.logger)
+	fullKVFile := store.NewCompleteFileInfo(s.name, moduleInitBlock, exclusiveEndBlock)
+	if err := kv.Load(ctx, fullKVFile); err != nil {
+		// A canceled load also surfaces as ErrInvalidFullKVFile; only delete
+		// when the context is still live so we never drop a valid store file.
+		if errors.Is(err, store.ErrInvalidFullKVFile) && ctx.Err() == nil {
+			s.logger.Warn("found a corrupted fullKV store, deleting it", zap.Error(err), zap.String("store_name", kv.Name()), zap.String("store_hash", kv.ModuleHash()), zap.String("filename", fullKVFile.Filename))
+			if delErr := kv.Delete(ctx, fullKVFile); delErr != nil {
+				s.logger.Error("cannot delete corrupted fullKV store", zap.Error(delErr), zap.String("store_name", kv.Name()), zap.String("store_hash", kv.ModuleHash()), zap.String("filename", fullKVFile.Filename))
+			}
+		}
+		kv.Close()
+		return nil, err
+	}
+	return kv, nil
+}
+
 //type MergeState int
 //
 //const (

orchestrator/stage/squash.go

@@ -7,6 +7,7 @@ import (
 
 	"go.uber.org/zap"
 
+	"github.com/dustin/go-humanize"
 	"github.com/hashicorp/go-multierror"
 	"github.com/streamingfast/derr"
 	"github.com/streamingfast/substreams/block"
@@ -69,7 +70,7 @@ func getPartialOrFullKV(ctx context.Context, modState *StoreModuleState, rng *bl
 	}()
 
 	go func() {
-		nextFull, err := modState.getStore(ctx, rng.ExclusiveEndBlock)
+		nextFull, err := modState.tryLoadFullKV(ctx, rng.ExclusiveEndBlock)
 		results <- Result{fullKVStore: nextFull, error: err}
 	}()
 
@@ -130,6 +131,9 @@ func (s *Stages) singleSquash(stage *Stage, modState *StoreModuleState, mergeUni
 	}
 
 	if newFullKV != nil {
+		if fullKV != nil {
+			fullKV.Close()
+		}
 		modState.cachedStore = newFullKV
 		modState.lastBlockInStore = rng.ExclusiveEndBlock
 		s.logger.Debug("squashing time metrics (skipped, loaded from full kv)", meter.logFields()...)
@@ -148,6 +152,12 @@ func (s *Stages) singleSquash(stage *Stage, modState *StoreModuleState, mergeUni
 	modState.lastBlockInStore = rng.ExclusiveEndBlock
 	meter.mergeEnd = time.Now()
 
+	s.logger.Info("merged partial into full store",
+		zap.String("store", modState.name),
+		zap.Uint64("up_to_block", rng.ExclusiveEndBlock),
+		zap.String("store_size", humanize.IBytes(fullKV.SizeBytes())),
+	)
+
 	s.logger.Info("deleting partial store", zap.Stringer("store", partialKV))
 
 	// Flush full store

orchestrator/stage/stages.go

@@ -148,6 +148,14 @@ func NewStages(
 	return out
 }
 
+func (s *Stages) Close() {
+	for _, stage := range s.stages {
+		for _, modState := range stage.storeModuleStates {
+			modState.Close()
+		}
+	}
+}
+
 func layerKind(layer exec.LayerModules) Kind {
 	if layer.IsStoreLayer() {
 		return KindStore
@@ -717,6 +725,14 @@ func (s *Stages) FinalStoreMap(exclusiveEndBlock uint64) (store.Map, error) {
 		reqHandler.AdjustFullKVSize(actualRequestStoresSize)
 	}
 
+	// Ownership of the loaded stores transfers to the returned map: the linear
+	// pipeline keeps writing to them and closes them at request end. Detach
+	// them from the module states so Stages.Close() does not close stores
+	// still in use.
+	for _, modState := range storeModuleStates {
+		modState.cachedStore = nil
+	}
+
 	return out, nil
 }
 

pipeline/pipeline.go

@@ -298,6 +298,13 @@ func (p *Pipeline) initStoresFromQuickload(ctx context.Context, reqPlan *plan.Re
 
 	if err := storeMap.QuickLoad(ctx, cursor.Block); err != nil {
 		p.stores.logger.Info("no temporary store files found", zap.Error(err))
+		for _, st := range storeMap.All() {
+			if closer, ok := st.(interface{ Close() error }); ok {
+				if cerr := closer.Close(); cerr != nil {
+					p.stores.logger.Warn("failed to close store after quickload failure", zap.String("store", st.Name()), zap.Error(cerr))
+				}
+			}
+		}
 		return false
 	}
 	reqctx.Logger(ctx).Info("skipping backprocessing, reading from temporary files", zap.Strings("stores", storeMap.Names()), zap.Uint64("block_num", cursor.Block.Num()), zap.String("block_id", cursor.Block.ID()))
@@ -390,6 +397,17 @@ func (p *Pipeline) setupSubrequestStores(ctx context.Context) (storeMap store.Ma
 	logger := reqctx.Logger(ctx)
 
 	storeMap = store.NewMap()
+	defer func() {
+		if err != nil {
+			for _, st := range storeMap.All() {
+				if closer, ok := st.(interface{ Close() error }); ok {
+					if cerr := closer.Close(); cerr != nil {
+						logger.Warn("failed to close store after setup error", zap.String("store", st.Name()), zap.Error(cerr))
+					}
+				}
+			}
+		}
+	}()
 
 	type loadable struct {
 		fullKVStore *store.FullKV
@@ -480,7 +498,11 @@ func (p *Pipeline) setupSubrequestStores(ctx context.Context) (storeMap store.Ma
 		}
 		egLoad.Go(func() error {
 			if err := loadable.fullKVStore.Load(ctx, loadable.fileInfo); err != nil {
-				if errors.Is(err, store.ErrInvalidFullKVFile) {
+				// A canceled context surfaces as ErrInvalidFullKVFile because the
+				// streaming unmarshal wraps the read error as a string, so only
+				// treat the file as corrupt when the context is still live —
+				// otherwise a canceled request would delete a valid store file.
+				if errors.Is(err, store.ErrInvalidFullKVFile) && ctx.Err() == nil {
 					logger.Warn("found a corrupted fullKV store, deleting it", zap.Error(err), zap.String("store_name", loadable.fullKVStore.Name()), zap.String("store_hash", loadable.fullKVStore.ModuleHash()), zap.String("filename", loadable.fileInfo.Filename))
 					if err := loadable.fullKVStore.Delete(ctx, loadable.fileInfo); err != nil {
 						logger.Error("cannot delete corrupted fullKV store", zap.Error(err), zap.String("store_name", loadable.fullKVStore.Name()), zap.String("store_hash", loadable.fullKVStore.ModuleHash()), zap.String("filename", loadable.fileInfo.Filename))
@@ -500,12 +522,21 @@ func (p *Pipeline) setupSubrequestStores(ctx context.Context) (storeMap store.Ma
 			}
 			actualRequestStoresSize += actualSize
 			loadMu.Unlock()
+			// Capture only the values the metadata write needs, NOT the whole
+			// fullKVStore — otherwise this un-awaited goroutine pins the entire
+			// loaded (multi-GB) KV map until SetMetadata returns. Use a bounded
+			// timeout detached from the request ctx so a canceled request neither
+			// pins the store nor leaves the goroutine hanging indefinitely.
+			metaStore := loadable.fullKVStore.Store()
+			metaFilename := loadable.fullKVStore.Filename()
+			metaStoreName := loadable.fullKVStore.Name()
 			go func() {
-				err := loadable.fullKVStore.Store().SetMetadata(ctx, loadable.fullKVStore.Filename(), met)
-				if err != nil {
+				metaCtx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+				defer cancel()
+				if err := metaStore.SetMetadata(metaCtx, metaFilename, met); err != nil {
 					logger.Debug("failed to set metadata on store",
-						zap.String("store_name", loadable.fullKVStore.Name()),
-						zap.String("filename", loadable.fullKVStore.Filename()),
+						zap.String("store_name", metaStoreName),
+						zap.String("filename", metaFilename),
 						zap.Error(err))
 				}
 			}()

pipeline/pipeline_test.go

@@ -189,7 +189,7 @@ func testConfigMap(t *testing.T, configs []testStoreConfig) store2.ConfigMap {
 	objStore := dstore.NewMockStore(nil)
 
 	for _, conf := range configs {
-		newStore, err := store2.NewConfig(conf.name, conf.initBlock, conf.name, pbsubstreams.Module_KindStore_UPDATE_POLICY_SET, "string", objStore, nil, 0)
+		newStore, err := store2.NewConfig(conf.name, conf.initBlock, conf.name, pbsubstreams.Module_KindStore_UPDATE_POLICY_SET, "string", objStore, nil, 0, "", "")
 		require.NoError(t, err)
 		confMap[newStore.Name()] = newStore
 

pipeline/snapshot.go

@@ -36,10 +36,17 @@ func (p *Pipeline) sendSnapshots(storeMap store.Map, snapshotModules []string) e
 
 		store.Iter(func(k string, v []byte) error {
 			count++
+			// Copy v: on the mmap backend Iter yields a slice aliasing bbolt's
+			// page buffer, valid only during the read transaction. This delta is
+			// accumulated and sent asynchronously (every 100 entries / at the
+			// end), outliving the scan, so the alias would be overwritten before
+			// it is serialized.
+			newValue := make([]byte, len(v))
+			copy(newValue, v)
 			accum = append(accum, &pbsubstreamsrpc.StoreDelta{
 				Operation: pbsubstreamsrpc.StoreDelta_CREATE,
 				Key:       k,
-				NewValue:  v,
+				NewValue:  newValue,
 			})
 
 			if count%100 == 0 {