feat: ECS P2 hardening — taskdef validation, logging signal, execution role, rolling deployments (#472)

- RegisterTaskDefinition rejects container secrets[] (InvalidParameterException)
  and warns on unsupported log drivers instead of silently dropping them.
- Capture logConfiguration; only host-side json-file is collected, other drivers
  warn at register and reach the agent for parity. Document json-file retrieval.
- Plumb executionRoleArn taskdef -> assign -> agent; the agent assumes it to
  authorize ECR pulls, falling back to the instance role when unset.
- Add deployment objects with a health-gated rolling update honoring
  deploymentConfiguration minimumHealthyPercent/maximumPercent, plus a deployment
  circuit breaker with optional rollback to the last-good task definition.

cmd/ecs-agent/credendpoint.go

@@ -192,6 +192,45 @@ func (c *credEndpoint) fetch(ctx context.Context, credID string) (stsauth.Creden
 	return creds, roleARN, nil
 }
 
+// AssumeProvider returns a CredentialsProvider that mints credentials by assuming
+// roleARN over the gateway — the execution-role path for ECR image pulls. It
+// caches the assumed set and re-assumes within the refresh margin. session scopes
+// the STS session name.
+func (c *credEndpoint) AssumeProvider(roleARN, session string) credentials.CredentialsProvider {
+	return &assumeProvider{ep: c, roleARN: roleARN, session: session}
+}
+
+// assumeProvider adapts a credEndpoint's AssumeRole path to the agent's
+// CredentialsProvider interface, caching the assumed credentials.
+type assumeProvider struct {
+	ep      *credEndpoint
+	roleARN string
+	session string
+
+	mu     sync.Mutex
+	cached stsauth.Credentials
+}
+
+var _ credentials.CredentialsProvider = (*assumeProvider)(nil)
+
+func (p *assumeProvider) Retrieve(ctx context.Context) (credentials.Credentials, error) {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	if !credValid(p.cached, credRefreshMargin) {
+		creds, err := p.ep.assume(ctx, p.roleARN, p.session)
+		if err != nil {
+			return credentials.Credentials{}, err
+		}
+		p.cached = creds
+	}
+	return credentials.Credentials{
+		AccessKeyID:     p.cached.AccessKeyID,
+		SecretAccessKey: p.cached.SecretAccessKey,
+		SessionToken:    p.cached.SessionToken,
+		Expiration:      p.cached.Expiration,
+	}, nil
+}
+
 // assumeOverGateway mints credentials for roleARN by SigV4-signing AssumeRole
 // against the gateway with the agent's instance credentials.
 func (c *credEndpoint) assumeOverGateway(ctx context.Context, roleARN, sessionName string) (stsauth.Credentials, error) {

cmd/ecs-agent/taskrun.go

@@ -65,10 +65,11 @@ func (a *Agent) runTask(ctx context.Context, as *bus.Assign) {
 	}
 
 	credID := a.registerTaskCreds(as)
+	resolver := a.pullResolver(as)
 
 	statuses := make([]bus.ContainerStatus, 0, len(as.Containers))
 	for _, c := range as.Containers {
-		if _, err := a.puller.Pull(ctx, ctrruntime.PullSpec{Ref: c.Image}, a.resolver); err != nil {
+		if _, err := a.puller.Pull(ctx, ctrruntime.PullSpec{Ref: c.Image}, resolver); err != nil {
 			slog.Error("ecs-agent: pull failed", "task", as.TaskID, "image", c.Image, "err", err)
 			a.teardownTaskNetns(as)
 			a.reportTaskState(as, bus.TaskStatusStopped, "image pull failed: "+err.Error(), statuses)
@@ -137,6 +138,18 @@ func (a *Agent) teardownTaskNetns(as *bus.Assign) {
 	}
 }
 
+// pullResolver returns the ECR resolver used for a task's image pulls. When the
+// assign carries an execution role, it authorizes pulls as that role (assumed
+// over the gateway); an empty role or a nil credential endpoint (unit tests)
+// falls back to the instance-role resolver, so a pull is never worse off.
+func (a *Agent) pullResolver(as *bus.Assign) ctrruntime.Resolver {
+	if as.ExecutionRoleARN == "" || a.cred == nil {
+		return a.resolver
+	}
+	prov := a.cred.AssumeProvider(as.ExecutionRoleARN, sessionName(as.TaskID))
+	return newLazyECRResolver(prov, a.cfg.Region, a.cfg.GatewayURL, a.cfg.GatewayCA)
+}
+
 // registerTaskCreds registers the task's credID -> role mapping with the
 // credential endpoint and returns the credID (empty when the task has no role).
 func (a *Agent) registerTaskCreds(as *bus.Assign) string {

cmd/ecs-agent/taskrun_test.go

@@ -227,3 +227,32 @@ func TestContainerID(t *testing.T) {
 		t.Errorf("containerID = %q, want t-1-web", got)
 	}
 }
+
+// sentinelResolver is an identity-comparable Resolver for pullResolver tests.
+type sentinelResolver struct{}
+
+func (sentinelResolver) Authorize(context.Context, string) (string, string, string, error) {
+	return "", "", "", nil
+}
+
+// TestPullResolver covers siv-459: an assign with no execution role (or no
+// credential endpoint) uses the instance resolver; with both, a distinct
+// exec-role resolver is built.
+func TestPullResolver(t *testing.T) {
+	fake := sentinelResolver{}
+	a := newAgent(config{Region: "us-east-1"}, testIdentity(), &fakeCP{}, nil, nil, fake)
+
+	if got := a.pullResolver(&bus.Assign{TaskID: "t1"}); got != fake {
+		t.Fatalf("no execution role: want instance resolver fallback")
+	}
+	execAssign := &bus.Assign{TaskID: "t1", ExecutionRoleARN: "arn:aws:iam::1:role/e"}
+	if got := a.pullResolver(execAssign); got != fake {
+		t.Fatalf("nil cred endpoint: want instance resolver fallback")
+	}
+
+	a.cred = &credEndpoint{}
+	got := a.pullResolver(execAssign)
+	if got == nil || got == fake {
+		t.Fatalf("execution role + cred endpoint: want distinct exec-role resolver, got %v", got)
+	}
+}

docs/containers/ecs/README.md

@@ -62,11 +62,32 @@ A **task definition** describes one or more containers (image, CPU/memory, ports
 ECS v1 is deliberately minimal. Keep these in mind:
 
 - **No service discovery.** `serviceRegistries` / Cloud Map integration is not implemented — reach a service through its load balancer, not a DNS name.
-- **`awslogs` log driver is discarded.** Container stdout/stderr is not shipped to CloudWatch Logs; the log configuration is accepted but dropped.
-- **No rolling deployments.** The reconciler keeps the desired count running but ignores `deploymentConfiguration` (`minimumHealthyPercent` / `maximumPercent`); an `UpdateService` to a new revision replaces tasks without a health-gated rollout or circuit-breaker.
+- **Only the `json-file` log driver is collected.** Container stdout/stderr is captured host-side on the container instance (see [Logging](#logging)); it is not shipped to CloudWatch Logs. Any other driver (e.g. `awslogs`) is accepted for parity but its logs are discarded — `RegisterTaskDefinition` logs a warning naming the container so the drop is not silent.
 - **No capacity providers or managed scaling.** Capacity is the static total of your registered instances; there is no scale-out/in or ASG binding — you manage the instance count.
-- **No distinct execution role.** ECR pulls use the instance role directly; `executionRoleArn` is not honoured separately.
-- **`secrets[]` are silently dropped**, and tags set via `TagResource` are not persisted. Health-check settings beyond the target group default are not forwarded.
+- **`secrets[]` are rejected.** A task definition that declares container `secrets[]` fails `RegisterTaskDefinition` with `InvalidParameterException` rather than running without the secrets it expects. Tags set via `TagResource` are not persisted, and health-check settings beyond the target group default are not forwarded.
+
+### Logging
+
+Spinifex honours the host-side **`json-file`** log driver (the containerd default). A container's stdout/stderr is written on its container instance and retrievable there — no CloudWatch Logs path exists.
+
+To read a task's logs, find the container instance running it (`aws ecs describe-tasks` → `containerInstanceArn` → the EC2 instance), then on that host inspect the containerd task output. Containers are named `{taskId}-{containerName}` and carry `mulga.ecs.*` labels:
+
+```bash
+# On the container instance:
+ctr -n default containers ls                 # find {taskId}-{containerName}
+ctr -n default tasks ls
+journalctl -u containerd | grep <taskId>     # container stdout/stderr via the host journal
+```
+
+A task definition may still declare `awslogs` (or any other driver) for AWS compatibility — Spinifex accepts it, warns at registration that the driver is not implemented, and falls back to the host-side `json-file` behaviour.
+
+### Deployments
+
+An `UpdateService` to a new task-definition revision performs a **health-gated rolling update** honouring `deploymentConfiguration`: `minimumHealthyPercent` keeps that fraction of the desired count running while `maximumPercent` bounds how many extra tasks launch during the roll. Enabling the **deployment circuit breaker** fails a rollout whose tasks repeatedly fail to start, and — with `rollback` set — automatically reverts to the last-good task definition.
+
+### Execution role
+
+If a task definition sets `executionRoleArn`, the agent assumes that role to authorise ECR image pulls (instead of the container-instance role). When it is unset, pulls fall back to the instance role.
 
 ## Prerequisites
 
@@ -267,4 +288,4 @@ tofu apply -var 'gateway_url=https://<host-lan-ip>:9999'
 
 **Container cannot assume its task role.** Confirm the task definition sets `taskRoleArn` and the role is trusted by `ecs-tasks.amazonaws.com`. The agent injects `AWS_CONTAINER_CREDENTIALS_RELATIVE_URI`; a container that overrides this variable, or an SDK too old to honour it, will not pick up the credentials.
 
-**Expected a feature that is missing.** ECS v1 omits service discovery, `awslogs` shipping, rolling deployments, capacity providers, and a distinct execution role. See the Limitations in the Overview.
+**Expected a feature that is missing.** ECS v1 omits service discovery, CloudWatch Logs (`awslogs`) shipping, and capacity providers / managed scaling. See the Limitations in the Overview.

spinifex/handlers/ecs/bus/messages.go

@@ -65,6 +65,9 @@ type AssignContainer struct {
 	Command      []string          `json:"command,omitempty"`
 	Environment  map[string]string `json:"environment,omitempty"`
 	PortMappings []PortMapping     `json:"portMappings,omitempty"`
+	// LogDriver is the requested container log driver; only json-file is honored
+	// (host-side). Any other value means logs are discarded (warned at register).
+	LogDriver string `json:"logDriver,omitempty"`
 }
 
 // Assign is published on AssignSubject when the scheduler places a task on this
@@ -89,9 +92,13 @@ type Assign struct {
 	// TaskRoleARN is the task's IAM role (from the taskdef); when set, the agent
 	// serves its credentials at 169.254.170.2 under CredID, defaulting CredID to
 	// the taskID when the scheduler leaves it empty.
-	CredID      string    `json:"credId,omitempty"`
-	TaskRoleARN string    `json:"taskRoleArn,omitempty"`
-	AssignedAt  time.Time `json:"assignedAt"`
+	CredID      string `json:"credId,omitempty"`
+	TaskRoleARN string `json:"taskRoleArn,omitempty"`
+	// ExecutionRoleARN is the task's execution role (from the taskdef); when set,
+	// the agent assumes it to authorize ECR image pulls instead of the container-
+	// instance role. Empty falls back to the instance role.
+	ExecutionRoleARN string    `json:"executionRoleArn,omitempty"`
+	AssignedAt       time.Time `json:"assignedAt"`
 }
 
 // ContainerStatus is a single container's reported lifecycle state.

spinifex/handlers/ecs/conv.go

@@ -82,6 +82,15 @@ func containerDefsFromAWS(in []*ecs.ContainerDefinition) []ContainerDef {
 				Protocol:      aws.StringValue(p.Protocol),
 			})
 		}
+		if lc := c.LogConfiguration; lc != nil {
+			def.LogDriver = aws.StringValue(lc.LogDriver)
+			for k, v := range lc.Options {
+				if def.LogOptions == nil {
+					def.LogOptions = map[string]string{}
+				}
+				def.LogOptions[k] = aws.StringValue(v)
+			}
+		}
 		out = append(out, def)
 	}
 	return out
@@ -115,6 +124,16 @@ func (c ContainerDef) toAWS() *ecs.ContainerDefinition {
 		}
 		cd.PortMappings = append(cd.PortMappings, pm)
 	}
+	if c.LogDriver != "" {
+		lc := &ecs.LogConfiguration{LogDriver: aws.String(c.LogDriver)}
+		if len(c.LogOptions) > 0 {
+			lc.Options = map[string]*string{}
+			for k, v := range c.LogOptions {
+				lc.Options[k] = aws.String(v)
+			}
+		}
+		cd.LogConfiguration = lc
+	}
 	return cd
 }
 
@@ -129,5 +148,6 @@ func (c ContainerDef) toAssignContainer() bus.AssignContainer {
 		Command:      c.Command,
 		Environment:  c.Environment,
 		PortMappings: c.PortMappings,
+		LogDriver:    c.LogDriver,
 	}
 }

spinifex/handlers/ecs/deployments.go

@@ -0,0 +1,240 @@
+package handlers_ecs
+
+import (
+	"log/slog"
+	"slices"
+	"strings"
+	"time"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/service/ecs"
+	"github.com/google/uuid"
+)
+
+// deploymentStartedByPrefix is the AWS StartedBy prefix a service stamps on its
+// tasks; the suffix is the deployment ID, so the reconciler maps a task back to
+// the deployment that launched it.
+const deploymentStartedByPrefix = "ecs-svc/"
+
+// deploymentSupersededReason is the stop reason for an old-deployment task drained
+// during a rolling update.
+const deploymentSupersededReason = "Superseded by new deployment"
+
+func deploymentStartedBy(id string) string { return deploymentStartedByPrefix + id }
+
+func deploymentIDFromStartedBy(startedBy string) string {
+	id, _ := strings.CutPrefix(startedBy, deploymentStartedByPrefix)
+	return id
+}
+
+// ceilPercent returns count*percent/100 rounded up — used for minimumHealthyPercent
+// (AWS rounds the healthy floor up).
+func ceilPercent(count, percent int) int {
+	if count <= 0 {
+		return 0
+	}
+	return (count*percent + 99) / 100
+}
+
+// applyDeploymentConfig sets the service's deploymentConfiguration, defaulting to
+// the AWS REPLICA defaults (100% healthy / 200% max) when a field is omitted.
+func applyDeploymentConfig(rec *ServiceRecord, dc *ecs.DeploymentConfiguration) {
+	rec.MinimumHealthyPercent = defaultMinimumHealthyPercent
+	rec.MaximumPercent = defaultMaximumPercent
+	if dc == nil {
+		return
+	}
+	if dc.MinimumHealthyPercent != nil {
+		rec.MinimumHealthyPercent = int(*dc.MinimumHealthyPercent)
+	}
+	if dc.MaximumPercent != nil {
+		rec.MaximumPercent = int(*dc.MaximumPercent)
+	}
+	if cb := dc.DeploymentCircuitBreaker; cb != nil {
+		rec.CircuitBreakerEnable = aws.BoolValue(cb.Enable)
+		rec.CircuitBreakerRollback = aws.BoolValue(cb.Rollback)
+	}
+}
+
+// normalizeDeploymentConfig backfills the deploymentConfiguration defaults for a
+// legacy service record written before rolling deployments existed.
+func (r *ServiceRecord) normalizeDeploymentConfig() {
+	if r.MinimumHealthyPercent == 0 {
+		r.MinimumHealthyPercent = defaultMinimumHealthyPercent
+	}
+	if r.MaximumPercent == 0 {
+		r.MaximumPercent = defaultMaximumPercent
+	}
+}
+
+// newPrimaryDeployment builds an IN_PROGRESS PRIMARY deployment for a task def.
+func newPrimaryDeployment(id string, td *TaskDefRecord, desired int) Deployment {
+	now := time.Now().UTC()
+	return Deployment{
+		ID:              id,
+		Status:          DeploymentStatusPrimary,
+		TaskDefARN:      td.ARN,
+		TaskDefFamily:   td.Family,
+		TaskDefRevision: td.Revision,
+		DesiredCount:    desired,
+		RolloutState:    RolloutStateInProgress,
+		CreatedAt:       now,
+		UpdatedAt:       now,
+	}
+}
+
+// ensurePrimaryDeployment synthesizes a PRIMARY deployment for a legacy service
+// record that predates deployment tracking, keyed by its existing DeploymentID so
+// its already-running tasks (StartedBy=ecs-svc/{DeploymentID}) map to it.
+func (r *ServiceRecord) ensurePrimaryDeployment() {
+	if r.primaryDeployment() != nil {
+		return
+	}
+	r.Deployments = append(r.Deployments, Deployment{
+		ID:              r.DeploymentID,
+		Status:          DeploymentStatusPrimary,
+		TaskDefARN:      r.TaskDefARN,
+		TaskDefFamily:   r.TaskDefFamily,
+		TaskDefRevision: r.TaskDefRevision,
+		DesiredCount:    r.DesiredCount,
+		RolloutState:    RolloutStateInProgress,
+		CreatedAt:       r.CreatedAt,
+		UpdatedAt:       time.Now().UTC(),
+	})
+	if r.LastGoodTaskDefARN == "" {
+		r.LastGoodTaskDefARN = r.TaskDefARN
+	}
+}
+
+// startDeployment demotes the current PRIMARY to ACTIVE (draining) and installs a
+// new PRIMARY for td, mirroring the taskdef onto the service record.
+func (r *ServiceRecord) startDeployment(td *TaskDefRecord) {
+	r.demotePrimary()
+	r.DeploymentID = uuid.NewString()
+	r.TaskDefFamily = td.Family
+	r.TaskDefRevision = td.Revision
+	r.TaskDefARN = td.ARN
+	r.NetworkMode = resolveNetworkMode(td)
+	r.Deployments = append(r.Deployments, newPrimaryDeployment(r.DeploymentID, td, r.DesiredCount))
+}
+
+func (r *ServiceRecord) demotePrimary() {
+	for i := range r.Deployments {
+		if r.Deployments[i].Status == DeploymentStatusPrimary {
+			r.Deployments[i].Status = DeploymentStatusActive
+			r.Deployments[i].UpdatedAt = time.Now().UTC()
+		}
+	}
+}
+
+// pruneCompletedDeployments drops the drained (non-PRIMARY) deployments once the
+// rollout has completed, leaving a single PRIMARY at steady state.
+func (r *ServiceRecord) pruneCompletedDeployments() {
+	r.Deployments = slices.DeleteFunc(r.Deployments, func(d Deployment) bool {
+		return d.Status != DeploymentStatusPrimary
+	})
+}
+
+// tallyDeployments zeroes then recomputes each deployment's RUNNING/PENDING counts
+// from the service's live tasks and returns the service-wide totals.
+func tallyDeployments(svc *ServiceRecord, tasks []TaskRecord) (running, pending int) {
+	byDep := make(map[string]*Deployment, len(svc.Deployments))
+	for i := range svc.Deployments {
+		svc.Deployments[i].RunningCount = 0
+		svc.Deployments[i].PendingCount = 0
+		byDep[svc.Deployments[i].ID] = &svc.Deployments[i]
+	}
+	for i := range tasks {
+		d := byDep[deploymentIDFromStartedBy(tasks[i].StartedBy)]
+		switch tasks[i].LastStatus {
+		case TaskStatusRunning:
+			running++
+			if d != nil {
+				d.RunningCount++
+			}
+		case TaskStatusPending:
+			pending++
+			if d != nil {
+				d.PendingCount++
+			}
+		}
+	}
+	return running, pending
+}
+
+// updateRolloutState advances the PRIMARY deployment's rollout state. It completes
+// (and prunes drained deployments, recording the taskdef as last-good) once the
+// primary is fully up and no old tasks remain; otherwise it stays IN_PROGRESS. A
+// FAILED deployment is left as-is (the circuit breaker owns that transition).
+func updateRolloutState(svc *ServiceRecord, primary *Deployment, desired int) {
+	primary.DesiredCount = desired
+	if primary.RolloutState == RolloutStateFailed {
+		return
+	}
+	remainingOld := 0
+	for i := range svc.Deployments {
+		if svc.Deployments[i].Status != DeploymentStatusPrimary {
+			remainingOld += svc.Deployments[i].RunningCount + svc.Deployments[i].PendingCount
+		}
+	}
+	if primary.RunningCount >= desired && remainingOld == 0 {
+		if primary.RolloutState != RolloutStateCompleted {
+			primary.RolloutState = RolloutStateCompleted
+			primary.RolloutReason = "ECS deployment completed."
+			primary.UpdatedAt = time.Now().UTC()
+			slog.Info("ECS deployment completed", "service", svc.Name, "deployment", primary.ID, "taskDef", primary.TaskDefARN)
+		}
+		svc.LastGoodTaskDefARN = primary.TaskDefARN
+		svc.pruneCompletedDeployments()
+		return
+	}
+	primary.RolloutState = RolloutStateInProgress
+}
+
+// tripCircuitBreaker fails a PRIMARY deployment whose task launches have failed
+// past the threshold when the breaker is enabled, optionally rolling back to the
+// last-good task definition. Returns true when it acted so the caller can persist
+// and defer further launches to the next tick.
+func tripCircuitBreaker(svc *ServiceRecord, primary *Deployment) bool {
+	if !svc.CircuitBreakerEnable || primary.RolloutState == RolloutStateFailed {
+		return false
+	}
+	if primary.FailedTasks < circuitBreakerFailureThreshold {
+		return false
+	}
+	primary.RolloutState = RolloutStateFailed
+	primary.RolloutReason = "ECS deployment circuit breaker: tasks failed to start."
+	primary.UpdatedAt = time.Now().UTC()
+	slog.Warn("ECS deployment circuit breaker tripped", "service", svc.Name,
+		"deployment", primary.ID, "failedTasks", primary.FailedTasks)
+
+	if !svc.CircuitBreakerRollback || svc.LastGoodTaskDefARN == "" || svc.LastGoodTaskDefARN == primary.TaskDefARN {
+		return true
+	}
+	svc.rollbackToLastGood()
+	return true
+}
+
+// rollbackToLastGood demotes the failed PRIMARY and starts a fresh PRIMARY from
+// the last-good task definition ARN.
+func (r *ServiceRecord) rollbackToLastGood() {
+	family, rev := parseTaskDefRef(r.LastGoodTaskDefARN)
+	r.demotePrimary()
+	now := time.Now().UTC()
+	r.DeploymentID = uuid.NewString()
+	r.Deployments = append(r.Deployments, Deployment{
+		ID:              r.DeploymentID,
+		Status:          DeploymentStatusPrimary,
+		TaskDefARN:      r.LastGoodTaskDefARN,
+		TaskDefFamily:   family,
+		TaskDefRevision: rev,
+		DesiredCount:    r.DesiredCount,
+		RolloutState:    RolloutStateInProgress,
+		RolloutReason:   "ECS deployment rolled back to last good task definition.",
+		CreatedAt:       now,
+		UpdatedAt:       now,
+	})
+	r.TaskDefARN = r.LastGoodTaskDefARN
+	r.TaskDefFamily = family
+	r.TaskDefRevision = rev
+}

spinifex/handlers/ecs/deployments_test.go

@@ -0,0 +1,225 @@
+package handlers_ecs
+
+import (
+	"testing"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/service/ecs"
+	"github.com/mulgadc/spinifex/spinifex/handlers/ecs/bus"
+	"github.com/nats-io/nats.go"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// reloadService re-reads a service record from KV so a test sees writes made by
+// the reconciler / failure accounting.
+func reloadService(t *testing.T, kv nats.KeyValue, cluster, name string) *ServiceRecord {
+	t.Helper()
+	var rec ServiceRecord
+	found, err := getJSON(kv, ServiceKey(cluster, name), &rec)
+	require.NoError(t, err)
+	require.True(t, found)
+	return &rec
+}
+
+// driveRunning reports RUNNING for every PENDING task of a service.
+func driveRunning(t *testing.T, svc *Service, kv nats.KeyValue, cluster, name string) {
+	t.Helper()
+	tasks, err := svc.listServiceTasks(kv, cluster, name)
+	require.NoError(t, err)
+	for i := range tasks {
+		if tasks[i].LastStatus != TaskStatusPending {
+			continue
+		}
+		require.NoError(t, svc.recordTaskState(&bus.TaskState{
+			AccountID: testAccountID, ClusterName: cluster, TaskID: tasks[i].TaskID,
+			LastStatus: TaskStatusRunning,
+		}))
+	}
+}
+
+// failPending reports STOPPED (never-RUNNING) for every PENDING task, simulating
+// tasks that fail to start — the deployment circuit-breaker signal.
+func failPending(t *testing.T, svc *Service, kv nats.KeyValue, cluster, name string) {
+	t.Helper()
+	tasks, err := svc.listServiceTasks(kv, cluster, name)
+	require.NoError(t, err)
+	for i := range tasks {
+		if tasks[i].LastStatus != TaskStatusPending {
+			continue
+		}
+		require.NoError(t, svc.recordTaskState(&bus.TaskState{
+			AccountID: testAccountID, ClusterName: cluster, TaskID: tasks[i].TaskID,
+			LastStatus: TaskStatusStopped, Reason: "image pull failed",
+		}))
+	}
+}
+
+func TestDeployment_CreateService_SeedsPrimary(t *testing.T) {
+	svc, _, kv := serviceTestRig(t)
+	out, err := svc.CreateService(&ecs.CreateServiceInput{
+		Cluster: aws.String("web"), ServiceName: aws.String("web"),
+		TaskDefinition: aws.String("app"), DesiredCount: aws.Int64(2),
+	}, testAccountID)
+	require.NoError(t, err)
+
+	require.Len(t, out.Service.Deployments, 1)
+	assert.Equal(t, DeploymentStatusPrimary, aws.StringValue(out.Service.Deployments[0].Status))
+	assert.Equal(t, RolloutStateInProgress, aws.StringValue(out.Service.Deployments[0].RolloutState))
+
+	rec := reloadService(t, kv, "web", "web")
+	require.NotNil(t, rec.primaryDeployment())
+	assert.Equal(t, defaultMinimumHealthyPercent, rec.MinimumHealthyPercent)
+	assert.Equal(t, defaultMaximumPercent, rec.MaximumPercent)
+	// Every launched task carries the primary deployment id in StartedBy.
+	tasks, err := svc.listServiceTasks(kv, "web", "web")
+	require.NoError(t, err)
+	require.Len(t, tasks, 2)
+	for i := range tasks {
+		assert.Equal(t, rec.DeploymentID, deploymentIDFromStartedBy(tasks[i].StartedBy))
+	}
+}
+
+func TestDeployment_RollingUpdate_ReplacesOldWithNew(t *testing.T) {
+	svc, _, kv := serviceTestRig(t)
+	_, err := svc.CreateService(&ecs.CreateServiceInput{
+		Cluster: aws.String("web"), ServiceName: aws.String("web"),
+		TaskDefinition: aws.String("app"), DesiredCount: aws.Int64(2),
+	}, testAccountID)
+	require.NoError(t, err)
+
+	// Complete the initial deployment.
+	driveRunning(t, svc, kv, "web", "web")
+	require.NoError(t, svc.reconcileService(kv, testAccountID, reloadService(t, kv, "web", "web")))
+	rec := reloadService(t, kv, "web", "web")
+	require.Len(t, rec.Deployments, 1)
+	assert.Equal(t, RolloutStateCompleted, rec.primaryDeployment().RolloutState)
+	firstDeployID := rec.DeploymentID
+
+	// New taskdef revision starts a rolling deployment.
+	registerTaskDef(t, svc, "app", 128, 256) // app:2
+	upd, err := svc.UpdateService(&ecs.UpdateServiceInput{
+		Cluster: aws.String("web"), Service: aws.String("web"),
+		TaskDefinition: aws.String("app:2"),
+	}, testAccountID)
+	require.NoError(t, err)
+	require.Len(t, upd.Service.Deployments, 2) // PRIMARY app:2 + ACTIVE app:1
+
+	rec = reloadService(t, kv, "web", "web")
+	assert.NotEqual(t, firstDeployID, rec.DeploymentID)
+	// maximumPercent=200 lets 2 new tasks launch alongside the 2 old running ones.
+	newTasks := 0
+	tasks, err := svc.listServiceTasks(kv, "web", "web")
+	require.NoError(t, err)
+	for i := range tasks {
+		if deploymentIDFromStartedBy(tasks[i].StartedBy) == rec.DeploymentID {
+			newTasks++
+		}
+	}
+	assert.Equal(t, 2, newTasks)
+
+	// New tasks come up; a reconcile drains the old ones and completes the rollout.
+	driveRunning(t, svc, kv, "web", "web")
+	require.NoError(t, svc.reconcileService(kv, testAccountID, reloadService(t, kv, "web", "web")))
+	rec = reloadService(t, kv, "web", "web")
+	require.Len(t, rec.Deployments, 1)
+	assert.Equal(t, RolloutStateCompleted, rec.primaryDeployment().RolloutState)
+	assert.Equal(t, rec.DeploymentID, rec.primaryDeployment().ID)
+
+	// Only new-deployment tasks remain live.
+	tasks, err = svc.listServiceTasks(kv, "web", "web")
+	require.NoError(t, err)
+	require.Len(t, tasks, 2)
+	for i := range tasks {
+		assert.Equal(t, rec.DeploymentID, deploymentIDFromStartedBy(tasks[i].StartedBy))
+	}
+}
+
+func TestDeployment_MinimumHealthyPercent_GatesDrain(t *testing.T) {
+	svc, _, kv := serviceTestRig(t)
+	// min=100, max=150: with desired=2 the rollout may run up to 3 tasks and must
+	// keep 2 healthy, so old tasks only drain as new ones become healthy.
+	_, err := svc.CreateService(&ecs.CreateServiceInput{
+		Cluster: aws.String("web"), ServiceName: aws.String("web"),
+		TaskDefinition: aws.String("app"), DesiredCount: aws.Int64(2),
+		DeploymentConfiguration: &ecs.DeploymentConfiguration{
+			MinimumHealthyPercent: aws.Int64(100), MaximumPercent: aws.Int64(150),
+		},
+	}, testAccountID)
+	require.NoError(t, err)
+	driveRunning(t, svc, kv, "web", "web")
+	require.NoError(t, svc.reconcileService(kv, testAccountID, reloadService(t, kv, "web", "web")))
+
+	registerTaskDef(t, svc, "app", 128, 256) // app:2
+	_, err = svc.UpdateService(&ecs.UpdateServiceInput{
+		Cluster: aws.String("web"), Service: aws.String("web"), TaskDefinition: aws.String("app:2"),
+	}, testAccountID)
+	require.NoError(t, err)
+
+	// Ceiling of 3 tasks total: 2 old running + 1 new pending. No old task drained
+	// yet because healthy running (2) must not drop below minCount (2).
+	tasks, err := svc.listServiceTasks(kv, "web", "web")
+	require.NoError(t, err)
+	assert.Len(t, tasks, 3)
+	rec := reloadService(t, kv, "web", "web")
+	assert.Equal(t, 2, rec.RunningCount)
+	assert.Equal(t, 1, rec.PendingCount)
+}
+
+func TestDeployment_CircuitBreaker_RollsBackToLastGood(t *testing.T) {
+	svc, _, kv := serviceTestRig(t)
+	_, err := svc.CreateService(&ecs.CreateServiceInput{
+		Cluster: aws.String("web"), ServiceName: aws.String("web"),
+		TaskDefinition: aws.String("app"), DesiredCount: aws.Int64(1),
+		DeploymentConfiguration: &ecs.DeploymentConfiguration{
+			DeploymentCircuitBreaker: &ecs.DeploymentCircuitBreaker{
+				Enable: aws.Bool(true), Rollback: aws.Bool(true),
+			},
+		},
+	}, testAccountID)
+	require.NoError(t, err)
+	driveRunning(t, svc, kv, "web", "web")
+	require.NoError(t, svc.reconcileService(kv, testAccountID, reloadService(t, kv, "web", "web")))
+	goodARN := reloadService(t, kv, "web", "web").LastGoodTaskDefARN
+	require.NotEmpty(t, goodARN)
+
+	// Roll out a revision whose tasks always fail to start.
+	registerTaskDef(t, svc, "app", 128, 256) // app:2
+	_, err = svc.UpdateService(&ecs.UpdateServiceInput{
+		Cluster: aws.String("web"), Service: aws.String("web"), TaskDefinition: aws.String("app:2"),
+	}, testAccountID)
+	require.NoError(t, err)
+
+	// Each cycle: fail the pending new task, then reconcile relaunches. After the
+	// failure threshold the breaker trips on the next reconcile and rolls back.
+	for range circuitBreakerFailureThreshold {
+		failPending(t, svc, kv, "web", "web")
+		require.NoError(t, svc.reconcileService(kv, testAccountID, reloadService(t, kv, "web", "web")))
+	}
+
+	rec := reloadService(t, kv, "web", "web")
+	primary := rec.primaryDeployment()
+	require.NotNil(t, primary)
+	assert.Equal(t, goodARN, primary.TaskDefARN)
+	assert.Equal(t, goodARN, rec.TaskDefARN)
+}
+
+func TestDeployment_LegacyServiceSynthesizesPrimary(t *testing.T) {
+	svc, _, kv := serviceTestRig(t)
+	// A record written before deployment tracking existed: no Deployments slice.
+	rec := &ServiceRecord{
+		Name: "legacy", ARN: ServiceARN(testRegion, testAccountID, "web", "legacy"),
+		Cluster: "web", TaskDefFamily: "app", TaskDefRevision: 1,
+		TaskDefARN:   TaskDefARN(testRegion, testAccountID, "app", 1),
+		DesiredCount: 1, Status: ServiceStatusActive,
+		SchedulingStrategy: SchedulingStrategyReplica, DeploymentID: "legacy-dep",
+	}
+	require.NoError(t, putJSON(kv, ServiceKey("web", "legacy"), rec))
+
+	require.NoError(t, svc.reconcileService(kv, testAccountID, rec))
+	reloaded := reloadService(t, kv, "web", "legacy")
+	primary := reloaded.primaryDeployment()
+	require.NotNil(t, primary)
+	assert.Equal(t, "legacy-dep", primary.ID)
+	assert.Equal(t, defaultMinimumHealthyPercent, reloaded.MinimumHealthyPercent)
+}

spinifex/handlers/ecs/instances.go

@@ -243,6 +243,11 @@ func (s *Service) recordTaskState(msg *bus.TaskState) error {
 	// Deregister targets, release the public IP, release capacity + reclaim the
 	// task ENI once, on the transition into STOPPED.
 	if msg.LastStatus == TaskStatusStopped && prev != TaskStatusStopped {
+		// A task that stops without ever reaching RUNNING is a deployment failure;
+		// feed the owning deployment's circuit breaker.
+		if task.StartedAt.IsZero() {
+			s.recordDeploymentFailure(kv, msg.ClusterName, &task)
+		}
 		s.deregisterServiceTargets(kv, msg.AccountID, &task)
 		s.releaseTaskPublicIP(msg.AccountID, &task)
 		s.reclaimAssignInbox(kv, msg.ClusterName, task.ContainerInstanceID, msg.TaskID)
@@ -286,6 +291,32 @@ func (s *Service) SubmitTaskStateChange(input *ecs.SubmitTaskStateChangeInput, a
 	return &ecs.SubmitTaskStateChangeOutput{Acknowledgment: aws.String("OK")}, nil
 }
 
+// recordDeploymentFailure increments the failed-task counter on the deployment
+// that launched a task which stopped before ever running, driving the service's
+// deployment circuit breaker. No-op for a non-service task or an unknown deployment.
+func (s *Service) recordDeploymentFailure(kv nats.KeyValue, cluster string, task *TaskRecord) {
+	name := serviceNameFromGroup(task.Group)
+	depID := deploymentIDFromStartedBy(task.StartedBy)
+	if name == "" || depID == "" {
+		return
+	}
+	var svc ServiceRecord
+	found, err := getJSON(kv, ServiceKey(cluster, name), &svc)
+	if err != nil || !found {
+		return
+	}
+	for i := range svc.Deployments {
+		if svc.Deployments[i].ID == depID {
+			svc.Deployments[i].FailedTasks++
+			svc.Deployments[i].UpdatedAt = time.Now().UTC()
+			if perr := putJSON(kv, ServiceKey(cluster, name), &svc); perr != nil {
+				slog.Error("ECS deployment failure accounting: persist failed", "service", name, "err", perr)
+			}
+			return
+		}
+	}
+}
+
 // releaseReservation returns a stopped task's capacity to its instance under CAS.
 func (s *Service) releaseReservation(kv nats.KeyValue, cluster, instanceID, taskID string, cpu, mem int) error {
 	for range reservePlacementRetries {

spinifex/handlers/ecs/records.go

@@ -32,8 +32,45 @@ const (
 	// DAEMON is rejected at CreateService.
 	SchedulingStrategyReplica = "REPLICA"
 	SchedulingStrategyDaemon  = "DAEMON"
+
+	// Deployment status: PRIMARY is the deployment being rolled out (or steady);
+	// ACTIVE is a superseded deployment still draining its tasks.
+	DeploymentStatusPrimary = "PRIMARY"
+	DeploymentStatusActive  = "ACTIVE"
+
+	// Deployment rollout state (AWS deploymentController rollout enum subset).
+	RolloutStateInProgress = "IN_PROGRESS"
+	RolloutStateCompleted  = "COMPLETED"
+	RolloutStateFailed     = "FAILED"
+
+	// deploymentConfiguration defaults for a REPLICA service when unset.
+	defaultMinimumHealthyPercent = 100
+	defaultMaximumPercent        = 200
+
+	// circuitBreakerFailureThreshold trips the breaker once this many of the
+	// primary deployment's task launches stop before ever reaching RUNNING.
+	circuitBreakerFailureThreshold = 3
 )
 
+// Deployment is one rollout of a service's task definition. A service has exactly
+// one PRIMARY deployment plus zero or more ACTIVE (superseded, draining) ones
+// while a rolling update is in flight; steady state is a single PRIMARY.
+type Deployment struct {
+	ID              string    `json:"id"`
+	Status          string    `json:"status"`
+	TaskDefARN      string    `json:"taskDefArn"`
+	TaskDefFamily   string    `json:"taskDefFamily"`
+	TaskDefRevision int       `json:"taskDefRevision"`
+	DesiredCount    int       `json:"desiredCount"`
+	RunningCount    int       `json:"runningCount"`
+	PendingCount    int       `json:"pendingCount"`
+	FailedTasks     int       `json:"failedTasks"`
+	RolloutState    string    `json:"rolloutState"`
+	RolloutReason   string    `json:"rolloutStateReason,omitempty"`
+	CreatedAt       time.Time `json:"createdAt"`
+	UpdatedAt       time.Time `json:"updatedAt"`
+}
+
 // ARN builders for the ECS resource shapes (ecs-v1.md §1). Region + accountID
 // scope every ARN; the partition is fixed to "aws" to match the rest of the
 // stack.
@@ -94,20 +131,30 @@ type ContainerDef struct {
 	Command      []string          `json:"command,omitempty"`
 	Environment  map[string]string `json:"environment,omitempty"`
 	PortMappings []bus.PortMapping `json:"portMappings,omitempty"`
+	// LogDriver / LogOptions capture the container's logConfiguration. Only the
+	// host-side json-file default is honored; any other driver is accepted for
+	// parity but warned at register time (logs are discarded).
+	LogDriver  string            `json:"logDriver,omitempty"`
+	LogOptions map[string]string `json:"logOptions,omitempty"`
 }
 
+// LogDriverJSONFile is the only log driver the agent honors: containerd's task IO
+// lands in the host journal/file, retrievable per ecs-logging.md.
+const LogDriverJSONFile = "json-file"
+
 // TaskDefRecord is the persisted task definition revision at TaskDefRevKey.
 type TaskDefRecord struct {
-	Family       string         `json:"family"`
-	Revision     int            `json:"revision"`
-	ARN          string         `json:"arn"`
-	NetworkMode  string         `json:"networkMode,omitempty"`
-	CPU          string         `json:"cpu,omitempty"`
-	Memory       string         `json:"memory,omitempty"`
-	TaskRoleArn  string         `json:"taskRoleArn,omitempty"`
-	Containers   []ContainerDef `json:"containers"`
-	Status       string         `json:"status"`
-	RegisteredAt time.Time      `json:"registeredAt"`
+	Family           string         `json:"family"`
+	Revision         int            `json:"revision"`
+	ARN              string         `json:"arn"`
+	NetworkMode      string         `json:"networkMode,omitempty"`
+	CPU              string         `json:"cpu,omitempty"`
+	Memory           string         `json:"memory,omitempty"`
+	TaskRoleArn      string         `json:"taskRoleArn,omitempty"`
+	ExecutionRoleArn string         `json:"executionRoleArn,omitempty"`
+	Containers       []ContainerDef `json:"containers"`
+	Status           string         `json:"status"`
+	RegisteredAt     time.Time      `json:"registeredAt"`
 }
 
 // reservedCPU/reservedMemory sum the task definition's per-container reservations
@@ -239,6 +286,26 @@ type ServiceRecord struct {
 	DeploymentID       string               `json:"deploymentId"`
 	RunningCount       int                  `json:"runningCount"`
 	PendingCount       int                  `json:"pendingCount"`
-	CreatedAt          time.Time            `json:"createdAt"`
-	UpdatedAt          time.Time            `json:"updatedAt"`
+	// Rolling-update configuration (deploymentConfiguration) and its live state.
+	// MinimumHealthyPercent / MaximumPercent gate the rollout; the circuit breaker
+	// trips a failing deployment and optionally rolls back to LastGoodTaskDefARN.
+	MinimumHealthyPercent  int          `json:"minimumHealthyPercent,omitempty"`
+	MaximumPercent         int          `json:"maximumPercent,omitempty"`
+	CircuitBreakerEnable   bool         `json:"deploymentCircuitBreakerEnable,omitempty"`
+	CircuitBreakerRollback bool         `json:"deploymentCircuitBreakerRollback,omitempty"`
+	LastGoodTaskDefARN     string       `json:"lastGoodTaskDefArn,omitempty"`
+	Deployments            []Deployment `json:"deployments,omitempty"`
+	CreatedAt              time.Time    `json:"createdAt"`
+	UpdatedAt              time.Time    `json:"updatedAt"`
+}
+
+// primaryDeployment returns a pointer to the service's PRIMARY deployment, or nil
+// when none exists (a legacy record before ensurePrimaryDeployment runs).
+func (r *ServiceRecord) primaryDeployment() *Deployment {
+	for i := range r.Deployments {
+		if r.Deployments[i].Status == DeploymentStatusPrimary {
+			return &r.Deployments[i]
+		}
+	}
+	return nil
 }

spinifex/handlers/ecs/service.go

@@ -4,6 +4,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"log/slog"
 	"maps"
 	"slices"
 	"strconv"
@@ -334,6 +335,10 @@ func (s *Service) RegisterTaskDefinition(input *ecs.RegisterTaskDefinitionInput,
 	if family == "" {
 		return nil, errors.New(awserrors.ErrorECSInvalidParameter)
 	}
+	if err := validateContainerDefs(input.ContainerDefinitions); err != nil {
+		return nil, err
+	}
+	warnUnsupportedLogDrivers(family, input.ContainerDefinitions)
 	kv, err := s.bucket(accountID)
 	if err != nil {
 		return nil, err
@@ -345,16 +350,17 @@ func (s *Service) RegisterTaskDefinition(input *ecs.RegisterTaskDefinitionInput,
 	}
 
 	rec := TaskDefRecord{
-		Family:       family,
-		Revision:     rev,
-		ARN:          TaskDefARN(s.region, accountID, family, rev),
-		NetworkMode:  aws.StringValue(input.NetworkMode),
-		CPU:          aws.StringValue(input.Cpu),
-		Memory:       aws.StringValue(input.Memory),
-		TaskRoleArn:  aws.StringValue(input.TaskRoleArn),
-		Status:       TaskDefStatusActive,
-		RegisteredAt: time.Now().UTC(),
-		Containers:   containerDefsFromAWS(input.ContainerDefinitions),
+		Family:           family,
+		Revision:         rev,
+		ARN:              TaskDefARN(s.region, accountID, family, rev),
+		NetworkMode:      aws.StringValue(input.NetworkMode),
+		CPU:              aws.StringValue(input.Cpu),
+		Memory:           aws.StringValue(input.Memory),
+		TaskRoleArn:      aws.StringValue(input.TaskRoleArn),
+		ExecutionRoleArn: aws.StringValue(input.ExecutionRoleArn),
+		Status:           TaskDefStatusActive,
+		RegisteredAt:     time.Now().UTC(),
+		Containers:       containerDefsFromAWS(input.ContainerDefinitions),
 	}
 	if err := putJSON(kv, TaskDefRevKey(family, rev), &rec); err != nil {
 		return nil, err
@@ -365,6 +371,35 @@ func (s *Service) RegisterTaskDefinition(input *ecs.RegisterTaskDefinitionInput,
 	return &ecs.RegisterTaskDefinitionOutput{TaskDefinition: rec.toAWS()}, nil
 }
 
+// validateContainerDefs hard-rejects taskdef features the data plane cannot honor.
+// secrets[] is dropped silently by the assign path, so a task would run believing
+// it has secrets it never receives (ecs-v1 Q18) — reject at register instead.
+func validateContainerDefs(defs []*ecs.ContainerDefinition) error {
+	for _, c := range defs {
+		if c != nil && len(c.Secrets) > 0 {
+			return errors.New(awserrors.ErrorECSInvalidParameter)
+		}
+	}
+	return nil
+}
+
+// warnUnsupportedLogDrivers emits the honest "logs discarded" operator signal for
+// any container whose log driver is not the host-side json-file default. The
+// taskdef is still accepted for parity; only json-file is actually collected.
+func warnUnsupportedLogDrivers(family string, defs []*ecs.ContainerDefinition) {
+	for _, c := range defs {
+		if c == nil || c.LogConfiguration == nil {
+			continue
+		}
+		driver := aws.StringValue(c.LogConfiguration.LogDriver)
+		if driver == "" || driver == LogDriverJSONFile {
+			continue
+		}
+		slog.Warn("ECS RegisterTaskDefinition: logDriver not implemented; container logs will be discarded",
+			"family", family, "container", aws.StringValue(c.Name), "logDriver", driver)
+	}
+}
+
 // nextRevision reads the family's latest-rev and returns latest+1 (1 if absent).
 func (s *Service) nextRevision(kv nats.KeyValue, family string) (int, error) {
 	var latest int
@@ -521,6 +556,9 @@ func (r *TaskDefRecord) toAWS() *ecs.TaskDefinition {
 	if r.TaskRoleArn != "" {
 		td.TaskRoleArn = aws.String(r.TaskRoleArn)
 	}
+	if r.ExecutionRoleArn != "" {
+		td.ExecutionRoleArn = aws.String(r.ExecutionRoleArn)
+	}
 	for _, c := range r.Containers {
 		td.ContainerDefinitions = append(td.ContainerDefinitions, c.toAWS())
 	}

spinifex/handlers/ecs/services.go

@@ -183,6 +183,9 @@ func (s *Service) CreateService(input *ecs.CreateServiceInput, accountID string)
 		rec.SecurityGroups = awsStringSlice(v.SecurityGroups)
 		rec.AssignPublicIP = aws.StringValue(v.AssignPublicIp)
 	}
+	applyDeploymentConfig(&rec, input.DeploymentConfiguration)
+	rec.LastGoodTaskDefARN = taskDef.ARN
+	rec.Deployments = []Deployment{newPrimaryDeployment(rec.DeploymentID, taskDef, rec.DesiredCount)}
 	if err := putJSON(kv, ServiceKey(cluster, name), &rec); err != nil {
 		return nil, err
 	}
@@ -210,6 +213,10 @@ func (s *Service) UpdateService(input *ecs.UpdateServiceInput, accountID string)
 		return nil, errors.New(awserrors.ErrorECSServiceNotFound)
 	}
 
+	rec.ensurePrimaryDeployment()
+	if input.DeploymentConfiguration != nil {
+		applyDeploymentConfig(&rec, input.DeploymentConfiguration)
+	}
 	if input.DesiredCount != nil {
 		rec.DesiredCount = int(aws.Int64Value(input.DesiredCount))
 	}
@@ -218,11 +225,11 @@ func (s *Service) UpdateService(input *ecs.UpdateServiceInput, accountID string)
 		if terr != nil {
 			return nil, terr
 		}
-		rec.TaskDefFamily = taskDef.Family
-		rec.TaskDefRevision = taskDef.Revision
-		rec.TaskDefARN = taskDef.ARN
-		rec.NetworkMode = resolveNetworkMode(taskDef)
-		rec.DeploymentID = uuid.NewString()
+		// A new task definition starts a rolling deployment only when it differs
+		// from the current PRIMARY; a no-op re-apply keeps the deployment steady.
+		if primary := rec.primaryDeployment(); primary == nil || primary.TaskDefARN != taskDef.ARN {
+			rec.startDeployment(taskDef)
+		}
 	}
 	rec.UpdatedAt = time.Now().UTC()
 	if err := putJSON(kv, ServiceKey(cluster, name), &rec); err != nil {
@@ -314,59 +321,56 @@ func (s *Service) ListServices(input *ecs.ListServicesInput, accountID string) (
 
 // --- Reconciliation ---
 
-// reconcileService converges a service's running task count on its desired count:
-// it launches the shortfall via RunTask (tagged with the service group so the
-// task counts toward the service) and force-stops any surplus. Counts on the
-// record are refreshed as a side effect.
+// reconcileService drives a service's rolling deployment toward its desired
+// count: it launches PRIMARY-deployment tasks within the maximumPercent ceiling
+// and drains superseded (old) tasks while healthy running stays at/above
+// minimumHealthyPercent, then advances the rollout / circuit-breaker state. Per-
+// deployment and overall counts are refreshed as a side effect.
 func (s *Service) reconcileService(kv nats.KeyValue, accountID string, svc *ServiceRecord) error {
 	if svc.Status != ServiceStatusActive {
 		return nil
 	}
+	svc.normalizeDeploymentConfig()
+	svc.ensurePrimaryDeployment()
+
 	tasks, err := s.listServiceTasks(kv, svc.Cluster, svc.Name)
 	if err != nil {
 		return err
 	}
+	running, pending := tallyDeployments(svc, tasks)
 
-	running, pending := 0, 0
-	for i := range tasks {
-		switch tasks[i].LastStatus {
-		case TaskStatusRunning:
-			running++
-		case TaskStatusPending:
-			pending++
+	primary := svc.primaryDeployment()
+	if primary != nil {
+		// A failing deployment trips the breaker (and may roll back); persist and
+		// let the next tick roll the replacement in.
+		if tripCircuitBreaker(svc, primary) {
+			svc.RunningCount, svc.PendingCount = running, pending
+			return putJSON(kv, ServiceKey(svc.Cluster, svc.Name), svc)
 		}
+		desired := svc.DesiredCount
+		maxCount := desired * svc.MaximumPercent / 100
+		minCount := ceilPercent(desired, svc.MinimumHealthyPercent)
+
+		primaryActive := primary.RunningCount + primary.PendingCount
+		if primaryActive < desired && primary.RolloutState != RolloutStateFailed {
+			n := min(desired-primaryActive, max(maxCount-(running+pending), 0))
+			if n > 0 {
+				s.launchDeploymentTasks(accountID, svc, primary, n)
+			}
+		}
+		s.stopSurplusTasks(kv, accountID, tasks, primary.ID, desired, running, minCount)
+
+		// Re-tally after launch/stop so rollout state + persisted counts are current.
+		if fresh, ferr := s.listServiceTasks(kv, svc.Cluster, svc.Name); ferr == nil {
+			running, pending = tallyDeployments(svc, fresh)
+		}
+		updateRolloutState(svc, primary, desired)
 	}
 
-	active := running + pending
-	switch {
-	case active < svc.DesiredCount:
-		s.launchServiceTasks(accountID, svc, svc.DesiredCount-active)
-	case active > svc.DesiredCount:
-		s.stopServiceSurplus(kv, accountID, tasks, active-svc.DesiredCount)
-	}
-
-	// Recount after launch/stop so the persisted counts reflect current state.
-	svc.RunningCount, svc.PendingCount = s.countServiceTasks(kv, svc.Cluster, svc.Name)
+	svc.RunningCount, svc.PendingCount = running, pending
 	return putJSON(kv, ServiceKey(svc.Cluster, svc.Name), svc)
 }
 
-// countServiceTasks returns a service's current RUNNING and PENDING task counts.
-func (s *Service) countServiceTasks(kv nats.KeyValue, cluster, name string) (running, pending int) {
-	tasks, err := s.listServiceTasks(kv, cluster, name)
-	if err != nil {
-		return 0, 0
-	}
-	for i := range tasks {
-		switch tasks[i].LastStatus {
-		case TaskStatusRunning:
-			running++
-		case TaskStatusPending:
-			pending++
-		}
-	}
-	return running, pending
-}
-
 // reconcileAllServices is the scheduler-tick fan-out: every ACTIVE service in
 // every ECS account bucket is reconciled. Runs only on the scheduler leader.
 func (s *Service) reconcileAllServices() {
@@ -395,15 +399,16 @@ func (s *Service) reconcileAllServices() {
 	}
 }
 
-// launchServiceTasks places n replacement tasks via the standard RunTask flow,
-// tagged with the service group + deployment so they count toward the service.
-func (s *Service) launchServiceTasks(accountID string, svc *ServiceRecord, n int) {
+// launchDeploymentTasks places n tasks for a specific deployment via the standard
+// RunTask flow, tagged with the service group + the deployment ID (StartedBy) so
+// the reconciler maps them back to the deployment they belong to.
+func (s *Service) launchDeploymentTasks(accountID string, svc *ServiceRecord, dep *Deployment, n int) {
 	in := &ecs.RunTaskInput{
 		Cluster:        aws.String(svc.Cluster),
-		TaskDefinition: aws.String(svc.TaskDefARN),
+		TaskDefinition: aws.String(dep.TaskDefARN),
 		Count:          aws.Int64(int64(n)),
 		Group:          aws.String(serviceTaskGroup(svc.Name)),
-		StartedBy:      aws.String("ecs-svc/" + svc.DeploymentID),
+		StartedBy:      aws.String(deploymentStartedBy(dep.ID)),
 	}
 	if svc.PlacementStrategy != "" {
 		in.PlacementStrategy = []*ecs.PlacementStrategy{{Type: aws.String(svc.PlacementStrategy)}}
@@ -429,26 +434,57 @@ func (s *Service) launchServiceTasks(accountID string, svc *ServiceRecord, n int
 	}
 }
 
-// stopServiceSurplus force-stops n of a service's tasks, preferring PENDING tasks
-// (least disruptive) before RUNNING ones.
-func (s *Service) stopServiceSurplus(kv nats.KeyValue, accountID string, tasks []TaskRecord, n int) {
-	order := make([]int, 0, len(tasks))
+// stopSurplusTasks drains a service's excess tasks: superseded (non-primary)
+// deployment tasks toward zero, and the primary's own surplus on a scale-in.
+// PENDING surplus is stopped freely; RUNNING surplus is gated so healthy running
+// never drops below minimumHealthyPercent (minCount).
+func (s *Service) stopSurplusTasks(kv nats.KeyValue, accountID string, tasks []TaskRecord, primaryID string, desired, runningTotal, minCount int) {
+	var oldPending, oldRunning, primaryPending, primaryRunning []int
 	for i := range tasks {
-		if tasks[i].LastStatus == TaskStatusPending {
-			order = append(order, i)
+		isPrimary := deploymentIDFromStartedBy(tasks[i].StartedBy) == primaryID
+		switch tasks[i].LastStatus {
+		case TaskStatusPending:
+			if isPrimary {
+				primaryPending = append(primaryPending, i)
+			} else {
+				oldPending = append(oldPending, i)
+			}
+		case TaskStatusRunning:
+			if isPrimary {
+				primaryRunning = append(primaryRunning, i)
+			} else {
+				oldRunning = append(oldRunning, i)
+			}
 		}
 	}
-	for i := range tasks {
-		if tasks[i].LastStatus == TaskStatusRunning {
-			order = append(order, i)
-		}
+	primarySurplus := max(len(primaryPending)+len(primaryRunning)-desired, 0)
+	runBudget := max(runningTotal-minCount, 0)
+	stop := func(i int, reason string) { s.forceStopTask(kv, accountID, &tasks[i], reason) }
+
+	for _, i := range oldPending {
+		stop(i, deploymentSupersededReason)
 	}
-	for _, idx := range order {
-		if n <= 0 {
+	for _, i := range primaryPending {
+		if primarySurplus <= 0 {
 			break
 		}
-		s.forceStopTask(kv, accountID, &tasks[idx], "Service scaled in")
-		n--
+		stop(i, "Service scaled in")
+		primarySurplus--
+	}
+	for _, i := range oldRunning {
+		if runBudget <= 0 {
+			break
+		}
+		stop(i, deploymentSupersededReason)
+		runBudget--
+	}
+	for _, i := range primaryRunning {
+		if primarySurplus <= 0 || runBudget <= 0 {
+			break
+		}
+		stop(i, "Service scaled in")
+		primarySurplus--
+		runBudget--
 	}
 }
 
@@ -580,5 +616,34 @@ func (s *Service) serviceToAWS(accountID string, r *ServiceRecord) *ecs.Service
 			ContainerPort:  aws.Int64(int64(lb.ContainerPort)),
 		})
 	}
+	if r.MinimumHealthyPercent > 0 || r.MaximumPercent > 0 {
+		svc.DeploymentConfiguration = &ecs.DeploymentConfiguration{
+			MinimumHealthyPercent: aws.Int64(int64(r.MinimumHealthyPercent)),
+			MaximumPercent:        aws.Int64(int64(r.MaximumPercent)),
+			DeploymentCircuitBreaker: &ecs.DeploymentCircuitBreaker{
+				Enable:   aws.Bool(r.CircuitBreakerEnable),
+				Rollback: aws.Bool(r.CircuitBreakerRollback),
+			},
+		}
+	}
+	for i := range r.Deployments {
+		d := &r.Deployments[i]
+		svc.Deployments = append(svc.Deployments, &ecs.Deployment{
+			Id:             aws.String(d.ID),
+			Status:         aws.String(d.Status),
+			TaskDefinition: aws.String(d.TaskDefARN),
+			DesiredCount:   aws.Int64(int64(d.DesiredCount)),
+			RunningCount:   aws.Int64(int64(d.RunningCount)),
+			PendingCount:   aws.Int64(int64(d.PendingCount)),
+			FailedTasks:    aws.Int64(int64(d.FailedTasks)),
+			RolloutState:   aws.String(d.RolloutState),
+			RolloutStateReason: func() *string {
+				if d.RolloutReason == "" {
+					return nil
+				}
+				return aws.String(d.RolloutReason)
+			}(),
+		})
+	}
 	return svc
 }

spinifex/handlers/ecs/taskdef_validation_test.go

@@ -0,0 +1,91 @@
+package handlers_ecs
+
+import (
+	"testing"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/service/ecs"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// TestRegisterTaskDefinition_RejectsSecrets covers siv-458: a container secrets[]
+// is hard-rejected rather than silently dropped.
+func TestRegisterTaskDefinition_RejectsSecrets(t *testing.T) {
+	svc, _ := newTestService(t)
+	_, err := svc.RegisterTaskDefinition(&ecs.RegisterTaskDefinitionInput{
+		Family: aws.String("app"),
+		ContainerDefinitions: []*ecs.ContainerDefinition{{
+			Name: aws.String("app"), Image: aws.String("registry/app:1"), Essential: aws.Bool(true),
+			Secrets: []*ecs.Secret{{
+				Name: aws.String("DB_PASSWORD"), ValueFrom: aws.String("arn:aws:ssm:::parameter/db"),
+			}},
+		}},
+	}, testAccountID)
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "InvalidParameterException")
+}
+
+// TestRegisterTaskDefinition_AcceptsUnsupportedLogDriver covers siv-455/458: a
+// non-json-file driver is accepted for parity (warned, not rejected) and the
+// driver round-trips through DescribeTaskDefinition.
+func TestRegisterTaskDefinition_AcceptsUnsupportedLogDriver(t *testing.T) {
+	svc, _ := newTestService(t)
+	_, err := svc.RegisterTaskDefinition(&ecs.RegisterTaskDefinitionInput{
+		Family: aws.String("app"),
+		ContainerDefinitions: []*ecs.ContainerDefinition{{
+			Name: aws.String("app"), Image: aws.String("registry/app:1"), Essential: aws.Bool(true),
+			LogConfiguration: &ecs.LogConfiguration{
+				LogDriver: aws.String("awslogs"),
+				Options:   map[string]*string{"awslogs-group": aws.String("/ecs/app")},
+			},
+		}},
+	}, testAccountID)
+	require.NoError(t, err)
+
+	d, err := svc.DescribeTaskDefinition(&ecs.DescribeTaskDefinitionInput{
+		TaskDefinition: aws.String("app"),
+	}, testAccountID)
+	require.NoError(t, err)
+	require.Len(t, d.TaskDefinition.ContainerDefinitions, 1)
+	lc := d.TaskDefinition.ContainerDefinitions[0].LogConfiguration
+	require.NotNil(t, lc)
+	assert.Equal(t, "awslogs", aws.StringValue(lc.LogDriver))
+	assert.Equal(t, "/ecs/app", aws.StringValue(lc.Options["awslogs-group"]))
+}
+
+// TestRunTask_AssignCarriesExecutionRoleAndLogDriver covers siv-459 (execution
+// role plumbed to the agent) and siv-455 (log driver reaches the assign).
+func TestRunTask_AssignCarriesExecutionRoleAndLogDriver(t *testing.T) {
+	svc, _ := newTestService(t)
+	_, err := svc.CreateCluster(&ecs.CreateClusterInput{ClusterName: aws.String("web")}, testAccountID)
+	require.NoError(t, err)
+	execARN := "arn:aws:iam::123456789012:role/exec-app"
+	_, err = svc.RegisterTaskDefinition(&ecs.RegisterTaskDefinitionInput{
+		Family:           aws.String("app"),
+		ExecutionRoleArn: aws.String(execARN),
+		ContainerDefinitions: []*ecs.ContainerDefinition{{
+			Name: aws.String("app"), Image: aws.String("registry/app:1"),
+			Cpu: aws.Int64(128), Memory: aws.Int64(256), Essential: aws.Bool(true),
+			LogConfiguration: &ecs.LogConfiguration{LogDriver: aws.String(LogDriverJSONFile)},
+		}},
+	}, testAccountID)
+	require.NoError(t, err)
+	registerInstance(t, svc, "web", "i-1", 1024, 2048)
+
+	d, err := svc.DescribeTaskDefinition(&ecs.DescribeTaskDefinitionInput{TaskDefinition: aws.String("app")}, testAccountID)
+	require.NoError(t, err)
+	assert.Equal(t, execARN, aws.StringValue(d.TaskDefinition.ExecutionRoleArn))
+
+	_, err = svc.RunTask(&ecs.RunTaskInput{
+		Cluster: aws.String("web"), TaskDefinition: aws.String("app"), Count: aws.Int64(1),
+	}, testAccountID)
+	require.NoError(t, err)
+
+	poll, err := svc.PollAssignments(&PollAssignmentsInput{Cluster: "web", ContainerInstance: "i-1"}, testAccountID)
+	require.NoError(t, err)
+	require.Len(t, poll.Assignments, 1)
+	assert.Equal(t, execARN, poll.Assignments[0].ExecutionRoleARN)
+	require.Len(t, poll.Assignments[0].Containers, 1)
+	assert.Equal(t, LogDriverJSONFile, poll.Assignments[0].Containers[0].LogDriver)
+}

spinifex/handlers/ecs/tasks.go

@@ -192,19 +192,20 @@ func (s *Service) newTaskRecord(accountID, cluster, taskID string, td *TaskDefRe
 // an unacked assign survives an agent crash and is re-delivered on the next poll.
 func (s *Service) publishAssign(kv nats.KeyValue, accountID, cluster, instanceID string, rec *TaskRecord, td *TaskDefRecord) error {
 	msg := bus.Assign{
-		AccountID:       accountID,
-		ClusterName:     cluster,
-		InstanceID:      instanceID,
-		TaskID:          rec.TaskID,
-		TaskARN:         rec.ARN,
-		TaskDefFamily:   td.Family,
-		TaskDefRevision: td.Revision,
-		TaskRoleARN:     td.TaskRoleArn,
-		ENIID:           rec.ENIID,
-		ENIMacAddress:   rec.ENIMacAddress,
-		ENIPrivateIP:    rec.ENIPrivateIP,
-		ENISubnetID:     rec.ENISubnetID,
-		AssignedAt:      time.Now().UTC(),
+		AccountID:        accountID,
+		ClusterName:      cluster,
+		InstanceID:       instanceID,
+		TaskID:           rec.TaskID,
+		TaskARN:          rec.ARN,
+		TaskDefFamily:    td.Family,
+		TaskDefRevision:  td.Revision,
+		TaskRoleARN:      td.TaskRoleArn,
+		ExecutionRoleARN: td.ExecutionRoleArn,
+		ENIID:            rec.ENIID,
+		ENIMacAddress:    rec.ENIMacAddress,
+		ENIPrivateIP:     rec.ENIPrivateIP,
+		ENISubnetID:      rec.ENISubnetID,
+		AssignedAt:       time.Now().UTC(),
 	}
 	for _, c := range td.Containers {
 		msg.Containers = append(msg.Containers, c.toAssignContainer())