fix(run): a hop route to a module that starts after the hop is frozen, not refused

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,21 +254,30 @@ 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 {
 		return err
 	}
 	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

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) {

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