Detect rapid submission replays

internal/anomalydetection/anomalydetection.go

@@ -29,6 +29,8 @@ const (
 	maxDisplayItems               = 10
 	maxCorrelationDisplay         = 5
 	maxCorrelationPackages        = 8
+	rapidReplayWindow             = 2 * time.Hour
+	rapidReplayThreshold          = 0.1
 )
 
 type (
@@ -62,6 +64,11 @@ type (
 		PackagesAboveThreshold []PackageRatio
 	}
 
+	RapidReplayResult struct {
+		Reports      int
+		TotalReports int
+	}
+
 	DetectionResult struct {
 		CountCorrelations   []CountCorrelation
 		NewPackageSpikes    []Spike
@@ -70,6 +77,7 @@ type (
 		SystemArchAnomalies []GrowthAnomaly
 		OSArchAnomalies     []GrowthAnomaly
 		BasePackageResult   BasePackageResult
+		RapidReplays        RapidReplayResult
 	}
 )
 
@@ -94,10 +102,22 @@ func (r *DetectionResult) HasExtremeMirrorGrowth() bool {
 	return false
 }
 
+func (r *RapidReplayResult) Share() float64 {
+	if r.TotalReports == 0 {
+		return 0
+	}
+	return float64(r.Reports) / float64(r.TotalReports) * 100
+}
+
+func (r *RapidReplayResult) NeedsInvestigation() bool {
+	return r.Share() >= rapidReplayThreshold
+}
+
 func (r *DetectionResult) IsHighConfidence() bool {
 	return r.BasePackageResult.HasAnomalies() ||
 		(r.HasMirrorAnomalies() && r.HasArchitectureAnomalies()) ||
-		r.HasExtremeMirrorGrowth()
+		r.HasExtremeMirrorGrowth() ||
+		r.RapidReplays.NeedsInvestigation()
 }
 
 // Run executes the detect-anomalies subcommand. args are os.Args[2:].
@@ -234,6 +254,11 @@ func detect(ctx context.Context, db *sql.DB, targetMonth, baselineStart, baselin
 		return nil, fmt.Errorf("base package anomalies: %w", err)
 	}
 
+	rapidReplays, err := detectRapidReplays(ctx, db, targetMonth)
+	if err != nil {
+		return nil, fmt.Errorf("rapid replays: %w", err)
+	}
+
 	return &DetectionResult{
 		CountCorrelations:   countCorrelations,
 		NewPackageSpikes:    newPackageSpikes,
@@ -242,9 +267,33 @@ func detect(ctx context.Context, db *sql.DB, targetMonth, baselineStart, baselin
 		SystemArchAnomalies: systemArchAnomalies,
 		OSArchAnomalies:     osArchAnomalies,
 		BasePackageResult:   basePackageResult,
+		RapidReplays:        rapidReplays,
 	}, nil
 }
 
+func detectRapidReplays(ctx context.Context, db *sql.DB, month int) (RapidReplayResult, error) {
+	result := RapidReplayResult{}
+	windowSeconds := int(rapidReplayWindow.Seconds())
+	err := db.QueryRowContext(ctx, `
+		WITH reports AS (
+			SELECT timestamp, ip, payload_hash, json_extract(headers, '$.User-Agent') AS user_agent
+			FROM submission_log
+			WHERE month = ?
+		), ordered AS (
+			SELECT timestamp, LAG(timestamp) OVER (
+				PARTITION BY ip, payload_hash, user_agent
+				ORDER BY timestamp
+			) AS previous_timestamp
+			FROM reports
+		)
+		SELECT COUNT(*), COALESCE(SUM(timestamp - previous_timestamp < ?), 0)
+		FROM ordered`, month, windowSeconds).Scan(&result.TotalReports, &result.Reports)
+	if err != nil {
+		return RapidReplayResult{}, err
+	}
+	return result, nil
+}
+
 func detectCountCorrelations(ctx context.Context, db *sql.DB, targetMonth, previousMonth int) ([]CountCorrelation, error) {
 	query := `
 		WITH deltas AS (
@@ -514,6 +563,7 @@ func findPackagesAboveBaseThreshold(ctx context.Context, db *sql.DB, targetMonth
 
 func renderResults(result *DetectionResult) {
 	renderBasePackageAnomalies(&result.BasePackageResult)
+	renderRapidReplays(&result.RapidReplays)
 	renderGrowthAnomalies("Mirror Anomalies", result.MirrorAnomalies)
 	renderSpikes("New Mirror Spikes", result.NewMirrorSpikes)
 	renderArchitectureAnomalies(result)
@@ -526,6 +576,19 @@ func renderResults(result *DetectionResult) {
 	renderSummary(result)
 }
 
+func renderRapidReplays(result *RapidReplayResult) {
+	if result.Reports == 0 {
+		return
+	}
+
+	severity := "WARNING"
+	if result.NeedsInvestigation() {
+		severity = "ERROR"
+	}
+	fmt.Printf("%s: %d rapid exact replays within %s (%.3f%% of %d logged reports)\n\n",
+		severity, result.Reports, rapidReplayWindow, result.Share(), result.TotalReports)
+}
+
 func renderBasePackageAnomalies(result *BasePackageResult) {
 	if !result.HasAnomalies() {
 		return
@@ -631,6 +694,9 @@ func renderSummary(result *DetectionResult) {
 		if result.BasePackageResult.HasAnomalies() {
 			typeCount++
 		}
+		if result.RapidReplays.NeedsInvestigation() {
+			typeCount++
+		}
 		fmt.Printf("ERROR: High-confidence anomalies detected (%d types) - requires investigation\n", typeCount)
 	case result.HasMirrorAnomalies() || result.HasArchitectureAnomalies():
 		fmt.Println("WARNING: Minor anomalies detected (single mirror or architecture spike - may be legitimate)")

internal/anomalydetection/anomalydetection_test.go

@@ -155,3 +155,34 @@ func TestDetectionResult_IsHighConfidence(t *testing.T) {
 		}
 	})
 }
+
+func TestDetectRapidReplays(t *testing.T) {
+	db, err := database.New(":memory:")
+	if err != nil {
+		t.Fatalf("failed to create database: %v", err)
+	}
+	defer func() { _ = db.Close() }()
+
+	for _, timestamp := range []int{0, 60, 120, 8000, 8001} {
+		_, err := db.Exec(`
+			INSERT INTO submission_log (month, timestamp, ip, headers, payload, payload_hash, country)
+			VALUES (202501, ?, '203.0.113.1', '{"User-Agent":"pkgstats/3.5.4"}', '{}', 'hash', 'DE')`, timestamp)
+		if err != nil {
+			t.Fatalf("insert submission log: %v", err)
+		}
+	}
+
+	result, err := detectRapidReplays(context.Background(), db, 202501)
+	if err != nil {
+		t.Fatalf("detect rapid replays: %v", err)
+	}
+	if result.TotalReports != 5 {
+		t.Errorf("total reports = %d, want 5", result.TotalReports)
+	}
+	if result.Reports != 3 {
+		t.Errorf("rapid replays = %d, want 3", result.Reports)
+	}
+	if !result.NeedsInvestigation() {
+		t.Error("rapid replays should require investigation")
+	}
+}

justfile

@@ -121,8 +121,8 @@ fixtures:
     go run ./cmd/fixtures
 
 # detect anomalies in submission data
-detect-anomalies:
-    go run . detect-anomalies
+detect-anomalies *args:
+    go run . detect-anomalies {{ args }}
 
 # prune submission log entries past the retention window
 prune-submission-log: