1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130

import hashlib
from dataclasses import dataclass
from enum import Enum
from typing import List, Optional
import zstandard as zstd
 
class IntegrityStatus(Enum):
    OK = 'ok'
    MISSING_CHUNK = 'missing_chunk'
    HASH_MISMATCH = 'hash_mismatch'
    DECOMPRESS_FAILED = 'decompress_failed'
    REFERENCE_COUNT_ERROR = 'reference_count_error'
 
@dataclass
class VerificationResult:
    status: IntegrityStatus
    chunk_hash: Optional[str] = None
    expected_hash: Optional[str] = None
    computed_hash: Optional[str] = None
    error_message: Optional[str] = None
 
class ChunkIntegrityVerifier:
    """
    Comprehensive chunk integrity verification.
    
    Implements multiple layers of protection:
    1. Existence check (chunk exists in storage)
    2. Hash verification (content matches fingerprint)
    3. Decompression test (compressed data is valid)
    4. Reference count validation (refcount >= actual references)
    """
    
    def __init__(self, chunk_storage, chunk_index, file_registry):
        self.storage = chunk_storage
        self.index = chunk_index
        self.registry = file_registry
        self.decompressor = zstd.ZstdDecompressor()
    
    def verify_chunk(self, chunk_hash: str) -> VerificationResult:
        """
        Full verification of a single chunk.
        """
        # Step 1: Existence check
        if not self.storage.chunk_exists(chunk_hash):
            return VerificationResult(
                status=IntegrityStatus.MISSING_CHUNK,
                chunk_hash=chunk_hash,
            )
        
        # Step 2: Read chunk data
        chunk_data = self.storage.read_chunk(chunk_hash)
        metadata = self.storage.read_chunk_metadata(chunk_hash)
        
        # Step 3: Hash verification
        if metadata.get('is_compressed', False):
            # Hash is of uncompressed content
            try:
                uncompressed = self.decompressor.decompress(chunk_data)
            except Exception as e:
                return VerificationResult(
                    status=IntegrityStatus.DECOMPRESS_FAILED,
                    chunk_hash=chunk_hash,
                    error_message=str(e),
                )
            computed_hash = hashlib.sha256(uncompressed).hexdigest()
        else:
            computed_hash = hashlib.sha256(chunk_data).hexdigest()
        
        if computed_hash != chunk_hash:
            return VerificationResult(
                status=IntegrityStatus.HASH_MISMATCH,
                chunk_hash=chunk_hash,
                expected_hash=chunk_hash,
                computed_hash=computed_hash,
            )
        
        # Step 4: Reference count validation
        stored_refcount = self.index.get_refcount(chunk_hash)
        actual_refs = self.registry.count_references_to_chunk(chunk_hash)
        
        if stored_refcount < actual_refs:
            # This is a serious bug - refcount too low could cause premature deletion
            return VerificationResult(
                status=IntegrityStatus.REFERENCE_COUNT_ERROR,
                chunk_hash=chunk_hash,
                error_message=f'Stored refcount {stored_refcount} < actual refs {actual_refs}',
            )
        
        return VerificationResult(status=IntegrityStatus.OK)
    
    def verify_file(self, file_id: str) -> List[VerificationResult]:
        """
        Verify all chunks referenced by a file.
        """
        results = []
        recipe = self.registry.get_file_recipe(file_id)
        
        for chunk_hash in recipe['chunks']:
            result = self.verify_chunk(chunk_hash)
            if result.status != IntegrityStatus.OK:
                results.append(result)
        
        return results  # Empty list means all OK
    
    def full_scrub(self, sample_rate: float = 1.0) -> dict:
        """
        Full storage scrub - verify all (or sampled) chunks.
        
        Should be run periodically (daily/weekly) as background maintenance.
        """
        import random
        
        all_chunks = self.index.list_all_chunks()
        errors = []
        checked = 0
        
        for chunk_hash in all_chunks:
            if random.random() > sample_rate:
                continue
            
            checked += 1
            result = self.verify_chunk(chunk_hash)
            if result.status != IntegrityStatus.OK:
                errors.append(result)
        
        return {
            'chunks_checked': checked,
            'errors_found': len(errors),
            'errors': errors,
        }

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153

from dataclasses import dataclass
from typing import List, Set, Dict
import logging
 
logger = logging.getLogger(__name__)
 
@dataclass
class RecoveryStats:
    orphaned_chunks: int = 0
    refcount_corrections: int = 0
    missing_chunks: int = 0
    corrupted_chunks: int = 0
 
class StorageRecovery:
    """
    Recovery operations for deduplicated storage.
    """
    
    def __init__(self, chunk_storage, chunk_index, file_registry):
        self.storage = chunk_storage
        self.index = chunk_index
        self.registry = file_registry
    
    def rebuild_reference_counts(self) -> RecoveryStats:
        """
        Full mark-and-sweep to rebuild reference counts.
        
        This is the authoritative recovery operation that fixes
        any refcount inconsistencies. Expensive but comprehensive.
        
        Process:
        1. Traverse all file recipes
        2. Count actual references to each chunk
        3. Compare to stored refcounts
        4. Fix any discrepancies
        """
        stats = RecoveryStats()
        
        # Phase 1: Mark - count actual references
        actual_refcounts: Dict[str, int] = {}
        
        logger.info("Phase 1: Scanning all file recipes...")
        for file_id in self.registry.list_all_files():
            recipe = self.registry.get_file_recipe(file_id)
            for chunk_hash in recipe['chunks']:
                actual_refcounts[chunk_hash] = actual_refcounts.get(chunk_hash, 0) + 1
        
        logger.info(f"Found {len(actual_refcounts)} referenced chunks")
        
        # Phase 2: Compare and fix
        logger.info("Phase 2: Verifying reference counts...")
        for chunk_hash, actual_count in actual_refcounts.items():
            stored_count = self.index.get_refcount(chunk_hash)
            
            if stored_count != actual_count:
                logger.warning(
                    f"Refcount mismatch for {chunk_hash[:16]}: "
                    f"stored={stored_count}, actual={actual_count}"
                )
                self.index.set_refcount(chunk_hash, actual_count)
                stats.refcount_corrections += 1
        
        # Phase 3: Find orphans (chunks with no references)
        logger.info("Phase 3: Identifying orphaned chunks...")
        for chunk_hash in self.index.list_all_chunks():
            if chunk_hash not in actual_refcounts:
                stats.orphaned_chunks += 1
                # Don't delete immediately - quarantine for safety
                self.index.mark_for_deletion(chunk_hash, delay_days=30)
        
        logger.info(f"Recovery complete: {stats}")
        return stats
    
    def rebuild_index_from_chunks(self) -> int:
        """
        Rebuild dedup index by scanning all stored chunks.
        
        Used when index is completely lost or corrupted.
        Very expensive - scans and hashes all stored data.
        """
        logger.info("Rebuilding index from chunk storage...")
        count = 0
        
        for chunk_path in self.storage.scan_all_chunks():
            # Read chunk
            data, metadata = self.storage.read_chunk_raw(chunk_path)
            
            # Compute hash
            if metadata.get('is_compressed', False):
                uncompressed = self._decompress(data, metadata)
                chunk_hash = self._hash(uncompressed)
            else:
                chunk_hash = self._hash(data)
            
            # Add to index
            self.index.add(chunk_hash, chunk_path, refcount=0)
            count += 1
            
            if count % 100000 == 0:
                logger.info(f"Indexed {count} chunks...")
        
        logger.info(f"Index rebuilt with {count} chunks")
        return count
    
    def verify_and_repair_file(self, file_id: str) -> dict:
        """
        Attempt to verify and repair a specific file.
        
        Returns status and any actions taken.
        """
        result = {
            'file_id': file_id,
            'status': 'ok',
            'actions': [],
            'unrecoverable_chunks': [],
        }
        
        recipe = self.registry.get_file_recipe(file_id)
        
        for i, chunk_hash in enumerate(recipe['chunks']):
            # Check chunk exists
            if not self.storage.chunk_exists(chunk_hash):
                # Try to recover from replica or backup
                recovered = self._attempt_chunk_recovery(chunk_hash)
                
                if recovered:
                    result['actions'].append(f"Recovered chunk {i} from replica")
                else:
                    result['unrecoverable_chunks'].append({
                        'index': i,
                        'hash': chunk_hash,
                    })
                    result['status'] = 'partial_loss'
            
            # Verify chunk integrity
            elif not self._verify_chunk_integrity(chunk_hash):
                # Chunk corrupted - try recovery
                recovered = self._attempt_chunk_recovery(chunk_hash)
                
                if recovered:
                    result['actions'].append(f"Recovered corrupted chunk {i}")
                else:
                    result['unrecoverable_chunks'].append({
                        'index': i,
                        'hash': chunk_hash,
                        'reason': 'corruption',
                    })
                    result['status'] = 'partial_loss'
        
        if result['unrecoverable_chunks']:
            logger.error(f"File {file_id} has {len(result['unrecoverable_chunks'])} unrecoverable chunks")
        
        return result

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118

from dataclasses import dataclass
from typing import Callable, Dict, Optional
from enum import Enum
 
class MigrationStrategy(Enum):
    COPY_ON_READ = 'copy_on_read'
    BACKGROUND = 'background'
    GENERATIONAL = 'generational'
 
@dataclass
class CompressionVersion:
    """Definition of a compression version."""
    version_id: int
    algorithm: str
    level: int
    compressor: Callable[[bytes], bytes]
    decompressor: Callable[[bytes], bytes]
    deprecated: bool = False
 
class VersionedCompressionManager:
    """
    Manage multiple compression versions for smooth upgrades.
    """
    
    def __init__(self):
        self.versions: Dict[int, CompressionVersion] = {}
        self.current_version: Optional[int] = None
        self.migration_strategy = MigrationStrategy.COPY_ON_READ
    
    def register_version(self, version: CompressionVersion):
        """Register a compression version (for reading old data)."""
        self.versions[version.version_id] = version
    
    def set_current(self, version_id: int):
        """Set the version to use for new writes."""
        if version_id not in self.versions:
            raise ValueError(f"Unknown version: {version_id}")
        self.current_version = version_id
    
    def compress(self, data: bytes) -> tuple[bytes, int]:
        """Compress using current version. Returns (compressed, version_id)."""
        version = self.versions[self.current_version]
        compressed = version.compressor(data)
        return compressed, self.current_version
    
    def decompress(self, data: bytes, version_id: int) -> bytes:
        """Decompress using specified version."""
        if version_id not in self.versions:
            raise ValueError(f"Unknown version {version_id} - cannot decompress")
        version = self.versions[version_id]
        return version.decompressor(data)
    
    def should_migrate(self, chunk_version: int) -> bool:
        """Check if a chunk should be migrated to current version."""
        if chunk_version == self.current_version:
            return False
        old_version = self.versions.get(chunk_version)
        return old_version and old_version.deprecated
    
    def migrate_chunk(
        self,
        chunk_data: bytes,
        old_version: int,
        chunk_storage,
        chunk_index,
    ) -> Optional[bytes]:
        """
        Migrate a chunk to current compression version.
        
        Returns new compressed data, or None if migration not needed.
        """
        if not self.should_migrate(old_version):
            return None
        
        # Decompress with old algorithm
        decompressed = self.decompress(chunk_data, old_version)
        
        # Recompress with current algorithm
        recompressed, new_version = self.compress(decompressed)
        
        # Only migrate if it actually helps
        if len(recompressed) >= len(chunk_data):
            return None  # New algorithm isn't better for this data
        
        return recompressed
 
 
# Example: Setting up versions
def setup_compression_versions():
    import lz4.frame
    import zstandard as zstd
    
    manager = VersionedCompressionManager()
    zstd_ctx = zstd.ZstdCompressor(level=3)
    zstd_dctx = zstd.ZstdDecompressor()
    
    # Legacy: LZ4 (still readable, no longer written)
    manager.register_version(CompressionVersion(
        version_id=1,
        algorithm='lz4',
        level=0,
        compressor=lz4.frame.compress,
        decompressor=lz4.frame.decompress,
        deprecated=True,  # Will migrate on read
    ))
    
    # Current: Zstandard level 3
    manager.register_version(CompressionVersion(
        version_id=2,
        algorithm='zstd',
        level=3,
        compressor=zstd_ctx.compress,
        decompressor=zstd_dctx.decompress,
        deprecated=False,
    ))
    
    manager.set_current(2)
    return manager

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141

import asyncio
from dataclasses import dataclass
from typing import List, Dict, Optional
import hashlib
from concurrent.futures import ThreadPoolExecutor
import bisect
 
class BatchedIndexLookup:
    """
    Batch multiple index lookups together to amortize overhead.
    
    Instead of individual queries, collect hashes and query in batch.
    Many index backends (Redis, databases) are much more efficient
    with batch operations.
    """
    
    def __init__(self, index_backend, batch_size: int = 1000):
        self.index = index_backend
        self.batch_size = batch_size
    
    def lookup_chunks(self, chunk_hashes: List[str]) -> Dict[str, Optional[str]]:
        """
        Lookup multiple chunks at once.
        
        Returns dict mapping hash -> location (None if not found).
        """
        results = {}
        
        # Process in batches
        for i in range(0, len(chunk_hashes), self.batch_size):
            batch = chunk_hashes[i:i + self.batch_size]
            
            # Single batch query to backend
            batch_results = self.index.mget(batch)
            
            for hash_val, location in zip(batch, batch_results):
                results[hash_val] = location
        
        return results
 
 
class ParallelHasher:
    """
    Parallelize hash computation across CPU cores.
    
    For large files, split into segments and hash in parallel.
    """
    
    def __init__(self, num_workers: int = 4):
        self.executor = ThreadPoolExecutor(max_workers=num_workers)
    
    def hash_chunks(self, chunks: List[bytes]) -> List[str]:
        """Hash multiple chunks in parallel."""
        def hash_one(data: bytes) -> str:
            return hashlib.sha256(data).hexdigest()
        
        futures = [self.executor.submit(hash_one, chunk) for chunk in chunks]
        return [f.result() for f in futures]
 
 
class ChunkContainerPacker:
    """
    Pack frequently co-accessed chunks into containers.
    
    When chunks from the same file are stored together, reading
    the file requires fewer I/O operations.
    """
    
    def __init__(self, container_size: int = 4 * 1024 * 1024):  # 4MB containers
        self.container_size = container_size
        self.current_container_id = 0
        self.current_container = bytearray()
        self.container_index: List[tuple] = []  # [(chunk_hash, offset, length), ...]
    
    def add_chunk(self, chunk_hash: str, chunk_data: bytes) -> tuple[int, int]:
        """
        Add chunk to current container.
        
        Returns (container_id, offset) for later retrieval.
        """
        if len(self.current_container) + len(chunk_data) > self.container_size:
            # Seal container and start new one
            self._seal_container()
        
        offset = len(self.current_container)
        self.current_container.extend(chunk_data)
        
        self.container_index.append((chunk_hash, offset, len(chunk_data)))
        
        return self.current_container_id, offset
    
    def _seal_container(self):
        """Finalize container and prepare for new one."""
        # Write container to storage...
        self.current_container_id += 1
        self.current_container = bytearray()
        self.container_index = []
 
 
class ReadAheadBuffer:
    """
    Prefetch chunks predicted to be needed soon.
    
    When reading a file sequentially, chunks later in the recipe
    are likely to be needed soon. Prefetch them during idle time.
    """
    
    def __init__(self, chunk_storage, buffer_chunks: int = 32):
        self.storage = chunk_storage
        self.buffer_size = buffer_chunks
        self.buffer: Dict[str, bytes] = {}
        self.prefetch_queue: List[str] = []
    
    async def read_chunk(self, chunk_hash: str) -> bytes:
        """Read chunk, using buffer if available."""
        if chunk_hash in self.buffer:
            return self.buffer.pop(chunk_hash)
        
        # Synchronous read (cache miss)
        return await self.storage.read_chunk_async(chunk_hash)
    
    def hint_upcoming(self, chunk_hashes: List[str]):
        """Hint that these chunks will be needed soon."""
        for h in chunk_hashes:
            if h not in self.buffer and h not in self.prefetch_queue:
                self.prefetch_queue.append(h)
        
        # Trigger background prefetch
        self._start_prefetch()
    
    def _start_prefetch(self):
        """Start background prefetch task."""
        while self.prefetch_queue and len(self.buffer) < self.buffer_size:
            chunk_hash = self.prefetch_queue.pop(0)
            # Schedule async read...
            asyncio.create_task(self._prefetch_one(chunk_hash))
    
    async def _prefetch_one(self, chunk_hash: str):
        data = await self.storage.read_chunk_async(chunk_hash)
        if len(self.buffer) < self.buffer_size:
            self.buffer[chunk_hash] = data

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144

from dataclasses import dataclass, field
from typing import Dict, List
import time
from prometheus_client import Gauge, Counter, Histogram
 
# Prometheus metrics for storage system
METRICS = {
    # Capacity
    'physical_bytes_used': Gauge('storage_physical_bytes_used', 'Physical storage used'),
    'logical_bytes_stored': Gauge('storage_logical_bytes_stored', 'Logical data stored'),
    'unique_chunks': Gauge('storage_unique_chunks', 'Number of unique chunks'),
    'dedup_ratio': Gauge('storage_dedup_ratio', 'Current deduplication ratio'),
    'compression_ratio': Gauge('storage_compression_ratio', 'Current compression ratio'),
    
    # Performance
    'write_bytes': Counter('storage_write_bytes_total', 'Total bytes written'),
    'read_bytes': Counter('storage_read_bytes_total', 'Total bytes read'),
    'hash_operations': Counter('storage_hash_operations_total', 'Total hash operations'),
    'index_lookups': Counter('storage_index_lookups_total', 'Total index lookups'),
    'index_hits': Counter('storage_index_hits_total', 'Index lookup cache hits'),
    
    'write_latency': Histogram('storage_write_latency_seconds', 'Write latency'),
    'read_latency': Histogram('storage_read_latency_seconds', 'Read latency'),
    'decompress_latency': Histogram('storage_decompress_latency_seconds', 'Decompression latency'),
    
    # Health
    'verification_errors': Counter('storage_verification_errors_total', 'Verification errors found'),
    'recovery_operations': Counter('storage_recovery_ops_total', 'Recovery operations performed'),
    'gc_chunks_reclaimed': Counter('storage_gc_chunks_reclaimed', 'Chunks reclaimed by GC'),
}
 
@dataclass
class AlertThreshold:
    """Alert threshold configuration."""
    metric: str
    warning: float
    critical: float
    comparison: str = 'gt'  # 'gt', 'lt', 'eq'
 
class StorageAlerting:
    """
    Alerting system for storage health.
    """
    
    # Default thresholds
    THRESHOLDS = [
        AlertThreshold('capacity_percent', warning=80, critical=90),
        AlertThreshold('dedup_ratio', warning=1.5, critical=1.2, comparison='lt'),
        AlertThreshold('index_hit_rate', warning=0.7, critical=0.5, comparison='lt'),
        AlertThreshold('verification_error_rate', warning=0.001, critical=0.01),
        AlertThreshold('days_until_full', warning=30, critical=7, comparison='lt'),
    ]
    
    def __init__(self, metrics_collector, alert_channels: List):
        self.metrics = metrics_collector
        self.channels = alert_channels
    
    def check_thresholds(self) -> List[dict]:
        """Check all thresholds and generate alerts."""
        alerts = []
        
        current_metrics = self.metrics.collect_current()
        
        for threshold in self.THRESHOLDS:
            value = current_metrics.get(threshold.metric)
            if value is None:
                continue
            
            severity = self._evaluate_threshold(value, threshold)
            
            if severity:
                alerts.append({
                    'metric': threshold.metric,
                    'value': value,
                    'severity': severity,
                    'threshold': threshold.critical if severity == 'critical' else threshold.warning,
                    'timestamp': time.time(),
                })
        
        # Send alerts
        for alert in alerts:
            for channel in self.channels:
                channel.send(alert)
        
        return alerts
    
    def _evaluate_threshold(self, value: float, threshold: AlertThreshold) -> str:
        """Evaluate value against threshold."""
        if threshold.comparison == 'gt':
            if value >= threshold.critical:
                return 'critical'
            elif value >= threshold.warning:
                return 'warning'
        elif threshold.comparison == 'lt':
            if value <= threshold.critical:
                return 'critical'
            elif value <= threshold.warning:
                return 'warning'
        return None
 
 
class OperationalDashboard:
    """
    Key metrics for operational dashboard.
    """
    
    def generate_summary(self, storage_system) -> dict:
        """Generate dashboard summary."""
        stats = storage_system.get_stats()
        
        return {
            # Capacity overview
            'physical_capacity_tb': stats['physical_capacity'] / (1024**4),
            'physical_used_tb': stats['physical_used'] / (1024**4),
            'logical_stored_tb': stats['logical_stored'] / (1024**4),
            'capacity_percent': (stats['physical_used'] / stats['physical_capacity']) * 100,
            
            # Efficiency
            'data_reduction_ratio': stats['logical_stored'] / max(stats['physical_used'], 1),
            'dedup_ratio': stats['dedup_ratio'],
            'compression_ratio': stats['compression_ratio'],
            
            # Performance (last hour)
            'write_throughput_mbps': stats['write_bytes_last_hour'] / (1024**2) / 3600,
            'read_throughput_mbps': stats['read_bytes_last_hour'] / (1024**2) / 3600,
            'avg_read_latency_ms': stats['avg_read_latency'] * 1000,
            'p99_read_latency_ms': stats['p99_read_latency'] * 1000,
            
            # Health
            'chunk_count': stats['unique_chunks'],
            'index_cache_hit_rate': stats['index_cache_hit_rate'],
            'pending_gc_chunks': stats['gc_queue_size'],
            'last_scrub_errors': stats['last_scrub_errors'],
            'days_until_full': self._project_capacity(stats),
        }
    
    def _project_capacity(self, stats: dict) -> float:
        """Project days until capacity exhaustion."""
        remaining = stats['physical_capacity'] - stats['physical_used']
        daily_growth = stats.get('daily_growth_bytes', 0)
        
        if daily_growth <= 0:
            return float('inf')
        return remaining / daily_growth