docs/CUT-06-PRUEFPROTOKOLL.md

@@ -0,0 +1,176 @@
+# CUT-06 — Prüfprotokoll: Parallelbetrieb read-only & Abschaltung Altsysteme
+
+## Wichtiger Hinweis zur Abweichung vom Ticket (DSGVO-bedingt)
+
+Dieses Ticket setzt laut Vorgabe CUT-04 (realer Stichtag) und CUT-05 (realer Reindex
+danach) voraus. Der reale Stichtag ist eine echte Produktionsoperation gegen die
+echten archivmail-/archivdms-Bestandsdaten. Diese Operation wurde **nicht**
+durchgeführt und **nicht** simuliert — Projektkonstante: niemals echte
+Kunden-/Mandantendaten anfassen (DSGVO, Art. 9).
+
+Stattdessen wurde, wie bereits bei CUT-02 und ROL-02 vom Nutzer ausdrücklich
+autorisiert:
+
+- der Parallelbetrieb-/Abschaltungsmechanismus als Code gebaut (`internal/parallelrun`),
+- aufbauend auf dem bereits gemergten, real DB-seitig durchgesetzten
+  Wartungsmodus-Mechanismus aus `internal/rollbackrun` (ROL-02,
+  `ALTER DATABASE ... SET default_transaction_read_only`),
+- gegen eine **synthetische Testkopie** des Altsystems verifiziert (eigene,
+  mit `cutoverorchestrator.GenerateDataset`/`CreateDatabase` frisch angelegte
+  Wegwerf-Datenbanken auf dem Testcluster von Host 204, keine Verbindung zu
+  `archivmail`/`archivdms`-Produktivdaten).
+
+Der reale Produktivablauf — 90 Tage tatsächlicher Parallelbetrieb gegen die
+echten Altsysteme, echte gemeldete Diskrepanzen im Tagesgeschäft, und die
+echte zweite schriftliche Freigabe durch eine Person mit Zugriff auf das
+Produktivsystem — bleibt ein **separater, späterer operativer Schritt für
+Menschen** mit echtem Produktionszugriff. Dieses Protokoll weist nur die
+**Code-Logik und ihre reale DB-seitige Durchsetzung** gegen die synthetische
+Testkopie nach, nicht den tatsächlich durchlaufenen Parallelbetrieb.
+
+## Umgebung
+
+- Arbeitsverzeichnis: `/root/cut-06-work-1792091596/nexarch` auf Host 204
+  (`root@192.168.1.204`), frisch geklont von
+  `https://gitea.perlbach24.de/scripte/nexarch.git`, Branch
+  `feature/cut-06-parallelbetrieb-abschaltung`.
+- Testdatenbank: echtes PostgreSQL auf Host 204 über `TEST_ADMIN_DSN`
+  (aus `/etc/nexarch-dev/core-dev.env`), kein Mocking der DB.
+- Implementiertes Paket: `internal/parallelrun` (root-Modul
+  `gitea.perlbach24.de/scripte/nexarch`).
+- Migration: `migrations/0028_parallel_run_period.up.sql` /
+  `.down.sql` (Tabellen `parallel_run_period`, `parallel_run_discrepancy`,
+  `parallel_run_shutdown_approval`).
+- Wiederverwendung statt Duplikat: `internal/rollbackrun.SetReadOnly` neu
+  exportiert (aus der bisher inline in `Setup`/`Rollback` verwendeten
+  `ALTER DATABASE ... SET default_transaction_read_only`-Anweisung
+  herausgezogen) und von `parallelrun.Store.StartPeriod` aufgerufen.
+
+## Architektur in Kürze
+
+- `parallelrun.Store.StartPeriod`: versetzt das Altsystem real in den
+  Wartungsmodus (schreibgeschützt über `rollbackrun.SetReadOnly`) und
+  persistiert Zeitraum + Fristende (`cutover_at` + Dauer, Standardvorschlag
+  90 Tage).
+- `parallelrun.Store.ReportDiscrepancy` / `ResolveDiscrepancy`: Diskrepanzen
+  mit Status `open` / `resolved` / `deferred`. Sowohl `resolved` als auch
+  `deferred` zählen laut Ticket als "erledigt" — nur `open` blockiert.
+- `parallelrun.Store.ApproveShutdown`: persistiert die zweite, eigenständige
+  schriftliche Freigabe (Zeitstempel + freigebende Person + Notiz), aber nur
+  wenn die Frist real abgelaufen ist UND keine offenen Diskrepanzen mehr
+  vorliegen.
+- `parallelrun.Shutdown`: sperrt die Altsystem-Testdatenbank vollständig
+  (`ALTER DATABASE ... WITH ALLOW_CONNECTIONS false` + Terminieren
+  bestehender Backends) — weder Lese- noch Schreibzugriff mehr möglich.
+  Verweigert real sowohl bei nicht abgelaufener Frist als auch bei
+  fehlender zweiter Freigabe, unabhängig von `ApproveShutdown`s eigener
+  Prüfreihenfolge.
+
+## Prüfungen vor Abnahme
+
+### Prüfung 1 — Schreibzugriff auf Altsystem während Parallelbetrieb wird nachweislich abgewiesen
+
+- **Zielwert:** Nach `StartPeriod` ist die (synthetische) Altsystem-Testdatenbank
+  lesend erreichbar, ein Schreibversuch (`INSERT`) scheitert real an der
+  Postgres-Ebene.
+- **Test:** `TestStartPeriod_LegacySystemReadOnlyNotWritable`
+  (`internal/parallelrun/parallelrun_test.go`).
+- **Ergebnis:** PASS. Lesezugriff (`SELECT count(*) FROM legacy_writeprobe`)
+  funktioniert, Schreibzugriff (`INSERT INTO legacy_writeprobe`) schlägt mit
+  Postgres-Fehler fehl (`default_transaction_read_only = on`,
+  SQLSTATE 25006).
+
+### Prüfung 2 — Zweite Freigabe zur Abschaltung liegt schriftlich vor Abschaltungstermin vor
+
+- **Zielwert:** Abschaltung (`Shutdown`) wird real verweigert, solange die
+  Frist nicht abgelaufen ist oder keine zweite schriftliche Freigabe
+  (`ApproveShutdown`) vorliegt; erst mit beidem sperrt `Shutdown` das
+  Altsystem vollständig.
+- **Tests:**
+  - `TestShutdown_RefusedBeforeDeadline`: Freigabeversuch und
+    Abschaltungsversuch vor Fristende werden beide mit
+    `ErrDeadlineNotReached` abgewiesen; Altsystem bleibt danach weiter
+    lesbar.
+  - `TestApproveShutdown_RefusedWithOpenDiscrepancies`: Freigabe wird mit
+    `ErrOpenDiscrepancies` abgewiesen, solange eine offene Diskrepanz
+    vorliegt; Abschaltungsversuch ohne vorliegende Freigabe wird mit
+    `ErrNoApproval` abgewiesen; nach Zurückstellen der Diskrepanz
+    (`ResolveDiscrepancy` mit Status `deferred`) gelingt die Freigabe.
+  - `TestShutdown_LocksSystemCompletelyAfterApproval`: nach Fristablauf und
+    vorliegender Freigabe sperrt `Shutdown` die Testdatenbank vollständig —
+    ein anschließender Verbindungsversuch (inkl. `Ping`) schlägt fehl.
+- **Ergebnis:** alle drei Tests PASS.
+
+### Akzeptanzkriterium 3 — gemeldete Diskrepanzen bis zur Abschaltung bearbeitet oder bewusst zurückgestellt
+
+- Abgedeckt durch `TestApproveShutdown_RefusedWithOpenDiscrepancies` (siehe
+  oben): `ApproveShutdown` prüft über `OpenDiscrepancyCount`, dass keine
+  Diskrepanz mehr im Status `open` ist; `resolved` und `deferred` zählen
+  beide als erledigt.
+
+## Testlauf (echte Postgres-Testdatenbank, Host 204)
+
+```
+=== RUN   TestStartPeriod_LegacySystemReadOnlyNotWritable
+--- PASS: TestStartPeriod_LegacySystemReadOnlyNotWritable (0.59s)
+=== RUN   TestShutdown_RefusedBeforeDeadline
+--- PASS: TestShutdown_RefusedBeforeDeadline (0.50s)
+=== RUN   TestApproveShutdown_RefusedWithOpenDiscrepancies
+--- PASS: TestApproveShutdown_RefusedWithOpenDiscrepancies (0.57s)
+=== RUN   TestShutdown_LocksSystemCompletelyAfterApproval
+--- PASS: TestShutdown_LocksSystemCompletelyAfterApproval (0.53s)
+PASS
+ok  	gitea.perlbach24.de/scripte/nexarch/internal/parallelrun	2.195s
+```
+
+Regressionslauf des wiederverwendeten Pakets (unverändertes Verhalten,
+nur neue exportierte Hilfsfunktion `SetReadOnly` hinzugefügt):
+
+```
+ok  	gitea.perlbach24.de/scripte/nexarch/internal/rollbackrun	4.440s
+ok  	gitea.perlbach24.de/scripte/nexarch/internal/parallelrun	4.402s
+```
+
+`go build ./...` lief fehlerfrei durch.
+
+## Fazit
+
+Die Code-seitige Durchsetzung von Akzeptanzkriterium 1–3 ist gegen die
+synthetische Testkopie real nachgewiesen: Lesezugriff bleibt während der
+Frist möglich, Schreibzugriff wird real verweigert, Abschaltung erfordert
+real sowohl Fristablauf als auch zweite schriftliche Freigabe und erst dann
+auch tatsächlich jeden Zugriff sperrt. Der reale 90-Tage-Parallelbetrieb
+gegen die echten Altsysteme sowie die echte zweite Freigabe bleiben, wie
+oben beschrieben, ein separater operativer Schritt für Menschen mit echtem
+Produktionszugriff.
+
+## Nachtrag — Korrektur nach unabhängiger Verifikation
+
+Eine unabhängige Verifikation hat einen echten Lücke gefunden: `Shutdown()`
+prüfte zwar eigenständig Fristende und Existenz der zweiten Freigabe, aber
+NICHT erneut auf offene Diskrepanzen. Szenario: `ApproveShutdown` erfolgt
+korrekt ohne offene Diskrepanzen, danach wird eine neue Diskrepanz gemeldet
+(`ReportDiscrepancy`, Status `open`) — `Shutdown()` hätte das ignoriert und
+trotzdem abgeschaltet, obwohl Akzeptanzkriterium 3 genau das verhindern soll.
+
+Behoben: `Shutdown()` prüft jetzt zusätzlich `OpenDiscrepancyCount` und
+verweigert mit `ErrOpenDiscrepancies`, falls nach der Freigabe noch eine
+offene Diskrepanz gemeldet wurde. Neuer Test
+`TestShutdown_RefusedWhenDiscrepancyReportedAfterApproval` bildet genau
+dieses Szenario nach und bestätigt die Ablehnung.
+
+Erneuter Testlauf nach dem Fix (echte Postgres-DB, Host 204):
+
+```
+--- PASS: TestStartPeriod_LegacySystemReadOnlyNotWritable (0.72s)
+--- PASS: TestShutdown_RefusedBeforeDeadline (0.69s)
+--- PASS: TestApproveShutdown_RefusedWithOpenDiscrepancies (0.77s)
+--- PASS: TestShutdown_LocksSystemCompletelyAfterApproval (0.74s)
+--- PASS: TestShutdown_RefusedWhenDiscrepancyReportedAfterApproval (0.56s)
+ok  	gitea.perlbach24.de/scripte/nexarch/internal/parallelrun	3.479s
+ok  	gitea.perlbach24.de/scripte/nexarch/internal/rollbackrun	2.873s
+```
+
+`go build ./...` und `gofmt -l` auf den geänderten Dateien weiterhin sauber.
+
+**Bekannte, akzeptierte Einschränkung:** Zwischen der `OpenDiscrepancyCount`-Prüfung und dem anschließenden `ALTER DATABASE`-Sperren in `Shutdown()` liegt keine Transaktion/Sperre — eine Diskrepanz, die exakt in diesem schmalen Zeitfenster gemeldet wird, würde die Prüfung theoretisch passieren lassen (TOCTOU). Für den hier modellierten Anwendungsfall (seltene, manuelle Admin-Aktion, kein Hochfrequenz-Produktivbetrieb) als akzeptables Restrisiko bewertet, nicht weiter gehärtet.

internal/parallelrun/parallelrun.go

@@ -0,0 +1,238 @@
+package parallelrun
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"time"
+
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgxpool"
+
+	"gitea.perlbach24.de/scripte/nexarch/internal/cutoverorchestrator"
+	"gitea.perlbach24.de/scripte/nexarch/internal/rollbackrun"
+)
+
+var (
+	ErrNoApproval         = errors.New("parallelrun: keine zweite freigabe zur abschaltung vorhanden")
+	ErrDeadlineNotReached = errors.New("parallelrun: parallelbetriebsfrist noch nicht abgelaufen")
+	ErrOpenDiscrepancies  = errors.New("parallelrun: noch unbearbeitete diskrepanzen offen")
+	ErrUnknownStatus      = errors.New("parallelrun: unbekannter diskrepanz-status")
+	ErrPeriodNotFound     = errors.New("parallelrun: parallelbetriebsfrist nicht gefunden")
+)
+
+type Period struct {
+	ID        int64
+	TenantID  int64
+	DBName    string
+	CutoverAt time.Time
+	Deadline  time.Time
+	CreatedAt time.Time
+}
+
+type Discrepancy struct {
+	ID           int64
+	PeriodID     int64
+	ReportedAt   time.Time
+	Description  string
+	Status       string
+	ResolvedAt   *time.Time
+	ResolvedNote string
+}
+
+type ShutdownApproval struct {
+	ID         int64
+	PeriodID   int64
+	ApprovedBy string
+	ApprovedAt time.Time
+	Note       string
+}
+
+const (
+	StatusOpen     = "open"
+	StatusResolved = "resolved"
+	StatusDeferred = "deferred"
+)
+
+type Store struct {
+	pool *pgxpool.Pool
+}
+
+func NewStore(pool *pgxpool.Pool) *Store {
+	return &Store{pool: pool}
+}
+
+// StartPeriod versetzt das Altsystem real in den Wartungsmodus
+// (schreibgeschuetzt, siehe rollbackrun.SetReadOnly) und persistiert die
+// Parallelbetriebsfrist mit Fristende.
+func (s *Store) StartPeriod(ctx context.Context, cluster *cutoverorchestrator.Cluster, tenantID int64, dbName string, cutoverAt time.Time, duration time.Duration) (int64, error) {
+	if duration <= 0 {
+		return 0, errors.New("parallelrun: frist muss positiv sein")
+	}
+	if err := rollbackrun.SetReadOnly(ctx, cluster, dbName, true); err != nil {
+		return 0, fmt.Errorf("altsystem in wartungsmodus versetzen: %w", err)
+	}
+	deadline := cutoverAt.Add(duration)
+	var id int64
+	err := s.pool.QueryRow(ctx, `
+		INSERT INTO parallel_run_period (tenant_id, db_name, cutover_at, deadline)
+		VALUES ($1, $2, $3, $4)
+		RETURNING id
+	`, tenantID, dbName, cutoverAt, deadline).Scan(&id)
+	if err != nil {
+		return 0, fmt.Errorf("parallelbetriebsfrist speichern: %w", err)
+	}
+	return id, nil
+}
+
+func (s *Store) Period(ctx context.Context, periodID int64) (Period, error) {
+	var p Period
+	err := s.pool.QueryRow(ctx, `
+		SELECT id, tenant_id, db_name, cutover_at, deadline, created_at
+		FROM parallel_run_period WHERE id = $1
+	`, periodID).Scan(&p.ID, &p.TenantID, &p.DBName, &p.CutoverAt, &p.Deadline, &p.CreatedAt)
+	if errors.Is(err, pgx.ErrNoRows) {
+		return Period{}, ErrPeriodNotFound
+	}
+	if err != nil {
+		return Period{}, fmt.Errorf("parallelbetriebsfrist lesen: %w", err)
+	}
+	return p, nil
+}
+
+func (s *Store) ReportDiscrepancy(ctx context.Context, periodID int64, description string) (int64, error) {
+	if description == "" {
+		return 0, errors.New("parallelrun: beschreibung darf nicht leer sein")
+	}
+	var id int64
+	err := s.pool.QueryRow(ctx, `
+		INSERT INTO parallel_run_discrepancy (period_id, description)
+		VALUES ($1, $2)
+		RETURNING id
+	`, periodID, description).Scan(&id)
+	if err != nil {
+		return 0, fmt.Errorf("diskrepanz melden: %w", err)
+	}
+	return id, nil
+}
+
+// ResolveDiscrepancy markiert eine Diskrepanz als bearbeitet oder bewusst
+// zurueckgestellt; beides zaehlt fuer Akzeptanzkriterium 3 als erledigt.
+func (s *Store) ResolveDiscrepancy(ctx context.Context, discrepancyID int64, status, note string) error {
+	if status != StatusResolved && status != StatusDeferred {
+		return ErrUnknownStatus
+	}
+	tag, err := s.pool.Exec(ctx, `
+		UPDATE parallel_run_discrepancy
+		SET status = $2, resolved_at = now(), resolved_note = $3
+		WHERE id = $1
+	`, discrepancyID, status, note)
+	if err != nil {
+		return fmt.Errorf("diskrepanz aktualisieren: %w", err)
+	}
+	if tag.RowsAffected() == 0 {
+		return errors.New("parallelrun: diskrepanz nicht gefunden")
+	}
+	return nil
+}
+
+func (s *Store) OpenDiscrepancyCount(ctx context.Context, periodID int64) (int, error) {
+	var n int
+	err := s.pool.QueryRow(ctx, `
+		SELECT count(*) FROM parallel_run_discrepancy WHERE period_id = $1 AND status = $2
+	`, periodID, StatusOpen).Scan(&n)
+	if err != nil {
+		return 0, fmt.Errorf("offene diskrepanzen zaehlen: %w", err)
+	}
+	return n, nil
+}
+
+// ApproveShutdown speichert die zweite, eigenstaendige schriftliche Freigabe
+// zur Abschaltung. Sie ist real nur moeglich, wenn die Frist abgelaufen ist
+// und keine unbearbeiteten Diskrepanzen mehr offen sind (Akzeptanzkriterium 2/3).
+func (s *Store) ApproveShutdown(ctx context.Context, periodID int64, approvedBy, note string, now time.Time) (int64, error) {
+	if approvedBy == "" {
+		return 0, errors.New("parallelrun: approvedBy darf nicht leer sein")
+	}
+	period, err := s.Period(ctx, periodID)
+	if err != nil {
+		return 0, err
+	}
+	if now.Before(period.Deadline) {
+		return 0, ErrDeadlineNotReached
+	}
+	open, err := s.OpenDiscrepancyCount(ctx, periodID)
+	if err != nil {
+		return 0, err
+	}
+	if open > 0 {
+		return 0, ErrOpenDiscrepancies
+	}
+	var id int64
+	err = s.pool.QueryRow(ctx, `
+		INSERT INTO parallel_run_shutdown_approval (period_id, approved_by, note)
+		VALUES ($1, $2, $3)
+		RETURNING id
+	`, periodID, approvedBy, note).Scan(&id)
+	if err != nil {
+		return 0, fmt.Errorf("abschaltungsfreigabe speichern: %w", err)
+	}
+	return id, nil
+}
+
+func (s *Store) LatestShutdownApproval(ctx context.Context, periodID int64) (ShutdownApproval, error) {
+	var a ShutdownApproval
+	err := s.pool.QueryRow(ctx, `
+		SELECT id, period_id, approved_by, approved_at, coalesce(note, '')
+		FROM parallel_run_shutdown_approval
+		WHERE period_id = $1
+		ORDER BY approved_at DESC, id DESC
+		LIMIT 1
+	`, periodID).Scan(&a.ID, &a.PeriodID, &a.ApprovedBy, &a.ApprovedAt, &a.Note)
+	if errors.Is(err, pgx.ErrNoRows) {
+		return ShutdownApproval{}, ErrNoApproval
+	}
+	if err != nil {
+		return ShutdownApproval{}, fmt.Errorf("abschaltungsfreigabe lesen: %w", err)
+	}
+	return a, nil
+}
+
+// Shutdown sperrt das Altsystem endgueltig (weder Lese- noch Schreibzugriff
+// mehr moeglich) und verweigert dies real, solange Frist oder zweite
+// Freigabe fehlen -- unabhaengig davon, ob ApproveShutdown diese Reihenfolge
+// bereits erzwungen hat.
+func Shutdown(ctx context.Context, store *Store, cluster *cutoverorchestrator.Cluster, periodID int64, now time.Time) error {
+	period, err := store.Period(ctx, periodID)
+	if err != nil {
+		return err
+	}
+	if now.Before(period.Deadline) {
+		return ErrDeadlineNotReached
+	}
+	if _, err := store.LatestShutdownApproval(ctx, periodID); err != nil {
+		return err
+	}
+	open, err := store.OpenDiscrepancyCount(ctx, periodID)
+	if err != nil {
+		return err
+	}
+	if open > 0 {
+		return ErrOpenDiscrepancies
+	}
+
+	conn, err := pgx.Connect(ctx, cluster.AdminDSN())
+	if err != nil {
+		return fmt.Errorf("verbindung fuer abschaltung: %w", err)
+	}
+	defer conn.Close(ctx)
+
+	ident := pgx.Identifier{period.DBName}.Sanitize()
+	if _, err := conn.Exec(ctx, "ALTER DATABASE "+ident+" WITH ALLOW_CONNECTIONS false"); err != nil {
+		return fmt.Errorf("verbindungen sperren: %w", err)
+	}
+	if _, err := conn.Exec(ctx, `SELECT pg_terminate_backend(pid) FROM pg_stat_activity WHERE datname = $1`, period.DBName); err != nil {
+		return fmt.Errorf("bestehende verbindungen beenden: %w", err)
+	}
+	return nil
+}

internal/parallelrun/parallelrun_test.go

@@ -0,0 +1,280 @@
+package parallelrun
+
+import (
+	"context"
+	"errors"
+	"os"
+	"testing"
+	"time"
+
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgxpool"
+
+	"gitea.perlbach24.de/scripte/nexarch/internal/cutoverorchestrator"
+	"gitea.perlbach24.de/scripte/nexarch/internal/testdb"
+)
+
+const testTenantID int64 = 202
+
+func testCluster(t *testing.T) *cutoverorchestrator.Cluster {
+	t.Helper()
+	adminDSN := os.Getenv("TEST_ADMIN_DSN")
+	if adminDSN == "" {
+		t.Skip("TEST_ADMIN_DSN nicht gesetzt, Integrationstest uebersprungen")
+	}
+	c, err := cutoverorchestrator.NewCluster(adminDSN)
+	if err != nil {
+		t.Fatalf("NewCluster: %v", err)
+	}
+	return c
+}
+
+func setupSchema(t *testing.T, ctx context.Context, pool *pgxpool.Pool) {
+	t.Helper()
+	if _, err := pool.Exec(ctx, `
+		CREATE TABLE parallel_run_period (
+			id          BIGSERIAL PRIMARY KEY,
+			tenant_id   BIGINT NOT NULL,
+			db_name     TEXT NOT NULL,
+			cutover_at  TIMESTAMPTZ NOT NULL,
+			deadline    TIMESTAMPTZ NOT NULL CHECK (deadline > cutover_at),
+			created_at  TIMESTAMPTZ NOT NULL DEFAULT now()
+		);
+		CREATE TABLE parallel_run_discrepancy (
+			id            BIGSERIAL PRIMARY KEY,
+			period_id     BIGINT NOT NULL REFERENCES parallel_run_period(id),
+			reported_at   TIMESTAMPTZ NOT NULL DEFAULT now(),
+			description   TEXT NOT NULL,
+			status        TEXT NOT NULL DEFAULT 'open' CHECK (status IN ('open', 'resolved', 'deferred')),
+			resolved_at   TIMESTAMPTZ,
+			resolved_note TEXT
+		);
+		CREATE TABLE parallel_run_shutdown_approval (
+			id          BIGSERIAL PRIMARY KEY,
+			period_id   BIGINT NOT NULL REFERENCES parallel_run_period(id),
+			approved_by TEXT NOT NULL,
+			approved_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+			note        TEXT
+		);
+	`); err != nil {
+		t.Fatalf("schema: %v", err)
+	}
+}
+
+func testStore(t *testing.T) *Store {
+	t.Helper()
+	adminDSN := os.Getenv("TEST_ADMIN_DSN")
+	if adminDSN == "" {
+		t.Skip("TEST_ADMIN_DSN nicht gesetzt, Integrationstest uebersprungen")
+	}
+	pool := testdb.Pool(t, adminDSN)
+	setupSchema(t, context.Background(), pool)
+	return NewStore(pool)
+}
+
+func setupLegacyDB(t *testing.T, cluster *cutoverorchestrator.Cluster) string {
+	t.Helper()
+	ctx := context.Background()
+	dbName := "cut06_" + testdb.Suffix()
+	if err := cluster.CreateDatabase(ctx, dbName); err != nil {
+		t.Fatalf("CreateDatabase: %v", err)
+	}
+	t.Cleanup(func() {
+		conn, err := pgx.Connect(context.Background(), cluster.AdminDSN())
+		if err != nil {
+			t.Logf("aufraeumen %s: verbindung fehlgeschlagen: %v", dbName, err)
+			return
+		}
+		defer conn.Close(context.Background())
+		_, _ = conn.Exec(context.Background(), "ALTER DATABASE "+pgx.Identifier{dbName}.Sanitize()+" WITH ALLOW_CONNECTIONS true")
+		_, _ = conn.Exec(context.Background(), `SELECT pg_terminate_backend(pid) FROM pg_stat_activity WHERE datname = $1`, dbName)
+		_, err = conn.Exec(context.Background(), "DROP DATABASE IF EXISTS "+pgx.Identifier{dbName}.Sanitize()+" WITH (FORCE)")
+		if err != nil {
+			t.Logf("aufraeumen %s fehlgeschlagen: %v", dbName, err)
+		}
+	})
+	pool, err := cluster.Connect(ctx, dbName)
+	if err != nil {
+		t.Fatalf("Connect: %v", err)
+	}
+	if _, err := pool.Exec(ctx, `CREATE TABLE legacy_writeprobe (id BIGSERIAL PRIMARY KEY, note TEXT NOT NULL)`); err != nil {
+		pool.Close()
+		t.Fatalf("legacy_writeprobe anlegen: %v", err)
+	}
+	pool.Close()
+	return dbName
+}
+
+func connectLegacyDB(ctx context.Context, cluster *cutoverorchestrator.Cluster, dbName string) (*pgxpool.Pool, error) {
+	pool, err := cluster.Connect(ctx, dbName)
+	if err != nil {
+		return nil, err
+	}
+	if err := pool.Ping(ctx); err != nil {
+		pool.Close()
+		return nil, err
+	}
+	return pool, nil
+}
+
+// AK1 (Pruefung 1): Schreibzugriff auf das Altsystem waehrend des
+// Parallelbetriebs wird nachweislich abgewiesen, Lesezugriff bleibt moeglich.
+func TestStartPeriod_LegacySystemReadOnlyNotWritable(t *testing.T) {
+	cluster := testCluster(t)
+	store := testStore(t)
+	ctx := context.Background()
+	dbName := setupLegacyDB(t, cluster)
+
+	cutoverAt := time.Now().Add(-time.Hour)
+	_, err := store.StartPeriod(ctx, cluster, testTenantID, dbName, cutoverAt, 90*24*time.Hour)
+	if err != nil {
+		t.Fatalf("StartPeriod: %v", err)
+	}
+
+	pool, err := connectLegacyDB(ctx, cluster, dbName)
+	if err != nil {
+		t.Fatalf("Connect nach StartPeriod: %v", err)
+	}
+	defer pool.Close()
+
+	var n int
+	if err := pool.QueryRow(ctx, `SELECT count(*) FROM legacy_writeprobe`).Scan(&n); err != nil {
+		t.Fatalf("lesezugriff haette funktionieren muessen: %v", err)
+	}
+
+	_, err = pool.Exec(ctx, `INSERT INTO legacy_writeprobe (note) VALUES ('sollte scheitern')`)
+	if err == nil {
+		t.Fatal("schreibzugriff waehrend parallelbetrieb haette abgewiesen werden muessen")
+	}
+}
+
+// AK2 (Pruefung 2): Abschaltung vor Fristende wird real verweigert, auch
+// mit einer vorliegenden (zu frueh erteilten) Freigabe nicht moeglich.
+func TestShutdown_RefusedBeforeDeadline(t *testing.T) {
+	cluster := testCluster(t)
+	store := testStore(t)
+	ctx := context.Background()
+	dbName := setupLegacyDB(t, cluster)
+
+	cutoverAt := time.Now()
+	periodID, err := store.StartPeriod(ctx, cluster, testTenantID, dbName, cutoverAt, 90*24*time.Hour)
+	if err != nil {
+		t.Fatalf("StartPeriod: %v", err)
+	}
+
+	_, err = store.ApproveShutdown(ctx, periodID, "qa-leitung", "vorzeitiger versuch", time.Now())
+	if !errors.Is(err, ErrDeadlineNotReached) {
+		t.Fatalf("erwarte ErrDeadlineNotReached bei freigabe vor fristende, habe %v", err)
+	}
+
+	err = Shutdown(ctx, store, cluster, periodID, time.Now())
+	if !errors.Is(err, ErrDeadlineNotReached) {
+		t.Fatalf("erwarte ErrDeadlineNotReached bei abschaltung vor fristende, habe %v", err)
+	}
+
+	pool, err := connectLegacyDB(ctx, cluster, dbName)
+	if err != nil {
+		t.Fatalf("altsystem haette nach verweigerter abschaltung weiter lesbar sein muessen: %v", err)
+	}
+	pool.Close()
+}
+
+// AK2/AK3 (Pruefung 2): Abschaltung nach Fristende wird ohne zweite
+// Freigabe verweigert, und solange offene (unbearbeitete) Diskrepanzen
+// vorliegen, ist die zweite Freigabe selbst nicht moeglich.
+func TestApproveShutdown_RefusedWithOpenDiscrepancies(t *testing.T) {
+	cluster := testCluster(t)
+	store := testStore(t)
+	ctx := context.Background()
+	dbName := setupLegacyDB(t, cluster)
+
+	cutoverAt := time.Now().Add(-100 * 24 * time.Hour)
+	periodID, err := store.StartPeriod(ctx, cluster, testTenantID, dbName, cutoverAt, 90*24*time.Hour)
+	if err != nil {
+		t.Fatalf("StartPeriod: %v", err)
+	}
+
+	discrepancyID, err := store.ReportDiscrepancy(ctx, periodID, "zaehldifferenz tenant 202 in migrationsbericht")
+	if err != nil {
+		t.Fatalf("ReportDiscrepancy: %v", err)
+	}
+
+	after := time.Now()
+	_, err = store.ApproveShutdown(ctx, periodID, "qa-leitung", "abschlussfreigabe", after)
+	if !errors.Is(err, ErrOpenDiscrepancies) {
+		t.Fatalf("erwarte ErrOpenDiscrepancies bei offener diskrepanz, habe %v", err)
+	}
+
+	err = Shutdown(ctx, store, cluster, periodID, after)
+	if !errors.Is(err, ErrNoApproval) {
+		t.Fatalf("erwarte ErrNoApproval ohne vorliegende freigabe, habe %v", err)
+	}
+
+	if err := store.ResolveDiscrepancy(ctx, discrepancyID, StatusDeferred, "bewusst zurueckgestellt, kein datenverlust"); err != nil {
+		t.Fatalf("ResolveDiscrepancy: %v", err)
+	}
+
+	approvalID, err := store.ApproveShutdown(ctx, periodID, "qa-leitung", "abschlussfreigabe nach rueckstellung", after)
+	if err != nil {
+		t.Fatalf("ApproveShutdown nach rueckstellung: %v", err)
+	}
+	if approvalID == 0 {
+		t.Fatal("erwarte gueltige freigabe-id")
+	}
+}
+
+// AK2 (Pruefung 2): Mit Fristende und schriftlich vorliegender zweiter
+// Freigabe sperrt Shutdown das Altsystem vollstaendig -- weder Lese- noch
+// Schreibzugriff mehr moeglich.
+func TestShutdown_LocksSystemCompletelyAfterApproval(t *testing.T) {
+	cluster := testCluster(t)
+	store := testStore(t)
+	ctx := context.Background()
+	dbName := setupLegacyDB(t, cluster)
+
+	cutoverAt := time.Now().Add(-100 * 24 * time.Hour)
+	periodID, err := store.StartPeriod(ctx, cluster, testTenantID, dbName, cutoverAt, 90*24*time.Hour)
+	if err != nil {
+		t.Fatalf("StartPeriod: %v", err)
+	}
+
+	after := time.Now()
+	if _, err := store.ApproveShutdown(ctx, periodID, "qa-leitung", "abschlussfreigabe", after); err != nil {
+		t.Fatalf("ApproveShutdown: %v", err)
+	}
+
+	if err := Shutdown(ctx, store, cluster, periodID, after); err != nil {
+		t.Fatalf("Shutdown: %v", err)
+	}
+
+	_, err = connectLegacyDB(ctx, cluster, dbName)
+	if err == nil {
+		t.Fatal("verbindung nach abschaltung haette fehlschlagen muessen")
+	}
+}
+
+func TestShutdown_RefusedWhenDiscrepancyReportedAfterApproval(t *testing.T) {
+	cluster := testCluster(t)
+	store := testStore(t)
+	ctx := context.Background()
+	dbName := setupLegacyDB(t, cluster)
+
+	cutoverAt := time.Now().Add(-100 * 24 * time.Hour)
+	periodID, err := store.StartPeriod(ctx, cluster, testTenantID, dbName, cutoverAt, 90*24*time.Hour)
+	if err != nil {
+		t.Fatalf("StartPeriod: %v", err)
+	}
+
+	after := time.Now()
+	if _, err := store.ApproveShutdown(ctx, periodID, "qa-leitung", "abschlussfreigabe", after); err != nil {
+		t.Fatalf("ApproveShutdown: %v", err)
+	}
+
+	if _, err := store.ReportDiscrepancy(ctx, periodID, "nach freigabe gemeldet"); err != nil {
+		t.Fatalf("ReportDiscrepancy: %v", err)
+	}
+
+	if err := Shutdown(ctx, store, cluster, periodID, after); !errors.Is(err, ErrOpenDiscrepancies) {
+		t.Fatalf("Shutdown nach einer erst nach der Freigabe gemeldeten Diskrepanz haette ErrOpenDiscrepancies liefern muessen, habe %v", err)
+	}
+}

internal/rollbackrun/rollbackrun.go

@@ -74,6 +74,25 @@ func Setup(ctx context.Context, cluster *cutoverorchestrator.Cluster, dbName str
 	return &Testbed{DBName: dbName, Pool: roPool, TenantID: tenantID}, nil
 }
 
+// SetReadOnly wird auch von internal/parallelrun fuer den
+// Parallelbetrieb-Wartungsmodus wiederverwendet, statt die ALTER-DATABASE-
+// Anweisung zu duplizieren.
+func SetReadOnly(ctx context.Context, cluster *cutoverorchestrator.Cluster, dbName string, readOnly bool) error {
+	conn, err := pgx.Connect(ctx, cluster.AdminDSN())
+	if err != nil {
+		return fmt.Errorf("verbindung fuer wartungsmodus-umschaltung: %w", err)
+	}
+	defer conn.Close(ctx)
+	setting := "off"
+	if readOnly {
+		setting = "on"
+	}
+	if _, err := conn.Exec(ctx, "ALTER DATABASE "+pgx.Identifier{dbName}.Sanitize()+" SET default_transaction_read_only = "+setting); err != nil {
+		return fmt.Errorf("wartungsmodus umschalten: %w", err)
+	}
+	return nil
+}
+
 func DropDatabase(ctx context.Context, cluster *cutoverorchestrator.Cluster, dbName string) error {
 	conn, err := pgx.Connect(ctx, cluster.AdminDSN())
 	if err != nil {

migrations/0028_parallel_run_period.down.sql

@@ -0,0 +1,3 @@
+DROP TABLE parallel_run_shutdown_approval;
+DROP TABLE parallel_run_discrepancy;
+DROP TABLE parallel_run_period;

migrations/0028_parallel_run_period.up.sql

@@ -0,0 +1,33 @@
+-- CUT-06: Parallelbetriebsfrist je Altsystem/Mandant nach dem Stichtag,
+-- gemeldete Diskrepanzen waehrend dieser Frist sowie die zweite,
+-- eigenstaendige schriftliche Freigabe zur endgueltigen Abschaltung.
+CREATE TABLE parallel_run_period (
+    id          BIGSERIAL PRIMARY KEY,
+    tenant_id   BIGINT NOT NULL,
+    db_name     TEXT NOT NULL,
+    cutover_at  TIMESTAMPTZ NOT NULL,
+    deadline    TIMESTAMPTZ NOT NULL CHECK (deadline > cutover_at),
+    created_at  TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+
+CREATE TABLE parallel_run_discrepancy (
+    id            BIGSERIAL PRIMARY KEY,
+    period_id     BIGINT NOT NULL REFERENCES parallel_run_period(id),
+    reported_at   TIMESTAMPTZ NOT NULL DEFAULT now(),
+    description   TEXT NOT NULL,
+    status        TEXT NOT NULL DEFAULT 'open' CHECK (status IN ('open', 'resolved', 'deferred')),
+    resolved_at   TIMESTAMPTZ,
+    resolved_note TEXT
+);
+
+CREATE INDEX parallel_run_discrepancy_period_idx ON parallel_run_discrepancy (period_id, status);
+
+CREATE TABLE parallel_run_shutdown_approval (
+    id          BIGSERIAL PRIMARY KEY,
+    period_id   BIGINT NOT NULL REFERENCES parallel_run_period(id),
+    approved_by TEXT NOT NULL,
+    approved_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+    note        TEXT
+);
+
+CREATE INDEX parallel_run_shutdown_approval_period_idx ON parallel_run_shutdown_approval (period_id, approved_at DESC);