fix(compaction): add manifest-based tracking to prevent data duplication

[khalid244]

Summary

Fixes #157 - Compaction causes data duplication when deletion fails or pod restarts.

This PR implements manifest-based tracking to ensure atomic compaction operations with crash recovery support.

Problem

The compaction process was not atomic. When a pod crashed after uploading the compacted file but before deleting source files, both original and compacted files remained in storage, causing:

• Data duplication in query results (400k rows → 800k rows)
• Storage waste from duplicate files
• Compounding duplication in subsequent compaction cycles

Solution

Implement a manifest file system that tracks compaction state and enables crash recovery:

_compaction_state/
├── hourly/
│   └── {database}/
│       └── {partition}_{job_id}.json
└── daily/
    └── {database}/
        └── {partition}_{job_id}.json

Compaction Flow (Modified)

1. Download input files
2. Compact → create output file locally
3. Write manifest: { output_path, output_size, input_files[], status: "pending" }
4. Upload output file
5. Delete input files
6. Delete manifest (compaction complete)

Recovery Flow (On Startup / Each Cycle)

1. Scan for orphaned manifests in _compaction_state/
2. For each manifest:
   - Output missing → delete manifest, retry compaction next cycle
   - Output exists + size matches → complete deletion of inputs, delete manifest
   - Output size mismatch → delete partial output, delete manifest, retry
3. Filter candidates to exclude files tracked by manifests

Changes

File	Changes
internal/compaction/manifest.go	New ManifestManager for read/write/recovery
internal/compaction/manifest_test.go	Comprehensive tests for manifest operations
internal/compaction/job.go	Write manifest before upload, delete after success
internal/compaction/manager.go	Run recovery at cycle start, filter candidates
internal/compaction/subprocess.go	Create ManifestManager for subprocess jobs

Manifest File Format

{
  "output_path": "metrics/cpu/2025/01/15/10/cpu_compacted.parquet",
  "output_size": 52428800,
  "input_files": [
    "metrics/cpu/2025/01/15/10/file1.parquet",
    "metrics/cpu/2025/01/15/10/file2.parquet"
  ],
  "database": "metrics",
  "measurement": "cpu",
  "partition_path": "metrics/cpu/2025/01/15/10",
  "tier": "hourly",
  "status": "pending",
  "created_at": "2025-01-15T10:30:00Z",
  "job_id": "metrics_cpu_..._1737834567890"
}

Recovery Scenarios

Crash Point	Output	Inputs	Recovery Action
Before upload	Missing	Exist	Delete manifest, retry next cycle
During upload	Partial	Exist	Delete partial + manifest, retry
After upload	Valid	Exist	Delete inputs, delete manifest ✓
After delete	Valid	Gone	Just delete manifest

Test plan

• All existing compaction tests pass
• New manifest tests pass:
  • TestManifestManager_WriteAndRead
  • TestManifestManager_ListManifests
  • TestManifestManager_GetFilesInManifests
  • TestManifestManager_RecoverOrphanedManifests_OutputMissing
  • TestManifestManager_RecoverOrphanedManifests_OutputExists
  • TestManifestManager_IsFileInManifest
• Deployed to test cluster (registry.gyoom.co/arc:manifest-fix)
• Health check passing

[xe-nvdk]

Approved: Manifest-Based Compaction Recovery

This PR correctly addresses issue #157 (data duplication during compaction failures). The manifest-based approach is the right solution.

What I Like

1. Manifests stored in S3 - Preserves compute/storage separation. Any pod can pick up recovery.
2. Size validation - Uses ObjectLister to detect partial uploads
3. Comprehensive tests - Good coverage of recovery scenarios
4. Candidate filtering - Files tracked by manifests are excluded from re-compaction

Minor Suggestions (Non-blocking)

1. Stale manifest TTL - Consider adding a max age (e.g., 7 days) after which manifests are deleted. Very old manifests likely indicate a deeper problem.

2. Metrics - Consider adding a Prometheus counter for compaction_manifests_recovered_total to track recovery frequency.

3. Documentation - The _compaction_state/ directory should be mentioned in the architecture docs for users doing bucket migrations or disaster recovery.

Note for Follow-up

The local temp directory (./data/compaction/) cleanup on startup is a separate concern - I'll open a separate issue for that.

Nice work on the design and implementation!