Support -parallel > 1 for real apply

README.md

@@ -166,17 +166,39 @@ This performs the same validation as `apply` and `run`, with a slight overhead,
 
 ### Parallel Execution
 
-The `diff` command and `apply -dry-run` support parallel execution for improved performance with many rules:
+The `diff` and `apply` commands support parallel execution for improved performance with many rules:
 
 ```console
 % ecschedule -conf ecschedule.yaml diff -all -parallel 10
 % ecschedule -conf ecschedule.yaml apply -all -dry-run -parallel 10
+% ecschedule -conf ecschedule.yaml apply -all -parallel 10
 ```
 
 - Default: `parallel=1` (sequential, backward compatible)
-- Recommended: 1-10 (due to AWS API rate limits)
+- Recommended: 1-10 (due to AWS API rate limits; quotas are per-account per-region and adjustable via Service Quotas — large rule sets should expect throttling-driven retries)
 - Note: Output order is not guaranteed when parallel > 1
-- Note: `-parallel` is only effective with `-dry-run` for the `apply` command. Using it without `-dry-run` returns an error.
+
+For real `apply`, the failure semantics depend on `-parallel`:
+
+| | behavior on a rule failure |
+|---|---|
+| `-parallel 1` (default) | stops at the first error (unchanged) |
+| `-parallel N` (N > 1) | all rules are validated **before any write**; every changed rule is then attempted; failures are aggregated into an end-of-run summary and the exit status is non-zero |
+
+Notes for parallel apply (`-parallel > 1`):
+
+- A rule that has started writing always completes its `PutRule` → `PutTargets` → `TagResource` sequence (bounded by a 2-minute per-rule timeout). The first Ctrl-C stops starting new rules and waits for in-flight ones; a second Ctrl-C force-quits. In CI, make sure the cancellation grace period exceeds the per-rule timeout or a hard kill may still interrupt a write.
+- Re-running `apply` after a partial failure converges, with two exceptions: changing a `targetId` leaves the old target attached remotely (the rule fires both targets and subsequently diffs as a new rule), and a rule whose tagging (`TagResource`) failed is NOT healed by re-run — the failure message prints the exact `aws events tag-resource` command to repair it.
+- If every rule retries for ~2 minutes and then fails with `LimitExceededException`, you hit the rules-per-bus quota — raise it via Service Quotas instead of waiting out retries.
+- Rule names must be unique: configurations containing duplicate rule names are rejected at load time (for every subcommand).
+- `-prune` orphan detection only searches the region of your ambient AWS configuration; rules in per-rule `region` overrides are outside its view.
+- With `-parallel > 1` the SDK retry budget is shared per region (not per rule), so under sustained failures some rules may fail fast with a rate-limit-token error instead of the underlying one — two error texts can appear for a single root cause.
+
+### Signal Handling
+
+Since the introduction of parallel apply, all subcommands handle SIGINT/SIGTERM gracefully. The first signal stops work at API-call boundaries and the command exits through the normal error path (exit status 1 with a `💢` message) instead of dying to the signal (previously exit status ~130). A second signal restores default fatal handling and kills the process immediately.
+
+One part is deliberately exempt from that first signal: a rule whose write has already begun always completes its `PutRule` → `PutTargets` → `TagResource` sequence. That sequence is detached from the signal (it runs under `context.WithoutCancel`) and bounded only by the per-rule timeout, so no rule is ever left half-written. This holds for any real `apply`, with or without `-parallel`. What the first signal does stop in a parallel apply is admission: no further rules start, and the command waits for the in-flight ones.
 
 ## Log Format
 
@@ -190,7 +212,7 @@ ecschedule supports unified diff format (similar to `git diff`) with `-u` flag f
 ```
 
 The `diff` command with `-u` flag outputs pure diff content without log prefixes or headers, making it suitable for piping to other tools.
-The `apply` command includes progress logs even with `-u` flag.
+The `apply` command includes progress logs even with `-u` flag. With `-parallel > 1`, each rule's *result* — the diff and the resulting YAML — is written as a single indivisible block, so it never interleaves with another rule's result. The short `applying rule "..."` progress lines are emitted as each rule starts, so those do appear between result blocks.
 
 ### Color control
 

cmd/ecschedule/main.go

@@ -5,13 +5,24 @@ import (
 	"flag"
 	"log"
 	"os"
+	"os/signal"
+	"syscall"
 
 	"github.com/Songmu/ecschedule"
 )
 
 func main() {
 	log.SetFlags(0)
-	err := ecschedule.Run(context.Background(), os.Args[1:], os.Stdout, os.Stderr)
+	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
+	defer stop()
+	// After the first signal cancels ctx, unregister so a second
+	// SIGINT/SIGTERM gets default (fatal) handling — the escape hatch
+	// when an in-flight AWS call refuses to finish.
+	go func() {
+		<-ctx.Done()
+		stop()
+	}()
+	err := ecschedule.Run(ctx, os.Args[1:], os.Stdout, os.Stderr)
 	if err != nil && err != flag.ErrHelp {
 		log.Printf("💢 %s\n", err)
 		exitCode := 1

cmd_apply.go

@@ -31,7 +31,7 @@ var cmdApply = &runnerImpl{
 			prune    = fs.Bool("prune", false, "prune orphaned rules after apply")
 			unified  = fs.Bool("u", false, "output diff in unified format (colored, similar to git diff)")
 			noColor  = fs.Bool("no-color", false, "disable colored output (Unified diff format only)")
-			parallel = fs.Int("parallel", 1, "number of parallel workers for dry-run (default: 1, only effective with -dry-run, recommended: 1-10 due to AWS API rate limits. Note: output order is not guaranteed when parallel > 1)")
+			parallel = fs.Int("parallel", 1, "number of parallel workers (default: 1, recommended: 1-10 due to AWS API rate limits. Note: output order is not guaranteed when parallel > 1)")
 		)
 		if err := fs.Parse(argv); err != nil {
 			return err
@@ -42,6 +42,9 @@ var cmdApply = &runnerImpl{
 		if !*all && *rule == "" {
 			return errors.New("-rule or -all option required")
 		}
+		if *parallel < 1 {
+			return errors.New("-parallel must be at least 1")
+		}
 		a := getApp(ctx)
 		c := a.Config
 		if *conf != "" {
@@ -68,13 +71,6 @@ var cmdApply = &runnerImpl{
 			}
 		}
 
-		if *parallel < 1 {
-			return errors.New("-parallel must be at least 1")
-		}
-		if *parallel > 1 && !*dryRun {
-			return errors.New("-parallel can only be used with -dry-run (apply parallelization is not yet supported)")
-		}
-
 		var dryRunSuffix string
 		if *dryRun {
 			dryRunSuffix = " (dry-run)"
@@ -83,13 +79,16 @@ var cmdApply = &runnerImpl{
 		format := selectDiffFormat(*unified)
 
 		if *dryRun {
+			// Shared across workers so rules pointing at the same task
+			// definition describe it once instead of once per rule.
+			tdv := newTaskDefValidator()
 			processApplyDryRunJob := func(ctx context.Context, ruleName string) (applyDryRunResult, error) {
 				ru := c.GetRuleByName(ruleName)
 				if ru == nil {
 					return applyDryRunResult{}, fmt.Errorf("no rules found for %s", ruleName)
 				}
 				log.Printf("applying the rule %q%s", ruleName, dryRunSuffix)
-				if err := ru.applyInternal(ctx, a.AwsConf, true, format); err != nil {
+				if err := ru.applyInternalWith(ctx, a.AwsConf, true, format, tdv); err != nil {
 					return applyDryRunResult{}, err
 				}
 				for _, v := range ru.ContainerOverrides {
@@ -106,6 +105,10 @@ var cmdApply = &runnerImpl{
 			if err := <-errChan; err != nil {
 				return err
 			}
+		} else if *parallel > 1 {
+			if err := runParallelApply(ctx, a.AwsConf, c, ruleNames, *parallel, format); err != nil {
+				return err
+			}
 		} else {
 			for _, rule := range ruleNames {
 				ru := c.GetRuleByName(rule)

cmd_diff.go

@@ -86,6 +86,10 @@ var cmdDiff = &runnerImpl{
 
 		var hasValidationError atomic.Bool
 
+		// Shared across workers so rules pointing at the same task
+		// definition describe it once instead of once per rule.
+		tdv := newTaskDefValidator()
+
 		processDiffJob := func(ctx context.Context, ruleName string) (diffResult, error) {
 			result := diffResult{ruleName: ruleName}
 
@@ -104,7 +108,7 @@ var cmdDiff = &runnerImpl{
 				if err := ru.validateSSM(); err != nil {
 					result.validationErrors = append(result.validationErrors, fmt.Sprintf("  ssm: %s", err))
 				}
-				if err := ru.validateTaskDefinition(ctx, a.AwsConf); err != nil {
+				if err := tdv.validate(ctx, ru, a.AwsConf); err != nil {
 					result.validationErrors = append(result.validationErrors, fmt.Sprintf("  task definition: %s", err))
 				}
 

config.go

@@ -75,6 +75,29 @@ func (c *Config) cronValidate() error {
 	return nil
 }
 
+// validateUniqueRuleNames rejects configurations containing multiple rules
+// with the same name. GetRuleByName resolves names to the first match, so
+// duplicates silently shadow each other and, worse, hand the same *Rule to
+// multiple parallel workers (a data race).
+func (c *Config) validateUniqueRuleNames() error {
+	seen := map[string]bool{}
+	reported := map[string]bool{}
+	var dups []string
+	for _, r := range c.Rules {
+		if seen[r.Name] && !reported[r.Name] {
+			dups = append(dups, r.Name)
+			reported[r.Name] = true
+		}
+		seen[r.Name] = true
+	}
+	if len(dups) > 0 {
+		return fmt.Errorf(
+			"duplicate rule name(s) in configuration: %s (rule names must be unique; regenerate a clean config with `ecschedule dump -region <region> -cluster <cluster>`)",
+			strings.Join(dups, ", "))
+	}
+	return nil
+}
+
 func validateCronExpression(exp string) error {
 	if strings.HasPrefix(exp, "rate(") && strings.HasSuffix(exp, ")") {
 		return nil
@@ -175,6 +198,9 @@ func LoadConfig(ctx context.Context, r io.Reader, accountID string, confPath str
 	for _, r := range c.Rules {
 		r.mergeBaseConfig(c.BaseConfig, c.Role)
 	}
+	if err := c.validateUniqueRuleNames(); err != nil {
+		return nil, err
+	}
 	return &c, nil
 }
 

config_test.go

@@ -5,6 +5,7 @@ import (
 	"os"
 	"path/filepath"
 	"reflect"
+	"strings"
 	"testing"
 	"text/template"
 
@@ -279,3 +280,36 @@ func TestCronValidate(t *testing.T) {
 		t.Errorf("unexpected error message\nwant:\n%s\n\ngot:\n%s", e, g)
 	}
 }
+
+func TestLoadConfigDuplicateRuleNames(t *testing.T) {
+	conf := `region: us-east-1
+cluster: api
+rules:
+- name: dup-task
+  scheduleExpression: cron(0 0 * * ? *)
+  taskDefinition: task1
+- name: dup-task
+  scheduleExpression: cron(5 0 * * ? *)
+  taskDefinition: task2
+`
+	dir := t.TempDir()
+	path := filepath.Join(dir, "dup.yaml")
+	if err := os.WriteFile(path, []byte(conf), 0644); err != nil {
+		t.Fatal(err)
+	}
+	f, err := os.Open(path)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer f.Close()
+	_, err = LoadConfig(context.Background(), f, "334", path)
+	if err == nil {
+		t.Fatal("expected duplicate rule name error, got nil")
+	}
+	if !strings.Contains(err.Error(), "dup-task") {
+		t.Errorf("error should name the duplicate rule, got: %s", err)
+	}
+	if !strings.Contains(err.Error(), "ecschedule dump") {
+		t.Errorf("error should mention the dump escape hatch, got: %s", err)
+	}
+}

parallel.go

@@ -10,6 +10,17 @@ import (
 	"golang.org/x/sync/errgroup"
 )
 
+// executeJobsInParallel is the original fail-fast runner: all jobs share
+// one errgroup.WithContext, so the first error cancels the context and the
+// rest stop early. diff and apply -dry-run still use it.
+//
+// New callers should prefer executeJobsInParallelContinueOnError, the more
+// general of the two: fail-fast can be layered on top of it by cancelling
+// a derived context when the first error arrives, while the reverse is not
+// possible. Moving the remaining callers over would change what they
+// report — every rule's error instead of only the first — so that is left
+// out of this change. The intent is to consolidate on that runner and drop
+// this function once the behavior change is acceptable.
 func executeJobsInParallel[T any](
 	ctx context.Context,
 	ruleNames []string,
@@ -65,3 +76,68 @@ func executeJobsInParallel[T any](
 
 	return results, errChan
 }
+
+// jobOutcome is the result of one named job run by
+// executeJobsInParallelContinueOnError.
+type jobOutcome[T any] struct {
+	Index   int // position in names; summaries sort by this
+	Name    string
+	Result  T
+	Err     error
+	Skipped bool // admission saw a canceled ctx; jobFunc never ran
+}
+
+// executeJobsInParallelContinueOnError runs jobFunc for every name with
+// at most `parallel` workers and returns a channel of outcomes in
+// completion order.
+//
+// Contract:
+//   - The channel is buffered to len(names); sends never block.
+//   - Submission, g.Wait(), and close all run in one background goroutine;
+//     the function returns the channel immediately, so callers can consume
+//     outcomes while jobs are still being admitted.
+//   - One job's failure never cancels sibling jobs (plain errgroup.Group,
+//     errors are recorded in the outcome, never returned to the group).
+//   - Every name yields exactly one outcome on every path — success,
+//     error, admission skip, panic — via a single deferred send. The
+//     submission loop never breaks early, so Index coverage is total.
+//   - Each job checks ctx at admission; if already canceled it emits
+//     Skipped without running jobFunc. Jobs are admitted in names order
+//     (g.Go blocks while saturated, blocking the background goroutine).
+//   - A panic in jobFunc is recovered and recorded as that job's error,
+//     with the stack embedded.
+func executeJobsInParallelContinueOnError[T any](
+	ctx context.Context,
+	names []string,
+	parallel int,
+	jobFunc func(ctx context.Context, name string) (T, error),
+) <-chan jobOutcome[T] {
+	outcomes := make(chan jobOutcome[T], len(names))
+	var g errgroup.Group
+	g.SetLimit(parallel)
+
+	go func() {
+		for i, name := range names {
+			g.Go(func() error {
+				out := jobOutcome[T]{Index: i, Name: name}
+				defer func() {
+					if rec := recover(); rec != nil {
+						out.Err = fmt.Errorf("panic in worker for rule %q: %v\n%s",
+							name, rec, debug.Stack())
+					}
+					outcomes <- out
+				}()
+				if ctx.Err() != nil {
+					out.Skipped = true
+					return nil
+				}
+				out.Result, out.Err = jobFunc(ctx, name)
+				return nil
+			})
+		}
+		_ = g.Wait()
+		close(outcomes)
+	}()
+
+	return outcomes
+}