fix: answer STARTED only once the HTTP endpoint serves

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",

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)

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)
+}

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