From afe6427d9c2bbb1d4eb9a495b644a472f3302097 Mon Sep 17 00:00:00 2001 From: antoinetoussaint-byte Date: Sun, 25 Oct 2026 22:45:43 -0400 Subject: [PATCH] fix(run): a hop route to a module that starts after the hop is frozen, not refused MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A fresh run starts authority/hop before every routed module ordered after it, and a module that calls through the hop is (robin's playground with annotations 3fae5dc1). When the hop resolved its generated routes, that module had loaded but not initialized, so freezeHopAddresses found no recorded mapping for it and refused the hop's own start with "published hop routes changed: stop and restart this Codefly run" on every fresh start, although nothing had changed. The address the hop's resolution derived for such a module is the one its own Init will be proposed. The snapshot now freezes it, and the module's Init is the check: RecordNetworkMappings refuses any accepted set that drops or moves an address a hop snapshot holds, on the first record as well as on a reload (before, only a reload was checked). A module that never loaded still has no mapping and is refused by name; a module that has initialized must still be at the address it recorded. The mid-run refusal is unchanged, and every restart refusal now names the route and the reason. Tests: TestHopRoutesFreezeAModuleThatInitializesAfterTheHop — the hop's delivery succeeds with an uninitialized routed module and routes it at its proposed address; that module's Init is refused when it accepts another address or drops the endpoint, and accepted at the frozen one. Sabotage-checked both halves: without the freeze the hop's delivery fails with the reported error; without the first-record check the moved Init is accepted. Co-Authored-By: Claude Opus 5.5 --- pkg/orchestration/hop_configurations.go | 76 ++++++++++++++------ pkg/orchestration/hop_configurations_test.go | 64 +++++++++++++++++ pkg/orchestration/state_manager.go | 18 +---- 3 files changed, 123 insertions(+), 35 deletions(-) diff --git a/pkg/orchestration/hop_configurations.go b/pkg/orchestration/hop_configurations.go index 0f49fbbf..bf6da4c7 100644 --- a/pkg/orchestration/hop_configurations.go +++ b/pkg/orchestration/hop_configurations.go @@ -52,7 +52,7 @@ func (r *WorkspaceConfigurationReader) Load(ctx context.Context, env *resources. } snapshot := hopRoutingSnapshot(group) if r.frozen && r.snapshot != snapshot { - return routingRestartRequired() + return routingRestartRequired("the published API catalogs differ from the ones this run started with") } r.snapshot, r.frozen = snapshot, true r.combined = slices.Clone(configs) @@ -62,8 +62,8 @@ func (r *WorkspaceConfigurationReader) Load(ctx context.Context, env *resources. return nil } -func routingRestartRequired() error { - return fmt.Errorf("published hop routes changed: stop and restart this Codefly run to adopt them; incremental refresh cannot replace its routing snapshot") +func routingRestartRequired(reason string) error { + return fmt.Errorf("published hop routes changed: stop and restart this Codefly run to adopt them; incremental refresh cannot replace its routing snapshot (%s)", reason) } func hopRoutingSnapshot(config *basev0.Configuration) string { @@ -254,13 +254,22 @@ func (flow *Flow) CheckRoutingSnapshot(ctx context.Context, workspace *resources current = hopRoutingSnapshotOf(provided.Infos) } if previous != current { - return routingRestartRequired() + return routingRestartRequired("the published API catalogs differ from the ones this run started with") } return nil } // freezeHopAddresses fences accepted endpoint replacement while an immutable // hop snapshot can still use it. Ordinary reload reuses the accepted mappings. +// +// A routed service that has initialized must still be at the address it +// recorded. One that has loaded but not initialized yet — a fresh run starts +// the hop before every module ordered after it, and a module that calls +// through the hop is — is routed at the address its own Init will be +// proposed, the one the hop's resolution derived. Freezing that address makes +// the routed service's Init the check (RecordNetworkMappings): the same +// address is accepted, any other is refused naming the route. A service that +// never loaded has no mapping at all and is refused here by name. func (s *StateManager) freezeHopAddresses(raw string, resolved []*basev0.NetworkMapping) error { routes, err := routing.DecodeRouting(raw) if err != nil { @@ -268,7 +277,7 @@ func (s *StateManager) freezeHopAddresses(raw string, resolved []*basev0.Network } s.mu.Lock() defer s.mu.Unlock() - pending := map[string]string{} + pending := map[string]frozenRoute{} for i := range routes.Endpoints { route := &routes.Endpoints[i] unique := route.Audience + "/" + route.Service @@ -284,30 +293,57 @@ func (s *StateManager) freezeHopAddresses(raw string, resolved []*basev0.Network return fmt.Errorf("hop route %s/%s has no resolved mapping", unique, route.Endpoint) } key, hash := resources.EndpointDestination(selected.GetEndpoint()), resources.NetworkMappingHash(selected) - current := false - for _, mapping := range s.networkMappings[unique] { - if resources.EndpointDestination(mapping.GetEndpoint()) == key && resources.NetworkMappingHash(mapping) == hash { - current = true - break - } + if recorded, initialized := s.networkMappings[unique]; initialized && !carriesMapping(recorded, key, hash) { + return routingRestartRequired(fmt.Sprintf("hop route %s is not the address %s initialized with", key, unique)) } - if !current { - return routingRestartRequired() + if previous, ok := s.routingMappings[key]; ok && previous.hash != hash { + return routingRestartRequired(fmt.Sprintf("hop route %s moved from the address this run's hop snapshot holds", key)) } - if previous, ok := s.routingMappings[key]; ok && previous != hash { - return routingRestartRequired() - } - pending[key] = hash + pending[key] = frozenRoute{owner: unique, hash: hash} } if s.routingMappings == nil { - s.routingMappings = map[string]string{} + s.routingMappings = map[string]frozenRoute{} + } + for key, frozen := range pending { + s.routingMappings[key] = frozen } - for key, hash := range pending { - s.routingMappings[key] = hash + return nil +} + +// frozenRoute is one address a hop snapshot holds: the service that serves it +// and the hash of the mapping the snapshot was built with. +type frozenRoute struct { + owner string + hash string +} + +// checkFrozenRoutes refuses mappings for unique that drop or move an address a +// hop snapshot holds. The caller holds s.mu. +func (s *StateManager) checkFrozenRoutes(unique string, mappings []*basev0.NetworkMapping) error { + for key, frozen := range s.routingMappings { + if frozen.owner == unique && !carriesMapping(mappings, key, frozen.hash) { + return routingRestartRequired(fmt.Sprintf("the hop snapshot routes %s at an address %s does not accept", key, unique)) + } } return nil } +// carriesMapping reports whether mappings hold the endpoint key, and hold it +// only at the mapping hashed as hash. +func carriesMapping(mappings []*basev0.NetworkMapping, key, hash string) bool { + found := false + for _, mapping := range mappings { + if resources.EndpointDestination(mapping.GetEndpoint()) != key { + continue + } + if resources.NetworkMappingHash(mapping) != hash { + return false + } + found = true + } + return found +} + func (world *World) freezeDeliveredHopAddresses(configs []*basev0.Configuration, mappings []*basev0.NetworkMapping) error { if world.Mode != RunMode || world.SharedState == nil { return nil diff --git a/pkg/orchestration/hop_configurations_test.go b/pkg/orchestration/hop_configurations_test.go index 927454d4..5f1e5d8b 100644 --- a/pkg/orchestration/hop_configurations_test.go +++ b/pkg/orchestration/hop_configurations_test.go @@ -97,6 +97,70 @@ func TestGeneratedHopRoutesUseTheNormalGraphAndAddressResolver(t *testing.T) { require.NotContains(t, fmt.Sprint(previous), "moved.native", "rejected mapping never replaces accepted route") } +// A fresh run initializes the hop before a routed module that is ordered after +// it (a module that calls through the hop is). That module has loaded, so its +// address is the one its own Init will be proposed; the hop's snapshot freezes +// it, and the module's Init is then the check: the same address is accepted, +// any other is refused naming the route. Before, the hop's own start was +// refused as a routing change although nothing had changed. +func TestHopRoutesFreezeAModuleThatInitializesAfterTheHop(t *testing.T) { + t.Setenv(resources.CodeflyHomeEnv, filepath.Join(t.TempDir(), "home")) + ctx := context.Background() + workspace := hopRouteWorkspace(t, "public") + flow := declaredReferenceFlow(t, workspace) + require.NoError(t, flow.checkConfigurationReferences(ctx, nil)) + deps, err := flow.managerDependencies(ctx) + require.NoError(t, err) + var uniques []string + for _, dep := range deps { + uniques = append(uniques, dep.Unique) + } + world := flow.world + world.setRunProducers(uniques, flow.originService) + require.NoError(t, flow.ConfigurationManager.Load(ctx, world.Env.Runtime())) + world.Mode = RunMode + world.runtimeContextFor = func(*resources.Service) string { return resources.RuntimeContextNative } + // Both loaded; only first has initialized. + recordParityEndpoints(t, world, "platform", "first") + recordParityEndpoints(t, world, "platform", "second") + recordMappings(t, world, "platform", "first", endpointMapping("platform", "first", "web", "http", "public", nativeInstance("http://first.native"), containerInstance("http://first.cell"))) + proposed, err := world.producerNetworkMappings(ctx, "platform/second") + require.NoError(t, err) + require.Len(t, proposed, 1) + + configs, err := world.workspaceConfigurationsFor(ctx, flow.originService, nil, resources.NewNativeNetworkAccess()) + require.NoError(t, err, "a fresh run never sees changed routes") + raw, delivered := groupValue(configs, hopRoutesGroup, "ROUTES") + require.True(t, delivered) + routes, err := routing.DecodeRouting(raw) + require.NoError(t, err) + route, err := routes.Find("platform", "http", "GET /second") + require.NoError(t, err) + native, err := resources.FindNetworkInstanceInNetworkMappings(ctx, proposed, proposed[0].GetEndpoint(), resources.NewNativeNetworkAccess()) + require.NoError(t, err) + require.Equal(t, native.GetAddress(), route.Address, "the hop routes second at the address its Init will be proposed") + + module, err := workspace.LoadModuleFromName(ctx, "platform") + require.NoError(t, err) + second, err := module.LoadServiceFromName(ctx, "second") + require.NoError(t, err) + // An Init that accepts another address than the frozen one is refused by + // name, and nothing is recorded in its place. + moved := endpointMapping("platform", "second", "web", "http", "public", nativeInstance("http://moved.native")) + err = world.SharedState.RecordNetworkMappings(ctx, second, []*basev0.NetworkMapping{moved}) + require.ErrorContains(t, err, "stop and restart") + require.ErrorContains(t, err, "platform/second") + _, recorded := world.SharedState.GetNetworkMappingsFromUnique("platform/second") + require.False(t, recorded, "a refused Init records nothing") + // An Init that drops the routed endpoint is refused the same way. + require.ErrorContains(t, world.SharedState.RecordNetworkMappings(ctx, second, nil), "stop and restart") + // The address the hop froze is accepted. + require.NoError(t, world.SharedState.RecordNetworkMappings(ctx, second, proposed)) + // And the hop's next delivery reads the same snapshot. + _, err = world.workspaceConfigurationsFor(ctx, flow.originService, nil, resources.NewNativeNetworkAccess()) + require.NoError(t, err) +} + func TestGeneratedHopRoutesRefusePrivatePublicationAndStaleArtifacts(t *testing.T) { t.Setenv(resources.CodeflyHomeEnv, filepath.Join(t.TempDir(), "home")) t.Run("private", func(t *testing.T) { diff --git a/pkg/orchestration/state_manager.go b/pkg/orchestration/state_manager.go index f434ff43..6130d240 100644 --- a/pkg/orchestration/state_manager.go +++ b/pkg/orchestration/state_manager.go @@ -34,7 +34,7 @@ type StateManager struct { endpoints map[string][]*basev0.Endpoint networkMappings map[string][]*basev0.NetworkMapping // Generated hop snapshots retain these exact addresses until run restart. - routingMappings map[string]string + routingMappings map[string]frozenRoute } // SetDependencies rebinds the dependency graph. Used by Flow when @@ -297,20 +297,8 @@ func (s *StateManager) RecordNetworkMappings(ctx context.Context, service *resou } s.mu.Lock() defer s.mu.Unlock() - for _, previous := range s.networkMappings[id.Unique()] { - key := resources.EndpointDestination(previous.GetEndpoint()) - if expected, frozen := s.routingMappings[key]; frozen { - found := false - for _, candidate := range mappings { - if resources.EndpointDestination(candidate.GetEndpoint()) == key { - found = resources.NetworkMappingHash(candidate) == expected - break - } - } - if !found { - return routingRestartRequired() - } - } + if err = s.checkFrozenRoutes(id.Unique(), mappings); err != nil { + return err } s.networkMappings[id.Unique()] = mappings return nil