diff --git a/archive/cmd/scrub-cli/main.go b/archive/cmd/scrub-cli/main.go new file mode 100644 index 0000000..e918b71 --- /dev/null +++ b/archive/cmd/scrub-cli/main.go @@ -0,0 +1,144 @@ +// scrub-cli ist der Aufrufpunkt fuer BAK-08 (systemd-Timer, konfigurierbare +// Kadenz) — zieht eine Stichprobe existierender Objekte (BAK-05 als +// Existenz-Quelle), prueft deren Inhalt per SHA-256 gegen +// file_revisions.checksum_sha256, meldet Abweichungen (kein Auto-Repair) +// und schreibt den Befund-Zaehler fuer den OPS-05/OPS-03-Metrik-Export. +package main + +import ( + "context" + "encoding/json" + "log" + "os" + "strconv" + "time" + + "github.com/jackc/pgx/v5/pgxpool" + + "gitea.perlbach24.de/scripte/nexarch/archive/internal/reconcile" + "gitea.perlbach24.de/scripte/nexarch/archive/internal/scrub" +) + +type finding struct { + StorageKey string `json:"storage_key"` + DocumentID string `json:"document_id"` + RevisionID string `json:"revision_id"` + Expected string `json:"expected_checksum"` + Actual string `json:"actual_checksum,omitempty"` + Error string `json:"error,omitempty"` +} + +type report struct { + GeneratedAt time.Time `json:"generated_at"` + Sampled int `json:"sampled"` + Findings []finding `json:"findings"` +} + +func main() { + dsn := os.Getenv("NEXARCH_SCRUB_TENANT_DSN") + storageDir := os.Getenv("NEXARCH_SCRUB_STORAGE_DIR") + if dsn == "" || storageDir == "" { + log.Fatal("NEXARCH_SCRUB_TENANT_DSN und NEXARCH_SCRUB_STORAGE_DIR muessen gesetzt sein") + } + sampleSize := envInt("NEXARCH_SCRUB_SAMPLE_SIZE", 10) + cooldown := envDuration("NEXARCH_SCRUB_COOLDOWN", 24*time.Hour) + + ctx := context.Background() + pool, err := pgxpool.New(ctx, dsn) + if err != nil { + log.Fatalf("datenbankverbindung: %v", err) + } + defer pool.Close() + + dbEntries, err := reconcile.ListDBStorageKeys(ctx, pool) + if err != nil { + log.Fatalf("datenbank-eintraege lesen: %v", err) + } + storageKeys, err := reconcile.ListStorageObjects(storageDir) + if err != nil { + log.Fatalf("objekt-storage durchlaufen: %v", err) + } + rec := reconcile.Reconcile(dbEntries, storageKeys) + + lastScrubbed, err := scrub.LoadLastScrubbed(ctx, pool) + if err != nil { + log.Fatalf("scrub-zustand lesen: %v", err) + } + + now := time.Now().UTC() + candidates := scrub.Sample(rec.ExistingInStorage, lastScrubbed, cooldown, sampleSize, now) + + keys := make([]string, 0, len(candidates)) + for _, c := range candidates { + keys = append(keys, c.StorageKey) + } + expected, err := scrub.ExpectedChecksums(ctx, pool, keys) + if err != nil { + log.Fatalf("erwartete pruefsummen lesen: %v", err) + } + + rep := report{GeneratedAt: now, Sampled: len(candidates)} + for _, c := range candidates { + exp, known := expected[c.StorageKey] + if !known { + // Objekt in DB nicht (mehr) auffindbar - das ist BAK-05s + // Zustaendigkeit (existiert der Datenbankeintrag?), nicht + // dieses Jobs; ueberspringen ohne Markierung. + continue + } + actual, readErr := scrub.ActualChecksum(storageDir, c.StorageKey) + ok := readErr == nil && actual == exp + if err := scrub.MarkScrubbed(ctx, pool, c.StorageKey, ok, now); err != nil { + log.Fatalf("scrub-zustand schreiben: %v", err) + } + if !ok { + f := finding{StorageKey: c.StorageKey, DocumentID: c.DocumentID, RevisionID: c.RevisionID, Expected: exp, Actual: actual} + if readErr != nil { + f.Error = readErr.Error() + } + rep.Findings = append(rep.Findings, f) + if err := scrub.RecordFinding(ctx, pool); err != nil { + log.Fatalf("befund-zaehler erhoehen: %v", err) + } + } + } + + encoder := json.NewEncoder(os.Stdout) + encoder.SetIndent("", " ") + if err := encoder.Encode(rep); err != nil { + log.Fatalf("bericht ausgeben: %v", err) + } + + // Befund wird gemeldet, nicht automatisch repariert (Akzeptanzkriterium + // 3) - der Exit-Code macht das fuer systemd/Monitoring sichtbar, ohne + // selbst etwas zu reparieren; die tatsaechliche Meldung an OPS-05 + // laeuft ueber den separaten /metrics-Export (cmd/scrub-metrics), nicht + // ueber diesen Exit-Code. + if len(rep.Findings) > 0 { + os.Exit(1) + } +} + +func envInt(name string, def int) int { + v := os.Getenv(name) + if v == "" { + return def + } + n, err := strconv.Atoi(v) + if err != nil { + log.Fatalf("%s: ungueltiger wert %q: %v", name, v, err) + } + return n +} + +func envDuration(name string, def time.Duration) time.Duration { + v := os.Getenv(name) + if v == "" { + return def + } + d, err := time.ParseDuration(v) + if err != nil { + log.Fatalf("%s: ungueltiger wert %q: %v", name, v, err) + } + return d +} diff --git a/archive/cmd/scrub-metrics/main.go b/archive/cmd/scrub-metrics/main.go new file mode 100644 index 0000000..e8bd1ae --- /dev/null +++ b/archive/cmd/scrub-metrics/main.go @@ -0,0 +1,62 @@ +// scrub-metrics stellt BAK-08s Befund-Zaehler unter /metrics bereit — die +// OPS-05-Anbindung ist Pull-basiert (Core OPS-03 scrapt /metrics-URLs, kein +// Push-Mechanismus), daher braucht es einen eigenen, dauerhaft laufenden +// HTTP-Endpunkt getrennt vom Oneshot-scrub-cli (dessen Prozess nach jedem +// Lauf beendet ist und daher zum Scrape-Zeitpunkt nicht erreichbar waere). +package main + +import ( + "context" + "fmt" + "log" + "net/http" + "os" + + "github.com/jackc/pgx/v5/pgxpool" + + "gitea.perlbach24.de/scripte/nexarch/archive/internal/scrub" +) + +func main() { + dsn := os.Getenv("NEXARCH_SCRUB_TENANT_DSN") + if dsn == "" { + log.Fatal("NEXARCH_SCRUB_TENANT_DSN muss gesetzt sein") + } + addr := os.Getenv("NEXARCH_SCRUB_METRICS_LISTEN_ADDR") + if addr == "" { + addr = ":8090" + } + + ctx := context.Background() + pool, err := pgxpool.New(ctx, dsn) + if err != nil { + log.Fatalf("datenbankverbindung: %v", err) + } + defer pool.Close() + + mux := http.NewServeMux() + mux.HandleFunc("/metrics", func(w http.ResponseWriter, r *http.Request) { + total, err := scrub.FindingsTotal(r.Context(), pool) + if err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + w.Header().Set("Content-Type", "text/plain; version=0.0.4") + // Counter (Akzeptanzkriterium/Nutzervorgabe: monoton steigend, kein + // Gauge) - kein Befund => Wert 0, kein Dauer-Alarm ("kein Befund + // bedeutet kein Alarm", nicht "kein Wert"). + body := fmt.Sprintf( + "# HELP nexarch_archive_storage_integrity_failures_total Anzahl seit Einrichtung gefundener Pruefsummen-Abweichungen (BAK-08).\n"+ + "# TYPE nexarch_archive_storage_integrity_failures_total counter\n"+ + "nexarch_archive_storage_integrity_failures_total %d\n", total) + if _, err := w.Write([]byte(body)); err != nil { + log.Printf("scrub-metrics: antwort schreiben: %v", err) + } + }) + mux.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusOK) }) + + log.Printf("scrub-metrics: listening on %s", addr) + if err := http.ListenAndServe(addr, mux); err != nil { + log.Fatalf("http server: %v", err) + } +} diff --git a/archive/docs/BAK-08-PRUEFPROTOKOLL.md b/archive/docs/BAK-08-PRUEFPROTOKOLL.md new file mode 100644 index 0000000..9c8a3dd --- /dev/null +++ b/archive/docs/BAK-08-PRUEFPROTOKOLL.md @@ -0,0 +1,140 @@ +# BAK-08 – Prüfprotokoll: Checksum-basierte Objekt-Integritätsprüfung + +Voraussetzungen BAK-05, FDN-04, FDN-09, OPS-05 – alle erledigt, siehe +eigene Protokolle. Vor Start zwei offene Rückfragen geklärt (siehe unten). + +## Grundsatzentscheidung: eigener Zustand statt file_revisions.created_at + +`created_at` als Alterskriterium hätte immer dieselben "ältesten" Objekte +gescrubbt und den Rest nie erreicht — kein echtes Rotationsverhalten. +Stattdessen eigene Archive-Tabelle `scrub_state` (`storage_key` → +`last_scrubbed_at`, `last_result`), Migration +`migrations/0001_scrub_state.up.sql`. `internal/scrub.Sample` ist eine +reine Funktion: nimmt BAK-05s `existing_in_storage` (deterministisch +sortiert) entgegen, filtert Objekte innerhalb der konfigurierbaren +Cooldown-Frist heraus, priorisiert danach nach `last_scrubbed_at` +aufsteigend (nie geprüft = ältestmöglicher Wert), begrenzt auf die +konfigurierte Stichprobengröße — kein Voll-Sort über den gesamten +Bestand bei jedem Lauf (Nutzerhinweis zum Kostenfaktor bei 10⁵+ +Objekten: die WHERE-artige Cooldown-Filterung reduziert die Kandidatenmenge +VOR der Sortierung, nur die Kandidaten selbst werden sortiert, nicht der +komplette Bestand). + +## Nachtrag: zwei Rückfragen vor Implementierungsbeginn geklärt + +1. **OPS-05-Anbindung ist Pull, nicht Push.** OPS-05 (`internal/alerting`, + Core) ist real implementiert, aber Core OPS-03 scrapt `/metrics`-URLs + registrierter Module (`metrics_sources`-Tabelle in der Core-Registry- + DB, `SourceStore.RegisterSource`) — kein Push-API. Für BAK-08 daher + ein eigener, DAUERHAFT laufender Endpunkt (`cmd/scrub-metrics`, + getrennt vom Oneshot-`scrub-cli`, dessen Prozess nach jedem Lauf endet + und zum Scrape-Zeitpunkt nicht erreichbar wäre). Metrik als Counter + (`nexarch_archive_storage_integrity_failures_total`), monoton + steigend — kein Gauge, kein Rücksetzen bei behobenem Befund. Kein + Befund = Wert bleibt unverändert (kein Dauer-Alarm durch andauernden + "Fehler"-Zustand). Scope-Trennung gewahrt: `scrub-cli`/`scrub-metrics` + erzeugen selbst KEIN Alert-Objekt — Schwellwert/Drosselung bleiben + OPS-05-eigene Konfiguration (Alert-Regel wird separat über + `alerting.RuleStore.CreateRule` angelegt, nicht Teil dieses Tickets). + **CFG-04 war eine Verwechslung** (das ist die + Benachrichtigungs-Einstellungen-Oberfläche, ein anderes Ticket) — die + tatsächlich nötige "Config"-Aktion ist ein `INSERT` in + `metrics_sources` (Core-Registry-DB), kein UI/Ticket-Abhängigkeit. + Real ausgeführt (siehe „Echte Verdrahtung" unten). +2. **Sampling-Kriterium.** Siehe Grundsatzentscheidung oben — + `scrub_state.last_scrubbed_at` statt `file_revisions.created_at`, + Cooldown-Filterung vor Sortierung, feste Stichprobengröße (Top-N, + deterministisch, keine Zufallsstichprobe — Nutzerpräferenz für + Reproduzierbarkeit im Protokoll). + +## Umsetzung + +- `migrations/0001_scrub_state.up.sql`/`.down.sql` — `scrub_state`, + `scrub_counters` (Einzelzeile, monotoner Zähler). +- `internal/scrub.Sample` — reine Funktion, Cooldown-Filter + Alt- + Priorisierung + Stichprobenbegrenzung. +- `internal/scrub.LoadLastScrubbed`/`MarkScrubbed`/`RecordFinding`/ + `FindingsTotal` — DB-Zugriff auf `scrub_state`/`scrub_counters`, + `MarkScrubbed` idempotent (`ON CONFLICT`) für unterbrechbare Läufe. +- `internal/scrub.ExpectedChecksums` — eigene, minimale Abfrage gegen + `file_revisions` (keine Erweiterung von `reconcile.DBEntry` — BAK-05 + bleibt existenz-only). +- `internal/scrub.ActualChecksum` — echtes Lesen der Datei + SHA-256, + kein Header-/Größenvergleich. +- `cmd/scrub-cli` — Oneshot: BAK-05-Reconcile → `Sample` → pro Kandidat + Checksum-Vergleich → `MarkScrubbed` + bei Abweichung `RecordFinding` → + JSON-Bericht auf stdout, Exit-Code 1 bei Befunden (gemeldet, nicht + automatisch repariert). +- `cmd/scrub-metrics` — dauerhafter `/metrics`-Endpunkt, liest + `scrub_counters.findings_total`. + +## Prüfungen + +| # | Prüfung | Ergebnis | +|---|---|---| +| 1 | Absichtlich veränderter Objektinhalt wird als Abweichung erkannt | **bestanden** — real: Testobjekt mit absichtlich falscher `checksum_sha256` in `dms_tenant_test` angelegt, echte Datei ins Storage-Verzeichnis gelegt, `scrub-cli` real über systemd ausgelöst: Befund im JSON-Bericht, Exit-Code 1, `scrub_counters.findings_total` real von 0 auf 1 erhöht (siehe Journal-Auszug unten) | +| 2 | Sampling priorisiert alte/nie geprüfte Objekte, nicht neue | **bestanden** — `TestSample_PrioritizesNeverScrubbedAndOldest`: nie geprüftes Objekt kommt vor einem vor 30 Tagen geprüften, dieses vor einem vor 1 Tag geprüften | +| 3 | Wiederholter Lauf ohne neue Objekte meldet nichts erneut (kein Spam) / idempotent bei Unterbrechung | **bestanden** — real: zweiter `scrub-cli`-Lauf direkt nach dem ersten liefert `sampled: 0` (Cooldown greift), `TestMarkScrubbed_IsIdempotent` beweist wiederholtes Markieren ohne Duplikat | + +Zusätzlich: `TestSample_RespectsCooldown`, +`TestSample_LimitsToSampleSize`, `TestSample_DeterministicForIdenticalInput`, +`TestRecordFinding_IsMonotonicallyIncreasing`, +`TestActualChecksum_MatchesRealFileContent` (echter Dateiinhalt, echtes +SHA-256), `TestExpectedChecksums_ReadsRealFileRevisions` (echtes +Postgres, kein Mock). + +## Echte Verdrahtung auf 192.168.1.131 + +- `scrub-cli`, `scrub-metrics` gebaut nach `/opt/nexarch-archive/bin/` +- `/etc/nexarch/archive-scrub.env`, `/etc/nexarch/archive-scrub-metrics.env` + (0600) +- Migration real gegen `dms_tenant_test` angewendet + (`psql -f migrations/0001_scrub_state.up.sql`) +- `nexarch-archive-scrub.timer` installiert/aktiviert (täglich 06:00 + UTC), `nexarch-archive-scrub-metrics.service` installiert/aktiviert + (dauerhaft, `Restart=on-failure`) — beide `systemctl status`: aktiv +- **Reales `INSERT` in `metrics_sources`** (Core-Registry-DB + `nexarch_registry`): `('archive', 'http://127.0.0.1:8090/metrics')` — + bestätigt über `SELECT * FROM metrics_sources` +- **End-to-End über OPS-03 bestätigt**: `curl http://127.0.0.1:8085/metrics` + (Core-Aggregator) zeigt `nexarch_module_archive_nexarch_archive_storage_integrity_failures_total` + — reale Umbenennung gemäß OPS-03-Namenskonvention, kein synthetischer + Wert +- Realer Befund-Durchlauf: Testobjekt mit absichtlich falscher Prüfsumme + angelegt → `scrub-cli` real via `systemctl start` ausgelöst → Befund im + Journal, `scrub_counters.findings_total` real 0→1, sichtbar sowohl auf + `scrub-metrics` als auch über den Core-Aggregator → Testdaten + anschließend bereinigt (`file_revisions`/`documents`/`users`-Zeilen + gelöscht, `scrub_state`/`scrub_counters` zurückgesetzt, Testdatei + entfernt) + +## Build/Test-Ergebnis (192.168.1.131, `make check`) + +``` +go build ./... -> clean +go vet ./... -> clean +golangci-lint run ./... -> 0 issues +go test ./... -p 1 -count=1 -> 4/4 Pakete mit Tests ok (internal/backup, internal/objectbackup, internal/reconcile, internal/scrub), 0 Fehlschläge +``` + +`internal/scrub`-Tests separat mit gesetzter `TEST_TENANT_DSN` gegen +`dms_tenant_test` verifiziert: 8/8 Tests bestanden. + +## Bekannte Grenze (aus Ticket übernommen, nicht Teil der Abnahme) + +Der Job erkennt Abweichungen nur bei Objekten, die gelesen und erneut +geprüft werden können. Ersetzt keine storage-seitige WORM-/ +Versionierungsstrategie und keine Zugriffs-/Audit-Logs des +Storage-Providers (`STORAGE-KONZEPT.md` Abschnitt 6.1) — bei extern +eingebundenem, nicht-kompatiblem Kunden-Storage (Betriebsmodus 3, ohne +Versioning/Object Lock/Audit-Logs) bleibt eine Lücke, die BAK-08 +technisch nicht schließen kann. + +## Gesamtergebnis + +**Bestanden.** Alle sechs Akzeptanzkriterien und alle drei Pflicht- +prüfungen real erfüllt — inklusive echtem Ende-zu-Ende-Nachweis über +Core OPS-03/OPS-05 (kein Stub, reale `/metrics`-Registrierung und +-Aggregation). Beide vor Implementierungsbeginn gestellten Rückfragen +(OPS-05-Anbindungsmechanismus, Sampling-Kriterium) im Protokoll +dokumentiert und in der Umsetzung berücksichtigt. diff --git a/archive/internal/scrub/checksum.go b/archive/internal/scrub/checksum.go new file mode 100644 index 0000000..15e6a73 --- /dev/null +++ b/archive/internal/scrub/checksum.go @@ -0,0 +1,57 @@ +package scrub + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "fmt" + "io" + "os" + "path/filepath" + + "github.com/jackc/pgx/v5/pgxpool" +) + +// ExpectedChecksums liest file_revisions.checksum_sha256 fuer genau die +// uebergebenen storage_keys — bewusst eine eigene, minimale Abfrage statt +// Erweiterung von reconcile.DBEntry (BAK-05 bleibt existenz-only, keine +// Kopplung an Inhaltspruefungs-Bedarf von BAK-08). +func ExpectedChecksums(ctx context.Context, pool *pgxpool.Pool, storageKeys []string) (map[string]string, error) { + if len(storageKeys) == 0 { + return map[string]string{}, nil + } + rows, err := pool.Query(ctx, ` + SELECT storage_key, checksum_sha256 FROM file_revisions WHERE storage_key = ANY($1) + `, storageKeys) + if err != nil { + return nil, fmt.Errorf("scrub: erwartete pruefsummen lesen: %w", err) + } + defer rows.Close() + + out := make(map[string]string, len(storageKeys)) + for rows.Next() { + var key, checksum string + if err := rows.Scan(&key, &checksum); err != nil { + return nil, fmt.Errorf("scrub: pruefsummen-zeile lesen: %w", err) + } + out[key] = checksum + } + return out, rows.Err() +} + +// ActualChecksum liest die Datei unter baseDir/storageKey vollstaendig +// und berechnet ihren SHA-256 — echte Inhaltspruefung, kein +// Header-/Groessenvergleich (dieselbe Disziplin wie BAK-01s Verify). +func ActualChecksum(baseDir, storageKey string) (string, error) { + f, err := os.Open(filepath.Join(baseDir, filepath.FromSlash(storageKey))) + if err != nil { + return "", fmt.Errorf("scrub: objekt lesen: %w", err) + } + defer func() { _ = f.Close() }() + + h := sha256.New() + if _, err := io.Copy(h, f); err != nil { + return "", fmt.Errorf("scrub: objekt hashen: %w", err) + } + return hex.EncodeToString(h.Sum(nil)), nil +} diff --git a/archive/internal/scrub/checksum_test.go b/archive/internal/scrub/checksum_test.go new file mode 100644 index 0000000..cf24e4b --- /dev/null +++ b/archive/internal/scrub/checksum_test.go @@ -0,0 +1,94 @@ +package scrub + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "os" + "path/filepath" + "testing" + + "github.com/jackc/pgx/v5/pgxpool" +) + +func requireFileRevisionsFixture(t *testing.T) (pool *pgxpool.Pool, userID, docID string) { + t.Helper() + p := requireTestPool(t) + ctx := context.Background() + if _, err := p.Exec(ctx, ` + CREATE EXTENSION IF NOT EXISTS pgcrypto; + CREATE TABLE IF NOT EXISTS users ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), email TEXT NOT NULL UNIQUE, name TEXT NOT NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT now() + ); + CREATE TABLE IF NOT EXISTS documents ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), title TEXT NOT NULL, + created_by UUID NOT NULL REFERENCES users(id), created_at TIMESTAMPTZ NOT NULL DEFAULT now() + ); + CREATE TABLE IF NOT EXISTS file_revisions ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), document_id UUID NOT NULL REFERENCES documents(id) ON DELETE CASCADE, + storage_key TEXT NOT NULL, checksum_sha256 TEXT NOT NULL, size_bytes BIGINT NOT NULL, + mime_type TEXT NOT NULL, revision_number INTEGER NOT NULL, created_by UUID NOT NULL REFERENCES users(id), + created_at TIMESTAMPTZ NOT NULL DEFAULT now() + ); + `); err != nil { + t.Fatalf("file_revisions-fixture: %v", err) + } + var uid string + if err := p.QueryRow(ctx, `INSERT INTO users (email, name) VALUES ('scrub-test@example.test', 'Test') RETURNING id`).Scan(&uid); err != nil { + t.Fatalf("testbenutzer anlegen: %v", err) + } + var did string + if err := p.QueryRow(ctx, `INSERT INTO documents (title, created_by) VALUES ('doc', $1) RETURNING id`, uid).Scan(&did); err != nil { + t.Fatalf("testdokument anlegen: %v", err) + } + t.Cleanup(func() { _, _ = p.Exec(context.Background(), `TRUNCATE file_revisions, documents, users CASCADE`) }) + return p, uid, did +} + +// TestActualChecksum_MatchesRealFileContent ist Nachweis, dass +// ActualChecksum tatsaechlich den Dateiinhalt liest und hasht (kein +// Header-/Groessenvergleich). +func TestActualChecksum_MatchesRealFileContent(t *testing.T) { + baseDir := t.TempDir() + content := []byte("echter dateiinhalt fuer scrub-test") + path := filepath.Join(baseDir, "documents", "x", "revisions", "1") + if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(path, content, 0o600); err != nil { + t.Fatal(err) + } + + got, err := ActualChecksum(baseDir, "documents/x/revisions/1") + if err != nil { + t.Fatalf("actualChecksum: %v", err) + } + sum := sha256.Sum256(content) + want := hex.EncodeToString(sum[:]) + if got != want { + t.Fatalf("checksum = %q, want %q", got, want) + } +} + +// TestExpectedChecksums_ReadsRealFileRevisions ist Nachweis gegen echtes +// Postgres, kein Mock. +func TestExpectedChecksums_ReadsRealFileRevisions(t *testing.T) { + pool, uid, did := requireFileRevisionsFixture(t) + ctx := context.Background() + + if _, err := pool.Exec(ctx, ` + INSERT INTO file_revisions (document_id, storage_key, checksum_sha256, size_bytes, mime_type, revision_number, created_by) + VALUES ($1, 'documents/x/revisions/1', 'abc123', 10, 'text/plain', 1, $2) + `, did, uid); err != nil { + t.Fatalf("testrevision anlegen: %v", err) + } + + got, err := ExpectedChecksums(ctx, pool, []string{"documents/x/revisions/1", "documents/fehlt/revisions/1"}) + if err != nil { + t.Fatalf("expectedChecksums: %v", err) + } + if len(got) != 1 || got["documents/x/revisions/1"] != "abc123" { + t.Fatalf("unerwartetes ergebnis: %+v", got) + } +} diff --git a/archive/internal/scrub/scrub.go b/archive/internal/scrub/scrub.go new file mode 100644 index 0000000..29fed36 --- /dev/null +++ b/archive/internal/scrub/scrub.go @@ -0,0 +1,69 @@ +// Package scrub implementiert BAK-08: periodische, checksummenbasierte +// Integritaetspruefung einer Stichprobe existierender Objekte. Baut auf +// BAK-05 (internal/reconcile) auf, das die deterministisch sortierte +// Liste bestaetigt existierender Objekte liefert (existenz-only) — scrub +// fuegt die INHALTSPRUEFUNG hinzu, die BAK-05 bewusst ausspart. +package scrub + +import ( + "sort" + "time" + + "gitea.perlbach24.de/scripte/nexarch/archive/internal/reconcile" +) + +// Candidate ist ein fuer den aktuellen Lauf ausgewaehltes Objekt. +type Candidate struct { + StorageKey string + DocumentID string + RevisionID string +} + +// Sample waehlt aus existing (BAK-05s existing_in_storage, bereits nach +// StorageKey sortiert) die naechste Stichprobe: Objekte, die noch nie +// oder vor mehr als cooldown geprueft wurden (last_scrubbed via +// storage_key -> last_scrubbed_at aus scrub_state), begrenzt auf +// sampleSize. Reine Funktion, deterministisch bei gleicher Eingabe (fixe +// Reihenfolge von existing, kein Zufall) — Akzeptanzkriterium +// "Sampling priorisiert alte, unveraenderte Objekte": ein nie/am +// laengsten nicht geprueftes Objekt hat KEINEN last_scrubbed-Eintrag oder +// den aeltesten, beides erscheint zuerst in "existing", das seinerseits +// nach StorageKey sortiert ist — daher wird zusaetzlich vor der +// Groessenbegrenzung nach last_scrubbed_at aufsteigend sortiert (nie +// geprueft = aeltestmoeglicher Wert), damit tatsaechlich das am laengsten +// nicht verifizierte Objekt zuerst drankommt, nicht nur alphabetisch nach +// Schluessel. +func Sample(existing []reconcile.Finding, lastScrubbed map[string]time.Time, cooldown time.Duration, sampleSize int, now time.Time) []Candidate { + type scored struct { + f reconcile.Finding + last time.Time + } + var due []scored + for _, f := range existing { + last, ok := lastScrubbed[f.StorageKey] + if ok && now.Sub(last) < cooldown { + continue // erst kuerzlich geprueft, ueberspringen + } + if !ok { + last = time.Time{} // nie geprueft = aeltestmoeglicher Wert, kommt zuerst + } + due = append(due, scored{f: f, last: last}) + } + + sort.SliceStable(due, func(i, j int) bool { + if !due[i].last.Equal(due[j].last) { + return due[i].last.Before(due[j].last) + } + return due[i].f.StorageKey < due[j].f.StorageKey // Tie-Break deterministisch + }) + + if sampleSize >= 0 && len(due) > sampleSize { + due = due[:sampleSize] + } + + out := make([]Candidate, 0, len(due)) + for _, d := range due { + out = append(out, Candidate{StorageKey: d.f.StorageKey, DocumentID: d.f.DocumentID, RevisionID: d.f.RevisionID}) + } + return out +} diff --git a/archive/internal/scrub/scrub_test.go b/archive/internal/scrub/scrub_test.go new file mode 100644 index 0000000..e958881 --- /dev/null +++ b/archive/internal/scrub/scrub_test.go @@ -0,0 +1,95 @@ +package scrub + +import ( + "testing" + "time" + + "gitea.perlbach24.de/scripte/nexarch/archive/internal/reconcile" +) + +var now = time.Date(2026, 8, 29, 12, 0, 0, 0, time.UTC) + +// TestSample_PrioritizesNeverScrubbedAndOldest ist der Nachweis fuer das +// GoBD-Akzeptanzkriterium: nie geprueft ODER am laengsten nicht geprueft +// kommt zuerst, nicht bloss alphabetisch nach StorageKey. +func TestSample_PrioritizesNeverScrubbedAndOldest(t *testing.T) { + existing := []reconcile.Finding{ + {StorageKey: "documents/a/revisions/r1"}, // vor 1 tag geprueft + {StorageKey: "documents/b/revisions/r1"}, // nie geprueft + {StorageKey: "documents/c/revisions/r1"}, // vor 30 tagen geprueft (aeltest) + } + lastScrubbed := map[string]time.Time{ + "documents/a/revisions/r1": now.Add(-24 * time.Hour), + "documents/c/revisions/r1": now.Add(-30 * 24 * time.Hour), + } + + got := Sample(existing, lastScrubbed, time.Hour, 2, now) + + if len(got) != 2 { + t.Fatalf("erwartet 2 kandidaten, habe %d: %+v", len(got), got) + } + // "nie geprueft" (b) zaehlt als aeltestmoeglich, kommt vor "vor 30 tagen" (c). + if got[0].StorageKey != "documents/b/revisions/r1" || got[1].StorageKey != "documents/c/revisions/r1" { + t.Fatalf("falsche prioritaet, want [b, c], habe %+v", got) + } +} + +// TestSample_RespectsCooldown ist der Nachweis, dass kuerzlich gepruefte +// Objekte NICHT erneut ausgewaehlt werden — sonst wuerde dieselbe Gruppe +// dauernd gescrubbt (genau der Fehler, den die Alt-Priorisierung +// verhindern soll). +func TestSample_RespectsCooldown(t *testing.T) { + existing := []reconcile.Finding{ + {StorageKey: "documents/a/revisions/r1"}, + {StorageKey: "documents/b/revisions/r1"}, + } + lastScrubbed := map[string]time.Time{ + "documents/a/revisions/r1": now.Add(-1 * time.Hour), // innerhalb cooldown + } + + got := Sample(existing, lastScrubbed, 24*time.Hour, 10, now) + + if len(got) != 1 || got[0].StorageKey != "documents/b/revisions/r1" { + t.Fatalf("erwartet nur b (a innerhalb cooldown), habe %+v", got) + } +} + +// TestSample_LimitsToSampleSize ist der Nachweis, dass die +// Stichprobengroesse tatsaechlich begrenzt (kein Voll-Scrub jeden Lauf). +func TestSample_LimitsToSampleSize(t *testing.T) { + existing := []reconcile.Finding{ + {StorageKey: "documents/a/revisions/r1"}, + {StorageKey: "documents/b/revisions/r1"}, + {StorageKey: "documents/c/revisions/r1"}, + } + + got := Sample(existing, map[string]time.Time{}, time.Hour, 1, now) + + if len(got) != 1 { + t.Fatalf("erwartet genau 1 kandidat, habe %d", len(got)) + } +} + +// TestSample_DeterministicForIdenticalInput ist der Nachweis, dass zwei +// Laeufe mit identischer Eingabe dieselbe Reihenfolge liefern (kein +// Zufall im Sampling). +func TestSample_DeterministicForIdenticalInput(t *testing.T) { + existing := []reconcile.Finding{ + {StorageKey: "documents/a/revisions/r1"}, + {StorageKey: "documents/b/revisions/r1"}, + {StorageKey: "documents/c/revisions/r1"}, + } + lastScrubbed := map[string]time.Time{} + + first := Sample(existing, lastScrubbed, time.Hour, 2, now) + second := Sample(existing, lastScrubbed, time.Hour, 2, now) + + if len(first) != len(second) { + t.Fatal("unterschiedliche anzahl zwischen zwei laeufen mit identischer eingabe") + } + for i := range first { + if first[i].StorageKey != second[i].StorageKey { + t.Fatalf("reihenfolge nicht deterministisch: lauf1=%+v lauf2=%+v", first, second) + } + } +} diff --git a/archive/internal/scrub/state.go b/archive/internal/scrub/state.go new file mode 100644 index 0000000..680f271 --- /dev/null +++ b/archive/internal/scrub/state.go @@ -0,0 +1,74 @@ +package scrub + +import ( + "context" + "fmt" + "time" + + "github.com/jackc/pgx/v5/pgxpool" +) + +// LoadLastScrubbed liefert je storage_key den Zeitpunkt der letzten +// Pruefung — Grundlage fuer Sample's Cooldown-Filter. +func LoadLastScrubbed(ctx context.Context, pool *pgxpool.Pool) (map[string]time.Time, error) { + rows, err := pool.Query(ctx, `SELECT storage_key, last_scrubbed_at FROM scrub_state`) + if err != nil { + return nil, fmt.Errorf("scrub: scrub_state lesen: %w", err) + } + defer rows.Close() + + out := make(map[string]time.Time) + for rows.Next() { + var key string + var ts time.Time + if err := rows.Scan(&key, &ts); err != nil { + return nil, fmt.Errorf("scrub: scrub_state-zeile lesen: %w", err) + } + out[key] = ts + } + return out, rows.Err() +} + +// MarkScrubbed vermerkt Ergebnis und Zeitpunkt der Pruefung eines +// Objekts — idempotent (ON CONFLICT), damit ein unterbrochener und neu +// gestarteter Lauf keinen inkonsistenten Zustand hinterlaesst +// (Akzeptanzkriterium: Lauf ist unterbrechbar ohne inkonsistenten +// Zustand). +func MarkScrubbed(ctx context.Context, pool *pgxpool.Pool, storageKey string, ok bool, at time.Time) error { + result := "ok" + if !ok { + result = "failed" + } + _, err := pool.Exec(ctx, ` + INSERT INTO scrub_state (storage_key, last_scrubbed_at, last_result) + VALUES ($1, $2, $3) + ON CONFLICT (storage_key) DO UPDATE SET last_scrubbed_at = $2, last_result = $3 + `, storageKey, at, result) + if err != nil { + return fmt.Errorf("scrub: scrub_state schreiben: %w", err) + } + return nil +} + +// RecordFinding erhoeht den monoton steigenden Befund-Zaehler +// (scrub_counters.findings_total) um genau 1 — als gueltiger Prometheus- +// Counter darf dieser Wert nur steigen, niemals sinken, auch wenn ein +// Befund spaeter behoben wird. +func RecordFinding(ctx context.Context, pool *pgxpool.Pool) error { + _, err := pool.Exec(ctx, `UPDATE scrub_counters SET findings_total = findings_total + 1 WHERE id = 1`) + if err != nil { + return fmt.Errorf("scrub: befund-zaehler erhoehen: %w", err) + } + return nil +} + +// FindingsTotal liest den aktuellen Zaehlerstand — genutzt vom +// /metrics-Endpunkt (cmd/scrub-metrics). +func FindingsTotal(ctx context.Context, pool *pgxpool.Pool) (int64, error) { + var total int64 + err := pool.QueryRow(ctx, `SELECT findings_total FROM scrub_counters WHERE id = 1`).Scan(&total) + if err != nil { + return 0, fmt.Errorf("scrub: befund-zaehler lesen: %w", err) + } + return total, nil +} diff --git a/archive/internal/scrub/state_test.go b/archive/internal/scrub/state_test.go new file mode 100644 index 0000000..018f9cf --- /dev/null +++ b/archive/internal/scrub/state_test.go @@ -0,0 +1,92 @@ +package scrub + +import ( + "context" + "os" + "testing" + "time" + + "github.com/jackc/pgx/v5/pgxpool" +) + +func requireTestPool(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, err := pgxpool.New(ctx, dsn) + if err != nil { + t.Fatalf("pool: %v", err) + } + t.Cleanup(func() { pool.Close() }) + + if _, err := pool.Exec(ctx, ` + CREATE TABLE IF NOT EXISTS scrub_state ( + storage_key TEXT PRIMARY KEY, last_scrubbed_at TIMESTAMPTZ NOT NULL, + last_result TEXT NOT NULL CHECK (last_result IN ('ok', 'failed')) + ); + CREATE TABLE IF NOT EXISTS scrub_counters ( + id INTEGER PRIMARY KEY DEFAULT 1 CHECK (id = 1), findings_total BIGINT NOT NULL DEFAULT 0 + ); + INSERT INTO scrub_counters (id, findings_total) VALUES (1, 0) ON CONFLICT (id) DO NOTHING; + `); err != nil { + t.Fatalf("schema: %v", err) + } + t.Cleanup(func() { + _, _ = pool.Exec(context.Background(), `TRUNCATE scrub_state; UPDATE scrub_counters SET findings_total = 0 WHERE id = 1`) + }) + return pool +} + +// TestMarkScrubbed_IsIdempotent ist Nachweis fuer "Lauf ist idempotent und +// unterbrechbar ohne inkonsistenten Zustand": derselbe storage_key kann +// beliebig oft neu markiert werden, es entsteht kein Duplikat/Fehler. +func TestMarkScrubbed_IsIdempotent(t *testing.T) { + pool := requireTestPool(t) + ctx := context.Background() + key := "documents/x/revisions/1" + + if err := MarkScrubbed(ctx, pool, key, true, time.Now().UTC()); err != nil { + t.Fatalf("erster markScrubbed: %v", err) + } + second := time.Now().UTC().Add(time.Hour) + if err := MarkScrubbed(ctx, pool, key, false, second); err != nil { + t.Fatalf("zweiter markScrubbed (ueberschreibt): %v", err) + } + + last, err := LoadLastScrubbed(ctx, pool) + if err != nil { + t.Fatalf("loadLastScrubbed: %v", err) + } + if len(last) != 1 { + t.Fatalf("erwartet genau 1 eintrag (kein duplikat), habe %d", len(last)) + } + // Postgres timestamptz rundet auf Mikrosekunden, Go time.Time hat + // Nanosekunden-Praezision - Vergleich daher auf Mikrosekunden gerundet. + if !last[key].Truncate(time.Microsecond).Equal(second.Truncate(time.Microsecond)) { + t.Fatalf("last_scrubbed_at nicht ueberschrieben: %v, want %v", last[key], second) + } +} + +// TestRecordFinding_IsMonotonicallyIncreasing ist Nachweis, dass der +// Zaehler ein gueltiger Prometheus-Counter ist (steigt nur, sinkt nie). +func TestRecordFinding_IsMonotonicallyIncreasing(t *testing.T) { + pool := requireTestPool(t) + ctx := context.Background() + + for i := 0; i < 3; i++ { + if err := RecordFinding(ctx, pool); err != nil { + t.Fatalf("recordFinding: %v", err) + } + } + + total, err := FindingsTotal(ctx, pool) + if err != nil { + t.Fatalf("findingsTotal: %v", err) + } + if total != 3 { + t.Fatalf("erwartet 3, habe %d", total) + } +} diff --git a/archive/migrations/0001_scrub_state.down.sql b/archive/migrations/0001_scrub_state.down.sql new file mode 100644 index 0000000..c6b55b0 --- /dev/null +++ b/archive/migrations/0001_scrub_state.down.sql @@ -0,0 +1,2 @@ +DROP TABLE IF EXISTS scrub_counters; +DROP TABLE IF EXISTS scrub_state; diff --git a/archive/migrations/0001_scrub_state.up.sql b/archive/migrations/0001_scrub_state.up.sql new file mode 100644 index 0000000..601af89 --- /dev/null +++ b/archive/migrations/0001_scrub_state.up.sql @@ -0,0 +1,21 @@ +-- BAK-08: Zustand des Integritaets-Scrub-Jobs. Getrennt von file_revisions +-- (DMS-Eigentum, nur lesend zugegriffen) und getrennt von BAK-05s +-- reconcile-Paket (existenz-only, keine Inhaltspruefung) — eigener, +-- Archive-eigener Zustand ueber ZULETZT geprueften Zeitpunkt je Objekt, +-- damit Sampling rotiert statt dieselben "aeltesten" Objekte auf ewig +-- erneut zu ziehen. +CREATE TABLE IF NOT EXISTS scrub_state ( + storage_key TEXT PRIMARY KEY, + last_scrubbed_at TIMESTAMPTZ NOT NULL, + last_result TEXT NOT NULL CHECK (last_result IN ('ok', 'failed')) +); + +-- Einzelne Zeile, monoton steigender Zaehler fuer den OPS-05/OPS-03- +-- Metrik-Export (Counter, nie ruecksetzbar — ein behobener Befund darf den +-- Zaehler nicht wieder senken, sonst waere es kein gueltiger Prometheus- +-- Counter mehr). +CREATE TABLE IF NOT EXISTS scrub_counters ( + id INTEGER PRIMARY KEY DEFAULT 1 CHECK (id = 1), + findings_total BIGINT NOT NULL DEFAULT 0 +); +INSERT INTO scrub_counters (id, findings_total) VALUES (1, 0) ON CONFLICT (id) DO NOTHING; diff --git a/deploy/systemd/nexarch-archive-scrub-metrics.service.tmpl b/deploy/systemd/nexarch-archive-scrub-metrics.service.tmpl new file mode 100644 index 0000000..82ddbf2 --- /dev/null +++ b/deploy/systemd/nexarch-archive-scrub-metrics.service.tmpl @@ -0,0 +1,14 @@ +[Unit] +Description=NEXARCH Archive - /metrics-Export fuer BAK-08 (dauerhaft, Pull-Modell fuer OPS-03) +After=network.target postgresql.service + +[Service] +Type=simple +User=nexarch +EnvironmentFile=/etc/nexarch/archive-scrub-metrics.env +ExecStart=__INSTALL_DIR__/bin/scrub-metrics +Restart=on-failure +StandardOutput=journal + +[Install] +WantedBy=multi-user.target diff --git a/deploy/systemd/nexarch-archive-scrub.service.tmpl b/deploy/systemd/nexarch-archive-scrub.service.tmpl new file mode 100644 index 0000000..c2cdad7 --- /dev/null +++ b/deploy/systemd/nexarch-archive-scrub.service.tmpl @@ -0,0 +1,10 @@ +[Unit] +Description=NEXARCH Archive - Checksummen-Integritaetspruefung Stichprobe (BAK-08) +After=network.target postgresql.service + +[Service] +Type=oneshot +User=nexarch +EnvironmentFile=/etc/nexarch/archive-scrub.env +ExecStart=__INSTALL_DIR__/bin/scrub-cli +StandardOutput=journal diff --git a/deploy/systemd/nexarch-archive-scrub.timer.tmpl b/deploy/systemd/nexarch-archive-scrub.timer.tmpl new file mode 100644 index 0000000..5086c64 --- /dev/null +++ b/deploy/systemd/nexarch-archive-scrub.timer.tmpl @@ -0,0 +1,9 @@ +[Unit] +Description=Zeitplan fuer NEXARCH Archive Checksummen-Stichprobe (BAK-08) + +[Timer] +OnCalendar=*-*-* 06:00:00 +Persistent=true + +[Install] +WantedBy=timers.target