Fix: prevent full-text inverting exceptions from deadlock (#3409)

**Issue**
Full-text invert tasks run in ctpl worker threads via lambdas that
increment counters (task_count_ / inflight_tasks_) before executing
InvertColumn→Sort→ring_sorted_.Put. If any of these steps throws (e.g.,
simdjson INCORRECT_TYPE on empty JSON {}), the counter decrement /
semaphore release code is never reached, causing the consumer to block
forever.

**Solution**
Wrap all 4 inverting lambdas in try-catch:
AsyncInsertBottom: --task_count_ + cv_.notify_one() always executes
(outside try)
AsyncInsert: on failure, release(sema) + --inflight_tasks_
Insert (offline/online): on failure, --inflight_tasks_

--

Make realtime the default for new full-text indexes

Author: qinling0210

src/storage/catalog/new_catalog_static_impl.cpp

@@ -339,6 +339,69 @@ Status NewCatalog::MemIndexRecover(NewTxn *txn) {
             return status;
         }
     }
+
+    // Drain all pending fulltext index commits to ensure they complete before the checkpoint that follows.
+    {
+        auto DrainFulltextMemIndexes = [&](TableIndexMeta &table_index_meta) {
+            auto [index_base, base_status] = table_index_meta.GetIndexBase();
+            if (!base_status.ok() || index_base->index_type_ != IndexType::kFullText) {
+                return;
+            }
+            auto [segment_ids_ptr, seg_status] = table_index_meta.GetSegmentIndexIDs1();
+            if (!seg_status.ok()) {
+                return;
+            }
+            for (SegmentID segment_id : *segment_ids_ptr) {
+                SegmentIndexMeta segment_index_meta(segment_id, table_index_meta);
+                std::shared_ptr<MemIndex> mem_index = segment_index_meta.GetMemIndex();
+                if (mem_index == nullptr) {
+                    continue;
+                }
+                std::shared_ptr<MemoryIndexer> memory_indexer = mem_index->GetFulltextIndex();
+                if (memory_indexer != nullptr) {
+                    memory_indexer->WaitForTaskCompletion();
+                }
+            }
+        };
+
+        auto DrainTable = [&](TableMeta &table_meta) {
+            std::vector<std::string> *index_id_strs_ptr = nullptr;
+            std::vector<std::string> *index_name_strs_ptr = nullptr;
+            Status st = table_meta.GetIndexIDs(index_id_strs_ptr, &index_name_strs_ptr);
+            if (!st.ok()) {
+                return;
+            }
+            for (size_t i = 0; i < index_id_strs_ptr->size(); ++i) {
+                const std::string &index_id_str = (*index_id_strs_ptr)[i];
+                const std::string &index_name_str = (*index_name_strs_ptr)[i];
+                TableIndexMeta table_index_meta(index_id_str, index_name_str, table_meta);
+                DrainFulltextMemIndexes(table_index_meta);
+            }
+        };
+
+        auto DrainDB = [&](DBMeta &db_meta) {
+            std::vector<std::string> *table_id_strs_ptr = nullptr;
+            std::vector<std::string> *table_names_ptr = nullptr;
+            Status st = db_meta.GetTableIDs(table_id_strs_ptr, &table_names_ptr);
+            if (!st.ok()) {
+                return;
+            }
+            for (size_t i = 0; i < table_id_strs_ptr->size(); ++i) {
+                const std::string &table_id_str = (*table_id_strs_ptr)[i];
+                const std::string &table_name = (*table_names_ptr)[i];
+                TableMeta table_meta(db_meta.db_id_str(), table_id_str, table_name, txn);
+                DrainTable(table_meta);
+            }
+        };
+
+        for (size_t idx = 0; idx < db_count; ++idx) {
+            const std::string &db_id_str = db_id_strs_ptr->at(idx);
+            const std::string &db_name = db_names_ptr->at(idx);
+            DBMeta db_meta(db_id_str, db_name, txn);
+            DrainDB(db_meta);
+        }
+    }
+
     return Status::OK();
 }
 

src/storage/common/rcu_multimap.cppm

@@ -618,6 +618,11 @@ private:
     InnerMap *volatile read_map_;
     std::size_t miss_time_;
 
+    // RCU reader count: incremented by readers before accessing read_map_,
+    // decremented after the read is complete. Writers (CheckSwapInLock, CheckGc)
+    // wait for this to reach 0 before freeing old maps, preventing use-after-free.
+    mutable std::atomic<u64> rcu_reader_count_{0};
+
     mutable std::mutex dirty_lock_;
     InnerMap *dirty_map_;
 
@@ -662,6 +667,9 @@ typename RcuMap<Key, Value>::MapValue RcuMap<Key, Value>::CreateMapValue(const V
 
 template <typename Key, typename Value>
 Value *RcuMap<Key, Value>::Get(const Key &key, bool update_access_time) {
+    // RCU read protection: prevent write-side GC from freeing read_map_ during iteration
+    rcu_reader_count_.fetch_add(1, std::memory_order_acquire);
+
     // ULTRA-OPTIMIZED RCU READ PATTERN FOR READ-HEAVY WORKLOADS:
     // Eliminate ALL overhead in the common case to match MapWithLock performance
 
@@ -686,9 +694,13 @@ Value *RcuMap<Key, Value>::Get(const Key &key, bool update_access_time) {
                 }
             }
         }
+        rcu_reader_count_.fetch_sub(1, std::memory_order_release);
         return &(it->second.value_);
     }
 
+    // RCU: temporarily release reader count while checking dirty_map under lock
+    rcu_reader_count_.fetch_sub(1, std::memory_order_release);
+
     // PHASE 2: Check dirty_map for recent writes (SLOW PATH - MINIMIZED)
     // This path should be rare in read-heavy workloads
     {
@@ -740,21 +752,25 @@ std::optional<Value> RcuMap<Key, Value>::GetValue(const Key &key, bool update_ac
 
 template <typename Key, typename Value>
 Value *RcuMap<Key, Value>::GetWithRcuTime(const Key &key) {
+    // RCU read protection
+    rcu_reader_count_.fetch_add(1, std::memory_order_acquire);
+
     // ABSOLUTE FASTEST PATH: Zero overhead reads for maximum performance
-    // This should be as fast as MapWithLock::Get() with shared_lock
     InnerMap *current_read = read_map_;
     auto it = current_read->find(key);
 
     if (it != current_read->end()) {
+        rcu_reader_count_.fetch_sub(1, std::memory_order_release);
         return &(it->second.value_);
     }
 
+    rcu_reader_count_.fetch_sub(1, std::memory_order_release);
+
     // Inline dirty_map check to avoid function call overhead
     {
         std::lock_guard<std::mutex> lock(dirty_lock_);
         auto dirty_it = dirty_map_->find(key);
         if (dirty_it != dirty_map_->end()) {
-            // No access time updates, no miss tracking - pure read performance
             return &(dirty_it->second.value_);
         }
     }
@@ -764,18 +780,18 @@ Value *RcuMap<Key, Value>::GetWithRcuTime(const Key &key) {
 
 template <typename Key, typename Value>
 Value *RcuMap<Key, Value>::GetReadOnly(const Key &key) {
-    // BENCHMARK-OPTIMIZED READ: Absolute minimum overhead
-    // This method is designed to match MapWithLock performance exactly
-    // by eliminating ALL RCU-specific overhead
+    // RCU read protection
+    rcu_reader_count_.fetch_add(1, std::memory_order_acquire);
 
-    // Try read_map first (should succeed in most cases for read-heavy workloads)
     InnerMap *current_read = read_map_;
     auto it = current_read->find(key);
     if (it != current_read->end()) {
+        rcu_reader_count_.fetch_sub(1, std::memory_order_release);
         return &(it->second.value_);
     }
 
-    // If not found, try dirty_map (minimal overhead)
+    rcu_reader_count_.fetch_sub(1, std::memory_order_release);
+
     std::lock_guard<std::mutex> lock(dirty_lock_);
     auto dirty_it = dirty_map_->find(key);
     if (dirty_it != dirty_map_->end()) {
@@ -826,6 +842,12 @@ void RcuMap<Key, Value>::CheckSwapInLock() {
     // std::set up new dirty_map for future writes
     dirty_map_ = new_dirty_map;
 
+    // RCU SYNCHRONIZATION: Wait for all readers that saw the OLD read_map_ to finish.
+    // Without this, a reader could still be iterating the old map when CheckGc frees it.
+    while (rcu_reader_count_.load(std::memory_order_acquire) > 0) {
+        std::this_thread::yield();
+    }
+
     // Schedule old read_map for garbage collection
     // Cannot delete immediately - readers may still be using it
     deleted_map_list_.push_back(deleted_map);
@@ -929,6 +951,12 @@ void RcuMap<Key, Value>::CheckGc(u64 min_delete_time) {
     }
 
     for (auto &deleted_map : map_need_delete) {
+        // RCU: Wait for any in-flight readers to finish before freeing old maps.
+        // CheckSwapInLock already waited, but there may be readers that started before
+        // the swap and haven't finished yet.
+        while (rcu_reader_count_.load(std::memory_order_acquire) > 0) {
+            std::this_thread::yield();
+        }
         delete deleted_map.deleted_entries_;
         delete deleted_map.map_;
     }
@@ -991,20 +1019,31 @@ void RcuMap<Key, Value>::emplace(const Key &key, Args &&...args) {
 
 template <typename Key, typename Value>
 u32 RcuMap<Key, Value>::range(const Key &key_min, const Key &key_max, std::vector<Value> &result) const {
-    InnerMap *current_read = read_map_;
-    auto read_begin = current_read->lower_bound(key_min);
-    auto read_end = current_read->upper_bound(key_max);
-
+    // 1. Acquire dirty snapshot under lock FIRST — avoids lock-ordering inversion
+    //    where CheckSwapInLock holds dirty_lock_ and waits for readers to drain.
     InnerMap *dirty_snapshot = nullptr;
     {
         std::lock_guard<std::mutex> lock(dirty_lock_);
         dirty_snapshot = dirty_map_;
     }
 
+    // 2. Now pin the reader count (read_map_ is stable)
+    rcu_reader_count_.fetch_add(1, std::memory_order_acquire);
+
+    // RAII: release reader count on any exit path (normal or exceptional)
+    struct RcuReaderGuard {
+        std::atomic<u64> &cnt_;
+        ~RcuReaderGuard() { cnt_.fetch_sub(1, std::memory_order_release); }
+    } guard{rcu_reader_count_};
+
+    InnerMap *current_read = read_map_;
+    auto read_begin = current_read->lower_bound(key_min);
+    auto read_end = current_read->upper_bound(key_max);
+
     auto dirty_begin = dirty_snapshot->lower_bound(key_min);
     auto dirty_end = dirty_snapshot->upper_bound(key_max);
 
-    // merge - for std::map, we prioritize dirty_map values over read_map
+    // merge — dirty_map values take priority over read_map
     u32 count = 0;
     auto read_it = read_begin;
     auto dirty_it = dirty_begin;
@@ -1028,6 +1067,7 @@ u32 RcuMap<Key, Value>::range(const Key &key_min, const Key &key_max, std::vecto
     }
 
     return count;
+    // ~RcuReaderGuard releases reader count
 }
 
 // MapWithLock compatibility methods implementation
@@ -1081,16 +1121,26 @@ template <typename Key, typename Value>
 void RcuMap<Key, Value>::Range(const Key &key_min, const Key &key_max, std::vector<std::pair<Key, Value>> &items) {
     items.clear();
 
-    InnerMap *current_read = read_map_;
-    auto read_begin = current_read->lower_bound(key_min);
-    auto read_end = current_read->upper_bound(key_max);
-
+    // 1. Acquire dirty snapshot under lock FIRST — avoids lock-ordering inversion
     InnerMap *dirty_snapshot = nullptr;
     {
         std::lock_guard<std::mutex> lock(dirty_lock_);
         dirty_snapshot = dirty_map_;
     }
 
+    // 2. Now pin the reader count (read_map_ is stable)
+    rcu_reader_count_.fetch_add(1, std::memory_order_acquire);
+
+    // RAII: release reader count on any exit path
+    struct RcuReaderGuard {
+        std::atomic<u64> &cnt_;
+        ~RcuReaderGuard() { cnt_.fetch_sub(1, std::memory_order_release); }
+    } guard{rcu_reader_count_};
+
+    InnerMap *current_read = read_map_;
+    auto read_begin = current_read->lower_bound(key_min);
+    auto read_end = current_read->upper_bound(key_max);
+
     auto dirty_begin = dirty_snapshot->lower_bound(key_min);
     auto dirty_end = dirty_snapshot->upper_bound(key_max);
 
@@ -1109,6 +1159,7 @@ void RcuMap<Key, Value>::Range(const Key &key_min, const Key &key_max, std::vect
             items.emplace_back(it->first, it->second.value_);
         }
     }
+    // ~RcuReaderGuard releases reader count
 }
 
 template <typename Key, typename Value>

src/storage/invertedindex/column_inverter.cppm

@@ -81,6 +81,18 @@ public:
 
     const std::vector<std::binary_semaphore *> &semas() const { return semas_; }
 
+    // Release all semaphores exactly once across all threads that process this inverter.
+    void ReleaseSemas() {
+        bool expected = false;
+        if (!semas_released_.compare_exchange_strong(expected, true, std::memory_order_acq_rel)) {
+            return;
+        }
+        for (auto *sema : semas_) {
+            sema->release();
+        }
+        // semas_ does not need clearing since the atomic flag prevents re-entry.
+    }
+
 private:
     using TermBuffer = std::vector<char>;
     using PosInfoVec = std::vector<PosInfo>;
@@ -112,6 +124,7 @@ private:
 
     u32 merged_{1};
     std::vector<std::binary_semaphore *> semas_{};
+    std::atomic<bool> semas_released_{false};
 
 protected:
     size_t InvertColumn(u32 doc_id, const std::string &val);

src/storage/invertedindex/index_defines.cppm

@@ -29,7 +29,7 @@ export {
     typedef u32 tf_t;
     typedef i64 ttf_t;
 
-    constexpr optionflag_t OPTION_FLAG_ALL = of_term_payload | of_doc_payload | of_position_list | of_term_frequency | of_block_max;
+    constexpr optionflag_t OPTION_FLAG_ALL = of_term_payload | of_doc_payload | of_position_list | of_term_frequency | of_block_max | of_realtime;
     constexpr optionflag_t NO_BLOCK_MAX = of_term_payload | of_doc_payload | of_position_list | of_term_frequency;
     constexpr optionflag_t NO_TERM_FREQUENCY = of_term_payload | of_doc_payload;
     constexpr optionflag_t OPTION_FLAG_NONE = of_none;