Fix session teardown races and run cli tests under -race

.changeset/recreate-settle-retry.md

@@ -0,0 +1,5 @@
+---
+"@kurrent/gaffer": patch
+---
+
+`gaffer recreate` (and the `deploy_recreate` MCP tool) no longer strands a projection when the server is slow to settle the delete. KurrentDB deletes projections asynchronously, so the rebuild's create could bounce off the still-registered name with a Conflict and leave the projection deleted but not recreated. The create now retries over a ten-second settle window before giving up with the recovery instructions.

.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/cmd/recreate.go

@@ -135,7 +135,8 @@ func runRecreate(cmd *cobra.Command, name string, opts recreateOpts) error {
 	// recovery messages, live in internal/deploy (shared in shape with deploy's
 	// rebuild); here we only bind each step to the client and its option mapping.
 	if err := deploy.Recreate(ctx, name, deploy.RecreateSteps{
-		Disable: func(ctx context.Context) error { return r.Disable(ctx, name) },
+		RetryCreate: remote.IsCreateConflict,
+		Disable:     func(ctx context.Context) error { return r.Disable(ctx, name) },
 		Delete: func(ctx context.Context) error {
 			return r.Delete(ctx, name, remote.DeleteOptions{
 				DeleteStateStream:      true,

cli/cmd/status_integration_test.go

@@ -8,6 +8,7 @@ import (
 	"os"
 	"path/filepath"
 	"testing"
+	"time"
 
 	"github.com/kurrent-io/gaffer/cli/internal/cliout"
 	"github.com/kurrent-io/gaffer/cli/internal/remote"
@@ -133,7 +134,21 @@ func TestStatus_Integration(t *testing.T) {
 			t.Fatal(err)
 		}
 
-		report := runStatusReportJSON(t)
+		// The drift check degrades to "no drift" by design when the
+		// /info/options read misses its 3s budget (it's advisory - see
+		// nodeOptionsHTTPTimeout), and a race-instrumented CI runner under
+		// full parallel-package load can miss that window. Retry the whole
+		// command a few times so the assertion tests the envelope plumbing,
+		// not one 3-second slot on a loaded runner.
+		var report cliout.StatusReportJSON
+		for attempt := 1; ; attempt++ {
+			report = runStatusReportJSON(t)
+			if len(report.ConfigDrift) > 0 || attempt == 3 {
+				break
+			}
+			t.Logf("attempt %d: configDrift empty (advisory check likely timed out), retrying", attempt)
+			time.Sleep(2 * time.Second)
+		}
 		if len(report.ConfigDrift) != 1 || report.ConfigDrift[0].Knob != "max_state_size" || report.ConfigDrift[0].Local != 1 {
 			t.Fatalf("configDrift = %+v, want the max_state_size divergence", report.ConfigDrift)
 		}

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/deploy/apply.go

@@ -102,6 +102,13 @@ type RecreateSteps struct {
 	Disable Step
 	Delete  Step
 	Create  Step
+	// RetryCreate, when set, marks a Create error as worth retrying over the
+	// settle budget. The server deletes asynchronously - the Delete RPC
+	// returns while the name can still be registered - so an immediate
+	// Create can bounce off the lingering registration (the projections
+	// subsystem replies Conflict; see remote.IsCreateConflict). Optional:
+	// nil means one attempt.
+	RetryCreate func(error) bool
 }
 
 func (s RecreateSteps) validate() error {
@@ -112,11 +119,18 @@ func (s RecreateSteps) validate() error {
 	)
 }
 
+// createSettleBudget bounds how long Recreate keeps retrying a Create that
+// bounces off the deleted projection's lingering registration, on top of the
+// per-attempt RPCTimeout.
+const createSettleBudget = 10 * time.Second
+
 // Recreate destroys a projection and rebuilds it from local config: stop it,
 // delete it (with its state and checkpoint streams), then create it fresh,
 // reprocessing from zero. The destroy precedes the create, so a failure after
 // Delete leaves the projection gone: each step names the recovery rather than a
-// bare error. There's no auto-rollback.
+// bare error. There's no auto-rollback. A Create refused while the server is
+// still settling the delete retries over createSettleBudget (see
+// RecreateSteps.RetryCreate) before giving up with the recovery message.
 func Recreate(ctx context.Context, name string, s RecreateSteps) error {
 	if err := s.validate(); err != nil {
 		return fmt.Errorf("recreate %s: %w", name, err)
@@ -127,7 +141,19 @@ func Recreate(ctx context.Context, name string, s RecreateSteps) error {
 	if err := bound(ctx, s.Delete); err != nil {
 		return fmt.Errorf("could not delete %s before recreating: %w", name, err)
 	}
-	if err := bound(ctx, s.Create); err != nil {
+	err := bound(ctx, s.Create)
+	if s.RetryCreate != nil {
+		deadline := time.Now().Add(createSettleBudget)
+		for delay := 250 * time.Millisecond; err != nil && s.RetryCreate(err) && time.Now().Before(deadline); delay = min(delay*2, 2*time.Second) {
+			select {
+			case <-ctx.Done():
+				return fmt.Errorf("%s was deleted but recreating it failed - re-run gaffer recreate %s, or gaffer deploy %s: %w", name, name, name, ctx.Err())
+			case <-time.After(delay):
+			}
+			err = bound(ctx, s.Create)
+		}
+	}
+	if err != nil {
 		return fmt.Errorf("%s was deleted but recreating it failed - re-run gaffer recreate %s, or gaffer deploy %s: %w", name, name, name, err)
 	}
 	return nil

cli/internal/deploy/apply_test.go

@@ -2,6 +2,7 @@ package deploy
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"strings"
 	"testing"
@@ -129,6 +130,65 @@ func TestRecreateCreateFailureNamesBothRecoveries(t *testing.T) {
 	}
 }
 
+// TestRecreateRetriesCreateWhileSettling pins the delete-settle retry: the
+// server deletes asynchronously, so a create can bounce off the lingering
+// registration (the envelope's Conflict reply) and must be retried rather
+// than stranding the projection deleted-but-not-recreated.
+func TestRecreateRetriesCreateWhileSettling(t *testing.T) {
+	r := &recorder{}
+	attempts := 0
+	steps := r.recreateSteps()
+	create := steps.Create
+	steps.Create = func(ctx context.Context) error {
+		attempts++
+		if attempts < 3 {
+			return errors.New("rpc error: code = Unknown desc = Envelope callback expected Updated, received Conflict instead")
+		}
+		return create(ctx)
+	}
+	steps.RetryCreate = func(err error) bool { return strings.Contains(err.Error(), "received Conflict") }
+
+	if err := Recreate(context.Background(), "orders", steps); err != nil {
+		t.Fatalf("Recreate should have retried through the settle window: %v", err)
+	}
+	if attempts != 3 {
+		t.Errorf("create attempts = %d, want 3 (two conflicts, then success)", attempts)
+	}
+}
+
+// A non-retryable create failure keeps the single-attempt behaviour even with
+// the predicate set.
+func TestRecreateDoesNotRetryOtherCreateFailures(t *testing.T) {
+	r := &recorder{failOn: "create"}
+	steps := r.recreateSteps()
+	steps.RetryCreate = func(err error) bool { return strings.Contains(err.Error(), "received Conflict") }
+
+	if err := Recreate(context.Background(), "orders", steps); err == nil {
+		t.Fatal("expected the create failure to surface")
+	}
+	if got := strings.Join(r.calls, ","); got != "disable,delete,create" {
+		t.Errorf("calls = %q, want a single create attempt", got)
+	}
+}
+
+// A cancelled context ends the retry loop with the recovery message instead
+// of sleeping out the settle budget.
+func TestRecreateRetryStopsOnCancel(t *testing.T) {
+	r := &recorder{}
+	ctx, cancel := context.WithCancel(context.Background())
+	steps := r.recreateSteps()
+	steps.Create = func(context.Context) error {
+		cancel()
+		return errors.New("received Conflict")
+	}
+	steps.RetryCreate = func(err error) bool { return strings.Contains(err.Error(), "received Conflict") }
+
+	err := Recreate(ctx, "orders", steps)
+	if err == nil || !strings.Contains(err.Error(), "orders was deleted but recreating it failed") {
+		t.Fatalf("got %v, want the deleted-but-not-recreated recovery", err)
+	}
+}
+
 func TestNilStepRejectedBeforeAnyCall(t *testing.T) {
 	// A wiring gap (a nil step) must be refused before any step runs: these
 	// sequences are destructive, so a nil caught mid-flight could strand a

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 {