diff --git a/cmd/gha-fleet/main.go b/cmd/gha-fleet/main.go index 7f51449..125812f 100644 --- a/cmd/gha-fleet/main.go +++ b/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 [options]") + fmt.Fprintln(writer, "usage: gha-fleet [options]") } diff --git a/internal/providerretry/recovery_linux.go b/internal/providerretry/recovery_linux.go index 156301a..66466ad 100644 --- a/internal/providerretry/recovery_linux.go +++ b/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::. Concrete // :job/:instance: retry keys are deliberately not recoverable through the diff --git a/internal/providerretry/recovery_linux_test.go b/internal/providerretry/recovery_linux_test.go index 3925c34..e5900bb 100644 --- a/internal/providerretry/recovery_linux_test.go +++ b/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")