fix(compaction): unexport metric fields to prevent mutex bypass

RELEASE_NOTES_2026.06.2.md

@@ -28,6 +28,8 @@ Performance: CSV imports now parse each numeric value once (the type-inference a
 
 Line Protocol and TLE imports are unchanged. CSV/Parquet imports no longer depend on the DuckDB sandbox's allowed-directories list.
 
+**Compaction metric fields are no longer exported.** `Manager` (`TotalJobsCompleted`, `TotalJobsFailed`, `TotalFilesCompacted`, `TotalBytesSaved`, `TotalManifestsRecov`) and `BaseTier` (`TotalCompactions`, `TotalFilesCompacted`, `TotalBytesSaved`) previously had exported struct fields that could be read or written without going through the mutex-protected `Stats()` / `GetBaseStats()` accessors, creating a potential data race. The fields are now unexported; the `Stats()` and `GetBaseStats()` methods remain the sole access path and continue to hold the mutex. `BaseTier` also gains a thread-safe `RecordCompaction(filesCompacted int, bytesSaved int64)` helper for tier implementations to increment metrics under the lock.
+
 ## Performance improvements
 
 **Compaction cleanup now uses batch-delete APIs on S3 and Azure.** `deleteOldFiles` previously called `StorageBackend.Delete` once per compacted source file — on large compaction cycles (hundreds of files), this produced hundreds of sequential S3/Azure API calls. Both cloud backends already implemented `DeleteBatch` (S3 `DeleteObjects`, Azure `BlobBatch`) but the compaction cleanup path never used it. The path now prefers `BatchDeleter.DeleteBatch`, falling back to per-file `Delete` if the backend does not support batching. This reduces S3 DELETE API calls by up to 1000× and Azure calls by up to 256× on large compactions, with a proportional drop in cleanup-phase latency. Local-storage deployments are unaffected (a per-file loop is the correct implementation for a local filesystem).

internal/compaction/manager.go

@@ -53,11 +53,11 @@ type Manager struct {
 	cycleID      atomic.Int64
 
 	// Metrics
-	TotalJobsCompleted  int
-	TotalJobsFailed     int
-	TotalFilesCompacted int
-	TotalBytesSaved     int64
-	TotalManifestsRecov int // Number of manifests recovered
+	totalJobsCompleted  int
+	totalJobsFailed     int
+	totalFilesCompacted int
+	totalBytesSaved     int64
+	totalManifestsRecov int // Number of manifests recovered
 
 	// Callback invoked after a successful compaction job (in parent process).
 	// Used to invalidate DuckDB and query caches after files are deleted.
@@ -293,11 +293,11 @@ func (m *Manager) CompactPartition(ctx context.Context, candidate Candidate) err
 
 	m.mu.Lock()
 	if shouldInvalidateCache {
-		m.TotalJobsCompleted++
-		m.TotalFilesCompacted += result.FilesCompacted
-		m.TotalBytesSaved += (result.BytesBefore - result.BytesAfter)
+		m.totalJobsCompleted++
+		m.totalFilesCompacted += result.FilesCompacted
+		m.totalBytesSaved += (result.BytesBefore - result.BytesAfter)
 	} else {
-		m.TotalJobsFailed++
+		m.totalJobsFailed++
 	}
 
 	// Build job stats for history
@@ -534,7 +534,7 @@ func (m *Manager) runCycleInternal(ctx context.Context, filterDatabases []string
 		}
 		if recovered > 0 {
 			m.mu.Lock()
-			m.TotalManifestsRecov += recovered
+			m.totalManifestsRecov += recovered
 			m.mu.Unlock()
 			m.logger.Info().Int("recovered", recovered).Msg("Recovered orphaned compaction manifests")
 		}
@@ -897,12 +897,12 @@ func (m *Manager) Stats() map[string]interface{} {
 	defer m.mu.Unlock()
 
 	stats := map[string]interface{}{
-		"total_jobs_completed":    m.TotalJobsCompleted,
-		"total_jobs_failed":       m.TotalJobsFailed,
-		"total_files_compacted":   m.TotalFilesCompacted,
-		"total_bytes_saved":       m.TotalBytesSaved,
-		"total_bytes_saved_mb":    float64(m.TotalBytesSaved) / 1024 / 1024,
-		"total_manifests_recover": m.TotalManifestsRecov,
+		"total_jobs_completed":    m.totalJobsCompleted,
+		"total_jobs_failed":       m.totalJobsFailed,
+		"total_files_compacted":   m.totalFilesCompacted,
+		"total_bytes_saved":       m.totalBytesSaved,
+		"total_bytes_saved_mb":    float64(m.totalBytesSaved) / 1024 / 1024,
+		"total_manifests_recover": m.totalManifestsRecov,
 		"cycle_running":           m.cycleRunning.Load(),
 		"current_cycle_id":        m.cycleID.Load(),
 	}

internal/compaction/tier.go

@@ -97,9 +97,9 @@ type BaseTier struct {
 	Enabled        bool
 
 	// Metrics
-	TotalCompactions    int
-	TotalFilesCompacted int
-	TotalBytesSaved     int64
+	totalCompactions    int
+	totalFilesCompacted int
+	totalBytesSaved     int64
 
 	Logger zerolog.Logger
 	mu     sync.Mutex
@@ -143,13 +143,22 @@ func (t *BaseTier) GetBaseStats(tierName string) map[string]interface{} {
 		"min_age_hours":         t.MinAgeHours,
 		"min_files":             t.MinFiles,
 		"target_size_mb":        t.TargetSizeMB,
-		"total_compactions":     t.TotalCompactions,
-		"total_files_compacted": t.TotalFilesCompacted,
-		"total_bytes_saved":     t.TotalBytesSaved,
-		"total_bytes_saved_mb":  float64(t.TotalBytesSaved) / 1024 / 1024,
+		"total_compactions":     t.totalCompactions,
+		"total_files_compacted": t.totalFilesCompacted,
+		"total_bytes_saved":     t.totalBytesSaved,
+		"total_bytes_saved_mb":  float64(t.totalBytesSaved) / 1024 / 1024,
 	}
 }
 
+// RecordCompaction updates the tier's compaction metrics in a thread-safe manner.
+func (t *BaseTier) RecordCompaction(filesCompacted int, bytesSaved int64) {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	t.totalCompactions++
+	t.totalFilesCompacted += filesCompacted
+	t.totalBytesSaved += bytesSaved
+}
+
 // ShouldCompactByFileSuffix determines if compaction is needed based on file classification.
 // This is a shared helper that implements the common compaction decision logic:
 //   - compactedSuffix: suffix for files already compacted at this tier (e.g., "_compacted.parquet")