/* ** Copyright (C) 2020-2026 Dirk-Jan C. Binnema ** ** This program is free software; you can redistribute it and/or modify it ** under the terms of the GNU General Public License as published by the ** Free Software Foundation; either version 3, or (at your option) any ** later version. ** ** This program is distributed in the hope that it will be useful, ** but WITHOUT ANY WARRANTY; without even the implied warranty of ** MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the ** GNU General Public License for more details. ** ** You should have received a copy of the GNU General Public License ** along with this program; if not, write to the Free Software Foundation, ** Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA. ** */ #include "mu-indexer.hh" #include #include #include #include #include #include #include #include #include using namespace std::chrono_literals; #include "mu-store.hh" #include "mu-scanner.hh" #include "utils/mu-error.hh" #include "utils/mu-utils-file.hh" using namespace Mu; struct IndexState { enum State { Idle, Scanning, Finishing, Cleaning, Aborting, }; static const char* name(State s) { switch (s) { case Idle: return "idle"; case Scanning: return "scanning"; case Finishing: return "finishing"; case Cleaning: return "cleaning"; case Aborting: return "aborting"; default: return ""; } } bool operator==(State rhs) const { return state_.load() == rhs; } bool operator!=(State rhs) const { return state_.load() != rhs; } void change_to(State new_state) { mu_debug("changing indexer state {}->{}", name((State)state_), name((State)new_state)); state_.store(new_state); } private: std::atomic state_{Idle}; }; struct Indexer::Private { Private(Mu::Store& store) : store_{store}, scanner_{store_.root_maildir(), [this](auto&& path, auto&& statbuf, auto&& info) { return handler(path, statbuf, info); }}, max_message_size_{store_.config().get()}, was_empty_{store.empty()} { mu_message("created indexer for {} -> " "{} (batch-size: {}; was-empty: {}; ngrams: {})", store.root_maildir(), store.path(), store.config().get(), was_empty_, store.config().get()); } ~Private() { stop(); } bool dir_predicate(const std::string& path, const struct dirent* dirent) const; bool handler(const std::string& fullpath, struct stat* statbuf, Scanner::HandleType htype); void maybe_start_worker(); void prefetch_path_terms(bool for_cleanup); bool mark_seen(const std::string& path); void mark_seen_dir(const std::string& dir_path); void scan_worker(); bool add_message(const std::string& path); void cleanup(); bool start(const Indexer::Config& conf, bool block); bool stop(); bool is_running() const { return state_ != IndexState::Idle; } Indexer::Config conf_; Store& store_; Scanner scanner_; const size_t max_message_size_; ::time_t dirstamp_{}; std::thread scanner_worker_; struct WorkItem { std::string full_path; enum Type { Dir, File }; Type type; }; void handle_item(WorkItem&& item); Progress progress_{}; IndexState state_{}; std::mutex lock_, w_lock_; Option started_; std::atomic completed_{}; bool was_empty_{}; uint64_t last_index_{}; using PathTerm = std::pair; std::vector db_path_terms_; /**< all path-terms in the store (at scan-start), in Xapian's (ascending) and a seen marker */ bool use_db_path_terms_{}; }; void Indexer::Private::prefetch_path_terms(bool for_cleanup) { use_db_path_terms_ = false; db_path_terms_.clear(); // The in-memory path-terms serve two purposes: // // i. checking whether some message is already in the store // without db access // ii. unless we need it for cleanup: whatever // is left unmarked after scanning is an orphan. if (!for_cleanup && conf_.lazy_check) return; db_path_terms_.reserve(store_.size()); store_.for_each_term(Field::Id::Path, [&](const std::string& term) { db_path_terms_.emplace_back(term, false/*!seen*/); return true; }); use_db_path_terms_ = true; mu_debug("prefetched {} path-term(s)", db_path_terms_.size()); } bool Indexer::Private::mark_seen(const std::string& path) { if (!use_db_path_terms_) return false; // N.B. Xapian yields terms in ascending byte-lexicographic order, so // db_path_terms_ is sorted and we can binary-search. const auto term{field_from_id(Field::Id::Path).xapian_term(path)}; const auto it{std::ranges::lower_bound(db_path_terms_, term, {}, &PathTerm::first)}; if (it == db_path_terms_.end() || it->first != term) return false; // not in the store. it->second = true; return true; } void Indexer::Private::mark_seen_dir(const std::string& dir_path) { if (!use_db_path_terms_) return; // mark all store messages under dir_path as seen const auto prefix{field_from_id(Field::Id::Path) .xapian_term(dir_path + "/")}; for (auto it = std::ranges::lower_bound(db_path_terms_, prefix, {}, &PathTerm::first); it != db_path_terms_.end() && it->first.starts_with(prefix); ++it) it->second = true; } bool Indexer::Private::handler(const std::string& fullpath, struct stat* statbuf, Scanner::HandleType htype) { switch (htype) { case Scanner::HandleType::EnterDir: case Scanner::HandleType::EnterNewCur: { if (fullpath.length() > MaxTermLength) { // currently the db uses the path as a key, and // therefore it cannot be too long. We'd get an error // later anyway but for now it's useful for surviving // circular symlinks mu_warning("'{}' is too long; ignore", fullpath); return false; } // in lazy-mode, we ignore this dir if its dirstamp suggest it // is up-to-date (this is _not_ always true; hence we call it // lazy-mode); only for actual message dirs, since the dir // tstamps may not bubble up. dirstamp_ = store_.dirstamp(fullpath); if (conf_.lazy_check && dirstamp_ >= statbuf->st_ctime && htype == Scanner::HandleType::EnterNewCur) { mu_debug("skip {} (seems up-to-date: {:%FT%T} >= {:%FT%T})", fullpath, mu_time(dirstamp_), mu_time(statbuf->st_ctime)); mark_seen_dir(fullpath); return false; } // don't index dirs with '.noindex' auto noindex = ::access((fullpath + "/.noindex").c_str(), F_OK) == 0; if (noindex) { mu_debug("skip {} (has .noindex)", fullpath); return false; // don't descend into this dir. } // don't index dirs with '.noupdate', unless we do a full // (re)index. if (!conf_.ignore_noupdate) { auto noupdate = ::access((fullpath + "/.noupdate").c_str(), F_OK) == 0; if (noupdate) { mu_debug("skip {} (has .noupdate)", fullpath); mark_seen_dir(fullpath); return false; } } mu_debug("checked {}", fullpath); return true; } case Scanner::HandleType::LeaveDir: { // don't touch dirstamps in a cleanup-only run: nothing was // indexed, so the dir is not to be considered up-to-date. if (conf_.scan) handle_item({fullpath, WorkItem::Type::Dir}); return true; } case Scanner::HandleType::File: { ++progress_.checked; // this file is present in the file-system; also remember // whether it is in the store already. const auto in_store{mark_seen(fullpath)}; if (!conf_.scan) return false; // cleanup-only run; only mark. if (conf_.lazy_check && static_cast(statbuf->st_ctime) < last_index_) { // in lazy mode, ignore the file if it has not changed // since the last indexing op. return false; } if (static_cast(statbuf->st_size) > max_message_size_) { mu_debug("skip {} (too big: {} bytes)", fullpath, statbuf->st_size); return false; } // if the message is not in the db yet, or not up-to-date, queue // it for updating/inserting. if (statbuf->st_ctime <= dirstamp_ && (use_db_path_terms_ ? in_store : store_.contains_message(fullpath))) return false; handle_item({fullpath, WorkItem::Type::File}); return true; } default: g_return_val_if_reached(false); return false; } } bool Indexer::Private::add_message(const std::string& path) { /* * Having the lock here makes things a _lot_ slower. * * The reason for having the lock is some helgrind warnings; * but it believed those are _false alarms_ * https://gitlab.gnome.org/GNOME/glib/-/issues/2662 */ //std::unique_lock lock{w_lock_}; auto msg{Message::make_from_path(path, store_.message_options())}; if (!msg) { mu_warning("failed to create message from {}: {}", path, msg.error().what()); return false; } // if the store was empty, we know that the message is completely new // and can use the fast path (Xapians 'add_document' rather than // 'replace_document) auto res = store_.consume_message(std::move(msg.value()), was_empty_); if (!res) { mu_warning("failed to add message @ {}: {}", path, res.error().what()); return false; } return true; } void Indexer::Private::handle_item(WorkItem&& item) { try { switch (item.type) { case WorkItem::Type::File: { if (G_LIKELY(add_message(item.full_path))) ++progress_.updated; } break; case WorkItem::Type::Dir: store_.set_dirstamp(item.full_path, started_.value()); break; default: g_warn_if_reached(); break; } } catch (const Mu::Error& er) { mu_warning("error adding message @ {}: {}", item.full_path, er.what()); } } void Indexer::Private::cleanup() { mu_debug("starting cleanup after scan"); // during the scan, each file (and each wholesale-skipped // directory) marked its path-term; whatever is left unmarked is // an orphan: a message in the store without a file in the maildir. // // N.B. this snapshot was taken at scan-start, so messages that // appeared in the store _during_ the scan are not in it, and thus // cannot be orphaned by mistake. std::vector orphan_terms; for (auto&& [term, seen] : db_path_terms_) if (!seen) orphan_terms.emplace_back(std::move(term)); if (orphan_terms.empty()) mu_debug("nothing to clean up"); else { size_t nr_removed = store_.remove_messages_by_term( orphan_terms, [&](size_t nr_cleaned_up_so_far) { progress_.removed = nr_cleaned_up_so_far; }); mu_debug("removing {} stale message(s) from store", nr_removed); progress_.removed = nr_removed; if (nr_removed != orphan_terms.size()) mu_warning("should have removed {} messages but actually removed {}", orphan_terms.size(), nr_removed); } } void Indexer::Private::scan_worker() { progress_.reset(); started_ = time(NULL); // cleanup gets its orphans by marking the in-memory path-terms // during the file-system scan, so we need the scanner even for a // cleanup-only run. if (conf_.scan || conf_.cleanup) { prefetch_path_terms(conf_.cleanup); mu_debug("starting scanner"); if (!scanner_.start()) { // blocks. mu_warning("failed to start scanner"); state_.change_to(IndexState::Idle); return; } mu_debug("scanner finished"); } bool aborted = state_ == IndexState::Aborting; state_.change_to(IndexState::Cleaning); if (!conf_.cleanup) mu_debug("cleanup: not running as requested"); else if (aborted) mu_debug("cleanup: disabling because indexer aborted"); else cleanup(); // release the in-memory path-terms. use_db_path_terms_ = false; db_path_terms_ = {}; aborted = state_ == IndexState::Aborting; if (!aborted && conf_.scan) { // Store started time, not ending time, so that next time we run we know to scan // anything that appeared during our scan. store_.config().set(started_.value()); } completed_ = ::time({}); // attempt to commit to disk. store_.xapian_db().request_commit(true); state_.change_to(IndexState::Idle); started_ = Nothing; } bool Indexer::Private::start(const Indexer::Config& conf, bool block) { stop(); conf_ = conf; // refresh: the indexer may be re-used for multiple runs, and the // add_document fast-path is only valid while the store is empty. was_empty_ = store_.empty(); if (was_empty_ && conf_.lazy_check) { mu_debug("turn off lazy check since we have an empty store"); conf_.lazy_check = false; } mu_debug("starting indexer"); mu_debug("indexing: {}; clean-up: {}", conf_.scan ? "yes" : "no", conf_.cleanup ? "yes" : "no"); // remember the _previous_ indexing, so in lazy mode we can skip // those files. last_index_ = store_.config().get(); state_.change_to(IndexState::Scanning); /* kick the disk-scanner thread */ scanner_worker_ = std::thread([this] { scan_worker(); }); mu_debug("started indexer in {}-mode", block ? "blocking" : "non-blocking"); if (block) { while(is_running()) { using namespace std::chrono_literals; std::this_thread::sleep_for(100ms); } } return true; } bool Indexer::Private::stop() { if (state_ != IndexState::Idle) state_.change_to(IndexState::Aborting); scanner_.stop(); if (scanner_worker_.joinable()) scanner_worker_.join(); state_.change_to(IndexState::Idle); return true; } Indexer::Indexer(Store& store) : priv_{std::make_unique(store)} {} Indexer::~Indexer() = default; bool Indexer::start(const Indexer::Config& conf, bool block) { const auto mdir{priv_->store_.root_maildir()}; if (G_UNLIKELY(access(mdir.c_str(), R_OK) != 0)) { mu_critical("'{}' is not readable: {}", mdir, g_strerror(errno)); return false; } std::lock_guard lock(priv_->lock_); if (is_running()) return true; return priv_->start(conf, block); } bool Indexer::stop() { std::lock_guard lock{priv_->lock_}; if (!is_running()) return true; mu_debug("stopping indexer"); return priv_->stop(); } bool Indexer::is_running() const { return priv_->is_running(); } const Indexer::Progress& Indexer::progress() const { priv_->progress_.running = priv_->state_ == IndexState::Idle ? false : true; return priv_->progress_; } ::time_t Indexer::completed() const { return priv_->completed_; } #if BUILD_TESTS #include "mu-test-utils.hh" static void test_index_basic() { allow_warnings(); TempDir tdir; auto store = Store::make_new(tdir.path(), MU_TESTMAILDIR2); assert_valid_result(store); g_assert_true(store->empty()); Indexer& idx{store->indexer()}; g_assert_false(idx.is_running()); g_assert_true(idx.stop()); g_assert_cmpuint(idx.completed(),==, 0); const auto& prog{idx.progress()}; g_assert_false(prog.running); g_assert_cmpuint(prog.checked,==, 0); g_assert_cmpuint(prog.updated,==, 0); g_assert_cmpuint(prog.removed,==, 0); mu_test_skip_valgrind_return(); Indexer::Config conf{}; conf.ignore_noupdate = true; { const auto start{time({})}; g_assert_true(idx.start(conf)); while (idx.is_running()) g_usleep(10000); g_assert_false(idx.is_running()); g_assert_true(idx.stop()); g_assert_cmpuint(idx.completed() - start, <, 5); g_assert_false(prog.running); g_assert_cmpuint(prog.checked,==, 14); g_assert_cmpuint(prog.updated,==, 14); g_assert_cmpuint(prog.removed,==, 0); g_assert_cmpuint(store->size(),==,14); } conf.lazy_check = true; conf.ignore_noupdate = false; { const auto start{time({})}; g_assert_true(idx.start(conf)); while (idx.is_running()) g_usleep(10000); g_assert_false(idx.is_running()); g_assert_true(idx.stop()); g_assert_cmpuint(idx.completed() - start, <, 3); g_assert_false(prog.running); g_assert_cmpuint(prog.checked,==, 0); g_assert_cmpuint(prog.updated,==, 0); g_assert_cmpuint(prog.removed,==, 0); g_assert_cmpuint(store->size(),==, 14); } } static void test_index_lazy() { allow_warnings(); mu_test_skip_valgrind_return(); TempDir tdir; auto store = Store::make_new(tdir.path(), MU_TESTMAILDIR2); assert_valid_result(store); g_assert_true(store->empty()); Indexer& idx{store->indexer()}; Indexer::Config conf{}; conf.lazy_check = true; conf.ignore_noupdate = false; const auto start{time({})}; g_assert_true(idx.start(conf)); while (idx.is_running()) g_usleep(10000); g_assert_false(idx.is_running()); g_assert_true(idx.stop()); g_assert_cmpuint(idx.completed() - start, <, 3); const auto& prog{idx.progress()}; g_assert_false(prog.running); g_assert_cmpuint(prog.checked,==, 6); g_assert_cmpuint(prog.updated,==, 6); g_assert_cmpuint(prog.removed,==, 0); g_assert_cmpuint(store->size(),==, 6); } static void test_index_cleanup() { allow_warnings(); TempDir tdir; auto mdir = join_paths(tdir.path(), "Test"); { auto res = run_command({"cp", "-r", MU_TESTMAILDIR2, mdir}); assert_valid_result(res); g_assert_cmpuint(res->exit_code,==, 0); } auto store = Store::make_new(tdir.path(), mdir); assert_valid_result(store); g_assert_true(store->empty()); Indexer& idx{store->indexer()}; Indexer::Config conf{}; conf.ignore_noupdate = true; g_assert_true(idx.start(conf)); while (idx.is_running()) g_usleep(10000); g_assert_false(idx.is_running()); g_assert_true(idx.stop()); g_assert_cmpuint(store->size(),==, 14); // remove a message { auto mpath = join_paths(mdir, "bar", "cur", "mail6"); auto res = run_command({"rm", mpath}); assert_valid_result(res); g_assert_cmpuint(res->exit_code,==, 0); } // no cleanup, # stays the same conf.cleanup = false; g_assert_true(idx.start(conf)); while (idx.is_running()) g_usleep(10000); g_assert_false(idx.is_running()); g_assert_true(idx.stop()); g_assert_cmpuint(store->size(),==, 14); // cleanup, message is gone from store. conf.cleanup = true; g_assert_true(idx.start(conf)); while (idx.is_running()) g_usleep(10000); g_assert_false(idx.is_running()); g_assert_true(idx.stop()); g_assert_cmpuint(store->size(),==, 13); // remove another message { auto mpath = join_paths(mdir, "bar", "cur", "mail5"); auto res = run_command({"rm", mpath}); assert_valid_result(res); g_assert_cmpuint(res->exit_code,==, 0); } // cleanup-only run (no scan); message is gone from store. conf.scan = false; g_assert_true(idx.start(conf)); while (idx.is_running()) g_usleep(10000); g_assert_false(idx.is_running()); g_assert_true(idx.stop()); g_assert_cmpuint(store->size(),==, 12); } int main(int argc, char* argv[]) { mu_test_init(&argc, &argv); g_test_add_func("/index/basic", test_index_basic); g_test_add_func("/index/lazy", test_index_lazy); g_test_add_func("/index/cleanup", test_index_cleanup); return g_test_run(); } #endif /*BUILD_TESTS*/