fix: serialize Init against Start (#24)

init_start_test.go

@@ -0,0 +1,193 @@
+package main
+
+import (
+	"context"
+	"os"
+	"path/filepath"
+	"slices"
+	"sync"
+	"testing"
+	"time"
+
+	basev0 "github.com/codefly-dev/core/generated/go/codefly/base/v0"
+	runtimev0 "github.com/codefly-dev/core/generated/go/codefly/services/runtime/v0"
+	"github.com/stretchr/testify/require"
+)
+
+// initStartTestRuntime builds on initTestRuntime and adds what an Init/Start
+// overlap needs: a runner environment Start can launch against without a real
+// uv, a cache location for the dependency hashes, and a configuration, so Init
+// still has environment-manager work to do after its runner section — the
+// window a Start can land in.
+func initStartTestRuntime(t *testing.T) (*Runtime, *fakeRunnerEnvironment, *runtimev0.InitRequest) {
+	t.Helper()
+
+	runtime, req := initTestRuntime(t)
+	runtime.cacheLocation = t.TempDir()
+	env := &fakeRunnerEnvironment{}
+	runtime.runnerEnvironment = env
+	req.Configuration = &basev0.Configuration{
+		Origin: "mod/probe",
+		Infos: []*basev0.ConfigurationInformation{{
+			Name:                "postgres",
+			ConfigurationValues: []*basev0.ConfigurationValue{{Key: "url", Value: "postgres://localhost"}},
+		}},
+	}
+	return runtime, env, req
+}
+
+// uvicornProc returns the process Start launched, told apart from the ones Init
+// runs on the same environment by the binary it invokes.
+func uvicornProc(t *testing.T, env *fakeRunnerEnvironment) *fakeProc {
+	t.Helper()
+	for _, proc := range env.started() {
+		if slices.Contains(proc.args, "uvicorn") {
+			return proc
+		}
+	}
+	t.Fatal("Start never launched uvicorn")
+	return nil
+}
+
+// uvicornPort returns the port Start bound the process to.
+func uvicornPort(t *testing.T, env *fakeRunnerEnvironment) string {
+	t.Helper()
+	args := uvicornProc(t, env).args
+	i := slices.Index(args, "--port")
+	require.GreaterOrEqual(t, i, 0, "uvicorn was launched without a port: %v", args)
+	require.Less(t, i+1, len(args), "uvicorn was launched with a bare --port: %v", args)
+	return args[i+1]
+}
+
+// gatingEnv holds Init inside its runner section until the test releases it, so
+// a Start can be issued while Init is demonstrably mid-flight.
+type gatingEnv struct {
+	*fakeRunnerEnvironment
+	entered chan struct{}
+	release chan struct{}
+}
+
+func (e *gatingEnv) Init(context.Context) error {
+	close(e.entered)
+	<-e.release
+	return nil
+}
+
+// TestStartWaitsForInitToPublishItsPorts is the contract: Init publishes the
+// port Start binds uvicorn to, so a Start landing in the middle of an Init runs
+// after it and launches on the published port rather than on whatever the field
+// happened to hold.
+//
+// It is also where `-race` sees the two calls overlap on the state Init
+// publishes. Gating Init inside its runner section is what makes that reliable:
+// released from a bare start gate the two bodies order themselves through
+// runnerMu, and the detector reports nothing at all.
+func TestStartWaitsForInitToPublishItsPorts(t *testing.T) {
+	for range 20 {
+		runInitAgainstStart(t)
+	}
+}
+
+func runInitAgainstStart(t *testing.T) {
+	t.Helper()
+	runtime, fake, req := initStartTestRuntime(t)
+	env := &gatingEnv{fakeRunnerEnvironment: fake, entered: make(chan struct{}), release: make(chan struct{})}
+	runtime.runnerEnvironment = env
+
+	var wg sync.WaitGroup
+	var initResp *runtimev0.InitResponse
+	var initErr error
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		initResp, initErr = runtime.Init(context.Background(), req)
+	}()
+
+	<-env.entered
+
+	startDone := make(chan struct{})
+	var startResp *runtimev0.StartResponse
+	var startErr error
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		startResp, startErr = runtime.Start(context.Background(), &runtimev0.StartRequest{})
+		close(startDone)
+	}()
+
+	// Nothing observable says the Start goroutine reached the lock, so give it
+	// the room an unserialized Start would need. Observing startDone is only
+	// safe on the failing side: receiving from it would order the Start
+	// goroutine ahead of the release below and hide the very race this test
+	// exists to surface, so the good path must fall through the timeout having
+	// synchronized with nothing.
+	select {
+	case <-startDone:
+		t.Fatal("Start ran to completion while Init was still inside its runner section")
+	case <-time.After(50 * time.Millisecond):
+	}
+	close(env.release)
+	wg.Wait()
+
+	require.NoError(t, initErr)
+	require.Equal(t, runtimev0.InitStatus_READY, initResp.GetStatus().GetState(), initResp.GetStatus().GetMessage())
+	require.NoError(t, startErr)
+	require.Equal(t, runtimev0.StartStatus_STARTED, startResp.GetStatus().GetState(), startResp.GetStatus().GetMessage())
+	require.Equal(t, "9999", uvicornPort(t, fake))
+}
+
+// TestInitIsSafeWithConcurrentStartOnAColdRuntime covers the path the gated
+// test cannot reach: the first Init, which runs CreateRunnerEnvironment. That
+// function reads the network mappings and the whole environment manager, and
+// publishes ActiveEnv — none of it exercised once a runner environment is
+// already in place.
+//
+// Start is pointed at a source location that does not exist, so it does all of
+// its reading and then fails in exec rather than launching a real uvicorn.
+func TestInitIsSafeWithConcurrentStartOnAColdRuntime(t *testing.T) {
+	isolateWorkingDir(t)
+	for range 20 {
+		runColdInitAgainstStart(t)
+	}
+}
+
+func runColdInitAgainstStart(t *testing.T) {
+	t.Helper()
+	runtime, req := initTestRuntime(t)
+	// Init creates and publishes the runner environment — the section under
+	// test — and then stops on the missing project file, well before uv.
+	require.NoError(t, os.Remove(filepath.Join(runtime.Service.SourceLocation, "pyproject.toml")))
+	runtime.Service.SourceLocation = filepath.Join(runtime.Service.SourceLocation, "does-not-exist")
+
+	var initResp *runtimev0.InitResponse
+	var initErr error
+	var startResp *runtimev0.StartResponse
+
+	gate := make(chan struct{})
+	var wg sync.WaitGroup
+	wg.Add(2)
+	go func() {
+		defer wg.Done()
+		<-gate
+		initResp, initErr = runtime.Init(context.Background(), req)
+	}()
+	go func() {
+		defer wg.Done()
+		<-gate
+		// Land in the window Init is vulnerable in. Before the environment is
+		// published Start refuses at once, having read none of the state Init
+		// is writing, and the accesses that would race never execute.
+		waitForRunnerEnvironment(runtime)
+		startResp, _ = runtime.Start(context.Background(), &runtimev0.StartRequest{})
+	}()
+	close(gate)
+	wg.Wait()
+
+	require.NoError(t, initErr)
+	require.NotNil(t, initResp, "Init must not panic when a Start lands mid-flight")
+	require.Contains(t, initResp.GetStatus().GetMessage(), "no pyproject.toml",
+		"Init must run past its runner section, or this tests nothing")
+	require.NotNil(t, startResp, "Start must not panic against a concurrent Init")
+	require.DirExists(t, filepath.Join(runtime.Location, ".cache/local"),
+		"Init must have created and published the runner environment")
+}

runtime.go

@@ -73,15 +73,55 @@ type Runtime struct {
 	// runnerCreateMu single-flights the creation of the runner environment.
 	// runnerMu cannot do that job: publishing is cheap but building is a
 	// docker pull, and holding runnerMu across it would park every Stop.
-	// Without it two concurrent Inits each build an environment, one gets
-	// published and the other is left initialized — holding a docker client
-	// and a log stream — with nothing referencing it to shut it down.
+	// Without it two callers each build an environment, one gets published and
+	// the other is left initialized — holding a docker client and a log stream
+	// — with nothing referencing it to shut it down.
+	//
+	// Init no longer reaches it concurrently: initMu below serializes Init
+	// against itself, so runnerEnv's only production caller is single-threaded.
+	// It stays because it guards runnerEnv rather than Init — the invariant
+	// belongs to the function that publishes the field, not to the one caller
+	// that happens to hold a bigger lock today.
 	runnerCreateMu sync.Mutex
 
+	// initMu serializes Init against Start. Init publishes what Start consumes
+	// — the network mappings, port and grpcPort, and the entries it folds into
+	// EnvironmentVariables — none of which runnerMu covers, so an overlapping
+	// Start can otherwise launch uvicorn on a half-written port, or read the
+	// environment manager while Init is appending to it. The manager is core's
+	// resources.EnvironmentVariableManager and carries no synchronization of
+	// its own, so nothing finer than serializing the two calls covers it.
+	//
+	// It is a lock of its own rather than runnerMu because Init's runner
+	// section can pull an image or run `uv sync`: Stop and Destroy take only
+	// runnerMu, so a teardown is never queued behind that, and Init carries the
+	// environment it captured through it. Start has nothing to run against
+	// until Init has published its ports, so Start is the side that waits.
+	// Holding it for the whole of Init also serializes Init against itself.
+	//
+	// LOCK ORDER for all three mutexes on this struct:
+	//
+	//	initMu -> runnerCreateMu -> runnerMu
+	//
+	// Every acquisition follows it and nothing takes them the other way around.
+	// The trap is that runnerMu is the innermost: a helper that already holds
+	// it must not call anything reaching for the other two, and in particular
+	// must not call CreateRunnerEnvironment, which takes runnerMu itself to
+	// publish. Callers reach that through runnerEnv, holding none of the three.
+	initMu sync.Mutex
+
 	// startInputs renders the StartRequest fields the running process was
 	// launched with, so a later Start can tell whether they still match.
 	startInputs string
 
+	// launchInputs renders everything that shapes the launched process: the
+	// StartRequest fields above plus the agent-owned argv state Init publishes.
+	// It answers a different question from startInputs — "is the process that
+	// is running still the one this request describes", rather than "did the
+	// request change" — and the two diverge whenever an Init republishes the
+	// ports without the StartRequest changing.
+	launchInputs string
+
 	// appliedOverrides counts, per key, how many override entries this agent
 	// has handed to the environment manager. The manager appends overrides and
 	// never removes them, so the counts are what lets a withdrawn key be
@@ -175,6 +215,11 @@ func (s *Runtime) resolveRuntimeImage() (*resources.DockerImage, error) {
 // CreateRunnerEnvironment builds the execution environment and the cache
 // location that belongs to it, publishes both under runnerMu and returns them
 // so the caller can keep working against locals.
+//
+// It must be called with runnerMu NOT held — it takes that lock to publish —
+// and through runnerEnv, which single-flights it under runnerCreateMu. Calling
+// it directly from a path already holding runnerMu deadlocks; calling it
+// directly from two goroutines builds two environments and strands one.
 func (s *Runtime) CreateRunnerEnvironment(ctx context.Context) (runners.RunnerEnvironment, string, error) {
 	s.Wool.Debug("creating runner environment in", wool.DirField(s.Identity.WorkspacePath))
 	image, err := s.resolveRuntimeImage()
@@ -330,6 +375,9 @@ func (s *Runtime) Init(ctx context.Context, req *runtimev0.InitRequest) (*runtim
 
 	s.Base.Runtime.LogInitRequest(req)
 
+	s.initMu.Lock()
+	defer s.initMu.Unlock()
+
 	// Init opens a new lifecycle, so a runtime torn down earlier accepts work
 	// again — otherwise Init would succeed and every later Start refuse.
 	s.runnerMu.Lock()
@@ -448,6 +496,17 @@ func (s *Runtime) Start(ctx context.Context, req *runtimev0.StartRequest) (*runt
 	defer s.Wool.Catch()
 	ctx = s.Wool.Inject(ctx)
 
+	s.initMu.Lock()
+	defer s.initMu.Unlock()
+
+	// initMu is not cancellable, and Init can hold it across a docker pull and
+	// a `uv sync`. The process launched below outlives this RPC on a context of
+	// its own, so a Start that only reaches here after its caller gave up would
+	// leave a service running for a request nobody is waiting on.
+	if err := ctx.Err(); err != nil {
+		return s.Base.Runtime.StartError(err)
+	}
+
 	s.runnerMu.Lock()
 	defer s.runnerMu.Unlock()
 
@@ -458,6 +517,7 @@ func (s *Runtime) Start(ctx context.Context, req *runtimev0.StartRequest) (*runt
 	}
 
 	inputs := startInputs(req)
+	launch := launchInputs(inputs, s.port, s.grpcPort, s.FastAPI.Settings.HotReload)
 
 	if s.runner != nil {
 		switch {
@@ -466,10 +526,13 @@ func (s *Runtime) Start(ctx context.Context, req *runtimev0.StartRequest) (*runt
 			// again is the chance to bring the service back, not something to
 			// answer from a stale handle.
 			s.Infof("fastapi app is no longer running, starting it again")
-		case inputs == s.startInputs:
-			// A running process keeps the environment it was launched with and
-			// uvicorn --reload already covers source edits, so an unchanged
-			// request describes exactly what is running: nothing to do.
+		case launch == s.launchInputs:
+			// A running process keeps the environment AND the command line it
+			// was launched with, and uvicorn --reload already covers source
+			// edits, so nothing to do. The request alone cannot answer this:
+			// an Init between two identical Starts can republish the ports,
+			// and the process bound to the old one is not what this request
+			// describes however unchanged the request is.
 			return s.Base.Runtime.StartResponse()
 		default:
 			// Different inputs only reach the service through a new process:
@@ -529,6 +592,7 @@ func (s *Runtime) Start(ctx context.Context, req *runtimev0.StartRequest) (*runt
 
 	s.runner = handle
 	s.startInputs = inputs
+	s.launchInputs = launch
 
 	if s.FastAPI.Settings.HotReload {
 		conf := services.NewWatchConfiguration(requirements)
@@ -722,6 +786,19 @@ func startInputs(req *runtimev0.StartRequest) string {
 	return strings.Join(append(lines, dependencies...), "\n")
 }
 
+// launchInputs extends the StartRequest fingerprint with the state this agent
+// owns rather than receives: the ports Init resolves from the proposed network
+// mappings, and the hot-reload setting. Both reach uvicorn's command line, so a
+// process launched under different values is not serving what the caller asked
+// for even when the StartRequest is byte-identical.
+//
+// It is deliberately separate from startInputs: that one gates applyStartInputs,
+// which must fire only when the request itself changed, because the environment
+// manager appends overrides and endpoints without ever removing any.
+func launchInputs(requestInputs string, port uint16, grpcPort uint16, hotReload bool) string {
+	return fmt.Sprintf("%s\nport %d\ngrpc-port %d\nhot-reload %t", requestInputs, port, grpcPort, hotReload)
+}
+
 // applyStartInputs folds the StartRequest into the environment manager. Callers
 // must only invoke it when the inputs actually changed: the manager appends
 // overrides and endpoints without ever removing any, so re-folding an unchanged