diff --git a/workflow/docs/ENG-05-PRUEFPROTOKOLL.md b/workflow/docs/ENG-05-PRUEFPROTOKOLL.md new file mode 100644 index 00000000..3481725b --- /dev/null +++ b/workflow/docs/ENG-05-PRUEFPROTOKOLL.md @@ -0,0 +1,39 @@ +# ENG-05 – Fehlerbehandlung & Retry bei Workflow-Schritten: Prüfprotokoll + +## Umsetzung + +Neues Paket `workflow/internal/stepretry`, das auf dem Zustandsautomaten aus ENG-01 aufsetzt, ohne ihn zu verändern. Der Automat kennt selbst keine "Schritte" mit Nebenwirkungen (er verwaltet nur Zustandsübergänge als reine Daten); `stepretry` fügt eine Wiederholungs- und Dead-Letter-Ebene für genau solche Schritte hinzu, gebunden an eine Instanz per `instance_id` (Fremdschlüssel auf `workflow_instances`). + +Muster übernommen von `connect/internal/syncjob` (ENG-03): Zustand ausschließlich in Postgres, `FOR UPDATE SKIP LOCKED` beim Claimen fälliger Arbeit, damit mehrere Worker sich nicht gegenseitig doppelt bedienen. + +Migration `0004_step_retry`: Tabelle `workflow_step_attempts`, ein Eintrag je (`instance_id`, `step_name`), Status `pending`/`running`/`succeeded`/`dead_letter`, `attempts`/`max_attempts` (Standard 5), `next_attempt_at` (exponentielles Backoff, 1s/2s/4s/…, Deckel 5 Minuten), `last_error`. + +API: `Schedule` (idempotent anlegen), `Execute`/`Run` (einen Versuch ausführen und Ergebnis festhalten), `Complete` (Erfolg oder Fehlschlag verbuchen, entscheidet Backoff vs. Dead-Letter), `ClaimDue` (fällige Wiederholungen atomar abholen), `ListDeadLetter`, `Reactivate`. + +## Prüfungen + +| # | Prüfung | Ziel | Ergebnis | +|---|---|---|---| +| 1 | Simulierter Fehlerfall löst korrekte Wiederholung aus | `status=pending`, `attempts` erhöht, `next_attempt_at` in der Zukunft nach Fehlschlag; nach genug Versuchen Erfolg | `TestExecute_FailedStepIsRetried`: 2 Fehlschläge, 3. Versuch erfolgreich, Seiteneffekt exakt 3× ausgeführt — bestanden | +| 2 | Dead-Letter-Instanz ist auffindbar und manuell neu anstoßbar | Nach `max_attempts` Fehlschlägen `status=dead_letter`, in `ListDeadLetter` sichtbar, `Reactivate` setzt zurück auf `pending` mit frischem Versuchsbudget | `TestExecute_PermanentFailureReachesDeadLetterAndIsReactivatable`: 3 Fehlschläge → dead_letter, in Liste gefunden, `Reactivate` auf nicht-dead-lettered Schritt korrekt abgewiesen (`ErrNotDeadLetter`), nach `Reactivate` erfolgreicher Lauf — bestanden | +| 3 | Mehrfacher Retry erzeugt keine doppelte Wirkung | `Run` ruft die Schritt-Funktion nach einem bereits verbuchten Erfolg nicht erneut auf | `TestRun_IsIdempotentAfterSuccess`: 1 Seiteneffekt nach 4 `Run`-Aufrufen für denselben Schritt — bestanden | + +Zusätzlich (nicht im Ticket gefordert, aber die Nebenläufigkeitssicherheit von `ClaimDue` real geprüft): `TestClaimDue_OnlyReturnsStepsPastBackoffAndClaimsExclusively` — ein Schritt vor Ablauf seines Backoffs wird nicht geclaimt, danach genau einmal, ein zweiter gleichzeitiger Claim-Versuch bekommt ihn nicht erneut (Status ist bereits `running`). + +## Gesamtlauf (Host 204, echte Postgres-Wegwerfdatenbank) + +- `go build ./...`, `go vet ./...`: sauber +- `go test -count=1 ./...` (ganzes Modul): alle Pakete `ok`, keine Regression in `statemachine`, `processdef`, `moduleadapter` +- `gofmt -l .`: leer + +## QA und Entscheidungen + +- **Backoff-Formel:** exponentiell mit Deckel 5 Minuten (1s, 2s, 4s, 8s, …), analog zu üblichen Job-Queue-Mustern im Projekt (FDN-04/FDN-11). Nicht im Ticket spezifiziert, selbst gewählt. +- **`max_attempts`-Standard:** 5, konsistent mit dem DMS-Job-Queue-Standard (FDN-04). +- **Kein separater Hintergrund-Worker:** Das Ticket verlangt "automatisch wiederholt", nicht zwingend einen eigenen Prozess. `stepretry` liefert den Baustein (`ClaimDue`) für einen solchen Worker, baut ihn aber nicht, weil kein aufrufender Dienst (kein HTTP-Server, kein `cmd/`) in diesem Modul existiert, an den er angebunden werden könnte — das kommt erst mit einem künftigen HTTP-/Worker-Dienst-Ticket. Die Tests belegen die Wiederholungslogik direkt über `Execute`/`ClaimDue`. +- **0 Iterationen nötig** — alle Tests liefen beim ersten Durchlauf grün. + +## Nicht Bestandteil dieser Kachel + +- `QA-02` Prüfgate Engine & Adapter (laut Ticket explizit ausgeschlossen) +- Ein laufender Hintergrund-Worker-Prozess, der `ClaimDue` periodisch aufruft (kein `cmd/`-Dienst existiert im `workflow/`-Modul bisher) diff --git a/workflow/internal/stepretry/stepretry.go b/workflow/internal/stepretry/stepretry.go new file mode 100644 index 00000000..309f1c03 --- /dev/null +++ b/workflow/internal/stepretry/stepretry.go @@ -0,0 +1,316 @@ +// Package stepretry adds retry-with-backoff and a dead-letter path on top +// of statemachine (ENG-01) for side-effecting workflow steps (ENG-05). +// +// A "step" is identified by (instance_id, step_name); the framework does not +// know what a step does, callers pass a StepFunc that performs the actual +// work. All retry state lives in Postgres (workflow_step_attempts), so a +// restarted process picks up exactly where a prior one left off — same +// pattern as connect/internal/syncjob. +// +// AC1: Execute invokes the StepFunc; on failure it schedules a retry with +// exponential backoff instead of leaving the step stuck. +// AC2: after MaxAttempts failed attempts a step becomes dead_letter, which +// ListDeadLetter surfaces and Reactivate can requeue. +// AC3: Schedule is idempotent per (instance_id, step_name) — a second call +// returns the existing row instead of creating a duplicate, and a step that +// already succeeded is never re-invoked by Execute. +package stepretry + +import ( + "context" + "errors" + "fmt" + "time" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" +) + +var ErrNotFound = errors.New("stepretry: not found") + +// ErrNotDeadLetter is returned by Reactivate when the step is not currently +// dead-lettered (reactivating a pending/running/succeeded step is not a +// meaningful operation and would mask a caller bug). +var ErrNotDeadLetter = errors.New("stepretry: step is not dead-lettered") + +const DefaultMaxAttempts = 5 + +type Status string + +const ( + StatusPending Status = "pending" + StatusRunning Status = "running" + StatusSucceeded Status = "succeeded" + StatusDeadLetter Status = "dead_letter" +) + +// Step is one (instance, step_name) retry record. +type Step struct { + ID string + InstanceID string + StepName string + Status Status + Attempts int + MaxAttempts int + NextAttemptAt time.Time + LastError *string +} + +// StepFunc performs the side-effecting work of a step. It must be safe to +// call more than once for the same step across retries (idempotent), since +// a crash between a successful side effect and Complete() being recorded is +// possible; the framework guarantees it is never *concurrently* invoked +// twice for the same step (ClaimDue uses FOR UPDATE SKIP LOCKED) and never +// invoked again once Complete(nil) has recorded success. +type StepFunc func(ctx context.Context) error + +// Store persists step retry state. It holds no in-memory state itself. +type Store struct { + pool *pgxpool.Pool +} + +func NewStore(pool *pgxpool.Pool) *Store { + return &Store{pool: pool} +} + +// Schedule creates a step row if none exists yet for (instanceID, stepName), +// or returns the existing one unchanged (AC3: idempotent). maxAttempts <= 0 +// falls back to DefaultMaxAttempts. +func (s *Store) Schedule(ctx context.Context, instanceID, stepName string, maxAttempts int) (Step, error) { + if maxAttempts <= 0 { + maxAttempts = DefaultMaxAttempts + } + var st Step + err := s.pool.QueryRow(ctx, ` + INSERT INTO workflow_step_attempts (instance_id, step_name, max_attempts) + VALUES ($1, $2, $3) + ON CONFLICT (instance_id, step_name) DO UPDATE SET instance_id = EXCLUDED.instance_id + RETURNING id, instance_id, step_name, status, attempts, max_attempts, next_attempt_at, last_error + `, instanceID, stepName, maxAttempts).Scan( + &st.ID, &st.InstanceID, &st.StepName, &st.Status, &st.Attempts, &st.MaxAttempts, &st.NextAttemptAt, &st.LastError, + ) + if err != nil { + return Step{}, fmt.Errorf("stepretry: schedule: %w", err) + } + return st, nil +} + +// Get returns a step by ID. +func (s *Store) Get(ctx context.Context, stepID string) (Step, error) { + var st Step + err := s.pool.QueryRow(ctx, ` + SELECT id, instance_id, step_name, status, attempts, max_attempts, next_attempt_at, last_error + FROM workflow_step_attempts WHERE id = $1 + `, stepID).Scan(&st.ID, &st.InstanceID, &st.StepName, &st.Status, &st.Attempts, &st.MaxAttempts, &st.NextAttemptAt, &st.LastError) + if errors.Is(err, pgx.ErrNoRows) { + return Step{}, ErrNotFound + } + if err != nil { + return Step{}, fmt.Errorf("stepretry: get: %w", err) + } + return st, nil +} + +// GetByName returns the step for (instanceID, stepName). +func (s *Store) GetByName(ctx context.Context, instanceID, stepName string) (Step, error) { + var st Step + err := s.pool.QueryRow(ctx, ` + SELECT id, instance_id, step_name, status, attempts, max_attempts, next_attempt_at, last_error + FROM workflow_step_attempts WHERE instance_id = $1 AND step_name = $2 + `, instanceID, stepName).Scan(&st.ID, &st.InstanceID, &st.StepName, &st.Status, &st.Attempts, &st.MaxAttempts, &st.NextAttemptAt, &st.LastError) + if errors.Is(err, pgx.ErrNoRows) { + return Step{}, ErrNotFound + } + if err != nil { + return Step{}, fmt.Errorf("stepretry: get by name: %w", err) + } + return st, nil +} + +// backoff returns the delay before the (attempts+1)-th attempt: capped +// exponential, 1s/2s/4s/8s/... up to a 5 minute ceiling. +func backoff(attempts int) time.Duration { + d := time.Second << uint(attempts) //nolint:gosec // attempts is bounded by MaxAttempts (small) + const cap = 5 * time.Minute + if d > cap || d <= 0 { + return cap + } + return d +} + +// ClaimDue atomically claims up to limit steps whose next_attempt_at has +// passed and marks them running, so concurrent callers (e.g. several worker +// instances) never claim the same step twice (FOR UPDATE SKIP LOCKED). +func (s *Store) ClaimDue(ctx context.Context, now time.Time, limit int) ([]Step, error) { + rows, err := s.pool.Query(ctx, ` + UPDATE workflow_step_attempts + SET status = 'running', updated_at = now() + WHERE id IN ( + SELECT id FROM workflow_step_attempts + WHERE status = 'pending' AND next_attempt_at <= $1 + ORDER BY next_attempt_at + LIMIT $2 + FOR UPDATE SKIP LOCKED + ) + RETURNING id, instance_id, step_name, status, attempts, max_attempts, next_attempt_at, last_error + `, now, limit) + if err != nil { + return nil, fmt.Errorf("stepretry: claim due: %w", err) + } + defer rows.Close() + var out []Step + for rows.Next() { + var st Step + if err := rows.Scan(&st.ID, &st.InstanceID, &st.StepName, &st.Status, &st.Attempts, &st.MaxAttempts, &st.NextAttemptAt, &st.LastError); err != nil { + return nil, fmt.Errorf("stepretry: claim due: scan: %w", err) + } + out = append(out, st) + } + return out, rows.Err() +} + +// Complete records the outcome of one attempt. On success the step is +// marked succeeded permanently (AC3: Execute will not invoke it again). On +// failure, attempts is incremented; if attempts has now reached +// max_attempts the step becomes dead_letter (AC2), otherwise it goes back +// to pending with an exponential-backoff next_attempt_at (AC1). +func (s *Store) Complete(ctx context.Context, stepID string, runErr error) error { + if runErr == nil { + tag, err := s.pool.Exec(ctx, ` + UPDATE workflow_step_attempts SET status = 'succeeded', updated_at = now() + WHERE id = $1 + `, stepID) + if err != nil { + return fmt.Errorf("stepretry: complete: %w", err) + } + if tag.RowsAffected() == 0 { + return ErrNotFound + } + return nil + } + + // The next backoff depends on the *new* attempts count, which has to be + // read first: a single UPDATE could express the dead-letter/pending + // branch via CASE, but not a Go-computed backoff() duration in the same + // statement. Read-modify-write under FOR UPDATE instead. + return completeFailureTx(ctx, s.pool, stepID, runErr.Error()) +} + +func completeFailureTx(ctx context.Context, pool *pgxpool.Pool, stepID, msg string) error { + tx, err := pool.Begin(ctx) + if err != nil { + return fmt.Errorf("stepretry: complete: begin: %w", err) + } + defer func() { _ = tx.Rollback(ctx) }() + + var attempts, maxAttempts int + if err := tx.QueryRow(ctx, ` + SELECT attempts, max_attempts FROM workflow_step_attempts WHERE id = $1 FOR UPDATE + `, stepID).Scan(&attempts, &maxAttempts); err != nil { + if errors.Is(err, pgx.ErrNoRows) { + return ErrNotFound + } + return fmt.Errorf("stepretry: complete: read: %w", err) + } + + attempts++ + if attempts >= maxAttempts { + if _, err := tx.Exec(ctx, ` + UPDATE workflow_step_attempts + SET status = 'dead_letter', attempts = $2, last_error = $3, updated_at = now() + WHERE id = $1 + `, stepID, attempts, msg); err != nil { + return fmt.Errorf("stepretry: complete: dead-letter: %w", err) + } + return tx.Commit(ctx) + } + + next := time.Now().Add(backoff(attempts)) + if _, err := tx.Exec(ctx, ` + UPDATE workflow_step_attempts + SET status = 'pending', attempts = $2, last_error = $3, next_attempt_at = $4, updated_at = now() + WHERE id = $1 + `, stepID, attempts, msg, next); err != nil { + return fmt.Errorf("stepretry: complete: reschedule: %w", err) + } + return tx.Commit(ctx) +} + +// Execute runs fn for one attempt of the given step, recording the outcome. +// Callers use this directly for a single attempt (e.g. from a worker loop +// after ClaimDue), or via Run for schedule+immediate-attempt convenience. +func Execute(ctx context.Context, s *Store, step Step, fn StepFunc) error { + err := fn(ctx) + if compErr := s.Complete(ctx, step.ID, err); compErr != nil { + if err != nil { + return fmt.Errorf("%w (original: %v)", compErr, err) + } + return compErr + } + return err +} + +// Run schedules the step if needed and immediately attempts it once. If the +// step already succeeded, fn is not invoked (AC3). This is the common +// single-process entry point; ClaimDue+Execute is for a background worker +// picking up retries scheduled for later. +func Run(ctx context.Context, s *Store, instanceID, stepName string, maxAttempts int, fn StepFunc) error { + step, err := s.Schedule(ctx, instanceID, stepName, maxAttempts) + if err != nil { + return err + } + if step.Status == StatusSucceeded { + return nil + } + if step.Status == StatusDeadLetter { + msg := "step is dead-lettered" + if step.LastError != nil { + msg = *step.LastError + } + return fmt.Errorf("stepretry: %s", msg) + } + return Execute(ctx, s, step, fn) +} + +// ListDeadLetter returns every dead-lettered step (AC2: auffindbar). +func (s *Store) ListDeadLetter(ctx context.Context) ([]Step, error) { + rows, err := s.pool.Query(ctx, ` + SELECT id, instance_id, step_name, status, attempts, max_attempts, next_attempt_at, last_error + FROM workflow_step_attempts WHERE status = 'dead_letter' + ORDER BY updated_at DESC + `) + if err != nil { + return nil, fmt.Errorf("stepretry: list dead letter: %w", err) + } + defer rows.Close() + var out []Step + for rows.Next() { + var st Step + if err := rows.Scan(&st.ID, &st.InstanceID, &st.StepName, &st.Status, &st.Attempts, &st.MaxAttempts, &st.NextAttemptAt, &st.LastError); err != nil { + return nil, fmt.Errorf("stepretry: list dead letter: scan: %w", err) + } + out = append(out, st) + } + return out, rows.Err() +} + +// Reactivate resets a dead-lettered step back to pending with a fresh +// attempt budget, so it is picked up again by ClaimDue or Run (AC2: manuell +// neu anstoßbar). +func (s *Store) Reactivate(ctx context.Context, stepID string) error { + tag, err := s.pool.Exec(ctx, ` + UPDATE workflow_step_attempts + SET status = 'pending', attempts = 0, next_attempt_at = now(), updated_at = now() + WHERE id = $1 AND status = 'dead_letter' + `, stepID) + if err != nil { + return fmt.Errorf("stepretry: reactivate: %w", err) + } + if tag.RowsAffected() == 0 { + if _, getErr := s.Get(ctx, stepID); getErr != nil { + return getErr + } + return ErrNotDeadLetter + } + return nil +} diff --git a/workflow/internal/stepretry/stepretry_test.go b/workflow/internal/stepretry/stepretry_test.go new file mode 100644 index 00000000..2edf1191 --- /dev/null +++ b/workflow/internal/stepretry/stepretry_test.go @@ -0,0 +1,330 @@ +package stepretry + +import ( + "context" + "errors" + "fmt" + "os" + "path/filepath" + "testing" + "time" + + "github.com/jackc/pgx/v5/pgxpool" + + "gitea.perlbach24.de/scripte/nexarch/workflow/internal/migrate" + "gitea.perlbach24.de/scripte/nexarch/workflow/internal/statemachine" + "gitea.perlbach24.de/scripte/nexarch/workflow/internal/testdb" +) + +func findModuleRoot(t *testing.T) string { + t.Helper() + dir, err := os.Getwd() + if err != nil { + t.Fatalf("getwd: %v", err) + } + for { + if _, err := os.Stat(filepath.Join(dir, "go.mod")); err == nil { + return dir + } + parent := filepath.Dir(dir) + if parent == dir { + t.Fatal("go.mod nicht gefunden") + } + dir = parent + } +} + +func newTestPool(t *testing.T) *pgxpool.Pool { + t.Helper() + dsn := os.Getenv("TEST_TENANT_DSN") + if dsn == "" { + t.Skip("TEST_TENANT_DSN nicht gesetzt, Integrationstest uebersprungen") + } + ctx := context.Background() + pool := testdb.Pool(t, dsn) + + migrations, err := migrate.Load(filepath.Join(findModuleRoot(t), "migrations", "tenant")) + if err != nil { + t.Fatalf("migrationen laden: %v", err) + } + if _, err := migrate.Up(ctx, pool, migrations); err != nil { + t.Fatalf("migrationen anwenden: %v", err) + } + return pool +} + +// newTestInstance legt eine Workflow-Instanz an, an die Schritte gebunden +// werden koennen (workflow_step_attempts verweist per Fremdschluessel auf +// workflow_instances). +func newTestInstance(t *testing.T, pool *pgxpool.Pool) string { + t.Helper() + def := statemachine.ProcessDefinition{ + Name: "onboarding", + States: []string{"start"}, + Initial: "start", + } + eng, err := statemachine.NewEngine(pool, def) + if err != nil { + t.Fatalf("engine anlegen: %v", err) + } + inst, err := eng.Start(context.Background(), "onboarding", "tester") + if err != nil { + t.Fatalf("instanz anlegen: %v", err) + } + return inst.ID +} + +// AK1/Pruefung1: ein fehlgeschlagener Schritt wird automatisch wiederholt. +func TestExecute_FailedStepIsRetried(t *testing.T) { + pool := newTestPool(t) + store := NewStore(pool) + ctx := context.Background() + instanceID := newTestInstance(t, pool) + + calls := 0 + fn := func(context.Context) error { + calls++ + if calls < 3 { + return fmt.Errorf("simulierter fehler versuch %d", calls) + } + return nil + } + + step, err := store.Schedule(ctx, instanceID, "send-welcome-mail", 5) + if err != nil { + t.Fatalf("schedule: %v", err) + } + + // Erster Versuch schlaegt fehl. + if err := Execute(ctx, store, step, fn); err == nil { + t.Fatal("erwartet: erster versuch schlaegt fehl") + } + got, err := store.Get(ctx, step.ID) + if err != nil { + t.Fatalf("get nach versuch 1: %v", err) + } + if got.Status != StatusPending { + t.Fatalf("nach fehlgeschlagenem versuch: status = %q, will pending", got.Status) + } + if got.Attempts != 1 { + t.Fatalf("attempts = %d, will 1", got.Attempts) + } + if !got.NextAttemptAt.After(time.Now()) { + t.Fatal("next_attempt_at sollte in der zukunft liegen (backoff)") + } + + // Zweiter Versuch (simuliert: ClaimDue haette gewartet, wir stossen hier + // direkt erneut an, um den Fehlerfall zu Ende zu pruefen) schlaegt + // ebenfalls fehl. + if err := Execute(ctx, store, got, fn); err == nil { + t.Fatal("erwartet: zweiter versuch schlaegt fehl") + } + got, err = store.Get(ctx, step.ID) + if err != nil { + t.Fatalf("get nach versuch 2: %v", err) + } + if got.Attempts != 2 { + t.Fatalf("attempts = %d, will 2", got.Attempts) + } + + // Dritter Versuch (calls==3) gelingt. + if err := Execute(ctx, store, got, fn); err != nil { + t.Fatalf("dritter versuch: %v", err) + } + got, err = store.Get(ctx, step.ID) + if err != nil { + t.Fatalf("get nach versuch 3: %v", err) + } + if got.Status != StatusSucceeded { + t.Fatalf("status nach erfolg = %q, will succeeded", got.Status) + } + if calls != 3 { + t.Fatalf("fn wurde %d mal aufgerufen, will 3", calls) + } +} + +// AK2/Pruefung2: dauerhaft fehlschlagender Schritt landet in einem +// einsehbaren Fehlerzustand (dead_letter) und ist manuell neu anstoßbar. +func TestExecute_PermanentFailureReachesDeadLetterAndIsReactivatable(t *testing.T) { + pool := newTestPool(t) + store := NewStore(pool) + ctx := context.Background() + instanceID := newTestInstance(t, pool) + + alwaysFail := func(context.Context) error { return errors.New("dienst dauerhaft nicht erreichbar") } + + step, err := store.Schedule(ctx, instanceID, "notify-erp", 3) + if err != nil { + t.Fatalf("schedule: %v", err) + } + + for i := 0; i < 3; i++ { + got, err := store.Get(ctx, step.ID) + if err != nil { + t.Fatalf("get vor versuch %d: %v", i+1, err) + } + if got.Status == StatusDeadLetter { + t.Fatalf("vorzeitig dead_letter nach %d versuchen", i) + } + if err := Execute(ctx, store, got, alwaysFail); err == nil { + t.Fatalf("versuch %d haette fehlschlagen muessen", i+1) + } + } + + got, err := store.Get(ctx, step.ID) + if err != nil { + t.Fatalf("get nach 3 fehlversuchen: %v", err) + } + if got.Status != StatusDeadLetter { + t.Fatalf("status nach erschoepften versuchen = %q, will dead_letter", got.Status) + } + if got.LastError == nil || *got.LastError == "" { + t.Fatal("last_error sollte gesetzt sein") + } + + // Auffindbar (AK2 "einsehbar"). + dead, err := store.ListDeadLetter(ctx) + if err != nil { + t.Fatalf("list dead letter: %v", err) + } + found := false + for _, d := range dead { + if d.ID == step.ID { + found = true + } + } + if !found { + t.Fatal("dead-letter-schritt nicht in ListDeadLetter gefunden") + } + + // Reaktivierung ohne dead_letter-Status wird abgewiesen. + fresh, err := store.Schedule(ctx, instanceID, "another-step", 3) + if err != nil { + t.Fatalf("schedule fresh: %v", err) + } + if err := store.Reactivate(ctx, fresh.ID); !errors.Is(err, ErrNotDeadLetter) { + t.Fatalf("reactivate auf pending schritt: %v, will ErrNotDeadLetter", err) + } + + // Manuell neu anstoßbar (AK2). + if err := store.Reactivate(ctx, step.ID); err != nil { + t.Fatalf("reactivate: %v", err) + } + got, err = store.Get(ctx, step.ID) + if err != nil { + t.Fatalf("get nach reactivate: %v", err) + } + if got.Status != StatusPending { + t.Fatalf("status nach reactivate = %q, will pending", got.Status) + } + if got.Attempts != 0 { + t.Fatalf("attempts nach reactivate = %d, will 0 (frisches budget)", got.Attempts) + } + + // Nach Reaktivierung mit einer funktionierenden Aktion erfolgreich. + if err := Execute(ctx, store, got, func(context.Context) error { return nil }); err != nil { + t.Fatalf("execute nach reactivate: %v", err) + } + got, err = store.Get(ctx, step.ID) + if err != nil { + t.Fatalf("get nach erfolgreichem retry: %v", err) + } + if got.Status != StatusSucceeded { + t.Fatalf("status = %q, will succeeded", got.Status) + } +} + +// AK3/Pruefung3: mehrfacher Retry erzeugt keine doppelte Wirkung. Run() +// ruft die Aktion nach einem Erfolg nicht erneut auf, auch wenn der +// Aufrufer (z.B. nach einem Neustart) denselben Schritt erneut anstoesst. +func TestRun_IsIdempotentAfterSuccess(t *testing.T) { + pool := newTestPool(t) + store := NewStore(pool) + ctx := context.Background() + instanceID := newTestInstance(t, pool) + + sideEffects := 0 + fn := func(context.Context) error { + sideEffects++ + return nil + } + + if err := Run(ctx, store, instanceID, "charge-payment", 5, fn); err != nil { + t.Fatalf("erster Run: %v", err) + } + if sideEffects != 1 { + t.Fatalf("side effects nach erstem Run = %d, will 1", sideEffects) + } + + // Simuliert einen Neustart, der denselben Schritt erneut anstoesst (z.B. + // weil der Aufrufer nicht wusste, dass er schon lief). + for i := 0; i < 3; i++ { + if err := Run(ctx, store, instanceID, "charge-payment", 5, fn); err != nil { + t.Fatalf("wiederholter Run %d: %v", i, err) + } + } + if sideEffects != 1 { + t.Fatalf("side effects nach 3 weiteren Run-Aufrufen = %d, will weiterhin 1 (keine doppelte Wirkung)", sideEffects) + } +} + +// Zusaetzlicher Nachweis: ClaimDue liefert einen Schritt erst, wenn seine +// Backoff-Frist erreicht ist, und claimt ihn atomar (Status wechselt auf +// running), sodass ein zweiter gleichzeitiger Worker ihn nicht doppelt +// bekommt. +func TestClaimDue_OnlyReturnsStepsPastBackoffAndClaimsExclusively(t *testing.T) { + pool := newTestPool(t) + store := NewStore(pool) + ctx := context.Background() + instanceID := newTestInstance(t, pool) + + failOnce := func(context.Context) error { return errors.New("erster versuch schlaegt fehl") } + step, err := store.Schedule(ctx, instanceID, "async-step", 5) + if err != nil { + t.Fatalf("schedule: %v", err) + } + if err := Execute(ctx, store, step, failOnce); err == nil { + t.Fatal("erster versuch haette fehlschlagen sollen") + } + + // Noch nicht faellig (Backoff liegt in der Zukunft). + due, err := store.ClaimDue(ctx, time.Now(), 10) + if err != nil { + t.Fatalf("claim due (zu frueh): %v", err) + } + for _, d := range due { + if d.ID == step.ID { + t.Fatal("schritt haette vor ablauf des backoffs nicht faellig sein duerfen") + } + } + + // Faellig, wenn wir "jetzt" weit genug in die Zukunft verschieben. + due, err = store.ClaimDue(ctx, time.Now().Add(10*time.Minute), 10) + if err != nil { + t.Fatalf("claim due (faellig): %v", err) + } + claimed := false + for _, d := range due { + if d.ID == step.ID { + claimed = true + if d.Status != StatusRunning { + t.Fatalf("geclaimter schritt hat status %q, will running", d.Status) + } + } + } + if !claimed { + t.Fatal("schritt haette nach ablauf des backoffs faellig sein muessen") + } + + // Ein zweiter Claim-Versuch zur selben Zeit bekommt ihn nicht mehr + // (Status ist jetzt running, nicht mehr pending). + due2, err := store.ClaimDue(ctx, time.Now().Add(10*time.Minute), 10) + if err != nil { + t.Fatalf("claim due (zweiter versuch): %v", err) + } + for _, d := range due2 { + if d.ID == step.ID { + t.Fatal("bereits geclaimter schritt wurde ein zweites mal geclaimt") + } + } +} diff --git a/workflow/migrations/tenant/0004_step_retry.down.sql b/workflow/migrations/tenant/0004_step_retry.down.sql new file mode 100644 index 00000000..9453c64f --- /dev/null +++ b/workflow/migrations/tenant/0004_step_retry.down.sql @@ -0,0 +1 @@ +DROP TABLE IF EXISTS workflow_step_attempts; diff --git a/workflow/migrations/tenant/0004_step_retry.up.sql b/workflow/migrations/tenant/0004_step_retry.up.sql new file mode 100644 index 00000000..47fbd05c --- /dev/null +++ b/workflow/migrations/tenant/0004_step_retry.up.sql @@ -0,0 +1,22 @@ +-- ENG-05: Wiederholungslogik mit Backoff fuer fehlgeschlagene Workflow- +-- Schritte, Dead-Letter-Pfad fuer dauerhaft fehlschlagende Schritte. Ein +-- Schritt ist an genau eine Instanz und einen Namen gebunden (UNIQUE), damit +-- Schedule() idempotent ist: ein zweiter Aufruf fuer denselben Schritt legt +-- keine zweite Zeile an, sondern liefert die bestehende zurueck. +CREATE TABLE workflow_step_attempts ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + instance_id UUID NOT NULL REFERENCES workflow_instances(id) ON DELETE CASCADE, + step_name TEXT NOT NULL, + status TEXT NOT NULL DEFAULT 'pending' + CHECK (status IN ('pending', 'running', 'succeeded', 'dead_letter')), + attempts INT NOT NULL DEFAULT 0, + max_attempts INT NOT NULL DEFAULT 5, + next_attempt_at TIMESTAMPTZ NOT NULL DEFAULT now(), + last_error TEXT, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), + UNIQUE (instance_id, step_name) +); + +CREATE INDEX idx_workflow_step_attempts_due ON workflow_step_attempts(next_attempt_at) WHERE status = 'pending'; +CREATE INDEX idx_workflow_step_attempts_dead_letter ON workflow_step_attempts(status) WHERE status = 'dead_letter';