From 8f24383e42dc27d44fa24ec7e3e9c16ec1d5665a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ste=CC=81phane=20Duchesneau?= Date: Sun, 5 Jul 2026 19:58:43 -0400 Subject: [PATCH 1/5] Return error on nil Modules in ProcessRange The InvalidArgument error was built and logged but never returned, so the request fell through to a nil pointer dereference. --- service/tier2.go | 1 + service/tier2_processrange_test.go | 51 ++++++++++++++++++++++++++++++ 2 files changed, 52 insertions(+) create mode 100644 service/tier2_processrange_test.go diff --git a/service/tier2.go b/service/tier2.go index 78d228446..74ad4a8f6 100644 --- a/service/tier2.go +++ b/service/tier2.go @@ -291,6 +291,7 @@ func (s *Tier2Service) ProcessRange(request *pbssinternal.ProcessRangeRequest, s err := status.Error(codes.InvalidArgument, "missing modules in request") fields = append(fields, zap.Error(err)) logger.Info("refusing Substreams ProcessRange request", fields...) + return err } moduleNames := make([]string, len(request.Modules.Modules)) for i := 0; i < len(moduleNames); i++ { diff --git a/service/tier2_processrange_test.go b/service/tier2_processrange_test.go new file mode 100644 index 000000000..8891b6c6b --- /dev/null +++ b/service/tier2_processrange_test.go @@ -0,0 +1,51 @@ +package service + +import ( + "context" + "testing" + + "github.com/streamingfast/substreams/metrics" + pbssinternal "github.com/streamingfast/substreams/pb/sf/substreams/intern/v2" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "go.uber.org/zap" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/metadata" + "google.golang.org/grpc/status" +) + +type fakeProcessRangeServer struct { + ctx context.Context +} + +func (s *fakeProcessRangeServer) Send(*pbssinternal.ProcessRangeResponse) error { return nil } +func (s *fakeProcessRangeServer) SetHeader(metadata.MD) error { return nil } +func (s *fakeProcessRangeServer) SendHeader(metadata.MD) error { return nil } +func (s *fakeProcessRangeServer) SetTrailer(metadata.MD) {} +func (s *fakeProcessRangeServer) Context() context.Context { return s.ctx } +func (s *fakeProcessRangeServer) SendMsg(any) error { return nil } +func (s *fakeProcessRangeServer) RecvMsg(any) error { return nil } + +// TestProcessRange_NilModules ensures that a malformed ProcessRangeRequest +// with no Modules is refused with InvalidArgument instead of panicking on a +// nil pointer dereference when building the module names list. +func TestProcessRange_NilModules(t *testing.T) { + metrics.DeclareTier2Metrics(zap.NewNop()) // tier2 metrics are lazily declared by the app, ProcessRange needs them + + svc, err := NewTier2( + context.Background(), + zap.NewNop(), + func() bool { return false }, + WithReadinessFunc(func(bool) {}), + ) + require.NoError(t, err) + + streamSrv := &fakeProcessRangeServer{ctx: context.Background()} + + err = svc.ProcessRange(&pbssinternal.ProcessRangeRequest{Modules: nil}, streamSrv) + require.Error(t, err) + + st, ok := status.FromError(err) + require.True(t, ok, "expected a gRPC status error, got: %v", err) + assert.Equal(t, codes.InvalidArgument, st.Code()) +} From 42dd10c1c9705a6f4576a4889a1ac982a3b03c62 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ste=CC=81phane=20Duchesneau?= Date: Sun, 5 Jul 2026 19:59:04 -0400 Subject: [PATCH 2/5] Fix map race in GetExecutionPlan cleanup goroutine The goroutine closing exec output readers on cancellation iterated existingExecOuts while the main loop was still populating it, a fatal concurrent map iteration and write. --- service/tier2.go | 15 ++++- service/tier2_execplan_test.go | 100 +++++++++++++++++++++++++++++++++ 2 files changed, 113 insertions(+), 2 deletions(-) create mode 100644 service/tier2_execplan_test.go diff --git a/service/tier2.go b/service/tier2.go index 74ad4a8f6..adad8c647 100644 --- a/service/tier2.go +++ b/service/tier2.go @@ -864,8 +864,13 @@ func GetExecutionPlan( } } + // existingExecOuts is written by the loop below while the cleanup goroutine + // iterates over it on context cancellation: guard all accesses with a mutex. + var existingExecOutsLock sync.Mutex go func() { <-ctx.Done() + existingExecOutsLock.Lock() + defer existingExecOutsLock.Unlock() for _, eo := range existingExecOuts { if err := eo.Close(); err != nil { logger.Info("error closing reader", zap.String("filename", eo.Filename()), zap.Error(err)) @@ -874,6 +879,12 @@ func GetExecutionPlan( // Writers don't need to be closed, canceling the context is enough }() + setExistingExecOut := func(name string, file execout.FileReader) { + existingExecOutsLock.Lock() + defer existingExecOutsLock.Unlock() + existingExecOuts[name] = file + } + for _, mod := range usedModules { if mod.InitialBlock >= stopBlock { continue @@ -912,7 +923,7 @@ func GetExecutionPlan( requiredModules[name] = usedModules[name] break } - existingExecOuts[name] = file + setExistingExecOut(name, file) case pbsubstreams.ModuleKindStore: file, readErr := c.OpenFileReader(ctx, &block.Range{StartBlock: moduleStartBlock, ExclusiveEndBlock: stopBlock}) @@ -922,7 +933,7 @@ func GetExecutionPlan( } requiredModules[name] = usedModules[name] } else { - existingExecOuts[name] = file + setExistingExecOut(name, file) } // if either full or partial kv exists, we can skip the module diff --git a/service/tier2_execplan_test.go b/service/tier2_execplan_test.go new file mode 100644 index 000000000..bb3ff6762 --- /dev/null +++ b/service/tier2_execplan_test.go @@ -0,0 +1,100 @@ +package service + +import ( + "context" + "fmt" + "testing" + "time" + + "github.com/streamingfast/dstore" + pbsubstreams "github.com/streamingfast/substreams/pb/sf/substreams/v1" + "github.com/streamingfast/substreams/pipeline/exec" + "github.com/streamingfast/substreams/storage/execout" + "github.com/streamingfast/substreams/storage/index" + "github.com/streamingfast/substreams/storage/store" + "github.com/stretchr/testify/require" + "go.uber.org/zap" +) + +// TestGetExecutionPlan_CancelledContextNoMapRace exercises the cleanup +// goroutine of GetExecutionPlan, which closes the already-opened exec output +// readers when the context is cancelled. That goroutine iterates over the +// existingExecOuts map while the main loop is still populating it: without +// synchronization this is a concurrent map iteration/write (caught by -race, +// and a fatal "concurrent map iteration and map write" in production). +func TestGetExecutionPlan_CancelledContextNoMapRace(t *testing.T) { + logger := zap.NewNop() + + const moduleCount = 20 + modules := make([]*pbsubstreams.Module, moduleCount) + for i := 0; i < moduleCount; i++ { + mod := &pbsubstreams.Module{ + Name: fmt.Sprintf("mod_%d", i), + Kind: &pbsubstreams.Module_KindMap_{}, + } + if i == 0 { + mod.Inputs = []*pbsubstreams.Module_Input{ + { + Input: &pbsubstreams.Module_Input_Source_{ + Source: &pbsubstreams.Module_Input_Source{Type: "test.Block"}, + }, + }, + } + } else { + mod.Inputs = []*pbsubstreams.Module_Input{ + { + Input: &pbsubstreams.Module_Input_Map_{ + Map: &pbsubstreams.Module_Input_Map{ModuleName: fmt.Sprintf("mod_%d", i-1)}, + }, + }, + } + } + modules[i] = mod + } + + outputModule := fmt.Sprintf("mod_%d", moduleCount-1) + execGraph, err := exec.NewOutputModuleGraph(outputModule, true, &pbsubstreams.Modules{ + Modules: modules, + Binaries: []*pbsubstreams.Binary{ + {Type: "test", Content: []byte("some-fake-binary-data")}, + }, + }, 0) + require.NoError(t, err) + + const stage = uint32(0) + const startBlock = uint64(0) + const stopBlock = uint64(10) + + // Pre-write an (empty, thus valid and "ordered") exec output file for + // every module so that the main loop of GetExecutionPlan writes each of + // them into the existingExecOuts map. + baseStore := dstore.NewMockStore(nil) + for _, mod := range execGraph.UsedModulesUpToStage(int(stage)) { + hash := execGraph.ModuleHashes()[mod.Name] + baseStore.SetFile(fmt.Sprintf("%s/outputs/%010d-%010d.output", hash, startBlock, stopBlock), []byte{}) + } + + execoutConfigs, err := execout.NewConfigs(baseStore, execGraph.UsedModulesUpToStage(int(stage)), execGraph.ModuleHashes(), stopBlock, 0, logger) + require.NoError(t, err) + + storeConfigs, err := store.NewConfigMap(baseStore, nil, execGraph.Stores(), execGraph.ModuleHashes(), 0, 0) + require.NoError(t, err) + + indexConfigs, err := index.NewConfigs(baseStore, execGraph.UsedIndexesModulesUpToStage(int(stage)), execGraph.ModuleHashes(), 0, logger) + require.NoError(t, err) + + // The context is already cancelled when GetExecutionPlan starts: the + // cleanup goroutine fires immediately and iterates the map concurrently + // with the population loop. + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + plan, err := GetExecutionPlan(ctx, logger, execGraph, stage, startBlock, stopBlock, outputModule, execoutConfigs, indexConfigs, storeConfigs) + require.NoError(t, err) + require.NotNil(t, plan) + require.Len(t, plan.ExistingExecOuts, moduleCount) + + // Give the cleanup goroutine time to run its iteration before the test + // ends, so the race detector observes both sides. + time.Sleep(100 * time.Millisecond) +} From 7d74e540a4ec91834c68a89e802ba8cd5db14df2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ste=CC=81phane=20Duchesneau?= Date: Sun, 5 Jul 2026 19:59:17 -0400 Subject: [PATCH 3/5] Fix reversed errors.Is args on stream error errors.Is(context.Canceled, streamErr) never matches a wrapped cancellation, so context.Cause was never substituted. --- service/tier2.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/service/tier2.go b/service/tier2.go index adad8c647..d206c1e90 100644 --- a/service/tier2.go +++ b/service/tier2.go @@ -654,7 +654,7 @@ excludable: streamErr = blockStream.Run(ctx) span.EndWithErr(&streamErr) - if errors.Is(context.Canceled, streamErr) { + if errors.Is(streamErr, context.Canceled) { streamErr = context.Cause(ctx) } From 316b9d3e47d2c67a544a11f67831789b9204014f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ste=CC=81phane=20Duchesneau?= Date: Sun, 5 Jul 2026 19:59:17 -0400 Subject: [PATCH 4/5] Fix connection and goroutine leaks in live back filler Close the tier2 client connection on all requestBackProcessing exit paths (it leaked on errors and was closed twice on success), and stop RequestBackProcessing from blocking forever on its result channel after Start has returned on context cancellation. --- service/live_back_filler.go | 28 +++--- service/live_back_filler_leak_test.go | 125 ++++++++++++++++++++++++++ 2 files changed, 140 insertions(+), 13 deletions(-) create mode 100644 service/live_back_filler_leak_test.go diff --git a/service/live_back_filler.go b/service/live_back_filler.go index eecc193e3..fcf79257c 100644 --- a/service/live_back_filler.go +++ b/service/live_back_filler.go @@ -84,7 +84,10 @@ func RequestBackProcessing(ctx context.Context, logger *zap.Logger, startBlock u return nil }) - jobResult <- err + select { + case jobResult <- err: + case <-ctx.Done(): + } } func requestBackProcessing(ctx context.Context, logger *zap.Logger, liveCachingRequest *pbssinternal.ProcessRangeRequest, clientFactory client.InternalClientFactory) error { @@ -94,6 +97,11 @@ func requestBackProcessing(ctx context.Context, logger *zap.Logger, liveCachingR if err != nil { return fmt.Errorf("failed to create live cache grpc client: %w", err) } + defer func() { + if err := closeFunc(); err != nil { + logger.Warn("closing client connection", zap.Error(err)) + } + }() if headers.IsSet() { ctx = metadata.AppendToOutgoingContext(ctx, headers.ToArray()...) @@ -103,6 +111,11 @@ func requestBackProcessing(ctx context.Context, logger *zap.Logger, liveCachingR if err != nil { return fmt.Errorf("getting stream: %w", err) } + defer func() { + if err := stream.CloseSend(); err != nil { + logger.Warn("closing stream", zap.Error(err)) + } + }() for { _, err = stream.Recv() @@ -114,17 +127,6 @@ func requestBackProcessing(ctx context.Context, logger *zap.Logger, liveCachingR } } - defer func() { - if err = stream.CloseSend(); err != nil { - logger.Warn("closing stream", zap.Error(err)) - } - if err = closeFunc(); err != nil { - logger.Warn("closing stream", zap.Error(err)) - } - }() - - closeFunc() - return nil } @@ -149,7 +151,7 @@ func (l *LiveBackFiller) Start(ctx context.Context) { var jobFailed bool var jobProcessing bool var blockNumber uint64 - jobResult := make(chan error) + jobResult := make(chan error, 1) for { select { case <-ctx.Done(): diff --git a/service/live_back_filler_leak_test.go b/service/live_back_filler_leak_test.go new file mode 100644 index 000000000..5e027b4ab --- /dev/null +++ b/service/live_back_filler_leak_test.go @@ -0,0 +1,125 @@ +package service + +import ( + "context" + "errors" + "io" + "testing" + "time" + + pbssinternal "github.com/streamingfast/substreams/pb/sf/substreams/intern/v2" + "github.com/streamingfast/substreams/client" + "github.com/streamingfast/substreams/reqctx" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "go.uber.org/zap" + "google.golang.org/grpc" + "google.golang.org/grpc/metadata" +) + +type fakeProcessRangeClientStream struct { + recvErr error +} + +func (s *fakeProcessRangeClientStream) Recv() (*pbssinternal.ProcessRangeResponse, error) { + return nil, s.recvErr +} +func (s *fakeProcessRangeClientStream) Header() (metadata.MD, error) { return nil, nil } +func (s *fakeProcessRangeClientStream) Trailer() metadata.MD { return nil } +func (s *fakeProcessRangeClientStream) CloseSend() error { return nil } +func (s *fakeProcessRangeClientStream) Context() context.Context { return context.Background() } +func (s *fakeProcessRangeClientStream) SendMsg(any) error { return nil } +func (s *fakeProcessRangeClientStream) RecvMsg(any) error { return nil } + +type fakeSubstreamsClient struct { + processRangeErr error + recvErr error +} + +func (c *fakeSubstreamsClient) ProcessRange(ctx context.Context, in *pbssinternal.ProcessRangeRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[pbssinternal.ProcessRangeResponse], error) { + if c.processRangeErr != nil { + return nil, c.processRangeErr + } + return &fakeProcessRangeClientStream{recvErr: c.recvErr}, nil +} + +func newCountingClientFactory(fakeClient *fakeSubstreamsClient, closeCount *int) client.InternalClientFactory { + return func() (pbssinternal.SubstreamsClient, func() error, []grpc.CallOption, client.Headers, error) { + closeFunc := func() error { + *closeCount++ + return nil + } + return fakeClient, closeFunc, nil, nil, nil + } +} + +// TestRequestBackProcessing_ConnectionClosed verifies that the grpc client +// connection created by the clientFactory is closed exactly once on every +// code path of requestBackProcessing: previously, error returns leaked the +// connection and the success path closed it twice. +func TestRequestBackProcessing_ConnectionClosed(t *testing.T) { + cases := []struct { + name string + fakeClient *fakeSubstreamsClient + expectError bool + }{ + { + name: "error getting stream", + fakeClient: &fakeSubstreamsClient{processRangeErr: errors.New("connection refused")}, + expectError: true, + }, + { + name: "error receiving from stream", + fakeClient: &fakeSubstreamsClient{recvErr: errors.New("stream broken")}, + expectError: true, + }, + { + name: "success", + fakeClient: &fakeSubstreamsClient{recvErr: io.EOF}, + expectError: false, + }, + } + + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + closeCount := 0 + factory := newCountingClientFactory(c.fakeClient, &closeCount) + + request := &pbssinternal.ProcessRangeRequest{SegmentSize: 10} + err := requestBackProcessing(context.Background(), zap.NewNop(), request, factory) + if c.expectError { + require.Error(t, err) + } else { + require.NoError(t, err) + } + assert.Equal(t, 1, closeCount, "clientFactory closeFunc must be called exactly once") + }) + } +} + +// TestRequestBackProcessing_NoGoroutineLeakOnCancel verifies that +// RequestBackProcessing does not block forever on its result channel when the +// context is cancelled and nobody is reading the channel anymore (which is +// what happens when LiveBackFiller.Start returns on ctx.Done()). +func TestRequestBackProcessing_NoGoroutineLeakOnCancel(t *testing.T) { + ctx := reqctx.WithRequest(context.Background(), &reqctx.RequestDetails{}) + ctx = reqctx.WithTier2RequestParameters(ctx, reqctx.Tier2RequestParameters{StateBundleSize: 10}) + ctx, cancel := context.WithCancel(ctx) + cancel() + + closeCount := 0 + factory := newCountingClientFactory(&fakeSubstreamsClient{processRangeErr: errors.New("unreachable")}, &closeCount) + + jobResult := make(chan error) // unbuffered and never read, like after Start() has returned + done := make(chan struct{}) + go func() { + RequestBackProcessing(ctx, zap.NewNop(), 0, 0, factory, jobResult) + close(done) + }() + + select { + case <-done: + case <-time.After(5 * time.Second): + t.Fatal("RequestBackProcessing leaked: still blocked sending its result after context cancellation") + } +} From 404324c7fd51854ddd14f9b755c340079ac90dd8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ste=CC=81phane=20Duchesneau?= Date: Sun, 5 Jul 2026 19:59:24 -0400 Subject: [PATCH 5/5] Propagate cancellation in cursor resolver When the context is cancelled the source shutdown may win the select via src.Terminated(), falling through to the error fallback which emitted an undo-to-LIB signal instead of the context error (#399). --- pipeline/resolve.go | 7 +++++ pipeline/resolve_cancel_test.go | 45 +++++++++++++++++++++++++++++++++ 2 files changed, 52 insertions(+) create mode 100644 pipeline/resolve_cancel_test.go diff --git a/pipeline/resolve.go b/pipeline/resolve.go index 427d9421e..f4109f9cb 100644 --- a/pipeline/resolve.go +++ b/pipeline/resolve.go @@ -316,6 +316,13 @@ func NewCursorResolver(hub *hub.ForkableHub, mergedBlocksStore, forkedBlocksStor src.Shutdown(ctx.Err()) return nil, nil, ctx.Err() case <-src.Terminated(): + // On cancellation, the shutdown goroutine above may terminate the + // source before this select observes ctx.Done(): in that case we + // must propagate the cancellation instead of falling through to + // the fallback below, which would emit an undo-to-LIB signal. + if ctx.Err() != nil { + return nil, nil, ctx.Err() + } } if !errors.Is(src.Err(), ErrDone) { diff --git a/pipeline/resolve_cancel_test.go b/pipeline/resolve_cancel_test.go new file mode 100644 index 000000000..ff8e9510e --- /dev/null +++ b/pipeline/resolve_cancel_test.go @@ -0,0 +1,45 @@ +package pipeline + +import ( + "context" + "errors" + "testing" + + "github.com/streamingfast/bstream" + "github.com/streamingfast/bstream/hub" + "github.com/streamingfast/dstore" + "github.com/stretchr/testify/require" +) + +// TestCursorResolver_CancellationDoesNotEmitUndo covers the residual race of +// issue #399: when the context is cancelled, the source shutdown goroutine may +// terminate the source before the ctx.Done() select case is picked. If the +// src.Terminated() case wins, the resolver used to fall through to the error +// fallback and emit an undo-to-LIB signal (junction = cursor.LIB, err = nil) +// instead of propagating the cancellation. +// +// The select choice between the two ready channels is randomized by the +// runtime, so we run several iterations: every single one must surface the +// context error, never an undo signal. +func TestCursorResolver_CancellationDoesNotEmitUndo(t *testing.T) { + mergedBlocksStore := dstore.NewMockStore(nil) + forkedBlocksStore := dstore.NewMockStore(nil) + + resolver := NewCursorResolver((*hub.ForkableHub)(nil), mergedBlocksStore, forkedBlocksStore) + + cursor := &bstream.Cursor{ + Step: bstream.StepNew, + Block: bstream.NewBlockRef("bb", 105), + HeadBlock: bstream.NewBlockRef("bb", 105), + LIB: bstream.NewBlockRef("aa", 100), + } + + for i := 0; i < 30; i++ { + ctx, cancel := context.WithCancel(context.Background()) + cancel() // cancelled before resolution starts + + junction, head, err := resolver(ctx, cursor) + require.Error(t, err, "iteration %d: cancellation was swallowed, got junction=%v head=%v (undo-to-LIB signal)", i, junction, head) + require.True(t, errors.Is(err, context.Canceled), "iteration %d: expected context.Canceled, got: %v", i, err) + } +}