Merge branch 'dev' of github.com:mulgadc/spinifex into dev

docs/COMMANDS.md

@@ -113,7 +113,7 @@ Operational commands for inspecting cluster state. These fan out NATS requests t
 
 | Command | Implemented Flags | Missing Flags | Status |
 |---------|-------------------|---------------|--------|
-| `run-instances` | `--image-id`, `--instance-type`, `--count`, `--key-name`, `--user-data`, `--subnet-id`, `--security-group-ids`, `--tag-specifications` (instance-scoped), `--block-device-mappings` (DeviceName, VolumeSize, VolumeType, Iops, DeleteOnTermination), `--placement` (GroupName), `--iam-instance-profile` (Name/Arn) | `--dry-run`, `--client-token`, `--disable-api-termination`, `--ebs-optimized`, `--network-interfaces`, `--private-ip-address`, `--monitoring`, `--credit-specification`, `--cpu-options`, `--metadata-options`, `--launch-template`, `--hibernate-options` | **DONE** |
+| `run-instances` | `--image-id`, `--instance-type`, `--count`, `--key-name`, `--user-data`, `--subnet-id`, `--security-group-ids`, `--tag-specifications` (instance-scoped), `--block-device-mappings` (DeviceName, VolumeSize, VolumeType, Iops, DeleteOnTermination), `--placement` (GroupName), `--iam-instance-profile` (Name/Arn), `--capacity-reservation-specification` (CapacityReservationTarget.CapacityReservationId, targeted-by-id only) | `--dry-run`, `--client-token`, `--disable-api-termination`, `--ebs-optimized`, `--network-interfaces`, `--private-ip-address`, `--monitoring`, `--credit-specification`, `--cpu-options`, `--metadata-options`, `--launch-template`, `--hibernate-options` | **DONE** |
 | `describe-instances` | `--instance-ids`, `--filters` (instance-state-name, instance-id, instance-type, vpc-id, subnet-id, tag:*, tag-key, tag-value) | `--max-results`, `--next-token`, `--dry-run` | **DONE** |
 | `start-instances` | `--instance-ids` | `--dry-run`, `--force` | **DONE** |
 | `stop-instances` | `--instance-ids` | `--force`, `--hibernate`, `--dry-run` | **DONE** |
@@ -132,6 +132,11 @@ Operational commands for inspecting cluster state. These fan out NATS requests t
 role ARN and rejects cross-account profile ARNs. Placement strategies: `spread`
 (1 per node, atomic CAS) and `cluster` (pin all to one node).
 
+`run-instances --capacity-reservation-specification` consumes a reserved slot on
+the node owning the targeted reservation (targeted-by-id only; `open`
+auto-matching is not supported). The combination with `--placement` (group) is
+rejected. `describe-instances` then reports the instance's `CapacityReservationId`.
+
 ### EC2 — IAM Instance Profile Associations
 
 Stored as `vm.VM.IamInstanceProfileArn` + `IamInstanceProfileAssociationId`

docs/service-interfaces.yaml

@@ -292,6 +292,14 @@ suites:
     # via vpcd, AMI via predastore, root volume via viperblockd, lifecycle via
     # spinifex_daemon. Single-mode, VM-booting leg.
 
+  e2e-ecr:
+    path: tests/e2e/ecr
+    covers: [awsgw, spinifex_daemon, predastore]
+    # ECR control plane (handlers/ecr, gateway/ecr*) + OCI data plane: blobs and
+    # manifests in predastore, control + auth bridge in awsgw, KV via
+    # spinifex_daemon. docker/crane/skopeo data-plane subtests skip when the
+    # client binary or registry-host resolution is absent.
+
   e2e-ddil:
     path: tests/e2e/ddil
     covers: [awsgw, spinifex_daemon, vpcd, viperblockd, predastore]

spinifex/daemon/capacityreservation.go

@@ -4,6 +4,8 @@ import (
 	"errors"
 	"time"
 
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/service/ec2"
 	"github.com/mulgadc/spinifex/spinifex/awserrors"
 )
 
@@ -22,10 +24,13 @@ type capacityReservation struct {
 	Tenancy               string
 	InstancePlatform      string
 	CreateDate            time.Time
+	ConsumedCount         int // instances launched into this reservation; slots moved reservedCR*->allocated*
 }
 
-// reservationCompute returns the headline compute carve-out for a reservation:
-// InstanceCount × catalog (vCPU, memory), with no nbdkit overhead.
+// reservationCompute returns the full compute carve-out for a reservation:
+// TotalInstanceCount × per-instance (vCPU, memory). MemGBPerInstance already
+// folds in nbdkit overhead (set at create), so the reservedCR->allocated swap on
+// launch is net-zero.
 func (r *capacityReservation) reservationCompute() (vcpu int, memGB float64) {
 	return r.TotalInstanceCount * r.VCPUPerInstance, float64(r.TotalInstanceCount) * r.MemGBPerInstance
 }
@@ -52,10 +57,10 @@ func (rm *ResourceManager) CreateReservation(rec *capacityReservation) error {
 	return nil
 }
 
-// CancelReservation releases the carve-out for an account-owned reservation and
-// removes it from the map, then refreshes subscriptions. Returns the removed
-// record and true on success, or (nil, false) if no reservation with that id is
-// owned by the account.
+// CancelReservation releases the unconsumed remainder of an account-owned
+// reservation's carve-out and removes it; consumed slots already moved to
+// allocated* and keep running. Returns the removed record and true on success,
+// or (nil, false) if no reservation with that id is owned by the account.
 func (rm *ResourceManager) CancelReservation(id, accountID string) (*capacityReservation, bool) {
 	rm.mu.Lock()
 	rec, ok := rm.reservations[id]
@@ -63,9 +68,9 @@ func (rm *ResourceManager) CancelReservation(id, accountID string) (*capacityRes
 		rm.mu.Unlock()
 		return nil, false
 	}
-	freedVCPU, freedMem := rec.reservationCompute()
-	rm.reservedCRVCPU -= freedVCPU
-	rm.reservedCRMem -= freedMem
+	remaining := rec.TotalInstanceCount - rec.ConsumedCount
+	rm.reservedCRVCPU -= remaining * rec.VCPUPerInstance
+	rm.reservedCRMem -= float64(remaining) * rec.MemGBPerInstance
 	delete(rm.reservations, id)
 	rm.mu.Unlock()
 
@@ -73,16 +78,105 @@ func (rm *ResourceManager) CancelReservation(id, accountID string) (*capacityRes
 	return rec, true
 }
 
-// ListReservations returns a snapshot of the account's reservations on this node.
-// Records are immutable, so the returned pointers are safe to read concurrently.
-func (rm *ResourceManager) ListReservations(accountID string) []*capacityReservation {
+// AllocateFromReservation launches one instance into reservation crID via a
+// net-zero swap under rm.mu: the per-instance compute already held in reservedCR*
+// at create moves to allocated* and ConsumedCount bumps, so schedulable capacity
+// is unchanged by construction. Errors: unknown/foreign id ->
+// InvalidCapacityReservationId.NotFound; type mismatch -> InvalidParameterValue;
+// already full -> ReservationCapacityExceeded.
+func (rm *ResourceManager) AllocateFromReservation(crID, accountID string, it *ec2.InstanceTypeInfo) error {
+	rm.mu.Lock()
+	rec, ok := rm.reservations[crID]
+	if !ok || rec.AccountID != accountID {
+		rm.mu.Unlock()
+		return errors.New(awserrors.ErrorInvalidCapacityReservationIdNotFound)
+	}
+	if rec.InstanceType != aws.StringValue(it.InstanceType) {
+		rm.mu.Unlock()
+		return errors.New(awserrors.ErrorInvalidParameterValue)
+	}
+	if rec.ConsumedCount >= rec.TotalInstanceCount {
+		rm.mu.Unlock()
+		return errors.New(awserrors.ErrorReservationCapacityExceeded)
+	}
+	rm.reservedCRVCPU -= rec.VCPUPerInstance
+	rm.reservedCRMem -= rec.MemGBPerInstance
+	rm.allocatedVCPU += rec.VCPUPerInstance
+	rm.allocatedMem += rec.MemGBPerInstance
+	rec.ConsumedCount++
+	rm.mu.Unlock()
+
+	rm.updateInstanceSubscriptions()
+	return nil
+}
+
+// ReleaseToReservation returns a launched instance's slot. If the reservation
+// still exists it is the inverse net-zero swap (allocated* -> reservedCR*,
+// ConsumedCount--); if the reservation is gone (cancelled or lost on restart) the
+// allocation is freed to the general pool using the catalog charge so it never
+// overcounts. Best-effort: a missing reservation is not an error.
+func (rm *ResourceManager) ReleaseToReservation(crID string, it *ec2.InstanceTypeInfo) {
+	rm.mu.Lock()
+	if rec, ok := rm.reservations[crID]; ok && rec.ConsumedCount > 0 {
+		rm.allocatedVCPU -= rec.VCPUPerInstance
+		rm.allocatedMem -= rec.MemGBPerInstance
+		rm.reservedCRVCPU += rec.VCPUPerInstance
+		rm.reservedCRMem += rec.MemGBPerInstance
+		rec.ConsumedCount--
+	} else {
+		rm.allocatedVCPU -= int(instanceTypeVCPUs(it))
+		rm.allocatedMem -= float64(rm.instanceMemChargeMiB(it)) / 1024.0
+	}
+	rm.mu.Unlock()
+
+	rm.updateInstanceSubscriptions()
+}
+
+// ReservationAvailable reports how many more instances can launch into crID
+// (Total - Consumed), or 0 when the reservation is unknown, owned by another
+// account, or for a different instance type. The daemon handler's up-front check
+// turns those zero cases into precise errors before the launch loop.
+func (rm *ResourceManager) ReservationAvailable(crID, accountID string, it *ec2.InstanceTypeInfo) int {
+	rm.mu.RLock()
+	defer rm.mu.RUnlock()
+	rec, ok := rm.reservations[crID]
+	if !ok || rec.AccountID != accountID || rec.InstanceType != aws.StringValue(it.InstanceType) {
+		return 0
+	}
+	return rec.TotalInstanceCount - rec.ConsumedCount
+}
+
+// ValidateReservationTarget is the owning daemon's up-front semantic check for a
+// targeted launch, run before the launch loop. It returns
+// InvalidCapacityReservationId.NotFound for an unknown or foreign id and
+// InvalidParameterValue on an instance-type mismatch. The "full" case is left to
+// the count gate in PrepareRunInstances, which caps the launch at the available
+// slots and returns ReservationCapacityExceeded when none remain.
+func (rm *ResourceManager) ValidateReservationTarget(crID, accountID string, it *ec2.InstanceTypeInfo) error {
+	rm.mu.RLock()
+	defer rm.mu.RUnlock()
+	rec, ok := rm.reservations[crID]
+	if !ok || rec.AccountID != accountID {
+		return errors.New(awserrors.ErrorInvalidCapacityReservationIdNotFound)
+	}
+	if rec.InstanceType != aws.StringValue(it.InstanceType) {
+		return errors.New(awserrors.ErrorInvalidParameterValue)
+	}
+	return nil
+}
+
+// ListReservations returns value-copy snapshots of the account's reservations on
+// this node, taken under the read lock. Copies rather than the live pointers
+// because ConsumedCount is mutated under rm.mu by the launch/release paths, so a
+// caller reading the shared record lock-free would race.
+func (rm *ResourceManager) ListReservations(accountID string) []capacityReservation {
 	rm.mu.RLock()
 	defer rm.mu.RUnlock()
 
-	var out []*capacityReservation
+	var out []capacityReservation
 	for _, rec := range rm.reservations {
 		if rec.AccountID == accountID {
-			out = append(out, rec)
+			out = append(out, *rec)
 		}
 	}
 	return out

spinifex/daemon/capacityreservation_test.go

@@ -169,3 +169,187 @@ func TestCapacityReservation_ConcurrentNoOvercommit(t *testing.T) {
 	assert.Equal(t, sumVCPU, rm.reservedCRVCPU)
 	assert.Equal(t, sumMem, rm.reservedCRMem)
 }
+
+// AllocateFromReservation moves a slot reservedCR*->allocated* without touching
+// schedulable capacity, bumps ConsumedCount, and drops AvailableInstanceCount.
+func TestAllocateFromReservation_NetZeroSwap(t *testing.T) {
+	rm := newReservationTestRM()
+	mt := microType()
+	rec := microReservation("cr-1", "acct-a", 2)
+	require.NoError(t, rm.CreateReservation(rec))
+
+	before := rm.canAllocate(mt, 100)
+	require.Equal(t, 2, before, "carve-out of 2 leaves room for 2 general launches")
+
+	require.NoError(t, rm.AllocateFromReservation("cr-1", "acct-a", mt))
+
+	rm.mu.RLock()
+	assert.Equal(t, 2, rm.reservedCRVCPU, "one slot left reservedCR*")
+	assert.Equal(t, 1.0, rm.reservedCRMem)
+	assert.Equal(t, 2, rm.allocatedVCPU, "consumed slot now in allocated*")
+	assert.Equal(t, 1.0, rm.allocatedMem)
+	assert.Equal(t, 1, rec.ConsumedCount)
+	rm.mu.RUnlock()
+
+	assert.Equal(t, before, rm.canAllocate(mt, 100), "schedulable capacity unchanged by the swap")
+	assert.Equal(t, int64(1), aws.Int64Value(rec.toAWSCapacityReservation().AvailableInstanceCount))
+}
+
+// The semantic checks reject unknown/foreign ids, type mismatch, and over-consume
+// with their mapped AWS error codes, leaving accounting untouched.
+func TestAllocateFromReservation_Errors(t *testing.T) {
+	rm := newReservationTestRM()
+	mt := microType()
+	require.NoError(t, rm.CreateReservation(microReservation("cr-1", "acct-a", 1)))
+
+	err := rm.AllocateFromReservation("cr-missing", "acct-a", mt)
+	require.Error(t, err)
+	assert.Equal(t, awserrors.ErrorInvalidCapacityReservationIdNotFound, err.Error())
+
+	err = rm.AllocateFromReservation("cr-1", "acct-b", mt)
+	require.Error(t, err)
+	assert.Equal(t, awserrors.ErrorInvalidCapacityReservationIdNotFound, err.Error(), "foreign account is not found")
+
+	wrongType := &ec2.InstanceTypeInfo{
+		InstanceType: aws.String("t3.large"),
+		VCpuInfo:     &ec2.VCpuInfo{DefaultVCpus: aws.Int64(2)},
+		MemoryInfo:   &ec2.MemoryInfo{SizeInMiB: aws.Int64(8192)},
+	}
+	err = rm.AllocateFromReservation("cr-1", "acct-a", wrongType)
+	require.Error(t, err)
+	assert.Equal(t, awserrors.ErrorInvalidParameterValue, err.Error())
+
+	require.NoError(t, rm.AllocateFromReservation("cr-1", "acct-a", mt), "first slot consumes")
+	err = rm.AllocateFromReservation("cr-1", "acct-a", mt)
+	require.Error(t, err)
+	assert.Equal(t, awserrors.ErrorReservationCapacityExceeded, err.Error(), "second exceeds Total=1")
+}
+
+// ReleaseToReservation is the inverse swap when the reservation still exists:
+// allocated*->reservedCR*, ConsumedCount--, AvailableInstanceCount restored.
+func TestReleaseToReservation_RestoresSlot(t *testing.T) {
+	rm := newReservationTestRM()
+	mt := microType()
+	rec := microReservation("cr-1", "acct-a", 2)
+	require.NoError(t, rm.CreateReservation(rec))
+	require.NoError(t, rm.AllocateFromReservation("cr-1", "acct-a", mt))
+
+	rm.ReleaseToReservation("cr-1", mt)
+
+	rm.mu.RLock()
+	assert.Equal(t, 4, rm.reservedCRVCPU, "slot returned to reservedCR*")
+	assert.Equal(t, 2.0, rm.reservedCRMem)
+	assert.Equal(t, 0, rm.allocatedVCPU)
+	assert.Equal(t, 0.0, rm.allocatedMem)
+	assert.Equal(t, 0, rec.ConsumedCount)
+	rm.mu.RUnlock()
+
+	assert.Equal(t, int64(2), aws.Int64Value(rec.toAWSCapacityReservation().AvailableInstanceCount))
+}
+
+// After the reservation is cancelled, a still-running consumer's release frees to
+// the general pool (catalog charge) instead of a now-gone reservedCR*, never
+// overcounting. Cancel itself frees only the unconsumed remainder.
+func TestReleaseToReservation_AfterCancelFreesToGeneralPool(t *testing.T) {
+	rm := newReservationTestRM()
+	mt := microType()
+	require.NoError(t, rm.CreateReservation(microReservation("cr-1", "acct-a", 2)))
+	require.NoError(t, rm.AllocateFromReservation("cr-1", "acct-a", mt))
+
+	_, ok := rm.CancelReservation("cr-1", "acct-a")
+	require.True(t, ok)
+
+	rm.mu.RLock()
+	assert.Equal(t, 0, rm.reservedCRVCPU, "cancel freed only the unconsumed slot")
+	assert.Equal(t, 0.0, rm.reservedCRMem)
+	assert.Equal(t, 2, rm.allocatedVCPU, "consumed slot keeps running on general capacity")
+	rm.mu.RUnlock()
+
+	rm.ReleaseToReservation("cr-1", mt)
+
+	rm.mu.RLock()
+	assert.Equal(t, 0, rm.allocatedVCPU, "release frees the dangling consumer to general pool")
+	assert.Equal(t, 0.0, rm.allocatedMem)
+	assert.Equal(t, 0, rm.reservedCRVCPU, "no phantom reservedCR* re-added")
+	rm.mu.RUnlock()
+
+	assert.Equal(t, 4, rm.canAllocate(mt, 100), "general capacity fully restored")
+}
+
+// ReservationAvailable reports Total-Consumed, and 0 for unknown/foreign-account
+// /type-mismatch so the count site degrades to the daemon's up-front error.
+func TestReservationAvailable(t *testing.T) {
+	rm := newReservationTestRM()
+	mt := microType()
+	require.NoError(t, rm.CreateReservation(microReservation("cr-1", "acct-a", 2)))
+
+	assert.Equal(t, 2, rm.ReservationAvailable("cr-1", "acct-a", mt))
+	assert.Equal(t, 0, rm.ReservationAvailable("cr-missing", "acct-a", mt))
+	assert.Equal(t, 0, rm.ReservationAvailable("cr-1", "acct-b", mt), "foreign account sees nothing")
+
+	wrongType := &ec2.InstanceTypeInfo{InstanceType: aws.String("t3.large")}
+	assert.Equal(t, 0, rm.ReservationAvailable("cr-1", "acct-a", wrongType), "type mismatch sees nothing")
+
+	require.NoError(t, rm.AllocateFromReservation("cr-1", "acct-a", mt))
+	assert.Equal(t, 1, rm.ReservationAvailable("cr-1", "acct-a", mt), "consume drops availability")
+}
+
+// ValidateReservationTarget is the up-front semantic check: unknown/foreign ids
+// map to NotFound, a type mismatch to InvalidParameterValue, and a matching
+// reservation passes (the full case is left to the launch-side count gate).
+func TestValidateReservationTarget(t *testing.T) {
+	rm := newReservationTestRM()
+	mt := microType()
+	require.NoError(t, rm.CreateReservation(microReservation("cr-1", "acct-a", 1)))
+
+	require.NoError(t, rm.ValidateReservationTarget("cr-1", "acct-a", mt))
+
+	err := rm.ValidateReservationTarget("cr-missing", "acct-a", mt)
+	require.Error(t, err)
+	assert.Equal(t, awserrors.ErrorInvalidCapacityReservationIdNotFound, err.Error())
+
+	err = rm.ValidateReservationTarget("cr-1", "acct-b", mt)
+	require.Error(t, err)
+	assert.Equal(t, awserrors.ErrorInvalidCapacityReservationIdNotFound, err.Error(), "foreign account is not found")
+
+	wrongType := &ec2.InstanceTypeInfo{InstanceType: aws.String("t3.large")}
+	err = rm.ValidateReservationTarget("cr-1", "acct-a", wrongType)
+	require.Error(t, err)
+	assert.Equal(t, awserrors.ErrorInvalidParameterValue, err.Error())
+
+	// A full reservation still validates — the count gate handles capacity.
+	require.NoError(t, rm.AllocateFromReservation("cr-1", "acct-a", mt))
+	require.NoError(t, rm.ValidateReservationTarget("cr-1", "acct-a", mt), "full reservation passes the semantic check")
+}
+
+// Concurrent general allocate and AllocateFromReservation share rm.mu; the swap is
+// net-zero on reservedCR*+allocated*, so committed compute never exceeds the host
+// and ConsumedCount never passes Total.
+func TestAllocateFromReservation_ConcurrentNoOvercommit(t *testing.T) {
+	rm := newReservationTestRM()
+	mt := microType()
+	rec := microReservation("cr-1", "acct-a", 2)
+	require.NoError(t, rm.CreateReservation(rec))
+
+	const workers = 32
+	var wg sync.WaitGroup
+	wg.Add(workers)
+	for i := range workers {
+		go func(i int) {
+			defer wg.Done()
+			if i%2 == 0 {
+				_ = rm.AllocateFromReservation("cr-1", "acct-a", mt)
+			} else {
+				_ = rm.allocate(mt)
+			}
+		}(i)
+	}
+	wg.Wait()
+
+	rm.mu.RLock()
+	defer rm.mu.RUnlock()
+	assert.LessOrEqual(t, rm.reservedCRVCPU+rm.allocatedVCPU, rm.hostVCPU, "vCPU never overcommitted")
+	assert.LessOrEqual(t, rm.reservedCRMem+rm.allocatedMem, rm.hostMemGB, "memory never overcommitted")
+	assert.LessOrEqual(t, rec.ConsumedCount, rec.TotalInstanceCount, "ConsumedCount never passes Total")
+	assert.GreaterOrEqual(t, rec.ConsumedCount, 0)
+}

spinifex/daemon/daemon.go

@@ -2141,6 +2141,8 @@ func (rm *ResourceManager) deallocate(instanceType *ec2.InstanceTypeInfo) {
 	rm.updateInstanceSubscriptions()
 }
 
+var _ handlers_ec2_instance.InstanceTypeAllocator = (*ResourceManager)(nil)
+
 // Allocate, Deallocate, CanAllocate satisfy handlers_ec2_instance.InstanceTypeAllocator.
 func (rm *ResourceManager) Allocate(it *ec2.InstanceTypeInfo) error { return rm.allocate(it) }
 func (rm *ResourceManager) Deallocate(it *ec2.InstanceTypeInfo)     { rm.deallocate(it) }

spinifex/daemon/daemon_handlers_capacityreservation.go

@@ -50,7 +50,7 @@ func (d *Daemon) handleEC2CreateCapacityReservation(msg *nats.Msg) {
 		AvailabilityZone:      aws.StringValue(input.AvailabilityZone),
 		TotalInstanceCount:    int(aws.Int64Value(input.InstanceCount)),
 		VCPUPerInstance:       int(instanceTypeVCPUs(it)),
-		MemGBPerInstance:      float64(instanceTypeMemoryMiB(it)) / 1024.0,
+		MemGBPerInstance:      float64(d.resourceMgr.instanceMemChargeMiB(it)) / 1024.0,
 		InstanceMatchCriteria: matchCriteria,
 		Tenancy:               tenancy,
 		InstancePlatform:      aws.StringValue(input.InstancePlatform),
@@ -62,6 +62,19 @@ func (d *Daemon) handleEC2CreateCapacityReservation(msg *nats.Msg) {
 		return
 	}
 
+	// Subscribe the per-reservation launch subject before responding: the SUB and
+	// this reply share d.natsConn, so the server registers the responder before
+	// the gateway sees the new id and can route a targeted launch to it. A failed
+	// subscribe would leave the reservation holding capacity no targeted launch
+	// could ever reach, so roll it back rather than return a dead id.
+	if err := d.subscribeReservationLaunch(rec.ID); err != nil {
+		slog.Error("Failed to subscribe reservation launch subject, rolling back reservation",
+			"crId", rec.ID, "err", err)
+		d.resourceMgr.CancelReservation(rec.ID, accountID)
+		respondWithError(msg, awserrors.ErrorServerInternal)
+		return
+	}
+
 	respondWithJSON(msg, rec.toAWSCapacityReservation())
 }
 
@@ -92,22 +105,71 @@ func (d *Daemon) handleEC2CancelCapacityReservation(msg *nats.Msg) {
 	}
 
 	_, found := d.resourceMgr.CancelReservation(aws.StringValue(input.CapacityReservationId), accountID)
+	if found {
+		// Drop the launch subject so a targeted launch against the cancelled id
+		// hits no responder → InvalidCapacityReservationId.NotFound at the gateway.
+		d.unsubscribeReservationLaunch(aws.StringValue(input.CapacityReservationId))
+	}
 	respondWithJSON(msg, &ec2.CancelCapacityReservationOutput{Return: aws.Bool(found)})
 }
 
+// reservationLaunchSubject is the per-reservation subject the owning daemon
+// subscribes so the gateway routes a targeted launch straight to it, with no
+// cr→node lookup. ErrNoResponders on this subject therefore means the
+// reservation is gone (cancelled or lost to a restart).
+func reservationLaunchSubject(crID string) string {
+	return "ec2.RunInstances.cr." + crID
+}
+
+// subscribeReservationLaunch wires the cr launch subject to the standard
+// RunInstances handler — the target id rides in the input, so handleEC2RunInstances
+// serves it unchanged. Tracked in natsSubscriptions (keyed by cr id) so cancel
+// and shutdown can drop it. Idempotent; returns the subscribe error so the
+// caller can roll the reservation back rather than leave it unlaunchable.
+func (d *Daemon) subscribeReservationLaunch(crID string) error {
+	if d.natsConn == nil {
+		return nil
+	}
+	d.mu.Lock()
+	defer d.mu.Unlock()
+	if _, exists := d.natsSubscriptions[crID]; exists {
+		return nil
+	}
+	sub, err := d.natsConn.Subscribe(reservationLaunchSubject(crID), d.handleEC2RunInstances)
+	if err != nil {
+		return err
+	}
+	d.natsSubscriptions[crID] = sub
+	return nil
+}
+
+// unsubscribeReservationLaunch drops the cr launch subscription on cancel.
+func (d *Daemon) unsubscribeReservationLaunch(crID string) {
+	d.mu.Lock()
+	sub, ok := d.natsSubscriptions[crID]
+	if ok {
+		delete(d.natsSubscriptions, crID)
+	}
+	d.mu.Unlock()
+	if ok {
+		if err := sub.Unsubscribe(); err != nil {
+			slog.Warn("Failed to unsubscribe reservation launch subject", "crId", crID, "err", err)
+		}
+	}
+}
+
 // toAWSCapacityReservation renders the in-memory record as the AWS API type.
-// Reservations are always active with AvailableInstanceCount equal to the total,
+// Reservations are always active with AvailableInstanceCount = Total − Consumed,
 // and never expire (EndDateType=unlimited).
 func (r *capacityReservation) toAWSCapacityReservation() *ec2.CapacityReservation {
-	count := int64(r.TotalInstanceCount)
 	return &ec2.CapacityReservation{
 		CapacityReservationId:  aws.String(r.ID),
 		OwnerId:                aws.String(r.AccountID),
 		InstanceType:           aws.String(r.InstanceType),
 		InstancePlatform:       aws.String(r.InstancePlatform),
 		AvailabilityZone:       aws.String(r.AvailabilityZone),
-		TotalInstanceCount:     aws.Int64(count),
-		AvailableInstanceCount: aws.Int64(count),
+		TotalInstanceCount:     aws.Int64(int64(r.TotalInstanceCount)),
+		AvailableInstanceCount: aws.Int64(int64(r.TotalInstanceCount - r.ConsumedCount)),
 		State:                  aws.String(ec2.CapacityReservationStateActive),
 		StartDate:              aws.Time(r.CreateDate),
 		CreateDate:             aws.Time(r.CreateDate),

spinifex/daemon/daemon_handlers_capacityreservation_test.go

@@ -249,3 +249,61 @@ func TestHandleEC2CancelCapacityReservation_ScopingAndAck(t *testing.T) {
 
 	assert.False(t, cancel(testAccountID), "a second cancel reports not-found")
 }
+
+// Create subscribes the per-reservation launch subject (so a targeted launch has
+// a responder) and Cancel drops it (so a launch against the cancelled id returns
+// ErrNoResponders → InvalidCapacityReservationId.NotFound at the gateway).
+func TestHandleEC2CapacityReservation_LaunchSubjectLifecycle(t *testing.T) {
+	daemon := createTestDaemon(t, sharedNATSURL)
+
+	instType := getTestInstanceType(t)
+	it := daemon.resourceMgr.instanceTypes[instType]
+	require.NotNil(t, it)
+	require.GreaterOrEqual(t, daemon.resourceMgr.canAllocate(it, 1000), 1)
+
+	createSubject := "ec2.CreateCapacityReservation." + daemon.node
+	createSub, err := daemon.natsConn.Subscribe(createSubject, daemon.handleEC2CreateCapacityReservation)
+	require.NoError(t, err)
+	defer func() { _ = createSub.Unsubscribe() }()
+	cancelSub, err := daemon.natsConn.Subscribe("ec2.CancelCapacityReservation", daemon.handleEC2CancelCapacityReservation)
+	require.NoError(t, err)
+	defer func() { _ = cancelSub.Unsubscribe() }()
+
+	reqData, _ := json.Marshal(&ec2.CreateCapacityReservationInput{
+		InstanceType: aws.String(instType), InstanceCount: aws.Int64(1),
+		AvailabilityZone: aws.String("ap-southeast-2a"), InstancePlatform: aws.String("Linux/UNIX"),
+	})
+	reply, err := natsRequest(daemon.natsConn, createSubject, reqData, 5*time.Second)
+	require.NoError(t, err)
+	require.Empty(t, replyErrCode(t, reply.Data), "create should not error: %s", reply.Data)
+	var cr ec2.CapacityReservation
+	require.NoError(t, json.Unmarshal(reply.Data, &cr))
+	crID := aws.StringValue(cr.CapacityReservationId)
+
+	daemon.mu.Lock()
+	_, tracked := daemon.natsSubscriptions[crID]
+	daemon.mu.Unlock()
+	assert.True(t, tracked, "create tracks the cr launch subscription")
+
+	// An empty payload makes handleEC2RunInstances reject on MissingParameter
+	// before any allocation; the point is only that a responder exists (no
+	// ErrNoResponders) — the SUB raced ahead of the create reply on the same conn.
+	launchSubject := reservationLaunchSubject(crID)
+	_, err = natsRequest(daemon.natsConn, launchSubject, []byte("{}"), 2*time.Second)
+	require.NotErrorIs(t, err, nats.ErrNoResponders, "cr launch subject has a responder after create")
+
+	cancelData, _ := json.Marshal(&ec2.CancelCapacityReservationInput{CapacityReservationId: aws.String(crID)})
+	creply, err := natsRequest(daemon.natsConn, "ec2.CancelCapacityReservation", cancelData, 5*time.Second)
+	require.NoError(t, err)
+	var cout ec2.CancelCapacityReservationOutput
+	require.NoError(t, json.Unmarshal(creply.Data, &cout))
+	require.True(t, aws.BoolValue(cout.Return), "owner cancel succeeds")
+
+	daemon.mu.Lock()
+	_, tracked = daemon.natsSubscriptions[crID]
+	daemon.mu.Unlock()
+	assert.False(t, tracked, "cancel drops the cr launch subscription")
+
+	_, err = natsRequest(daemon.natsConn, launchSubject, []byte("{}"), 1*time.Second)
+	assert.ErrorIs(t, err, nats.ErrNoResponders, "cr launch subject has no responder after cancel")
+}

spinifex/daemon/daemon_handlers_instance.go

@@ -50,7 +50,21 @@ func (d *Daemon) handleEC2RunInstances(msg *nats.Msg) {
 		return
 	}
 
-	reservation, instances, instanceType, err := d.instanceService.PrepareRunInstances(input, accountID)
+	// Targeted launch: the gateway routes only when an explicit reservation id is
+	// present, but the id rides in the input either way. Validate semantics
+	// up-front (the per-instance atomic re-check under rm.mu covers the race);
+	// the count gate in PrepareRunInstances handles a full reservation.
+	reservationID := capacityReservationTargetID(input)
+	if reservationID != "" {
+		if it, ok := d.resourceMgr.instanceTypes[aws.StringValue(input.InstanceType)]; ok {
+			if vErr := d.resourceMgr.ValidateReservationTarget(reservationID, accountID, it); vErr != nil {
+				respondWithError(msg, awserrors.ValidErrorCode(vErr.Error()))
+				return
+			}
+		}
+	}
+
+	reservation, instances, instanceType, err := d.instanceService.PrepareRunInstances(input, accountID, reservationID)
 	if err != nil {
 		respondWithError(msg, awserrors.ValidErrorCode(err.Error()))
 		return
@@ -68,7 +82,11 @@ func (d *Daemon) handleEC2RunInstances(msg *nats.Msg) {
 		slog.Error("handleEC2RunInstances failed to marshal reservation", "err", err)
 		respondWithError(msg, awserrors.ErrorServerInternal)
 		for range instances {
-			d.resourceMgr.deallocate(instanceType)
+			if reservationID == "" {
+				d.resourceMgr.deallocate(instanceType)
+			} else {
+				d.resourceMgr.ReleaseToReservation(reservationID, instanceType)
+			}
 		}
 		return
 	}
@@ -100,6 +118,16 @@ func (d *Daemon) handleEC2RunInstances(msg *nats.Msg) {
 	d.instanceService.LaunchRunInstances(instances, input, instanceType)
 }
 
+// capacityReservationTargetID returns the explicit targeted-launch reservation id
+// from the input, or "" when the launch is untargeted (general path).
+func capacityReservationTargetID(input *ec2.RunInstancesInput) string {
+	spec := input.CapacityReservationSpecification
+	if spec == nil || spec.CapacityReservationTarget == nil {
+		return ""
+	}
+	return aws.StringValue(spec.CapacityReservationTarget.CapacityReservationId)
+}
+
 // describeInstancesValidFilters defines the set of filter names accepted by DescribeInstances.
 var describeInstancesValidFilters = map[string]bool{
 	"instance-state-name": true,
@@ -307,6 +335,19 @@ func (d *Daemon) handleEC2DescribeInstances(msg *nats.Msg) {
 					}
 				}
 
+				// Echo the consumed capacity reservation so targeted-launch
+				// Terraform converges — without it the instance reports no
+				// reservation and the plan never settles.
+				if instance.CapacityReservationId != "" {
+					instanceCopy.CapacityReservationId = aws.String(instance.CapacityReservationId)
+					instanceCopy.CapacityReservationSpecification = &ec2.CapacityReservationSpecificationResponse{
+						CapacityReservationPreference: aws.String(ec2.CapacityReservationPreferenceOpen),
+						CapacityReservationTarget: &ec2.CapacityReservationTargetResponse{
+							CapacityReservationId: aws.String(instance.CapacityReservationId),
+						},
+					}
+				}
+
 				// Apply filters against the fully-built instance copy
 				if len(parsedFilters) > 0 && !instanceMatchesFilters(instance, &instanceCopy, parsedFilters) {
 					continue

spinifex/daemon/daemon_handlers_instance_test.go

@@ -595,3 +595,82 @@ func TestHandleEC2DescribeStoppedInstances_ReservationGrouping(t *testing.T) {
 	assert.Equal(t, "r-shared", *output.Reservations[0].ReservationId)
 	assert.Len(t, output.Reservations[0].Instances, 2)
 }
+
+// DescribeInstances echoes the consumed capacity reservation so a targeted
+// launch reports a CapacityReservationId (without it, Terraform never converges).
+func TestHandleEC2DescribeInstances_CapacityReservationEcho(t *testing.T) {
+	daemon := createTestDaemon(t, sharedNATSURL)
+
+	reservation := &ec2.Reservation{}
+	reservation.SetReservationId("r-cr-echo")
+	reservation.SetOwnerId(testAccountID)
+	instance := &ec2.Instance{}
+	instance.SetInstanceId("i-cr-echo")
+	instance.SetInstanceType("t3.micro")
+
+	daemon.vmMgr.Insert(&vm.VM{
+		ID:                    "i-cr-echo",
+		Status:                vm.StateRunning,
+		AccountID:             testAccountID,
+		CapacityReservationId: "cr-0123456789abcdef0",
+		Reservation:           reservation,
+		Instance:              instance,
+	})
+
+	sub, err := daemon.natsConn.Subscribe("ec2.DescribeInstances", daemon.handleEC2DescribeInstances)
+	require.NoError(t, err)
+	defer func() { _ = sub.Unsubscribe() }()
+
+	reqData, _ := json.Marshal(&ec2.DescribeInstancesInput{})
+	reply, err := natsRequest(daemon.natsConn, "ec2.DescribeInstances", reqData, 5*time.Second)
+	require.NoError(t, err)
+	var out ec2.DescribeInstancesOutput
+	require.NoError(t, json.Unmarshal(reply.Data, &out))
+	require.Len(t, out.Reservations, 1)
+	require.Len(t, out.Reservations[0].Instances, 1)
+
+	got := out.Reservations[0].Instances[0]
+	assert.Equal(t, "cr-0123456789abcdef0", aws.StringValue(got.CapacityReservationId))
+	require.NotNil(t, got.CapacityReservationSpecification)
+	assert.Equal(t, ec2.CapacityReservationPreferenceOpen,
+		aws.StringValue(got.CapacityReservationSpecification.CapacityReservationPreference))
+	require.NotNil(t, got.CapacityReservationSpecification.CapacityReservationTarget)
+	assert.Equal(t, "cr-0123456789abcdef0",
+		aws.StringValue(got.CapacityReservationSpecification.CapacityReservationTarget.CapacityReservationId))
+}
+
+// An instance with no reservation reports no capacity-reservation fields.
+func TestHandleEC2DescribeInstances_NoCapacityReservationEcho(t *testing.T) {
+	daemon := createTestDaemon(t, sharedNATSURL)
+
+	reservation := &ec2.Reservation{}
+	reservation.SetReservationId("r-plain")
+	reservation.SetOwnerId(testAccountID)
+	instance := &ec2.Instance{}
+	instance.SetInstanceId("i-plain")
+	instance.SetInstanceType("t3.micro")
+
+	daemon.vmMgr.Insert(&vm.VM{
+		ID:          "i-plain",
+		Status:      vm.StateRunning,
+		AccountID:   testAccountID,
+		Reservation: reservation,
+		Instance:    instance,
+	})
+
+	sub, err := daemon.natsConn.Subscribe("ec2.DescribeInstances", daemon.handleEC2DescribeInstances)
+	require.NoError(t, err)
+	defer func() { _ = sub.Unsubscribe() }()
+
+	reqData, _ := json.Marshal(&ec2.DescribeInstancesInput{})
+	reply, err := natsRequest(daemon.natsConn, "ec2.DescribeInstances", reqData, 5*time.Second)
+	require.NoError(t, err)
+	var out ec2.DescribeInstancesOutput
+	require.NoError(t, json.Unmarshal(reply.Data, &out))
+	require.Len(t, out.Reservations, 1)
+	require.Len(t, out.Reservations[0].Instances, 1)
+
+	got := out.Reservations[0].Instances[0]
+	assert.Empty(t, aws.StringValue(got.CapacityReservationId))
+	assert.Nil(t, got.CapacityReservationSpecification)
+}

spinifex/daemon/daemon_system_instance.go

@@ -427,7 +427,7 @@ func (d *Daemon) launchAMISystemInstance(input *sysinstance.SystemInstanceInput)
 		}}
 	}
 
-	_, instances, instanceType, err := d.instanceService.PrepareRunInstances(runInput, input.AccountID)
+	_, instances, instanceType, err := d.instanceService.PrepareRunInstances(runInput, input.AccountID, "")
 	if err != nil {
 		return nil, err
 	}

spinifex/daemon/vm_adapters.go

@@ -313,6 +313,14 @@ func (a *resourceControllerAdapter) Deallocate(instanceType string) {
 	a.rm.deallocate(it)
 }
 
+func (a *resourceControllerAdapter) ReleaseToReservation(reservationID, instanceType string) {
+	it := a.rm.instanceTypes[instanceType]
+	if it == nil {
+		return
+	}
+	a.rm.ReleaseToReservation(reservationID, it)
+}
+
 func (a *resourceControllerAdapter) CanAllocate(instanceType string, count int) int {
 	it := a.rm.instanceTypes[instanceType]
 	if it == nil {

spinifex/gateway/ec2/instance/RunInstances.go

@@ -104,6 +104,25 @@ func runInstancesInner(input *ec2.RunInstancesInput, natsConn *nats.Conn, iamSvc
 		return reservation, err
 	}
 
+	// Targeted capacity-reservation launch routes straight to the owning node's
+	// cr-subject. The gateway does input-only checks (malformed id, the PG+CR
+	// combination); the owning daemon does the semantic checks (account, type,
+	// full) only it can see, and ErrNoResponders surfaces a gone id as NotFound.
+	if crID := capacityReservationTargetID(input); crID != "" {
+		if !strings.HasPrefix(crID, "cr-") {
+			return reservation, errors.New(awserrors.ErrorInvalidCapacityReservationIdMalformed)
+		}
+		if placementGroupName(input) != "" {
+			return reservation, errors.New(awserrors.ErrorInvalidParameterValue)
+		}
+		reservationPtr, err := runIntoReservation(input, natsConn, accountID, crID)
+		if err != nil {
+			return reservation, err
+		}
+		enrichReservationWithProfileID(reservationPtr, resolvedProfile)
+		return *reservationPtr, nil
+	}
+
 	groupName := placementGroupName(input)
 	if groupName != "" {
 		strategy, err := lookupPlacementGroupStrategy(natsConn, accountID, groupName)
@@ -198,6 +217,17 @@ func placementGroupName(input *ec2.RunInstancesInput) string {
 	return ""
 }
 
+// capacityReservationTargetID returns the explicit targeted-launch reservation id
+// from the input, or "" when the launch is untargeted (general path). Preference
+// is ignored — only a present target id routes to the cr-subject.
+func capacityReservationTargetID(input *ec2.RunInstancesInput) string {
+	spec := input.CapacityReservationSpecification
+	if spec == nil || spec.CapacityReservationTarget == nil {
+		return ""
+	}
+	return aws.StringValue(spec.CapacityReservationTarget.CapacityReservationId)
+}
+
 // lookupPlacementGroupStrategy returns the strategy of a placement group, or an error if absent/unavailable.
 func lookupPlacementGroupStrategy(natsConn *nats.Conn, accountID, groupName string) (string, error) {
 	pgSvc := handlers_ec2_placementgroup.NewNATSPlacementGroupService(natsConn)

spinifex/gateway/ec2/instance/RunInstancesIntoReservation.go

@@ -0,0 +1,28 @@
+package gateway_ec2_instance
+
+import (
+	"errors"
+	"time"
+
+	"github.com/aws/aws-sdk-go/service/ec2"
+	"github.com/mulgadc/spinifex/spinifex/awserrors"
+	"github.com/mulgadc/spinifex/spinifex/utils"
+	"github.com/nats-io/nats.go"
+)
+
+// runIntoReservation routes a targeted launch straight to the owning daemon's
+// per-reservation subject. Only the owner subscribes ec2.RunInstances.cr.<crID>,
+// so NATS addresses it without any cr→node lookup. ErrNoResponders means the
+// reservation is gone (cancelled or lost to a restart) → NotFound; any semantic
+// error (type mismatch, full) rides back as the daemon's awserror code.
+func runIntoReservation(input *ec2.RunInstancesInput, natsConn *nats.Conn, accountID, crID string) (*ec2.Reservation, error) {
+	subject := "ec2.RunInstances.cr." + crID
+	reservation, err := utils.NATSRequest[ec2.Reservation](natsConn, subject, input, 5*time.Minute, accountID)
+	if err != nil {
+		if errors.Is(err, nats.ErrNoResponders) {
+			return nil, errors.New(awserrors.ErrorInvalidCapacityReservationIdNotFound)
+		}
+		return nil, err
+	}
+	return reservation, nil
+}

spinifex/gateway/ec2/instance/RunInstancesIntoReservation_test.go

@@ -0,0 +1,124 @@
+package gateway_ec2_instance
+
+import (
+	"encoding/json"
+	"testing"
+	"time"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/service/ec2"
+	"github.com/mulgadc/spinifex/spinifex/awserrors"
+	"github.com/mulgadc/spinifex/spinifex/utils"
+	"github.com/nats-io/nats.go"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+const testCRID = "cr-0123456789abcdef0"
+
+// crTargetedInput clones the valid defaults and targets a capacity reservation.
+func crTargetedInput(crID string) *ec2.RunInstancesInput {
+	in := defaults
+	in.CapacityReservationSpecification = &ec2.CapacityReservationSpecification{
+		CapacityReservationTarget: &ec2.CapacityReservationTarget{
+			CapacityReservationId: aws.String(crID),
+		},
+	}
+	return &in
+}
+
+func TestCapacityReservationTargetID(t *testing.T) {
+	assert.Empty(t, capacityReservationTargetID(&ec2.RunInstancesInput{}), "no spec → general path")
+	assert.Empty(t, capacityReservationTargetID(&ec2.RunInstancesInput{
+		CapacityReservationSpecification: &ec2.CapacityReservationSpecification{
+			CapacityReservationPreference: aws.String(ec2.CapacityReservationPreferenceOpen),
+		},
+	}), "open without target → general path")
+	assert.Equal(t, testCRID, capacityReservationTargetID(crTargetedInput(testCRID)))
+}
+
+// A malformed reservation id is rejected at the gateway before any NATS call.
+func TestRunInstancesInner_TargetedMalformedID(t *testing.T) {
+	_, err := runInstancesInner(crTargetedInput("bogus-id"), nil, nil, "123456789012", nil, 1)
+	require.Error(t, err)
+	assert.Equal(t, awserrors.ErrorInvalidCapacityReservationIdMalformed, err.Error())
+}
+
+// The placement-group + capacity-reservation combination is rejected (Phase 2).
+func TestRunInstancesInner_TargetedWithPlacementGroup(t *testing.T) {
+	input := crTargetedInput(testCRID)
+	input.Placement = &ec2.Placement{GroupName: aws.String("pg-cluster")}
+	_, err := runInstancesInner(input, nil, nil, "123456789012", nil, 1)
+	require.Error(t, err)
+	assert.Equal(t, awserrors.ErrorInvalidParameterValue, err.Error())
+}
+
+// No responder on the cr-subject (cancelled / restarted reservation) maps the
+// transport-level ErrNoResponders to InvalidCapacityReservationId.NotFound.
+func TestRunIntoReservation_NoResponder(t *testing.T) {
+	_, nc := startTestNATSServer(t)
+	_, err := runIntoReservation(crTargetedInput(testCRID), nc, "123456789012", testCRID)
+	require.Error(t, err)
+	assert.Equal(t, awserrors.ErrorInvalidCapacityReservationIdNotFound, err.Error())
+}
+
+// A targeted launch routes to ec2.RunInstances.cr.<crID> and returns the owning
+// daemon's reservation verbatim.
+func TestRunIntoReservation_Success(t *testing.T) {
+	_, nc := startTestNATSServer(t)
+
+	sub, err := nc.Subscribe("ec2.RunInstances.cr."+testCRID, func(msg *nats.Msg) {
+		data, _ := json.Marshal(ec2.Reservation{
+			ReservationId: aws.String("r-cr"),
+			Instances:     []*ec2.Instance{{InstanceId: aws.String("i-cr1")}},
+		})
+		_ = msg.Respond(data)
+	})
+	require.NoError(t, err)
+	defer func() { _ = sub.Unsubscribe() }()
+	time.Sleep(50 * time.Millisecond)
+
+	reservation, err := runIntoReservation(crTargetedInput(testCRID), nc, "123456789012", testCRID)
+	require.NoError(t, err)
+	require.Len(t, reservation.Instances, 1)
+	assert.Equal(t, "i-cr1", aws.StringValue(reservation.Instances[0].InstanceId))
+}
+
+// A semantic error from the owning daemon (e.g. a full reservation) rides back
+// as its awserror code, not the generic NotFound.
+func TestRunIntoReservation_DaemonErrorPropagates(t *testing.T) {
+	_, nc := startTestNATSServer(t)
+
+	sub, err := nc.Subscribe("ec2.RunInstances.cr."+testCRID, func(msg *nats.Msg) {
+		_ = msg.Respond(utils.GenerateErrorPayload(awserrors.ErrorReservationCapacityExceeded))
+	})
+	require.NoError(t, err)
+	defer func() { _ = sub.Unsubscribe() }()
+	time.Sleep(50 * time.Millisecond)
+
+	_, err = runIntoReservation(crTargetedInput(testCRID), nc, "123456789012", testCRID)
+	require.Error(t, err)
+	assert.Equal(t, awserrors.ErrorReservationCapacityExceeded, err.Error())
+}
+
+// The full runInstancesInner branch routes a targeted launch to the cr-subject
+// (not the general distribute path) and enriches the result.
+func TestRunInstancesInner_TargetedRoutesToCRSubject(t *testing.T) {
+	_, nc := startTestNATSServer(t)
+
+	sub, err := nc.Subscribe("ec2.RunInstances.cr."+testCRID, func(msg *nats.Msg) {
+		data, _ := json.Marshal(ec2.Reservation{
+			ReservationId: aws.String("r-cr"),
+			Instances:     []*ec2.Instance{{InstanceId: aws.String("i-cr1")}},
+		})
+		_ = msg.Respond(data)
+	})
+	require.NoError(t, err)
+	defer func() { _ = sub.Unsubscribe() }()
+	time.Sleep(50 * time.Millisecond)
+
+	reservation, err := runInstancesInner(crTargetedInput(testCRID), nc, nil, "123456789012", nil, 1)
+	require.NoError(t, err)
+	require.Len(t, reservation.Instances, 1)
+	assert.Equal(t, "i-cr1", aws.StringValue(reservation.Instances[0].InstanceId))
+}

spinifex/gateway/ecr/registry.go

@@ -516,7 +516,7 @@ func (reg *Registry) getManifest(w http.ResponseWriter, r *http.Request, name, r
 		WriteError(w, http.StatusNotFound, "MANIFEST_UNKNOWN", "manifest unknown")
 		return
 	}
-	if !acceptsType(r.Header.Get("Accept"), meta.MediaType) {
+	if !acceptsType(strings.Join(r.Header.Values("Accept"), ","), meta.MediaType) {
 		WriteError(w, http.StatusNotAcceptable, "MANIFEST_INVALID", "no acceptable manifest media type")
 		return
 	}

spinifex/gateway/ecr/registry_test.go

@@ -208,6 +208,29 @@ func TestManifest_Get_AcceptMismatch(t *testing.T) {
 	assert.Equal(t, http.StatusNotAcceptable, w.Code)
 }
 
+// TestManifest_Get_MultipleAcceptHeaders reproduces skopeo, which sends each
+// acceptable manifest media type as its own Accept header line rather than one
+// comma-joined value. Negotiation must consider every line, not just the first.
+func TestManifest_Get_MultipleAcceptHeaders(t *testing.T) {
+	reg := newTestRegistry()
+	repo := "team/app"
+	cfg := pushBlob(t, reg, repo, []byte("c"))
+	manifest := fmt.Appendf(nil,
+		`{"mediaType":"%s","config":{"digest":"%s"},"layers":[]}`, mediaTypeDockerManifest, cfg)
+	do(reg, http.MethodPut, "/v2/"+repo+"/manifests/v1", manifest,
+		map[string]string{"Content-Type": mediaTypeDockerManifest})
+
+	r := httptest.NewRequest(http.MethodGet, "/v2/"+repo+"/manifests/v1", nil)
+	r.Header.Add("Accept", mediaTypeOCIIndex)
+	r.Header.Add("Accept", mediaTypeOCIManifest)
+	r.Header.Add("Accept", mediaTypeDockerManifest)
+	w := httptest.NewRecorder()
+	reg.ServeHTTP(w, r)
+
+	assert.Equal(t, http.StatusOK, w.Code, "body: %s", w.Body.String())
+	assert.Equal(t, manifest, w.Body.Bytes())
+}
+
 func TestManifest_Get_Unknown(t *testing.T) {
 	reg := newTestRegistry()
 	w := do(reg, http.MethodGet, "/v2/team/app/manifests/missing", nil, nil)

spinifex/handlers/ec2/instance/service.go

@@ -59,6 +59,9 @@ type InstanceTypeAllocator interface {
 	Allocate(instanceType *ec2.InstanceTypeInfo) error
 	Deallocate(instanceType *ec2.InstanceTypeInfo)
 	CanAllocate(instanceType *ec2.InstanceTypeInfo, count int) int
+	AllocateFromReservation(reservationID, accountID string, instanceType *ec2.InstanceTypeInfo) error
+	ReleaseToReservation(reservationID string, instanceType *ec2.InstanceTypeInfo)
+	ReservationAvailable(reservationID, accountID string, instanceType *ec2.InstanceTypeInfo) int
 	InstanceTypes() map[string]*ec2.InstanceTypeInfo
 }
 

spinifex/handlers/ec2/instance/service_impl.go

@@ -572,7 +572,13 @@ func instanceTagsFromSpec(specs []*ec2.TagSpecification) []*ec2.Tag {
 // PrepareRunInstances validates input, allocates capacity, creates VM metadata,
 // auto-creates the primary ENI, and auto-assigns a public IP when needed.
 // Does NOT touch vmMgr or NATS — callers insert VMs then call LaunchRunInstances.
-func (s *InstanceServiceImpl) PrepareRunInstances(input *ec2.RunInstancesInput, accountID string) (*ec2.Reservation, []*vm.VM, *ec2.InstanceTypeInfo, error) {
+//
+// A non-empty reservationID makes this a targeted launch into an On-Demand
+// Capacity Reservation: capacity is consumed from that reservation (net-zero
+// swap), the count is capped at the reservation's free slots (never spilling
+// onto general capacity), and every rollback site returns slots to the
+// reservation rather than the general pool.
+func (s *InstanceServiceImpl) PrepareRunInstances(input *ec2.RunInstancesInput, accountID, reservationID string) (*ec2.Reservation, []*vm.VM, *ec2.InstanceTypeInfo, error) {
 	if accountID == "" {
 		return nil, nil, nil, errors.New(awserrors.ErrorServerInternal)
 	}
@@ -620,10 +626,21 @@ func (s *InstanceServiceImpl) PrepareRunInstances(input *ec2.RunInstancesInput,
 	minCount := int(*input.MinCount)
 	maxCount := int(*input.MaxCount)
 
-	allocatableCount := s.resourceMgr.CanAllocate(instanceType, maxCount)
+	var allocatableCount int
+	if reservationID == "" {
+		allocatableCount = s.resourceMgr.CanAllocate(instanceType, maxCount)
+	} else {
+		// Targeted launch: confined to the reservation. Cap at its free slots so
+		// the overflow never spills onto the node's general capacity.
+		allocatableCount = min(s.resourceMgr.ReservationAvailable(reservationID, accountID, instanceType), maxCount)
+	}
 	if allocatableCount < minCount {
-		slog.Error("PrepareRunInstances: insufficient capacity", "requested", minCount, "available", allocatableCount, "InstanceType", *input.InstanceType)
-		return nil, nil, nil, errors.New(awserrors.ErrorInsufficientInstanceCapacity)
+		errCode := awserrors.ErrorInsufficientInstanceCapacity
+		if reservationID != "" {
+			errCode = awserrors.ErrorReservationCapacityExceeded
+		}
+		slog.Error("PrepareRunInstances: insufficient capacity", "requested", minCount, "available", allocatableCount, "InstanceType", *input.InstanceType, "reservationId", reservationID)
+		return nil, nil, nil, errors.New(errCode)
 	}
 
 	launchCount := allocatableCount
@@ -633,18 +650,32 @@ func (s *InstanceServiceImpl) PrepareRunInstances(input *ec2.RunInstancesInput,
 	// allocatableCount — the < minCount branch below rolls back.
 	var allocatedCount int
 	for i := 0; i < launchCount; i++ {
-		if err := s.resourceMgr.Allocate(instanceType); err != nil {
-			slog.Error("PrepareRunInstances: allocate failed mid-allocation", "allocated", allocatedCount, "err", err)
+		var allocErr error
+		if reservationID == "" {
+			allocErr = s.resourceMgr.Allocate(instanceType)
+		} else {
+			allocErr = s.resourceMgr.AllocateFromReservation(reservationID, accountID, instanceType)
+		}
+		if allocErr != nil {
+			slog.Error("PrepareRunInstances: allocate failed mid-allocation", "allocated", allocatedCount, "err", allocErr)
 			break
 		}
 		allocatedCount++
 	}
 	if allocatedCount < minCount {
 		for i := 0; i < allocatedCount; i++ {
-			s.resourceMgr.Deallocate(instanceType)
+			if reservationID == "" {
+				s.resourceMgr.Deallocate(instanceType)
+			} else {
+				s.resourceMgr.ReleaseToReservation(reservationID, instanceType)
+			}
+		}
+		errCode := awserrors.ErrorInsufficientInstanceCapacity
+		if reservationID != "" {
+			errCode = awserrors.ErrorReservationCapacityExceeded
 		}
 		slog.Error("PrepareRunInstances: insufficient capacity after allocation", "allocated", allocatedCount, "minCount", minCount)
-		return nil, nil, nil, errors.New(awserrors.ErrorInsufficientInstanceCapacity)
+		return nil, nil, nil, errors.New(errCode)
 	}
 	launchCount = allocatedCount
 
@@ -657,7 +688,11 @@ func (s *InstanceServiceImpl) PrepareRunInstances(input *ec2.RunInstancesInput,
 		if err != nil {
 			slog.Error("PrepareRunInstances: RunInstance failed", "index", i, "err", err)
 			lastRunErr = err
-			s.resourceMgr.Deallocate(instanceType)
+			if reservationID == "" {
+				s.resourceMgr.Deallocate(instanceType)
+			} else {
+				s.resourceMgr.ReleaseToReservation(reservationID, instanceType)
+			}
 			continue
 		}
 		instance.BootMode = amiMeta.BootMode
@@ -685,14 +720,22 @@ func (s *InstanceServiceImpl) PrepareRunInstances(input *ec2.RunInstancesInput,
 				slog.Error("PrepareRunInstances: pre-created ENI specified but ENI service unavailable",
 					"instanceId", instance.ID, "eniId", preExistingENIID)
 				lastRunErr = errors.New(awserrors.ErrorServerInternal)
-				s.resourceMgr.Deallocate(instanceType)
+				if reservationID == "" {
+					s.resourceMgr.Deallocate(instanceType)
+				} else {
+					s.resourceMgr.ReleaseToReservation(reservationID, instanceType)
+				}
 				continue
 			}
 			if err := s.attachPreCreatedENI(accountID, preExistingENIID, instance, ec2Instance); err != nil {
 				slog.Error("PrepareRunInstances: pre-created ENI attach failed",
 					"instanceId", instance.ID, "eniId", preExistingENIID, "err", err)
 				lastRunErr = err
-				s.resourceMgr.Deallocate(instanceType)
+				if reservationID == "" {
+					s.resourceMgr.Deallocate(instanceType)
+				} else {
+					s.resourceMgr.ReleaseToReservation(reservationID, instanceType)
+				}
 				continue
 			}
 			ec2Instance.SetAmiLaunchIndex(int64(len(allEC2Instances)))
@@ -718,7 +761,11 @@ func (s *InstanceServiceImpl) PrepareRunInstances(input *ec2.RunInstancesInput,
 			if eniErr != nil {
 				slog.Error("PrepareRunInstances: auto-create ENI failed", "instanceId", instance.ID, "subnetId", *input.SubnetId, "err", eniErr)
 				lastRunErr = eniErr
-				s.resourceMgr.Deallocate(instanceType)
+				if reservationID == "" {
+					s.resourceMgr.Deallocate(instanceType)
+				} else {
+					s.resourceMgr.ReleaseToReservation(reservationID, instanceType)
+				}
 				continue
 			}
 
@@ -738,7 +785,11 @@ func (s *InstanceServiceImpl) PrepareRunInstances(input *ec2.RunInstancesInput,
 						"eniId", instance.ENIId, "err", delErr)
 				}
 				lastRunErr = attachErr
-				s.resourceMgr.Deallocate(instanceType)
+				if reservationID == "" {
+					s.resourceMgr.Deallocate(instanceType)
+				} else {
+					s.resourceMgr.ReleaseToReservation(reservationID, instanceType)
+				}
 				continue
 			}
 			ec2Instance.SetPrivateIpAddress(*eni.PrivateIpAddress)
@@ -794,7 +845,11 @@ func (s *InstanceServiceImpl) PrepareRunInstances(input *ec2.RunInstancesInput,
 								"eniId", *eni.NetworkInterfaceId, "err", delErr)
 						}
 						lastRunErr = errors.New(awserrors.ErrorInsufficientAddressCapacity)
-						s.resourceMgr.Deallocate(instanceType)
+						if reservationID == "" {
+							s.resourceMgr.Deallocate(instanceType)
+						} else {
+							s.resourceMgr.ReleaseToReservation(reservationID, instanceType)
+						}
 						continue
 					} else {
 						if updateErr := s.eniCreator.UpdateENIPublicIP(accountID, *eni.NetworkInterfaceId, publicIP, poolName); updateErr != nil {
@@ -808,7 +863,11 @@ func (s *InstanceServiceImpl) PrepareRunInstances(input *ec2.RunInstancesInput,
 							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)
+							if reservationID == "" {
+								s.resourceMgr.Deallocate(instanceType)
+							} else {
+								s.resourceMgr.ReleaseToReservation(reservationID, instanceType)
+							}
 							continue
 						}
 						ec2Instance.PublicIpAddress = aws.String(publicIP)
@@ -832,7 +891,11 @@ func (s *InstanceServiceImpl) PrepareRunInstances(input *ec2.RunInstancesInput,
 
 	if len(instances) < minCount {
 		for range instances {
-			s.resourceMgr.Deallocate(instanceType)
+			if reservationID == "" {
+				s.resourceMgr.Deallocate(instanceType)
+			} else {
+				s.resourceMgr.ReleaseToReservation(reservationID, instanceType)
+			}
 		}
 		errCode := awserrors.ErrorServerInternal
 		if lastRunErr != nil {
@@ -852,6 +915,9 @@ func (s *InstanceServiceImpl) PrepareRunInstances(input *ec2.RunInstancesInput,
 	for _, instance := range instances {
 		instance.Reservation = reservation
 		instance.AccountID = accountID
+		if reservationID != "" {
+			instance.CapacityReservationId = reservationID
+		}
 		if input.Placement != nil && input.Placement.GroupName != nil && *input.Placement.GroupName != "" {
 			instance.PlacementGroupName = *input.Placement.GroupName
 		}
@@ -987,7 +1053,7 @@ func (s *InstanceServiceImpl) LaunchRunInstances(instances []*vm.VM, input *ec2.
 // RunInstances is for non-daemon callers (tests). The daemon calls
 // PrepareRunInstances + LaunchRunInstances directly to respond before launching.
 func (s *InstanceServiceImpl) RunInstances(input *ec2.RunInstancesInput, accountID string) (*ec2.Reservation, error) {
-	reservation, instances, instanceType, err := s.PrepareRunInstances(input, accountID)
+	reservation, instances, instanceType, err := s.PrepareRunInstances(input, accountID, "")
 	if err != nil {
 		return nil, err
 	}

spinifex/handlers/ec2/instance/service_impl_test.go

@@ -972,6 +972,11 @@ type fakeResourceCapacityProvider struct {
 	allocated      []*ec2.InstanceTypeInfo
 	deallocated    []*ec2.InstanceTypeInfo
 	canAllocFn     func(*ec2.InstanceTypeInfo, int) int
+
+	reservationAllocErr  error
+	reservationAllocated []*ec2.InstanceTypeInfo
+	reservationReleased  []*ec2.InstanceTypeInfo
+	reservationAvailFn   func(string, string, *ec2.InstanceTypeInfo) int
 }
 
 func (f *fakeResourceCapacityProvider) GetAvailableInstanceTypeInfos(showCapacity bool) []*ec2.InstanceTypeInfo {
@@ -1007,6 +1012,25 @@ func (f *fakeResourceCapacityProvider) CanAllocate(it *ec2.InstanceTypeInfo, cou
 	return count
 }
 
+func (f *fakeResourceCapacityProvider) AllocateFromReservation(_, _ string, it *ec2.InstanceTypeInfo) error {
+	if f.reservationAllocErr != nil {
+		return f.reservationAllocErr
+	}
+	f.reservationAllocated = append(f.reservationAllocated, it)
+	return nil
+}
+
+func (f *fakeResourceCapacityProvider) ReleaseToReservation(_ string, it *ec2.InstanceTypeInfo) {
+	f.reservationReleased = append(f.reservationReleased, it)
+}
+
+func (f *fakeResourceCapacityProvider) ReservationAvailable(reservationID, accountID string, it *ec2.InstanceTypeInfo) int {
+	if f.reservationAvailFn != nil {
+		return f.reservationAvailFn(reservationID, accountID, it)
+	}
+	return 0
+}
+
 func (f *fakeResourceCapacityProvider) InstanceTypes() map[string]*ec2.InstanceTypeInfo {
 	return f.instanceTypes
 }
@@ -2375,14 +2399,14 @@ func defaultPrepareInstanceTypes() (map[string]*ec2.InstanceTypeInfo, *ec2.Insta
 
 func TestPrepareRunInstances_MissingAccountID(t *testing.T) {
 	svc := &InstanceServiceImpl{}
-	_, _, _, err := svc.PrepareRunInstances(&ec2.RunInstancesInput{InstanceType: aws.String("t3.micro")}, "")
+	_, _, _, err := svc.PrepareRunInstances(&ec2.RunInstancesInput{InstanceType: aws.String("t3.micro")}, "", "")
 	require.Error(t, err)
 	assert.Equal(t, awserrors.ErrorServerInternal, err.Error())
 }
 
 func TestPrepareRunInstances_MissingInstanceType(t *testing.T) {
 	svc := &InstanceServiceImpl{}
-	_, _, _, err := svc.PrepareRunInstances(&ec2.RunInstancesInput{}, "acc")
+	_, _, _, err := svc.PrepareRunInstances(&ec2.RunInstancesInput{}, "acc", "")
 	require.Error(t, err)
 	assert.Equal(t, awserrors.ErrorMissingParameter, err.Error())
 }
@@ -2391,7 +2415,7 @@ func TestPrepareRunInstances_InvalidInstanceType(t *testing.T) {
 	svc := &InstanceServiceImpl{instanceTypes: map[string]*ec2.InstanceTypeInfo{}}
 	_, _, _, err := svc.PrepareRunInstances(&ec2.RunInstancesInput{
 		InstanceType: aws.String("unknown.type"),
-	}, "acc")
+	}, "acc", "")
 	require.Error(t, err)
 	assert.Equal(t, awserrors.ErrorInvalidInstanceType, err.Error())
 }
@@ -2401,7 +2425,7 @@ func TestPrepareRunInstances_MissingImageID(t *testing.T) {
 	svc := &InstanceServiceImpl{instanceTypes: types}
 	_, _, _, err := svc.PrepareRunInstances(&ec2.RunInstancesInput{
 		InstanceType: aws.String("t3.micro"),
-	}, "acc")
+	}, "acc", "")
 	require.Error(t, err)
 	assert.Equal(t, awserrors.ErrorMissingParameter, err.Error())
 }
@@ -2412,7 +2436,7 @@ func TestPrepareRunInstances_NilAMILoader(t *testing.T) {
 	_, _, _, err := svc.PrepareRunInstances(&ec2.RunInstancesInput{
 		InstanceType: aws.String("t3.micro"),
 		ImageId:      aws.String("ami-1"),
-	}, "acc")
+	}, "acc", "")
 	require.Error(t, err)
 	assert.Equal(t, awserrors.ErrorServerInternal, err.Error())
 }
@@ -2426,7 +2450,7 @@ func TestPrepareRunInstances_AMINotFound(t *testing.T) {
 	_, _, _, err := svc.PrepareRunInstances(&ec2.RunInstancesInput{
 		InstanceType: aws.String("t3.micro"),
 		ImageId:      aws.String("ami-missing"),
-	}, "acc")
+	}, "acc", "")
 	require.Error(t, err)
 	assert.Equal(t, awserrors.ErrorInvalidAMIIDNotFound, err.Error())
 }
@@ -2442,7 +2466,7 @@ func TestPrepareRunInstances_AMINotOwnedByCaller(t *testing.T) {
 	_, _, _, err := svc.PrepareRunInstances(&ec2.RunInstancesInput{
 		InstanceType: aws.String("t3.micro"),
 		ImageId:      aws.String("ami-other"),
-	}, "111122223333")
+	}, "111122223333", "")
 	require.Error(t, err)
 	assert.Equal(t, awserrors.ErrorInvalidAMIIDNotFound, err.Error())
 }
@@ -2462,7 +2486,7 @@ func TestPrepareRunInstances_KeyPairNotFound(t *testing.T) {
 		KeyName:      aws.String("nope"),
 		MinCount:     aws.Int64(1),
 		MaxCount:     aws.Int64(1),
-	}, "acc")
+	}, "acc", "")
 	require.Error(t, err)
 	assert.Equal(t, awserrors.ErrorInvalidKeyPairNotFound, err.Error())
 }
@@ -2486,7 +2510,7 @@ func TestPrepareRunInstances_InsufficientCapacity(t *testing.T) {
 		ImageId:      aws.String("ami-1"),
 		MinCount:     aws.Int64(1),
 		MaxCount:     aws.Int64(5),
-	}, "acc")
+	}, "acc", "")
 	require.Error(t, err)
 	assert.Equal(t, awserrors.ErrorInsufficientInstanceCapacity, err.Error())
 }
@@ -2510,7 +2534,7 @@ func TestPrepareRunInstances_HappyPathNoENI(t *testing.T) {
 		ImageId:      aws.String("ami-1"),
 		MinCount:     aws.Int64(2),
 		MaxCount:     aws.Int64(2),
-	}, "acc")
+	}, "acc", "")
 	require.NoError(t, err)
 	require.NotNil(t, reservation)
 	require.Len(t, instances, 2)
@@ -2522,6 +2546,124 @@ func TestPrepareRunInstances_HappyPathNoENI(t *testing.T) {
 	}
 }
 
+// A targeted launch consumes slots from the reservation (not the general pool)
+// and stamps the reservation id onto every prepared VM so terminate can restore.
+func TestPrepareRunInstances_ConsumesReservation(t *testing.T) {
+	types, _ := defaultPrepareInstanceTypes()
+	prov := &fakeResourceCapacityProvider{
+		instanceTypes:      types,
+		reservationAvailFn: func(_, _ string, _ *ec2.InstanceTypeInfo) int { return 3 },
+	}
+	svc := &InstanceServiceImpl{
+		config:        &config.Config{},
+		instanceTypes: types,
+		amiLoader: &fakeAMILoader{byID: map[string]viperblock.AMIMetadata{
+			"ami-1": {ImageOwnerAlias: "acc"},
+		}},
+		resourceMgr: prov,
+	}
+	_, instances, _, err := svc.PrepareRunInstances(&ec2.RunInstancesInput{
+		InstanceType: aws.String("t3.micro"),
+		ImageId:      aws.String("ami-1"),
+		MinCount:     aws.Int64(2),
+		MaxCount:     aws.Int64(2),
+	}, "acc", "cr-123")
+	require.NoError(t, err)
+	require.Len(t, instances, 2)
+	assert.Len(t, prov.reservationAllocated, 2, "slots consumed from the reservation")
+	assert.Empty(t, prov.allocated, "targeted launch must not touch general capacity")
+	for _, inst := range instances {
+		assert.Equal(t, "cr-123", inst.CapacityReservationId, "instance stamped with reservation id")
+	}
+}
+
+// A targeted launch is capped at the reservation's free slots and never spills
+// onto general capacity: MaxCount past Available yields exactly Available.
+func TestPrepareRunInstances_ReservationCapsNoSpill(t *testing.T) {
+	types, _ := defaultPrepareInstanceTypes()
+	prov := &fakeResourceCapacityProvider{
+		instanceTypes:      types,
+		reservationAvailFn: func(_, _ string, _ *ec2.InstanceTypeInfo) int { return 2 },
+	}
+	svc := &InstanceServiceImpl{
+		config:        &config.Config{},
+		instanceTypes: types,
+		amiLoader: &fakeAMILoader{byID: map[string]viperblock.AMIMetadata{
+			"ami-1": {ImageOwnerAlias: "acc"},
+		}},
+		resourceMgr: prov,
+	}
+	_, instances, _, err := svc.PrepareRunInstances(&ec2.RunInstancesInput{
+		InstanceType: aws.String("t3.micro"),
+		ImageId:      aws.String("ami-1"),
+		MinCount:     aws.Int64(1),
+		MaxCount:     aws.Int64(5),
+	}, "acc", "cr-123")
+	require.NoError(t, err)
+	assert.Len(t, instances, 2, "launch capped at the 2 available reservation slots")
+	assert.Empty(t, prov.allocated, "no general-pool allocation for a targeted launch")
+}
+
+// When the reservation has fewer free slots than MinCount the launch is rejected
+// with ReservationCapacityExceeded and nothing is allocated.
+func TestPrepareRunInstances_ReservationExceeded(t *testing.T) {
+	types, _ := defaultPrepareInstanceTypes()
+	prov := &fakeResourceCapacityProvider{
+		instanceTypes:      types,
+		reservationAvailFn: func(_, _ string, _ *ec2.InstanceTypeInfo) int { return 1 },
+	}
+	svc := &InstanceServiceImpl{
+		config:        &config.Config{},
+		instanceTypes: types,
+		amiLoader: &fakeAMILoader{byID: map[string]viperblock.AMIMetadata{
+			"ami-1": {ImageOwnerAlias: "acc"},
+		}},
+		resourceMgr: prov,
+	}
+	_, _, _, err := svc.PrepareRunInstances(&ec2.RunInstancesInput{
+		InstanceType: aws.String("t3.micro"),
+		ImageId:      aws.String("ami-1"),
+		MinCount:     aws.Int64(2),
+		MaxCount:     aws.Int64(2),
+	}, "acc", "cr-123")
+	require.Error(t, err)
+	assert.Equal(t, awserrors.ErrorReservationCapacityExceeded, err.Error())
+	assert.Empty(t, prov.reservationAllocated, "nothing allocated when below MinCount")
+}
+
+// Invariant guardrail: a mid-launch failure on the reservation path must return
+// every consumed slot to the reservation (reservationReleased == reservationAllocated)
+// and never leak one onto the general pool (Deallocate untouched). Each per-instance
+// ENI create fails, exercising a rollback site inside the launch loop.
+func TestPrepareRunInstances_ReservationRollbackNoGeneralPoolLeak(t *testing.T) {
+	types, _ := defaultPrepareInstanceTypes()
+	prov := &fakeResourceCapacityProvider{
+		instanceTypes:      types,
+		reservationAvailFn: func(_, _ string, _ *ec2.InstanceTypeInfo) int { return 2 },
+	}
+	eni := &fakeENICreator{createErr: errors.New("eni create failed")} // fails every call
+	svc := &InstanceServiceImpl{
+		config:        &config.Config{Region: "us-east-1", AZ: "us-east-1a"},
+		instanceTypes: types,
+		amiLoader: &fakeAMILoader{byID: map[string]viperblock.AMIMetadata{
+			"ami-1": {ImageOwnerAlias: "acc"},
+		}},
+		resourceMgr: prov,
+		eniCreator:  eni,
+	}
+	_, _, _, err := svc.PrepareRunInstances(&ec2.RunInstancesInput{
+		InstanceType: aws.String("t3.micro"),
+		ImageId:      aws.String("ami-1"),
+		SubnetId:     aws.String("subnet-1"),
+		MinCount:     aws.Int64(2),
+		MaxCount:     aws.Int64(2),
+	}, "acc", "cr-123")
+	require.Error(t, err)
+	assert.Len(t, prov.reservationAllocated, 2, "both slots allocated from the reservation")
+	assert.Len(t, prov.reservationReleased, 2, "both reservation slots restored on rollback")
+	assert.Empty(t, prov.deallocated, "no reservation-bound slot may leak to the general pool")
+}
+
 // TestPrepareRunInstances_AmiLaunchIndexContiguous pins that ami-launch-index is
 // assigned per successful launch (0..n-1) so it flows to DescribeInstances and the
 // IMDS identity document. Survivors of a mid-loop failure stay contiguous (no gap).
@@ -2545,7 +2687,7 @@ func TestPrepareRunInstances_AmiLaunchIndexContiguous(t *testing.T) {
 			ImageId:      aws.String("ami-1"),
 			MinCount:     aws.Int64(3),
 			MaxCount:     aws.Int64(3),
-		}, "acc")
+		}, "acc", "")
 		require.NoError(t, err)
 		require.Len(t, instances, 3)
 		require.Len(t, reservation.Instances, 3)
@@ -2578,7 +2720,7 @@ func TestPrepareRunInstances_AmiLaunchIndexContiguous(t *testing.T) {
 			SubnetId:     aws.String("subnet-1"),
 			MinCount:     aws.Int64(2),
 			MaxCount:     aws.Int64(3),
-		}, "acc")
+		}, "acc", "")
 		require.NoError(t, err)
 		require.Len(t, instances, 2)
 		require.Len(t, reservation.Instances, 2)
@@ -2621,7 +2763,7 @@ func TestPrepareRunInstances_BootModePropagated(t *testing.T) {
 				ImageId:      aws.String("ami-1"),
 				MinCount:     aws.Int64(1),
 				MaxCount:     aws.Int64(1),
-			}, "acc")
+			}, "acc", "")
 			require.NoError(t, err)
 			require.Len(t, instances, 1)
 			assert.Equal(t, tc.wantBootMode, instances[0].BootMode)
@@ -2826,7 +2968,7 @@ func TestPrepareRunInstances_DefaultSubnetResolved(t *testing.T) {
 		ImageId:      aws.String("ami-1"),
 		MinCount:     aws.Int64(1),
 		MaxCount:     aws.Int64(1),
-	}, "acc")
+	}, "acc", "")
 	require.NoError(t, err)
 	require.Len(t, instances, 1)
 	assert.Equal(t, "eni-1", instances[0].ENIId)
@@ -2882,7 +3024,7 @@ func TestPrepareRunInstances_PublicIPAutoAssigned(t *testing.T) {
 				}
 			}
 
-			_, instances, _, err := svc.PrepareRunInstances(input, "acc")
+			_, instances, _, err := svc.PrepareRunInstances(input, "acc", "")
 			require.NoError(t, err)
 			require.Len(t, instances, 1)
 			if tc.wantPublic {
@@ -2949,7 +3091,7 @@ func TestPrepareRunInstances_NATFailureRollsBackPublicIP(t *testing.T) {
 		SubnetId:     aws.String("subnet-1"),
 		MinCount:     aws.Int64(1),
 		MaxCount:     aws.Int64(1),
-	}, "acc")
+	}, "acc", "")
 
 	// MinCount=1 with a single launch attempt that NAT-failed must drop
 	// below minimum and return ServerInternal — the wrapped natErr is not
@@ -3018,7 +3160,7 @@ func TestPrepareRunInstances_PublicIPAllocFailureAbortsLaunch(t *testing.T) {
 		SubnetId:     aws.String("subnet-1"),
 		MinCount:     aws.Int64(1),
 		MaxCount:     aws.Int64(1),
-	}, "acc")
+	}, "acc", "")
 
 	require.Error(t, err)
 	assert.Equal(t, awserrors.ErrorInsufficientAddressCapacity, err.Error(),
@@ -3053,7 +3195,7 @@ func TestPrepareRunInstances_ENICreateFailureDeallocates(t *testing.T) {
 		SubnetId:     aws.String("subnet-1"),
 		MinCount:     aws.Int64(1),
 		MaxCount:     aws.Int64(1),
-	}, "acc")
+	}, "acc", "")
 	require.Error(t, err)
 	// MinCount not satisfied → InsufficientInstanceCapacity (no known
 	// AWS code for raw ENI failure).
@@ -3086,7 +3228,7 @@ func TestPrepareRunInstances_ENIAttachFailureRollsBack(t *testing.T) {
 		SubnetId:     aws.String("subnet-1"),
 		MinCount:     aws.Int64(1),
 		MaxCount:     aws.Int64(1),
-	}, "acc")
+	}, "acc", "")
 
 	require.Error(t, err)
 	assert.Equal(t, awserrors.ErrorServerInternal, err.Error())
@@ -3122,7 +3264,7 @@ func TestPrepareRunInstances_NetworkInterfaceLifted(t *testing.T) {
 				Groups:   []*string{aws.String("sg-1")},
 			},
 		},
-	}, "acc")
+	}, "acc", "")
 	require.NoError(t, err)
 	require.Len(t, instances, 1)
 	assert.Equal(t, "eni-3", instances[0].ENIId)
@@ -3148,7 +3290,7 @@ func TestPrepareRunInstances_PlacementGroup(t *testing.T) {
 		MinCount:     aws.Int64(1),
 		MaxCount:     aws.Int64(1),
 		Placement:    &ec2.Placement{GroupName: aws.String("pg-1")},
-	}, "acc")
+	}, "acc", "")
 	require.NoError(t, err)
 	require.Len(t, instances, 1)
 	assert.Equal(t, "pg-1", instances[0].PlacementGroupName)
@@ -3174,7 +3316,7 @@ func TestPrepareRunInstances_AllocateFailsMidLoop(t *testing.T) {
 		ImageId:      aws.String("ami-1"),
 		MinCount:     aws.Int64(1),
 		MaxCount:     aws.Int64(3),
-	}, "acc")
+	}, "acc", "")
 	require.Error(t, err)
 	assert.Equal(t, awserrors.ErrorInsufficientInstanceCapacity, err.Error())
 }
@@ -3530,7 +3672,7 @@ func TestPrepareRunInstances_PreCreatedENIAttachedSkipsAutoCreate(t *testing.T)
 			NetworkInterfaceId: aws.String("eni-pre"),
 			DeviceIndex:        aws.Int64(0),
 		}},
-	}, "acc")
+	}, "acc", "")
 	require.NoError(t, err)
 	require.Len(t, instances, 1)
 	assert.Equal(t, "eni-pre", instances[0].ENIId)
@@ -3558,7 +3700,7 @@ func TestPrepareRunInstances_PreCreatedENIInUseRejected(t *testing.T) {
 		NetworkInterfaces: []*ec2.InstanceNetworkInterfaceSpecification{{
 			NetworkInterfaceId: aws.String("eni-busy"),
 		}},
-	}, "acc")
+	}, "acc", "")
 	require.Error(t, err)
 	assert.Equal(t, awserrors.ErrorInvalidNetworkInterfaceInUse, err.Error())
 	assert.Equal(t, 0, eni.attachCalls, "in-use ENI must not be attached")
@@ -3579,7 +3721,7 @@ func TestPrepareRunInstances_PreCreatedENILookupErrorSurfaced(t *testing.T) {
 		NetworkInterfaces: []*ec2.InstanceNetworkInterfaceSpecification{{
 			NetworkInterfaceId: aws.String("eni-ghost"),
 		}},
-	}, "acc")
+	}, "acc", "")
 	require.Error(t, err)
 	assert.Equal(t, awserrors.ErrorInvalidNetworkInterfaceIDNotFound, err.Error())
 	assert.Equal(t, 0, eni.attachCalls)
@@ -3610,7 +3752,7 @@ func TestPrepareRunInstances_PreCreatedENIAttachErrorSurfaced(t *testing.T) {
 		NetworkInterfaces: []*ec2.InstanceNetworkInterfaceSpecification{{
 			NetworkInterfaceId: aws.String("eni-pre"),
 		}},
-	}, "acc")
+	}, "acc", "")
 	require.Error(t, err)
 	assert.Equal(t, awserrors.ErrorServerInternal, err.Error())
 	assert.Equal(t, 1, eni.attachCalls)

spinifex/vm/crash_recovery.go

@@ -114,7 +114,16 @@ func (m *Manager) HandleCrash(instance *VM, waitErr error) {
 	if m.deps.Resources != nil && instance.InstanceType != "" {
 		slog.Info("Deallocating resources for crashed instance",
 			"instance", instance.ID, "type", instance.InstanceType)
-		m.deps.Resources.Deallocate(instance.InstanceType)
+		m.deallocateResources(instance)
+	}
+
+	// The slot just returned to the reservation (deallocateResources above), but
+	// the restart re-allocates from the general pool, so drop the binding under the
+	// manager lock. This keeps the crash release and restart allocation on the same
+	// pool; leaving it set would drift the shared reservation counter on the later
+	// stop/terminate while other instances still consume the reservation.
+	if instance.CapacityReservationId != "" {
+		m.UpdateState(instance.ID, func(v *VM) { v.CapacityReservationId = "" })
 	}
 
 	if instance.Config.QMPSocket != "" {
@@ -246,7 +255,7 @@ func (m *Manager) RestartCrashedInstance(instance *VM) {
 		if err := m.deps.TransitionState(instance, StatePending); err != nil {
 			slog.Error("Failed to transition instance to pending for restart",
 				"instance", instance.ID, "err", err)
-			m.deps.Resources.Deallocate(instance.InstanceType)
+			m.deallocateResources(instance)
 			return
 		}
 	}

spinifex/vm/crash_recovery_test.go

@@ -161,6 +161,53 @@ func TestHandleCrash_CoreFlow(t *testing.T) {
 	assert.True(t, os.IsNotExist(err), "QMP socket must be removed")
 }
 
+// A reservation-bound instance returns its slot to the reservation through the
+// deallocateResources chokepoint, never to the general pool (net-zero swap).
+func TestHandleCrash_ReservationBoundReleasesToReservation(t *testing.T) {
+	m, rc, _, _ := crashTestManager(t)
+
+	tmpDir := t.TempDir()
+	qmpPath := filepath.Join(tmpDir, "qmp.sock")
+	require.NoError(t, os.WriteFile(qmpPath, []byte("dummy"), 0o600))
+
+	instance := &VM{
+		ID:                    "i-cr",
+		Status:                StateRunning,
+		InstanceType:          "t3.micro",
+		CapacityReservationId: "cr-0123456789abcdef0",
+		Config:                Config{QMPSocket: qmpPath},
+	}
+	m.Insert(instance)
+
+	m.HandleCrash(instance, fmt.Errorf("test crash"))
+
+	assert.Equal(t, 1, rc.releaseCount(), "slot must return to the reservation")
+	assert.Zero(t, rc.deallocateCount("t3.micro"), "reservation-bound release must not hit the general pool")
+}
+
+// A crashed reservation-bound instance drops its binding: the slot went back to
+// the reservation, and the restart runs on general capacity, so a later
+// stop/terminate must not double-credit the reservation's shared counter.
+func TestHandleCrash_ReservationBoundClearsBindingForRestart(t *testing.T) {
+	m, rc, _, _ := crashTestManager(t)
+
+	instance := &VM{
+		ID:                    "i-cr-clear",
+		Status:                StateRunning,
+		InstanceType:          "t3.micro",
+		CapacityReservationId: "cr-0123456789abcdef0",
+	}
+	m.Insert(instance)
+
+	m.HandleCrash(instance, fmt.Errorf("test crash"))
+
+	assert.Equal(t, 1, rc.releaseCount(), "slot must return to the reservation on crash")
+	assert.Empty(t, instance.CapacityReservationId, "binding must be cleared so restart uses general capacity")
+	stored, ok := m.Get(instance.ID)
+	require.True(t, ok)
+	assert.Empty(t, stored.CapacityReservationId, "cleared binding must be persisted on the stored VM")
+}
+
 func TestHandleCrash_FirstCrashSetsTime(t *testing.T) {
 	m, _, _, _ := crashTestManager(t)
 

spinifex/vm/fakes_test.go

@@ -9,6 +9,7 @@ type fakeResourceController struct {
 	mu             sync.Mutex
 	allocated      map[string]int
 	deallocated    []string
+	released       []string // "<reservationID>/<type>" per ReleaseToReservation call
 	allocateErr    error
 	canAllocateRet int
 }
@@ -36,6 +37,21 @@ func (f *fakeResourceController) Deallocate(typeName string) {
 	}
 }
 
+func (f *fakeResourceController) ReleaseToReservation(reservationID, typeName string) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.released = append(f.released, reservationID+"/"+typeName)
+	if f.allocated[typeName] > 0 {
+		f.allocated[typeName]--
+	}
+}
+
+func (f *fakeResourceController) releaseCount() int {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	return len(f.released)
+}
+
 func (f *fakeResourceController) CanAllocate(typeName string, count int) int {
 	f.mu.Lock()
 	defer f.mu.Unlock()

spinifex/vm/instance_type_resolver.go

@@ -21,6 +21,11 @@ type InstanceTypeResolver interface {
 type ResourceController interface {
 	Allocate(instanceType string) error
 	Deallocate(instanceType string)
+	// ReleaseToReservation returns instanceType's compute to the capacity
+	// reservation reservationID instead of the general pool — the net-zero
+	// inverse of the launch-time swap, used when a reservation-bound instance
+	// stops or terminates. Frees to the general pool if the reservation is gone.
+	ReleaseToReservation(reservationID, instanceType string)
 	// CanAllocate returns how many instances of instanceType could be
 	// allocated right now (0 to count). Used by the crash-recovery
 	// scheduler to avoid restarting a VM the host can't fit.

spinifex/vm/shutdown.go

@@ -282,6 +282,14 @@ func (m *Manager) stopCleanup(instance *VM) {
 		}
 	}
 	m.deallocateResources(instance)
+
+	// The slot just returned to the reservation; detach the binding (under the
+	// manager lock, mirroring crash recovery) so a later start re-allocates from
+	// the general pool and terminate frees there too — never double-counting the
+	// reservation while the stopped instance no longer holds a slot.
+	if instance.CapacityReservationId != "" {
+		m.UpdateState(instance.ID, func(v *VM) { v.CapacityReservationId = "" })
+	}
 }
 
 // terminateCleanup is stopCleanup plus the AWS-resource cleanup that
@@ -398,11 +406,20 @@ func (m *Manager) cleanupExtraENITaps(instance *VM) {
 }
 
 // deallocateResources releases the per-instance vCPU/memory reservation
-// back to the resource controller.
+// back to the resource controller. The single release/restore chokepoint for
+// stop, terminate and crash recovery.
 func (m *Manager) deallocateResources(instance *VM) {
 	if m.deps.Resources == nil || instance.InstanceType == "" {
 		return
 	}
+	// A reservation-bound instance returns its slot to the reservation, not the
+	// general pool. CapacityReservationId is set at launch and only cleared (under
+	// the manager lock, on crash before a general-capacity restart), so the
+	// stop/terminate/crash reads here do not overlap that clear.
+	if instance.CapacityReservationId != "" {
+		m.deps.Resources.ReleaseToReservation(instance.CapacityReservationId, instance.InstanceType)
+		return
+	}
 	m.deps.Resources.Deallocate(instance.InstanceType)
 }
 

spinifex/vm/shutdown_test.go

@@ -555,6 +555,30 @@ func TestStopCleanup_InvokesReleaseGPU(t *testing.T) {
 	}
 }
 
+// TestStopCleanup_ReservationBoundReleasesAndClears proves a reservation-bound
+// instance returns its slot to the reservation on stop and then detaches the
+// binding, so a later start re-allocates from the general pool and terminate
+// frees there too — never double-counting the reservation.
+func TestStopCleanup_ReservationBoundReleasesAndClears(t *testing.T) {
+	t.Setenv("XDG_RUNTIME_DIR", t.TempDir())
+	rc := &countingResourceController{}
+	m := NewManagerWithDeps(Deps{Resources: rc})
+	instance := &VM{ID: "i-cr", InstanceType: "t3.micro", CapacityReservationId: "cr-123"}
+	m.Insert(instance)
+
+	m.stopCleanup(instance)
+
+	if got := rc.releasedToCRID; len(got) != 1 || got[0] != "cr-123" {
+		t.Fatalf("stop must release the slot to the reservation: got %v, want [cr-123]", got)
+	}
+	if rc.deallocations != 0 {
+		t.Fatalf("stop must not free a reservation-bound instance to the general pool: deallocations=%d", rc.deallocations)
+	}
+	if instance.CapacityReservationId != "" {
+		t.Fatalf("stop must clear the reservation binding so start uses general capacity: got %q", instance.CapacityReservationId)
+	}
+}
+
 func TestTerminateCleanup_InvokesReleaseGPU(t *testing.T) {
 	t.Setenv("XDG_RUNTIME_DIR", t.TempDir())
 	cleaner := &recordingInstanceCleaner{}
@@ -1082,13 +1106,20 @@ func TestClassifyRestoredInstances_StateErrorSkipsRelaunch(t *testing.T) {
 }
 
 // countingResourceController counts how often Allocate fires so a test
-// can prove the restore path skipped its resource-allocation branch.
+// can prove the restore path skipped its resource-allocation branch. It
+// also records general deallocations and reservation releases so stop
+// tests can assert which pool a reservation-bound instance returns to.
 type countingResourceController struct {
-	allocations int
+	allocations    int
+	deallocations  int
+	releasedToCRID []string
 }
 
-func (c *countingResourceController) Allocate(_ string) error         { c.allocations++; return nil }
-func (c *countingResourceController) Deallocate(_ string)             {}
+func (c *countingResourceController) Allocate(_ string) error { c.allocations++; return nil }
+func (c *countingResourceController) Deallocate(_ string)     { c.deallocations++ }
+func (c *countingResourceController) ReleaseToReservation(crID, _ string) {
+	c.releasedToCRID = append(c.releasedToCRID, crID)
+}
 func (c *countingResourceController) CanAllocate(_ string, n int) int { return n }
 
 // signalingStore is a no-op StateStore that fires onSave once per

spinifex/vm/vm.go

@@ -140,6 +140,11 @@ type VM struct {
 	// the visibility window, after which it may reclaim it early; the bucket's
 	// 1h TTL bounds visibility regardless.
 	TerminatedAt time.Time `json:"terminated_at,omitzero"`
+
+	// CapacityReservationId is the On-Demand Capacity Reservation this instance
+	// was launched into (targeted launch). Empty for instances on general
+	// capacity. Drives slot restore to the reservation on stop/terminate.
+	CapacityReservationId string `json:"capacity_reservation_id,omitempty"`
 }
 
 // ResetNodeLocalState zeroes node-specific fields after deserializing a VM

tests/e2e/Makefile

@@ -1,4 +1,4 @@
-PKG         ?= ./cert/... ./lb/... ./eks/... ./iam/... ./multinode/... ./baremetal/... ./reboot/... ./tofu/...
+PKG         ?= ./cert/... ./lb/... ./eks/... ./iam/... ./ecr/... ./multinode/... ./baremetal/... ./reboot/... ./tofu/...
 ARTIFACT_DIR ?= /tmp/spinifex-e2e-artifacts
 TIMEOUT     ?= 30m
 MODE        ?= single
@@ -11,9 +11,9 @@ SUITE_TAG   ?= $(notdir $(patsubst %/...,%,$(firstword $(PKG))))
 # named suite into BIN_DIR/<suite>.test so CI can ship binaries (not source +
 # go toolchain) to the remote VM.
 BIN_DIR      ?= $(CURDIR)/_bin
-SUITES_BUILD ?= cert lb eks single iam gpu
+SUITES_BUILD ?= cert lb eks single iam ecr gpu
 
-.PHONY: e2e e2e-cert e2e-lb e2e-eks e2e-single e2e-iam e2e-multinode e2e-bm e2e-reboot e2e-tofu e2e-gpu e2e-all clean e2e-build
+.PHONY: e2e e2e-cert e2e-lb e2e-eks e2e-single e2e-iam e2e-ecr e2e-multinode e2e-bm e2e-reboot e2e-tofu e2e-gpu e2e-all clean e2e-build
 
 ifeq ($(GOTESTSUM),1)
 E2E_CMD = gotestsum \
@@ -37,6 +37,7 @@ e2e-lb:         ; $(MAKE) e2e PKG=./lb/...
 e2e-eks:        ; $(MAKE) e2e PKG=./eks/...
 e2e-single:     ; $(MAKE) e2e PKG=./single/...
 e2e-iam:        ; $(MAKE) e2e PKG=./iam/...
+e2e-ecr:        ; $(MAKE) e2e PKG=./ecr/...
 e2e-multinode:  ; $(MAKE) e2e PKG=./multinode/... MODE=multinode
 e2e-bm:         ; $(MAKE) e2e PKG=./baremetal/... MODE=baremetal
 e2e-reboot:     ; $(MAKE) e2e PKG=./reboot/...

tests/e2e/ecr/capacity_test.go

@@ -0,0 +1,14 @@
+//go:build e2e
+
+package ecr
+
+import "testing"
+
+// TestECRCapacityLoad is the predastore capacity/load leg of Sprint 2f (parallel
+// crane push of many large images, asserting predastore throughput with no
+// manifest corruption). Deferred: it depends on predastore fixes mulga-siv-359
+// (':'-in-key SigV4 canonicalization) and mulga-siv-356 (404 NoSuchBucket),
+// which are owned elsewhere. Unskip once those land.
+func TestECRCapacityLoad(t *testing.T) {
+	t.Skip("deferred: blocked on predastore mulga-siv-359 / mulga-siv-356")
+}

tests/e2e/ecr/crane_skopeo_test.go

@@ -0,0 +1,114 @@
+//go:build e2e
+
+package ecr
+
+import (
+	"archive/tar"
+	"crypto/rand"
+	"os"
+	"path/filepath"
+	"testing"
+	"time"
+
+	"github.com/mulgadc/spinifex/tests/e2e/harness"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+const ociCmdTimeout = 2 * time.Minute
+
+// writeLayerTar writes a single-file tar (an OCI layer) with random contents and
+// returns its path.
+func writeLayerTar(t *testing.T, dir string) string {
+	t.Helper()
+	path := filepath.Join(dir, "layer.tar")
+	fh, err := os.Create(path) //nolint:gosec // test-controlled path under TmpDir
+	require.NoError(t, err)
+	defer fh.Close()
+
+	payload := make([]byte, 8192)
+	_, err = rand.Read(payload)
+	require.NoError(t, err)
+
+	tw := tar.NewWriter(fh)
+	require.NoError(t, tw.WriteHeader(&tar.Header{
+		Name: "payload", Mode: 0o600, Size: int64(len(payload)),
+	}))
+	_, err = tw.Write(payload)
+	require.NoError(t, err)
+	require.NoError(t, tw.Close())
+	return path
+}
+
+// craneLogin authenticates crane against host, writing the auth file under
+// dockerConfig so subsequent crane calls reuse it.
+func craneLogin(t *testing.T, crane *harness.ExternalCLI, host, pass string) {
+	t.Helper()
+	out, err := crane.Run(ociCmdTimeout, "auth", "login", host, "-u", "AWS", "-p", pass)
+	require.NoErrorf(t, err, "crane auth login: %s", out)
+}
+
+// TestCranePushPull pushes a scratch image with `crane append` and pulls it
+// back, exercising the Bearer /v2/token flow crane follows.
+func TestCranePushPull(t *testing.T) {
+	f := requireECRFixture(t)
+	host := harness.ECRRegistryHost(f.Account)
+	harness.RequireRegistryResolves(t, host)
+
+	dockerConfig := filepath.Join(f.TmpDir, "crane")
+	require.NoError(t, os.MkdirAll(dockerConfig, 0o700))
+	crane := harness.NewCrane(t, dockerConfig)
+	craneLogin(t, crane, host, harness.ECRGetLoginPassword(t, f.AWS))
+
+	repo := uniqueRepo("crane")
+	harness.CreateECRRepository(t, f.AWS, repo)
+	ref := harness.ECRRepositoryURI(f.Account, repo) + ":v1"
+
+	layer := writeLayerTar(t, f.TmpDir)
+	out, err := crane.Run(ociCmdTimeout, "append", "-f", layer, "-t", ref)
+	require.NoErrorf(t, err, "crane append: %s", out)
+
+	require.True(t, harness.ECRWaitImageTag(t, f.AWS, repo, "v1", 30*time.Second),
+		"crane-pushed tag v1 not visible via DescribeImages")
+
+	digest, err := crane.Run(ociCmdTimeout, "digest", ref)
+	require.NoErrorf(t, err, "crane digest: %s", digest)
+	assert.Contains(t, digest, "sha256:")
+
+	out, err = crane.Run(ociCmdTimeout, "pull", ref, filepath.Join(f.TmpDir, "pulled.tar"))
+	require.NoErrorf(t, err, "crane pull: %s", out)
+}
+
+// TestSkopeoCopy seeds an image with crane, then `skopeo copy` retags it within
+// the registry — exercising skopeo's Bearer auth on both read and write.
+func TestSkopeoCopy(t *testing.T) {
+	f := requireECRFixture(t)
+	host := harness.ECRRegistryHost(f.Account)
+	harness.RequireRegistryResolves(t, host)
+
+	dockerConfig := filepath.Join(f.TmpDir, "skopeo")
+	require.NoError(t, os.MkdirAll(dockerConfig, 0o700))
+	crane := harness.NewCrane(t, dockerConfig)
+	skopeo := harness.NewSkopeo(t, dockerConfig)
+
+	pass := harness.ECRGetLoginPassword(t, f.AWS)
+	craneLogin(t, crane, host, pass)
+
+	repo := uniqueRepo("skopeo")
+	harness.CreateECRRepository(t, f.AWS, repo)
+	srcRef := harness.ECRRepositoryURI(f.Account, repo) + ":src"
+	dstRef := harness.ECRRepositoryURI(f.Account, repo) + ":dst"
+
+	layer := writeLayerTar(t, f.TmpDir)
+	out, err := crane.Run(ociCmdTimeout, "append", "-f", layer, "-t", srcRef)
+	require.NoErrorf(t, err, "crane append: %s", out)
+
+	creds := "AWS:" + pass
+	out, err = skopeo.Run(ociCmdTimeout, "copy",
+		"--src-creds", creds, "--dest-creds", creds,
+		"docker://"+srcRef, "docker://"+dstRef)
+	require.NoErrorf(t, err, "skopeo copy: %s", out)
+
+	require.True(t, harness.ECRWaitImageTag(t, f.AWS, repo, "dst", 30*time.Second),
+		"skopeo-copied tag dst not visible via DescribeImages")
+}

tests/e2e/ecr/docker_push_test.go

@@ -0,0 +1,80 @@
+//go:build e2e
+
+package ecr
+
+import (
+	"crypto/rand"
+	"os"
+	"path/filepath"
+	"testing"
+	"time"
+
+	"github.com/mulgadc/spinifex/tests/e2e/harness"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+const dockerCmdTimeout = 2 * time.Minute
+
+// buildScratchContext writes a hermetic FROM-scratch build context (no network
+// pulls) with a random payload layer, returning the context dir.
+func buildScratchContext(t *testing.T, dir string) string {
+	t.Helper()
+	payload := make([]byte, 4096)
+	_, err := rand.Read(payload)
+	require.NoError(t, err)
+	require.NoError(t, os.WriteFile(filepath.Join(dir, "payload"), payload, 0o600))
+	require.NoError(t, os.WriteFile(filepath.Join(dir, "Dockerfile"),
+		[]byte("FROM scratch\nCOPY payload /payload\n"), 0o600))
+	return dir
+}
+
+// TestDockerPushPull drives the canonical workflow: aws ecr get-login-password
+// -> docker login -> docker push -> DescribeImages -> docker pull (clean daemon)
+// against the cluster registry.
+func TestDockerPushPull(t *testing.T) {
+	f := requireECRFixture(t)
+	host := harness.ECRRegistryHost(f.Account)
+	harness.RequireRegistryResolves(t, host)
+
+	dockerConfig := filepath.Join(f.TmpDir, "docker")
+	require.NoError(t, os.MkdirAll(dockerConfig, 0o700))
+	docker := harness.NewDocker(t, dockerConfig)
+
+	repo := uniqueRepo("docker")
+	harness.CreateECRRepository(t, f.AWS, repo)
+	ref := harness.ECRRepositoryURI(f.Account, repo) + ":v1"
+
+	t.Run("docker login", func(t *testing.T) {
+		pass := harness.ECRGetLoginPassword(t, f.AWS)
+		out, err := docker.RunStdin(dockerCmdTimeout, pass,
+			"login", host, "-u", "AWS", "--password-stdin")
+		require.NoErrorf(t, err, "docker login: %s", out)
+		assert.Contains(t, out, "Login Succeeded")
+	})
+
+	t.Run("docker build + push", func(t *testing.T) {
+		buildDir := filepath.Join(f.TmpDir, "build")
+		require.NoError(t, os.MkdirAll(buildDir, 0o700))
+		ctxDir := buildScratchContext(t, buildDir)
+
+		out, err := docker.Run(dockerCmdTimeout, "build", "-t", ref, ctxDir)
+		require.NoErrorf(t, err, "docker build: %s", out)
+
+		out, err = docker.Run(dockerCmdTimeout, "push", ref)
+		require.NoErrorf(t, err, "docker push: %s", out)
+	})
+
+	t.Run("DescribeImages shows the pushed tag", func(t *testing.T) {
+		require.True(t, harness.ECRWaitImageTag(t, f.AWS, repo, "v1", 30*time.Second),
+			"pushed tag v1 not visible via DescribeImages")
+	})
+
+	t.Run("docker pull round-trips from a clean daemon", func(t *testing.T) {
+		out, err := docker.Run(dockerCmdTimeout, "rmi", "-f", ref)
+		require.NoErrorf(t, err, "docker rmi: %s", out)
+
+		out, err = docker.Run(dockerCmdTimeout, "pull", ref)
+		require.NoErrorf(t, err, "docker pull: %s", out)
+	})
+}

tests/e2e/ecr/ecr_test.go

@@ -0,0 +1,106 @@
+//go:build e2e
+
+package ecr
+
+import (
+	"fmt"
+	"time"
+
+	"testing"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/service/ecr"
+	"github.com/mulgadc/spinifex/tests/e2e/harness"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// uniqueRepo returns a collision-resistant repo name scoped to a test.
+func uniqueRepo(prefix string) string {
+	return fmt.Sprintf("e2e/%s-%d", prefix, time.Now().UnixNano())
+}
+
+// TestECRControlPlane round-trips the ECR control-plane surface: repository
+// CRUD, repository policy, and lifecycle policy. Zero VMs, pure SDK.
+func TestECRControlPlane(t *testing.T) {
+	f := requireECRFixture(t)
+	c := f.AWS
+	repo := uniqueRepo("ctl")
+	harness.CreateECRRepository(t, c, repo)
+
+	t.Run("DescribeRepositories lists the new repo", func(t *testing.T) {
+		out, err := c.ECR.DescribeRepositories(&ecr.DescribeRepositoriesInput{
+			RepositoryNames: []*string{aws.String(repo)},
+		})
+		require.NoError(t, err)
+		require.Len(t, out.Repositories, 1)
+		assert.Equal(t, repo, aws.StringValue(out.Repositories[0].RepositoryName))
+		assert.Contains(t, aws.StringValue(out.Repositories[0].RepositoryUri), repo)
+	})
+
+	t.Run("repository policy round-trips", func(t *testing.T) {
+		policy := `{"Version":"2012-10-17","Statement":[{"Sid":"allow-pull","Effect":"Allow","Principal":"*","Action":["ecr:GetDownloadUrlForLayer","ecr:BatchGetImage"]}]}`
+		_, err := c.ECR.SetRepositoryPolicy(&ecr.SetRepositoryPolicyInput{
+			RepositoryName: aws.String(repo),
+			PolicyText:     aws.String(policy),
+		})
+		require.NoError(t, err)
+
+		got, err := c.ECR.GetRepositoryPolicy(&ecr.GetRepositoryPolicyInput{
+			RepositoryName: aws.String(repo),
+		})
+		require.NoError(t, err)
+		assert.JSONEq(t, policy, aws.StringValue(got.PolicyText))
+	})
+
+	t.Run("lifecycle policy round-trips", func(t *testing.T) {
+		policy := `{"rules":[{"rulePriority":1,"selection":{"tagStatus":"untagged","countType":"sinceImagePushed","countUnit":"days","countNumber":14},"action":{"type":"expire"}}]}`
+		_, err := c.ECR.PutLifecyclePolicy(&ecr.PutLifecyclePolicyInput{
+			RepositoryName:      aws.String(repo),
+			LifecyclePolicyText: aws.String(policy),
+		})
+		require.NoError(t, err)
+
+		got, err := c.ECR.GetLifecyclePolicy(&ecr.GetLifecyclePolicyInput{
+			RepositoryName: aws.String(repo),
+		})
+		require.NoError(t, err)
+		assert.JSONEq(t, policy, aws.StringValue(got.LifecyclePolicyText))
+	})
+
+	t.Run("DeleteRepository removes it", func(t *testing.T) {
+		_, err := c.ECR.DeleteRepository(&ecr.DeleteRepositoryInput{
+			RepositoryName: aws.String(repo),
+			Force:          aws.Bool(true),
+		})
+		require.NoError(t, err)
+
+		_, err = c.ECR.DescribeRepositories(&ecr.DescribeRepositoriesInput{
+			RepositoryNames: []*string{aws.String(repo)},
+		})
+		require.Error(t, err, "deleted repo must not be describable")
+		assert.Contains(t, err.Error(), "RepositoryNotFoundException")
+	})
+}
+
+// TestECRGetLoginPassword confirms GetAuthorizationToken returns a usable
+// AWS:<jwt> credential — the input to docker login.
+func TestECRGetLoginPassword(t *testing.T) {
+	f := requireECRFixture(t)
+	pass := harness.ECRGetLoginPassword(t, f.AWS)
+	assert.NotEmpty(t, pass)
+	// JWT shape: three dot-separated segments.
+	assert.Len(t, splitDots(pass), 3, "authorization password should be a JWT")
+}
+
+func splitDots(s string) []string {
+	var out []string
+	start := 0
+	for i := 0; i < len(s); i++ {
+		if s[i] == '.' {
+			out = append(out, s[start:i])
+			start = i + 1
+		}
+	}
+	return append(out, s[start:])
+}

tests/e2e/ecr/main_test.go

@@ -0,0 +1,71 @@
+//go:build e2e
+
+// Package ecr is the standalone ECR E2E suite: control-plane CRUD over the
+// awsgw ECR surface plus the OCI data plane (docker/crane/skopeo push-pull)
+// against a running cluster. The data-plane subtests skip when the client
+// binary is missing or the registry host does not resolve, so the suite still
+// runs its control-plane leg on a bare runner.
+package ecr
+
+import (
+	"os"
+	"sync"
+	"testing"
+
+	"github.com/mulgadc/spinifex/tests/e2e/harness"
+)
+
+// Package-scoped singleton fixture, built lazily by requireECRFixture and torn
+// down by TestMain after the run.
+var (
+	pkgFixOnce sync.Once
+	pkgFix     *Fixture
+	pkgFixErr  error
+)
+
+// Fixture carries per-process state shared across every Test* in this package.
+type Fixture struct {
+	Env     *harness.Env
+	AWS     *harness.AWSClient
+	Account string
+	TmpDir  string
+}
+
+func TestMain(m *testing.M) {
+	code := m.Run()
+	if pkgFix != nil && pkgFix.TmpDir != "" {
+		_ = os.RemoveAll(pkgFix.TmpDir)
+	}
+	os.Exit(code)
+}
+
+// requireECRFixture returns the package-scoped Fixture, building it on first
+// call. Skips the calling test if SPINIFEX_E2E is unset.
+func requireECRFixture(t *testing.T) *Fixture {
+	t.Helper()
+	pkgFixOnce.Do(func() {
+		if os.Getenv("SPINIFEX_E2E") == "" {
+			return
+		}
+		env := harness.LoadEnv(t)
+		awsCli := harness.NewAWSClient(t, env)
+		tmpDir, terr := os.MkdirTemp("", "ecr-pkgfix-*")
+		if terr != nil {
+			pkgFixErr = terr
+			return
+		}
+		pkgFix = &Fixture{
+			Env:     env,
+			AWS:     awsCli,
+			Account: harness.IAMAccountID(t, awsCli),
+			TmpDir:  tmpDir,
+		}
+	})
+	if pkgFixErr != nil {
+		t.Fatalf("ecr fixture init failed: %v", pkgFixErr)
+	}
+	if pkgFix == nil {
+		t.Skip("ecr fixture unavailable (SPINIFEX_E2E unset)")
+	}
+	return pkgFix
+}

tests/e2e/harness/aws.go

@@ -16,6 +16,7 @@ import (
 	"github.com/aws/aws-sdk-go/service/acm"
 	"github.com/aws/aws-sdk-go/service/ec2"
 	"github.com/aws/aws-sdk-go/service/ec2/ec2iface"
+	"github.com/aws/aws-sdk-go/service/ecr"
 	"github.com/aws/aws-sdk-go/service/eks"
 	"github.com/aws/aws-sdk-go/service/elbv2"
 	"github.com/aws/aws-sdk-go/service/iam"
@@ -35,6 +36,7 @@ type AWSClient struct {
 	STS     *sts.STS
 	EKS     *eks.EKS
 	ACM     *acm.ACM
+	ECR     *ecr.ECR
 }
 
 // NewAWSClient builds clients pointed at the spinifex gateway using the
@@ -131,6 +133,7 @@ func newAWSClient(t *testing.T, env *Env, accessKey, secretKey, sessionToken str
 		STS:     sts.New(sess),
 		EKS:     eks.New(sess),
 		ACM:     acm.New(sess),
+		ECR:     ecr.New(sess),
 	}
 }
 
@@ -145,5 +148,6 @@ func (c *AWSClient) IgnoreCertErrors() {
 	c.IAM.Config.HTTPClient = hc
 	c.STS.Config.HTTPClient = hc
 	c.ACM.Config.HTTPClient = hc
+	c.ECR.Config.HTTPClient = hc
 	_ = (*x509.CertPool)(nil) // silence unused-import on hardened paths
 }

tests/e2e/harness/ecr.go

@@ -0,0 +1,112 @@
+//go:build e2e
+
+package harness
+
+import (
+	"encoding/base64"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/service/ecr"
+	"github.com/stretchr/testify/require"
+)
+
+// ECRRegistrySuffix is the internal DNS suffix the registry host is built from.
+// Defaults to the gateway's DefaultAWSInternalSuffix; override per cluster with
+// SPINIFEX_ECR_SUFFIX.
+func ECRRegistrySuffix() string { return getenv("SPINIFEX_ECR_SUFFIX", "mulga.internal") }
+
+// ECRRegistryRegion is the region label in the registry host. Mirrors the AWS
+// client region.
+func ECRRegistryRegion() string { return getenv("SPINIFEX_AWS_REGION", "ap-southeast-2") }
+
+// ECRRegistryPort is the awsgw port appended to the registry host for OCI
+// clients. Empty by default (production parity, :443); set SPINIFEX_ECR_PORT on
+// dev clusters where awsgw binds a non-standard port (e.g. 9999).
+func ECRRegistryPort() string { return getenv("SPINIFEX_ECR_PORT", "") }
+
+// ECRRegistryHost returns {account}.dkr.ecr.{region}.{suffix}[:port] — the OCI
+// registry endpoint docker/crane/skopeo address. The bare host must resolve to
+// the awsgw bind IP (DNS or /etc/hosts) for the data-plane subtests to run.
+func ECRRegistryHost(account string) string {
+	host := account + ".dkr.ecr." + ECRRegistryRegion() + "." + ECRRegistrySuffix()
+	if p := ECRRegistryPort(); p != "" {
+		host += ":" + p
+	}
+	return host
+}
+
+// ECRRepositoryURI returns the full push/pull reference for repo under account.
+func ECRRepositoryURI(account, repo string) string {
+	return ECRRegistryHost(account) + "/" + repo
+}
+
+// CreateECRRepository creates repo (idempotent: an existing repo is treated as
+// success) and registers best-effort teardown.
+func CreateECRRepository(t *testing.T, c *AWSClient, repo string) {
+	t.Helper()
+	_, err := c.ECR.CreateRepository(&ecr.CreateRepositoryInput{
+		RepositoryName: aws.String(repo),
+	})
+	if err != nil && !strings.Contains(err.Error(), "RepositoryAlreadyExistsException") {
+		require.NoError(t, err)
+	}
+	t.Cleanup(func() { DeleteECRRepositoryBestEffort(c, repo) })
+}
+
+// DeleteECRRepositoryBestEffort force-deletes repo and all its images, swallowing
+// errors — used for cleanup.
+func DeleteECRRepositoryBestEffort(c *AWSClient, repo string) {
+	_, _ = c.ECR.DeleteRepository(&ecr.DeleteRepositoryInput{
+		RepositoryName: aws.String(repo),
+		Force:          aws.Bool(true),
+	})
+}
+
+// ECRGetLoginPassword returns the decoded registry password (the JWT half of the
+// authorization token) for docker login -u AWS --password-stdin.
+func ECRGetLoginPassword(t *testing.T, c *AWSClient) string {
+	t.Helper()
+	out, err := c.ECR.GetAuthorizationToken(&ecr.GetAuthorizationTokenInput{})
+	require.NoError(t, err)
+	require.NotEmpty(t, out.AuthorizationData, "no authorization data returned")
+	raw, err := base64.StdEncoding.DecodeString(aws.StringValue(out.AuthorizationData[0].AuthorizationToken))
+	require.NoError(t, err)
+	user, pass, found := strings.Cut(string(raw), ":")
+	require.True(t, found && user == "AWS", "unexpected authorization token shape")
+	return pass
+}
+
+// ECRDescribeImageTags returns the set of image tags currently in repo.
+func ECRDescribeImageTags(t *testing.T, c *AWSClient, repo string) []string {
+	t.Helper()
+	out, err := c.ECR.DescribeImages(&ecr.DescribeImagesInput{
+		RepositoryName: aws.String(repo),
+	})
+	require.NoError(t, err)
+	var tags []string
+	for _, img := range out.ImageDetails {
+		for _, tag := range img.ImageTags {
+			tags = append(tags, aws.StringValue(tag))
+		}
+	}
+	return tags
+}
+
+// ECRWaitImageTag polls DescribeImages until tag appears in repo or the deadline
+// passes.
+func ECRWaitImageTag(t *testing.T, c *AWSClient, repo, tag string, timeout time.Duration) bool {
+	t.Helper()
+	deadline := time.Now().Add(timeout)
+	for time.Now().Before(deadline) {
+		for _, got := range ECRDescribeImageTags(t, c, repo) {
+			if got == tag {
+				return true
+			}
+		}
+		time.Sleep(500 * time.Millisecond)
+	}
+	return false
+}

tests/e2e/harness/registry.go

@@ -0,0 +1,115 @@
+//go:build e2e
+
+package harness
+
+import (
+	"bytes"
+	"net"
+	"os"
+	"os/exec"
+	"strings"
+	"testing"
+	"time"
+)
+
+// ExternalCLI wraps an OCI client binary (docker/crane/skopeo) for the ECR
+// data-plane subtests. Env is appended to the process environment (e.g.
+// DOCKER_CONFIG to isolate the auth file).
+type ExternalCLI struct {
+	Name string
+	Bin  string
+	Env  []string
+}
+
+// newExternalCLI locates bin on PATH (overridable via <NAME>_BIN, upper-cased)
+// and SKIPS the calling test when absent — the data-plane subtests are optional
+// on runners without the client installed.
+func newExternalCLI(t *testing.T, name string, env []string) *ExternalCLI {
+	t.Helper()
+	bin := os.Getenv(strings.ToUpper(name) + "_BIN")
+	if bin == "" {
+		bin = name
+	}
+	resolved, err := exec.LookPath(bin)
+	if err != nil {
+		t.Skipf("%s not found on PATH (%v); skipping data-plane subtest", name, err)
+	}
+	return &ExternalCLI{Name: name, Bin: resolved, Env: env}
+}
+
+// NewDocker / NewCrane / NewSkopeo build wrappers for each client, skipping the
+// test if the binary is missing. dockerConfig isolates the docker auth file so
+// crane/skopeo can reuse it via DOCKER_CONFIG without touching ~/.docker.
+func NewDocker(t *testing.T, dockerConfig string) *ExternalCLI {
+	return newExternalCLI(t, "docker", []string{"DOCKER_CONFIG=" + dockerConfig})
+}
+
+func NewCrane(t *testing.T, dockerConfig string) *ExternalCLI {
+	return newExternalCLI(t, "crane", []string{"DOCKER_CONFIG=" + dockerConfig})
+}
+
+func NewSkopeo(t *testing.T, dockerConfig string) *ExternalCLI {
+	return newExternalCLI(t, "skopeo", []string{"REGISTRY_AUTH_FILE=" + dockerConfig + "/config.json"})
+}
+
+// Run executes the client with args and returns combined output. Errors are
+// returned (not fatal) so callers can assert on output or poll.
+func (c *ExternalCLI) Run(timeout time.Duration, args ...string) (string, error) {
+	cmd := exec.Command(c.Bin, args...) //nolint:gosec // bin is LookPath-resolved, args test-controlled
+	cmd.Env = append(os.Environ(), c.Env...)
+	var buf bytes.Buffer
+	cmd.Stdout = &buf
+	cmd.Stderr = &buf
+	if err := cmd.Start(); err != nil {
+		return "", err
+	}
+	done := make(chan error, 1)
+	go func() { done <- cmd.Wait() }()
+	select {
+	case err := <-done:
+		return buf.String(), err
+	case <-time.After(timeout):
+		_ = cmd.Process.Kill()
+		<-done
+		return buf.String(), &timeoutError{}
+	}
+}
+
+// RunStdin is Run with data piped to the process stdin (docker login
+// --password-stdin).
+func (c *ExternalCLI) RunStdin(timeout time.Duration, stdin string, args ...string) (string, error) {
+	cmd := exec.Command(c.Bin, args...) //nolint:gosec // bin is LookPath-resolved, args test-controlled
+	cmd.Env = append(os.Environ(), c.Env...)
+	cmd.Stdin = strings.NewReader(stdin)
+	var buf bytes.Buffer
+	cmd.Stdout = &buf
+	cmd.Stderr = &buf
+	if err := cmd.Start(); err != nil {
+		return "", err
+	}
+	done := make(chan error, 1)
+	go func() { done <- cmd.Wait() }()
+	select {
+	case err := <-done:
+		return buf.String(), err
+	case <-time.After(timeout):
+		_ = cmd.Process.Kill()
+		<-done
+		return buf.String(), &timeoutError{}
+	}
+}
+
+// RequireRegistryResolves SKIPS the calling test unless host resolves to an
+// address. host may carry a :port suffix (stripped before lookup). The
+// data-plane subtests need {account}.dkr.ecr.{region}.{suffix} to reach the
+// awsgw bind IP (Route53 in prod, /etc/hosts in dev); without it the client
+// cannot connect.
+func RequireRegistryResolves(t *testing.T, host string) {
+	t.Helper()
+	if h, _, err := net.SplitHostPort(host); err == nil {
+		host = h
+	}
+	if _, err := net.LookupHost(host); err != nil {
+		t.Skipf("registry host %q does not resolve (%v); seed /etc/hosts or DNS", host, err)
+	}
+}