refactor(did_index): improve compaction strategy

src/did_index.rs

···
     // LRU cache for hot shards
     shard_cache: Arc<RwLock<HashMap<u8, Arc<Shard>>>>,
     max_cache: usize,
-    max_segments_per_shard: usize,
 
     config: Arc<RwLock<Config>>,
 
···
             config_path,
             shard_cache: Arc::new(RwLock::new(HashMap::new())),
             max_cache: 5,
-            max_segments_per_shard: 8,
             config: Arc::new(RwLock::new(config.clone())),
             cache_hits: AtomicI64::new(0),
             cache_misses: AtomicI64::new(0),
···
                 }
 
                 let compacted = if meta_opt.is_some() {
-                    self.auto_compact_if_needed(shard_num)?
+                    self.compact_shard(shard_num)?;
+                    true
                 } else {
                     false
                 };
···
     }
 
     /// Compact pending delta segments. If `shards` is `None`, all shards are compacted.
+    /// Uses parallel execution across shards for faster compaction.
     pub fn compact_pending_segments(&self, shards: Option<Vec<u8>>) -> Result<()> {
+        use rayon::prelude::*;
+
         match shards {
             Some(list) if !list.is_empty() => {
-                for shard in list {
-                    self.compact_shard(shard)?;
-                }
+                list.into_par_iter()
+                    .map(|shard| self.compact_shard(shard))
+                    .collect::<Result<Vec<_>>>()?;
             }
             _ => {
-                for shard in 0..DID_SHARD_COUNT {
-                    self.compact_shard(shard as u8)?;
-                }
+                (0..DID_SHARD_COUNT)
+                    .into_par_iter()
+                    .map(|shard| self.compact_shard(shard as u8))
+                    .collect::<Result<Vec<_>>>()?;
             }
         }
 
···
         Ok(layers)
     }
 
-    fn auto_compact_if_needed(&self, shard_num: u8) -> Result<bool> {
-        let should_compact = {
-            let config = self.config.read().unwrap();
-            config
-                .shards
-                .get(shard_num as usize)
-                .map(|meta| meta.segments.len() >= self.max_segments_per_shard)
-                .unwrap_or(false)
-        };
-
-        if should_compact {
-            self.compact_shard(shard_num)?;
-            Ok(true)
-        } else {
-            Ok(false)
-        }
-    }
+    // removed: auto_compact_if_needed (compaction now happens immediately after updates)
 
     fn compact_shard(&self, shard_num: u8) -> Result<()> {
         use std::time::Instant;
···
                 .get("delta_segments")
                 .and_then(|v| v.as_u64())
                 .unwrap_or(0);
-            assert!(
-                delta_segments > 0,
-                "expected pending delta segments, got {}",
-                delta_segments
+            assert_eq!(
+                delta_segments, 0,
+                "expected no pending delta segments after immediate compaction"
             );
         }
 

src/manager.rs

···
     ///
     /// IMPORTANT: This method performs heavy blocking I/O and should be called from async
     /// contexts using spawn_blocking to avoid freezing the async runtime (and HTTP server).
-    pub fn batch_update_did_index(&self, start_bundle: u32, end_bundle: u32) -> Result<()> {
+    pub fn batch_update_did_index(
+        &self,
+        start_bundle: u32,
+        end_bundle: u32,
+        compact: bool,
+    ) -> Result<()> {
         use std::time::Instant;
 
         if start_bundle > end_bundle {
···
 
         let total_start = Instant::now();
         let bundle_count = end_bundle - start_bundle + 1;
+        if bundle_count > 10 {
+            use std::time::Instant;
+            eprintln!(
+                "[DID Index] Rebuild triggered for {} bundles ({} → {})",
+                bundle_count, start_bundle, end_bundle
+            );
+            let rebuild_start = Instant::now();
+            let _ = self.build_did_index(
+                crate::constants::DID_INDEX_FLUSH_INTERVAL,
+                Some(
+                    |current: u32, total: u32, bytes_processed: u64, total_bytes: u64| {
+                        let percent = if total_bytes > 0 {
+                            (bytes_processed as f64 / total_bytes as f64) * 100.0
+                        } else {
+                            0.0
+                        };
+                        eprintln!(
+                            "[DID Index] Rebuild progress: {}/{} ({:.1}%)",
+                            current, total, percent
+                        );
+                    },
+                ),
+                None,
+                None,
+            )?;
+            let dur = rebuild_start.elapsed();
+            eprintln!("[DID Index] Rebuild complete in {:.1}s", dur.as_secs_f64());
+            return Ok(());
+        }
 
         if *self.verbose.lock().unwrap() {
             log::info!(
···
             total_operations as f64 / update_duration.as_secs_f64()
         );
 
+        // Optionally compact all shards immediately to avoid leaving delta segments
+        if compact {
+            let idx_guard = self.did_index.read().unwrap();
+            if let Some(idx) = idx_guard.as_ref() {
+                idx.compact_pending_segments(None)?;
+            }
+        }
+
         let total_duration = total_start.elapsed();
 
         if *self.verbose.lock().unwrap() {
···
         &self,
         start_bundle: u32,
         end_bundle: u32,
+        compact: bool,
     ) -> Result<()> {
         let manager = self.clone_for_arc();
 
-        tokio::task::spawn_blocking(move || {
-            manager.batch_update_did_index(start_bundle, end_bundle)
+        // First perform the batch update in a blocking task
+        let _ = tokio::task::spawn_blocking(move || {
+            manager.batch_update_did_index(start_bundle, end_bundle, compact)
         })
         .await
-        .map_err(|e| anyhow::anyhow!("Batch DID index update task failed: {}", e))?
+        .map_err(|e| anyhow::anyhow!("Batch DID index update task failed: {}", e))?;
+
+        Ok(())
     }
 
     /// Fetch and save next bundle from PLC directory
···
         &self,
         client: &crate::plc_client::PLCClient,
         shutdown_rx: Option<tokio::sync::watch::Receiver<bool>>,
+        update_did_index: bool,
     ) -> Result<SyncResult> {
         use crate::sync::{get_boundary_cids, strip_boundary_duplicates};
         use std::time::Instant;
···
             }
 
             let fetch_op_start = Instant::now();
-            if let Some(ref rx) = shutdown_rx && *rx.borrow() {
+            if let Some(ref rx) = shutdown_rx
+                && *rx.borrow()
+            {
                 anyhow::bail!("Shutdown requested");
             }
             let (plc_ops, wait_dur, http_dur) = if let Some(rx) = shutdown_rx.clone() {
···
             index_write_time,
             did_index_compacted,
         ) = self
-            .save_bundle_with_timing(next_bundle_num, operations)
+            .save_bundle_with_timing(next_bundle_num, operations, update_did_index)
             .await?;
         let save_duration = save_start.elapsed();
 
···
         let total_duration_ms = (fetch_total_duration + save_duration).as_millis() as u64;
         let fetch_duration_ms = fetch_total_duration.as_millis() as u64;
 
-        // Calculate separate timings: bundle save (serialize + compress + hash) vs index (did_index + index_write)
-        let bundle_save_ms = (serialize_time + compress_time + hash_time).as_millis() as u64;
-        let index_ms = (did_index_time + index_write_time).as_millis() as u64;
+        // Calculate separate timings: bundle save vs index write/DID index
+        let (bundle_save_ms, index_ms) = if update_did_index {
+            (
+                (serialize_time + compress_time + hash_time).as_millis() as u64,
+                (did_index_time + index_write_time).as_millis() as u64,
+            )
+        } else {
+            (
+                (serialize_time + compress_time + hash_time + index_write_time).as_millis() as u64,
+                0,
+            )
+        };
 
         // Only log detailed info in verbose mode
         if *self.verbose.lock().unwrap() {
···
         let mut synced = 0;
 
         loop {
-            match self.sync_next_bundle(client, None).await {
+            match self.sync_next_bundle(client, None, true).await {
                 Ok(SyncResult::BundleCreated { .. }) => {
                     synced += 1;
 
···
         &self,
         bundle_num: u32,
         operations: Vec<Operation>,
+        update_did_index: bool,
     ) -> Result<(
         std::time::Duration,
         std::time::Duration,
···
             );
         }
 
-        // Update DID index (now fast with delta segments)
-        let did_index_start = Instant::now();
-        let did_ops: Vec<(String, bool)> = operations
-            .iter()
-            .map(|op| (op.did.clone(), op.nullified))
-            .collect();
+        let (did_index_time, did_index_compacted) = if update_did_index {
+            let did_index_start = Instant::now();
+            let did_ops: Vec<(String, bool)> = operations
+                .iter()
+                .map(|op| (op.did.clone(), op.nullified))
+                .collect();
 
-        self.ensure_did_index()?;
-        let did_index_compacted = self
-            .did_index
-            .write()
-            .unwrap()
-            .as_mut()
-            .unwrap()
-            .update_for_bundle(bundle_num, did_ops)?;
-        let did_index_time = did_index_start.elapsed();
+            self.ensure_did_index()?;
+            let compacted = self
+                .did_index
+                .write()
+                .unwrap()
+                .as_mut()
+                .unwrap()
+                .update_for_bundle(bundle_num, did_ops)?;
+            (did_index_start.elapsed(), compacted)
+        } else {
+            (std::time::Duration::from_millis(0), false)
+        };
 
         // Update main index
         let index_write_start = Instant::now();

src/plc_client.rs

···
         after: &str,
         count: usize,
     ) -> Result<(Vec<PLCOperation>, Duration, Duration)> {
-        self.fetch_operations_with_retry_cancelable(after, count, 5, None).await
+        self.fetch_operations_with_retry_cancelable(after, count, 5, None)
+            .await
     }
 
     pub async fn fetch_operations_cancelable(
···
         count: usize,
         shutdown_rx: Option<tokio::sync::watch::Receiver<bool>>,
     ) -> Result<(Vec<PLCOperation>, Duration, Duration)> {
-        self.fetch_operations_with_retry_cancelable(after, count, 5, shutdown_rx).await
+        self.fetch_operations_with_retry_cancelable(after, count, 5, shutdown_rx)
+            .await
     }
 
     async fn fetch_operations_with_retry_cancelable(
···
         let mut total_http = Duration::from_secs(0);
 
         for attempt in 1..=max_retries {
-            if let Some(ref rx) = shutdown_rx && *rx.borrow() {
+            if let Some(ref rx) = shutdown_rx
+                && *rx.borrow()
+            {
                 anyhow::bail!("Shutdown requested");
             }
-            let export_url = format!(
-                "{}/export?after={}&count={}",
-                self.base_url, after, count
-            );
+            let export_url = format!("{}/export?after={}&count={}", self.base_url, after, count);
 
             let permits = self.rate_limiter.available_permits();
             let requests_in_period = self.count_requests_in_period();
···
         let semaphore = std::sync::Arc::new(tokio::sync::Semaphore::new(0));
         let sem_clone = semaphore.clone();
 
-        let refill_rate = Duration::from_secs_f64(period.as_secs_f64() / requests_per_period as f64);
+        let refill_rate =
+            Duration::from_secs_f64(period.as_secs_f64() / requests_per_period as f64);
 
         // Spawn background task to refill permits at steady rate
         // CRITICAL: Add first permit immediately, then refill at steady rate

src/sync.rs

···
         } else {
             format!("{:.0}KB", size_kb)
         };
-
-        eprintln!(
-            "[INFO] → Bundle {:06} | {} | {} dids | {} | fetch: {:.2}s ({} reqs, {:.1}s wait) | save: {}ms | index: {}ms | {}",
+        let base = format!(
+            "[INFO] → Bundle {:06} | {} | {} dids | {} | fetch: {:.2}s ({} reqs, {:.1}s wait) | save: {}ms",
             bundle_num,
             hash,
             unique_dids,
···
             fetch_secs,
             fetch_requests,
             wait_secs,
-            bundle_save_ms,
-            index_ms,
-            age
+            bundle_save_ms
         );
+        if index_ms > 0 {
+            eprintln!("{} | index: {}ms | {}", base, index_ms, age);
+        } else {
+            eprintln!("{} | {}", base, age);
+        }
     }
 
     fn on_caught_up(
···
 
             match self
                 .manager
-                .sync_next_bundle(&self.client, None)
+                .sync_next_bundle(&self.client, None, true)
                 .await
             {
                 Ok(crate::manager::SyncResult::BundleCreated {
···
 
         let mut total_synced = 0u32;
         let mut is_initial_sync = true;
+        let mut did_index_batch_done = false;
+        let mut initial_sync_first_bundle: Option<u32> = None;
 
         // Notify logger that sync is starting
         if let Some(logger) = &self.logger {
···
 
             let sync_result = self
                 .manager
-                .sync_next_bundle(&self.client, self.config.shutdown_rx.clone())
+                .sync_next_bundle(
+                    &self.client,
+                    self.config.shutdown_rx.clone(),
+                    !is_initial_sync,
+                )
                 .await;
 
             match sync_result {
···
                     fetch_http_ms,
                 }) => {
                     total_synced += 1;
+                    if is_initial_sync && initial_sync_first_bundle.is_none() {
+                        initial_sync_first_bundle = Some(bundle_num);
+                    }
 
                     // Reset error counter on successful sync
                     use std::sync::atomic::{AtomicU32, Ordering};
···
                     }
 
                     // Caught up to the end of the chain
-                    // Mark initial sync as complete ONLY if we actually synced at least one bundle.
-                    // This prevents premature "initial sync complete" when we just have a full
-                    // mempool from a previous run but still have thousands of bundles to sync.
-                    if is_initial_sync && total_synced > 0 {
-                        is_initial_sync = false;
+                    // When initial sync finishes, perform a single batch DID index update if the index is empty
+                    // or if we created bundles during initial sync with per-bundle updates disabled.
+                    if is_initial_sync && !did_index_batch_done {
+                        let stats = self.manager.get_did_index_stats();
+                        let total_dids = stats
+                            .get("total_dids")
+                            .and_then(|v| v.as_u64())
+                            .unwrap_or(0);
+                        let total_entries = stats
+                            .get("total_entries")
+                            .and_then(|v| v.as_u64())
+                            .unwrap_or(0);
 
+                        let end_bundle = self.manager.get_last_bundle();
+                        let start_bundle = initial_sync_first_bundle.unwrap_or(1);
+
+                        // Only run batch update if there are bundles to process and either index is empty
+                        // or we created some bundles during this initial sync.
+                        let created_bundles = total_synced > 0;
+                        let index_is_empty = total_dids == 0 && total_entries == 0;
+                        if end_bundle >= start_bundle && (index_is_empty || created_bundles) {
+                            if self.config.verbose {
+                                eprintln!(
+                                    "[Sync] Performing batch DID index update: {} → {} (index empty={}, created_bundles={})",
+                                    start_bundle, end_bundle, index_is_empty, created_bundles
+                                );
+                            }
+                            if let Err(e) = self
+                                .manager
+                                .batch_update_did_index_async(start_bundle, end_bundle, true)
+                                .await
+                            {
+                                eprintln!(
+                                    "[Sync] Batch DID index update failed after initial sync: {}",
+                                    e
+                                );
+                            } else {
+                                did_index_batch_done = true;
+                            }
+                        }
+
+                        is_initial_sync = false;
                         self.handle_event(&SyncEvent::InitialSyncComplete {
                             total_bundles: total_synced,
                             mempool_count,

tests/manager.rs

···
 
     // Build DID index so DID lookups work
     manager
-        .batch_update_did_index_async(1, manager.get_last_bundle())
+        .batch_update_did_index_async(1, manager.get_last_bundle(), false)
         .await?;
 
     // Query DID operations and resolve DID
···
     common::add_dummy_bundle(&dir2_path)?;
     let manager2 = plcbundle::BundleManager::new(dir2_path.clone(), ())?;
     manager2
-        .batch_update_did_index_async(1, manager2.get_last_bundle())
+        .batch_update_did_index_async(1, manager2.get_last_bundle(), false)
         .await?;
 
     // Verify we can query DID operations from the newly built index

tests/server.rs

···
     let manager = Arc::new(manager);
     // Build DID index so the resolver can find operations in bundles
     manager
-        .batch_update_did_index_async(1, manager.get_last_bundle())
+        .batch_update_did_index_async(1, manager.get_last_bundle(), false)
         .await?;
     let port = 3032;
     let server_handle = common::start_test_server(Arc::clone(&manager), port).await?;
···
     let manager = Arc::new(manager);
     // Ensure DID index is available for data/op lookups
     manager
-        .batch_update_did_index_async(1, manager.get_last_bundle())
+        .batch_update_did_index_async(1, manager.get_last_bundle(), false)
         .await?;
     let port = 3031;
     let server_handle = common::start_test_server(Arc::clone(&manager), port).await?;