feat(recovery): add exact job retry CAS

cmd/gha-fleet/main.go

@@ -137,6 +137,8 @@ func run(args []string, stdout, stderr io.Writer) int {
 		return runRecoverQueueIntent(args[1:], stdout, stderr, true)
 	case "recover-provider-retry":
 		return runRecoverProviderRetry(args[1:], stdout, stderr)
+	case "recover-provider-job-retry":
+		return runRecoverProviderJobRetry(args[1:], stdout, stderr)
 	case "version":
 		if err := writeJSON(stdout, map[string]string{"version": version, "commit": commit}); err != nil {
 			fmt.Fprintf(stderr, "gha-fleet: %v\n", err)
@@ -505,6 +507,56 @@ func runRecoverProviderRetry(args []string, stdout, stderr io.Writer) int {
 	return 0
 }
 
+func runRecoverProviderJobRetry(args []string, stdout, stderr io.Writer) int {
+	flags := flag.NewFlagSet("recover-provider-job-retry", flag.ContinueOnError)
+	flags.SetOutput(stderr)
+	journalPath := flags.String("journal", "/var/lib/gha-fleet/create-retries.json", "exact GARM provider retry journal")
+	lockPath := flags.String("lock", "/var/lib/gha-fleet/create-retries.lock", "exact GARM provider retry lock")
+	queuePath := flags.String("queue", "/var/lib/gha-fleet/queue-intents.json", "exact durable queue journal")
+	key := flags.String("key", "", "exact terminal job retry key")
+	entityID := flags.String("entity-id", "", "exact forge entity UUID encoded in the retry key")
+	scaleSetID := flags.Uint("scale-set-id", 0, "exact GARM scale-set database ID encoded in the retry key")
+	errorClass := flags.String("error-class", "", "exact recoverable error class")
+	updatedAtText := flags.String("updated-at", "", "exact RFC3339Nano updated_at precondition")
+	apply := flags.Bool("apply", false, "remove the exact proven terminal job circuit")
+	if err := flags.Parse(args); err != nil {
+		return 2
+	}
+	if flags.NArg() != 0 || *key == "" || *entityID == "" || *scaleSetID == 0 || *errorClass == "" || *updatedAtText == "" {
+		fmt.Fprintln(stderr, "gha-fleet: recover-provider-job-retry requires --key, --entity-id, --scale-set-id, --error-class and --updated-at")
+		return 2
+	}
+	if os.Geteuid() == 0 {
+		fmt.Fprintln(stderr, "gha-fleet: recover-provider-job-retry must run as the garm service account")
+		return 1
+	}
+	active, err := garmServiceActive()
+	if err != nil || active {
+		if err == nil {
+			err = errors.New("garm.service must be stopped")
+		}
+		fmt.Fprintf(stderr, "gha-fleet: recover provider job retry: %v\n", err)
+		return 1
+	}
+	updatedAt, err := time.Parse(time.RFC3339Nano, *updatedAtText)
+	if err != nil {
+		fmt.Fprintf(stderr, "gha-fleet: recover provider job retry: invalid updated_at: %v\n", err)
+		return 2
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+	result, err := providerretry.RecoverExactJobTerminal(ctx, *journalPath, *lockPath, *queuePath, *key, *entityID, *scaleSetID, *errorClass, updatedAt, *apply)
+	if err != nil {
+		fmt.Fprintf(stderr, "gha-fleet: recover provider job retry: %v\n", err)
+		return 1
+	}
+	if err := writeJSON(stdout, result); err != nil {
+		fmt.Fprintf(stderr, "gha-fleet: %v\n", err)
+		return 1
+	}
+	return 0
+}
+
 var garmServiceActive = func() (bool, error) {
 	err := exec.Command("systemctl", "is-active", "--quiet", "garm.service").Run()
 	if err == nil {
@@ -1826,5 +1878,5 @@ func runCapacity(args []string, stdout, stderr io.Writer) int {
 }
 
 func printUsage(writer io.Writer) {
-	fmt.Fprintln(writer, "usage: gha-fleet <validate|validate-cache|validate-cache-broker|validate-telemetry|validate-rustfs-cache|validate-diagnostic-exporter|validate-diagnostic-storage|validate-tenant-registry|validate-queue-admission|validate-observability-rules|validate-observability-dashboards|render-openobserve-alerts|render-openobserve-dashboards|reconcile-openobserve-alerts|reconcile-openobserve-dashboards|render|admit|preflight|publish-pressure|reconcile-incus|reconcile-image|bootstrap-github-app|verify-github-app|reconcile-garm|reconcile-zot-credentials|reconcile-rustfs-cache|reconcile-diagnostic-storage|render-garm-build|provider-release|fleet-contract|capacity|version> [options]")
+	fmt.Fprintln(writer, "usage: gha-fleet <validate|validate-cache|validate-cache-broker|validate-telemetry|validate-rustfs-cache|validate-diagnostic-exporter|validate-diagnostic-storage|validate-tenant-registry|validate-queue-admission|validate-observability-rules|validate-observability-dashboards|render-openobserve-alerts|render-openobserve-dashboards|reconcile-openobserve-alerts|reconcile-openobserve-dashboards|render|admit|preflight|publish-pressure|reconcile-incus|reconcile-image|bootstrap-github-app|verify-github-app|reconcile-garm|reconcile-zot-credentials|reconcile-rustfs-cache|reconcile-diagnostic-storage|render-garm-build|provider-release|fleet-contract|capacity|recover-provider-retry|recover-provider-job-retry|version> [options]")
 }

internal/providerretry/recovery_linux.go

@@ -19,6 +19,8 @@ import (
 
 const minimumTerminalRecoveryAge = time.Minute
 
+const maximumQueueJournalBytes = 1024 * 1024
+
 type record struct {
 	JobID          string    `json:"job_id"`
 	Attempts       int       `json:"attempts"`
@@ -164,6 +166,30 @@ func RecoverTerminal(ctx context.Context, journalPath, lockPath, key, entityID s
 		(errorClass != "provider" && errorClass != "intent" && errorClass != "identity" && errorClass != "timeout") || expectedUpdatedAt.IsZero() {
 		return RecoveryResult{}, errors.New("exact scale-set retry key, entity_id, scale_set_id, recoverable error class and updated_at are required")
 	}
+	return recoverTerminalRecord(ctx, journalPath, lockPath, key, entityID, scaleSetID, errorClass, expectedUpdatedAt, 8, nil, apply)
+}
+
+func RecoverExactJobTerminal(ctx context.Context, journalPath, lockPath, queuePath, key, entityID string, scaleSetID uint, errorClass string, expectedUpdatedAt time.Time, apply bool) (RecoveryResult, error) {
+	jobID, parsedEntityID, parsedScaleSetID, err := parseExactJobKey(key)
+	if err != nil || parsedEntityID != entityID || parsedScaleSetID != scaleSetID ||
+		(errorClass != "provider" && errorClass != "identity" && errorClass != "timeout") || expectedUpdatedAt.IsZero() {
+		return RecoveryResult{}, errors.New("exact job retry key, entity_id, scale_set_id, recoverable error class and updated_at are required")
+	}
+	if !filepath.IsAbs(queuePath) || filepath.Dir(queuePath) != filepath.Dir(journalPath) {
+		return RecoveryResult{}, errors.New("queue journal must be an absolute retry-journal sibling")
+	}
+	return recoverTerminalRecord(ctx, journalPath, lockPath, key, entityID, scaleSetID, errorClass, expectedUpdatedAt, 3, func(now time.Time) error {
+		return requireActiveQueueJob(queuePath, jobID, now)
+	}, apply)
+}
+
+func recoverTerminalRecord(ctx context.Context, journalPath, lockPath, key, entityID string, scaleSetID uint, errorClass string, expectedUpdatedAt time.Time, attempts int, prove func(time.Time) error, apply bool) (RecoveryResult, error) {
+	if err := ctx.Err(); err != nil {
+		return RecoveryResult{}, err
+	}
+	if !filepath.IsAbs(journalPath) || !filepath.IsAbs(lockPath) || filepath.Dir(journalPath) != filepath.Dir(lockPath) || journalPath == lockPath {
+		return RecoveryResult{}, errors.New("retry journal and lock must be distinct absolute siblings")
+	}
 	parent := filepath.Dir(journalPath)
 	resolved, err := filepath.EvalSymlinks(parent)
 	if err != nil || filepath.Clean(resolved) != filepath.Clean(parent) {
@@ -191,7 +217,7 @@ func RecoverTerminal(ctx context.Context, journalPath, lockPath, key, entityID s
 		return RecoveryResult{}, err
 	}
 	retry, exists := state.Records[key]
-	if !exists || retry.JobID != key || retry.Attempts != 8 || retry.LastErrorClass != errorClass || retry.TerminalUntil.IsZero() {
+	if !exists || retry.JobID != key || retry.Attempts != attempts || retry.LastErrorClass != errorClass || retry.TerminalUntil.IsZero() {
 		return RecoveryResult{}, errors.New("retry record is not the exact terminal circuit")
 	}
 	if !retry.UpdatedAt.Equal(expectedUpdatedAt.UTC()) {
@@ -201,6 +227,11 @@ func RecoverTerminal(ctx context.Context, journalPath, lockPath, key, entityID s
 	if now.Sub(retry.UpdatedAt) < minimumTerminalRecoveryAge {
 		return RecoveryResult{}, errors.New("retry circuit is inside the recovery grace period")
 	}
+	if prove != nil {
+		if err := prove(now); err != nil {
+			return RecoveryResult{}, err
+		}
+	}
 	result := RecoveryResult{Key: key, EntityID: entityID, ScaleSetID: scaleSetID, ErrorClass: errorClass, ExpectedUpdatedAt: retry.UpdatedAt, PreviousGeneration: state.Generation, Generation: state.Generation, RecoveredAt: now}
 	if !apply {
 		return result, nil
@@ -251,6 +282,47 @@ func RecoverTerminal(ctx context.Context, journalPath, lockPath, key, entityID s
 	return result, nil
 }
 
+func parseExactJobKey(key string) (jobID, entityID string, scaleSetID uint, err error) {
+	domain, jobID, found := strings.Cut(key, ":job:")
+	if !found || jobID == "" || strings.Contains(jobID, ":") {
+		return "", "", 0, errors.New("retry key is not an exact job key")
+	}
+	entityID, scaleSetID, err = parseScaleSetDomainKey(domain)
+	return jobID, entityID, scaleSetID, err
+}
+
+func requireActiveQueueJob(path, jobID string, now time.Time) error {
+	info, err := os.Lstat(path)
+	if err != nil || !info.Mode().IsRegular() || info.Mode().Perm()&0o077 != 0 || info.Size() < 1 || info.Size() > maximumQueueJournalBytes {
+		return errors.New("queue journal must be a bounded private regular file")
+	}
+	data, err := os.ReadFile(path)
+	if err != nil {
+		return fmt.Errorf("read queue journal: %w", err)
+	}
+	var queue struct {
+		SchemaVersion int `json:"schema_version"`
+		Intents       map[string]struct {
+			JobID     string    `json:"job_id"`
+			State     string    `json:"state"`
+			ExpiresAt time.Time `json:"expires_at"`
+		} `json:"intents"`
+	}
+	if err := json.Unmarshal(data, &queue); err != nil || queue.SchemaVersion != 4 || queue.Intents == nil {
+		return errors.New("queue journal identity is invalid")
+	}
+	for _, intent := range queue.Intents {
+		if intent.JobID != jobID || !intent.ExpiresAt.After(now) {
+			continue
+		}
+		switch intent.State {
+		case "queued", "acquiring", "acquired", "assigned":
+			return nil
+		}
+	}
+	return errors.New("exact retry job is not active in the queue journal")
+}
+
 // parseScaleSetDomainKey accepts only the terminal failure domain written by
 // GARM: scale-set:<forge entity UUID>:<scale-set database ID>. Concrete
 // :job/:instance: retry keys are deliberately not recoverable through the

internal/providerretry/recovery_linux_test.go

@@ -5,6 +5,7 @@ package providerretry
 import (
 	"context"
 	"encoding/json"
+	"fmt"
 	"os"
 	"path/filepath"
 	"strings"
@@ -97,6 +98,84 @@ func TestRecoverTerminalRejectsWrongTenantOrConcreteRetryKey(t *testing.T) {
 	}
 }
 
+func TestRecoverExactJobTerminalRequiresLiveQueueProofAndCAS(t *testing.T) {
+	t.Parallel()
+	directory := t.TempDir()
+	journalPath := filepath.Join(directory, "create-retries.json")
+	lockPath := filepath.Join(directory, "create-retries.lock")
+	queuePath := filepath.Join(directory, "queue-intents.json")
+	updatedAt := time.Now().UTC().Add(-time.Hour)
+	key := "scale-set:entity-one:3:job:job-one"
+	state := journal{SchemaVersion: 2, Generation: 19, UpdatedAt: updatedAt, Records: map[string]record{key: {
+		JobID: key, Attempts: 3, LastErrorClass: "provider", UpdatedAt: updatedAt,
+		NextAllowedAt: updatedAt, TerminalUntil: updatedAt.Add(24 * time.Hour),
+	}}, Reservations: map[string]reservation{"runner-existing": {RetryKey: "scale-set:other:2:job:other", UpdatedAt: updatedAt}}}
+	content, _ := json.Marshal(state)
+	if err := os.WriteFile(journalPath, content, 0o600); err != nil {
+		t.Fatal(err)
+	}
+	queue := fmt.Sprintf(`{"schema_version":4,"intents":{"github-scale-set-job:v2:3:job-one":{"job_id":"job-one","state":"assigned","expires_at":%q}}}`,
+		time.Now().UTC().Add(time.Hour).Format(time.RFC3339Nano))
+	if err := os.WriteFile(queuePath, []byte(queue), 0o600); err != nil {
+		t.Fatal(err)
+	}
+	dry, err := RecoverExactJobTerminal(context.Background(), journalPath, lockPath, queuePath, key, "entity-one", 3, "provider", updatedAt, false)
+	if err != nil || dry.Applied || dry.Generation != 19 {
+		t.Fatalf("dry exact recovery=%#v err=%v", dry, err)
+	}
+	applied, err := RecoverExactJobTerminal(context.Background(), journalPath, lockPath, queuePath, key, "entity-one", 3, "provider", updatedAt, true)
+	if err != nil || !applied.Applied || applied.Generation != 20 {
+		t.Fatalf("applied exact recovery=%#v err=%v", applied, err)
+	}
+	observed, err := readJournal(journalPath)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, exists := observed.Records[key]; exists || observed.Generation != 20 {
+		t.Fatalf("exact recovery journal=%#v", observed)
+	}
+}
+
+func TestRecoverExactJobTerminalRejectsMissingOrInvalidQueueProof(t *testing.T) {
+	t.Parallel()
+	for _, test := range []struct {
+		name  string
+		queue string
+	}{
+		{name: "job absent", queue: `{"schema_version":4,"intents":{}}`},
+		{name: "job running", queue: `{"schema_version":4,"intents":{"job":{"job_id":"job-one","state":"running","expires_at":"2099-01-01T00:00:00Z"}}}`},
+		{name: "wrong schema", queue: `{"schema_version":3,"intents":{"job":{"job_id":"job-one","state":"assigned","expires_at":"2099-01-01T00:00:00Z"}}}`},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			directory := t.TempDir()
+			journalPath := filepath.Join(directory, "create-retries.json")
+			lockPath := filepath.Join(directory, "create-retries.lock")
+			queuePath := filepath.Join(directory, "queue-intents.json")
+			updatedAt := time.Now().UTC().Add(-time.Hour)
+			key := "scale-set:entity-one:3:job:job-one"
+			state := journal{SchemaVersion: 2, Generation: 21, UpdatedAt: updatedAt, Records: map[string]record{key: {
+				JobID: key, Attempts: 3, LastErrorClass: "provider", UpdatedAt: updatedAt,
+				NextAllowedAt: updatedAt, TerminalUntil: updatedAt.Add(24 * time.Hour),
+			}}, Reservations: map[string]reservation{"runner-existing": {RetryKey: "scale-set:other:2:job:other", UpdatedAt: updatedAt}}}
+			content, _ := json.Marshal(state)
+			if err := os.WriteFile(journalPath, content, 0o600); err != nil {
+				t.Fatal(err)
+			}
+			if err := os.WriteFile(queuePath, []byte(test.queue), 0o600); err != nil {
+				t.Fatal(err)
+			}
+			before, _ := os.ReadFile(journalPath)
+			if _, err := RecoverExactJobTerminal(context.Background(), journalPath, lockPath, queuePath, key, "entity-one", 3, "provider", updatedAt, true); err == nil {
+				t.Fatal("invalid queue proof recovered exact job retry")
+			}
+			after, _ := os.ReadFile(journalPath)
+			if string(after) != string(before) {
+				t.Fatal("failed exact recovery mutated retry journal")
+			}
+		})
+	}
+}
+
 func TestInspectCountsOnlyTerminalFailureDomains(t *testing.T) {
 	directory := t.TempDir()
 	journalPath := filepath.Join(directory, "create-retries.json")