Fix tier2/live-backfiller crash, race, leaks and cursor-resolver cancellation

pipeline/resolve.go

@@ -319,6 +319,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) {

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)
+	}
+}

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():

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")
+	}
+}

service/tier2.go

@@ -292,6 +292,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++ {
@@ -656,7 +657,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)
 	}
 
@@ -866,8 +867,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))
@@ -876,6 +882,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
@@ -914,7 +926,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})
@@ -924,7 +936,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