Fix session teardown races and run cli tests under -race

.changeset/runner-teardown-race.md

@@ -0,0 +1,5 @@
+---
+"@kurrent/gaffer": patch
+---
+
+Session teardown in the debug surfaces no longer races in-flight debug commands. Stopping an MCP run or ending/restarting a DAP debug session could free the native projection session while a step or resume from another goroutine was still executing inside it. That use-after-free could crash the process. The engine runner now refuses new session calls once teardown begins and waits out the in-flight ones before freeing the session.

cli/cmd/dev.go

@@ -637,7 +637,11 @@ func runDevDebug(
 		select {
 		case <-adapter.Ready():
 		case <-adapter.RestartRequested():
-			session.Destroy()
+			// Destroy through the runner (here and on every in-loop teardown
+			// below): the adapter's handlers issue session calls on their own
+			// goroutines, and Runner.Destroy waits those out where a raw
+			// session.Destroy would free the session under them.
+			r.Destroy()
 			session, info, err = engine.CreateSession(proj, true, includeShape)
 			if err != nil {
 				adapter.AckRestart()
@@ -654,7 +658,7 @@ func runDevDebug(
 			adapter.AckRestart()
 			continue
 		case <-ctx.Done():
-			session.Destroy()
+			r.Destroy()
 			return nil
 		}
 
@@ -689,7 +693,7 @@ func runDevDebug(
 		if err != nil {
 			innerCancel()
 			close(iterDone)
-			session.Destroy()
+			r.Destroy()
 			return err
 		}
 
@@ -722,7 +726,7 @@ func runDevDebug(
 		// events, and a restart won't fix it - surface the signal and stop.
 		if authErr := asAuthRequired(srcErr); authErr != nil {
 			writer.WriteAuthRequired(authErr.Env)
-			session.Destroy()
+			r.Destroy()
 			return silent(srcErr)
 		}
 
@@ -758,7 +762,7 @@ func runDevDebug(
 			// projection_errors_seen. The teardown-path branch below
 			// has its own r.Faulted() check; this one matches it.
 			recordProjectionFault(r, obs.onProjectionError)
-			session.Destroy()
+			r.Destroy()
 			session, info, err = engine.CreateSession(proj, true, includeShape)
 			if err != nil {
 				adapter.AckRestart()
@@ -787,7 +791,7 @@ func runDevDebug(
 
 		adapter.SendTerminated()
 		if runErr := finalizeRun(ctx, caughtUp, srcErr, r, os.Stderr); runErr != nil {
-			session.Destroy()
+			r.Destroy()
 			if code, ok := runErrorCode(runErr); ok {
 				writer.WriteRunError(code, runErr.Error())
 				return silent(runErr)
@@ -800,7 +804,7 @@ func runDevDebug(
 			fmt.Fprintf(os.Stderr, "warning: reading projection state: %v\n", stateErr)
 		}
 		writer.WriteSummary(r.Stats(), summary)
-		session.Destroy()
+		r.Destroy()
 		if r.Faulted() {
 			lastErr := r.LastError()
 			if lastErr != nil {

cli/internal/dap/adapter_test.go

@@ -55,6 +55,13 @@ func mustSetupDebugSessionUnbound(t *testing.T, source string) (*DebugAdapter, *
 			OnBreak: adapter.HandleBreak,
 		},
 	})
+	// Tear down through the Runner, which waits out in-flight session calls -
+	// the adapter's handlers issue step verbs on their own goroutines, and a
+	// raw session.Destroy here raced them (caught by the race detector). The
+	// session.Destroy cleanup above stays as the failure-path backstop for a
+	// t.Fatal between session and runner creation; it no-ops once the runner
+	// has destroyed the session. Cleanups run LIFO, so the runner goes first.
+	t.Cleanup(runner.Destroy)
 
 	handler := adapter.Handler()
 	srv, err := NewServer("127.0.0.1:0", handler)
@@ -335,6 +342,7 @@ func mustSetupDebugSessionWithHistory(t *testing.T) (*DebugAdapter, *engine.Runn
 	if err != nil {
 		t.Fatal(err)
 	}
+	t.Cleanup(func() { _ = store.Close() })
 
 	adapter := NewDebugAdapter(session, "/tmp/test/projection.js", "/tmp/test")
 	runner := engine.NewRunner(engine.RunnerConfig{
@@ -349,6 +357,9 @@ func mustSetupDebugSessionWithHistory(t *testing.T) (*DebugAdapter, *engine.Runn
 			OnBreak: adapter.HandleBreak,
 		},
 	})
+	// See mustSetupDebugSessionUnbound: teardown goes through the Runner so
+	// in-flight step verbs from the adapter's goroutines are waited out.
+	t.Cleanup(runner.Destroy)
 	adapter.SetRunner(runner)
 
 	handler := adapter.Handler()

cli/internal/engine/runner.go

@@ -63,10 +63,25 @@ type RunnerConfig struct {
 //     only to snapshot Runner fields the FFI call needs (e.g. the partition
 //     set), then released before crossing into the session.
 //
+// Teardown adds a third guarantee: every session-crossing method - including
+// ProcessOne - registers with r.ops (via beginSessionOp, or directly under
+// r.mu), and Destroy sets closed (refusing new ops), drains, then waits for
+// r.ops before freeing the session. Without it, Destroy races the in-flight
+// FFI call of a step verb issued from another goroutine - and two of those
+// goroutines are the Runner's own: the drain-resume and break_at auto-step
+// goroutines OnBreak spawns are, by construction, still alive at the moment
+// the feed goroutine they unpark exits, which is exactly when front-end
+// teardowns proceed to Destroy. With ProcessOne registered too, Destroy is
+// safe even against a feed still in flight: the drain unparks it, ops.Wait
+// sees it out, and the source loop's next ProcessOne returns stop. (The
+// OnBreak-spawned goroutines register unconditionally rather than through
+// the closed gate - see the comment there for why that cannot deadlock.)
+//
 // The mutable state is split into two concerns, both guarded by r.mu: run
 // holds what processing produces (stats, partitions, fault, status), control
 // holds the debug pause/break wiring. step is an atomic.Int64 read lock-free
-// (see Step), deliberately outside the mutex.
+// (see Step), deliberately outside the mutex. closed is lifecycle, guarded
+// by r.mu alongside them.
 type Runner struct {
 	feed          FeedFn
 	session       *gafferruntime.Session
@@ -78,10 +93,26 @@ type Runner struct {
 	onDiagnostic  func(code string)
 
 	step atomic.Int64
+	ops  sync.WaitGroup
 
 	mu      sync.Mutex
 	run     runState
 	control debugControl
+	closed  bool
+}
+
+// beginSessionOp registers an in-flight session-crossing call so Destroy can
+// wait for it, pairing with r.ops.Done() at the call's end. Returns false once
+// Destroy has begun: the session may already be freed, so the caller must skip
+// the session call entirely.
+func (r *Runner) beginSessionOp() bool {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	if r.closed {
+		return false
+	}
+	r.ops.Add(1)
+	return true
 }
 
 // runState is the Runner's per-run progress, mutated as events are processed
@@ -147,10 +178,26 @@ func NewRunner(cfg RunnerConfig) *Runner {
 			// step the break_at pause converts into - so the blocked
 			// Feed returns and the feed goroutine can exit. The resume
 			// error is irrelevant: the session is going away.
+			// The two goroutines spawned below register with r.ops directly,
+			// NOT via beginSessionOp: they must run even mid-Destroy (closed
+			// set), because they are what unparks the blocked Feed that
+			// Destroy's ops.Wait is waiting out - gating them on closed would
+			// deadlock the teardown. The unconditional Add is safe: this
+			// callback runs inside the feed goroutine's own registered op
+			// (ProcessOne), so the counter is non-zero and the session cannot
+			// be freed before these goroutines finish. Registration happens
+			// here, before the spawn, so ops.Wait can never pass between the
+			// spawn and a goroutine registering itself. That safety argument
+			// rests on Feed only ever being driven through ProcessOne - a
+			// caller feeding the session directly with debug wired would let
+			// these Adds race ops.Wait at counter zero. (WaitGroup.Go adds
+			// synchronously before spawning, preserving exactly that order.)
 			if r.control.draining {
 				r.control.paused = false
 				r.mu.Unlock()
-				go func() { _ = r.debug.Session.Continue() }()
+				r.ops.Go(func() {
+					_ = r.debug.Session.Continue()
+				})
 				return
 			}
 			if info.Reason == "pause" && r.control.breakAtStep > 0 {
@@ -160,11 +207,11 @@ func NewRunner(cfg RunnerConfig) *Runner {
 				// StepInto blocks on the engine thread that's currently running
 				// this callback. A returned error has no caller here, so route
 				// it through OnError (e.g. MCP's error channel).
-				go func() {
+				r.ops.Go(func() {
 					if err := r.debug.Session.StepInto(); err != nil && r.debug.OnError != nil {
 						r.debug.OnError(err)
 					}
-				}()
+				})
 				return
 			}
 			r.control.paused = true
@@ -177,6 +224,14 @@ func NewRunner(cfg RunnerConfig) *Runner {
 
 func (r *Runner) ProcessOne(eventJSON string) (stop bool) {
 	r.mu.Lock()
+	if r.closed {
+		// Teardown has begun and the session may already be freed: stop the
+		// source loop instead of feeding into it.
+		r.mu.Unlock()
+		return true
+	}
+	r.ops.Add(1)
+	defer r.ops.Done()
 	step := r.step.Add(1)
 	var pauseErr error
 	if r.debug != nil {

cli/internal/engine/runner_debug.go

@@ -31,6 +31,10 @@ func (r *Runner) SetBreakpoints(breakpoints []Breakpoint) ([]*gafferruntime.Snap
 	if r.debug == nil {
 		return nil, errors.New("debug not enabled")
 	}
+	if !r.beginSessionOp() {
+		return nil, gafferruntime.ErrSessionDestroyed
+	}
+	defer r.ops.Done()
 	if err := r.debug.Session.ClearBreakpoints(); err != nil {
 		return nil, fmt.Errorf("clearing breakpoints: %w", err)
 	}
@@ -53,6 +57,10 @@ func (r *Runner) ClearBreakpoints() error {
 	if r.debug == nil {
 		return nil
 	}
+	if !r.beginSessionOp() {
+		return nil
+	}
+	defer r.ops.Done()
 	return r.debug.Session.ClearBreakpoints()
 }
 
@@ -66,8 +74,16 @@ func (r *Runner) doStep(fn func() error) error {
 	// false on error also lets the caller's next Paused() guard reject cleanly
 	// rather than re-entering here and failing again.
 	r.mu.Lock()
+	if r.closed {
+		// Teardown has begun: the session may already be freed, and there is
+		// nothing left to step. No-op, like a step on a never-paused engine.
+		r.mu.Unlock()
+		return nil
+	}
+	r.ops.Add(1)
 	r.control.paused = false
 	r.mu.Unlock()
+	defer r.ops.Done()
 	return fn()
 }
 
@@ -98,6 +114,10 @@ func (r *Runner) Pause() error {
 	if r.debug == nil {
 		return nil
 	}
+	if !r.beginSessionOp() {
+		return nil
+	}
+	defer r.ops.Done()
 	return r.debug.Session.Pause()
 }
 
@@ -113,6 +133,18 @@ func (r *Runner) Pause() error {
 //
 // Safe to call when debug is disabled (no-op) or no breakpoint is set.
 func (r *Runner) Drain() {
+	if !r.beginSessionOp() {
+		// Destroy has begun, and its own drain covers the teardown.
+		return
+	}
+	defer r.ops.Done()
+	r.drain()
+}
+
+// drain is Drain without the session-op registration: Destroy calls it after
+// setting closed (which makes beginSessionOp refuse), sequenced on its own
+// goroutine before the session is freed.
+func (r *Runner) drain() {
 	if r.debug == nil {
 		return
 	}
@@ -137,14 +169,37 @@ func (r *Runner) Drain() {
 	}
 }
 
+// Destroy frees the session after draining and waiting out every in-flight
+// session-crossing call, an in-flight ProcessOne included (see the Runner
+// doc comment's teardown guarantee). Idempotent; a concurrent second call
+// no-ops immediately, returning before the first call's teardown completes.
+// Front-ends should still stop their source loop first as a matter of
+// hygiene, but a straggling feed no longer races the teardown.
+//
+// The wait trusts every registered op to finish. The runtime's debug-command
+// queue can strand a command (an unsynchronized paused check-then-add racing
+// the command loop's exit leaves the command's Done unsignalled), and a
+// stranded verb now blocks Destroy where it previously leaked a goroutine.
+// The window needs concurrent debug verbs on a paused engine; the fix is
+// runtime-side, in the command loop.
+//
+// The history store is NOT closed here: the Runner only borrows it, and in
+// the dev debug loop one store serves every restart iteration's runner.
+// Whoever created the store closes it.
 func (r *Runner) Destroy() {
-	r.Drain()
+	r.mu.Lock()
+	if r.closed {
+		r.mu.Unlock()
+		return
+	}
+	r.closed = true
+	r.mu.Unlock()
+
+	r.drain()
+	r.ops.Wait()
 	if r.session != nil {
 		r.session.Destroy()
 	}
-	if r.history != nil {
-		_ = r.history.Close()
-	}
 }
 
 // Inspection methods. These cross into the runtime session, which is not
@@ -157,27 +212,43 @@ func (r *Runner) Evaluate(expression string) (*gafferruntime.DebugVariable, erro
 	if r.debug == nil {
 		return nil, errors.New("debug not enabled")
 	}
+	if !r.beginSessionOp() {
+		return nil, gafferruntime.ErrSessionDestroyed
+	}
+	defer r.ops.Done()
 	return r.debug.Session.Evaluate(expression)
 }
 
 func (r *Runner) GetCallStack() ([]gafferruntime.DebugCallFrame, error) {
 	if r.debug == nil {
 		return nil, errors.New("debug not enabled")
 	}
+	if !r.beginSessionOp() {
+		return nil, gafferruntime.ErrSessionDestroyed
+	}
+	defer r.ops.Done()
 	return r.debug.Session.GetCallStack()
 }
 
 func (r *Runner) GetScopes(frameID int) ([]gafferruntime.DebugScopeInfo, error) {
 	if r.debug == nil {
 		return nil, errors.New("debug not enabled")
 	}
+	if !r.beginSessionOp() {
+		return nil, gafferruntime.ErrSessionDestroyed
+	}
+	defer r.ops.Done()
 	return r.debug.Session.GetScopes(frameID)
 }
 
 func (r *Runner) GetVariables(variablesReference int) ([]gafferruntime.DebugVariable, error) {
 	if r.debug == nil {
 		return nil, errors.New("debug not enabled")
 	}
+	if !r.beginSessionOp() {
+		return nil, gafferruntime.ErrSessionDestroyed
+	}
+	defer r.ops.Done()
 	return r.debug.Session.GetVariables(variablesReference)
 }
 
@@ -191,8 +262,16 @@ func (r *Runner) CollectState() (StateSummary, error) {
 		return StateSummary{}, nil
 	}
 	r.mu.Lock()
+	if r.closed {
+		// Mirrors the nil-session return: after teardown there is no state
+		// left to read.
+		r.mu.Unlock()
+		return StateSummary{}, nil
+	}
+	r.ops.Add(1)
 	partitions := maps.Clone(r.run.partitions)
 	r.mu.Unlock()
+	defer r.ops.Done()
 	return CollectState(r.session, r.info, partitions)
 }
 
@@ -204,6 +283,11 @@ func (r *Runner) GetPartitionState(partition string) (state *string, result *str
 	if r.session == nil {
 		return nil, nil, nil
 	}
+	if !r.beginSessionOp() {
+		// Mirrors the nil-session return, like CollectState.
+		return nil, nil, nil
+	}
+	defer r.ops.Done()
 	state, err = r.session.GetState(&partition)
 	if err != nil {
 		return nil, nil, fmt.Errorf("reading state for partition %q: %w", partition, err)