Skip to content

Commit b73a19b

Browse files
fix: close the gaps in StartRequest handling (#17)
Review of the first pass found six defects, each fixed at its cause. The fixture was written only when non-empty, which made it a latch: a request that withdrew it left the previous selector exported. The manager already falls back to the environment-level fixture on empty, so passing the value through unconditionally is the correct call and is what makes a withdrawal revert. Secrets were re-applied to the process after the resolved environment, so a `--set` naming a secret key was silently shadowed. All() already carries the secret values, making that second application a duplicate; dropping it puts secrets under the same override resolution as every other key. Overrides the manager accumulated could not be withdrawn either. The agent now counts the entries it has handed over, so a key absent from the current request has exactly its stale entries filtered out and the configuration value underneath is what the process sees. The restart decision compared the whole message, so a difference in Specs — a field this agent documents as unread — bounced a healthy process. It now compares a rendering of the honored fields only, which also drops the direct protobuf dependency. Start and Stop are concurrent gRPC handlers that both replace s.runner; overlapping restarts could each stop the same process and orphan a replacement on the bound port. Both now hold a mutex over the runner lifecycle. Dependency mappings whose instances match no access kind were dropped without a word, surfacing much later as a missing variable inside the service. Start now warns, naming the endpoints. Also narrowed the override resolution to the keys overrides actually claim (it previously collapsed every duplicate in the process environment), made the manager fold idempotent so a retry cannot grow it, moved the fold after the process is created, and gave endpoint failures their own error message. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
1 parent 2c71003 commit b73a19b

3 files changed

Lines changed: 340 additions & 46 deletions

File tree

‎go.mod‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -7,7 +7,6 @@ require (
77
github.com/codefly-dev/service-python v0.0.52
88
github.com/stretchr/testify v1.11.1
99
google.golang.org/grpc v1.83.0
10-
google.golang.org/protobuf v1.36.12
1110
gopkg.in/yaml.v3 v3.0.1
1211
)
1312

@@ -126,6 +125,7 @@ require (
126125
golang.org/x/text v0.41.0 // indirect
127126
google.golang.org/genproto/googleapis/api v0.0.0-20260817212433-ac3dfec99bb1 // indirect
128127
google.golang.org/genproto/googleapis/rpc v0.0.0-20260817212433-ac3dfec99bb1 // indirect
128+
google.golang.org/protobuf v1.36.12 // indirect
129129
gopkg.in/evanphx/json-patch.v4 v4.13.0 // indirect
130130
gopkg.in/warnings.v0 v0.1.2 // indirect
131131
k8s.io/kube-openapi v0.0.0-20260721132016-d427ff9ee9ad // indirect

‎runtime.go‎

Lines changed: 156 additions & 43 deletions
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,7 @@ import (
88
"path"
99
"slices"
1010
"strings"
11+
"sync"
1112

1213
"github.com/codefly-dev/core/agents/helpers/code"
1314
"github.com/codefly-dev/core/agents/services"
@@ -22,8 +23,6 @@ import (
2223
"github.com/codefly-dev/core/wool"
2324

2425
pythonruntime "github.com/codefly-dev/service-python/pkg/runtime"
25-
26-
"google.golang.org/protobuf/proto"
2726
)
2827

2928
// Runtime is the FastAPI specialization of the generic Python Runtime.
@@ -58,9 +57,21 @@ type Runtime struct {
5857
runnerEnvironment runners.RunnerEnvironment
5958
runner runners.Proc
6059

61-
// startRequest is what the running process was launched with, kept so a
62-
// later Start can tell whether the inputs it carries still match.
63-
startRequest *runtimev0.StartRequest
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
63+
// replacements running, unreachable, on the bound port.
64+
runnerMu sync.Mutex
65+
66+
// startInputs renders the StartRequest fields the running process was
67+
// launched with, so a later Start can tell whether they still match.
68+
startInputs string
69+
70+
// appliedOverrides counts, per key, how many override entries this agent
71+
// has handed to the environment manager. The manager appends overrides and
72+
// never removes them, so the counts are what lets a withdrawn key be
73+
// filtered back out of the process environment.
74+
appliedOverrides map[string]int
6475

6576
port uint16
6677

@@ -359,10 +370,15 @@ func (s *Runtime) Start(ctx context.Context, req *runtimev0.StartRequest) (*runt
359370
defer s.Wool.Catch()
360371
ctx = s.Wool.Inject(ctx)
361372

373+
s.runnerMu.Lock()
374+
defer s.runnerMu.Unlock()
375+
362376
s.Base.Runtime.LogStartRequest(req)
363377

378+
inputs := startInputs(req)
379+
364380
if s.runner != nil && s.FastAPI.Settings.HotReload {
365-
if proto.Equal(req, s.startRequest) {
381+
if inputs == s.startInputs {
366382
return s.Base.Runtime.StartResponse()
367383
}
368384
// Hot reload only covers source: uvicorn --reload re-executes the app
@@ -386,32 +402,39 @@ func (s *Runtime) Start(ctx context.Context, req *runtimev0.StartRequest) (*runt
386402
return s.Base.Runtime.StartError(s.Wool.NewError("runner environment not initialized (Init must run before Start)"))
387403
}
388404

389-
startEnvs, err := s.startEnvironment(ctx, req)
390-
if err != nil {
391-
return s.Base.Runtime.StartErrorf(err, "getting environment variables")
392-
}
393-
394405
proc, err := s.runnerEnvironment.NewProcess(
395406
"uv", "run", "uvicorn", "src.main:app",
396407
"--reload", "--host", "0.0.0.0", "--port", fmt.Sprintf("%d", s.port))
397408
if err != nil {
398409
return s.Base.Runtime.StartError(err)
399410
}
400411

412+
if inputs != s.startInputs {
413+
if err := s.applyStartInputs(ctx, req); err != nil {
414+
return s.Base.Runtime.StartErrorf(err, "applying start request")
415+
}
416+
}
417+
418+
startEnvs, err := s.processEnvironment(req)
419+
if err != nil {
420+
return s.Base.Runtime.StartErrorf(err, "getting environment variables")
421+
}
422+
401423
proc.WithOutput(s.Logger)
402424
proc.WithDir(s.Service.SourceLocation)
403425

404426
proc.WithEnvironmentVariables(ctx, startEnvs...)
405-
proc.WithEnvironmentVariables(ctx, s.EnvironmentVariables.Secrets()...)
406427
if s.grpcPort != 0 {
407428
// The FastAPI lifespan boots the grpc.aio listener on this port
408429
// (src/rpc/server.py); in container/k8s the app falls back to the
409-
// standard gRPC port when the variable is unset.
430+
// standard gRPC port when the variable is unset. The agent owns the
431+
// mapped port, so it is applied after the overrides and wins over a
432+
// `--set` naming the same key.
410433
proc.WithEnvironmentVariables(ctx, resources.Env("CODEFLY_GRPC_PORT", s.grpcPort))
411434
}
412435

413436
s.runner = proc
414-
s.startRequest = req
437+
s.startInputs = inputs
415438

416439
if s.FastAPI.Settings.HotReload {
417440
conf := services.NewWatchConfiguration(requirements)
@@ -431,52 +454,139 @@ func (s *Runtime) Start(ctx context.Context, req *runtimev0.StartRequest) (*runt
431454
return s.Base.Runtime.StartResponse()
432455
}
433456

434-
// startEnvironment folds the StartRequest into the environment manager and
435-
// returns what the uvicorn process runs with: the per-service overrides of
436-
// `codefly run --set`, the fixture selector, and the resolved addresses of the
437-
// endpoints this service depends on. req.Specs carries agent-specific runtime
438-
// knobs and fastapi defines none, so nothing is read from it.
439-
func (s *Runtime) startEnvironment(ctx context.Context, req *runtimev0.StartRequest) ([]*resources.EnvironmentVariable, error) {
457+
// startInputs renders the StartRequest fields this agent turns into process
458+
// environment: the per-service overrides of `codefly run --set`, the fixture
459+
// selector, and the addresses of the endpoints this service depends on.
460+
// req.Specs carries agent-specific runtime knobs and fastapi defines none, so
461+
// it is deliberately absent — a request differing only there produces an
462+
// identical process and must not cost a restart.
463+
func startInputs(req *runtimev0.StartRequest) string {
464+
lines := []string{fmt.Sprintf("fixture %s", req.GetFixture())}
465+
for _, key := range slices.Sorted(maps.Keys(req.GetOverrides())) {
466+
lines = append(lines, fmt.Sprintf("override %s=%s", key, req.GetOverrides()[key]))
467+
}
468+
var dependencies []string
469+
for _, mapping := range req.GetDependenciesNetworkMappings() {
470+
endpoint := mapping.GetEndpoint()
471+
for _, instance := range mapping.GetInstances() {
472+
dependencies = append(dependencies, fmt.Sprintf("dependency %s/%s/%s/%s %s %s",
473+
endpoint.GetModule(), endpoint.GetService(), endpoint.GetName(), endpoint.GetApi(),
474+
instance.GetAccess().GetKind(), instance.GetAddress()))
475+
}
476+
}
477+
slices.Sort(dependencies)
478+
return strings.Join(append(lines, dependencies...), "\n")
479+
}
480+
481+
// applyStartInputs folds the StartRequest into the environment manager. Callers
482+
// must only invoke it when the inputs actually changed: the manager appends
483+
// overrides and endpoints without ever removing any, so re-folding an unchanged
484+
// request would grow those slices on every restart.
485+
func (s *Runtime) applyStartInputs(ctx context.Context, req *runtimev0.StartRequest) error {
440486
overrides := req.GetOverrides()
441487
s.EnvironmentVariables.AddOverrides(overrides)
442-
443-
if fixture := req.GetFixture(); fixture != "" {
444-
s.EnvironmentVariables.SetFixture(fixture)
488+
if s.appliedOverrides == nil {
489+
s.appliedOverrides = make(map[string]int, len(overrides))
490+
}
491+
for key := range overrides {
492+
s.appliedOverrides[key]++
445493
}
446494

447-
if err := s.EnvironmentVariables.AddEndpoints(ctx, req.GetDependenciesNetworkMappings(),
448-
resources.NetworkAccessFromRuntimeContext(s.Base.Runtime.RuntimeContext)); err != nil {
449-
return nil, err
495+
// Unconditional: the manager falls back to the environment-level fixture
496+
// when this one is empty, so passing "" is how a withdrawn fixture reverts.
497+
s.EnvironmentVariables.SetFixture(req.GetFixture())
498+
499+
networkAccess := resources.NetworkAccessFromRuntimeContext(s.Base.Runtime.RuntimeContext)
500+
if unresolved := unresolvedDependencies(req.GetDependenciesNetworkMappings(), networkAccess); len(unresolved) > 0 {
501+
// AddEndpoints drops these silently, which surfaces much later as a
502+
// missing variable inside the service rather than as a start problem.
503+
s.Wool.Warn("dependency endpoints carry no address reachable from this runtime context and are not exported",
504+
wool.Field("access", networkAccess.GetKind()),
505+
wool.Field("endpoints", strings.Join(unresolved, ", ")))
506+
}
507+
if err := s.EnvironmentVariables.AddEndpoints(ctx, req.GetDependenciesNetworkMappings(), networkAccess); err != nil {
508+
return s.Wool.Wrapf(err, "cannot add dependency endpoints")
450509
}
451510

452511
s.EnvironmentVariables.SetRunning()
512+
return nil
513+
}
514+
515+
// unresolvedDependencies returns the dependency mappings that carry no instance
516+
// reachable from networkAccess, named module/service/endpoint. It mirrors the
517+
// filter AddEndpoints applies, which reports nothing when a mapping contributes
518+
// no address.
519+
func unresolvedDependencies(mappings []*basev0.NetworkMapping, networkAccess *basev0.NetworkAccess) []string {
520+
var unresolved []string
521+
for _, mapping := range mappings {
522+
if mapping == nil {
523+
continue
524+
}
525+
matched := false
526+
for _, instance := range mapping.GetInstances() {
527+
if instance.GetAccess().GetKind() == networkAccess.GetKind() {
528+
matched = true
529+
break
530+
}
531+
}
532+
if !matched {
533+
endpoint := mapping.GetEndpoint()
534+
unresolved = append(unresolved, fmt.Sprintf("%s/%s/%s",
535+
endpoint.GetModule(), endpoint.GetService(), endpoint.GetName()))
536+
}
537+
}
538+
return unresolved
539+
}
453540

541+
// processEnvironment returns what the service process runs with. All() already
542+
// carries the secret configuration values alongside the plain ones, so this is
543+
// the complete set and the override resolution below covers secrets too.
544+
func (s *Runtime) processEnvironment(req *runtimev0.StartRequest) ([]*resources.EnvironmentVariable, error) {
454545
envs, err := s.EnvironmentVariables.All()
455546
if err != nil {
456547
return nil, err
457548
}
458-
return withOverridesLast(envs, overrides), nil
549+
withdrawn := make(map[string]int, len(s.appliedOverrides))
550+
for key, count := range s.appliedOverrides {
551+
if _, current := req.GetOverrides()[key]; !current {
552+
withdrawn[key] = count
553+
}
554+
}
555+
return withOverrides(envs, req.GetOverrides(), withdrawn), nil
459556
}
460557

461-
// withOverridesLast re-appends the overrides at the end of envs and collapses
462-
// the result to a single entry per key, keeping the last value. The
463-
// environment manager emits overrides ahead of the configuration-derived
464-
// variables, so without the reordering a configuration would shadow a `--set`
465-
// naming the same key.
466-
func withOverridesLast(envs []*resources.EnvironmentVariable, overrides map[string]string) []*resources.EnvironmentVariable {
467-
for _, key := range slices.Sorted(maps.Keys(overrides)) {
468-
envs = append(envs, resources.Env(key, overrides[key]))
469-
}
470-
seen := make(map[string]bool, len(envs))
471-
resolved := make([]*resources.EnvironmentVariable, 0, len(envs))
472-
for i := len(envs) - 1; i >= 0; i-- {
473-
if seen[envs[i].Key] {
558+
// withOverrides resolves the keys this agent's overrides participate in, and
559+
// leaves every other entry exactly as the environment manager emitted it.
560+
//
561+
// The manager emits overrides ahead of the configuration-derived variables
562+
// (secrets included), so a configuration would otherwise shadow a `--set`
563+
// naming the same key: every earlier entry a current override claims is
564+
// dropped, and the override value is appended at the end.
565+
//
566+
// The manager also never forgets an override. withdrawn carries, per key, how
567+
// many stale entries an earlier start left behind; those are the first entries
568+
// the manager emits for that key, so dropping exactly that many reveals the
569+
// configuration value underneath — the value the service would have seen had
570+
// the override never been set.
571+
func withOverrides(envs []*resources.EnvironmentVariable, overrides map[string]string, withdrawn map[string]int) []*resources.EnvironmentVariable {
572+
if len(overrides) == 0 && len(withdrawn) == 0 {
573+
return envs
574+
}
575+
remaining := maps.Clone(withdrawn)
576+
resolved := make([]*resources.EnvironmentVariable, 0, len(envs)+len(overrides))
577+
for _, env := range envs {
578+
if _, claimed := overrides[env.Key]; claimed {
474579
continue
475580
}
476-
seen[envs[i].Key] = true
477-
resolved = append(resolved, envs[i])
581+
if remaining[env.Key] > 0 {
582+
remaining[env.Key]--
583+
continue
584+
}
585+
resolved = append(resolved, env)
586+
}
587+
for _, key := range slices.Sorted(maps.Keys(overrides)) {
588+
resolved = append(resolved, resources.Env(key, overrides[key]))
478589
}
479-
slices.Reverse(resolved)
480590
return resolved
481591
}
482592

@@ -488,6 +598,9 @@ func (s *Runtime) Stop(ctx context.Context, req *runtimev0.StopRequest) (*runtim
488598
defer s.Wool.Catch()
489599
ctx = s.Wool.Inject(ctx)
490600

601+
s.runnerMu.Lock()
602+
defer s.runnerMu.Unlock()
603+
491604
s.Wool.Debug("stopping service")
492605
if s.runner != nil {
493606
if err := s.runner.Stop(ctx); err != nil {

0 commit comments

Comments
 (0)