diff --git a/init_start_test.go b/init_start_test.go new file mode 100644 index 0000000..bbf156e --- /dev/null +++ b/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") +} diff --git a/runtime.go b/runtime.go index 3ba70ef..68de2b6 100644 --- a/runtime.go +++ b/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 diff --git a/start_request_test.go b/start_request_test.go index 51850f3..e826cd2 100644 --- a/start_request_test.go +++ b/start_request_test.go @@ -692,3 +692,71 @@ func TestTailWriterKeepsRecentLines(t *testing.T) { require.NoError(t, err) require.Equal(t, "dying words", unterminated.Tail()) } + +// TestStartRelaunchesWhenInitRepublishesThePort pins that the restart decision +// covers the ports Init owns, not just the StartRequest. An Init between two +// identical Starts can move the service's port; the process bound to the old +// one is not what the caller is being told is running. +func TestStartRelaunchesWhenInitRepublishesThePort(t *testing.T) { + ctx := context.Background() + runtime, env := startTestRuntime(t) + + runtime.port = 9000 + resp, err := runtime.Start(ctx, &runtimev0.StartRequest{}) + require.NoError(t, err) + require.Equal(t, runtimev0.StartStatus_STARTED, resp.GetStatus().GetState(), resp.GetStatus().GetMessage()) + require.Len(t, env.started(), 1) + require.Contains(t, env.started()[0].args, "9000") + + // What a re-Init resolving a different network mapping publishes. + runtime.port = 9100 + resp, err = runtime.Start(ctx, &runtimev0.StartRequest{}) + require.NoError(t, err) + require.Equal(t, runtimev0.StartStatus_STARTED, resp.GetStatus().GetState(), resp.GetStatus().GetMessage()) + + procs := env.started() + require.Len(t, procs, 2, "the process on the superseded port must be replaced") + require.True(t, procs[0].isStopped(), "the process on the old port must be stopped") + require.Contains(t, procs[1].args, "9100") +} + +// TestStartDoesNotRefoldAnUnchangedRequest pins the other half: the port change +// above must not be mistaken for a request change. The environment manager +// appends overrides and never removes them, so re-folding an identical request +// would duplicate every entry it carries. +func TestStartDoesNotRefoldAnUnchangedRequest(t *testing.T) { + ctx := context.Background() + runtime, env := startTestRuntime(t) + + req := &runtimev0.StartRequest{Overrides: map[string]string{"LOG_LEVEL": "debug"}} + runtime.port = 9000 + _, err := runtime.Start(ctx, req) + require.NoError(t, err) + + runtime.port = 9100 + _, err = runtime.Start(ctx, req) + require.NoError(t, err) + + procs := env.started() + require.Len(t, procs, 2) + require.Equal(t, []string{"debug"}, processEnv(procs[1], "LOG_LEVEL"), + "an unchanged request must be folded into the environment manager exactly once") + require.Equal(t, 1, runtime.appliedOverrides["LOG_LEVEL"]) +} + +// TestStartRefusesAnAbandonedRequest pins that Start does not launch a process +// for a caller that has already given up. Start can wait on initMu for as long +// as an Init takes — a docker pull, a `uv sync` — and the process it launches +// runs on a context of its own, so it would outlive the request entirely. +func TestStartRefusesAnAbandonedRequest(t *testing.T) { + runtime, env := startTestRuntime(t) + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + resp, err := runtime.Start(ctx, &runtimev0.StartRequest{}) + require.NoError(t, err) + require.Equal(t, runtimev0.StartStatus_ERROR, resp.GetStatus().GetState()) + require.Empty(t, env.started(), "no process may be launched for an abandoned request") + require.Nil(t, currentRunner(runtime)) +}