Skip to content

Commit 2e3f245

Browse files
fix: answer STARTED only once the HTTP endpoint serves (#67)
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: Antoine Toussaint <312536060+antoinetoussaint-byte@users.noreply.github.com> Co-authored-by: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
1 parent 833938c commit 2e3f245

4 files changed

Lines changed: 271 additions & 1 deletion

File tree

‎init_test.go‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -38,6 +38,9 @@ func initTestRuntime(t *testing.T) (*Runtime, *runtimev0.InitRequest) {
3838
runtime := NewRuntime(NewService())
3939
runtime.Location = t.TempDir()
4040
runtime.Logger = runtime.Wool
41+
// Init resolves a real endpoint address, but the processes these tests
42+
// start are stand-ins that serve nothing: count them as answering.
43+
runtime.answers = func(context.Context, string) error { return nil }
4144
// Load is what normally supplies these; these tests call Init directly.
4245
runtime.Identity = resources.ServiceIdentityFromProto(&basev0.ServiceIdentity{
4346
Name: "probe",

‎runtime.go‎

Lines changed: 111 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -6,12 +6,14 @@ import (
66
"fmt"
77
"io"
88
"maps"
9+
"net/http"
910
"os"
1011
"path"
1112
"slices"
1213
"strings"
1314
"sync"
1415
"sync/atomic"
16+
"time"
1517

1618
"github.com/codefly-dev/core/agents/helpers/code"
1719
"github.com/codefly-dev/core/agents/services"
@@ -132,6 +134,18 @@ type Runtime struct {
132134

133135
port uint16
134136

137+
// serveAddress is the address Init resolved for the HTTP endpoint, as the
138+
// host reaches it. Start answers STARTED only once an HTTP request to it gets
139+
// a response. Empty (no Init) skips the wait.
140+
serveAddress string
141+
142+
// serveTimeout bounds that wait; zero means defaultServeTimeout.
143+
serveTimeout time.Duration
144+
145+
// answers performs one attempt to reach the endpoint; nil means an HTTP
146+
// GET (httpAnswers). Tests whose process is a stand-in replace it.
147+
answers func(ctx context.Context, target string) error
148+
135149
// grpcPort is the mapped port for the service-owned gRPC listener, passed
136150
// to the Python process as CODEFLY_GRPC_PORT. Zero when gRPC is disabled.
137151
grpcPort uint16
@@ -438,6 +452,7 @@ func (s *Runtime) Init(ctx context.Context, req *runtimev0.InitRequest) (*runtim
438452

439453
s.Infof("will run on %s", net.Address)
440454
s.port = uint16(net.Port)
455+
s.serveAddress = net.Address
441456

442457
if s.FastAPI.GRPCEndpoint != nil {
443458
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
623638
return s.Base.Runtime.StartError(err)
624639
}
625640

641+
// STARTED is what the orchestrator starts dependents on, so it must mean
642+
// the endpoint answers, not that uv was launched: uvicorn binds only after
643+
// the app has imported and its lifespan startup has run, which for a real
644+
// service is seconds after the process exists.
645+
if err := s.awaitServing(ctx, handle); err != nil {
646+
s.runner = nil
647+
cancel()
648+
if stopErr := handle.proc.Stop(context.WithoutCancel(ctx)); stopErr != nil {
649+
s.Wool.Warn("cannot stop the fastapi app that never served", wool.ErrField(stopErr))
650+
}
651+
s.Base.StopWatcher()
652+
message := err.Error()
653+
if tail := handle.tail.Tail(); tail != "" {
654+
message = fmt.Sprintf("%s\n%s", message, tail)
655+
}
656+
return s.Base.Runtime.StartError(s.Wool.NewError("%s", message))
657+
}
658+
626659
// STARTED is recorded before the supervisor is armed: revoking a status
627660
// that was never committed would leave the orchestrator reading STARTED for
628661
// a process that is already gone.
@@ -633,6 +666,82 @@ func (s *Runtime) Start(ctx context.Context, req *runtimev0.StartRequest) (*runt
633666
return resp, err
634667
}
635668

669+
// defaultServeTimeout bounds how long Start waits for the HTTP endpoint to
670+
// answer. It covers `uv run` resolving the environment, the app import and its
671+
// lifespan startup; a service that has not answered by then is not coming up.
672+
const defaultServeTimeout = 3 * time.Minute
673+
674+
// servePollInterval is the pause between two attempts to reach the endpoint.
675+
const servePollInterval = 250 * time.Millisecond
676+
677+
// awaitServing returns once an HTTP request to the service's endpoint gets a
678+
// response — any status: readiness here means the endpoint accepts and serves
679+
// connections, and what the app answers on "/" is its own business. A bare TCP
680+
// accept is not enough: in a container runtime the published port is accepted
681+
// by the engine's proxy before anything listens behind it. It fails when the
682+
// process exits first, when ctx ends, or after the serve timeout.
683+
func (s *Runtime) awaitServing(ctx context.Context, handle *runnerHandle) error {
684+
if s.serveAddress == "" {
685+
return nil
686+
}
687+
target := s.serveAddress
688+
if !strings.Contains(target, "://") {
689+
target = "http://" + target
690+
}
691+
target = strings.TrimRight(target, "/") + "/"
692+
timeout := s.serveTimeout
693+
if timeout <= 0 {
694+
timeout = defaultServeTimeout
695+
}
696+
answers := s.answers
697+
if answers == nil {
698+
answers = httpAnswers
699+
}
700+
deadline := time.NewTimer(timeout)
701+
defer deadline.Stop()
702+
started := time.Now()
703+
var last error
704+
for {
705+
err := answers(ctx, target)
706+
if err == nil {
707+
s.Infof("fastapi app serving on %s after %s", s.serveAddress, time.Since(started).Round(100*time.Millisecond))
708+
return nil
709+
}
710+
last = err
711+
if !s.runnerAlive(ctx, handle) {
712+
return fmt.Errorf("fastapi app exited before its HTTP endpoint at %s answered", s.serveAddress)
713+
}
714+
select {
715+
case <-ctx.Done():
716+
return fmt.Errorf("start abandoned before the HTTP endpoint at %s answered: %w", s.serveAddress, ctx.Err())
717+
case <-deadline.C:
718+
return fmt.Errorf("the HTTP endpoint at %s did not answer within %s: %v", s.serveAddress, timeout, last)
719+
case <-time.After(servePollInterval):
720+
}
721+
}
722+
}
723+
724+
// serveClient is what httpAnswers probes with: one bounded attempt, and a
725+
// redirect counts as an answer since the endpoint is serving.
726+
var serveClient = &http.Client{
727+
Timeout: time.Second,
728+
CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse },
729+
}
730+
731+
// httpAnswers reports whether one GET to target got an HTTP response.
732+
func httpAnswers(ctx context.Context, target string) error {
733+
request, err := http.NewRequestWithContext(ctx, http.MethodGet, target, nil)
734+
if err != nil {
735+
return err
736+
}
737+
response, err := serveClient.Do(request)
738+
if err != nil {
739+
return err
740+
}
741+
_, _ = io.Copy(io.Discard, io.LimitReader(response.Body, 4<<10))
742+
return response.Body.Close()
743+
}
744+
636745
// uvicornArgs renders the `uv run` invocation. --reload is conditional: it
637746
// makes uvicorn fork a supervisor that re-executes the app on source edits,
638747
// which is not what a service configured without hot reload asked for.
@@ -698,7 +807,8 @@ func (s *Runtime) stopRunner(ctx context.Context) error {
698807
//
699808
// This is process supervision, not application readiness: with --reload the
700809
// uvicorn supervisor survives an application that fails to import, so a live
701-
// process is a weaker claim than a serving one. Readiness stays with the
810+
// process is a weaker claim than a serving one. Start established once that the
811+
// endpoint answered (awaitServing); readiness after that stays with the
702812
// orchestrator's probes against the endpoints the service declares.
703813
func (s *Runtime) superviseRunner(ctx context.Context, handle *runnerHandle) {
704814
err := handle.proc.Wait(ctx)

‎serve_readiness_test.go‎

Lines changed: 127 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,127 @@
1+
package main
2+
3+
import (
4+
"context"
5+
"errors"
6+
"net"
7+
"net/http"
8+
"net/http/httptest"
9+
"strings"
10+
"testing"
11+
"time"
12+
13+
runtimev0 "github.com/codefly-dev/core/generated/go/codefly/services/runtime/v0"
14+
"github.com/stretchr/testify/require"
15+
)
16+
17+
// closedAddress returns a loopback address nothing listens on.
18+
func closedAddress(t *testing.T) string {
19+
t.Helper()
20+
listener, err := net.Listen("tcp", "127.0.0.1:0")
21+
require.NoError(t, err)
22+
address := listener.Addr().String()
23+
require.NoError(t, listener.Close())
24+
return address
25+
}
26+
27+
// TestStartedMeansTheEndpointAnswers pins the contract a dependent is started
28+
// on: STARTED is not returned while the endpoint refuses connections, and is
29+
// returned once it answers, whatever the status.
30+
func TestStartedMeansTheEndpointAnswers(t *testing.T) {
31+
runtime, env := startTestRuntime(t)
32+
address := closedAddress(t)
33+
runtime.serveAddress = "http://" + address
34+
35+
served := make(chan struct{})
36+
go func() {
37+
time.Sleep(700 * time.Millisecond)
38+
listener, err := net.Listen("tcp", address)
39+
if err != nil {
40+
t.Errorf("cannot bind the late listener: %v", err)
41+
return
42+
}
43+
server := &http.Server{Handler: http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
44+
w.WriteHeader(http.StatusNotFound)
45+
}), ReadHeaderTimeout: time.Second}
46+
t.Cleanup(func() { _ = server.Close() })
47+
close(served)
48+
_ = server.Serve(listener)
49+
}()
50+
51+
began := time.Now()
52+
resp, err := runtime.Start(context.Background(), &runtimev0.StartRequest{})
53+
require.NoError(t, err)
54+
require.Equal(t, runtimev0.StartStatus_STARTED, resp.GetStatus().GetState(), resp.GetStatus().GetMessage())
55+
select {
56+
case <-served:
57+
default:
58+
t.Fatal("STARTED was returned before anything listened on the endpoint")
59+
}
60+
require.GreaterOrEqual(t, time.Since(began), 600*time.Millisecond)
61+
require.Len(t, env.started(), 1)
62+
require.False(t, env.started()[0].isStopped())
63+
}
64+
65+
// TestStartAcceptsAnyHTTPAnswer: a redirect or an error status is still an
66+
// endpoint serving HTTP.
67+
func TestStartAcceptsAnyHTTPAnswer(t *testing.T) {
68+
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
69+
http.Redirect(w, r, "/docs", http.StatusTemporaryRedirect)
70+
}))
71+
t.Cleanup(server.Close)
72+
73+
runtime, _ := startTestRuntime(t)
74+
// The carrier may hold host:port without a scheme.
75+
runtime.serveAddress = strings.TrimPrefix(server.URL, "http://")
76+
77+
resp, err := runtime.Start(context.Background(), &runtimev0.StartRequest{})
78+
require.NoError(t, err)
79+
require.Equal(t, runtimev0.StartStatus_STARTED, resp.GetStatus().GetState(), resp.GetStatus().GetMessage())
80+
}
81+
82+
// TestStartFailsWhenTheAppExitsBeforeServing: a process that dies before its
83+
// endpoint answers is a failed Start carrying what it printed, not a STARTED
84+
// the supervisor revokes later.
85+
func TestStartFailsWhenTheAppExitsBeforeServing(t *testing.T) {
86+
runtime, env := startTestRuntime(t)
87+
runtime.serveAddress = "http://" + closedAddress(t)
88+
89+
go func() {
90+
for len(env.started()) == 0 || !env.started()[0].isStarted() {
91+
time.Sleep(10 * time.Millisecond)
92+
}
93+
proc := env.started()[0]
94+
_, _ = proc.output.Write([]byte("ModuleNotFoundError: No module named 'src.main'\n"))
95+
proc.die(errors.New("exit status 1"))
96+
}()
97+
98+
resp, err := runtime.Start(context.Background(), &runtimev0.StartRequest{})
99+
require.NoError(t, err)
100+
require.Equal(t, runtimev0.StartStatus_ERROR, resp.GetStatus().GetState())
101+
require.Contains(t, resp.GetStatus().GetMessage(), "exited before its HTTP endpoint")
102+
require.Contains(t, resp.GetStatus().GetMessage(), "ModuleNotFoundError")
103+
require.Nil(t, currentRunner(runtime), "a process that never served must not stay the service's runner")
104+
}
105+
106+
// TestStartFailsWhenTheEndpointNeverAnswers: a live process whose endpoint
107+
// never answers is stopped and reported, bounded by the serve timeout.
108+
func TestStartFailsWhenTheEndpointNeverAnswers(t *testing.T) {
109+
runtime, env := startTestRuntime(t)
110+
runtime.serveAddress = "http://" + closedAddress(t)
111+
runtime.serveTimeout = 600 * time.Millisecond
112+
113+
resp, err := runtime.Start(context.Background(), &runtimev0.StartRequest{})
114+
require.NoError(t, err)
115+
require.Equal(t, runtimev0.StartStatus_ERROR, resp.GetStatus().GetState())
116+
require.Contains(t, resp.GetStatus().GetMessage(), "did not answer within")
117+
require.Len(t, env.started(), 1)
118+
require.True(t, env.started()[0].isStopped(), "the process that never served must be stopped")
119+
require.Nil(t, currentRunner(runtime))
120+
121+
// The next Start is a fresh attempt, not an answer from the failed one.
122+
runtime.serveAddress = ""
123+
resp, err = runtime.Start(context.Background(), &runtimev0.StartRequest{})
124+
require.NoError(t, err)
125+
require.Equal(t, runtimev0.StartStatus_STARTED, resp.GetStatus().GetState(), resp.GetStatus().GetMessage())
126+
require.Len(t, env.started(), 2)
127+
}

‎uvicorn_lifecycle_test.go‎

Lines changed: 30 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -148,6 +148,36 @@ func TestUvicornDeathRevokesStartedAndSourceFixRecovers(t *testing.T) {
148148
require.Equal(t, runtimev0.StartStatus_STARTED, runtimeStartState(t, runtime).GetState())
149149
}
150150

151+
// TestStartReturnsOnceUvicornServes is the contract dependents are started on:
152+
// with the endpoint address Init resolves, STARTED means the app answers, not
153+
// that uv was launched. Asserted with no Eventually: the answer must already be
154+
// there when Start returns.
155+
func TestStartReturnsOnceUvicornServes(t *testing.T) {
156+
ctx := context.Background()
157+
source := uvProject(t, servingApp)
158+
runtime := uvicornRuntime(t, source, false)
159+
runtime.serveAddress = fmt.Sprintf("127.0.0.1:%d", runtime.port)
160+
161+
resp, err := runtime.Start(ctx, &runtimev0.StartRequest{})
162+
require.NoError(t, err)
163+
require.Equal(t, runtimev0.StartStatus_STARTED, resp.GetStatus().GetState(), resp.GetStatus().GetMessage())
164+
require.True(t, serves(runtime.port), "STARTED was returned before uvicorn answered")
165+
}
166+
167+
// TestStartFailsWhenUvicornDiesBeforeServing: an import error is now a failed
168+
// Start carrying the traceback, rather than a STARTED revoked moments later.
169+
func TestStartFailsWhenUvicornDiesBeforeServing(t *testing.T) {
170+
ctx := context.Background()
171+
source := uvProject(t, brokenApp)
172+
runtime := uvicornRuntime(t, source, false)
173+
runtime.serveAddress = fmt.Sprintf("127.0.0.1:%d", runtime.port)
174+
175+
resp, err := runtime.Start(ctx, &runtimev0.StartRequest{})
176+
require.NoError(t, err)
177+
require.Equal(t, runtimev0.StartStatus_ERROR, resp.GetStatus().GetState())
178+
require.Contains(t, resp.GetStatus().GetMessage(), "codefly-lifecycle-boom")
179+
}
180+
151181
// TestRepeatedStartWithoutHotReloadOwnsOneProcess pins process ownership for
152182
// the setting that used to skip the reuse check: an unchanged request must keep
153183
// the process that is serving, and a changed one must reap it before binding

0 commit comments

Comments
 (0)