Merge Workflow ENG-05 (Fehlerbehandlung & Retry) in integration/gesamtprojekt

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)

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
+}

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")
+		}
+	}
+}

workflow/migrations/tenant/0004_step_retry.down.sql

@@ -0,0 +1 @@
+DROP TABLE IF EXISTS workflow_step_attempts;

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';