diff --git a/admin/duplicates.go b/admin/duplicates.go new file mode 100644 index 0000000..b47e4ee --- /dev/null +++ b/admin/duplicates.go @@ -0,0 +1,246 @@ +package admin + +import ( + "context" + "fmt" + "log" + "net/http" + "strings" + "sync" + "time" + + "gitea.tbdevent.eu/TBD/reforger_crawler_main/initializers" + "gitea.tbdevent.eu/TBD/reforger_crawler_main/models" + "github.com/gin-gonic/gin" + "gorm.io/gorm" +) + +type dupHash struct { + Hash string `json:"hash"` + AddonCount int64 `json:"addonCount"` + FileCount int64 `json:"fileCount"` + SamplePath string `json:"samplePath"` + Whitelisted bool `json:"whitelisted"` +} + +// dupState is the state of the background job that finds the hashes shared by +// the most addons. It runs separately from the cheap table statistics because +// it scans every live file row. +type dupState struct { + Running bool + StartedAt int64 + AsOf int64 // last successful run + TookMs int64 + Error string + ErrorAt int64 + DistinctHashes int64 + SharedHashes int64 + Items []dupHash +} + +const ( + dupTopN = 500 // keep more than shown so hiding whitelisted ones still leaves a full list + dupShown = 200 + dupTimeoutSeconds = 15 * 60 +) + +var ( + dupMu sync.Mutex + dups dupState + dupReqCh = make(chan struct{}, 1) +) + +func currentDups() dupState { + dupMu.Lock() + defer dupMu.Unlock() + return dups +} + +// requestDupRefresh asks the duplicate job to run as soon as possible. +func requestDupRefresh() { + select { + case dupReqCh <- struct{}{}: + default: + } +} + +// dupLoop runs the duplicate job on request and hourly. +func dupLoop(ctx context.Context) { + ticker := time.NewTicker(time.Hour) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + runDupJob() + case <-dupReqCh: + runDupJob() + } + } +} + +// runDupJob computes the duplicate statistics. Errors and panics are recorded +// for the dashboard, so the state never stays "running". +func runDupJob() { + dupMu.Lock() + if dups.Running { + dupMu.Unlock() + return + } + dups.Running = true + dups.StartedAt = time.Now().Unix() + dupMu.Unlock() + + started := time.Now() + log.Printf("admin: duplicate hashes: computing") + var ( + items []dupHash + distinct, shared int64 + err error + ) + func() { + defer func() { + if r := recover(); r != nil { + err = fmt.Errorf("panic: %v", r) + } + }() + items, distinct, shared, err = computeDuplicates() + }() + took := time.Since(started) + + dupMu.Lock() + dups.Running = false + dups.TookMs = took.Milliseconds() + if err != nil { + dups.Error, dups.ErrorAt = err.Error(), time.Now().Unix() + } else { + dups.Error, dups.ErrorAt = "", 0 + dups.AsOf = time.Now().Unix() + dups.Items, dups.DistinctHashes, dups.SharedHashes = items, distinct, shared + } + dupMu.Unlock() + invalidate("overview") + + if err != nil { + log.Printf("admin: duplicate hashes: failed after %s: %v", took.Round(time.Millisecond), err) + } else { + log.Printf("admin: duplicate hashes: computed in %s (%d shared of %d distinct)", took.Round(time.Millisecond), shared, distinct) + } +} + +// computeDuplicates scans the live file rows once. The inner DISTINCT over +// (hash, addon_id) can be answered from idx_addon_files_hash_addon_optimized, +// and the window functions return the totals alongside the top hashes. +func computeDuplicates() ([]dupHash, int64, int64, error) { + var rows []struct { + Hash string + AddonCount int64 + DistinctHashes int64 + SharedHashes int64 + } + err := withTimeout(dupTimeoutSeconds, func(tx *gorm.DB) error { + return tx.Raw(`SELECT hash, addon_count, + COUNT(*) OVER () AS distinct_hashes, + SUM(CASE WHEN addon_count > 1 THEN 1 ELSE 0 END) OVER () AS shared_hashes + FROM (SELECT hash, COUNT(*) AS addon_count + FROM (SELECT DISTINCT hash, addon_id FROM addon_files WHERE deleted_at IS NULL) d + GROUP BY hash) g + ORDER BY addon_count DESC, hash ASC + LIMIT ?`, dupTopN).Scan(&rows).Error + }) + if err != nil { + if strings.Contains(err.Error(), "statement timeout") { + err = fmt.Errorf("query timed out after %d minutes: %w", dupTimeoutSeconds/60, err) + } + return nil, 0, 0, err + } + + items := []dupHash{} + var distinct, shared int64 + for _, r := range rows { + distinct, shared = r.DistinctHashes, r.SharedHashes + if r.AddonCount > 1 { + items = append(items, dupHash{Hash: r.Hash, AddonCount: r.AddonCount}) + } + } + if len(items) == 0 { + return items, distinct, shared, nil + } + + hashes := make([]string, len(items)) + for i, d := range items { + hashes[i] = d.Hash + } + var details []struct { + Hash string + Files int64 + Path string + } + if err := initializers.DB.Raw(`SELECT hash, COUNT(*) AS files, MIN(path) AS path FROM addon_files + WHERE deleted_at IS NULL AND hash IN ? GROUP BY hash`, hashes).Scan(&details).Error; err != nil { + return nil, 0, 0, fmt.Errorf("file details: %w", err) + } + byHash := make(map[string]int, len(details)) + for i, d := range details { + byHash[d.Hash] = i + } + for i := range items { + if j, ok := byHash[items[i].Hash]; ok { + items[i].FileCount, items[i].SamplePath = details[j].Files, details[j].Path + } + } + return items, distinct, shared, nil +} + +// duplicateHashes lists the hashes shared by the most addons together with +// the state of the job that computes them. +func duplicateHashes(c *gin.Context) { + d := currentDups() + items := make([]dupHash, len(d.Items)) + copy(items, d.Items) + if len(items) > 0 { + hashes := make([]string, len(items)) + for i, it := range items { + hashes[i] = it.Hash + } + var wl []string + initializers.DB.Model(&models.WhitelistedHash{}).Where("hash IN ?", hashes).Pluck("hash", &wl) + set := make(map[string]bool, len(wl)) + for _, w := range wl { + set[w] = true + } + for i := range items { + items[i].Whitelisted = set[items[i].Hash] + } + } + if c.Query("hideWhitelisted") == "1" { + kept := items[:0] + for _, it := range items { + if !it.Whitelisted { + kept = append(kept, it) + } + } + items = kept + } + if len(items) > dupShown { + items = items[:dupShown] + } + c.JSON(http.StatusOK, gin.H{ + "running": d.Running, + "startedAt": d.StartedAt, + "asOf": d.AsOf, + "tookMs": d.TookMs, + "error": d.Error, + "errorAt": d.ErrorAt, + "distinctHashes": d.DistinctHashes, + "sharedHashes": d.SharedHashes, + "items": items, + }) +} + +func refreshDuplicates(c *gin.Context) { + requestDupRefresh() + audit(c, "duplicates_refresh", nil) + c.JSON(http.StatusAccepted, gin.H{"started": true}) +} diff --git a/admin/duplicates_test.go b/admin/duplicates_test.go new file mode 100644 index 0000000..be4c3f7 --- /dev/null +++ b/admin/duplicates_test.go @@ -0,0 +1,90 @@ +package admin + +import ( + "path/filepath" + "strings" + "testing" + "time" + + "gitea.tbdevent.eu/TBD/reforger_crawler_main/initializers" + "gitea.tbdevent.eu/TBD/reforger_crawler_main/models" + "gorm.io/driver/sqlite" + "gorm.io/gorm" + "gorm.io/gorm/logger" +) + +func setupDupDB(t *testing.T) { + t.Helper() + db, err := gorm.Open(sqlite.Open(filepath.Join(t.TempDir(), "d.db")), &gorm.Config{Logger: logger.Discard}) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { + if sqlDB, err := db.DB(); err == nil { + sqlDB.Close() + } + }) + if err := db.AutoMigrate(&models.AddonFile{}, &models.WhitelistedHash{}); err != nil { + t.Fatal(err) + } + initializers.DB = db + dupMu.Lock() + dups = dupState{} + dupMu.Unlock() +} + +func hash(c string) string { return strings.Repeat(c, 128) } + +func TestDuplicateJob(t *testing.T) { + setupDupDB(t) + files := []models.AddonFile{ + // hash a: in three addons (twice in addon 1) + {AddonID: "1", Hash: hash("a"), Path: "x/a.et"}, + {AddonID: "1", Hash: hash("a"), Path: "y/a.et"}, + {AddonID: "2", Hash: hash("a"), Path: "a.et"}, + {AddonID: "3", Hash: hash("a"), Path: "a.et"}, + // hash b: only in one addon + {AddonID: "1", Hash: hash("b"), Path: "b.et"}, + // hash c: in two addons, but one row is soft-deleted + {AddonID: "1", Hash: hash("c"), Path: "c.et"}, + {AddonID: "2", Hash: hash("c"), Path: "c.et"}, + } + if err := initializers.DB.Create(&files).Error; err != nil { + t.Fatal(err) + } + initializers.DB.Where("addon_id = ? AND hash = ?", "2", hash("c")).Delete(&models.AddonFile{}) + + runDupJob() + d := currentDups() + if d.Running || d.Error != "" || d.AsOf == 0 { + t.Fatalf("unexpected state %+v", d) + } + if d.DistinctHashes != 3 || d.SharedHashes != 1 { + t.Fatalf("distinct=%d shared=%d, want 3 and 1", d.DistinctHashes, d.SharedHashes) + } + if len(d.Items) != 1 { + t.Fatalf("items = %+v, want only hash a", d.Items) + } + it := d.Items[0] + if it.Hash != hash("a") || it.AddonCount != 3 || it.FileCount != 4 || it.SamplePath != "a.et" { + t.Fatalf("unexpected item %+v", it) + } +} + +func TestDuplicateJobRecordsErrors(t *testing.T) { + setupDupDB(t) + if err := initializers.DB.Migrator().DropTable(&models.AddonFile{}); err != nil { + t.Fatal(err) + } + runDupJob() + d := currentDups() + if d.Running { + t.Fatal("job must not stay running after a failure") + } + if d.Error == "" || d.ErrorAt == 0 || d.AsOf != 0 { + t.Fatalf("error not recorded: %+v", d) + } + if time.Now().Unix()-d.ErrorAt > 5 { + t.Fatalf("errorAt not set to now: %d", d.ErrorAt) + } +} diff --git a/admin/overview.go b/admin/overview.go index 31f0509..da3ef40 100644 --- a/admin/overview.go +++ b/admin/overview.go @@ -199,14 +199,6 @@ type tableInfo struct { SizeBytes int64 `json:"sizeBytes"` } -type dupHash struct { - Hash string `json:"hash"` - AddonCount int64 `json:"addonCount"` - FileCount int64 `json:"fileCount"` - SamplePath string `json:"samplePath"` - Whitelisted bool `json:"whitelisted"` -} - type heavyStats struct { AsOf int64 `json:"asOf"` FilesLive int64 `json:"filesLive"` @@ -217,45 +209,45 @@ type heavyStats struct { DBDriver string `json:"dbDriver"` DBSizeBytes int64 `json:"dbSizeBytes"` Tables []tableInfo `json:"tables"` - Duplicates []dupHash `json:"-"` - DuplicatesAsOf int64 `json:"duplicatesAsOf"` Error string `json:"error,omitempty"` } var ( heavyMu sync.Mutex heavy *heavyStats - refreshCh = make(chan bool, 1) + refreshCh = make(chan struct{}, 1) ) +// currentHeavy returns a copy of the table statistics with the distinct-hash +// count from the duplicate job, or nil before the first run. func currentHeavy() *heavyStats { heavyMu.Lock() defer heavyMu.Unlock() - return heavy + if heavy == nil { + return nil + } + h := *heavy + d := currentDups() + h.DistinctHashes, h.DistinctAsOf = d.DistinctHashes, d.AsOf + return &h } -// requestHeavyRefresh triggers a refresh; full=true also recomputes the -// expensive duplicate statistics. -func requestHeavyRefresh(full bool) { +// requestHeavyRefresh recomputes the table statistics soon. +func requestHeavyRefresh() { select { - case refreshCh <- full: + case refreshCh <- struct{}{}: default: } } func heavyStatsLoop(ctx context.Context) { - lastFull := time.Time{} - run := func(full bool) { + run := func() { defer func() { if r := recover(); r != nil { - log.Printf("admin: heavy stats panicked: %v", r) + log.Printf("admin: table stats panicked: %v", r) } }() - full = full || time.Since(lastFull) > time.Hour - computeHeavy(full) - if full { - lastFull = time.Now() - } + computeHeavy() } // give the server a moment to start before the first scan select { @@ -263,7 +255,10 @@ func heavyStatsLoop(ctx context.Context) { return case <-time.After(5 * time.Second): } - run(true) + // the slow duplicate scan runs in its own loop so it never delays these + go dupLoop(ctx) + requestDupRefresh() + run() ticker := time.NewTicker(10 * time.Minute) defer ticker.Stop() for { @@ -271,9 +266,9 @@ func heavyStatsLoop(ctx context.Context) { case <-ctx.Done(): return case <-ticker.C: - run(false) - case full := <-refreshCh: - run(full) + run() + case <-refreshCh: + run() } } } @@ -293,14 +288,11 @@ func withTimeout(seconds int, fn func(tx *gorm.DB) error) error { var ourTables = []string{"addons", "addon_files", "addon_dependencies", "whitelisted_hashes", "metric_series", "metric_points", "service_instances", "service_events", "daily_top_items"} -func computeHeavy(full bool) { +// computeHeavy refreshes table sizes and file row counts. The duplicate-hash +// statistics are computed separately by runDupJob. +func computeHeavy() { started := time.Now() h := &heavyStats{AsOf: time.Now().Unix(), DBDriver: initializers.DB.Dialector.Name(), Tables: []tableInfo{}} - prev := currentHeavy() - if prev != nil { - h.DistinctHashes, h.DistinctAsOf = prev.DistinctHashes, prev.DistinctAsOf - h.Duplicates, h.DuplicatesAsOf = prev.Duplicates, prev.DuplicatesAsOf - } var errs []string note := func(what string, err error) { @@ -342,52 +334,10 @@ func computeHeavy(full bool) { Scan(&h.FilesLive, &h.FilesSoftDeleted)) } - if full { - var distinct int64 - err := withTimeout(120, func(tx *gorm.DB) error { - return tx.Raw(`SELECT COUNT(DISTINCT hash) FROM addon_files WHERE deleted_at IS NULL`).Scan(&distinct).Error - }) - note("distinct hashes", err) - if err == nil { - h.DistinctHashes, h.DistinctAsOf = distinct, time.Now().Unix() - } - - var dups []dupHash - err = withTimeout(120, func(tx *gorm.DB) error { - return tx.Raw(`SELECT hash, COUNT(DISTINCT addon_id) AS addon_count, COUNT(*) AS file_count - FROM addon_files WHERE deleted_at IS NULL - GROUP BY hash HAVING COUNT(DISTINCT addon_id) > 1 - ORDER BY addon_count DESC, file_count DESC LIMIT 200`).Scan(&dups).Error - }) - note("duplicates", err) - if err == nil { - h.Duplicates, h.DuplicatesAsOf = dups, time.Now().Unix() - if len(dups) > 0 { - hashes := make([]string, len(dups)) - for i, d := range dups { - hashes[i] = d.Hash - } - var samples []struct { - Hash string - Path string - } - initializers.DB.Raw(`SELECT hash, MIN(path) AS path FROM addon_files - WHERE deleted_at IS NULL AND hash IN ? GROUP BY hash`, hashes).Scan(&samples) - byHash := map[string]string{} - for _, s := range samples { - byHash[s.Hash] = s.Path - } - for i := range h.Duplicates { - h.Duplicates[i].SamplePath = byHash[h.Duplicates[i].Hash] - } - } - } - } - if len(errs) > 0 { h.Error = errs[0] for _, e := range errs { - log.Printf("admin: heavy stats: %s", e) + log.Printf("admin: table stats: %s", e) } } heavyMu.Lock() @@ -395,7 +345,7 @@ func computeHeavy(full bool) { heavyMu.Unlock() invalidate("overview") if time.Since(started) > 10*time.Second { - log.Printf("admin: heavy stats took %s", time.Since(started).Round(time.Millisecond)) + log.Printf("admin: table stats took %s", time.Since(started).Round(time.Millisecond)) } } @@ -405,39 +355,3 @@ func sqlitePath() string { } return "crawler.db" } - -// duplicateHashes lists the hashes shared by the most addons. -func duplicateHashes(c *gin.Context) { - h := currentHeavy() - if h == nil { - c.JSON(http.StatusOK, gin.H{"asOf": 0, "items": []dupHash{}, "pending": true}) - return - } - items := make([]dupHash, len(h.Duplicates)) - copy(items, h.Duplicates) - if len(items) > 0 { - hashes := make([]string, len(items)) - for i, d := range items { - hashes[i] = d.Hash - } - var wl []string - initializers.DB.Model(&models.WhitelistedHash{}).Where("hash IN ?", hashes).Pluck("hash", &wl) - set := map[string]bool{} - for _, w := range wl { - set[w] = true - } - for i := range items { - items[i].Whitelisted = set[items[i].Hash] - } - } - if c.Query("hideWhitelisted") == "1" { - kept := items[:0] - for _, it := range items { - if !it.Whitelisted { - kept = append(kept, it) - } - } - items = kept - } - c.JSON(http.StatusOK, gin.H{"asOf": h.DuplicatesAsOf, "items": items}) -} diff --git a/admin/routes.go b/admin/routes.go index 22b763c..e1cf7e7 100644 --- a/admin/routes.go +++ b/admin/routes.go @@ -21,6 +21,7 @@ func Register(r *gin.Engine) { p.GET("/overview", overview) p.GET("/overview/duplicates", duplicateHashes) + p.POST("/overview/duplicates/refresh", refreshDuplicates) p.GET("/queue", queue) p.POST("/queue/reset-stuck", resetStuck) diff --git a/admin/system.go b/admin/system.go index e373125..7abb7c9 100644 --- a/admin/system.go +++ b/admin/system.go @@ -199,7 +199,8 @@ func maintenance(c *gin.Context) { } switch body.Action { case "refresh-stats": - requestHeavyRefresh(true) + requestHeavyRefresh() + requestDupRefresh() case "flush-metrics": metrics.FlushAll() case "rollup": @@ -240,5 +241,5 @@ func purgeDeletedFiles(days int) { metrics.Event("info", "purged deleted file rows", map[string]any{ "kind": "maintenance", "deleted": total, "olderThanDays": days, "tookSec": int64(time.Since(started).Seconds()), }) - requestHeavyRefresh(false) + requestHeavyRefresh() } diff --git a/docs/admin-dashboard.md b/docs/admin-dashboard.md index 10ab27d..1c27d7c 100644 --- a/docs/admin-dashboard.md +++ b/docs/admin-dashboard.md @@ -20,7 +20,7 @@ change data must also send `X-Admin-CSRF: 1`. | Area | Endpoints | |---|---| -| Stats | `GET overview`, `GET overview/duplicates`, `GET traffic?from&to`, `GET top?kind=addon\|search&days` | +| Stats | `GET overview`, `GET overview/duplicates` (job state + top shared hashes), `POST overview/duplicates/refresh`, `GET traffic?from&to`, `GET top?kind=addon\|search&days` | | Queue | `GET queue`, `POST queue/reset-stuck` | | Addons | `GET addons?q&status&type&sort&order&limit&offset`, `GET addons/:id`, `GET addons/:id/files` | | Addon actions | `POST addons/:id/{priority\|to-be-indexed\|reindex\|reset-indexing\|recheck\|restore}` with body `{"value": bool}`, `DELETE addons/:id`, `DELETE addons/:id/purge?confirm=`, `POST bulk/addons` | diff --git a/web/src/lib/types.ts b/web/src/lib/types.ts index 4ea43dc..fab92dd 100644 --- a/web/src/lib/types.ts +++ b/web/src/lib/types.ts @@ -60,7 +60,6 @@ export interface HeavyStats { dbDriver: string; dbSizeBytes: number; tables: TableInfo[]; - duplicatesAsOf: number; error?: string; } @@ -213,6 +212,19 @@ export interface DupHash { whitelisted: boolean; } +/** State and result of the background job that finds shared hashes. */ +export interface DupResponse { + running: boolean; + startedAt: number; + asOf: number; + tookMs: number; + error: string; + errorAt: number; + distinctHashes: number; + sharedHashes: number; + items: DupHash[]; +} + export interface ServiceInstance { id: number; service: string; diff --git a/web/src/pages/Whitelist.tsx b/web/src/pages/Whitelist.tsx index 2e3f457..39e91b4 100644 --- a/web/src/pages/Whitelist.tsx +++ b/web/src/pages/Whitelist.tsx @@ -1,10 +1,10 @@ import { useState } from "react"; import { useMutation, useQuery, useQueryClient, keepPreviousData } from "@tanstack/react-query"; -import { useNavigate } from "react-router"; -import { Plus, Search, ShieldCheck, Trash2 } from "lucide-react"; +import { useNavigate, useSearchParams } from "react-router"; +import { Plus, RefreshCw, Search, ShieldCheck, Trash2 } from "lucide-react"; import { api, qs } from "../lib/api"; -import type { DupHash, Paged, WhitelistRow } from "../lib/types"; -import { int, relative, shortHash } from "../lib/format"; +import type { DupResponse, Paged, WhitelistRow } from "../lib/types"; +import { int, ms, relative, seconds, shortHash } from "../lib/format"; import { Badge, Button, @@ -18,6 +18,7 @@ import { PageHeader, Pager, Segmented, + Spinner, Table, Td, Textarea, @@ -33,7 +34,9 @@ interface HashUsage { } export default function Whitelist() { - const [tab, setTab] = useState<"list" | "candidates">("list"); + const [sp, setSp] = useSearchParams(); + const tab: "list" | "candidates" = sp.get("tab") === "shared" ? "candidates" : "list"; + const setTab = (t: "list" | "candidates") => setSp(t === "candidates" ? { tab: "shared" } : {}, { replace: true }); const [inspect, setInspect] = useState(null); return (
@@ -190,7 +193,20 @@ function Candidates({ onInspect }: { onInspect: (h: string) => void }) { const [selected, setSelected] = useState>(new Set()); const dups = useQuery({ queryKey: ["dups", hide], - queryFn: () => api<{ asOf: number; items: DupHash[]; pending?: boolean }>(`/overview/duplicates${hide ? "?hideWhitelisted=1" : ""}`), + queryFn: () => api(`/overview/duplicates${hide ? "?hideWhitelisted=1" : ""}`), + // poll while the background job is running or has not produced anything yet + refetchInterval: (q) => { + const d = q.state.data; + return !d || d.running || (d.asOf === 0 && !d.error) ? 5000 : false; + }, + }); + const recompute = useMutation({ + mutationFn: () => api("/overview/duplicates/refresh", { method: "POST" }), + onSuccess: () => { + toast("Recomputing shared hashes", "neutral"); + setTimeout(() => qc.invalidateQueries({ queryKey: ["dups"] }), 500); + }, + onError: (e) => toast((e as Error).message, "critical"), }); const add = useMutation({ mutationFn: (hashes: string[]) => api<{ added: number }>("/whitelist", { method: "POST", body: { hashes, note: "from most shared hashes" } }), @@ -205,25 +221,54 @@ function Candidates({ onInspect }: { onInspect: (h: string) => void }) { if (dups.isLoading) return ; if (dups.error) return ; - const items = dups.data!.items; + const d = dups.data!; + const items = d.items; + + let subtitle: string; + if (d.running) subtitle = `Computing… started ${relative(d.startedAt)}`; + else if (d.asOf) + subtitle = `Computed ${relative(d.asOf)} in ${d.tookMs < 60000 ? ms(d.tookMs) : seconds(d.tookMs / 1000)} · ${int(d.sharedHashes)} shared of ${int(d.distinctHashes)} distinct hashes · refreshed hourly`; + else if (d.error) subtitle = "Not computed yet"; + else subtitle = "Waiting for the first run…"; return ( + {d.running && } + {subtitle} + + } actions={ <> + } > + {d.error && ( +
+ +
+ )} {items.length === 0 ? ( - No shared hashes + + {d.asOf + ? hide && d.sharedHashes > 0 + ? "All of the most shared hashes are whitelisted" + : "No hashes are shared between addons" + : d.running || !d.error + ? "Computing shared hashes — this scans every indexed file and can take a few minutes" + : "No result yet"} + ) : (