fix: Resolve critical bugs in cache, MySQL, and concurrency handling

## Cache Configuration
- Change CacheConfig from int max_memory_mb to size_t max_memory_bytes
  for precise byte-level memory control and overflow prevention
- Convert MB to bytes at config load time, bytes to MB at display time
- Fix LRUEvictionCleansUpMetadata test with byte-precision cache sizing

## Resource Management
- Add RAII pattern for MySQL result sets to prevent memory leaks
  (MysqlResultGuard in connection.cpp)

## Thread Safety & Concurrency
- Fix InvalidationManager::ClearTable deadlock by using lock_guard
- Add comprehensive test coverage for concurrent scenarios

## Test Reliability
- Add RESOURCE_LOCK to HTTP server tests to prevent port conflicts
  during parallel execution

This commit resolves critical issues including:
- Deadlock risks in cache invalidation
- Memory leaks in MySQL connection handling
- Race conditions in concurrent operations
- Test flakiness in parallel execution

Author: libraz

src/cache/cache_manager.cpp

@@ -15,12 +15,18 @@ CacheManager::CacheManager(const config::CacheConfig& cache_config,
     : enabled_(cache_config.enabled) {
   if (enabled_) {
     // Create query cache
-    const size_t max_memory_bytes = static_cast<size_t>(cache_config.max_memory_mb) * 1024 * 1024;
-    query_cache_ = std::make_unique<QueryCache>(max_memory_bytes, cache_config.min_query_cost_ms);
+    query_cache_ = std::make_unique<QueryCache>(cache_config.max_memory_bytes, cache_config.min_query_cost_ms);
 
     // Create invalidation manager
     invalidation_mgr_ = std::make_unique<InvalidationManager>(query_cache_.get());
 
+    // Set eviction callback to clean up invalidation metadata
+    query_cache_->SetEvictionCallback([this](const CacheKey& key) {
+      if (invalidation_mgr_) {
+        invalidation_mgr_->UnregisterCacheEntry(key);
+      }
+    });
+
     // Create invalidation queue with table_contexts for per-table ngram settings
     invalidation_queue_ =
         std::make_unique<InvalidationQueue>(query_cache_.get(), invalidation_mgr_.get(), table_contexts);

src/cache/invalidation_manager.cpp

@@ -71,9 +71,8 @@ std::unordered_set<CacheKey> InvalidationManager::InvalidateAffectedEntries(cons
   return affected_keys;
 }
 
-void InvalidationManager::UnregisterCacheEntry(const CacheKey& key) {
-  std::unique_lock lock(mutex_);
-
+// Internal helper: unregister cache entry without locking (assumes mutex is already held)
+void InvalidationManager::UnregisterCacheEntryUnlocked(const CacheKey& key) {
   // Find metadata
   auto metadata_it = cache_metadata_.find(key);
   if (metadata_it == cache_metadata_.end()) {
@@ -107,6 +106,11 @@ void InvalidationManager::UnregisterCacheEntry(const CacheKey& key) {
   cache_metadata_.erase(metadata_it);
 }
 
+void InvalidationManager::UnregisterCacheEntry(const CacheKey& key) {
+  std::unique_lock lock(mutex_);
+  UnregisterCacheEntryUnlocked(key);
+}
+
 void InvalidationManager::ClearTable(const std::string& table_name) {
   std::unique_lock lock(mutex_);
 
@@ -118,14 +122,12 @@ void InvalidationManager::ClearTable(const std::string& table_name) {
     }
   }
 
-  // Remove entries
-  lock.unlock();
+  // Remove entries while holding lock (use unlocked version to avoid deadlock)
   for (const auto& key : to_remove) {
-    UnregisterCacheEntry(key);
+    UnregisterCacheEntryUnlocked(key);
   }
 
-  // Remove table from reverse index
-  lock.lock();
+  // Remove table from reverse index (already holding lock)
   ngram_to_cache_keys_.erase(table_name);
 }
 

src/cache/invalidation_manager.h

@@ -120,6 +120,13 @@ class InvalidationManager {
   // Thread safety
   mutable std::shared_mutex mutex_;
 
+  /**
+   * @brief Internal helper: unregister cache entry without locking
+   * @param key Cache key to unregister
+   * @note Assumes mutex_ is already held by caller
+   */
+  void UnregisterCacheEntryUnlocked(const CacheKey& key);
+
   /**
    * @brief Extract ngrams from text
    * @param text Text to extract ngrams from

src/cache/invalidation_queue.cpp

@@ -38,6 +38,9 @@ void InvalidationQueue::Enqueue(const std::string& table_name, const std::string
 
       // Phase 2: Erase from cache immediately (no queuing)
       for (const auto& key : affected_keys) {
+        // Unregister metadata first to prevent memory leak even if Erase throws
+        invalidation_mgr_->UnregisterCacheEntry(key);
+
         if (cache_ != nullptr) {
           cache_->Erase(key);
         }
@@ -119,6 +122,11 @@ void InvalidationQueue::WorkerLoop() {
     const auto time_since_oldest = now - oldest_timestamp;
 
     if (pending_ngrams_.size() >= batch_size_ || time_since_oldest >= max_delay_) {
+      // Check running_ before processing to handle spurious wakeup and shutdown
+      if (!running_.load()) {
+        break;
+      }
+
       // Process batch
       lock.unlock();
       ProcessBatch();
@@ -127,6 +135,11 @@ void InvalidationQueue::WorkerLoop() {
       const auto remaining_delay = max_delay_ - time_since_oldest;
       queue_cv_.wait_for(lock, remaining_delay,
                          [this] { return !running_.load() || pending_ngrams_.size() >= batch_size_; });
+
+      // After wakeup, check running_ before continuing
+      if (!running_.load()) {
+        break;
+      }
     }
   }
 }

src/cache/query_cache.cpp

@@ -77,16 +77,17 @@ std::optional<std::vector<DocId>> QueryCache::Lookup(const CacheKey& key) {
     return std::nullopt;
   }
 
-  // Copy query_cost_ms before releasing lock to avoid use-after-free
+  // Copy query_cost_ms and created_at before releasing lock to avoid use-after-free
   const double query_cost_ms = entry.query_cost_ms;
+  const auto created_at = entry.metadata.created_at;
 
   // Update access time (need to upgrade to unique lock)
   lock.unlock();
   std::unique_lock write_lock(mutex_);
 
-  // Re-check existence (might have been evicted)
+  // Re-check existence and verify it's the same entry (not a new entry with same key)
   iter = cache_map_.find(key);
-  if (iter != cache_map_.end()) {
+  if (iter != cache_map_.end() && iter->second.first.metadata.created_at == created_at) {
     Touch(key);
     iter->second.first.metadata.last_accessed = std::chrono::steady_clock::now();
     iter->second.first.metadata.access_count++;
@@ -119,14 +120,14 @@ bool QueryCache::Insert(const CacheKey& key, const std::vector<DocId>& result, c
     return false;
   }
 
-  // Calculate memory usage
+  // Create cache entry to calculate accurate memory usage
+  CacheEntry temp_entry;
+  temp_entry.compressed = std::move(compressed);
+  temp_entry.metadata = metadata;
+
   const size_t original_count = result.size();  // Number of DocId elements, not bytes
-  const size_t compressed_size = compressed.size();
-  size_t ngrams_size = 0;
-  for (const auto& ngram : metadata.ngrams) {
-    ngrams_size += ngram.capacity();
-  }
-  const size_t entry_memory = sizeof(CacheEntry) + compressed_size + ngrams_size;
+  const size_t compressed_size = temp_entry.compressed.size();
+  const size_t entry_memory = temp_entry.MemoryUsage();
 
   // Don't cache if entry is too large
   if (entry_memory > max_memory_bytes_) {
@@ -148,24 +149,21 @@ bool QueryCache::Insert(const CacheKey& key, const std::vector<DocId>& result, c
     }
   }
 
-  // Create cache entry
-  CacheEntry entry;
-  entry.key = key;
-  entry.compressed = std::move(compressed);
-  entry.original_size = original_count;  // Store count, not bytes
-  entry.compressed_size = compressed_size;
-  entry.query_cost_ms = query_cost_ms;
-  entry.metadata = metadata;
-  entry.metadata.created_at = std::chrono::steady_clock::now();
-  entry.metadata.last_accessed = entry.metadata.created_at;
-  entry.invalidated.store(false);
+  // Complete cache entry (reuse temp_entry to maintain consistent memory calculation)
+  temp_entry.key = key;
+  temp_entry.original_size = original_count;  // Store count, not bytes
+  temp_entry.compressed_size = compressed_size;
+  temp_entry.query_cost_ms = query_cost_ms;
+  temp_entry.metadata.created_at = std::chrono::steady_clock::now();
+  temp_entry.metadata.last_accessed = temp_entry.metadata.created_at;
+  temp_entry.invalidated.store(false);
 
   // Insert into LRU list (front = most recent)
   lru_list_.push_front(key);
   auto lru_it = lru_list_.begin();
 
   // Insert into cache map using emplace to avoid copy
-  cache_map_.emplace(key, std::make_pair(std::move(entry), lru_it));
+  cache_map_.emplace(key, std::make_pair(std::move(temp_entry), lru_it));
 
   // Update memory tracking
   total_memory_bytes_ += entry_memory;
@@ -279,6 +277,11 @@ bool QueryCache::EvictForSpace(size_t required_bytes) {
     lru_list_.pop_back();
     cache_map_.erase(iter);
 
+    // Notify eviction callback (for InvalidationManager cleanup)
+    if (eviction_callback_) {
+      eviction_callback_(lru_key);
+    }
+
     // Update memory tracking
     total_memory_bytes_ -= entry_memory;
     stats_.current_entries--;

src/cache/query_cache.h

@@ -6,6 +6,7 @@
 #pragma once
 
 #include <atomic>
+#include <functional>
 #include <list>
 #include <memory>
 #include <mutex>
@@ -113,6 +114,12 @@ struct CacheStatistics {
  */
 class QueryCache {
  public:
+  /**
+   * @brief Callback type for eviction notifications
+   * @param key The cache key being evicted
+   */
+  using EvictionCallback = std::function<void(const CacheKey&)>;
+
   /**
    * @brief Constructor
    * @param max_memory_bytes Maximum memory usage in bytes
@@ -213,6 +220,12 @@ class QueryCache {
    */
   void IncrementInvalidationBatches() { stats_.invalidations_batches++; }
 
+  /**
+   * @brief Set callback to be notified when entries are evicted
+   * @param callback Function to call when an entry is evicted via LRU
+   */
+  void SetEvictionCallback(EvictionCallback callback) { eviction_callback_ = std::move(callback); }
+
  private:
   // LRU list: most recently used at front
   std::list<CacheKey> lru_list_;
@@ -233,6 +246,9 @@ class QueryCache {
   // Statistics
   CacheStatistics stats_;
 
+  // Eviction callback
+  EvictionCallback eviction_callback_;
+
   /**
    * @brief Evict entries to make room for new entry
    * @param required_bytes Bytes needed for new entry

src/config/config.cpp

@@ -569,7 +569,8 @@ Config ParseConfigFromJson(const json& root) {
       config.cache.enabled = cache["enabled"].get<bool>();
     }
     if (cache.contains("max_memory_mb")) {
-      config.cache.max_memory_mb = cache["max_memory_mb"].get<int>();
+      int max_memory_mb = cache["max_memory_mb"].get<int>();
+      config.cache.max_memory_bytes = static_cast<size_t>(max_memory_mb) * 1024 * 1024;
     }
     if (cache.contains("min_query_cost_ms")) {
       config.cache.min_query_cost_ms = cache["min_query_cost_ms"].get<double>();

src/config/config.h

@@ -243,7 +243,7 @@ struct LoggingConfig {
  */
 struct CacheConfig {
   bool enabled = true;                          ///< Enable/disable cache (default: true)
-  int max_memory_mb = 32;                       ///< Maximum cache memory in MB (default: 32)  // NOLINT
+  size_t max_memory_bytes = 32 * 1024 * 1024;  ///< Maximum cache memory in bytes (default: 32MB)  // NOLINT
   double min_query_cost_ms = 10.0;              ///< Minimum query cost to cache (default: 10ms)  // NOLINT
   int ttl_seconds = 3600;                       ///< Cache entry TTL (default: 1 hour, 0 = no TTL)  // NOLINT
   std::string invalidation_strategy = "ngram";  ///< Invalidation strategy: "ngram", "table"

src/config/config_help.cpp

@@ -210,7 +210,7 @@ nlohmann::json ConfigToJson(const Config& config) {
   // Cache configuration
   json["cache"] = {
       {"enabled", config.cache.enabled},
-      {"max_memory_mb", config.cache.max_memory_mb},
+      {"max_memory_mb", config.cache.max_memory_bytes / (1024 * 1024)},  // Convert bytes to MB for display
       {"min_query_cost_ms", config.cache.min_query_cost_ms},
       {"ttl_seconds", config.cache.ttl_seconds},
       {"invalidation_strategy", config.cache.invalidation_strategy},

src/index/index.cpp

@@ -445,12 +445,41 @@ Index::IndexStatistics Index::GetStatistics() const {
 }
 
 void Index::Optimize(uint64_t total_docs) {
-  for (auto& [term, posting] : term_postings_) {
+  // Check if already optimizing (prevent concurrent Optimize() calls)
+  bool expected = false;
+  if (!is_optimizing_.compare_exchange_strong(expected, true)) {
+    spdlog::warn("Optimization already in progress, ignoring Optimize() request");
+    return;
+  }
+
+  // RAII guard to ensure flag is cleared
+  struct OptimizationGuard {
+    std::atomic<bool>& flag;
+    explicit OptimizationGuard(std::atomic<bool>& f) : flag(f) {}
+    ~OptimizationGuard() { flag = false; }
+    OptimizationGuard(const OptimizationGuard&) = delete;
+    OptimizationGuard& operator=(const OptimizationGuard&) = delete;
+  };
+  OptimizationGuard guard(is_optimizing_);
+
+  // Take snapshot to prevent iterator invalidation while allowing concurrent searches
+  std::vector<std::pair<std::string, PostingList*>> postings_snapshot;
+  {
+    std::shared_lock<std::shared_mutex> lock(postings_mutex_);
+    postings_snapshot.reserve(term_postings_.size());
+    for (const auto& [term, posting] : term_postings_) {
+      postings_snapshot.emplace_back(term, posting.get());
+    }
+  }
+
+  // Optimize each posting
+  for (const auto& [term, posting] : postings_snapshot) {
     posting->Optimize(total_docs);
   }
+
   // NOLINTBEGIN(cppcoreguidelines-avoid-magic-numbers,readability-magic-numbers)
   // 1024: Standard conversion factor for bytes to KB to MB
-  spdlog::info("Optimized index: {} terms, {} MB", term_postings_.size(), MemoryUsage() / (1024 * 1024));
+  spdlog::info("Optimized index: {} terms, {} MB", postings_snapshot.size(), MemoryUsage() / (1024 * 1024));
   // NOLINTEND(cppcoreguidelines-avoid-magic-numbers,readability-magic-numbers)
 }