From 6dc54cd797cae2583c133dd9987381af03bf43ad Mon Sep 17 00:00:00 2001 From: Pierre Schmitz Date: Thu, 3 Sep 2026 14:36:24 +0200 Subject: [PATCH] Archive expired submission logs --- ARCHITECTURE.md | 4 +- internal/config/config.go | 18 +-- internal/submit/archive.go | 250 ++++++++++++++++++++++++++++++++++ internal/submit/log_test.go | 198 ++++++++++++++++++++++++++- internal/submit/prune.go | 5 +- internal/submit/repository.go | 8 +- 6 files changed, 463 insertions(+), 20 deletions(-) create mode 100644 internal/submit/archive.go diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 64ca5b70..404e87c4 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -31,7 +31,7 @@ cmd/ All data tables have the same shape: `( TEXT, month INT, count INT)` with `PRIMARY KEY (, month)`. The identifier column name varies by table (`name`, `code`, `url`, `id`). Month is encoded as `YEAR*100 + MONTH` (e.g. 202603). Each table maps 1:1 to a package in `internal/`. -The exception is `submission_log`: one row per accepted submission with client IP, HTTP headers and the raw JSON payload. It exists to analyze abusive submissions and recover the aggregate tables from data poisoning, and is pruned periodically. Payloads are plain JSON, so ad-hoc analysis works with SQLite's built-in JSON functions (e.g. `json_each(payload, '$.pacman.packages')`). +The exception is `submission_log`: one row per accepted submission with client IP, HTTP headers and the raw JSON payload. It exists to analyze abusive submissions and recover the aggregate tables from data poisoning. Payloads are plain JSON, so ad-hoc analysis works with SQLite's built-in JSON functions (e.g. `json_each(payload, '$.pacman.packages')`). Migrations are numbered sequential SQL files run automatically on startup via `golang-migrate`. When adding a new migration, use the next number after the highest existing one. @@ -92,7 +92,7 @@ Checks for count correlations, new entity spikes, mirror/arch growth anomalies, ## CLI Subcommand: Prune Submission Log -`pkgstatsd prune-submission-log` — deletes `submission_log` rows older than the retention window (the current plus two previous calendar months). Pruning is intentionally kept off the request path and is meant to be run periodically by an external scheduler, so retention is enforced on a schedule and its success is independently observable. +`pkgstatsd prune-submission-log` — archives `submission_log` rows older than the retention window (the current plus two previous calendar months) to gzip-compressed JSON Lines files, then deletes them from SQLite. Archives older than 12 months are removed. Set `SUBMISSION_LOG_ARCHIVE_DIR` to the archive location. Pruning is intentionally kept off the request path and is meant to be run periodically by an external scheduler, so retention is enforced on a schedule and its success is independently observable. ## Dev Workflow (`justfile`) diff --git a/internal/config/config.go b/internal/config/config.go index 74e57725..cdb48443 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -11,10 +11,11 @@ import ( var defaultExpectedPackages = []string{"pkgstats", "pacman"} type Config struct { - Database string - GeoIPDatabase string - Port string - ExpectedPackages []string + Database string + GeoIPDatabase string + Port string + ExpectedPackages []string + SubmissionLogArchiveDir string } func Load() (Config, error) { @@ -24,10 +25,11 @@ func Load() (Config, error) { } cfg := Config{ - Database: getEnv("DATABASE", ""), - GeoIPDatabase: getEnv("GEOIP_DATABASE", ""), - Port: getEnv("PORT", "8282"), - ExpectedPackages: expectedPackages, + Database: getEnv("DATABASE", ""), + GeoIPDatabase: getEnv("GEOIP_DATABASE", ""), + Port: getEnv("PORT", "8282"), + ExpectedPackages: expectedPackages, + SubmissionLogArchiveDir: getEnv("SUBMISSION_LOG_ARCHIVE_DIR", ""), } if cfg.Database == "" { diff --git a/internal/submit/archive.go b/internal/submit/archive.go new file mode 100644 index 00000000..95950f98 --- /dev/null +++ b/internal/submit/archive.go @@ -0,0 +1,250 @@ +package submit + +import ( + "bufio" + "compress/gzip" + "context" + "encoding/json" + "errors" + "fmt" + "io" + "log/slog" + "os" + "path/filepath" + "strconv" + "strings" + "syscall" + "time" +) + +const ( + archiveDirectoryMode = 0o750 + archiveFileMode = 0o600 + archiveRetentionMonths = 12 +) + +type archivedLogEntry struct { + ID int64 `json:"id"` + Month int `json:"month"` + Timestamp int64 `json:"timestamp"` + IP string `json:"ip"` + Headers string `json:"headers"` + Payload string `json:"payload"` + PayloadHash string `json:"payload_hash"` + Country string `json:"country"` +} + +type ArchivePruneResult struct { + ArchivedMonths int + PrunedEntries int64 + RemovedArchives int +} + +// ArchiveAndPruneLog archives every expired submission-log month before +// removing it from the live database. archiveDir is required only when there +// are expired entries to archive. +func (r *Repository) ArchiveAndPruneLog(ctx context.Context, archiveDir string) (ArchivePruneResult, error) { + now := r.now() + cutoff := retentionCutoff(now) + months, err := r.expiredLogMonths(ctx, cutoff) + if err != nil { + return ArchivePruneResult{}, fmt.Errorf("find expired submission log months: %w", err) + } + if len(months) > 0 && archiveDir == "" { + return ArchivePruneResult{}, errors.New("SUBMISSION_LOG_ARCHIVE_DIR is required to archive expired submission logs") + } + + release, err := lockArchiveDirectory(archiveDir) + if err != nil { + return ArchivePruneResult{}, fmt.Errorf("lock submission log archive: %w", err) + } + defer release() + + for _, month := range months { + if err := r.archiveLogMonth(ctx, archiveDir, month); err != nil { + return ArchivePruneResult{}, fmt.Errorf("archive submission log month %d: %w", month, err) + } + } + + deleted, err := r.pruneLog(ctx, cutoff, now) + if err != nil { + return ArchivePruneResult{}, err + } + removed, err := r.pruneArchives(archiveDir, archiveRetentionCutoff(now)) + if err != nil { + return ArchivePruneResult{}, fmt.Errorf("prune submission log archives: %w", err) + } + return ArchivePruneResult{ + ArchivedMonths: len(months), + PrunedEntries: deleted, + RemovedArchives: removed, + }, nil +} + +func lockArchiveDirectory(archiveDir string) (func(), error) { + if archiveDir == "" { + return func() {}, nil + } + if err := os.MkdirAll(archiveDir, archiveDirectoryMode); err != nil { + return nil, err + } + lock, err := os.OpenFile(filepath.Join(archiveDir, ".submission-log-archive.lock"), os.O_CREATE|os.O_RDWR, archiveFileMode) //nolint:gosec // archiveDir is operator configured + if err != nil { + return nil, err + } + if err := syscall.Flock(int(lock.Fd()), syscall.LOCK_EX|syscall.LOCK_NB); err != nil { + _ = lock.Close() + return nil, err + } + return func() { + _ = syscall.Flock(int(lock.Fd()), syscall.LOCK_UN) + _ = lock.Close() + }, nil +} + +func (r *Repository) pruneArchives(archiveDir string, cutoff int) (int, error) { + if archiveDir == "" { + return 0, nil + } + entries, err := os.ReadDir(archiveDir) + if errors.Is(err, os.ErrNotExist) { + return 0, nil + } + if err != nil { + return 0, err + } + removed := 0 + for _, entry := range entries { + month, ok := archiveMonth(entry.Name()) + if !ok || entry.IsDir() || month >= cutoff { + continue + } + if err := os.Remove(filepath.Join(archiveDir, entry.Name())); err != nil { + return 0, err + } + removed++ + } + return removed, nil +} + +func archiveRetentionCutoff(now time.Time) int { + return yearMonth(time.Date(now.Year(), now.Month()-archiveRetentionMonths, 1, 0, 0, 0, 0, now.Location())) +} + +func archiveMonth(name string) (int, bool) { + const prefix = "submission-log-" + const suffix = ".jsonl.gz" + if !strings.HasPrefix(name, prefix) || !strings.HasSuffix(name, suffix) { + return 0, false + } + month, err := strconv.Atoi(strings.TrimSuffix(strings.TrimPrefix(name, prefix), suffix)) + if err != nil || month < 100001 || month%monthMultiplier < 1 || month%monthMultiplier > 12 { + return 0, false + } + return month, true +} + +func (r *Repository) expiredLogMonths(ctx context.Context, cutoff int) ([]int, error) { + rows, err := r.db.QueryContext(ctx, + `SELECT DISTINCT month FROM submission_log WHERE month < ? ORDER BY month`, cutoff) + if err != nil { + return nil, err + } + defer func() { _ = rows.Close() }() + + var months []int + for rows.Next() { + var month int + if err := rows.Scan(&month); err != nil { + return nil, err + } + months = append(months, month) + } + return months, rows.Err() +} + +func (r *Repository) archiveLogMonth(ctx context.Context, archiveDir string, month int) error { + path := filepath.Join(archiveDir, fmt.Sprintf("submission-log-%d.jsonl.gz", month)) + if _, err := os.Lstat(path); err == nil { + slog.Warn("replacing existing submission log archive", "month", month, "path", path) + } else if !errors.Is(err, os.ErrNotExist) { + return fmt.Errorf("stat archive: %w", err) + } + temporary, err := os.CreateTemp(archiveDir, ".submission-log-*.jsonl.gz") + if err != nil { + return fmt.Errorf("create temporary archive: %w", err) + } + temporaryPath := temporary.Name() + defer func() { _ = os.Remove(temporaryPath) }() + if err := temporary.Chmod(archiveFileMode); err != nil { + _ = temporary.Close() + return fmt.Errorf("set archive permissions: %w", err) + } + + if err := r.writeLogArchive(ctx, temporary, month); err != nil { + _ = temporary.Close() + return err + } + if err := temporary.Sync(); err != nil { + _ = temporary.Close() + return fmt.Errorf("sync archive: %w", err) + } + if err := temporary.Close(); err != nil { + return fmt.Errorf("close archive: %w", err) + } + if err := os.Rename(temporaryPath, path); err != nil { + return fmt.Errorf("finalize archive: %w", err) + } + if err := syncDirectory(archiveDir); err != nil { + return fmt.Errorf("sync archive directory: %w", err) + } + return nil +} + +func (r *Repository) writeLogArchive(ctx context.Context, destination io.Writer, month int) error { + rows, err := r.db.QueryContext(ctx, ` + SELECT id, month, timestamp, ip, headers, payload, payload_hash, country + FROM submission_log + WHERE month = ? + ORDER BY id`, month) + if err != nil { + return fmt.Errorf("query submission log: %w", err) + } + defer func() { _ = rows.Close() }() + + compressed := gzip.NewWriter(destination) + writer := bufio.NewWriter(compressed) + encoder := json.NewEncoder(writer) + for rows.Next() { + entry := archivedLogEntry{} + if err := rows.Scan( + &entry.ID, &entry.Month, &entry.Timestamp, &entry.IP, &entry.Headers, + &entry.Payload, &entry.PayloadHash, &entry.Country, + ); err != nil { + return fmt.Errorf("scan submission log: %w", err) + } + if err := encoder.Encode(entry); err != nil { + return fmt.Errorf("encode submission log: %w", err) + } + } + if err := rows.Err(); err != nil { + return fmt.Errorf("iterate submission log: %w", err) + } + + if err := writer.Flush(); err != nil { + return fmt.Errorf("flush archive: %w", err) + } + if err := compressed.Close(); err != nil { + return fmt.Errorf("close gzip archive: %w", err) + } + return nil +} + +func syncDirectory(path string) error { + directory, err := os.Open(path) //nolint:gosec // path is the operator-configured archive directory + if err != nil { + return err + } + defer func() { _ = directory.Close() }() + return directory.Sync() +} diff --git a/internal/submit/log_test.go b/internal/submit/log_test.go index ee1dbce3..6005f96c 100644 --- a/internal/submit/log_test.go +++ b/internal/submit/log_test.go @@ -1,11 +1,14 @@ package submit import ( + "compress/gzip" "context" "crypto/sha256" "encoding/hex" "encoding/json" "net/http" + "os" + "path/filepath" "reflect" "testing" "time" @@ -164,12 +167,42 @@ func TestPrune(t *testing.T) { t.Fatalf("expected 2 rows before prune, got %d", before) } - deleted, err := NewRepository(db).PruneLog(context.Background()) + archiveDir := t.TempDir() + repository := NewRepository(db) + repository.now = func() time.Time { return time.Date(2001, time.January, 1, 0, 0, 0, 0, time.UTC) } + result, err := repository.ArchiveAndPruneLog(context.Background(), archiveDir) if err != nil { t.Fatalf("prune failed: %v", err) } - if deleted != 1 { - t.Errorf("expected 1 pruned entry, got %d", deleted) + if result.ArchivedMonths != 1 { + t.Errorf("expected 1 archived month, got %d", result.ArchivedMonths) + } + if result.PrunedEntries != 1 { + t.Errorf("expected 1 pruned entry, got %d", result.PrunedEntries) + } + if result.RemovedArchives != 0 { + t.Errorf("expected no expired archives to be removed, got %d", result.RemovedArchives) + } + + archive, err := os.Open(filepath.Join(archiveDir, "submission-log-200001.jsonl.gz")) + if err != nil { + t.Fatalf("open archive: %v", err) + } + defer func() { _ = archive.Close() }() + compressed, err := gzip.NewReader(archive) + if err != nil { + t.Fatalf("read gzip archive: %v", err) + } + defer func() { _ = compressed.Close() }() + var entry archivedLogEntry + if err := json.NewDecoder(compressed).Decode(&entry); err != nil { + t.Fatalf("decode archive entry: %v", err) + } + if entry.Month != 200001 { + t.Errorf("archive month = %d, want 200001", entry.Month) + } + if entry.Payload != "{}" || entry.Headers != "{}" || entry.Country != "" { + t.Errorf("archive entry = %#v, want original submission log values", entry) } var remaining int @@ -189,6 +222,165 @@ func TestPrune(t *testing.T) { } } +func TestArchiveAndPruneRequiresArchiveDirectory(t *testing.T) { + _, db := setupTestHandler(t) + _, err := db.Exec(` + INSERT INTO submission_log (month, timestamp, ip, headers, payload, payload_hash, country) + VALUES (200001, 0, '', '{}', '{}', '', '')`) + if err != nil { + t.Fatalf("insert expired submission log: %v", err) + } + + repository := NewRepository(db) + repository.now = func() time.Time { return time.Date(2001, time.January, 1, 0, 0, 0, 0, time.UTC) } + _, err = repository.ArchiveAndPruneLog(context.Background(), "") + if err == nil { + t.Fatal("expected missing archive directory error") + } + + var remaining int + if err := db.QueryRow(`SELECT COUNT(*) FROM submission_log WHERE month = 200001`).Scan(&remaining); err != nil { + t.Fatalf("count expired submission logs: %v", err) + } + if remaining != 1 { + t.Errorf("expected expired log entry to remain, found %d", remaining) + } +} + +func TestArchiveRetention(t *testing.T) { + archiveDir := t.TempDir() + for _, name := range []string{ + "submission-log-202409.jsonl.gz", + "submission-log-202509.jsonl.gz", + "unrelated.jsonl.gz", + } { + if err := os.WriteFile(filepath.Join(archiveDir, name), nil, archiveFileMode); err != nil { + t.Fatalf("create archive %s: %v", name, err) + } + } + + repository := NewRepository(nil) + repository.now = func() time.Time { return time.Date(2026, time.September, 3, 0, 0, 0, 0, time.UTC) } + removed, err := repository.pruneArchives(archiveDir, archiveRetentionCutoff(repository.now())) + if err != nil { + t.Fatalf("prune archives: %v", err) + } + if removed != 1 { + t.Errorf("removed archives = %d, want 1", removed) + } + for _, name := range []string{"submission-log-202509.jsonl.gz", "unrelated.jsonl.gz"} { + if _, err := os.Stat(filepath.Join(archiveDir, name)); err != nil { + t.Errorf("expected %s to remain: %v", name, err) + } + } +} + +func TestArchiveAndPruneReplacesExistingArchive(t *testing.T) { + _, db := setupTestHandler(t) + _, err := db.Exec(` + INSERT INTO submission_log (month, timestamp, ip, headers, payload, payload_hash, country) + VALUES (200001, 0, '', '{}', '{}', '', '')`) + if err != nil { + t.Fatalf("insert expired submission log: %v", err) + } + + archiveDir := t.TempDir() + archivePath := filepath.Join(archiveDir, "submission-log-200001.jsonl.gz") + if err := os.WriteFile(archivePath, []byte("not a gzip archive"), archiveFileMode); err != nil { + t.Fatalf("write old archive: %v", err) + } + repository := NewRepository(db) + repository.now = func() time.Time { return time.Date(2001, time.January, 1, 0, 0, 0, 0, time.UTC) } + result, err := repository.ArchiveAndPruneLog(context.Background(), archiveDir) + if err != nil { + t.Fatalf("archive and prune logs: %v", err) + } + if result.PrunedEntries != 1 { + t.Errorf("pruned entries = %d, want 1", result.PrunedEntries) + } + + archive, err := os.Open(archivePath) + if err != nil { + t.Fatalf("open archive: %v", err) + } + defer func() { _ = archive.Close() }() + compressed, err := gzip.NewReader(archive) + if err != nil { + t.Fatalf("read gzip archive: %v", err) + } + defer func() { _ = compressed.Close() }() + var entry archivedLogEntry + if err := json.NewDecoder(compressed).Decode(&entry); err != nil { + t.Fatalf("decode archive entry: %v", err) + } + if entry.Month != 200001 { + t.Errorf("archive month = %d, want 200001", entry.Month) + } +} + +func TestArchiveAndPruneRejectsConcurrentRun(t *testing.T) { + _, db := setupTestHandler(t) + _, err := db.Exec(` + INSERT INTO submission_log (month, timestamp, ip, headers, payload, payload_hash, country) + VALUES (200001, 0, '', '{}', '{}', '', '')`) + if err != nil { + t.Fatalf("insert expired submission log: %v", err) + } + + archiveDir := t.TempDir() + release, err := lockArchiveDirectory(archiveDir) + if err != nil { + t.Fatalf("lock archive directory: %v", err) + } + defer release() + repository := NewRepository(db) + repository.now = func() time.Time { return time.Date(2001, time.January, 1, 0, 0, 0, 0, time.UTC) } + if _, err := repository.ArchiveAndPruneLog(context.Background(), archiveDir); err == nil { + t.Fatal("expected concurrent archive error") + } + + var remaining int + if err := db.QueryRow(`SELECT COUNT(*) FROM submission_log WHERE month = 200001`).Scan(&remaining); err != nil { + t.Fatalf("count submission logs: %v", err) + } + if remaining != 1 { + t.Errorf("expected expired log entry to remain, found %d", remaining) + } +} + +func TestArchiveAndPruneUsesSingleCutoff(t *testing.T) { + _, db := setupTestHandler(t) + _, err := db.Exec(` + INSERT INTO submission_log (month, timestamp, ip, headers, payload, payload_hash, country) + VALUES + (202605, 0, '', '{}', '{}', '', ''), + (202606, 0, '', '{}', '{}', '', '')`) + if err != nil { + t.Fatalf("insert submission logs: %v", err) + } + + repository := NewRepository(db) + nowCalls := 0 + repository.now = func() time.Time { + nowCalls++ + if nowCalls == 1 { + return time.Date(2026, time.August, 1, 0, 0, 0, 0, time.UTC) + } + return time.Date(2026, time.September, 1, 0, 0, 0, 0, time.UTC) + } + if _, err := repository.ArchiveAndPruneLog(context.Background(), t.TempDir()); err != nil { + t.Fatalf("archive and prune logs: %v", err) + } + + var remaining int + if err := db.QueryRow(`SELECT COUNT(*) FROM submission_log WHERE month = 202606`).Scan(&remaining); err != nil { + t.Fatalf("count submission logs: %v", err) + } + if remaining != 1 { + t.Errorf("expected June log entry to remain, found %d", remaining) + } +} + func TestRetentionCutoff(t *testing.T) { tests := []struct { name string diff --git a/internal/submit/prune.go b/internal/submit/prune.go index d481efda..06b26ca2 100644 --- a/internal/submit/prune.go +++ b/internal/submit/prune.go @@ -21,12 +21,13 @@ func RunPruneLog(_ []string, cfg config.Config) int { } defer func() { _ = db.Close() }() - deleted, err := NewRepository(db).PruneLog(context.Background()) + result, err := NewRepository(db).ArchiveAndPruneLog(context.Background(), cfg.SubmissionLogArchiveDir) if err != nil { fmt.Fprintf(os.Stderr, "Error: %v\n", err) return 1 } - fmt.Printf("Pruned %d expired submission log entries.\n", deleted) + fmt.Printf("Archived %d months, pruned %d expired submission log entries, and removed %d expired archives.\n", + result.ArchivedMonths, result.PrunedEntries, result.RemovedArchives) return 0 } diff --git a/internal/submit/repository.go b/internal/submit/repository.go index 8f1920b4..c8964dc6 100644 --- a/internal/submit/repository.go +++ b/internal/submit/repository.go @@ -136,16 +136,14 @@ func retentionCutoff(now time.Time) int { return yearMonth(time.Date(now.Year(), now.Month()-retentionMonths, 1, 0, 0, 0, 0, now.Location())) } -// PruneLog deletes expired submission logs and deduplication fingerprints. It -// returns the number of log entries removed and runs as scheduled maintenance. -func (r *Repository) PruneLog(ctx context.Context) (int64, error) { +func (r *Repository) pruneLog(ctx context.Context, cutoff int, now time.Time) (int64, error) { result, err := r.db.ExecContext(ctx, - `DELETE FROM submission_log WHERE month < ?`, retentionCutoff(r.now())) + `DELETE FROM submission_log WHERE month < ?`, cutoff) if err != nil { return 0, fmt.Errorf("prune submission log: %w", err) } if _, err := r.db.ExecContext(ctx, - `DELETE FROM submission_dedup WHERE expires_at < ?`, r.now().Unix()); err != nil { + `DELETE FROM submission_dedup WHERE expires_at < ?`, now.Unix()); err != nil { return 0, fmt.Errorf("prune submission deduplication: %w", err) } return result.RowsAffected()