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)
+	}
+}
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")
+	}
+}
diff --git a/service/tier2.go b/service/tier2.go
index 78d228446..d206c1e90 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++ {
@@ -653,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)
 	}
 
@@ -863,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))
@@ -873,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
@@ -911,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})
@@ -921,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)
+}
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())
+}