From 0cd05d5cc44d0b49dadac8028953af630857f5b4 Mon Sep 17 00:00:00 2001 From: Antoine Toussaint <312536060+antoinetoussaint-byte@users.noreply.github.com> Date: Thu, 24 Sep 2026 19:12:20 -0400 Subject: [PATCH] fix: answer STARTED only once the HTTP endpoint serves Start returned STARTED as soon as `uv run uvicorn` was launched. The orchestrator starts dependents on that status, so a dependent could dial the service several seconds before uvicorn bound its port: uvicorn binds only after the app has imported and its lifespan startup has run. Start now waits, bounded, until an HTTP request to the endpoint address Init resolved gets a response (any status). A process that exits first is a failed Start carrying its output; one that never answers is stopped and reported. A bare TCP accept is not used as the signal, because a container runtime's published port is accepted by the engine's proxy before anything listens behind it. Co-Authored-By: Claude Opus 5.5 (1M context) --- init_test.go | 3 + runtime.go | 112 ++++++++++++++++++++++++++++++++- serve_readiness_test.go | 127 ++++++++++++++++++++++++++++++++++++++ uvicorn_lifecycle_test.go | 30 +++++++++ 4 files changed, 271 insertions(+), 1 deletion(-) create mode 100644 serve_readiness_test.go diff --git a/init_test.go b/init_test.go index cbb89be..060b1d8 100644 --- a/init_test.go +++ b/init_test.go @@ -38,6 +38,9 @@ func initTestRuntime(t *testing.T) (*Runtime, *runtimev0.InitRequest) { runtime := NewRuntime(NewService()) runtime.Location = t.TempDir() runtime.Logger = runtime.Wool + // Init resolves a real endpoint address, but the processes these tests + // start are stand-ins that serve nothing: count them as answering. + runtime.answers = func(context.Context, string) error { return nil } // Load is what normally supplies these; these tests call Init directly. runtime.Identity = resources.ServiceIdentityFromProto(&basev0.ServiceIdentity{ Name: "probe", diff --git a/runtime.go b/runtime.go index b8d0444..f7e86d6 100644 --- a/runtime.go +++ b/runtime.go @@ -6,12 +6,14 @@ import ( "fmt" "io" "maps" + "net/http" "os" "path" "slices" "strings" "sync" "sync/atomic" + "time" "github.com/codefly-dev/core/agents/helpers/code" "github.com/codefly-dev/core/agents/services" @@ -132,6 +134,18 @@ type Runtime struct { port uint16 + // serveAddress is the address Init resolved for the HTTP endpoint, as the + // host reaches it. Start answers STARTED only once an HTTP request to it gets + // a response. Empty (no Init) skips the wait. + serveAddress string + + // serveTimeout bounds that wait; zero means defaultServeTimeout. + serveTimeout time.Duration + + // answers performs one attempt to reach the endpoint; nil means an HTTP + // GET (httpAnswers). Tests whose process is a stand-in replace it. + answers func(ctx context.Context, target string) error + // grpcPort is the mapped port for the service-owned gRPC listener, passed // to the Python process as CODEFLY_GRPC_PORT. Zero when gRPC is disabled. grpcPort uint16 @@ -438,6 +452,7 @@ func (s *Runtime) Init(ctx context.Context, req *runtimev0.InitRequest) (*runtim s.Infof("will run on %s", net.Address) s.port = uint16(net.Port) + s.serveAddress = net.Address if s.FastAPI.GRPCEndpoint != nil { grpcNet, grpcErr := resources.FindNetworkInstanceInNetworkMappings(ctx, s.NetworkMappings, s.FastAPI.GRPCEndpoint, resources.NewNativeNetworkAccess()) @@ -623,6 +638,24 @@ func (s *Runtime) Start(ctx context.Context, req *runtimev0.StartRequest) (*runt return s.Base.Runtime.StartError(err) } + // STARTED is what the orchestrator starts dependents on, so it must mean + // the endpoint answers, not that uv was launched: uvicorn binds only after + // the app has imported and its lifespan startup has run, which for a real + // service is seconds after the process exists. + if err := s.awaitServing(ctx, handle); err != nil { + s.runner = nil + cancel() + if stopErr := handle.proc.Stop(context.WithoutCancel(ctx)); stopErr != nil { + s.Wool.Warn("cannot stop the fastapi app that never served", wool.ErrField(stopErr)) + } + s.Base.StopWatcher() + message := err.Error() + if tail := handle.tail.Tail(); tail != "" { + message = fmt.Sprintf("%s\n%s", message, tail) + } + return s.Base.Runtime.StartError(s.Wool.NewError("%s", message)) + } + // STARTED is recorded before the supervisor is armed: revoking a status // that was never committed would leave the orchestrator reading STARTED for // a process that is already gone. @@ -633,6 +666,82 @@ func (s *Runtime) Start(ctx context.Context, req *runtimev0.StartRequest) (*runt return resp, err } +// defaultServeTimeout bounds how long Start waits for the HTTP endpoint to +// answer. It covers `uv run` resolving the environment, the app import and its +// lifespan startup; a service that has not answered by then is not coming up. +const defaultServeTimeout = 3 * time.Minute + +// servePollInterval is the pause between two attempts to reach the endpoint. +const servePollInterval = 250 * time.Millisecond + +// awaitServing returns once an HTTP request to the service's endpoint gets a +// response — any status: readiness here means the endpoint accepts and serves +// connections, and what the app answers on "/" is its own business. A bare TCP +// accept is not enough: in a container runtime the published port is accepted +// by the engine's proxy before anything listens behind it. It fails when the +// process exits first, when ctx ends, or after the serve timeout. +func (s *Runtime) awaitServing(ctx context.Context, handle *runnerHandle) error { + if s.serveAddress == "" { + return nil + } + target := s.serveAddress + if !strings.Contains(target, "://") { + target = "http://" + target + } + target = strings.TrimRight(target, "/") + "/" + timeout := s.serveTimeout + if timeout <= 0 { + timeout = defaultServeTimeout + } + answers := s.answers + if answers == nil { + answers = httpAnswers + } + deadline := time.NewTimer(timeout) + defer deadline.Stop() + started := time.Now() + var last error + for { + err := answers(ctx, target) + if err == nil { + s.Infof("fastapi app serving on %s after %s", s.serveAddress, time.Since(started).Round(100*time.Millisecond)) + return nil + } + last = err + if !s.runnerAlive(ctx, handle) { + return fmt.Errorf("fastapi app exited before its HTTP endpoint at %s answered", s.serveAddress) + } + select { + case <-ctx.Done(): + return fmt.Errorf("start abandoned before the HTTP endpoint at %s answered: %w", s.serveAddress, ctx.Err()) + case <-deadline.C: + return fmt.Errorf("the HTTP endpoint at %s did not answer within %s: %v", s.serveAddress, timeout, last) + case <-time.After(servePollInterval): + } + } +} + +// serveClient is what httpAnswers probes with: one bounded attempt, and a +// redirect counts as an answer since the endpoint is serving. +var serveClient = &http.Client{ + Timeout: time.Second, + CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse }, +} + +// httpAnswers reports whether one GET to target got an HTTP response. +func httpAnswers(ctx context.Context, target string) error { + request, err := http.NewRequestWithContext(ctx, http.MethodGet, target, nil) + if err != nil { + return err + } + response, err := serveClient.Do(request) + if err != nil { + return err + } + _, _ = io.Copy(io.Discard, io.LimitReader(response.Body, 4<<10)) + return response.Body.Close() +} + // uvicornArgs renders the `uv run` invocation. --reload is conditional: it // makes uvicorn fork a supervisor that re-executes the app on source edits, // which is not what a service configured without hot reload asked for. @@ -698,7 +807,8 @@ func (s *Runtime) stopRunner(ctx context.Context) error { // // This is process supervision, not application readiness: with --reload the // uvicorn supervisor survives an application that fails to import, so a live -// process is a weaker claim than a serving one. Readiness stays with the +// process is a weaker claim than a serving one. Start established once that the +// endpoint answered (awaitServing); readiness after that stays with the // orchestrator's probes against the endpoints the service declares. func (s *Runtime) superviseRunner(ctx context.Context, handle *runnerHandle) { err := handle.proc.Wait(ctx) diff --git a/serve_readiness_test.go b/serve_readiness_test.go new file mode 100644 index 0000000..adc3af5 --- /dev/null +++ b/serve_readiness_test.go @@ -0,0 +1,127 @@ +package main + +import ( + "context" + "errors" + "net" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + runtimev0 "github.com/codefly-dev/core/generated/go/codefly/services/runtime/v0" + "github.com/stretchr/testify/require" +) + +// closedAddress returns a loopback address nothing listens on. +func closedAddress(t *testing.T) string { + t.Helper() + listener, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + address := listener.Addr().String() + require.NoError(t, listener.Close()) + return address +} + +// TestStartedMeansTheEndpointAnswers pins the contract a dependent is started +// on: STARTED is not returned while the endpoint refuses connections, and is +// returned once it answers, whatever the status. +func TestStartedMeansTheEndpointAnswers(t *testing.T) { + runtime, env := startTestRuntime(t) + address := closedAddress(t) + runtime.serveAddress = "http://" + address + + served := make(chan struct{}) + go func() { + time.Sleep(700 * time.Millisecond) + listener, err := net.Listen("tcp", address) + if err != nil { + t.Errorf("cannot bind the late listener: %v", err) + return + } + server := &http.Server{Handler: http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusNotFound) + }), ReadHeaderTimeout: time.Second} + t.Cleanup(func() { _ = server.Close() }) + close(served) + _ = server.Serve(listener) + }() + + began := time.Now() + resp, err := runtime.Start(context.Background(), &runtimev0.StartRequest{}) + require.NoError(t, err) + require.Equal(t, runtimev0.StartStatus_STARTED, resp.GetStatus().GetState(), resp.GetStatus().GetMessage()) + select { + case <-served: + default: + t.Fatal("STARTED was returned before anything listened on the endpoint") + } + require.GreaterOrEqual(t, time.Since(began), 600*time.Millisecond) + require.Len(t, env.started(), 1) + require.False(t, env.started()[0].isStopped()) +} + +// TestStartAcceptsAnyHTTPAnswer: a redirect or an error status is still an +// endpoint serving HTTP. +func TestStartAcceptsAnyHTTPAnswer(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + http.Redirect(w, r, "/docs", http.StatusTemporaryRedirect) + })) + t.Cleanup(server.Close) + + runtime, _ := startTestRuntime(t) + // The carrier may hold host:port without a scheme. + runtime.serveAddress = strings.TrimPrefix(server.URL, "http://") + + resp, err := runtime.Start(context.Background(), &runtimev0.StartRequest{}) + require.NoError(t, err) + require.Equal(t, runtimev0.StartStatus_STARTED, resp.GetStatus().GetState(), resp.GetStatus().GetMessage()) +} + +// TestStartFailsWhenTheAppExitsBeforeServing: a process that dies before its +// endpoint answers is a failed Start carrying what it printed, not a STARTED +// the supervisor revokes later. +func TestStartFailsWhenTheAppExitsBeforeServing(t *testing.T) { + runtime, env := startTestRuntime(t) + runtime.serveAddress = "http://" + closedAddress(t) + + go func() { + for len(env.started()) == 0 || !env.started()[0].isStarted() { + time.Sleep(10 * time.Millisecond) + } + proc := env.started()[0] + _, _ = proc.output.Write([]byte("ModuleNotFoundError: No module named 'src.main'\n")) + proc.die(errors.New("exit status 1")) + }() + + resp, err := runtime.Start(context.Background(), &runtimev0.StartRequest{}) + require.NoError(t, err) + require.Equal(t, runtimev0.StartStatus_ERROR, resp.GetStatus().GetState()) + require.Contains(t, resp.GetStatus().GetMessage(), "exited before its HTTP endpoint") + require.Contains(t, resp.GetStatus().GetMessage(), "ModuleNotFoundError") + require.Nil(t, currentRunner(runtime), "a process that never served must not stay the service's runner") +} + +// TestStartFailsWhenTheEndpointNeverAnswers: a live process whose endpoint +// never answers is stopped and reported, bounded by the serve timeout. +func TestStartFailsWhenTheEndpointNeverAnswers(t *testing.T) { + runtime, env := startTestRuntime(t) + runtime.serveAddress = "http://" + closedAddress(t) + runtime.serveTimeout = 600 * time.Millisecond + + resp, err := runtime.Start(context.Background(), &runtimev0.StartRequest{}) + require.NoError(t, err) + require.Equal(t, runtimev0.StartStatus_ERROR, resp.GetStatus().GetState()) + require.Contains(t, resp.GetStatus().GetMessage(), "did not answer within") + require.Len(t, env.started(), 1) + require.True(t, env.started()[0].isStopped(), "the process that never served must be stopped") + require.Nil(t, currentRunner(runtime)) + + // The next Start is a fresh attempt, not an answer from the failed one. + runtime.serveAddress = "" + resp, err = runtime.Start(context.Background(), &runtimev0.StartRequest{}) + require.NoError(t, err) + require.Equal(t, runtimev0.StartStatus_STARTED, resp.GetStatus().GetState(), resp.GetStatus().GetMessage()) + require.Len(t, env.started(), 2) +} diff --git a/uvicorn_lifecycle_test.go b/uvicorn_lifecycle_test.go index d2c38e2..8cab9d3 100644 --- a/uvicorn_lifecycle_test.go +++ b/uvicorn_lifecycle_test.go @@ -148,6 +148,36 @@ func TestUvicornDeathRevokesStartedAndSourceFixRecovers(t *testing.T) { require.Equal(t, runtimev0.StartStatus_STARTED, runtimeStartState(t, runtime).GetState()) } +// TestStartReturnsOnceUvicornServes is the contract dependents are started on: +// with the endpoint address Init resolves, STARTED means the app answers, not +// that uv was launched. Asserted with no Eventually: the answer must already be +// there when Start returns. +func TestStartReturnsOnceUvicornServes(t *testing.T) { + ctx := context.Background() + source := uvProject(t, servingApp) + runtime := uvicornRuntime(t, source, false) + runtime.serveAddress = fmt.Sprintf("127.0.0.1:%d", runtime.port) + + resp, err := runtime.Start(ctx, &runtimev0.StartRequest{}) + require.NoError(t, err) + require.Equal(t, runtimev0.StartStatus_STARTED, resp.GetStatus().GetState(), resp.GetStatus().GetMessage()) + require.True(t, serves(runtime.port), "STARTED was returned before uvicorn answered") +} + +// TestStartFailsWhenUvicornDiesBeforeServing: an import error is now a failed +// Start carrying the traceback, rather than a STARTED revoked moments later. +func TestStartFailsWhenUvicornDiesBeforeServing(t *testing.T) { + ctx := context.Background() + source := uvProject(t, brokenApp) + runtime := uvicornRuntime(t, source, false) + runtime.serveAddress = fmt.Sprintf("127.0.0.1:%d", runtime.port) + + resp, err := runtime.Start(ctx, &runtimev0.StartRequest{}) + require.NoError(t, err) + require.Equal(t, runtimev0.StartStatus_ERROR, resp.GetStatus().GetState()) + require.Contains(t, resp.GetStatus().GetMessage(), "codefly-lifecycle-boom") +} + // TestRepeatedStartWithoutHotReloadOwnsOneProcess pins process ownership for // the setting that used to skip the reuse check: an unchanged request must keep // the process that is serving, and a changed one must reap it before binding