Skip to content

Commit bdaee69

Browse files
fix: supervise the uvicorn process and make repeated Start safe (#19)
Start launched uvicorn and committed STARTED with nothing watching the process, so an app that died on an import error left the agent reporting a service that was gone. The reuse check was also gated on hot reload: without it a second Start built a second process and overwrote the handle to the first, leaving it running and unreachable on the bound port. The command passed --reload whatever the setting said. Each launch now owns a handle — the proc, the cancel function of the context it runs under, and the tail of its output — and a supervisor goroutine that revokes STARTED when the process exits without being asked to. Stop, Destroy and a replacement cancel before killing, so an intentional exit stays silent, and the supervisor compares handle identity under the runner mutex, so a dead generation cannot change the status of the one that replaced it. Repeated Start is now defined for both hot-reload settings: unchanged inputs against a process confirmed alive are idempotent, changed inputs reap the old process first, and a dead process is restarted instead of answered from a stale handle. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
1 parent ad9e9f4 commit bdaee69

3 files changed

Lines changed: 752 additions & 49 deletions

File tree

‎runtime.go‎

Lines changed: 198 additions & 29 deletions
Original file line numberDiff line numberDiff line change
@@ -3,12 +3,14 @@ package main
33
import (
44
"context"
55
"fmt"
6+
"io"
67
"maps"
78
"os"
89
"path"
910
"slices"
1011
"strings"
1112
"sync"
13+
"sync/atomic"
1214

1315
"github.com/codefly-dev/core/agents/helpers/code"
1416
"github.com/codefly-dev/core/agents/services"
@@ -55,11 +57,11 @@ type Runtime struct {
5557

5658
// internal
5759
runnerEnvironment runners.RunnerEnvironment
58-
runner runners.Proc
60+
runner *runnerHandle
5961

60-
// runnerMu serializes the runner lifecycle. Start and Stop are concurrent
61-
// gRPC handlers and both replace s.runner; without it two overlapping
62-
// Starts can each stop the same process and leave one of their
62+
// runnerMu serializes the runner lifecycle. Start, Stop and the supervisor
63+
// goroutines are concurrent and all replace s.runner; without it two
64+
// overlapping Starts can each stop the same process and leave one of their
6365
// replacements running, unreachable, on the bound port.
6466
runnerMu sync.Mutex
6567

@@ -377,22 +379,29 @@ func (s *Runtime) Start(ctx context.Context, req *runtimev0.StartRequest) (*runt
377379

378380
inputs := startInputs(req)
379381

380-
if s.runner != nil && s.FastAPI.Settings.HotReload {
381-
if inputs == s.startInputs {
382+
if s.runner != nil {
383+
switch {
384+
case !s.runnerAlive(ctx, s.runner):
385+
// The process died between two Starts — the orchestrator asking
386+
// again is the chance to bring the service back, not something to
387+
// answer from a stale handle.
388+
s.Infof("fastapi app is no longer running, starting it again")
389+
case inputs == s.startInputs:
390+
// A running process keeps the environment it was launched with and
391+
// uvicorn --reload already covers source edits, so an unchanged
392+
// request describes exactly what is running: nothing to do.
382393
return s.Base.Runtime.StartResponse()
394+
default:
395+
// Different inputs only reach the service through a new process:
396+
// --reload re-executes the app inside the process it was launched
397+
// in, which keeps that process's environment.
398+
s.Infof("start inputs changed, restarting fastapi app")
383399
}
384-
// Hot reload only covers source: uvicorn --reload re-executes the app
385-
// inside the process it was launched in, so that process keeps the
386-
// environment it was given. A request carrying different inputs
387-
// therefore needs a new process, or it would be accepted and never
388-
// take effect.
389-
s.Infof("start inputs changed, restarting fastapi app")
390-
if err := s.runner.Stop(ctx); err != nil {
400+
if err := s.stopRunner(ctx); err != nil {
391401
return s.Base.Runtime.StartError(err)
392402
}
393-
s.runner = nil
394-
// The restart registers a new watcher below, which would otherwise
395-
// leave this one running with no way to reach it.
403+
// The start below registers a new watcher, which would otherwise leave
404+
// this one running with no way to reach it.
396405
s.Base.StopWatcher()
397406
}
398407

@@ -402,9 +411,7 @@ func (s *Runtime) Start(ctx context.Context, req *runtimev0.StartRequest) (*runt
402411
return s.Base.Runtime.StartError(s.Wool.NewError("runner environment not initialized (Init must run before Start)"))
403412
}
404413

405-
proc, err := s.runnerEnvironment.NewProcess(
406-
"uv", "run", "uvicorn", "src.main:app",
407-
"--reload", "--host", "0.0.0.0", "--port", fmt.Sprintf("%d", s.port))
414+
proc, err := s.runnerEnvironment.NewProcess("uv", uvicornArgs(s.port, s.FastAPI.Settings.HotReload)...)
408415
if err != nil {
409416
return s.Base.Runtime.StartError(err)
410417
}
@@ -420,7 +427,8 @@ func (s *Runtime) Start(ctx context.Context, req *runtimev0.StartRequest) (*runt
420427
return s.Base.Runtime.StartErrorf(err, "getting environment variables")
421428
}
422429

423-
proc.WithOutput(s.Logger)
430+
tail := &tailWriter{max: runnerTailLines}
431+
proc.WithOutput(io.MultiWriter(s.Logger, tail))
424432
proc.WithDir(s.Service.SourceLocation)
425433

426434
proc.WithEnvironmentVariables(ctx, startEnvs...)
@@ -433,7 +441,13 @@ func (s *Runtime) Start(ctx context.Context, req *runtimev0.StartRequest) (*runt
433441
proc.WithEnvironmentVariables(ctx, resources.Env("CODEFLY_GRPC_PORT", s.grpcPort))
434442
}
435443

436-
s.runner = proc
444+
// The process must outlive this RPC, so it runs under a background context
445+
// — cancellable, so Stop, Destroy and a replacement can tell the supervisor
446+
// below that the exit it is about to see was asked for.
447+
runningContext, cancel := context.WithCancel(s.Wool.Inject(context.Background()))
448+
handle := &runnerHandle{proc: proc, cancel: cancel, tail: tail}
449+
450+
s.runner = handle
437451
s.startInputs = inputs
438452

439453
if s.FastAPI.Settings.HotReload {
@@ -444,14 +458,164 @@ func (s *Runtime) Start(ctx context.Context, req *runtimev0.StartRequest) (*runt
444458
}
445459

446460
s.Infof("starting fastapi app via uv")
447-
runningContext := s.Wool.Inject(context.Background())
448-
if err := s.runner.Start(runningContext); err != nil {
461+
if err := proc.Start(runningContext); err != nil {
462+
cancel()
449463
s.runner = nil // failed to start — don't leave a dead proc on the struct
450464
return s.Base.Runtime.StartError(err)
451465
}
452466

467+
// STARTED is recorded before the supervisor is armed: revoking a status
468+
// that was never committed would leave the orchestrator reading STARTED for
469+
// a process that is already gone.
470+
resp, err := s.Base.Runtime.StartResponse()
471+
go s.superviseRunner(runningContext, handle)
472+
453473
s.Wool.Debug("start done")
454-
return s.Base.Runtime.StartResponse()
474+
return resp, err
475+
}
476+
477+
// uvicornArgs renders the `uv run` invocation. --reload is conditional: it
478+
// makes uvicorn fork a supervisor that re-executes the app on source edits,
479+
// which is not what a service configured without hot reload asked for.
480+
func uvicornArgs(port uint16, hotReload bool) []string {
481+
args := []string{"run", "uvicorn", "src.main:app", "--host", "0.0.0.0", "--port", fmt.Sprintf("%d", port)}
482+
if hotReload {
483+
args = append(args, "--reload")
484+
}
485+
return args
486+
}
487+
488+
// runnerHandle is one launched process: the proc itself, the cancel function of
489+
// the context it runs under, and the tail of its output. Handles are never
490+
// reused — every Start that replaces a process installs a new one, and the
491+
// supervisor compares identity to tell whether the exit it observed still
492+
// describes the service.
493+
type runnerHandle struct {
494+
proc runners.Proc
495+
cancel context.CancelFunc
496+
tail *tailWriter
497+
498+
// exited is set by the supervisor as soon as the process is gone, so a
499+
// Start landing before the supervisor reaches the mutex still sees a dead
500+
// generation rather than a live-looking handle.
501+
exited atomic.Bool
502+
}
503+
504+
// runnerAlive reports whether the process behind the handle is still up. The
505+
// runner is asked directly because Start can land in the window between the
506+
// process dying and its supervisor being scheduled. A probe that cannot answer
507+
// counts as dead: the agent must not answer STARTED for a process whose
508+
// liveness it could not confirm.
509+
func (s *Runtime) runnerAlive(ctx context.Context, handle *runnerHandle) bool {
510+
if handle.exited.Load() {
511+
return false
512+
}
513+
running, err := handle.proc.IsRunning(ctx)
514+
if err != nil {
515+
s.Wool.Warn("cannot tell whether the fastapi app is running", wool.ErrField(err))
516+
return false
517+
}
518+
return running
519+
}
520+
521+
// stopRunner ends the current generation. It must be called with runnerMu held.
522+
// Cancelling before the kill is what tells the supervisor the exit was asked
523+
// for; dropping the handle first means even a failed Stop cannot leave a
524+
// generation that still owns the service's status.
525+
func (s *Runtime) stopRunner(ctx context.Context) error {
526+
handle := s.runner
527+
if handle == nil {
528+
return nil
529+
}
530+
s.runner = nil
531+
handle.cancel()
532+
return handle.proc.Stop(ctx)
533+
}
534+
535+
// superviseRunner turns the death of a uvicorn process into a revoked STARTED.
536+
// Start commits the status and returns; without this the agent would keep
537+
// reporting a service that exited on an import error, and `codefly run` has no
538+
// other way to learn about it (it polls Information).
539+
//
540+
// This is process supervision, not application readiness: with --reload the
541+
// uvicorn supervisor survives an application that fails to import, so a live
542+
// process is a weaker claim than a serving one. Readiness stays with the
543+
// orchestrator's probes against the endpoints the service declares.
544+
func (s *Runtime) superviseRunner(ctx context.Context, handle *runnerHandle) {
545+
err := handle.proc.Wait(ctx)
546+
handle.exited.Store(true)
547+
if ctx.Err() != nil {
548+
// Stop, Destroy or a replacement cancelled this generation: the exit
549+
// was asked for.
550+
return
551+
}
552+
s.reportRunnerExit(handle, err)
553+
}
554+
555+
// reportRunnerExit revokes STARTED for an unexpected exit — a non-zero one, and
556+
// a clean one just the same, since nothing asked this process to stop.
557+
func (s *Runtime) reportRunnerExit(handle *runnerHandle, err error) {
558+
s.runnerMu.Lock()
559+
defer s.runnerMu.Unlock()
560+
if s.runner != handle {
561+
// A replacement already owns the service; this generation's death says
562+
// nothing about the process that is running now.
563+
return
564+
}
565+
s.runner = nil
566+
567+
message := "fastapi app exited unexpectedly"
568+
if err != nil {
569+
message = fmt.Sprintf("%s: %v", message, err)
570+
}
571+
if tail := handle.tail.Tail(); tail != "" {
572+
message = fmt.Sprintf("%s\n%s", message, tail)
573+
}
574+
s.Wool.Error(message)
575+
s.Base.Runtime.MarkRunnerExited(s.Wool.NewError("%s", message))
576+
}
577+
578+
// runnerTailLines is how much process output the agent keeps to explain an
579+
// unexpected exit. A Python traceback is the diagnostic that matters here and
580+
// fits well inside it.
581+
const runnerTailLines = 20
582+
583+
// tailWriter mirrors process output into a bounded ring of its most recent
584+
// lines. An exit status alone says nothing about why uvicorn died; the
585+
// traceback it printed on the way out does.
586+
type tailWriter struct {
587+
mu sync.Mutex
588+
lines []string
589+
partial string
590+
max int
591+
}
592+
593+
func (w *tailWriter) Write(p []byte) (int, error) {
594+
w.mu.Lock()
595+
defer w.mu.Unlock()
596+
segments := strings.Split(w.partial+string(p), "\n")
597+
w.partial = segments[len(segments)-1]
598+
for _, line := range segments[:len(segments)-1] {
599+
line = strings.TrimRight(line, "\r")
600+
if strings.TrimSpace(line) == "" {
601+
continue
602+
}
603+
w.lines = append(w.lines, line)
604+
}
605+
if len(w.lines) > w.max {
606+
w.lines = slices.Clone(w.lines[len(w.lines)-w.max:])
607+
}
608+
return len(p), nil
609+
}
610+
611+
func (w *tailWriter) Tail() string {
612+
w.mu.Lock()
613+
defer w.mu.Unlock()
614+
lines := w.lines
615+
if strings.TrimSpace(w.partial) != "" {
616+
lines = append(slices.Clone(lines), w.partial)
617+
}
618+
return strings.Join(lines, "\n")
455619
}
456620

457621
// startInputs renders the StartRequest fields this agent turns into process
@@ -602,11 +766,8 @@ func (s *Runtime) Stop(ctx context.Context, req *runtimev0.StopRequest) (*runtim
602766
defer s.runnerMu.Unlock()
603767

604768
s.Wool.Debug("stopping service")
605-
if s.runner != nil {
606-
if err := s.runner.Stop(ctx); err != nil {
607-
return s.Base.Runtime.StopError(err)
608-
}
609-
s.runner = nil
769+
if err := s.stopRunner(ctx); err != nil {
770+
return s.Base.Runtime.StopError(err)
610771
}
611772
if s.runnerEnvironment != nil {
612773
if err := s.runnerEnvironment.Shutdown(ctx); err != nil {
@@ -626,6 +787,14 @@ func (s *Runtime) Destroy(ctx context.Context, req *runtimev0.DestroyRequest) (*
626787

627788
s.Wool.Debug("Destroying service")
628789

790+
// A Destroy that skips this wipes the cache and tears down the container
791+
// while the service process is still running against them.
792+
s.runnerMu.Lock()
793+
if err := s.stopRunner(ctx); err != nil {
794+
s.Wool.Warn("cannot stop the fastapi app", wool.ErrField(err))
795+
}
796+
s.runnerMu.Unlock()
797+
629798
s.Wool.Debug("removing cache")
630799
if err := shared.EmptyDir(ctx, s.cacheLocation); err != nil {
631800
// Best-effort: a failed cache wipe must NOT short-circuit Destroy and

0 commit comments

Comments
 (0)