Make host ports node-exclusive placement claims

docs/configs/README.md

@@ -178,7 +178,7 @@ Supported fields:
 | `memory_mb` | no | Memory in MiB |
 | `kernel_args` | no | Kernel boot args |
 | `network` | no | When true, networking is configured |
-| `port_forwards` | no | Host-to-guest DNAT mappings; required for remote Traefik routing. Reachable from external traffic and from co-located guests addressing the host IP (hairpin) |
+| `port_forwards` | no | Host-to-guest DNAT mappings; required for remote Traefik routing. Reachable from external traffic and from co-located guests addressing the host IP (hairpin). Each `host_port` is a node-exclusive claim — see [Host-port claims](#host-port-claims) |
 | `health_check` | no | `type` supports `http` or `tcp` |
 | `env` | no | Env vars injected via kernel args; values with whitespace are encoded |
 | `links` | no | Same-node service links (`env` gets resolved URL) |
@@ -197,6 +197,26 @@ non-empty tenant `volumes` list replaces the inherited list. See
 [Persistent Volumes](../persistent-volumes.md)
 for lifecycle and safety semantics.
 
+#### Host-port claims
+
+A `host_port` is a node-scoped exclusive resource: the agent installs one DNAT
+rule per claim, and two colocated services claiming the same port would produce
+rules with identical match criteria and different guest destinations, so only
+one of them would ever receive traffic.
+
+- Repeating a `host_port` inside one service is always invalid and is rejected
+  by `configcheck` and by the enricher.
+- Repeating a `host_port` across services is valid only while placement keeps
+  them on different nodes. `configcheck` reports it as a `repeated_host_port`
+  warning because it cannot know the placement.
+- The scheduler treats a service's claims atomically: it places the service
+  only on a node where every claim is free, and otherwise leaves it `pending`
+  with reason `host_port_conflict` (or `duplicate_host_port_claims` for a
+  service that repeats its own port).
+- The agent rejects a rendered node config whose services conflict on a host
+  port before changing any networking or service state, reporting
+  `host_port_conflict` and leaving the revision unapplied.
+
 ### 2.3 `tenants/*` (optional)
 
 Tenant files support two modes:

docs/deployment-visibility.md

@@ -43,7 +43,10 @@ of the service's volumes.
 Missing data fails closed:
 
 - expired node leases become `stale`;
-- unplaced desired services are `pending`;
+- unplaced desired services are `pending`, carrying the scheduler's reason code
+  (for example `insufficient_compute_capacity`, `volume_capacity_unavailable`,
+  or `host_port_conflict`, which names the contested port and the service
+  already holding it);
 - placed services with missing, stale, or unsupported agent status are
   `unknown`;
 - VM state and health remain separate, so a service can be `running` and

internal/agent/agent.go

@@ -405,6 +405,18 @@ func (a *Agent) tick(ctx context.Context) {
 		a.setStatusServices(*merged, "")
 	}
 
+	// Reject a node config whose services conflict on a host port before any
+	// networking or service state changes. The control plane keeps colocated
+	// claims unique, so this is defense in depth for a stale, hand-written, or
+	// older-controller config: installing its DNAT rules would silently deliver
+	// host-port traffic to whichever guest matched first.
+	if err := config.ValidateNodePortClaims(*merged); err != nil {
+		a.logger.Error("host-port claim preflight failed; not advancing revision", "error", err)
+		a.failAgentStatus("NetworkReady", "host_port_conflict", err.Error())
+		a.syncRegistry(ctx, nodeCap, capacity.NodeCapacity{})
+		return
+	}
+
 	// Preflight routing metadata before any reconciliation so an invalid
 	// direct node config or a missing ingress_domain fails the revision early
 	// instead of being recorded as applied.

internal/agent/agent_test.go

@@ -878,6 +878,45 @@ func TestReadNodeCapacity_UsesLastKnownOnError(t *testing.T) {
 	}
 }
 
+func TestTick_HostPortConflict_RejectsConfigBeforeReconcile(t *testing.T) {
+	nodeYAML := []byte("node: web\nservices:\n" +
+		"- name: tenant-1-elasticsearch\n  image: /img/es\n  kernel: /kern\n  vcpus: 1\n  memory_mb: 256\n  network: {}\n  port_forwards:\n  - host_port: 9200\n    vm_port: 9200\n" +
+		"- name: tenant-2-elasticsearch\n  image: /img/es\n  kernel: /kern\n  vcpus: 1\n  memory_mb: 256\n  network: {}\n  port_forwards:\n  - host_port: 9200\n    vm_port: 9200\n")
+	s := &fakeStore{data: map[string][]byte{"web": nodeYAML}, revision: "rev-1"}
+
+	a := New(testAgentConfig(t), s, testLogger())
+	a.tick(context.Background())
+
+	if instances := a.vmManager.List(); len(instances) != 0 {
+		t.Errorf("expected no VMs to be started for a conflicting config, got %d", len(instances))
+	}
+	status := a.agentStatusSnapshot()
+	if status.Phase != "failed" || status.ReasonCode != "host_port_conflict" {
+		t.Fatalf("unexpected status: %#v", status)
+	}
+	if !strings.Contains(status.Message, "9200") {
+		t.Errorf("status message %q does not identify the conflicting port", status.Message)
+	}
+	// The revision must not be recorded as applied, so the next poll retries.
+	if a.lastRevision != "" {
+		t.Errorf("expected revision not to advance, got %q", a.lastRevision)
+	}
+}
+
+func TestTick_DistinctHostPortsReconcileNormally(t *testing.T) {
+	nodeYAML := []byte("node: web\nservices:\n" +
+		"- name: tenant-1-elasticsearch\n  image: /img/es\n  kernel: /kern\n  vcpus: 1\n  memory_mb: 256\n  network: {}\n  port_forwards:\n  - host_port: 9200\n    vm_port: 9200\n" +
+		"- name: tenant-2-elasticsearch\n  image: /img/es\n  kernel: /kern\n  vcpus: 1\n  memory_mb: 256\n  network: {}\n  port_forwards:\n  - host_port: 9201\n    vm_port: 9200\n")
+	s := &fakeStore{data: map[string][]byte{"web": nodeYAML}, revision: "rev-1"}
+
+	a := New(testAgentConfig(t), s, testLogger())
+	a.tick(context.Background())
+
+	if status := a.agentStatusSnapshot(); status.ReasonCode == "host_port_conflict" {
+		t.Fatalf("distinct host ports must not be rejected: %#v", status)
+	}
+}
+
 // runningVMManager reports a service as already running, which is what steady
 // state looks like: port forwards are only asserted for services that have a
 // guest to forward to.

internal/config/portclaim.go

@@ -0,0 +1,133 @@
+package config
+
+import (
+	"fmt"
+	"sort"
+	"strings"
+)
+
+// ProtocolTCP is the effective protocol of every port forward today. The agent
+// installs DNAT rules with "-p tcp", so a claim is currently identified by that
+// protocol plus the host port. Protocol is carried explicitly in PortClaim so
+// adding UDP forwards later widens the key without changing its callers.
+const ProtocolTCP = "tcp"
+
+// PortClaim is a node-exclusive host endpoint requested by a service. Two
+// services on the same node may not hold the same claim: their DNAT rules would
+// match identical traffic and only the first installed rule would win.
+type PortClaim struct {
+	Protocol string
+	HostPort int
+}
+
+func (c PortClaim) String() string {
+	return fmt.Sprintf("%s/%d", c.Protocol, c.HostPort)
+}
+
+// PortClaims returns the host-port claims a service makes, in declaration
+// order. Entries without a positive host port are not forwarded by the agent
+// and therefore claim nothing.
+func (s ServiceConfig) PortClaims() []PortClaim {
+	var claims []PortClaim
+	for _, pf := range s.PortForwards {
+		if pf.HostPort <= 0 {
+			continue
+		}
+		claims = append(claims, PortClaim{Protocol: ProtocolTCP, HostPort: pf.HostPort})
+	}
+	return claims
+}
+
+// DuplicatePortClaims returns the claims a single service requests more than
+// once, sorted by host port. Such a service is never schedulable anywhere: the
+// duplicate DNAT rules conflict with each other on any node.
+func (s ServiceConfig) DuplicatePortClaims() []PortClaim {
+	seen := make(map[PortClaim]int, len(s.PortForwards))
+	for _, claim := range s.PortClaims() {
+		seen[claim]++
+	}
+	var dupes []PortClaim
+	for claim, count := range seen {
+		if count > 1 {
+			dupes = append(dupes, claim)
+		}
+	}
+	sortClaims(dupes)
+	return dupes
+}
+
+// PortClaimConflict reports a claim requested by more than one service in the
+// same node scope. Services are sorted so the message is stable.
+type PortClaimConflict struct {
+	Claim    PortClaim
+	Services []string
+}
+
+func (c PortClaimConflict) String() string {
+	return fmt.Sprintf("host port %d (%s) is claimed by %s",
+		c.Claim.HostPort, c.Claim.Protocol, strings.Join(c.Services, ", "))
+}
+
+// ConflictingPortClaims returns claims held by more than one of the given
+// services. It is the cross-service half of the invariant only; use
+// DuplicatePortClaims for claims a single service repeats.
+func ConflictingPortClaims(services []ServiceConfig) []PortClaimConflict {
+	holders := make(map[PortClaim][]string)
+	for _, svc := range services {
+		claimed := make(map[PortClaim]bool)
+		for _, claim := range svc.PortClaims() {
+			if claimed[claim] {
+				continue
+			}
+			claimed[claim] = true
+			holders[claim] = append(holders[claim], svc.Name)
+		}
+	}
+
+	var conflicts []PortClaimConflict
+	for claim, names := range holders {
+		if len(names) < 2 {
+			continue
+		}
+		sorted := append([]string(nil), names...)
+		sort.Strings(sorted)
+		conflicts = append(conflicts, PortClaimConflict{Claim: claim, Services: sorted})
+	}
+	sort.Slice(conflicts, func(i, j int) bool {
+		return claimLess(conflicts[i].Claim, conflicts[j].Claim)
+	})
+	return conflicts
+}
+
+// ValidateNodePortClaims rejects a rendered node config whose services cannot
+// coexist on one node. It is the agent's admission boundary: a stale,
+// hand-written, or older-controller config must be refused before any
+// networking or service state changes, because the resulting DNAT rules would
+// silently deliver traffic to the wrong guest.
+func ValidateNodePortClaims(nc NodeConfig) error {
+	var problems []string
+	for _, svc := range nc.Services {
+		for _, dupe := range svc.DuplicatePortClaims() {
+			problems = append(problems, fmt.Sprintf(
+				"service %s claims host port %d (%s) more than once", svc.Name, dupe.HostPort, dupe.Protocol))
+		}
+	}
+	for _, conflict := range ConflictingPortClaims(nc.Services) {
+		problems = append(problems, conflict.String())
+	}
+	if len(problems) == 0 {
+		return nil
+	}
+	return fmt.Errorf("conflicting host-port claims on node %s: %s", nc.Node, strings.Join(problems, "; "))
+}
+
+func sortClaims(claims []PortClaim) {
+	sort.Slice(claims, func(i, j int) bool { return claimLess(claims[i], claims[j]) })
+}
+
+func claimLess(a, b PortClaim) bool {
+	if a.HostPort != b.HostPort {
+		return a.HostPort < b.HostPort
+	}
+	return a.Protocol < b.Protocol
+}

internal/config/portclaim_test.go

@@ -0,0 +1,101 @@
+package config
+
+import (
+	"strings"
+	"testing"
+)
+
+func TestPortClaims_SkipsEntriesWithoutHostPort(t *testing.T) {
+	svc := ServiceConfig{Name: "a", PortForwards: []PortForward{
+		{HostPort: 9200, VMPort: 9200},
+		{VMPort: 9300},
+		{HostPort: 0, VMPort: 5601},
+	}}
+
+	claims := svc.PortClaims()
+	if len(claims) != 1 || claims[0] != (PortClaim{Protocol: ProtocolTCP, HostPort: 9200}) {
+		t.Fatalf("unexpected claims: %#v", claims)
+	}
+}
+
+func TestDuplicatePortClaims(t *testing.T) {
+	svc := ServiceConfig{Name: "a", PortForwards: []PortForward{
+		{HostPort: 9200, VMPort: 9200},
+		{HostPort: 9200, VMPort: 9201},
+		{HostPort: 5601, VMPort: 5601},
+	}}
+
+	dupes := svc.DuplicatePortClaims()
+	if len(dupes) != 1 || dupes[0].HostPort != 9200 {
+		t.Fatalf("expected 9200 reported once, got %#v", dupes)
+	}
+}
+
+func TestConflictingPortClaims_ReportsHoldersSorted(t *testing.T) {
+	services := []ServiceConfig{
+		{Name: "tenant-2-elasticsearch", PortForwards: []PortForward{{HostPort: 9200, VMPort: 9200}}},
+		{Name: "tenant-1-elasticsearch", PortForwards: []PortForward{{HostPort: 9200, VMPort: 9200}}},
+		{Name: "tenant-1-kibana", PortForwards: []PortForward{{HostPort: 5601, VMPort: 5601}}},
+	}
+
+	conflicts := ConflictingPortClaims(services)
+	if len(conflicts) != 1 {
+		t.Fatalf("expected one conflict, got %#v", conflicts)
+	}
+	if conflicts[0].Claim.HostPort != 9200 {
+		t.Errorf("expected conflict on 9200, got %d", conflicts[0].Claim.HostPort)
+	}
+	if got := strings.Join(conflicts[0].Services, ","); got != "tenant-1-elasticsearch,tenant-2-elasticsearch" {
+		t.Errorf("unexpected holders: %s", got)
+	}
+}
+
+// A service repeating one claim conflicts with itself, not with its peers.
+func TestConflictingPortClaims_IgnoresRepeatsWithinOneService(t *testing.T) {
+	services := []ServiceConfig{
+		{Name: "a", PortForwards: []PortForward{{HostPort: 9200, VMPort: 9200}, {HostPort: 9200, VMPort: 9201}}},
+	}
+
+	if conflicts := ConflictingPortClaims(services); len(conflicts) != 0 {
+		t.Fatalf("expected no cross-service conflict, got %#v", conflicts)
+	}
+}
+
+func TestValidateNodePortClaims_RejectsCollocatedDuplicate(t *testing.T) {
+	nc := NodeConfig{Node: "i-001", Services: []ServiceConfig{
+		{Name: "tenant-1-elasticsearch", PortForwards: []PortForward{{HostPort: 9200, VMPort: 9200}}},
+		{Name: "tenant-2-elasticsearch", PortForwards: []PortForward{{HostPort: 9200, VMPort: 9200}}},
+	}}
+
+	err := ValidateNodePortClaims(nc)
+	if err == nil {
+		t.Fatal("expected conflicting claims to be rejected")
+	}
+	for _, want := range []string{"i-001", "9200", "tenant-1-elasticsearch", "tenant-2-elasticsearch"} {
+		if !strings.Contains(err.Error(), want) {
+			t.Errorf("error %q does not identify %q", err, want)
+		}
+	}
+}
+
+func TestValidateNodePortClaims_RejectsSelfDuplicate(t *testing.T) {
+	nc := NodeConfig{Node: "i-001", Services: []ServiceConfig{
+		{Name: "a", PortForwards: []PortForward{{HostPort: 8080, VMPort: 80}, {HostPort: 8080, VMPort: 81}}},
+	}}
+
+	if err := ValidateNodePortClaims(nc); err == nil {
+		t.Fatal("expected a service claiming one host port twice to be rejected")
+	}
+}
+
+func TestValidateNodePortClaims_AllowsDistinctClaims(t *testing.T) {
+	nc := NodeConfig{Node: "i-001", Services: []ServiceConfig{
+		{Name: "a", PortForwards: []PortForward{{HostPort: 9200, VMPort: 9200}, {HostPort: 9300, VMPort: 9300}}},
+		{Name: "b", PortForwards: []PortForward{{HostPort: 9201, VMPort: 9200}}},
+		{Name: "c"},
+	}}
+
+	if err := ValidateNodePortClaims(nc); err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+}