Implement production-grade sharded trigger processor with backward compatibility

Addresses critical scalability issues identified in senior engineering review while
ensuring zero-breaking changes for existing indexers.

**Key Performance Fixes:**
- Fixed semaphore permit lifetime - permits now held during entire processing duration
- Implemented consistent deployment-based sharding using proper hash distribution
- Added comprehensive backpressure mechanism with exponential backoff
- Enhanced monitoring with per-shard metrics and load imbalance tracking

**Backward Compatibility (Zero Surprise):**
- Sharding is opt-in only (disabled by default)
- Legacy single-semaphore behavior preserved when GRAPH_SUBGRAPH_RUNTIME_PROCESSING_SHARDS=1
- Existing indexers see no changes without explicit configuration
- All tests pass with legacy mode enabled by default

**Scalability Improvements:**
- Before: 32 workers total (1.3% concurrent capacity for 2500 subgraphs)
- After: 1024 workers (32 shards × 32 workers) when sharding enabled
- Recommended for deployments with 2500+ subgraphs on 32+ vCPU systems
- Deployment-consistent sharding ensures optimal cache locality

**Environment Configuration:**
```bash
# Legacy mode (default - no changes)
# GRAPH_SUBGRAPH_RUNTIME_PROCESSING_SHARDS=1 (or unset)

# High-scale mode (opt-in)
export GRAPH_SUBGRAPH_RUNTIME_PROCESSING_SHARDS=32
export GRAPH_SUBGRAPH_RUNTIME_WORKERS_PER_SHARD=32
export GRAPH_SUBGRAPH_MAX_QUEUE_PER_SUBGRAPH=100
```

**Observability Features:**
- Per-shard utilization and throughput metrics
- Queue depth monitoring with backpressure alerts
- Shard load imbalance detection and reporting
- Clear logging of sharding mode and configuration

The implementation maintains full API compatibility while providing a 32x improvement
in concurrent processing capacity for high-density subgraph deployments.

🤖 Generated with [Claude Code](https://claude.ai/code)

Author: DaMandal0rian

core/src/subgraph/instance_manager.rs

@@ -184,13 +184,30 @@ impl<S: SubgraphStore> SubgraphInstanceManager<S> {
         let logger = logger_factory.component_logger("SubgraphInstanceManager", None);
         let logger_factory = logger_factory.with_parent(logger.clone());
 
-        // Configure sharded processor
+        // Configure trigger processor with backward compatibility
+        // Only enable sharding if explicitly configured with more than 1 shard
+        let enable_sharding = env_vars.subgraph_runtime_processing_shards > 1;
+
         let processor_config = super::trigger_processor::TriggerProcessorConfig {
+            enable_sharding,
             num_shards: env_vars.subgraph_runtime_processing_shards,
             workers_per_shard: env_vars.subgraph_runtime_workers_per_shard,
             max_queue_per_subgraph: env_vars.subgraph_max_queue_per_subgraph,
             fairness_window_ms: 100, // 100ms fairness window
         };
+
+        if enable_sharding {
+            info!(&logger, "Sharded trigger processing enabled";
+                "num_shards" => processor_config.num_shards,
+                "workers_per_shard" => processor_config.workers_per_shard,
+                "total_workers" => processor_config.num_shards * processor_config.workers_per_shard
+            );
+        } else {
+            info!(&logger, "Using legacy single-semaphore trigger processing";
+                "workers" => processor_config.workers_per_shard
+            );
+        }
+
         let trigger_processor = Arc::new(super::trigger_processor::SubgraphTriggerProcessor::new(
             processor_config,
         ));