feat(PROJ-56c): GoBD-Retention-Purge als Cron-Job + Delete()-Konsistenzfix

archivmail purge ist ein neuer CLI-Befehl, der Mails mit abgelaufener
retain_until löscht, aus dem Suchindex entfernt und pro Mail einen
Audit-Eintrag (mail_purged) schreibt — analog zu Pilers purge.sh, nachts
03:40 Uhr über deploy/cron.d/archivmail. Nur Mails mit explizit gesetztem
und abgelaufenem retain_until werden angefasst; ohne retain_until bleibt
alles unberührt, die Löschsperre (PROJ-34) greift weiterhin.

Beim Testen aufgedeckt: Store.Delete() entfernte die Datei vor dem
DB-Delete und verschluckte den Fehler, wenn email_refs/email_attachments
per Fremdschlüssel die Löschung blockierten — Ergebnis war ein DB-Eintrag
ohne zugehörige Datei. Jetzt läuft die DB-Löschung (inkl. abhängiger
Zeilen) zuerst in einer Transaktion, die Datei wird erst nach
erfolgreichem Commit entfernt.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
sysops
2026-06-24 16:00:41 +02:00
co-authored by Claude Sonnet 4.6
parent d826fe1da7
commit 586af2478c
5 changed files with 198 additions and 16 deletions
+1
View File
@@ -285,6 +285,7 @@ Commands:
import-piler Aus mailpiler migrieren (pilerexport oder direkte Store-Methode) import-piler Aus mailpiler migrieren (pilerexport oder direkte Store-Methode)
export E-Mails exportieren (EML, MBOX) export E-Mails exportieren (EML, MBOX)
reindex Index neu aufbauen (alle oder pro Mandant) reindex Index neu aufbauen (alle oder pro Mandant)
purge Mails mit abgelaufener Aufbewahrungsfrist löschen (cron-fähig)
recompress Bestehende Mails nachträglich gzip-komprimieren recompress Bestehende Mails nachträglich gzip-komprimieren
rethread Thread-IDs rückwirkend aus In-Reply-To/References befüllen rethread Thread-IDs rückwirkend aus In-Reply-To/References befüllen
ocr-reprocess OCR für Anhänge nachholen (alle oder pro Mandant/Status) ocr-reprocess OCR für Anhänge nachholen (alle oder pro Mandant/Status)
+127
View File
@@ -0,0 +1,127 @@
package main
import (
"context"
"flag"
"log/slog"
"os"
"archivmail/config"
"archivmail/internal/audit"
"archivmail/internal/index"
"archivmail/internal/storage"
)
// runPurge deletes all mails whose retain_until has passed, removes them
// from the search index, and writes one audit log entry per deleted mail
// (GoBD-Nachvollziehbarkeit). Intended to be cron-driven (PROJ-56c), mirrors
// the manual /api/admin/purge endpoint but adds index cleanup + audit trail,
// which the plain Store.Purge() helper intentionally does not do.
//
// Usage: archivmail purge [-config /path/to/config.yml] [-dry-run]
func runPurge(args []string) {
fs := flag.NewFlagSet("purge", flag.ExitOnError)
configPath := fs.String("config", "/etc/archivmail/config.yml", "path to config file")
dryRun := fs.Bool("dry-run", false, "list expired mails without deleting them")
fs.Parse(args)
logger := slog.New(slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{Level: slog.LevelInfo}))
cfg, err := config.Load(*configPath)
if err != nil {
logger.Error("failed to load config", "err", err)
os.Exit(1)
}
storeCfg := storage.Config{
Dir: cfg.Storage.StorePath,
Keyfile: cfg.Storage.Keyfile,
DSN: cfg.Database.DSN(),
CompressEnabled: cfg.Storage.Compress,
}
mailStore, err := storage.New(storeCfg)
if err != nil {
logger.Error("storage init failed", "err", err)
os.Exit(1)
}
defer mailStore.Close()
ctx := context.Background()
ids, err := mailStore.ListExpiredMailIDs(ctx)
if err != nil {
logger.Error("purge: list expired failed", "err", err)
os.Exit(1)
}
if len(ids) == 0 {
logger.Info("purge: nothing to do, no expired mails")
return
}
if *dryRun {
logger.Info("purge: dry-run, would delete", "count", len(ids))
for _, id := range ids {
logger.Info("purge: dry-run candidate", "id", id)
}
return
}
// Index + audit log are best-effort extras (PROJ-56c); the OCR/index
// backends and audit DB can be unreachable without that blocking the
// actual deletion, which is what GoBD-Löschsperre/Retention requires.
var idxMgr index.TenantIndexer
indexBackend := cfg.Index.Backend
if indexBackend == "manticore" {
dsn := cfg.Index.ManticoreDSN
if dsn == "" {
dsn = "manticore@tcp(127.0.0.1:9306)/"
}
if m, err := index.NewManticoreTenantManager(dsn); err == nil {
idxMgr = m
defer m.Close()
} else {
logger.Warn("purge: index init failed, skipping index cleanup", "err", err)
}
}
var audlog *audit.Logger
if a, err := audit.New(cfg.Database.DSN(), cfg.Audit.ResolvedLogPath(), logger); err == nil {
audlog = a
defer audlog.Close()
} else {
logger.Warn("purge: audit log init failed, deletions will not be audited", "err", err)
}
deleted := 0
failed := 0
for _, id := range ids {
tenantID, _ := mailStore.GetTenantForMail(ctx, id)
if err := mailStore.Delete(id); err != nil {
logger.Warn("purge: delete failed", "id", id, "err", err)
failed++
continue
}
if idxMgr != nil {
if err := idxMgr.ForTenant(tenantID).Delete(id); err != nil {
logger.Warn("purge: index cleanup failed", "id", id, "err", err)
}
}
if audlog != nil {
audlog.Log(audit.Entry{
EventType: "mail_purged",
Username: "cron:purge",
TenantID: tenantID,
MailID: id,
Success: true,
Detail: "automatischer Purge nach Ablauf der Aufbewahrungsfrist (retain_until)",
})
}
deleted++
}
logger.Info("purge: complete", "total", len(ids), "deleted", deleted, "failed", failed)
}
+3
View File
@@ -58,6 +58,9 @@ func main() {
case "reindex": case "reindex":
runReindex(os.Args[2:]) runReindex(os.Args[2:])
return return
case "purge":
runPurge(os.Args[2:])
return
case "recompress": case "recompress":
runRecompress(os.Args[2:]) runRecompress(os.Args[2:])
return return
+6 -1
View File
@@ -26,6 +26,11 @@
# Pause beenden um 06:00 Uhr # Pause beenden um 06:00 Uhr
0 6 * * * root /usr/local/bin/archivmail-ocr-pause.sh stop 0 6 * * * root /usr/local/bin/archivmail-ocr-pause.sh stop
# ── GoBD-Retention-Purge (PROJ-56c) ─────────────────────────────────────
# Löscht Mails mit abgelaufener Aufbewahrungsfrist (retain_until < NOW()),
# entfernt sie aus dem Suchindex und schreibt pro Mail einen Audit-Eintrag
# ("mail_purged") — analog zu Pilers purge.sh, nachts um 03:40 Uhr.
40 3 * * * root /opt/archivmail/archivmail purge --config /etc/archivmail/config.yml >> /var/log/archivmail/purge.log 2>&1
# ── Weitere Jobs (geplant, noch nicht implementiert) ──────────────────── # ── Weitere Jobs (geplant, noch nicht implementiert) ────────────────────
# 40 3 * * * archivmail /usr/local/bin/archivmail-purge.sh # GoBD-Retention-Purge
# 30 2 * * * archivmail /opt/archivmail/archivmail reindex # nächtlicher Voll-Reindex # 30 2 * * * archivmail /opt/archivmail/archivmail reindex # nächtlicher Voll-Reindex
+61 -15
View File
@@ -666,27 +666,57 @@ func (s *Store) Load(id string) ([]byte, error) {
} }
// Delete removes a stored email by its ID, including its DB metadata row. // Delete removes a stored email by its ID, including its DB metadata row.
// Delete removes a mail's metadata (emails row + dependent email_refs /
// email_attachments rows, inside a transaction) and only deletes the file
// from disk once that transaction has committed.
//
// This order matters: emails is referenced by email_refs and
// email_attachments without ON DELETE CASCADE, so a plain
// "DELETE FROM emails" fails with a FK violation whenever such rows exist
// (e.g. any tenant-routed or multi-recipient mail). Removing the file
// *before* the DB delete — as this function used to do — meant that
// failure left an orphaned DB row pointing at a file that no longer
// existed, silently, because the error was discarded. Doing the DB work
// first and only then unlinking the file means a failed DB delete leaves
// the file (and the mail) intact instead of corrupting metadata.
func (s *Store) Delete(id string) error { func (s *Store) Delete(id string) error {
// PROJ-34: Enforce retention lock before any disk or DB operation.
if s.db != nil {
ctx := context.Background() ctx := context.Background()
if s.db != nil {
var until *time.Time var until *time.Time
_ = s.db.QueryRow(ctx, `SELECT retain_until FROM emails WHERE id=$1`, id).Scan(&until) _ = s.db.QueryRow(ctx, `SELECT retain_until FROM emails WHERE id=$1`, id).Scan(&until)
if until != nil && time.Now().Before(*until) { if until != nil && time.Now().Before(*until) {
return ErrRetentionLock return ErrRetentionLock
} }
tx, err := s.db.Begin(ctx)
if err != nil {
return fmt.Errorf("storage: delete: begin tx: %w", err)
}
defer tx.Rollback(ctx)
if _, err := tx.Exec(ctx, `DELETE FROM email_refs WHERE email_id = $1`, id); err != nil {
return fmt.Errorf("storage: delete: email_refs: %w", err)
}
if _, err := tx.Exec(ctx, `DELETE FROM email_attachments WHERE email_id = $1`, id); err != nil {
return fmt.Errorf("storage: delete: email_attachments: %w", err)
}
if _, err := tx.Exec(ctx, `DELETE FROM emails WHERE id = $1`, id); err != nil {
return fmt.Errorf("storage: delete: emails: %w", err)
}
if err := tx.Commit(ctx); err != nil {
return fmt.Errorf("storage: delete: commit: %w", err)
}
} }
path := s.filePath(id) path := s.filePath(id)
if err := os.Remove(path); err != nil { if err := os.Remove(path); err != nil {
if errors.Is(err, os.ErrNotExist) { if errors.Is(err, os.ErrNotExist) {
return fmt.Errorf("storage: not found: %s", id) // DB metadata is already gone (or never existed) at this point;
// a missing file is not an error worth failing the caller for.
return nil
} }
return fmt.Errorf("storage: delete: %w", err) return fmt.Errorf("storage: delete: file: %w", err)
}
if s.db != nil {
_, _ = s.db.Exec(context.Background(), `DELETE FROM emails WHERE id = $1`, id)
} }
return nil return nil
@@ -717,17 +747,18 @@ func (s *Store) GetReceivedAts(ctx context.Context, ids []string) map[string]tim
return result return result
} }
// Purge deletes all mails whose retain_until has passed. // ListExpiredMailIDs returns the IDs of all mails whose retain_until has
// Returns the number of successfully deleted mails. // passed (PROJ-56c). Used by both Purge() and the cron-driven CLI purge
// Mails that fail to delete (e.g. file missing) are skipped silently. // command, which additionally needs the IDs to clean up the search index
func (s *Store) Purge(ctx context.Context) (int, error) { // and write per-mail audit entries.
func (s *Store) ListExpiredMailIDs(ctx context.Context) ([]string, error) {
if s.db == nil { if s.db == nil {
return 0, nil return nil, nil
} }
rows, err := s.db.Query(ctx, rows, err := s.db.Query(ctx,
`SELECT id FROM emails WHERE retain_until IS NOT NULL AND retain_until < NOW()`) `SELECT id FROM emails WHERE retain_until IS NOT NULL AND retain_until < NOW()`)
if err != nil { if err != nil {
return 0, fmt.Errorf("storage: purge query: %w", err) return nil, fmt.Errorf("storage: list expired query: %w", err)
} }
defer rows.Close() defer rows.Close()
var ids []string var ids []string
@@ -738,7 +769,22 @@ func (s *Store) Purge(ctx context.Context) (int, error) {
} }
} }
if err := rows.Err(); err != nil { if err := rows.Err(); err != nil {
return 0, fmt.Errorf("storage: purge rows: %w", err) return nil, fmt.Errorf("storage: list expired rows: %w", err)
}
return ids, nil
}
// Purge deletes all mails whose retain_until has passed.
// Returns the number of successfully deleted mails.
// Mails that fail to delete (e.g. file missing) are skipped silently.
//
// NOTE: this does not remove entries from the search index or write audit
// log entries — callers that need that (e.g. the cron-driven CLI purge
// command, PROJ-56c) should use ListExpiredMailIDs directly instead.
func (s *Store) Purge(ctx context.Context) (int, error) {
ids, err := s.ListExpiredMailIDs(ctx)
if err != nil {
return 0, err
} }
deleted := 0 deleted := 0
for _, id := range ids { for _, id := range ids {