feat: neutralise potentially-committed NAT rule on rollback (#249)

spinifex/daemon/daemon_system_instance.go

@@ -227,7 +227,22 @@ func (d *Daemon) LaunchSystemInstance(input *handlers_elbv2.SystemInstanceInput)
 				vpcID = *result.NetworkInterfaces[0].VpcId
 			}
 			portName := "port-" + instance.ENIId
-			utils.PublishNATEvent(d.natsConn, "vpc.add-nat", vpcID, publicIP, privateIP, portName, instance.ENIMac)
+			if natErr := utils.AddNAT(d.natsConn, vpcID, publicIP, privateIP, portName, instance.ENIMac); natErr != nil {
+				slog.Error("LaunchSystemInstance: vpc.add-nat failed for ALB public IP — rolling back to avoid surfacing an unreachable address",
+					"instanceId", instance.ID, "publicIp", publicIP, "pool", poolName, "err", natErr)
+				// Timeout may have committed the rule after our window; neutralise before releasing the IP.
+				utils.PublishNATEvent(d.natsConn, "vpc.delete-nat", vpcID, publicIP, privateIP, portName, instance.ENIMac)
+				if clearErr := d.vpcService.UpdateENIPublicIP(eniAccountID, instance.ENIId, "", ""); clearErr != nil {
+					slog.Warn("LaunchSystemInstance: failed to clear ENI public IP during NAT-failure rollback",
+						"eniId", instance.ENIId, "publicIp", publicIP, "err", clearErr)
+				}
+				if relErr := d.externalIPAM.ReleaseIP(poolName, publicIP); relErr != nil {
+					slog.Warn("LaunchSystemInstance: failed to release public IP during NAT-failure rollback",
+						"publicIp", publicIP, "pool", poolName, "err", relErr)
+				}
+				d.cleanupFailedSystemInstance(instance, instanceType)
+				return nil, fmt.Errorf("commit NAT rule for internet-facing ALB public IP %s: %w", publicIP, natErr)
+			}
 			instance.PublicIP = publicIP
 			instance.PublicIPPool = poolName
 			slog.Info("LaunchSystemInstance: allocated public IP via direct IPAM",

spinifex/daemon/daemon_system_instance_test.go

@@ -1,6 +1,7 @@
 package daemon
 
 import (
+	"encoding/json"
 	"fmt"
 	"os"
 	"path/filepath"
@@ -8,10 +9,14 @@ import (
 	"testing"
 	"time"
 
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/service/ec2"
+	handlers_ec2_vpc "github.com/mulgadc/spinifex/spinifex/handlers/ec2/vpc"
 	"github.com/mulgadc/spinifex/spinifex/handlers/elbv2"
 	"github.com/mulgadc/spinifex/spinifex/tags"
 	"github.com/mulgadc/spinifex/spinifex/testutil"
 	"github.com/mulgadc/spinifex/spinifex/vm"
+	"github.com/nats-io/nats-server/v2/server"
 	"github.com/nats-io/nats.go"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -449,3 +454,128 @@ func TestRefreshSystemInstanceState_WriteErrorPropagates(t *testing.T) {
 	require.Error(t, err)
 	assert.Contains(t, err.Error(), "rewrite fw_cfg blobs")
 }
+
+// TestLaunchSystemInstance_NATFailureRollsBackPublicIP regresses the silent
+// corruption where an internet-facing ALB launch committed the IPAM
+// allocation and surfaced the public IP on the ENI record even when
+// vpc.add-nat failed — the OVN dnat_and_snat rule was never created and the
+// public IP black-holed with no reconciler. With the fix, NAT NACK must roll
+// back the ENI public IP and release the IPAM lease before returning.
+func TestLaunchSystemInstance_NATFailureRollsBackPublicIP(t *testing.T) {
+	d := createVPCTestDaemon(t)
+
+	// Wire an ExternalIPAM against a fresh JS server. The pool gives us a
+	// known IP that the rollback must release back.
+	jsNS, err := server.NewServer(&server.Options{
+		Host:      "127.0.0.1",
+		Port:      -1,
+		JetStream: true,
+		StoreDir:  t.TempDir(),
+		NoLog:     true,
+		NoSigs:    true,
+	})
+	require.NoError(t, err)
+	go jsNS.Start()
+	require.True(t, jsNS.ReadyForConnections(5*time.Second))
+	t.Cleanup(func() { jsNS.Shutdown() })
+
+	jsNC, err := nats.Connect(jsNS.ClientURL())
+	require.NoError(t, err)
+	t.Cleanup(func() { jsNC.Close() })
+
+	js, err := jsNC.JetStream()
+	require.NoError(t, err)
+	ipam, err := handlers_ec2_vpc.NewExternalIPAM(jsNC, js, []handlers_ec2_vpc.ExternalPoolConfig{
+		{Name: "wan-test", RangeStart: "203.0.113.10", RangeEnd: "203.0.113.20", Gateway: "203.0.113.1", PrefixLen: 24},
+	})
+	require.NoError(t, err)
+	d.externalIPAM = ipam
+
+	// Create a VPC + subnet + ENI so LaunchSystemInstance has a real ENI to
+	// attach to and to update with the public IP.
+	vpcOut, err := d.vpcService.CreateVpc(&ec2.CreateVpcInput{
+		CidrBlock: aws.String("10.50.0.0/16"),
+	}, testAccountID)
+	require.NoError(t, err)
+	subnetOut, err := d.vpcService.CreateSubnet(&ec2.CreateSubnetInput{
+		VpcId:     vpcOut.Vpc.VpcId,
+		CidrBlock: aws.String("10.50.1.0/24"),
+	}, testAccountID)
+	require.NoError(t, err)
+	eniOut, err := d.vpcService.CreateNetworkInterface(&ec2.CreateNetworkInterfaceInput{
+		SubnetId:    subnetOut.Subnet.SubnetId,
+		Description: aws.String("nat-fail-test"),
+	}, testAccountID)
+	require.NoError(t, err)
+	eniID := aws.StringValue(eniOut.NetworkInterface.NetworkInterfaceId)
+	eniMac := aws.StringValue(eniOut.NetworkInterface.MacAddress)
+	eniIP := aws.StringValue(eniOut.NetworkInterface.PrivateIpAddress)
+
+	// Stand up a vpcd-shaped NACK responder on the daemon's NATS conn —
+	// utils.AddNAT publishes on d.natsConn, so the responder must live there.
+	sub, err := d.natsConn.Subscribe("vpc.add-nat", func(msg *nats.Msg) {
+		_ = msg.Respond([]byte(`{"success":false,"error":"northd unavailable"}`))
+	})
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = sub.Unsubscribe() })
+
+	// Capture vpc.delete-nat so we can assert the rollback neutralises any
+	// rule vpcd may have committed after the AddNAT timeout window.
+	deleteNATCh := make(chan map[string]string, 1)
+	delSub, err := d.natsConn.Subscribe("vpc.delete-nat", func(msg *nats.Msg) {
+		var p map[string]string
+		_ = json.Unmarshal(msg.Data, &p)
+		select {
+		case deleteNATCh <- p:
+		default:
+		}
+	})
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = delSub.Unsubscribe() })
+
+	input := &handlers_elbv2.SystemInstanceInput{
+		InstanceType: getTestInstanceType(t),
+		SubnetID:     aws.StringValue(subnetOut.Subnet.SubnetId),
+		Scheme:       handlers_elbv2.SchemeInternetFacing,
+		AccountID:    testAccountID,
+		ENIID:        eniID,
+		ENIMac:       eniMac,
+		ENIIP:        eniIP,
+	}
+
+	out, err := d.LaunchSystemInstance(input)
+	require.Error(t, err, "NAT NACK must fail the launch")
+	assert.Nil(t, out, "no SystemInstanceOutput should be returned on NAT failure")
+	assert.Contains(t, err.Error(), "northd unavailable", "error must propagate the vpcd reason")
+
+	// ENI public IP record must be cleared so subsequent DescribeNetworkInterfaces
+	// does not surface the unreachable address.
+	descOut, err := d.vpcService.DescribeNetworkInterfaces(&ec2.DescribeNetworkInterfacesInput{
+		NetworkInterfaceIds: []*string{aws.String(eniID)},
+	}, testAccountID)
+	require.NoError(t, err)
+	require.Len(t, descOut.NetworkInterfaces, 1)
+	// Association is built only when the underlying record has a public IP;
+	// rollback clears that field, so the helper omits Association entirely.
+	assert.Nil(t, descOut.NetworkInterfaces[0].Association,
+		"ENI must not carry the unreachable public IP after rollback")
+
+	// IPAM lease must be back in the pool so the next allocation can reuse
+	// it. .10 is the gateway (reserved at init); the first allocable IP is
+	// .11, which is what the launch grabbed and the rollback released.
+	rec, err := ipam.GetPoolRecord("wan-test")
+	require.NoError(t, err)
+	_, stillAllocated := rec.Allocated["203.0.113.11"]
+	assert.False(t, stillAllocated, "rollback must release the allocated IP back to the pool")
+
+	// vpc.delete-nat must be published with the same external IP that AddNAT
+	// targeted so vpcd reaps any rule committed after the request timeout.
+	select {
+	case got := <-deleteNATCh:
+		assert.Equal(t, "203.0.113.11", got["external_ip"])
+		assert.Equal(t, eniIP, got["logical_ip"])
+		assert.Equal(t, "port-"+eniID, got["port_name"])
+	case <-time.After(time.Second):
+		t.Fatal("rollback must publish vpc.delete-nat to neutralise a half-committed rule")
+	}
+}

spinifex/daemon/vpc_adapter.go

@@ -47,6 +47,10 @@ func (a *daemonENICreator) AttachENI(accountID, eniID, instanceID string, device
 	return a.d.vpcService.AttachENI(accountID, eniID, instanceID, deviceIndex)
 }
 
+func (a *daemonENICreator) DetachENI(accountID, eniID string) error {
+	return a.d.vpcService.DetachENI(accountID, eniID)
+}
+
 func (a *daemonENICreator) UpdateENIPublicIP(accountID, eniID, publicIP, poolName string) error {
 	return a.d.vpcService.UpdateENIPublicIP(accountID, eniID, publicIP, poolName)
 }

spinifex/handlers/ec2/instance/service.go

@@ -118,12 +118,16 @@ type SubnetInfo struct {
 
 // ENICreator covers the VPC/ENI operations RunInstances performs while
 // auto-attaching a primary interface. Implemented via an adapter over
-// handlers/ec2/vpc.VPCServiceImpl on the daemon.
+// handlers/ec2/vpc.VPCServiceImpl on the daemon. DetachENI lives here (not on
+// ENIDeleter) because the launch-time NAT-rollback path needs to flip the ENI
+// to "available" before ENIDeleter.DeleteNetworkInterface, which rejects
+// in-use ENIs.
 type ENICreator interface {
 	GetDefaultSubnet(accountID string) (*SubnetInfo, error)
 	GetSubnet(accountID, subnetID string) (*SubnetInfo, error)
 	CreateNetworkInterface(input *ec2.CreateNetworkInterfaceInput, accountID string) (*ec2.CreateNetworkInterfaceOutput, error)
 	AttachENI(accountID, eniID, instanceID string, deviceIndex int64) (string, error)
+	DetachENI(accountID, eniID string) error
 	UpdateENIPublicIP(accountID, eniID, publicIP, poolName string) error
 }
 

spinifex/handlers/ec2/instance/service_impl.go

@@ -680,7 +680,16 @@ func (s *InstanceServiceImpl) PrepareRunInstances(input *ec2.RunInstancesInput,
 							slog.Warn("PrepareRunInstances: failed to update ENI with public IP", "eniId", *eni.NetworkInterfaceId, "err", updateErr)
 						}
 						portName := "port-" + *eni.NetworkInterfaceId
-						utils.PublishNATEvent(s.natsConn, "vpc.add-nat", *eni.VpcId, publicIP, *eni.PrivateIpAddress, portName, *eni.MacAddress)
+						if natErr := utils.AddNAT(s.natsConn, *eni.VpcId, publicIP, *eni.PrivateIpAddress, portName, *eni.MacAddress); natErr != nil {
+							slog.Error("PrepareRunInstances: vpc.add-nat failed — rolling back public IP to avoid surfacing an unreachable address",
+								"instanceId", instance.ID, "publicIp", publicIP, "pool", poolName, "err", natErr)
+							// Timeout may have committed the rule after our window; neutralise before releasing the IP.
+							utils.PublishNATEvent(s.natsConn, "vpc.delete-nat", *eni.VpcId, publicIP, *eni.PrivateIpAddress, portName, *eni.MacAddress)
+							s.rollbackAutoAssignedPublicIP(accountID, instance.ID, *eni.NetworkInterfaceId, publicIP, poolName)
+							lastRunErr = natErr
+							s.resourceMgr.Deallocate(instanceType)
+							continue
+						}
 						ec2Instance.PublicIpAddress = aws.String(publicIP)
 						instance.PublicIP = publicIP
 						instance.PublicIPPool = poolName
@@ -2262,6 +2271,43 @@ func (s *InstanceServiceImpl) deleteInstanceVolumes(instance *vm.VM, instanceID
 	_ = instanceID
 }
 
+// rollbackAutoAssignedPublicIP unwinds an auto-assign that proceeded past
+// ENI create + attach + IPAM allocation + ENI update but failed at the
+// vpc.add-nat barrier. Order matters: clear the ENI record first so any
+// concurrent DescribeNetworkInterfaces stops surfacing the unreachable IP,
+// then release the IPAM lease so the pool slot can be reused, then detach
+// and delete the ENI itself — the failing instance never enters vmMgr, so
+// terminateCleanup will not run and the ENI would otherwise leak. Detach
+// must precede DeleteNetworkInterface; the latter rejects in-use ENIs.
+func (s *InstanceServiceImpl) rollbackAutoAssignedPublicIP(accountID, instanceID, eniID, publicIP, poolName string) {
+	if s.eniCreator != nil {
+		if err := s.eniCreator.UpdateENIPublicIP(accountID, eniID, "", ""); err != nil {
+			slog.Warn("PrepareRunInstances: failed to clear ENI public IP during NAT-failure rollback",
+				"eniId", eniID, "publicIp", publicIP, "err", err)
+		}
+	}
+	if s.ipReleaser != nil {
+		if err := s.ipReleaser.ReleaseIP(poolName, publicIP); err != nil {
+			slog.Warn("PrepareRunInstances: failed to release public IP during NAT-failure rollback",
+				"publicIp", publicIP, "pool", poolName, "err", err)
+		}
+	}
+	if s.eniCreator != nil {
+		if err := s.eniCreator.DetachENI(accountID, eniID); err != nil {
+			slog.Warn("PrepareRunInstances: failed to detach ENI during NAT-failure rollback",
+				"eniId", eniID, "instanceId", instanceID, "err", err)
+		}
+	}
+	if s.eniDeleter != nil {
+		if _, err := s.eniDeleter.DeleteNetworkInterface(&ec2.DeleteNetworkInterfaceInput{
+			NetworkInterfaceId: &eniID,
+		}, accountID); err != nil {
+			slog.Warn("PrepareRunInstances: failed to delete ENI during NAT-failure rollback",
+				"eniId", eniID, "instanceId", instanceID, "err", err)
+		}
+	}
+}
+
 func (s *InstanceServiceImpl) releaseInstancePublicIP(instance *vm.VM, instanceID string) {
 	if instance.PublicIP == "" || instance.PublicIPPool == "" || s.ipReleaser == nil {
 		return

spinifex/handlers/ec2/instance/service_impl_test.go

@@ -2366,6 +2366,8 @@ type fakeENICreator struct {
 	createErr     error
 	attachCalls   int
 	updateCalls   int
+	clearCalls    int // updateCalls where publicIP is ""
+	detachCalls   int
 }
 
 func (f *fakeENICreator) GetDefaultSubnet(_ string) (*SubnetInfo, error) {
@@ -2394,8 +2396,16 @@ func (f *fakeENICreator) AttachENI(_, _, _ string, _ int64) (string, error) {
 	return "attached", nil
 }
 
-func (f *fakeENICreator) UpdateENIPublicIP(_, _, _, _ string) error {
+func (f *fakeENICreator) DetachENI(_, _ string) error {
+	f.detachCalls++
+	return nil
+}
+
+func (f *fakeENICreator) UpdateENIPublicIP(_, _, publicIP, _ string) error {
 	f.updateCalls++
+	if publicIP == "" {
+		f.clearCalls++
+	}
 	return nil
 }
 
@@ -2476,6 +2486,14 @@ func TestPrepareRunInstances_PublicIPAutoAssigned(t *testing.T) {
 	ipam := &fakeIPAllocator{publicIP: "203.0.113.5", poolName: "pool-a"}
 	svc, _ := prepareSvcWithENI(t, eni, ipam)
 
+	// vpc.add-nat is now request-reply: the helper waits for vpcd to ack
+	// before the launch proceeds. The happy path needs a success responder.
+	sub, err := svc.natsConn.Subscribe("vpc.add-nat", func(msg *nats.Msg) {
+		_ = msg.Respond([]byte(`{"success":true}`))
+	})
+	require.NoError(t, err)
+	defer func() { _ = sub.Unsubscribe() }()
+
 	_, instances, _, err := svc.PrepareRunInstances(&ec2.RunInstancesInput{
 		InstanceType: aws.String("t3.micro"),
 		ImageId:      aws.String("ami-1"),
@@ -2490,6 +2508,112 @@ func TestPrepareRunInstances_PublicIPAutoAssigned(t *testing.T) {
 	assert.Equal(t, 1, eni.updateCalls)
 }
 
+// TestPrepareRunInstances_NATFailureRollsBackPublicIP regresses the silent
+// corruption where vpc.add-nat NACKs (or vpcd is offline) but the launch
+// returned an ec2.Instance carrying the unreachable IP, the ENI record was
+// updated with that IP, and the IPAM pool entry stayed allocated. With the
+// fix in place, NAT failure must drop the instance from the response, clear
+// the ENI public IP, release the IPAM lease, and deallocate capacity.
+func TestPrepareRunInstances_NATFailureRollsBackPublicIP(t *testing.T) {
+	eni := &fakeENICreator{
+		subnet: &SubnetInfo{SubnetID: "subnet-1", VpcID: "vpc-1", MapPublicIpOnLaunch: true},
+		createOut: &ec2.CreateNetworkInterfaceOutput{
+			NetworkInterface: &ec2.NetworkInterface{
+				NetworkInterfaceId: aws.String("eni-nat-fail"),
+				MacAddress:         aws.String("aa:bb:cc:dd:ee:ff"),
+				PrivateIpAddress:   aws.String("10.0.0.30"),
+				VpcId:              aws.String("vpc-1"),
+			},
+		},
+	}
+	ipam := &fakeIPAllocator{publicIP: "203.0.113.7", poolName: "pool-a"}
+	releaser := &fakePublicIPReleaser{}
+	deleter := &fakeENIDeleter{}
+	svc, prov := prepareSvcWithENI(t, eni, ipam)
+	svc.ipReleaser = releaser
+	svc.eniDeleter = deleter
+
+	// Stand up a vpcd-shaped responder that NACKs every add-nat — simulates
+	// vpcd online but unable to commit the OVN rule (northd lag, port
+	// missing, etc.).
+	sub, err := svc.natsConn.Subscribe("vpc.add-nat", func(msg *nats.Msg) {
+		_ = msg.Respond([]byte(`{"success":false,"error":"northd unavailable"}`))
+	})
+	require.NoError(t, err)
+	defer func() { _ = sub.Unsubscribe() }()
+
+	// Record vpc.delete-nat to assert the rollback neutralises any
+	// half-committed rule from a waitForFlowsHV timeout.
+	deleteNATCh := make(chan natWirePayload, 1)
+	delSub, err := svc.natsConn.Subscribe("vpc.delete-nat", func(msg *nats.Msg) {
+		var p natWirePayload
+		_ = json.Unmarshal(msg.Data, &p)
+		select {
+		case deleteNATCh <- p:
+		default:
+		}
+	})
+	require.NoError(t, err)
+	defer func() { _ = delSub.Unsubscribe() }()
+
+	_, instances, _, err := svc.PrepareRunInstances(&ec2.RunInstancesInput{
+		InstanceType: aws.String("t3.micro"),
+		ImageId:      aws.String("ami-1"),
+		SubnetId:     aws.String("subnet-1"),
+		MinCount:     aws.Int64(1),
+		MaxCount:     aws.Int64(1),
+	}, "acc")
+
+	// MinCount=1 with a single launch attempt that NAT-failed must drop
+	// below minimum and return ServerInternal — the wrapped natErr is not
+	// an AWS-known code, so the lookup misses and the wire fallback kicks
+	// in. The critical invariant is that no instance with the unreachable
+	// IP is surfaced.
+	require.Error(t, err)
+	assert.Equal(t, awserrors.ErrorServerInternal, err.Error())
+	assert.Empty(t, instances, "instance with failed NAT must not be returned")
+
+	// IPAM lease released back to the pool.
+	assert.Equal(t, "pool-a", releaser.pool, "IPAM ReleaseIP must be called with the originally allocated pool")
+	assert.Equal(t, "203.0.113.7", releaser.ip, "IPAM ReleaseIP must be called with the originally allocated IP")
+
+	// ENI record cleared: 2 update calls total (set + clear), 1 clear.
+	assert.Equal(t, 2, eni.updateCalls, "ENI public IP should be set once, then cleared on rollback")
+	assert.Equal(t, 1, eni.clearCalls, "ENI rollback should call UpdateENIPublicIP with empty publicIP")
+
+	// ENI itself must be detached + deleted; otherwise the dropped instance
+	// leaks an eniKV record (vmMgr.Insert never runs, so terminateCleanup
+	// never fires) and the subnet exhausts private IPs under vpcd brown-out.
+	assert.Equal(t, 1, eni.detachCalls, "rollback must detach the ENI before delete")
+	assert.Equal(t, []string{"eni-nat-fail"}, deleter.calls, "rollback must delete the auto-created ENI")
+
+	// Capacity deallocated so subsequent launches can use the slot.
+	require.Len(t, prov.deallocated, 1, "NAT failure must trigger Deallocate")
+
+	// vpc.delete-nat must be published so vpcd reaps any rule it committed
+	// after the AddNAT timeout — otherwise the orphan rule outlives the
+	// IPAM allocation and routes traffic for the next tenant that grabs
+	// the released IP.
+	select {
+	case got := <-deleteNATCh:
+		assert.Equal(t, "vpc-1", got.VpcId)
+		assert.Equal(t, "203.0.113.7", got.ExternalIP)
+		assert.Equal(t, "10.0.0.30", got.LogicalIP)
+		assert.Equal(t, "port-eni-nat-fail", got.PortName)
+	case <-time.After(time.Second):
+		t.Fatal("rollback must publish vpc.delete-nat to neutralise a half-committed rule")
+	}
+}
+
+// natWirePayload mirrors utils.natEvent for test-side decoding.
+type natWirePayload struct {
+	VpcId      string `json:"vpc_id"`
+	ExternalIP string `json:"external_ip"`
+	LogicalIP  string `json:"logical_ip"`
+	PortName   string `json:"port_name"`
+	MAC        string `json:"mac"`
+}
+
 func TestPrepareRunInstances_ENICreateFailureDeallocates(t *testing.T) {
 	eni := &fakeENICreator{createErr: errors.New("boom")}
 	svc, prov := prepareSvcWithENI(t, eni, nil)

spinifex/utils/nats.go

@@ -385,17 +385,42 @@ func PublishEvent(nc *nats.Conn, topic string, event any) {
 	}
 }
 
+// natEvent is the wire payload for vpc.add-nat / vpc.delete-nat.
+type natEvent struct {
+	VpcId      string `json:"vpc_id"`
+	ExternalIP string `json:"external_ip"`
+	LogicalIP  string `json:"logical_ip"`
+	PortName   string `json:"port_name"`
+	MAC        string `json:"mac"`
+}
+
+// AddNAT requests vpcd commit the OVN dnat_and_snat rule for (externalIP →
+// logicalIP) via NATS request-reply and waits up to 10 s for the ack. A
+// non-nil return means the rule was either NOT committed (NACK / no
+// responders) OR may have been committed after the timeout fired (vpcd's
+// waitForFlowsHV barrier is unbounded). Callers MUST roll back any state
+// that assumes the public IP is routable AND publish a best-effort
+// vpc.delete-nat for the same (vpcID, externalIP, logicalIP, portName) to
+// neutralise a half-committed rule — there is no reconciler that repairs
+// a black-holed external IP after the fact.
+func AddNAT(nc *nats.Conn, vpcID, externalIP, logicalIP, portName, mac string) error {
+	return RequestEvent(nc, "vpc.add-nat", natEvent{
+		VpcId: vpcID, ExternalIP: externalIP, LogicalIP: logicalIP,
+		PortName: portName, MAC: mac,
+	}, 10*time.Second)
+}
+
 // PublishNATEvent sends a NAT lifecycle event (vpc.add-nat or vpc.delete-nat)
 // to NATS. vpc.add-nat uses request-reply to commit the OVN NAT rule before
 // returning (prevents ARP propagation races); vpc.delete-nat is fire-and-forget.
+//
+// Callers that MUST react to an add-nat failure (e.g. roll back a launch)
+// should use AddNAT instead — this helper only logs the failure.
 func PublishNATEvent(nc *nats.Conn, topic, vpcID, externalIP, logicalIP, portName, mac string) {
-	evt := struct {
-		VpcId      string `json:"vpc_id"`
-		ExternalIP string `json:"external_ip"`
-		LogicalIP  string `json:"logical_ip"`
-		PortName   string `json:"port_name"`
-		MAC        string `json:"mac"`
-	}{VpcId: vpcID, ExternalIP: externalIP, LogicalIP: logicalIP, PortName: portName, MAC: mac}
+	evt := natEvent{
+		VpcId: vpcID, ExternalIP: externalIP, LogicalIP: logicalIP,
+		PortName: portName, MAC: mac,
+	}
 
 	if topic == "vpc.add-nat" {
 		if err := RequestEvent(nc, topic, evt, 10*time.Second); err != nil {

spinifex/utils/nats_test.go

@@ -821,3 +821,61 @@ func TestConnectNATSWithRetry_NoEscalation_BelowThreshold(t *testing.T) {
 	assert.NotContains(t, logs, "NATS still disconnected", "should not escalate before threshold")
 	assert.Contains(t, logs, "NATS not ready, retrying...", "should still log warn lines")
 }
+
+// TestAddNAT_Success pins that AddNAT returns nil only when vpcd acks the
+// add-nat request with {"success":true}. The wire payload must match the
+// natEvent shape vpcd unmarshals on the other end.
+func TestAddNAT_Success(t *testing.T) {
+	ns := startTestNATSServer(t)
+	nc, err := nats.Connect(ns.ClientURL())
+	require.NoError(t, err)
+	defer nc.Close()
+
+	var got natEvent
+	_, err = nc.Subscribe("vpc.add-nat", func(msg *nats.Msg) {
+		_ = json.Unmarshal(msg.Data, &got)
+		_ = msg.Respond([]byte(`{"success":true}`))
+	})
+	require.NoError(t, err)
+
+	err = AddNAT(nc, "vpc-1", "203.0.113.5", "10.0.0.5", "port-eni-1", "02:00:00:00:00:01")
+	require.NoError(t, err)
+	assert.Equal(t, "vpc-1", got.VpcId)
+	assert.Equal(t, "203.0.113.5", got.ExternalIP)
+	assert.Equal(t, "10.0.0.5", got.LogicalIP)
+	assert.Equal(t, "port-eni-1", got.PortName)
+	assert.Equal(t, "02:00:00:00:00:01", got.MAC)
+}
+
+// TestAddNAT_NACK is the core regression for the silent-corruption bug: when
+// vpcd reports failure, callers MUST see a non-nil error so they can roll
+// back the IPAM allocation and the ENI public IP record. Before the fix the
+// helper logged a warning and returned nothing, leaving an unreachable
+// public IP surfaced to the AWS client.
+func TestAddNAT_NACK(t *testing.T) {
+	ns := startTestNATSServer(t)
+	nc, err := nats.Connect(ns.ClientURL())
+	require.NoError(t, err)
+	defer nc.Close()
+
+	_, err = nc.Subscribe("vpc.add-nat", func(msg *nats.Msg) {
+		_ = msg.Respond([]byte(`{"success":false,"error":"northd unavailable"}`))
+	})
+	require.NoError(t, err)
+
+	err = AddNAT(nc, "vpc-1", "203.0.113.5", "10.0.0.5", "port-eni-1", "02:00:00:00:00:01")
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "northd unavailable")
+}
+
+// TestAddNAT_NoResponders ensures a vpcd outage (no subscriber on the topic)
+// surfaces as an error rather than a swallowed warning.
+func TestAddNAT_NoResponders(t *testing.T) {
+	ns := startTestNATSServer(t)
+	nc, err := nats.Connect(ns.ClientURL())
+	require.NoError(t, err)
+	defer nc.Close()
+
+	err = AddNAT(nc, "vpc-1", "203.0.113.5", "10.0.0.5", "port-eni-1", "02:00:00:00:00:01")
+	require.Error(t, err)
+}