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