feat: EC2 capacity reservations — launch into reservation (#397)

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`

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/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