refactor(vpcd): collapse SG subscribers onto network/{topology,policy} managers

Phase 2.6 / mulga-siv-129 slice 2. Rewrites the vpc.create-sg /
vpc.update-sg / vpc.delete-sg NATS subscribers in services/vpcd as
thin wrappers over topology.Manager (port group lifecycle) and
policy.SecurityGroupManager (ACL programming). Deletes the ~1000
lines of duplicated ACL builder and provisioning logic — its
authoritative copy lives in network/policy.

Surface additions:

- ovn.ErrPortGroupNotFound sentinel; mock + live GetPortGroup wrap
  not-found via fmt.Errorf("%w: ...") so callers can use errors.Is.
- topology.Manager grows EnsureSGPortGroup / DeleteSGPortGroup
  (resolves the apply.go TODO from §2.1 of the redesign plan).
  liveManager implements them with GetPortGroup-then-Create /
  GetPortGroup-then-Clear+Delete idempotency.

Behaviour delta: the old provisionSG rolled back the OVN port group
on AddACLs failure ("all-or-nothing"). The new path lets the port
group survive and relies on network/reconcile to retry EnsureSG on
its next sweep. The all-or-nothing fail-fast test cases are dropped;
ACL emission semantics are covered by network/policy/sg_test.go.

spinifex/network/ovn/client.go

@@ -16,6 +16,11 @@ import (
 // the same delete) match it with errors.Is and treat as success.
 var ErrNATNotFound = errors.New("NAT not found")
 
+// ErrPortGroupNotFound is returned by GetPortGroup when the named port group
+// is absent. EnsureSGPortGroup / DeleteSGPortGroup use errors.Is to make
+// existence checks and delete-then-recheck flows idempotent.
+var ErrPortGroupNotFound = errors.New("port group not found")
+
 // ACLSpec describes an OVN ACL rule for attachment to a port group.
 // Name, Severity, and Meter are optional — set Severity when Log is true.
 type ACLSpec struct {

spinifex/network/ovn/live.go

@@ -1068,7 +1068,7 @@ func (c *LiveClient) getPortGroup(ctx context.Context, name string) (*nbdb.PortG
 		return nil, fmt.Errorf("get port group: %w", err)
 	}
 	if len(pgs) == 0 {
-		return nil, fmt.Errorf("port group %q not found", name)
+		return nil, fmt.Errorf("%w: %q", ErrPortGroupNotFound, name)
 	}
 	return &pgs[0], nil
 }

spinifex/network/ovn/mock/mock.go

@@ -700,13 +700,14 @@ func (m *Client) ListPortGroupsForPort(_ context.Context, lspName string) ([]str
 }
 
 // GetPortGroup returns the port group with the given name, or an error if it
-// doesn't exist. Mirrors the live client.
+// doesn't exist. Mirrors the live client; missing rows surface as
+// ovn.ErrPortGroupNotFound so callers can use errors.Is for idempotent flows.
 func (m *Client) GetPortGroup(_ context.Context, name string) (*nbdb.PortGroup, error) {
 	m.Mu.Lock()
 	defer m.Mu.Unlock()
 	pg, exists := m.PortGroups[name]
 	if !exists {
-		return nil, fmt.Errorf("port group %q not found", name)
+		return nil, fmt.Errorf("%w: %q", ovn.ErrPortGroupNotFound, name)
 	}
 	return pg, nil
 }

spinifex/network/topology/manager.go

@@ -34,6 +34,12 @@ type Manager interface {
 	// memberships for a port. Manager computes the add/remove diff against
 	// current OVN state.
 	SetPortSecurityGroups(ctx context.Context, portID string, sgIDs []string) error
+
+	// SG port-group lifecycle. ACL programming on a port group lives in
+	// network/policy.SecurityGroupManager — topology only owns the empty
+	// OVN port-group row keyed by SecurityGroupPortGroup(groupID).
+	EnsureSGPortGroup(ctx context.Context, groupID string) error
+	DeleteSGPortGroup(ctx context.Context, groupID string) error
 }
 
 // VPCSpec describes a VPC at L2. AZSlice is reserved for Phase 3 per-AZ CIDR

spinifex/network/topology/manager_live.go

@@ -2,6 +2,7 @@ package topology
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"log/slog"
 	"net/netip"
@@ -333,6 +334,59 @@ func (m *liveManager) DeletePort(ctx context.Context, spec PortSpec) error {
 	return nil
 }
 
+// EnsureSGPortGroup creates the empty OVN port-group row for a security
+// group. Idempotent: a second call with the same groupID is a no-op. ACL
+// programming lives in network/policy.SecurityGroupManager — callers must
+// invoke it after EnsureSGPortGroup to attach the infrastructure and tenant
+// rule sets.
+func (m *liveManager) EnsureSGPortGroup(ctx context.Context, groupID string) error {
+	if m.ovn == nil {
+		return fmt.Errorf("OVN client not connected")
+	}
+	if groupID == "" {
+		return fmt.Errorf("EnsureSGPortGroup: empty groupID")
+	}
+	pgName := SecurityGroupPortGroup(groupID)
+	if _, err := m.ovn.GetPortGroup(ctx, pgName); err == nil {
+		return nil
+	} else if !errors.Is(err, ovn.ErrPortGroupNotFound) {
+		return fmt.Errorf("get port group %s: %w", pgName, err)
+	}
+	if err := m.ovn.CreatePortGroup(ctx, pgName, nil); err != nil {
+		return fmt.Errorf("create port group %s: %w", pgName, err)
+	}
+	slog.Info("topology: created SG port group", "pg", pgName, "group_id", groupID)
+	return nil
+}
+
+// DeleteSGPortGroup clears every ACL on the SG's port group and removes the
+// port group row. Idempotent: missing port group is treated as success.
+// Reference-integrity dictates ClearACLs before DeletePortGroup — libovsdb
+// rejects deleting a port group with dangling ACL references.
+func (m *liveManager) DeleteSGPortGroup(ctx context.Context, groupID string) error {
+	if m.ovn == nil {
+		return fmt.Errorf("OVN client not connected")
+	}
+	if groupID == "" {
+		return fmt.Errorf("DeleteSGPortGroup: empty groupID")
+	}
+	pgName := SecurityGroupPortGroup(groupID)
+	if _, err := m.ovn.GetPortGroup(ctx, pgName); err != nil {
+		if errors.Is(err, ovn.ErrPortGroupNotFound) {
+			return nil
+		}
+		return fmt.Errorf("get port group %s: %w", pgName, err)
+	}
+	if err := m.ovn.ClearACLs(ctx, pgName); err != nil {
+		return fmt.Errorf("clear ACLs on %s: %w", pgName, err)
+	}
+	if err := m.ovn.DeletePortGroup(ctx, pgName); err != nil {
+		return fmt.Errorf("delete port group %s: %w", pgName, err)
+	}
+	slog.Info("topology: deleted SG port group", "pg", pgName, "group_id", groupID)
+	return nil
+}
+
 // SetPortSecurityGroups reconciles the port's port-group memberships against
 // the declared list.
 func (m *liveManager) SetPortSecurityGroups(ctx context.Context, portID string, sgIDs []string) error {

spinifex/network/topology/manager_live_test.go

@@ -160,6 +160,47 @@ func TestLiveManager_DeleteSubnetAndVPC(t *testing.T) {
 	}
 }
 
+func TestLiveManager_EnsureSGPortGroup(t *testing.T) {
+	mgr, mockClient := newLiveManagerForTest(t)
+	ctx := context.Background()
+
+	if err := mgr.EnsureSGPortGroup(ctx, "sg-pg1"); err != nil {
+		t.Fatalf("EnsureSGPortGroup: %v", err)
+	}
+	pg, err := mockClient.GetPortGroup(ctx, SecurityGroupPortGroup("sg-pg1"))
+	if err != nil {
+		t.Fatalf("port group not present: %v", err)
+	}
+	if pg.Name != SecurityGroupPortGroup("sg-pg1") {
+		t.Errorf("port group name mismatch: %q", pg.Name)
+	}
+
+	// Idempotent: second call must not fail or churn state.
+	if err := mgr.EnsureSGPortGroup(ctx, "sg-pg1"); err != nil {
+		t.Fatalf("EnsureSGPortGroup second call: %v", err)
+	}
+}
+
+func TestLiveManager_DeleteSGPortGroup(t *testing.T) {
+	mgr, mockClient := newLiveManagerForTest(t)
+	ctx := context.Background()
+
+	if err := mgr.EnsureSGPortGroup(ctx, "sg-del1"); err != nil {
+		t.Fatalf("EnsureSGPortGroup: %v", err)
+	}
+	if err := mgr.DeleteSGPortGroup(ctx, "sg-del1"); err != nil {
+		t.Fatalf("DeleteSGPortGroup: %v", err)
+	}
+	if _, err := mockClient.GetPortGroup(ctx, SecurityGroupPortGroup("sg-del1")); err == nil {
+		t.Fatalf("port group still present after delete")
+	}
+
+	// Idempotent on already-absent.
+	if err := mgr.DeleteSGPortGroup(ctx, "sg-del1"); err != nil {
+		t.Fatalf("DeleteSGPortGroup second call: %v", err)
+	}
+}
+
 func TestLiveManager_SetPortSecurityGroups(t *testing.T) {
 	mgr, mockClient := newLiveManagerForTest(t)
 	ctx := context.Background()

spinifex/services/vpcd/acl.go

@@ -1,110 +0,0 @@
-package vpcd
-
-import (
-	"fmt"
-	"strings"
-)
-
-// SGRuleForACL mirrors the SGRule from handlers for vpcd's use.
-type SGRuleForACL struct {
-	IpProtocol string `json:"ip_protocol"` // "tcp", "udp", "icmp", "-1" (all)
-	FromPort   int64  `json:"from_port"`
-	ToPort     int64  `json:"to_port"`
-	CidrIp     string `json:"cidr_ip,omitempty"`
-	SourceSG   string `json:"source_sg,omitempty"`
-}
-
-// portGroupName converts a security group ID to an OVN port group name.
-// OVN port group names must match [a-zA-Z_][a-zA-Z0-9_]*, so hyphens
-// in sg-xxx IDs are replaced with underscores.
-func portGroupName(groupId string) string {
-	return strings.ReplaceAll(groupId, "-", "_")
-}
-
-// addressSetName is the OVN address set holding the IPv4 addresses of every
-// port that belongs to the given port group. SG-to-SG ACL match expressions
-// reference it via "ip4.src == $<name>" / "ip4.dst == $<name>". Lifecycle is
-// tied to the port group (created in handleCreateSG, deleted in handleDeleteSG);
-// reconcilePortSGs maintains its membership in step with port-group joins.
-func addressSetName(pgName string) string {
-	return pgName + "_ip4"
-}
-
-// BuildIngressACLMatch generates an OVN ACL match expression for an ingress rule.
-// Ingress rules use "outport == @{pgName}" because OVN ACLs with direction "to-lport"
-// match on the destination port (outport from the pipeline's perspective).
-//
-// Tenant-supplied CidrIp/SourceSG values MUST be validated by the handler before
-// reaching this builder (see ec2/vpc/security_group.go ipPermissionsToSGRules). This
-// builder interpolates them verbatim.
-func BuildIngressACLMatch(pgName string, rule SGRuleForACL) string {
-	parts := []string{fmt.Sprintf("outport == @%s", pgName), "ip4"}
-
-	// Protocol filter
-	parts = appendProtocolMatch(parts, rule)
-
-	// Source filter
-	if rule.CidrIp != "" && rule.CidrIp != "0.0.0.0/0" {
-		parts = append(parts, fmt.Sprintf("ip4.src == %s", rule.CidrIp))
-	}
-	if rule.SourceSG != "" {
-		parts = append(parts, fmt.Sprintf("ip4.src == $%s", addressSetName(portGroupName(rule.SourceSG))))
-	}
-
-	return strings.Join(parts, " && ")
-}
-
-// BuildEgressACLMatch generates an OVN ACL match expression for an egress rule.
-// Egress rules use "inport == @{pgName}" because OVN ACLs with direction "from-lport"
-// match on the source port (inport from the pipeline's perspective).
-func BuildEgressACLMatch(pgName string, rule SGRuleForACL) string {
-	parts := []string{fmt.Sprintf("inport == @%s", pgName), "ip4"}
-
-	// Protocol filter
-	parts = appendProtocolMatch(parts, rule)
-
-	// Destination filter
-	if rule.CidrIp != "" && rule.CidrIp != "0.0.0.0/0" {
-		parts = append(parts, fmt.Sprintf("ip4.dst == %s", rule.CidrIp))
-	}
-	if rule.SourceSG != "" {
-		parts = append(parts, fmt.Sprintf("ip4.dst == $%s", addressSetName(portGroupName(rule.SourceSG))))
-	}
-
-	return strings.Join(parts, " && ")
-}
-
-// appendProtocolMatch adds protocol-specific match clauses to the parts slice.
-func appendProtocolMatch(parts []string, rule SGRuleForACL) []string {
-	switch rule.IpProtocol {
-	case "tcp":
-		parts = appendPortMatch(parts, "tcp", rule.FromPort, rule.ToPort)
-	case "udp":
-		parts = appendPortMatch(parts, "udp", rule.FromPort, rule.ToPort)
-	case "icmp":
-		parts = append(parts, "icmp4")
-	case "-1", "":
-		// All protocols — no additional filter needed beyond "ip4"
-	default:
-		// Unknown protocol — treat as all traffic
-	}
-	return parts
-}
-
-// appendPortMatch adds TCP/UDP port match clauses.
-func appendPortMatch(parts []string, proto string, fromPort, toPort int64) []string {
-	if fromPort == 0 && toPort == 0 {
-		// All ports for this protocol
-		parts = append(parts, proto)
-		return parts
-	}
-	if fromPort == toPort {
-		// Single port
-		parts = append(parts, fmt.Sprintf("%s.dst == %d", proto, fromPort))
-	} else {
-		// Port range
-		parts = append(parts, fmt.Sprintf("%s.dst >= %d", proto, fromPort))
-		parts = append(parts, fmt.Sprintf("%s.dst <= %d", proto, toPort))
-	}
-	return parts
-}

spinifex/services/vpcd/acl_test.go

@@ -1,103 +0,0 @@
-package vpcd
-
-import (
-	"testing"
-
-	"github.com/stretchr/testify/assert"
-)
-
-func TestACL_TCPPortFromCIDR(t *testing.T) {
-	rule := SGRuleForACL{
-		IpProtocol: "tcp",
-		FromPort:   22,
-		ToPort:     22,
-		CidrIp:     "10.0.0.0/8",
-	}
-	match := BuildIngressACLMatch("sg_test", rule)
-	assert.Contains(t, match, "tcp.dst == 22")
-	assert.Contains(t, match, "ip4.src == 10.0.0.0/8")
-	assert.Contains(t, match, "outport == @sg_test")
-	assert.Contains(t, match, "ip4")
-}
-
-func TestACL_AllTrafficFromSG(t *testing.T) {
-	rule := SGRuleForACL{
-		IpProtocol: "-1",
-		SourceSG:   "sg-abc123",
-	}
-	match := BuildIngressACLMatch("sg_test", rule)
-	assert.Contains(t, match, "ip4.src == $sg_abc123_ip4")
-	assert.Contains(t, match, "outport == @sg_test")
-	assert.Contains(t, match, "ip4")
-}
-
-func TestACL_PortRange(t *testing.T) {
-	rule := SGRuleForACL{
-		IpProtocol: "udp",
-		FromPort:   1024,
-		ToPort:     65535,
-	}
-	match := BuildIngressACLMatch("sg_test", rule)
-	assert.Contains(t, match, "udp.dst >= 1024")
-	assert.Contains(t, match, "udp.dst <= 65535")
-}
-
-func TestACL_ICMP(t *testing.T) {
-	rule := SGRuleForACL{
-		IpProtocol: "icmp",
-		CidrIp:     "0.0.0.0/0",
-	}
-	match := BuildIngressACLMatch("sg_test", rule)
-	assert.Contains(t, match, "icmp4")
-	assert.NotContains(t, match, "tcp.dst")
-	assert.NotContains(t, match, "udp.dst")
-}
-
-func TestACL_AllProtocols(t *testing.T) {
-	rule := SGRuleForACL{
-		IpProtocol: "-1",
-		CidrIp:     "10.0.0.0/16",
-	}
-	match := BuildIngressACLMatch("sg_test", rule)
-	assert.Contains(t, match, "ip4")
-	assert.Contains(t, match, "ip4.src == 10.0.0.0/16")
-	assert.NotContains(t, match, "tcp")
-	assert.NotContains(t, match, "udp")
-	assert.NotContains(t, match, "icmp")
-}
-
-func TestACL_EgressAll(t *testing.T) {
-	rule := SGRuleForACL{
-		IpProtocol: "-1",
-		CidrIp:     "0.0.0.0/0",
-	}
-	match := BuildEgressACLMatch("sg_test", rule)
-	assert.Contains(t, match, "inport == @sg_test")
-	assert.NotContains(t, match, "outport")
-	assert.Contains(t, match, "ip4")
-}
-
-func TestACL_TCPSinglePort(t *testing.T) {
-	rule := SGRuleForACL{
-		IpProtocol: "tcp",
-		FromPort:   443,
-		ToPort:     443,
-		CidrIp:     "10.0.0.0/8",
-	}
-	match := BuildIngressACLMatch("sg_test", rule)
-	assert.Contains(t, match, "tcp.dst == 443")
-	assert.NotContains(t, match, "tcp.dst >=")
-	assert.NotContains(t, match, "tcp.dst <=")
-}
-
-func TestACL_NoSource(t *testing.T) {
-	rule := SGRuleForACL{
-		IpProtocol: "tcp",
-		FromPort:   80,
-		ToPort:     80,
-		CidrIp:     "0.0.0.0/0",
-	}
-	match := BuildIngressACLMatch("sg_test", rule)
-	assert.Contains(t, match, "tcp.dst == 80")
-	assert.NotContains(t, match, "ip4.src")
-}

spinifex/services/vpcd/security.go

@@ -1,58 +1,88 @@
 package vpcd
 
+// Phase 2.6 (mulga-siv-129) thin NATS wrappers for SG lifecycle. Each handler
+// deserialises the event, translates to the policy/topology layer's spec
+// types, and calls the manager. ACL semantics, OVN port-group lifecycle and
+// the full all-or-nothing convergence model live in network/topology and
+// network/policy; this file owns NATS plumbing only.
+
 import (
 	"context"
 	"encoding/json"
 	"fmt"
 	"log/slog"
 
+	"github.com/mulgadc/spinifex/spinifex/network/policy"
 	"github.com/nats-io/nats.go"
 )
 
-// NATS topics for security group lifecycle events.
-const (
-	TopicCreateSG = "vpc.create-sg"
-	TopicDeleteSG = "vpc.delete-sg"
-	TopicUpdateSG = "vpc.update-sg"
-)
-
-// denyACLSeverity is the syslog severity OVN uses when a packet matches a
-// default-deny ACL. "info" is loud enough to be captured by a syslog forwarder
-// yet quiet enough to avoid paging on every dropped scan packet. Operators can
-// promote it at their log collector if they want higher-priority alerts.
-const denyACLSeverity = "info"
-
-// SGEvent carries security group state from the handler to vpcd.
-type SGEvent struct {
-	GroupId      string         `json:"group_id"`
-	VpcId        string         `json:"vpc_id"`
-	IngressRules []SGRuleForACL `json:"ingress_rules,omitempty"`
-	EgressRules  []SGRuleForACL `json:"egress_rules,omitempty"`
+// sgRule mirrors the on-wire payload from handlers/ec2/vpc.SGRule. Kept local
+// so vpcd does not import the handlers package.
+type sgRule struct {
+	IpProtocol string `json:"ip_protocol"`
+	FromPort   int64  `json:"from_port"`
+	ToPort     int64  `json:"to_port"`
+	CidrIp     string `json:"cidr_ip,omitempty"`
+	SourceSG   string `json:"source_sg,omitempty"`
 }
 
-// handleCreateSG creates an OVN Port Group and initial ACLs for a new security group.
+// sgEvent mirrors handlers/ec2/vpc.SGEvent.
+type sgEvent struct {
+	GroupId      string   `json:"group_id"`
+	VpcId        string   `json:"vpc_id"`
+	IngressRules []sgRule `json:"ingress_rules,omitempty"`
+	EgressRules  []sgRule `json:"egress_rules,omitempty"`
+}
+
+func (e sgEvent) toSpec() policy.SGSpec {
+	return policy.SGSpec{
+		GroupID:      e.GroupId,
+		VPCID:        e.VpcId,
+		IngressRules: toPolicyRules(e.IngressRules),
+		EgressRules:  toPolicyRules(e.EgressRules),
+	}
+}
+
+func toPolicyRules(in []sgRule) []policy.Rule {
+	out := make([]policy.Rule, len(in))
+	for i, r := range in {
+		out[i] = policy.Rule{
+			IPProtocol: r.IpProtocol,
+			FromPort:   r.FromPort,
+			ToPort:     r.ToPort,
+			CIDR:       r.CidrIp,
+			SourceSG:   r.SourceSG,
+		}
+	}
+	return out
+}
+
+// handleCreateSG ensures the OVN port group exists then applies the ACL set
+// for a new security group. Two-step (topology + policy) because port-group
+// lifecycle is L2 and ACL programming is L3 — see plan §8.1.
 func (h *TopologyHandler) handleCreateSG(msg *nats.Msg) {
 	if h.ovn == nil {
 		respond(msg, fmt.Errorf("OVN client not connected"))
 		return
 	}
-
-	var evt SGEvent
+	var evt sgEvent
 	if err := json.Unmarshal(msg.Data, &evt); err != nil {
 		slog.Error("vpcd: failed to unmarshal vpc.create-sg event", "err", err)
 		respond(msg, err)
 		return
 	}
-
 	ctx := context.Background()
-	if err := h.provisionSG(ctx, evt.GroupId, evt.IngressRules, evt.EgressRules); err != nil {
-		slog.Error("vpcd: failed to provision security group", "group_id", evt.GroupId, "err", err)
+	if err := h.EnsureSGPortGroup(ctx, evt.GroupId); err != nil {
+		slog.Error("vpcd: EnsureSGPortGroup failed", "group_id", evt.GroupId, "err", err)
 		respond(msg, err)
 		return
 	}
-
-	slog.Info("vpcd: created security group port group",
-		"pg", portGroupName(evt.GroupId),
+	if err := h.securityGroupManager().EnsureSG(ctx, evt.toSpec()); err != nil {
+		slog.Error("vpcd: EnsureSG failed", "group_id", evt.GroupId, "err", err)
+		respond(msg, err)
+		return
+	}
+	slog.Info("vpcd: created security group",
 		"group_id", evt.GroupId,
 		"vpc_id", evt.VpcId,
 		"ingress_rules", len(evt.IngressRules),
@@ -61,129 +91,49 @@ func (h *TopologyHandler) handleCreateSG(msg *nats.Msg) {
 	respond(msg, nil)
 }
 
-// provisionSG creates the OVN port group, default-deny ACLs, and allow ACLs
-// for the given SG. Used by handleCreateSG (CreateSecurityGroup path) and the
-// reconciler's scan-2 (SG record without OVN port group).
-//
-// SG-to-SG match expressions reference `$<pg>_ip4` / `$<pg>_ip6` — these
-// resolve against the SB Address_Set rows that ovn-northd auto-derives from
-// each port group's port addresses (ovn-nb(5): "For each port group, there
-// are two address sets generated to the Address_Set table of the
-// OVN_Southbound database, ... with name being the name of the Port_Group
-// followed by a suffix _ip4 / _ip6"). vpcd must NOT create explicit NB
-// Address_Set rows with those names; that produces a duplicate sync to SB
-// and wedges northd in an `OVNSB commit failed` retry loop.
-//
-// All-or-nothing: on any failure after the port group exists, the port group
-// and any partial ACLs are torn down so the reconciler's scan-2 can recreate
-// the SG from scratch on its next pass. A partial state would otherwise look
-// healthy to scanMissingPortGroups (port group exists) while silently
-// dropping legitimate traffic via the default-deny ACL.
-func (h *TopologyHandler) provisionSG(ctx context.Context, groupId string, ingress, egress []SGRuleForACL) error {
-	pgName := portGroupName(groupId)
-
-	if err := h.ovn.CreatePortGroup(ctx, pgName, nil); err != nil {
-		return fmt.Errorf("create port group %s: %w", pgName, err)
-	}
-
-	done := false
-	defer func() {
-		if done {
-			return
-		}
-		if err := h.ovn.ClearACLs(ctx, pgName); err != nil {
-			slog.Warn("vpcd: provisionSG cleanup ClearACLs failed", "pg", pgName, "err", err)
-		}
-		if err := h.ovn.DeletePortGroup(ctx, pgName); err != nil {
-			slog.Warn("vpcd: provisionSG cleanup DeletePortGroup failed", "pg", pgName, "err", err)
-		}
-	}()
-
-	// Default deny ACLs (priority 900, logged for CMMC SC.L1-3.13.1),
-	// infrastructure DHCP allows (priority 1050), and tenant allow rules
-	// (priority 1000) go in one OVSDB transaction so a 60-rule SG creation
-	// is one round-trip, not 64.
-	specs := append(infrastructureACLs(pgName), ruleACLSpecs(pgName, ingress, egress)...)
-	if err := h.ovn.AddACLs(ctx, pgName, specs); err != nil {
-		return fmt.Errorf("add ACLs on %s: %w", pgName, err)
-	}
-	done = true
-	return nil
-}
-
-// handleDeleteSG deletes the OVN Port Group and all associated ACLs.
+// handleDeleteSG removes the SG's port group (and its ACLs in one OVN
+// transaction). Idempotent on already-absent.
 func (h *TopologyHandler) handleDeleteSG(msg *nats.Msg) {
 	if h.ovn == nil {
 		respond(msg, fmt.Errorf("OVN client not connected"))
 		return
 	}
-
-	var evt SGEvent
+	var evt sgEvent
 	if err := json.Unmarshal(msg.Data, &evt); err != nil {
 		slog.Error("vpcd: failed to unmarshal vpc.delete-sg event", "err", err)
 		respond(msg, err)
 		return
 	}
-
 	ctx := context.Background()
-	pgName := portGroupName(evt.GroupId)
-
-	// Clear all ACLs before deleting the port group. Fail-fast — leaving stale
-	// ACLs causes DeletePortGroup to be rejected by libovsdb (dangling ref).
-	if err := h.ovn.ClearACLs(ctx, pgName); err != nil {
-		slog.Error("vpcd: failed to clear ACLs", "pg", pgName, "err", err)
+	if err := h.DeleteSGPortGroup(ctx, evt.GroupId); err != nil {
+		slog.Error("vpcd: DeleteSGPortGroup failed", "group_id", evt.GroupId, "err", err)
 		respond(msg, err)
 		return
 	}
-
-	if err := h.ovn.DeletePortGroup(ctx, pgName); err != nil {
-		slog.Error("vpcd: failed to delete port group", "pg", pgName, "err", err)
-		respond(msg, err)
-		return
-	}
-
-	slog.Info("vpcd: deleted security group port group",
-		"pg", pgName,
-		"group_id", evt.GroupId,
-		"vpc_id", evt.VpcId,
-	)
+	slog.Info("vpcd: deleted security group", "group_id", evt.GroupId, "vpc_id", evt.VpcId)
 	respond(msg, nil)
 }
 
-// handleUpdateSG replaces all ACLs for a security group with the current rule set.
+// handleUpdateSG replaces the SG's ACL set with the rules from the event.
+// Port-group lifecycle is unaffected — UpdateSG only touches ACLs.
 func (h *TopologyHandler) handleUpdateSG(msg *nats.Msg) {
 	if h.ovn == nil {
 		respond(msg, fmt.Errorf("OVN client not connected"))
 		return
 	}
-
-	var evt SGEvent
+	var evt sgEvent
 	if err := json.Unmarshal(msg.Data, &evt); err != nil {
 		slog.Error("vpcd: failed to unmarshal vpc.update-sg event", "err", err)
 		respond(msg, err)
 		return
 	}
-
 	ctx := context.Background()
-	pgName := portGroupName(evt.GroupId)
-
-	// Clear existing ACLs
-	if err := h.ovn.ClearACLs(ctx, pgName); err != nil {
-		slog.Error("vpcd: failed to clear ACLs for update", "pg", pgName, "err", err)
+	if err := h.securityGroupManager().UpdateSG(ctx, evt.toSpec()); err != nil {
+		slog.Error("vpcd: UpdateSG failed", "group_id", evt.GroupId, "err", err)
 		respond(msg, err)
 		return
 	}
-
-	// Re-add default deny + DHCP-infra allow ACLs + current rules in one transaction.
-	specs := append(infrastructureACLs(pgName), ruleACLSpecs(pgName, evt.IngressRules, evt.EgressRules)...)
-	if err := h.ovn.AddACLs(ctx, pgName, specs); err != nil {
-		slog.Error("vpcd: failed to add ACLs for update", "pg", pgName, "err", err)
-		respond(msg, err)
-		return
-	}
-
-	slog.Info("vpcd: updated security group ACLs",
-		"pg", pgName,
+	slog.Info("vpcd: updated security group",
 		"group_id", evt.GroupId,
 		"vpc_id", evt.VpcId,
 		"ingress_rules", len(evt.IngressRules),
@@ -191,94 +141,3 @@ func (h *TopologyHandler) handleUpdateSG(msg *nats.Msg) {
 	)
 	respond(msg, nil)
 }
-
-// ruleACLSpecs builds the priority-1000 allow ACL specs for the given rules.
-// Allow rules are not logged — accept logging on a private network is high
-// volume and low signal; only denies carry Log=true.
-func ruleACLSpecs(pgName string, ingress, egress []SGRuleForACL) []ACLSpec {
-	specs := make([]ACLSpec, 0, len(ingress)+len(egress))
-	for _, rule := range ingress {
-		specs = append(specs, ACLSpec{Direction: "to-lport", Priority: 1000, Match: BuildIngressACLMatch(pgName, rule), Action: "allow-related"})
-	}
-	for _, rule := range egress {
-		specs = append(specs, ACLSpec{Direction: "from-lport", Priority: 1000, Match: BuildEgressACLMatch(pgName, rule), Action: "allow-related"})
-	}
-	return specs
-}
-
-// denyIngressACL builds the default-deny ingress ACL for a port group with
-// logging enabled (CMMC SC.L1-3.13.1 boundary monitoring).
-func denyIngressACL(pgName string) ACLSpec {
-	return ACLSpec{
-		Direction: "to-lport",
-		Priority:  900,
-		Match:     fmt.Sprintf("outport == @%s && ip4", pgName),
-		Action:    "drop",
-		Name:      pgName + "-deny-ingress",
-		Log:       true,
-		Severity:  denyACLSeverity,
-	}
-}
-
-// denyEgressACL is the egress counterpart to denyIngressACL.
-func denyEgressACL(pgName string) ACLSpec {
-	return ACLSpec{
-		Direction: "from-lport",
-		Priority:  900,
-		Match:     fmt.Sprintf("inport == @%s && ip4", pgName),
-		Action:    "drop",
-		Name:      pgName + "-deny-egress",
-		Log:       true,
-		Severity:  denyACLSeverity,
-	}
-}
-
-// infrastructureACLs returns the platform-level ACLs that every port group
-// carries regardless of tenant rules: the priority-900 default-deny pair and
-// the priority-1050 DHCPv4 client/server allow pair. OVN's built-in DHCP
-// responder runs inside the LS pipeline *after* the port-group ACL stage, so
-// without these allows a narrow-egress tenant SG (e.g. egress restricted to
-// the VPC CIDR) silently drops the guest's DHCPDISCOVER (dst=255.255.255.255)
-// against the default-deny and the VM never gets an IPv4 lease. AWS users
-// can't write a rule for the limited broadcast address, so the platform must
-// allow DHCP itself — matching AWS's implicit behaviour.
-//
-// Priority 1050 sits above the default-deny (900) and above tenant rules
-// (1000) so DHCP works regardless of tenant configuration; tenants in our
-// model don't write ACLs at >1000, so there is no collision.
-func infrastructureACLs(pgName string) []ACLSpec {
-	return []ACLSpec{
-		denyIngressACL(pgName),
-		denyEgressACL(pgName),
-		dhcpEgressACL(pgName),
-		dhcpIngressACL(pgName),
-	}
-}
-
-// dhcpEgressACL allows the guest's DHCP client traffic (DHCPDISCOVER,
-// DHCPREQUEST: udp.src=68, udp.dst=67) out of the port. Action is plain
-// "allow", not "allow-related" — DHCP is single-shot request/reply and
-// allow-related on broadcast UDP interacts oddly with OVN's CT zones.
-func dhcpEgressACL(pgName string) ACLSpec {
-	return ACLSpec{
-		Direction: "from-lport",
-		Priority:  1050,
-		Match:     fmt.Sprintf("inport == @%s && udp && udp.src == 68 && udp.dst == 67", pgName),
-		Action:    "allow",
-		Name:      pgName + "-allow-dhcp-egress",
-	}
-}
-
-// dhcpIngressACL allows the OVN-generated DHCP server reply (DHCPOFFER,
-// DHCPACK: udp.src=67, udp.dst=68) back to the guest. OVN's DHCP responder
-// emits the reply from inside the LS pipeline, so without this matching
-// to-lport allow the reply hits the priority-900 ingress default-deny.
-func dhcpIngressACL(pgName string) ACLSpec {
-	return ACLSpec{
-		Direction: "to-lport",
-		Priority:  1050,
-		Match:     fmt.Sprintf("outport == @%s && udp && udp.src == 67 && udp.dst == 68", pgName),
-		Action:    "allow",
-		Name:      pgName + "-allow-dhcp-ingress",
-	}
-}

spinifex/services/vpcd/security_test.go

@@ -1,5 +1,11 @@
 package vpcd
 
+// Phase 2.6 (mulga-siv-129) tests for the SG NATS wrappers in security.go.
+// ACL emission semantics live in network/policy/sg_test.go +
+// network/policy/acl_test.go; tests here cover the wrapper layer:
+// event-shape errors, OVN-client wiring, and end-to-end NATS → OVN happy
+// paths through the new managers.
+
 import (
 	"context"
 	"encoding/json"
@@ -9,12 +15,16 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
-func TestSecurity_CreatePortGroup(t *testing.T) {
-	_, nc := startTestNATS(t)
+func sgTopoFixture(t *testing.T) (*TopologyHandler, *MockOVNClient) {
+	t.Helper()
 	mock := NewMockOVNClient()
 	_ = mock.Connect(context.Background())
+	return NewTopologyHandler(mock), mock
+}
 
-	topo := NewTopologyHandler(mock)
+func TestSG_CreateSG_HappyPath(t *testing.T) {
+	_, nc := startTestNATS(t)
+	topo, mock := sgTopoFixture(t)
 	subs, err := topo.Subscribe(nc)
 	require.NoError(t, err)
 	defer func() {
@@ -23,125 +33,23 @@ func TestSecurity_CreatePortGroup(t *testing.T) {
 		}
 	}()
 
-	evt := SGEvent{
-		GroupId: "sg-abc123",
-		VpcId:   "vpc-test1",
-	}
+	evt := sgEvent{GroupId: "sg-create1", VpcId: "vpc-test"}
 	data, _ := json.Marshal(evt)
 	resp, err := nc.Request(TopicCreateSG, data, 5_000_000_000)
 	require.NoError(t, err)
-	assertSuccess(t, resp, "create SG port group")
+	assertSuccess(t, resp, "create SG")
 
-	// Verify port group was created in the mock
 	mock.Mu.Lock()
-	pg, exists := mock.PortGroups["sg_abc123"]
+	_, pgExists := mock.PortGroups["sg_create1"]
+	aclCount := len(mock.ACLs)
 	mock.Mu.Unlock()
-	assert.True(t, exists, "port group sg_abc123 should exist")
-	assert.NotNil(t, pg)
-
-	// Verify the infrastructure ACL set was emitted: 2 default-deny at
-	// priority 900 (logged for CMMC SC.L1-3.13.1) plus 2 DHCP-infra allows
-	// at priority 1050 (unlogged — high-volume, low-signal).
-	type aclSnapshot struct {
-		action    string
-		log       bool
-		severity  string
-		name      string
-		priority  int
-		direction string
-		match     string
-	}
-	var snapshots []aclSnapshot
-	mock.Mu.Lock()
-	aclCount := len(pg.ACLs)
-	for _, aclUUID := range pg.ACLs {
-		a := mock.ACLs[aclUUID]
-		if a == nil {
-			continue
-		}
-		snap := aclSnapshot{
-			action:    a.Action,
-			log:       a.Log,
-			priority:  a.Priority,
-			direction: a.Direction,
-			match:     a.Match,
-		}
-		if a.Severity != nil {
-			snap.severity = *a.Severity
-		}
-		if a.Name != nil {
-			snap.name = *a.Name
-		}
-		snapshots = append(snapshots, snap)
-	}
-	mock.Mu.Unlock()
-
-	assert.Equal(t, 4, aclCount, "should have 4 infra ACLs (2 deny + 2 DHCP allow)")
-	denyCount, dhcpAllowCount := 0, 0
-	for _, s := range snapshots {
-		switch s.action {
-		case "drop":
-			denyCount++
-			assert.True(t, s.log, "default deny ACL must have log=true for boundary monitoring")
-			assert.Equal(t, "info", s.severity, "default deny ACL must use info severity")
-			assert.Contains(t, s.name, "sg_abc123-deny-", "default deny ACL must have a name for syslog correlation")
-			assert.Equal(t, 900, s.priority, "default deny ACL must be priority 900")
-		case "allow":
-			dhcpAllowCount++
-			assert.False(t, s.log, "DHCP infra allow must not be logged (high-volume, low-signal)")
-			assert.Equal(t, 1050, s.priority, "DHCP infra allow must be priority 1050 (above tenant rules and default-deny)")
-			assert.Contains(t, s.name, "sg_abc123-allow-dhcp-", "DHCP allow must have a name for operator correlation")
-		}
-	}
-	assert.Equal(t, 2, denyCount, "should have 2 drop ACLs")
-	assert.Equal(t, 2, dhcpAllowCount, "should have 2 DHCP allow ACLs (egress client→server, ingress server→client)")
+	assert.True(t, pgExists, "port group sg_create1 must exist after create-sg")
+	assert.Equal(t, 4, aclCount, "infra ACL set (2 deny + 2 DHCP) must be attached")
 }
 
-// TestSecurity_DHCPInfraACLs locks the exact shape of the DHCPv4 infra-allow
-// ACLs that vpcd emits on every port group. This is the fix for the
-// "narrow-egress SG silently breaks DHCP" bug: OVN's DHCP responder runs
-// inside the LS pipeline *after* the port-group ACL stage, so without these
-// the guest's DHCPDISCOVER (dst=255.255.255.255) hits the priority-900
-// default-deny under any non-0.0.0.0/0 egress rule and the VM never gets a
-// lease.
-func TestSecurity_DHCPInfraACLs(t *testing.T) {
-	const pg = "sg_dhcp_shape"
-
-	egress := dhcpEgressACL(pg)
-	assert.Equal(t, "from-lport", egress.Direction)
-	assert.Equal(t, 1050, egress.Priority, "must outrank tenant rules (1000) and default-deny (900)")
-	assert.Equal(t, "allow", egress.Action, "plain allow — DHCP is single-shot, no conntrack")
-	assert.Equal(t, "inport == @sg_dhcp_shape && udp && udp.src == 68 && udp.dst == 67", egress.Match)
-	assert.Equal(t, "sg_dhcp_shape-allow-dhcp-egress", egress.Name)
-	assert.False(t, egress.Log)
-
-	ingress := dhcpIngressACL(pg)
-	assert.Equal(t, "to-lport", ingress.Direction)
-	assert.Equal(t, 1050, ingress.Priority)
-	assert.Equal(t, "allow", ingress.Action)
-	assert.Equal(t, "outport == @sg_dhcp_shape && udp && udp.src == 67 && udp.dst == 68", ingress.Match)
-	assert.Equal(t, "sg_dhcp_shape-allow-dhcp-ingress", ingress.Name)
-	assert.False(t, ingress.Log)
-}
-
-// TestSecurity_InfrastructureACLs_Order locks the deny-before-allow ordering
-// that provisionSG/handleUpdateSG rely on when concatenating infrastructure
-// ACLs with tenant rule ACLs.
-func TestSecurity_InfrastructureACLs_Order(t *testing.T) {
-	specs := infrastructureACLs("sg_order")
-	require.Len(t, specs, 4)
-	assert.Equal(t, "sg_order-deny-ingress", specs[0].Name)
-	assert.Equal(t, "sg_order-deny-egress", specs[1].Name)
-	assert.Equal(t, "sg_order-allow-dhcp-egress", specs[2].Name)
-	assert.Equal(t, "sg_order-allow-dhcp-ingress", specs[3].Name)
-}
-
-func TestSecurity_DeletePortGroup(t *testing.T) {
+func TestSG_CreateSG_Idempotent(t *testing.T) {
 	_, nc := startTestNATS(t)
-	mock := NewMockOVNClient()
-	_ = mock.Connect(context.Background())
-
-	topo := NewTopologyHandler(mock)
+	topo, mock := sgTopoFixture(t)
 	subs, err := topo.Subscribe(nc)
 	require.NoError(t, err)
 	defer func() {
@@ -150,39 +58,51 @@ func TestSecurity_DeletePortGroup(t *testing.T) {
 		}
 	}()
 
-	// Create first
-	createEvt := SGEvent{GroupId: "sg-del1", VpcId: "vpc-test2"}
-	data, _ := json.Marshal(createEvt)
+	data, _ := json.Marshal(sgEvent{GroupId: "sg-idem", VpcId: "vpc-test"})
+	for range 2 {
+		resp, err := nc.Request(TopicCreateSG, data, 5_000_000_000)
+		require.NoError(t, err)
+		assertSuccess(t, resp, "create SG retry")
+	}
+
+	mock.Mu.Lock()
+	pgCount := len(mock.PortGroups)
+	mock.Mu.Unlock()
+	assert.Equal(t, 1, pgCount, "duplicate create-sg must not create a second port group")
+}
+
+func TestSG_DeleteSG_HappyPath(t *testing.T) {
+	_, nc := startTestNATS(t)
+	topo, mock := sgTopoFixture(t)
+	subs, err := topo.Subscribe(nc)
+	require.NoError(t, err)
+	defer func() {
+		for _, s := range subs {
+			_ = s.Unsubscribe()
+		}
+	}()
+
+	data, _ := json.Marshal(sgEvent{GroupId: "sg-del1", VpcId: "vpc-test"})
 	resp, err := nc.Request(TopicCreateSG, data, 5_000_000_000)
 	require.NoError(t, err)
-	assertSuccess(t, resp, "create SG for delete test")
+	assertSuccess(t, resp, "create SG before delete")
 
-	// Verify it exists
-	mock.Mu.Lock()
-	_, exists := mock.PortGroups["sg_del1"]
-	mock.Mu.Unlock()
-	assert.True(t, exists)
-
-	// Delete
-	delEvt := SGEvent{GroupId: "sg-del1", VpcId: "vpc-test2"}
-	data, _ = json.Marshal(delEvt)
 	resp, err = nc.Request(TopicDeleteSG, data, 5_000_000_000)
 	require.NoError(t, err)
-	assertSuccess(t, resp, "delete SG port group")
+	assertSuccess(t, resp, "delete SG")
 
-	// Verify removed
 	mock.Mu.Lock()
-	_, exists = mock.PortGroups["sg_del1"]
+	_, pgExists := mock.PortGroups["sg_del1"]
 	mock.Mu.Unlock()
-	assert.False(t, exists, "port group sg_del1 should be deleted")
+	assert.False(t, pgExists, "port group must be gone after delete-sg")
 }
 
-func TestSecurity_UpdateSGAddRules(t *testing.T) {
+// Delete of a never-created SG must succeed: the new wrapper delegates to
+// topology.Manager.DeleteSGPortGroup which is idempotent on absent port
+// groups (the reconciler retries delete events that race with vm-cleanup).
+func TestSG_DeleteSG_IdempotentOnAbsent(t *testing.T) {
 	_, nc := startTestNATS(t)
-	mock := NewMockOVNClient()
-	_ = mock.Connect(context.Background())
-
-	topo := NewTopologyHandler(mock)
+	topo, _ := sgTopoFixture(t)
 	subs, err := topo.Subscribe(nc)
 	require.NoError(t, err)
 	defer func() {
@@ -191,83 +111,15 @@ func TestSecurity_UpdateSGAddRules(t *testing.T) {
 		}
 	}()
 
-	// Create SG first (no rules)
-	createEvt := SGEvent{GroupId: "sg-upd1", VpcId: "vpc-test3"}
-	data, _ := json.Marshal(createEvt)
-	resp, err := nc.Request(TopicCreateSG, data, 5_000_000_000)
+	data, _ := json.Marshal(sgEvent{GroupId: "sg-ghost", VpcId: "vpc-test"})
+	resp, err := nc.Request(TopicDeleteSG, data, 5_000_000_000)
 	require.NoError(t, err)
-	assertSuccess(t, resp, "create SG for update test")
-
-	// Update with ingress rules
-	updateEvt := SGEvent{
-		GroupId: "sg-upd1",
-		VpcId:   "vpc-test3",
-		IngressRules: []SGRuleForACL{
-			{IpProtocol: "tcp", FromPort: 22, ToPort: 22, CidrIp: "10.0.0.0/8"},
-			{IpProtocol: "tcp", FromPort: 443, ToPort: 443, CidrIp: "0.0.0.0/0"},
-		},
-	}
-	data, _ = json.Marshal(updateEvt)
-	resp, err = nc.Request(TopicUpdateSG, data, 5_000_000_000)
-	require.NoError(t, err)
-	assertSuccess(t, resp, "update SG with ingress rules")
-
-	// Verify ACLs were created: 2 default deny + 2 ingress allow = 4
-	type aclSnapshot struct {
-		action string
-		match  string
-		log    bool
-	}
-	var snapshots []aclSnapshot
-	mock.Mu.Lock()
-	pg := mock.PortGroups["sg_upd1"]
-	aclCount := len(pg.ACLs)
-	for _, aclUUID := range pg.ACLs {
-		a := mock.ACLs[aclUUID]
-		if a == nil {
-			continue
-		}
-		snapshots = append(snapshots, aclSnapshot{action: a.Action, match: a.Match, log: a.Log})
-	}
-	mock.Mu.Unlock()
-
-	assert.Equal(t, 6, aclCount, "should have 6 ACLs (2 deny + 2 DHCP infra allow + 2 ingress allow)")
-
-	// Check that at least one match contains tcp.dst == 22. Also verify
-	// logging policy: denies logged, allows not logged (CMMC SC.L1-3.13.1).
-	foundSSH := false
-	foundHTTPS := false
-	for _, s := range snapshots {
-		switch s.action {
-		case "drop":
-			assert.True(t, s.log, "deny ACL must be logged")
-		case "allow-related":
-			assert.False(t, s.log, "allow ACL must not be logged (high volume, low signal)")
-		}
-		if containsAll(s.match, "tcp.dst == 22", "ip4.src == 10.0.0.0/8") {
-			foundSSH = true
-		}
-		if containsAll(s.match, "tcp.dst == 443") {
-			foundHTTPS = true
-		}
-	}
-	assert.True(t, foundSSH, "should have SSH ACL with source CIDR")
-	assert.True(t, foundHTTPS, "should have HTTPS ACL")
+	assertSuccess(t, resp, "delete absent SG must be a no-op")
 }
 
-// TestSecurity_CreateSG_FailsFastOnAddACLError locks the Phase 1.1 invariant
-// that handleCreateSG returns an AddACLs error to the caller and tears down
-// the half-built SG. The mock fails the batched AddACLs call, and the test
-// asserts the handler responds with failure AND that the port group is
-// gone — so the next reconciler scan recreates from scratch instead of
-// observing a half-built PG with no ACLs.
-func TestSecurity_CreateSG_FailsFastOnAddACLError(t *testing.T) {
+func TestSG_UpdateSG_ReplacesACLSet(t *testing.T) {
 	_, nc := startTestNATS(t)
-	mock := NewMockOVNClient()
-	mock.AddACLErrAfter = 1
-	_ = mock.Connect(context.Background())
-
-	topo := NewTopologyHandler(mock)
+	topo, mock := sgTopoFixture(t)
 	subs, err := topo.Subscribe(nc)
 	require.NoError(t, err)
 	defer func() {
@@ -276,103 +128,40 @@ func TestSecurity_CreateSG_FailsFastOnAddACLError(t *testing.T) {
 		}
 	}()
 
-	evt := SGEvent{GroupId: "sg-failacl", VpcId: "vpc-failacl"}
-	data, _ := json.Marshal(evt)
-	resp, err := nc.Request(TopicCreateSG, data, 5_000_000_000)
-	require.NoError(t, err)
-	assertFailure(t, resp, "create SG must fail when AddACL fails")
-
-	mock.Mu.Lock()
-	_, pgExists := mock.PortGroups["sg_failacl"]
-	mock.Mu.Unlock()
-	assert.False(t, pgExists, "port group must be torn down on failed provisionSG")
-}
-
-// TestSecurity_UpdateSG_FailsFastOnAddACLError locks the same fail-fast policy
-// in handleUpdateSG: clear ACLs, re-add deny ACLs, then add allow ACLs — any
-// failure must propagate to the caller instead of being slog.Warn-ed away.
-func TestSecurity_UpdateSG_FailsFastOnAddACLError(t *testing.T) {
-	_, nc := startTestNATS(t)
-	mock := NewMockOVNClient()
-	_ = mock.Connect(context.Background())
-
-	topo := NewTopologyHandler(mock)
-	subs, err := topo.Subscribe(nc)
-	require.NoError(t, err)
-	defer func() {
-		for _, s := range subs {
-			_ = s.Unsubscribe()
-		}
-	}()
-
-	createEvt := SGEvent{GroupId: "sg-updfail", VpcId: "vpc-updfail"}
-	data, _ := json.Marshal(createEvt)
-	resp, err := nc.Request(TopicCreateSG, data, 5_000_000_000)
+	createData, _ := json.Marshal(sgEvent{GroupId: "sg-upd", VpcId: "vpc-test"})
+	resp, err := nc.Request(TopicCreateSG, createData, 5_000_000_000)
 	require.NoError(t, err)
 	assertSuccess(t, resp, "create SG before update")
 
-	mock.Mu.Lock()
-	mock.AddACLCalls = 0
-	mock.AddACLErrAfter = 1
-	mock.Mu.Unlock()
-
-	updateEvt := SGEvent{
-		GroupId: "sg-updfail",
-		VpcId:   "vpc-updfail",
-		IngressRules: []SGRuleForACL{
+	updateData, _ := json.Marshal(sgEvent{
+		GroupId: "sg-upd",
+		VpcId:   "vpc-test",
+		IngressRules: []sgRule{
 			{IpProtocol: "tcp", FromPort: 22, ToPort: 22, CidrIp: "10.0.0.0/8"},
-			{IpProtocol: "tcp", FromPort: 443, ToPort: 443, CidrIp: "0.0.0.0/0"},
 		},
-	}
-	data, _ = json.Marshal(updateEvt)
-	resp, err = nc.Request(TopicUpdateSG, data, 5_000_000_000)
+	})
+	resp, err = nc.Request(TopicUpdateSG, updateData, 5_000_000_000)
 	require.NoError(t, err)
-	assertFailure(t, resp, "update SG must fail when AddACL fails mid-batch")
-}
+	assertSuccess(t, resp, "update SG")
 
-// TestSecurity_CreateSG_WithRules_FailsFastOnAddACLError locks the fail-fast
-// policy when handleCreateSG carries allow rules: the batched AddACLs call
-// includes both deny ACLs and rule ACLs, and a failure must still tear down
-// the PG/AS so the reconciler doesn't mistake a half-built PG as healthy.
-func TestSecurity_CreateSG_WithRules_FailsFastOnAddACLError(t *testing.T) {
-	_, nc := startTestNATS(t)
-	mock := NewMockOVNClient()
-	mock.AddACLErrAfter = 1
-	_ = mock.Connect(context.Background())
-
-	topo := NewTopologyHandler(mock)
-	subs, err := topo.Subscribe(nc)
-	require.NoError(t, err)
-	defer func() {
-		for _, s := range subs {
-			_ = s.Unsubscribe()
+	mock.Mu.Lock()
+	pg := mock.PortGroups["sg_upd"]
+	aclCount := 0
+	hasTenantRule := false
+	if pg != nil {
+		aclCount = len(pg.ACLs)
+		for _, uuid := range pg.ACLs {
+			if acl := mock.ACLs[uuid]; acl != nil && acl.Priority == 1000 {
+				hasTenantRule = true
+			}
 		}
-	}()
-
-	evt := SGEvent{
-		GroupId: "sg-rulefail",
-		VpcId:   "vpc-rulefail",
-		IngressRules: []SGRuleForACL{
-			{IpProtocol: "tcp", FromPort: 22, ToPort: 22, CidrIp: "10.0.0.0/8"},
-			{IpProtocol: "tcp", FromPort: 443, ToPort: 443, CidrIp: "0.0.0.0/0"},
-		},
 	}
-	data, _ := json.Marshal(evt)
-	resp, err := nc.Request(TopicCreateSG, data, 5_000_000_000)
-	require.NoError(t, err)
-	assertFailure(t, resp, "create SG with rules must fail when AddACLs fails")
-
-	mock.Mu.Lock()
-	_, pgExists := mock.PortGroups["sg_rulefail"]
 	mock.Mu.Unlock()
-	assert.False(t, pgExists, "port group must be torn down when AddACLs fails")
+	assert.Equal(t, 5, aclCount, "update must leave 4 infra + 1 tenant ACL on the port group")
+	assert.True(t, hasTenantRule, "tenant priority-1000 ACL must be present after update")
 }
 
-// TestSecurity_HandleCreateSG_OVNNotConnected, ...HandleDeleteSG_..., and
-// ...HandleUpdateSG_... lock the defensive guard at the top of each handler:
-// when vpcd's OVN client failed to connect (or has been Closed), the SG event
-// must surface a clear error to the caller instead of nil-deref'ing.
-func TestSecurity_HandleCreateSG_OVNNotConnected(t *testing.T) {
+func TestSG_HandleCreateSG_OVNNotConnected(t *testing.T) {
 	_, nc := startTestNATS(t)
 	topo := NewTopologyHandler(nil)
 	subs, err := topo.Subscribe(nc)
@@ -383,13 +172,13 @@ func TestSecurity_HandleCreateSG_OVNNotConnected(t *testing.T) {
 		}
 	}()
 
-	data, _ := json.Marshal(SGEvent{GroupId: "sg-noovn", VpcId: "vpc-noovn"})
+	data, _ := json.Marshal(sgEvent{GroupId: "sg-noovn"})
 	resp, err := nc.Request(TopicCreateSG, data, 5_000_000_000)
 	require.NoError(t, err)
 	assertFailure(t, resp, "create SG must fail when ovn client is nil")
 }
 
-func TestSecurity_HandleDeleteSG_OVNNotConnected(t *testing.T) {
+func TestSG_HandleDeleteSG_OVNNotConnected(t *testing.T) {
 	_, nc := startTestNATS(t)
 	topo := NewTopologyHandler(nil)
 	subs, err := topo.Subscribe(nc)
@@ -400,13 +189,13 @@ func TestSecurity_HandleDeleteSG_OVNNotConnected(t *testing.T) {
 		}
 	}()
 
-	data, _ := json.Marshal(SGEvent{GroupId: "sg-noovn", VpcId: "vpc-noovn"})
+	data, _ := json.Marshal(sgEvent{GroupId: "sg-noovn"})
 	resp, err := nc.Request(TopicDeleteSG, data, 5_000_000_000)
 	require.NoError(t, err)
 	assertFailure(t, resp, "delete SG must fail when ovn client is nil")
 }
 
-func TestSecurity_HandleUpdateSG_OVNNotConnected(t *testing.T) {
+func TestSG_HandleUpdateSG_OVNNotConnected(t *testing.T) {
 	_, nc := startTestNATS(t)
 	topo := NewTopologyHandler(nil)
 	subs, err := topo.Subscribe(nc)
@@ -417,20 +206,15 @@ func TestSecurity_HandleUpdateSG_OVNNotConnected(t *testing.T) {
 		}
 	}()
 
-	data, _ := json.Marshal(SGEvent{GroupId: "sg-noovn", VpcId: "vpc-noovn"})
+	data, _ := json.Marshal(sgEvent{GroupId: "sg-noovn"})
 	resp, err := nc.Request(TopicUpdateSG, data, 5_000_000_000)
 	require.NoError(t, err)
 	assertFailure(t, resp, "update SG must fail when ovn client is nil")
 }
 
-// TestSecurity_HandleCreateSG_BadJSON / ..._HandleDeleteSG_BadJSON /
-// ..._HandleUpdateSG_BadJSON: malformed event payloads must surface as
-// failures, not be silently dropped — the handler-side caller relies on the
-// ack to know the request was processed.
-func TestSecurity_HandleCreateSG_BadJSON(t *testing.T) {
+func TestSG_HandleCreateSG_BadJSON(t *testing.T) {
 	_, nc := startTestNATS(t)
-	topo := NewTopologyHandler(NewMockOVNClient())
-	_ = topo.ovn.Connect(context.Background())
+	topo, _ := sgTopoFixture(t)
 	subs, err := topo.Subscribe(nc)
 	require.NoError(t, err)
 	defer func() {
@@ -444,10 +228,9 @@ func TestSecurity_HandleCreateSG_BadJSON(t *testing.T) {
 	assertFailure(t, resp, "create SG with malformed payload must fail")
 }
 
-func TestSecurity_HandleDeleteSG_BadJSON(t *testing.T) {
+func TestSG_HandleDeleteSG_BadJSON(t *testing.T) {
 	_, nc := startTestNATS(t)
-	topo := NewTopologyHandler(NewMockOVNClient())
-	_ = topo.ovn.Connect(context.Background())
+	topo, _ := sgTopoFixture(t)
 	subs, err := topo.Subscribe(nc)
 	require.NoError(t, err)
 	defer func() {
@@ -461,10 +244,9 @@ func TestSecurity_HandleDeleteSG_BadJSON(t *testing.T) {
 	assertFailure(t, resp, "delete SG with malformed payload must fail")
 }
 
-func TestSecurity_HandleUpdateSG_BadJSON(t *testing.T) {
+func TestSG_HandleUpdateSG_BadJSON(t *testing.T) {
 	_, nc := startTestNATS(t)
-	topo := NewTopologyHandler(NewMockOVNClient())
-	_ = topo.ovn.Connect(context.Background())
+	topo, _ := sgTopoFixture(t)
 	subs, err := topo.Subscribe(nc)
 	require.NoError(t, err)
 	defer func() {
@@ -477,122 +259,3 @@ func TestSecurity_HandleUpdateSG_BadJSON(t *testing.T) {
 	require.NoError(t, err)
 	assertFailure(t, resp, "update SG with malformed payload must fail")
 }
-
-// TestSecurity_HandleDeleteSG_FailsFastOnClearACLsError: handleDeleteSG must
-// surface the ClearACLs error rather than swallow it. We trigger this by
-// requesting a delete for an SG that was never created — the mock's ClearACLs
-// returns "port group not found", which must propagate as a failed response.
-func TestSecurity_HandleDeleteSG_FailsFastOnClearACLsError(t *testing.T) {
-	_, nc := startTestNATS(t)
-	mock := NewMockOVNClient()
-	_ = mock.Connect(context.Background())
-	topo := NewTopologyHandler(mock)
-	subs, err := topo.Subscribe(nc)
-	require.NoError(t, err)
-	defer func() {
-		for _, s := range subs {
-			_ = s.Unsubscribe()
-		}
-	}()
-
-	data, _ := json.Marshal(SGEvent{GroupId: "sg-never-created", VpcId: "vpc-x"})
-	resp, err := nc.Request(TopicDeleteSG, data, 5_000_000_000)
-	require.NoError(t, err)
-	assertFailure(t, resp, "delete must fail when port group does not exist")
-}
-
-// TestSecurity_HandleUpdateSG_FailsFastOnClearACLsError: handleUpdateSG must
-// also surface the ClearACLs error — silently swallowing it would leave the
-// caller thinking the rule update succeeded when in fact the old ACLs are
-// still in place.
-func TestSecurity_HandleUpdateSG_FailsFastOnClearACLsError(t *testing.T) {
-	_, nc := startTestNATS(t)
-	mock := NewMockOVNClient()
-	_ = mock.Connect(context.Background())
-	topo := NewTopologyHandler(mock)
-	subs, err := topo.Subscribe(nc)
-	require.NoError(t, err)
-	defer func() {
-		for _, s := range subs {
-			_ = s.Unsubscribe()
-		}
-	}()
-
-	data, _ := json.Marshal(SGEvent{
-		GroupId:      "sg-never-created",
-		VpcId:        "vpc-x",
-		IngressRules: []SGRuleForACL{{IpProtocol: "tcp", FromPort: 22, ToPort: 22, CidrIp: "0.0.0.0/0"}},
-	})
-	resp, err := nc.Request(TopicUpdateSG, data, 5_000_000_000)
-	require.NoError(t, err)
-	assertFailure(t, resp, "update must fail when port group does not exist")
-}
-
-// TestSecurity_HandleDeleteSG_Idempotent_AddressSetMissing: if the address set
-// has already been cleaned up out-of-band but the port group still exists, the
-// delete must NOT propagate the AS-not-found error — the goal state is "both
-// gone", and we've already achieved it for the AS. Locks the matching
-// fail-fast invariant in TestSecurity_DeleteSG_FailsFastOnAddressSetError
-// (already covered) by exercising the inverse direction: a fresh delete after
-// a successful create must succeed end-to-end without errors leaking.
-//
-// (kept here to avoid duplicating the create→delete fixture; the existing
-// happy-path test exercises the success branch but does not assert that
-// neither half-state lingers.)
-func TestSecurity_HandleDeleteSG_LeavesNoResidualState(t *testing.T) {
-	_, nc := startTestNATS(t)
-	mock := NewMockOVNClient()
-	_ = mock.Connect(context.Background())
-	topo := NewTopologyHandler(mock)
-	subs, err := topo.Subscribe(nc)
-	require.NoError(t, err)
-	defer func() {
-		for _, s := range subs {
-			_ = s.Unsubscribe()
-		}
-	}()
-
-	data, _ := json.Marshal(SGEvent{GroupId: "sg-residual", VpcId: "vpc-x"})
-	resp, err := nc.Request(TopicCreateSG, data, 5_000_000_000)
-	require.NoError(t, err)
-	assertSuccess(t, resp, "create SG")
-
-	resp, err = nc.Request(TopicDeleteSG, data, 5_000_000_000)
-	require.NoError(t, err)
-	assertSuccess(t, resp, "delete SG")
-
-	mock.Mu.Lock()
-	_, pgExists := mock.PortGroups["sg_residual"]
-	leftoverACLs := 0
-	for _, a := range mock.ACLs {
-		if a != nil && (a.Match == "outport == @sg_residual && ip4" || a.Match == "inport == @sg_residual && ip4") {
-			leftoverACLs++
-		}
-	}
-	mock.Mu.Unlock()
-	assert.False(t, pgExists, "port group must be gone")
-	assert.Zero(t, leftoverACLs, "no ACL rows must reference the deleted PG")
-}
-
-// containsAll checks if s contains all substrings.
-func containsAll(s string, subs ...string) bool {
-	for _, sub := range subs {
-		if !contains(s, sub) {
-			return false
-		}
-	}
-	return true
-}
-
-func contains(s, sub string) bool {
-	return len(s) >= len(sub) && searchString(s, sub)
-}
-
-func searchString(s, sub string) bool {
-	for i := 0; i <= len(s)-len(sub); i++ {
-		if s[i:i+len(sub)] == sub {
-			return true
-		}
-	}
-	return false
-}

spinifex/services/vpcd/topology.go

@@ -12,6 +12,7 @@ import (
 	"sync"
 	"time"
 
+	"github.com/mulgadc/spinifex/spinifex/network/policy"
 	"github.com/mulgadc/spinifex/spinifex/network/topology"
 	"github.com/mulgadc/spinifex/spinifex/services/vpcd/dhcp"
 	"github.com/mulgadc/spinifex/spinifex/services/vpcd/nbdb"
@@ -69,6 +70,9 @@ const (
 	TopicDeleteNAT        = "vpc.delete-nat"
 	TopicAddNATGateway    = "vpc.add-nat-gateway"
 	TopicDeleteNATGateway = "vpc.delete-nat-gateway"
+	TopicCreateSG         = "vpc.create-sg"
+	TopicDeleteSG         = "vpc.delete-sg"
+	TopicUpdateSG         = "vpc.update-sg"
 )
 
 // gatewayPortNetwork is the link-local CIDR every IGW gateway-LRP carries
@@ -179,6 +183,11 @@ type TopologyHandler struct {
 	// adapter call doesn't re-construct (Phase 2.6, mulga-siv-129).
 	lmOnce sync.Once
 	lm     topology.Manager
+
+	// sgm + sgmOnce cache the policy.SecurityGroupManager backed by h.ovn so
+	// SG subscribers reuse one manager instance across events.
+	sgmOnce sync.Once
+	sgm     policy.SecurityGroupManager
 }
 
 // NewTopologyHandler creates a new TopologyHandler with optional external network config.
@@ -841,7 +850,7 @@ func (h *TopologyHandler) handleCreatePort(msg *nats.Msg) {
 	// port group's port addresses once the LSP joins.
 	pgNames := make([]string, 0, len(evt.SecurityGroupIds))
 	for _, sgId := range evt.SecurityGroupIds {
-		pgNames = append(pgNames, portGroupName(sgId))
+		pgNames = append(pgNames, topology.SecurityGroupPortGroup(sgId))
 	}
 	if err := h.ovn.CreateLogicalSwitchPortInGroups(ctx, switchName, lsp, pgNames); err != nil {
 		slog.Error("vpcd: failed to create logical switch port", "port", portName, "switch", switchName, "err", err)
@@ -959,7 +968,7 @@ func (h *TopologyHandler) handleUpdatePortSGs(msg *nats.Msg) {
 func (h *TopologyHandler) reconcilePortSGs(ctx context.Context, lspName string, desiredSGs []string) (bool, error) {
 	desired := make(map[string]struct{}, len(desiredSGs))
 	for _, sgId := range desiredSGs {
-		desired[portGroupName(sgId)] = struct{}{}
+		desired[topology.SecurityGroupPortGroup(sgId)] = struct{}{}
 	}
 
 	currentNames, err := h.ovn.ListPortGroupsForPort(ctx, lspName)

spinifex/services/vpcd/topology_manager.go

@@ -14,6 +14,7 @@ import (
 	"context"
 	"strings"
 
+	"github.com/mulgadc/spinifex/spinifex/network/policy"
 	"github.com/mulgadc/spinifex/spinifex/network/topology"
 )
 
@@ -77,3 +78,23 @@ func (h *TopologyHandler) DeletePort(ctx context.Context, spec topology.PortSpec
 func (h *TopologyHandler) SetPortSecurityGroups(ctx context.Context, portID string, sgIDs []string) error {
 	return h.topologyManager().SetPortSecurityGroups(ctx, portID, sgIDs)
 }
+
+// EnsureSGPortGroup creates the OVN port group for a security group.
+func (h *TopologyHandler) EnsureSGPortGroup(ctx context.Context, groupID string) error {
+	return h.topologyManager().EnsureSGPortGroup(ctx, groupID)
+}
+
+// DeleteSGPortGroup removes the OVN port group for a security group.
+func (h *TopologyHandler) DeleteSGPortGroup(ctx context.Context, groupID string) error {
+	return h.topologyManager().DeleteSGPortGroup(ctx, groupID)
+}
+
+// securityGroupManager returns the lazily-constructed policy SG manager
+// backed by the same OVN client as the live topology manager. Subscribers
+// use it to apply ACL sets against the SG's port group.
+func (h *TopologyHandler) securityGroupManager() policy.SecurityGroupManager {
+	h.sgmOnce.Do(func() {
+		h.sgm = policy.NewSecurityGroupManager(h.ovn)
+	})
+	return h.sgm
+}

spinifex/services/vpcd/topology_test.go

@@ -10,6 +10,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/mulgadc/spinifex/spinifex/network/topology"
 	"github.com/mulgadc/spinifex/spinifex/services/vpcd/dhcp"
 	"github.com/mulgadc/spinifex/spinifex/services/vpcd/nbdb"
 	"github.com/mulgadc/spinifex/spinifex/testutil"
@@ -3570,7 +3571,7 @@ func setupPortSGFixture(t *testing.T, mock *MockOVNClient, ctx context.Context,
 		t.Fatalf("CreateDHCPOptions: %v", err)
 	}
 	for _, sgId := range sgIds {
-		pg := portGroupName(sgId)
+		pg := topology.SecurityGroupPortGroup(sgId)
 		if err := mock.CreatePortGroup(ctx, pg, nil); err != nil {
 			t.Fatalf("CreatePortGroup %s: %v", sgId, err)
 		}
@@ -3617,7 +3618,7 @@ func TestTopologyHandler_CreatePort_JoinsPortGroups(t *testing.T) {
 	}
 
 	for _, sgId := range []string{"sg-1111", "sg-2222"} {
-		pg := mock.PortGroups[portGroupName(sgId)]
+		pg := mock.PortGroups[topology.SecurityGroupPortGroup(sgId)]
 		if !slices.Contains(pg.Ports, lsp.UUID) {
 			t.Errorf("expected %s membership in port group %s, got %v", lsp.UUID, sgId, pg.Ports)
 		}
@@ -3695,7 +3696,7 @@ func TestTopologyHandler_DeletePort_ClearsPortGroupMembership(t *testing.T) {
 	resp, _ := nc.Request(TopicCreatePort, data, 5_000_000_000)
 	assertSuccess(t, resp, "create port")
 
-	if got := len(mock.PortGroups[portGroupName("sg-del")].Ports); got != 1 {
+	if got := len(mock.PortGroups[topology.SecurityGroupPortGroup("sg-del")].Ports); got != 1 {
 		t.Fatalf("setup: expected 1 member in port group, got %d", got)
 	}
 
@@ -3710,7 +3711,7 @@ func TestTopologyHandler_DeletePort_ClearsPortGroupMembership(t *testing.T) {
 	resp, _ = nc.Request(TopicDeletePort, data, 5_000_000_000)
 	assertSuccess(t, resp, "delete port")
 
-	if got := len(mock.PortGroups[portGroupName("sg-del")].Ports); got != 0 {
+	if got := len(mock.PortGroups[topology.SecurityGroupPortGroup("sg-del")].Ports); got != 0 {
 		t.Errorf("expected 0 members after delete-port, got %d", got)
 	}
 }
@@ -3792,7 +3793,7 @@ func TestTopologyHandler_UpdatePortSGs_Transitions(t *testing.T) {
 			}
 			expected := make(map[string]struct{}, len(tc.expected))
 			for _, s := range tc.expected {
-				expected[portGroupName(s)] = struct{}{}
+				expected[topology.SecurityGroupPortGroup(s)] = struct{}{}
 			}
 			for name, pg := range mock.PortGroups {
 				inGroup := slices.Contains(pg.Ports, lsp.UUID)
@@ -3836,7 +3837,7 @@ func TestTopologyHandler_CreatePort_ExistingLSP_ReconcilesSGMemberships(t *testi
 		Name:      "port-eni-recover",
 		Addresses: []string{"02:00:00:00:00:99 10.0.6.50"},
 	}))
-	pgName := portGroupName("sg-recover")
+	pgName := topology.SecurityGroupPortGroup("sg-recover")
 	require.NotContains(t, mock.PortGroups[pgName].Ports, mock.Ports["port-eni-recover"].UUID)
 
 	evt := PortEvent{