Describe this repository

Fixes https://github.com/valeriansaliou/sonic/issues/394.

CONFIGURATION.md

@@ -18,7 +18,7 @@ read from `SONIC_FOO__BAR__BAZ`.
 
 Under `[server]`:
 
-* `log_level` (type: _string_, allowed: `debug`, `info`, `warn`, `error`, default: `error`) — Verbosity of logging, set it to `error` in production
+* `log_level` (type: _string_, allowed: `debug`, `info`, `warn`, `error`, default: `warn`) — Verbosity of logging, set it to `error` in production
 
 ### Channel configuration
 
@@ -98,6 +98,7 @@ Under `[search]`:
 * `suggest_limit_maximum` (type: _integer_, allowed: numbers, default: `20`) — Maximum suggested words limit for a suggest command (if the LIMIT command modifier is being used when issuing a SUGGEST command)
 * `list_limit_default` (type: _integer_, allowed: numbers, default: `100`) — Default listed words limit for a list command (if the LIMIT command modifier is not used when issuing a LIST command)
 * `list_limit_maximum` (type: _integer_, allowed: numbers, default: `500`) — Maximum listed words limit for a list command (if the LIMIT command modifier is being used when issuing a LIST command)
+* `typo_factor_word_lengths` (type: _array_, allowed: word lengths in bytes, default: `[4, 7, 10]`) — Word lengths from which one more typo is corrected in query and suggest commands (the default allows no typo up to 3 letters, 1 typo from 4 letters, 2 typos from 7 letters and 3 typos from 10 letters; typo correction can be slow and yield unexpected results, remove values to allow fewer typos or use `[]` to disable it)
 
 ### KV store configuration
 
@@ -111,7 +112,7 @@ Under `[store.kv]`:
 * `database.flush_after` (type: _integer_, allowed: seconds, default: `900`) — Time after which pending database updates should be flushed from memory to disk (increase this delay if you encounter high-CPU usage issues when a flush task kicks-in; this value should be lower than `store.kv.pool.inactive_after`)
 * `database.parallelism` (type: _integer_, allowed: numbers, default: `2`) — Limit on the number of compaction and flush threads that can run at the same time
 * `database.max_files` (type: _integer_, allowed: numbers, no default) — Maximum number of database files kept open at the same time per-database (if any; otherwise there are no limits)
-* `database.write_buffer_size` (type: _integer_, allowed: numbers, default: `16384`) — Maximum size **in KiB** of the database write buffer, after which data gets flushed to disk (ie. `16384` is `16MiB`; the size should be a multiple of `1024`, eg. `128 * 1024 = 131072` for `128MiB`)
+* `database.write_buffer_size` (type: _integer_, allowed: numbers, default: `16384`) — Maximum size **in KiB** of the database write buffer, after which data gets flushed to disk (ie. `16384` is `16MiB`; to give the size in MiB, multiply it by `1024`, eg. `128 * 1024 = 131072` for `128MiB`)
 * `database.min_write_buffer_number` (type: _integer_, allowed: `≥1`, default: RocksDB default (`1`)) — Minimum number of memtables that must be written before they can be flushed together to disk (increasing this batches small flushes into fewer, larger ones, at the cost of more memory held before a flush)
 * `database.min_write_buffer_number_to_merge` (type: _integer_, allowed: `≥1`, default: RocksDB default (`1`)) — Minimum number of immutable memtables that must accumulate before they are merged and flushed together (higher values reduce write amplification from small flushes, at the cost of more memory usage)
 * `database.max_write_buffer_number` (type: _integer_, allowed: `≥2`, default: RocksDB default (`2`)) — Maximum number of memtables, both active and immutable, that can be held in memory at once (once this limit is reached, writes are stalled until a flush completes)

PROTOCOL.md

@@ -0,0 +1,216 @@
+# Sonic Protocol
+
+## ⚡️ Sonic Channel
+
+**Sonic Channel is the protocol used to perform searches and ingest index data. You can also use it for Sonic administration operations. Sonic listens on TCP port 1491 by default.**
+
+This document specifies the Sonic Channel protocol. Use it if you are looking to build your own Sonic Channel library, or if you are looking to debug Sonic using eg. `telnet` in command-line.
+
+To start a `telnet` session with your local Sonic instance, execute: `telnet ::1 1491`
+
+_Refer to sections below to interact with Sonic._
+
+---
+
+### 1️⃣ Before you start
+
+**Please consider the following upon integrating the Sonic Channel protocol:**
+
+1. Each command sent must be terminated with a new line character (`\n`) as to commit the command to the server;
+2. Upon starting a Sonic Channel session, your library should read the `buffer(20000)` parameter in the `STARTED` response, and use this value (in bytes) as to know when a command data should be truncated and split in multiple sub-commands (to avoid buffer overflows, ie. sending too much data in a single command);
+3. The `<collection>`, `<bucket>` and `<object>` values of any command must be 1 to 128 ASCII characters long, otherwise the command is rejected with `ERR invalid_argument(InvalidCollection)`, `ERR invalid_argument(InvalidBucket)` or `ERR invalid_argument(InvalidObject)`;
+
+---
+
+### 2️⃣ Sonic Channel (uninitialized)
+
+* `START <mode> <password>`: select mode to use for connection (either: `search`, `ingest` or `control`). The password is found in the `config.cfg` file at `channel.auth_password`.
+
+_Issuing any other command — eg. `QUIT` — in this mode will abort the TCP connection, effectively resulting in a `QUIT` with the `ENDED not_recognized` response._
+
+---
+
+### 3️⃣ Sonic Channel (Search mode)
+
+_The Sonic Channel Search mode is used for querying the search index. Once in this mode, you cannot switch to other modes or gain access to commands from other modes._
+
+**➡️ Available commands:**
+
+* `QUERY`: query database (syntax: `QUERY <collection> <bucket> "<terms>" [LIMIT(<count>)]? [OFFSET(<count>)]? [LANG(<locale>)]?`; time complexity: `O(1)` if enough exact word matches or `O(N)` if not enough exact matches where `N` is the number of alternate words tried, in practice it approaches `O(1)`)
+* `SUGGEST`: auto-completes word (syntax: `SUGGEST <collection> <bucket> "<word>" [LIMIT(<count>)]?`; time complexity: `O(1)`)
+* `LIST`: enumerates all words in an index (syntax: `LIST <collection> <bucket> [LIMIT(<count>)]? [OFFSET(<count>)]?`; time complexity: `O(N)` where `N` is the number of words enumerated, within provided limits)
+* `PING`: ping server (syntax: `PING`; time complexity: `O(1)`)
+* `HELP`: show help (syntax: `HELP [<manual>]?`; time complexity: `O(1)`)
+* `QUIT`: stop connection (syntax: `QUIT`; time complexity: `O(1)`)
+
+**⏩ Syntax terminology:**
+
+* `<collection>`: index collection (ie. what you search in, eg. `messages`, `products`, etc.), 1 to 128 ASCII characters long;
+* `<bucket>`: index bucket name (ie. user-specific search classifier in the collection if you have any eg. `user-1, user-2, ..`, otherwise use a common bucket name eg. `generic, default, common, ..`), 1 to 128 ASCII characters long;
+* `<terms>`: text for search terms (between quotes);
+* `<count>`: a positive integer number; set within allowed maximum & minimum limits;
+* `<locale>`: an ISO 639-3 locale code eg. `eng` for English (if set, the locale must be a valid ISO 639-3 code; if set to `none`, lexing will be disabled; if not set, the locale will be guessed from text);
+* `<manual>`: help manual to be shown (available manuals: `commands`);
+
+_Notice: the `bucket` terminology may confuse some Sonic users. As we are well-aware Sonic may be used in an environment where end-users may each hold their own search index in a given `collection`, we made it possible to manage per-end-user search indexes with `bucket`. If you only have a single index per `collection` (most Sonic users will), we advise you use a static generic name for your `bucket`, for instance: `default`._
+
+**⬇️ Search flow example (via `telnet`):**
+
+```bash
+T1: telnet sonic.local 1491
+T2: Trying ::1...
+T3: Connected to sonic.local.
+T4: Escape character is '^]'.
+T5: CONNECTED <sonic-server v1.0.0>
+T6: START search SecretPassword
+T7: STARTED search protocol(1) buffer(20000)
+T8: QUERY messages user:0dcde3a6 "valerian saliou" LIMIT(10)
+T9: PENDING Bt2m2gYa
+T10: EVENT QUERY Bt2m2gYa conversation:71f3d63b conversation:6501e83a
+T11: QUERY helpdesk user:0dcde3a6 "gdpr" LIMIT(50)
+T12: PENDING y57KaB2d
+T13: QUERY helpdesk user:0dcde3a6 "law" LIMIT(50) OFFSET(200)
+T14: PENDING CjPvE5t9
+T15: PING
+T16: PONG
+T17: EVENT QUERY CjPvE5t9
+T18: EVENT QUERY y57KaB2d article:28d79959
+T19: SUGGEST messages user:0dcde3a6 "val"
+T20: PENDING z98uDE0f
+T21: EVENT SUGGEST z98uDE0f valerian valala
+T22: QUIT
+T23: ENDED quit
+T24: Connection closed by foreign host.
+```
+
+_Notes on what happens:_
+
+* **T6:** we enter `search` mode (this is required to enable `search` commands);
+* **T8:** we query collection `messages`, in bucket for platform user `user:0dcde3a6` with search terms `valerian saliou` and a limit of `10` on returned results;
+* **T9:** Sonic received the query and stacked it for processing with marker `Bt2m2gYa` (the marker is used to track the asynchronous response);
+* **T10:** Sonic processed search query of T8 with marker `Bt2m2gYa` and sends 2 search results (those are conversation identifiers, that refer to a primary key in an external database);
+* **T11 + T13:** we query collection `helpdesk` twice (in the example, this one is heavy, so processing of results takes more time);
+* **T17 + T18:** we receive search results for search queries of T11 + T13 (this took a while!);
+
+---
+
+### 4️⃣ Sonic Channel (Ingest mode)
+
+_The Sonic Channel Ingest mode is used for altering the search index (push, pop and flush). Once in this mode, you cannot switch to other modes or gain access to commands from other modes._
+
+**➡️ Available commands:**
+
+* `PUSH`: Push search data in the index (syntax: `PUSH <collection> <bucket> <object> "<text>" [LANG(<locale>)]?`; time complexity: `O(1)`)
+* `POP`: Pop search data from the index (syntax: `POP <collection> <bucket> <object> "<text>"`; time complexity: `O(1)`)
+* `COUNTC`: Count buckets in a collection (syntax: `COUNT <collection>`; time complexity: `O(1)`)
+* `COUNTB`: Count objects in a bucket (syntax: `COUNT <collection> <bucket>`; time complexity: `O(1)`)
+* `COUNTO`: Count indexed terms in an object (syntax: `COUNT <collection> <bucket> <object>`; time complexity: `O(1)`)
+* `FLUSHC`: Flush all indexed data from a collection (syntax: `FLUSHC <collection>`; time complexity: `O(1)`)
+* `FLUSHB`: Flush all indexed data from a bucket in a collection (syntax: `FLUSHB <collection> <bucket>`; time complexity: `O(N)` where `N` is the number of bucket objects)
+* `FLUSHO`: Flush all indexed data from an object in a bucket in collection (syntax: `FLUSHO <collection> <bucket> <object>`; time complexity: `O(1)`)
+* `PING`: ping server (syntax: `PING`; time complexity: `O(1)`)
+* `HELP`: show help (syntax: `HELP [<manual>]?`; time complexity: `O(1)`)
+* `QUIT`: stop connection (syntax: `QUIT`; time complexity: `O(1)`)
+
+**👎 Deprecated commands:**
+
+* `COUNT`: Count indexed search data (syntax: `COUNT <collection> [<bucket> [<object>]?]?`; time complexity: `O(1)`)
+  * Deprecated because results for `<collection>`, `<collection> <bucket>` and `<collection> <bucket> <object>` are inconsistent. Use `COUNTC`/`COUNTB`/`COUNTO` instead.
+
+Note that Sonic doesn’t provide an `UPDATE` command, because of the lossy nature of its index. If you need to update an object, you have to `POP` it then `PUSH` the new data.
+
+**⏩ Syntax terminology:**
+
+* `<collection>`: index collection (ie. what you search in, eg. `messages`, `products`, etc.), 1 to 128 ASCII characters long;
+* `<bucket>`: index bucket name (ie. user-specific search classifier in the collection if you have any eg. `user-1, user-2, ..`, otherwise use a common bucket name eg. `generic, default, common, ..`), 1 to 128 ASCII characters long;
+* `<object>`: object identifier that refers to an entity in an external database, where the searched object is stored (eg. you use Sonic to index CRM contacts by name; full CRM contact data is stored in a MySQL database; in this case the object identifier in Sonic will be the MySQL primary key for the CRM contact), 1 to 128 ASCII characters long;
+* `<text>`: search text to be indexed (can be a single word, or a longer text; within maximum length safety limits; should be quoted using `"` quotes; internal quotes should be escaped using `\"`);
+* `<locale>`: an ISO 639-3 locale code eg. `eng` for English (if set, the locale must be a valid ISO 639-3 code; if set to `none`, lexing will be disabled; if not set, the locale will be guessed from text);
+* `<manual>`: help manual to be shown (available manuals: `commands`);
+
+_Notice: the `bucket` terminology may confuse some Sonic users. As we are well-aware Sonic may be used in an environment where end-users may each hold their own search index in a given `collection`, we made it possible to manage per-end-user search indexes with `bucket`. If you only have a single index per `collection` (most Sonic users will), we advise you use a static generic name for your `bucket`, for instance: `default`._
+
+**⬇️ Ingest flow example (via `telnet`):**
+
+```bash
+T1: telnet sonic.local 1491
+T2: Trying ::1...
+T3: Connected to sonic.local.
+T4: Escape character is '^]'.
+T5: CONNECTED <sonic-server v1.0.0>
+T6: START ingest SecretPassword
+T7: STARTED ingest protocol(1) buffer(20000)
+T8: PUSH messages user:0dcde3a6 conversation:71f3d63b Hey Valerian
+T9: ERR invalid_format(PUSH <collection> <bucket> <object> "<text>")
+T10: PUSH messages user:0dcde3a6 conversation:71f3d63b "Hello Valerian Saliou, how are you today?"
+T11: OK
+T12: COUNT messages user:0dcde3a6
+T13: RESULT 43
+T14: COUNT messages user:0dcde3a6 conversation:71f3d63b
+T15: RESULT 1
+T16: FLUSHO messages user:0dcde3a6 conversation:71f3d63b
+T17: RESULT 1
+T18: FLUSHB messages user:0dcde3a6
+T19: RESULT 42
+T20: PING
+T21: PONG
+T22: QUIT
+T23: ENDED quit
+T24: Connection closed by foreign host.
+```
+
+_Notes on what happens:_
+
+* **T6:** we enter `ingest` mode (this is required to enable `ingest` commands);
+* **T8:** we try to push text `Hey Valerian` to the index, in collection `messages`, bucket `user:0dcde3a6` and object `conversation:71f3d63b` (the syntax that was used is invalid);
+* **T9:** Sonic refuses the command we issued in T8, and provides us with the correct command format (notice that `<text>` should be quoted);
+* **T10:** we attempt to push another text in the same collection, bucket and object as in T8;
+* **T11:** this time, our push command in T10 was valid (Sonic acknowledges the push commit to the search index);
+* **T12:** we count the number of indexed terms in collection `messages` and bucket `user:0dcde3a6`;
+* **T13:** there are 43 terms (ie. words) in index for query in T12;
+* **T18:** we flush all index data from collection `messages` and bucket `user:0dcde3a6`;
+* **T19:** 42 terms have been flushed from index for command in T18;
+
+---
+
+### 5️⃣ Sonic Channel (Control mode)
+
+_The Sonic Channel Control mode is used for administration purposes. Once in this mode, you cannot switch to other modes or gain access to commands from other modes._
+
+**➡️ Available commands:**
+
+* `TRIGGER`: trigger an action (syntax: `TRIGGER [<action>]? [<data>]?`; time complexity: `O(1)`)
+* `INFO`: get server information (syntax: `INFO`; time complexity: `O(1)`)
+* `PING`: ping server (syntax: `PING`; time complexity: `O(1)`)
+* `HELP`: show help (syntax: `HELP [<manual>]?`; time complexity: `O(1)`)
+* `QUIT`: stop connection (syntax: `QUIT`; time complexity: `O(1)`)
+
+**⏩ Syntax terminology:**
+
+* `<action>`: action to be triggered (available actions: `consolidate`, `backup`, `restore`);
+* `<data>`: additional data to provide to the action (required for: `backup`, `restore`);
+* `<manual>`: help manual to be shown (available manuals: `commands`);
+
+**⬇️ Control flow example (via `telnet`):**
+
+```bash
+T1: telnet sonic.local 1491
+T2: Trying ::1...
+T3: Connected to sonic.local.
+T4: Escape character is '^]'.
+T5: CONNECTED <sonic-server v1.0.0>
+T6: START control SecretPassword
+T7: STARTED control protocol(1) buffer(20000)
+T8: TRIGGER consolidate
+T9: OK
+T10: PING
+T11: PONG
+T12: QUIT
+T13: ENDED quit
+T14: Connection closed by foreign host.
+```
+
+_Notes on what happens:_
+
+* **T6:** we enter `control` mode (this is required to enable `control` commands);
+* **T8:** we trigger a database consolidation (instead of waiting for the next automated consolidation tick);

README.md

@@ -0,0 +1,14 @@
+# Contributions to valeriansaliou/sonic
+
+🦔 Fast, lightweight & schema-less search backend. An alternative to Elasticsearch that runs on a few MBs of RAM.
+
+This repository holds only my commits to the upstream project, with their original author, date and diff.
+Each of my commits is preceded by a snapshot of the upstream files it touched, so every diff shows exactly what I changed.
+
+## Merged pull requests
+
+- [fix(core): Back up and restore FST buckets again](https://github.com/valeriansaliou/sonic/pull/432) (2026-10-05)
+- [docs: Document the 128 characters limit on collection, bucket and object names](https://github.com/valeriansaliou/sonic/pull/431) (2026-10-05)
+- [fix(core): Make `TRIGGER backup` work while stores are in use](https://github.com/valeriansaliou/sonic/pull/430) (2026-10-05)
+- [docs: clarify write_buffer_size](https://github.com/valeriansaliou/sonic/pull/418) (2026-09-24)
+- [Stop leaking a client slot when a channel read fails](https://github.com/valeriansaliou/sonic/pull/403) (2026-09-15)

config.cfg

@@ -0,0 +1,83 @@
+# Sonic
+# Fast, lightweight and schema-less search backend
+# Configuration file
+# Example: https://github.com/valeriansaliou/sonic/blob/master/config.cfg
+
+[server]
+log_level = "warn"
+
+
+[channel]
+inet = "[::1]:1491"
+tcp_timeout = 300
+
+# NOTE: You can pass this via `SONIC_CHANNEL__AUTH_PASSWORD`.
+auth_password = "SecretPassword"
+
+
+# WARN: After making changes to normalization steps, you must rebuild Sonic’s
+#   index by re-ingesting all data. To avoid such breaking change, new features
+#   are disabled by default (opt-in). On major releases, some steps might
+#   become enabled by default (opt-out). Sonic won’t enable unreasonable
+#   defaults, but override if you need consistency.
+[normalization]
+# TODO(major): Remove explicit override.
+unicode_normalization = "none"
+# TODO(major): Remove explicit override.
+diacritic_folding_enabled = false
+# WARN: Enabling stemming greatly affects the quality of Sonic results.
+#   Enable only if you have a good reason to.
+stemming_enabled = false
+
+
+# WARN: After making changes to tokenization steps, you must rebuild Sonic’s
+#   index by re-ingesting all data. To avoid such breaking change, new features
+#   are disabled by default (opt-in). On major releases, some steps might
+#   become enabled by default (opt-out). Sonic won’t enable unreasonable
+#   defaults, but override if you need consistency.
+[tokenization]
+# TODO(major): Remove explicit override.
+detect_special_patterns = true
+# TODO(major): Remove explicit override.
+compat_split_special_patterns = true
+
+
+[search]
+query_limit_default = 10
+query_limit_maximum = 100
+query_alternates_try = 4
+query_minimum_term_idf_default = 0.1
+query_minimum_term_idf_minimum_object_count = 100
+
+suggest_limit_default = 5
+suggest_limit_maximum = 20
+
+list_limit_default = 100
+list_limit_maximum = 500
+
+typo_factor_word_lengths = [4, 7, 10]
+
+
+[store.kv]
+path = "./data/store/kv/"
+
+retain_word_objects = 1000
+
+pool.inactive_after = 1800
+
+database.flush_after = 900
+database.compression_type = "zstd"
+database.parallelism = 2
+database.max_files = 100
+database.write_buffer_size = 16384
+database.write_ahead_log = true
+
+
+[store.fst]
+path = "./data/store/fst/"
+
+pool.inactive_after = 300
+
+graph.consolidate_after = 180
+graph.max_size = 2048
+graph.max_words = 250000

core/src/config.rs

@@ -0,0 +1,413 @@
+// Sonic
+//
+// Fast, lightweight and schema-less search backend
+// Copyright: 2019, Valerian Saliou <valerian@valeriansaliou.name>
+// Copyright: 2026, Rémi Bardon <remi@remibardon.name>
+// License: Mozilla Public License v2.0 (MPL v2.0)
+
+//! Sonic library configuration.
+//!
+//! It does not include server nor channel configuration, which are specific
+//! to the `sonic-server` binary.
+
+use std::collections::HashSet;
+use std::path::PathBuf;
+use std::sync::Arc;
+
+use serde::Deserialize;
+
+use crate::util::serde::env_var;
+
+#[derive(Debug, Deserialize)]
+pub struct Config {
+    pub normalization: NormalizationConfig,
+
+    pub tokenization: TokenizationConfig,
+
+    pub stopwords: StopwordsConfig,
+
+    pub search: SearchConfig,
+
+    pub store: StoreConfig,
+}
+
+impl Config {
+    pub fn validate(&self) {
+        // Check 'write_buffer' for KV
+        if self.store.kv.database.write_buffer_size == Some(0) {
+            panic!("write_buffer for kv must not be zero");
+        }
+
+        // Check 'flush_after' for KV
+        if self.store.kv.database.flush_after >= self.store.kv.pool.inactive_after {
+            panic!("flush_after for kv must be strictly lower than inactive_after");
+        }
+
+        // Check 'consolidate_after' for FST
+        if self.store.fst.graph.consolidate_after >= self.store.fst.pool.inactive_after {
+            panic!("consolidate_after for fst must be strictly lower than inactive_after");
+        }
+    }
+}
+
+/// Configuration group for normalization options (Unicode normalization,
+/// stemming, lemmatization…).
+#[derive(Debug, Deserialize, Clone, Copy)]
+pub struct NormalizationConfig {
+    #[serde(with = "crate::util::serde::none_string_as_none")]
+    pub unicode_normalization: Option<UnicodeNormalization>,
+
+    pub diacritic_folding_enabled: bool,
+
+    #[cfg(feature = "stemming")]
+    pub stemming_enabled: bool,
+}
+
+#[derive(Deserialize, Debug, Clone, Copy)]
+#[serde(rename_all = "snake_case")]
+pub enum UnicodeNormalization {
+    /// Unicode Normalization Form C.
+    Nfc,
+    /// Unicode Normalization Form KC.
+    Nfkc,
+}
+
+/// Configuration group for tokenization options.
+#[derive(Debug, Deserialize, Clone, Copy)]
+pub struct TokenizationConfig {
+    pub detect_special_patterns: bool,
+
+    #[serde(alias = "split_special_patterns")]
+    pub compat_split_special_patterns: bool,
+}
+
+#[derive(Debug, Deserialize, Clone, Default)]
+pub struct StopwordsConfig {
+    #[serde(deserialize_with = "to_stopwords")]
+    pub allow: HashSet<String>,
+
+    #[serde(deserialize_with = "to_stopwords")]
+    pub deny: HashSet<String>,
+}
+
+fn to_stopwords<'de, D>(deserializer: D) -> Result<HashSet<String>, D::Error>
+where
+    D: serde::de::Deserializer<'de>,
+{
+    use unicode_normalization::UnicodeNormalization as _;
+
+    let vec: Vec<Box<str>> = Deserialize::deserialize(deserializer)?;
+    let stopwords_iter = vec.into_iter().map(|s| s.nfkd().to_string());
+    Ok(HashSet::from_iter(stopwords_iter))
+}
+
+#[derive(Debug, Deserialize)]
+pub struct SearchConfig {
+    pub query_limit_default: u16,
+
+    pub query_limit_maximum: u16,
+
+    pub query_alternates_try: usize,
+
+    pub query_minimum_term_idf_default: f32,
+
+    pub query_minimum_term_idf_minimum_object_count: u64,
+
+    pub suggest_limit_default: u16,
+
+    pub suggest_limit_maximum: u16,
+
+    pub list_limit_default: u16,
+
+    pub list_limit_maximum: u16,
+
+    /// Word lengths from which one more typo is allowed when correcting words
+    /// (e.g. `[4, 7, 10]` allows 1 typo from 4 letters, 2 typos from 7 letters
+    /// and 3 typos from 10 letters). Empty means no typo correction.
+    pub typo_factor_word_lengths: Vec<usize>,
+}
+
+#[derive(Debug, Deserialize)]
+pub struct StoreConfig {
+    pub kv: Arc<KvStoreConfig>,
+
+    pub fst: Arc<FstStoreConfig>,
+}
+
+#[derive(Debug, Deserialize)]
+pub struct KvStoreConfig {
+    #[serde(deserialize_with = "env_var::path_buf")]
+    pub path: PathBuf,
+
+    pub retain_word_objects: usize,
+
+    pub pool: KvStorePoolConfig,
+
+    pub database: KvStoreDatabaseConfig,
+}
+
+#[derive(Debug, Deserialize)]
+pub struct KvStorePoolConfig {
+    pub inactive_after: u64,
+}
+
+#[derive(Debug, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct KvStoreDatabaseConfig {
+    pub flush_after: u64,
+
+    pub write_ahead_log: bool,
+
+    /// Whether or not to compress.
+    ///
+    /// Will get overriden if [`compression_type`](Self::compression_type) is
+    /// also specified.
+    #[serde(default)]
+    pub compress: Option<bool>,
+
+    #[serde(default)]
+    pub parallelism: Option<i32>,
+
+    #[serde(default)]
+    #[serde(alias = "max_files")]
+    pub max_open_files: Option<i32>,
+
+    // TODO(major): Make this MB, as in Kvrocks.
+    /// WARN: In KB!
+    #[serde(default = "default_write_buffer_size")]
+    #[serde(alias = "write_buffer")]
+    pub write_buffer_size: Option<usize>,
+
+    #[serde(default)]
+    pub max_write_buffer_number: Option<i32>,
+
+    #[serde(default)]
+    pub min_write_buffer_number: Option<i32>,
+
+    #[serde(default)]
+    pub min_write_buffer_number_to_merge: Option<i32>,
+
+    #[serde(default)]
+    pub block_cache_size: Option<u32>,
+
+    #[serde(default)]
+    pub cache_index_and_filter_blocks: Option<bool>,
+
+    #[serde(default)]
+    #[serde(alias = "compression")]
+    #[serde(deserialize_with = "to_rocksdb_compression_type_opt")]
+    pub compression_type: Option<rocksdb::DBCompressionType>,
+
+    #[serde(default)]
+    #[serde(alias = "wal_compression")]
+    #[serde(deserialize_with = "to_rocksdb_compression_type_opt")]
+    pub wal_compression_type: Option<rocksdb::DBCompressionType>,
+
+    #[serde(default)]
+    pub wal_ttl_seconds: Option<u64>,
+
+    #[serde(default)]
+    pub wal_size_limit_mb: Option<u64>,
+
+    #[serde(default)]
+    pub wal_bytes_per_sync: Option<u64>,
+
+    #[serde(default)]
+    #[serde(deserialize_with = "to_rocksdb_recovery_mode_opt")]
+    pub wal_recovery_mode: Option<rocksdb::DBRecoveryMode>,
+
+    #[serde(default)]
+    pub compression_level: Option<i32>,
+
+    #[serde(default)]
+    #[serde(alias = "compression_start_level")]
+    pub min_level_to_compress: Option<std::ffi::c_int>,
+
+    #[serde(default)]
+    #[serde(alias = "level0_file_num_compaction_trigger")]
+    pub level_zero_file_num_compaction_trigger: Option<i32>,
+
+    #[serde(default)]
+    #[serde(alias = "level0_slowdown_writes_trigger")]
+    pub level_zero_slowdown_writes_trigger: Option<i32>,
+
+    #[serde(default)]
+    #[serde(alias = "level0_stop_writes_trigger")]
+    pub level_zero_stop_writes_trigger: Option<i32>,
+
+    #[serde(default)]
+    pub max_bytes_for_level_base: Option<u64>,
+
+    #[serde(default)]
+    pub max_bytes_for_level_multiplier: Option<f64>,
+
+    #[serde(default)]
+    pub target_file_size_base: Option<u64>,
+
+    #[serde(default)]
+    pub max_background_jobs: Option<i32>,
+
+    #[serde(default)]
+    #[serde(alias = "max_compactions")]
+    pub max_subcompactions: Option<u32>,
+
+    #[serde(default)]
+    pub max_flushes: Option<i32>,
+
+    #[serde(default)]
+    pub stats_dump_period_sec: Option<u32>,
+}
+
+fn default_write_buffer_size() -> Option<usize> {
+    Some(16384)
+}
+
+fn parse_rocksdb_compression_type<E: serde::de::Error>(
+    str: &str,
+) -> Result<rocksdb::DBCompressionType, E> {
+    // NOTE: Some values are not available because not compiled in rocksdb
+    //   (feature flag is off).
+    match str.to_ascii_lowercase().as_str() {
+        "none" => Ok(rocksdb::DBCompressionType::None),
+        // "snappy" => Ok(rocksdb::DBCompressionType::Snappy),
+        // "zlib" => Ok(rocksdb::DBCompressionType::Zlib),
+        // "bz2" => Ok(rocksdb::DBCompressionType::Bz2),
+        // "lz4" => Ok(rocksdb::DBCompressionType::Lz4),
+        // "lz4hc" => Ok(rocksdb::DBCompressionType::Lz4hc),
+        "zstd" => Ok(rocksdb::DBCompressionType::Zstd),
+        _ => Err(serde::de::Error::unknown_variant(
+            str,
+            &[
+                "none",
+                // "snappy",
+                // "zlib",
+                // "bz2",
+                // "lz4",
+                // "lz4hc",
+                "zstd",
+            ],
+        )),
+    }
+}
+
+fn to_rocksdb_compression_type_opt<'de, D>(
+    deserializer: D,
+) -> Result<Option<rocksdb::DBCompressionType>, D::Error>
+where
+    D: serde::de::Deserializer<'de>,
+{
+    let str: Option<String> = Deserialize::deserialize(deserializer)?;
+    str.map(|s| parse_rocksdb_compression_type(&s)).transpose()
+}
+
+fn parse_rocksdb_recovery_mode<E: serde::de::Error>(
+    str: &str,
+) -> Result<rocksdb::DBRecoveryMode, E> {
+    match str.to_ascii_lowercase().as_str() {
+        "tolerate_corrupted_tail_records" | "TolerateCorruptedTailRecords" => {
+            Ok(rocksdb::DBRecoveryMode::TolerateCorruptedTailRecords)
+        }
+        "absolute_consistency" | "AbsoluteConsistency" => {
+            Ok(rocksdb::DBRecoveryMode::AbsoluteConsistency)
+        }
+        "point_in_time" | "PointInTime" => Ok(rocksdb::DBRecoveryMode::PointInTime),
+        "skip_any_corrupted_record" | "SkipAnyCorruptedRecord" => {
+            Ok(rocksdb::DBRecoveryMode::SkipAnyCorruptedRecord)
+        }
+        _ => Err(serde::de::Error::unknown_variant(
+            str,
+            &[
+                "tolerate_corrupted_tail_records",
+                "absolute_consistency",
+                "point_in_time",
+                "skip_any_corrupted_record",
+            ],
+        )),
+    }
+}
+
+fn to_rocksdb_recovery_mode_opt<'de, D>(
+    deserializer: D,
+) -> Result<Option<rocksdb::DBRecoveryMode>, D::Error>
+where
+    D: serde::de::Deserializer<'de>,
+{
+    let str: Option<String> = Deserialize::deserialize(deserializer)?;
+    str.map(|s| parse_rocksdb_recovery_mode(&s)).transpose()
+}
+
+#[derive(Debug, Deserialize)]
+pub struct FstStoreConfig {
+    #[serde(deserialize_with = "env_var::path_buf")]
+    pub path: PathBuf,
+
+    pub pool: FstStorePoolConfig,
+
+    pub graph: FstStoreGraphConfig,
+}
+
+#[derive(Debug, Deserialize)]
+pub struct FstStorePoolConfig {
+    pub inactive_after: u64,
+}
+
+#[derive(Debug, Deserialize)]
+pub struct FstStoreGraphConfig {
+    pub consolidate_after: u64,
+
+    pub max_size: usize,
+
+    pub max_words: usize,
+}
+
+#[cfg(test)]
+pub(crate) mod tests {
+    pub(crate) fn defaults_toml() -> &'static str {
+        r#"
+        [channel]
+        inet = "[::1]:1491"
+        tcp_timeout = 300
+
+        [normalization]
+        unicode_normalization = "none"
+        diacritic_folding_enabled = true
+        stemming_enabled = false
+
+        [tokenization]
+        detect_special_patterns = true
+        compat_split_special_patterns = false
+
+        [stopwords]
+        allow = []
+        deny = []
+
+        [search]
+        query_limit_default = 10
+        query_limit_maximum = 100
+        query_alternates_try = 4
+        query_minimum_term_idf_default = 0.1
+        query_minimum_term_idf_minimum_object_count = 100
+        suggest_limit_default = 5
+        suggest_limit_maximum = 20
+        list_limit_default = 100
+        list_limit_maximum = 500
+        typo_factor_word_lengths = [4, 7, 10]
+
+        [store.kv]
+        path = "./data/store/kv/"
+        retain_word_objects = 1000
+        pool.inactive_after = 1800
+        database.flush_after = 900
+        database.compression_type = "zstd"
+        database.parallelism = 2
+        database.write_ahead_log = true
+
+        [store.fst]
+        path = "./data/store/fst/"
+        pool.inactive_after = 300
+        graph.consolidate_after = 180
+        graph.max_size = 2048
+        graph.max_words = 250000
+        "#
+    }
+}

core/src/executor/search.rs

@@ -0,0 +1,616 @@
+// Sonic
+//
+// Fast, lightweight and schema-less search backend
+// Copyright: 2019, Valerian Saliou <valerian@valeriansaliou.name>
+// Copyright: 2026, Rémi Bardon <remi@remibardon.name>
+// License: Mozilla Public License v2.0 (MPL v2.0)
+
+use indexmap::IndexMap;
+
+use super::types::{QueryMatchScore, QueryResultScore, QuerySearchLimit, QuerySearchOffset};
+use crate::lexer::itertools::UniqueBy;
+use crate::lexer::preprocessor::{PreprocessorOutput, Token};
+use crate::store::StoreItemPart;
+use crate::store::fst::typo_factor;
+use crate::store::kv::KvStoreActionReadOnly;
+use crate::store::{StoreObjectIid, StoreTermHash};
+use crate::util::hash::NoopU32HasherBuilder;
+
+impl super::Executor {
+    pub fn search(
+        &self,
+        collection: StoreItemPart,
+        bucket: StoreItemPart,
+        input: PreprocessorOutput,
+        limit: QuerySearchLimit,
+        offset: QuerySearchOffset,
+    ) -> Result<Vec<String>, ()> {
+        // Important: acquire database access read lock, and reference it in context. This \
+        //   prevents the database from being erased while using it in this block.
+        let _kv_read_guard = self.kv_pool.lock_read_access();
+        let _fst_read_guard = self.fst_pool.lock_read_access();
+
+        let (Ok(kv_store), Ok(fst_store)) = (
+            self.kv_pool.acquire(false, collection, None, |_| {}),
+            self.fst_pool.acquire(collection, bucket),
+        ) else {
+            return Err(());
+        };
+
+        let Some(kv_store) = kv_store else {
+            tracing::debug!(
+                "collection store does not exist, consider {bucket:?} from {collection:?} empty"
+            );
+            return Ok(vec![]);
+        };
+
+        let (higher_limit, mut alternates_try) = (
+            self.app_conf.store.kv.retain_word_objects,
+            self.app_conf.search.query_alternates_try,
+        );
+
+        let (mut minimum_idf, idf_min_doc_count) = (
+            self.app_conf.search.query_minimum_term_idf_default,
+            (self.app_conf.search).query_minimum_term_idf_minimum_object_count,
+        );
+
+        let (prefix_matching_enabled, fuzzy_matching_enabled) = (
+            self.fst_pool.fst_action_config.prefix_matching_enabled,
+            self.fst_pool.fst_action_config.fuzzy_matching_enabled,
+        );
+
+        // Important: acquire bucket store read lock
+        executor_kv_lock_read!(kv_store);
+
+        let kv_action = kv_store.access_read_only(bucket);
+
+        let document_count = kv_action
+            .get_object_count()
+            .map_err(|error| tracing::warn!("{error:?}"))? as u64;
+
+        if document_count < idf_min_doc_count {
+            tracing::debug!(
+                "ignoring minimum_term_idf ({minimum_idf}) as document_count is too low ({document_count}<{idf_min_doc_count})"
+            );
+            minimum_idf = 0.;
+        }
+
+        // Collect all terms so we know the count right ahead.
+        // PERF: This helps allocating the correct amounts of memory.
+        let tokens: Vec<Token> =
+            UniqueBy::new_with_hasher(input.tokens(), Token::hash, NoopU32HasherBuilder).collect();
+        let term_count = tokens.len();
+
+        // Store scores for each found IID. Results will then be sorted by
+        // score before being returned. Scores are basically the sum of
+        // Levenshtein distances for each term in the query. Lower score
+        // means better result.
+        // NOTE: We use `IndexMap` instead of `HashMap` to preserve
+        //   insertion order, which correlates to reverse data ingestion
+        //   order.
+        // NOTE: `capacity = 24` to reduce initial grows.
+        let mut scoring_matrix: IndexMap<StoreObjectIid, Vec<Option<QueryMatchScore>>> =
+            IndexMap::with_capacity(24usize.min(usize::from(limit)));
+
+        // Look for exact matches.
+        'matches: for (idx, token) in tokens.iter().enumerate() {
+            let term_hash = token.hash();
+            let term = token.as_normalized();
+
+            let mut iids = kv_action
+                .get_term_to_iids(term_hash)
+                .unwrap_or(None)
+                .unwrap_or_default();
+
+            // Look for exact matches normalized differently if the Sonic
+            // index isn’t normalized.
+            if !token.is_special() && self.app_conf.normalization.unicode_normalization.is_none() {
+                use unicode_normalization::UnicodeNormalization as _;
+
+                let mut nfc = kv_action
+                    .get_term_to_iids(StoreTermHash::from(term.nfc().to_string().as_str()))
+                    .unwrap_or(None)
+                    .unwrap_or_default();
+                iids.append(&mut nfc);
+
+                let mut nfd = kv_action
+                    .get_term_to_iids(StoreTermHash::from(term.nfd().to_string().as_str()))
+                    .unwrap_or(None)
+                    .unwrap_or_default();
+                iids.append(&mut nfd);
+            };
+
+            tracing::debug!("got exact search executor iids: {iids:?} for term: {term:?}");
+
+            let document_frequency = document_frequency(term_hash, &kv_action);
+
+            // Filter out minimum IDF.
+            // PERF: Filtering `minimum_idf > 0` to save some computation.
+            if minimum_idf > 0. {
+                let idf = (document_count as f32 / document_frequency as f32).ln();
+                if idf < minimum_idf {
+                    tracing::debug!(
+                        "skipping term {term:?} because idf too low ({idf}<{minimum_idf})"
+                    );
+                    continue;
+                }
+            }
+
+            let bm25_score = bm25_lite_idf(document_count, document_frequency);
+
+            for iid in iids.into_iter() {
+                // Assign a base score of `1` as those are exact matches.
+                let inserted =
+                    update_score(&mut scoring_matrix, iid, 1. * bm25_score, idx, term_count);
+
+                if inserted {
+                    // Higher limit now reached?
+                    // Stop acquiring new suggested IIDs now.
+                    if scoring_matrix.len() >= higher_limit {
+                        tracing::trace!(?term, "got enough completed results for term");
+
+                        break 'matches;
+                    }
+                }
+            }
+        }
+
+        #[cfg(debug_assertions)]
+        tracing::debug!(?scoring_matrix);
+
+        // Look for words containing `term` as prefix.
+        if scoring_matrix.len() < higher_limit && alternates_try > 0 && prefix_matching_enabled {
+            tracing::debug!(
+                "not enough iids were found ({}/{higher_limit}), looking for prefixes",
+                scoring_matrix.len(),
+            );
+
+            'terms: for (idx, token) in tokens.iter().enumerate() {
+                let original_len = token.as_original().len();
+                let term = token.as_normalized();
+
+                let Some(suggestions) = fst_store.lookup_begins(term, original_len) else {
+                    tracing::trace!("did not get any completed word for term {term:?}");
+                    continue 'terms;
+                };
+
+                merge_suggestions(
+                    suggestions.map(|(w, distance)| (w, prefix_score(distance, original_len))),
+                    &mut scoring_matrix,
+                    term,
+                    idx,
+                    term_count,
+                    &kv_action,
+                    &mut alternates_try,
+                    higher_limit,
+                    document_count,
+                    minimum_idf,
+                );
+            }
+        }
+
+        #[cfg(debug_assertions)]
+        tracing::debug!(?scoring_matrix);
+
+        // Look for words like `term` (fuzzy matching).
+        if scoring_matrix.len() < higher_limit && alternates_try > 0 && fuzzy_matching_enabled {
+            tracing::debug!(
+                "not enough iids were found ({}/{higher_limit}), looking for fuzzy matches",
+                scoring_matrix.len(),
+            );
+
+            'terms: for (idx, token) in tokens.iter().enumerate() {
+                let original_word_len = token.as_original().len();
+                let term = token.as_normalized();
+
+                // Skip term if it’s special (we’d want exact matches only).
+                if token.is_special() {
+                    tracing::debug!("skipping fuzzy search for {term:?}: term is special");
+                    continue 'terms;
+                }
+
+                let max_typo_factor = typo_factor(
+                    original_word_len,
+                    &self.app_conf.search.typo_factor_word_lengths,
+                );
+                let mut typo_factor = 1u32;
+
+                // TODO: Rework the Levenshtein query feature to avoid repeating
+                //   the same query over and over again. Maybe try to see if
+                //   `fst_levenshtein` can return distances in its response.
+                while alternates_try > 0 && typo_factor <= max_typo_factor {
+                    let Some(suggestions) = fst_store.lookup_typos(term, typo_factor) else {
+                        tracing::trace!("did not get any completed word for term {term:?}");
+                        continue 'terms;
+                    };
+
+                    merge_suggestions(
+                        suggestions
+                            .map(|(w, distance)| (w, typo_score(distance, original_word_len))),
+                        &mut scoring_matrix,
+                        term,
+                        idx,
+                        term_count,
+                        &kv_action,
+                        &mut alternates_try,
+                        higher_limit,
+                        document_count,
+                        minimum_idf,
+                    );
+
+                    typo_factor += 1;
+                }
+            }
+        }
+
+        #[cfg(debug_assertions)]
+        tracing::debug!(?scoring_matrix);
+
+        // Switch to implicit `AND` if query contains a special token.
+        // NOTE: When a user queries for a special token (e.g. UUID),
+        //   they expect only exact matches to be returned. If one term is
+        //   considered special, we drop all results missing at least one
+        //   term. It’s not the most efficient (compared to not storing the
+        //   result in the first place) but it’s an edge case and the cost
+        //   is negligible.
+        let one_term_is_special = tokens.iter().any(Token::is_special);
+        if one_term_is_special {
+            let mut to_remove = Vec::<StoreObjectIid>::new();
+
+            for (&iid, scores) in scoring_matrix.iter() {
+                if scores.iter().any(Option::is_none) {
+                    to_remove.push(iid);
+                }
+            }
+
+            for iid in to_remove {
+                scoring_matrix.swap_remove(&iid);
+            }
+        }
+
+        // Flatten scores, taking into account missing matches (thanks to
+        // `None`).
+        let found_iids = scoring_matrix
+            .into_iter()
+            .map(|(iid, scores)| (iid, overall_score(&scores)));
+
+        // Sort found IIDs.
+        let all_iids = {
+            let mut all_iids = found_iids.collect::<Vec<_>>();
+            all_iids.sort_by(|a, b| a.1.total_cmp(&b.1).reverse());
+            all_iids.into_iter().map(|(iid, _score)| iid)
+        };
+
+        // Resolve OIDs from IIDs
+        // Notice: we also proceed paging from there
+        let (limit_usize, offset_usize) = (limit as usize, offset as usize);
+        let mut result_oids = Vec::with_capacity(limit_usize);
+
+        'paging: for (index, found_iid) in all_iids.skip(offset_usize).enumerate() {
+            // Stop there?
+            if index >= limit_usize {
+                break 'paging;
+            }
+
+            // Read IID-to-OID for this found IID
+            if let Ok(Some(oid)) = kv_action.get_iid_to_oid(found_iid) {
+                result_oids.push(oid);
+            } else {
+                tracing::error!("failed getting search executor iid-to-oid");
+            }
+        }
+
+        tracing::info!("got search executor final oids: {:?}", result_oids);
+
+        Ok(result_oids)
+    }
+}
+
+/// Inversely proportional to `lev_distance / word_len`, decreasing slowly
+/// towards `f(20) = 0.5`. Will never reach `0`.
+fn prefix_score(lev_distance: u16, word_len: usize) -> f32 {
+    // NOTE: Will be `> 1` in practice.
+    let lev_ratio = lev_distance as f32 / word_len as f32;
+
+    // NOTE: `20` means that auto-completed words 20 times longer than the
+    //   original word get a score of `0.5`. It’s just a magic number, it has
+    //   no further meaning. It just feels ok.
+    20. / (20. + lev_ratio)
+}
+
+#[cfg(test)]
+#[test]
+fn test_prefix_score() {
+    // Auto-complete 2 times longer.
+    for n in [2, 4, 8] {
+        assert_eq!(prefix_score(n, n as usize), 0.95238096);
+    }
+
+    // Auto-complete 4 times longer.
+    for n in [2, 4, 8] {
+        assert_eq!(prefix_score(3 * n, n as usize), 0.8695652);
+    }
+
+    // Auto-complete 1 character.
+    assert_eq!(prefix_score(1, 3), 0.9836065);
+    assert_eq!(prefix_score(1, 4), 0.9876543);
+    assert_eq!(prefix_score(1, 5), 0.99009895);
+    assert_eq!(prefix_score(1, 6), 0.9917356);
+
+    // Auto-complete 2 characters.
+    assert_eq!(prefix_score(2, 3), 0.96774197);
+    assert_eq!(prefix_score(2, 4), 0.9756098);
+    assert_eq!(prefix_score(2, 5), 0.98039216);
+    assert_eq!(prefix_score(2, 6), 0.9836065);
+
+    // More auto-complete means lower score.
+    for n in [1, 2, 4, 8] {
+        for word_len in [2, 4, 8, 10] {
+            assert!(prefix_score(n + 1, word_len as usize) < prefix_score(n, word_len as usize));
+        }
+    }
+}
+
+/// Levenshtein distance proportional to word length.
+fn typo_score(lev_distance: u16, word_len: usize) -> f32 {
+    debug_assert!(
+        (lev_distance as usize) < word_len,
+        "{lev_distance} >= {word_len}"
+    );
+
+    // SAFETY: `.min(1)` isn’t strictly necessary as `lev_distance` should
+    //   always be `< word_len`, but it’s there as a safety precaution.
+    let lev_ratio = (lev_distance as f32 / word_len as f32).min(1.);
+
+    1. - lev_ratio
+}
+
+#[cfg(test)]
+#[test]
+fn test_typo_score() {
+    // No typo.
+    assert_eq!(typo_score(0, 1), 1.);
+    assert_eq!(typo_score(0, 2), 1.);
+
+    // 1 typo.
+    assert_eq!(typo_score(1, 3), 0.6666666);
+    assert_eq!(typo_score(1, 4), 0.75);
+    assert_eq!(typo_score(1, 5), 0.8);
+    // 1 typo always scores lower than 0.
+    for n in 2..=8 {
+        assert!(typo_score(1, n) < typo_score(0, n), "n={n}");
+    }
+
+    // 2 typos.
+    assert_eq!(typo_score(2, 5), 0.6);
+    assert_eq!(typo_score(2, 6), 0.6666666);
+    assert_eq!(typo_score(2, 7), 0.71428573);
+    // 2 typos always scores lower than 1.
+    for n in 3..=8 {
+        assert!(typo_score(2, n) < typo_score(1, n), "n={n}");
+    }
+
+    // A lot of typos (length has no impact, proportion has).
+    assert_eq!(typo_score(1 * 20, 2 * 20), typo_score(1, 2));
+    assert_eq!(typo_score(3 * 20, 7 * 20), typo_score(3, 7));
+}
+
+fn overall_score(scores: &[Option<QueryMatchScore>]) -> QueryResultScore {
+    let total = scores.iter().map(|opt| opt.unwrap_or(0f32)).sum::<f32>();
+    let count = scores.len() as f32;
+
+    #[allow(clippy::let_and_return)]
+    let average = total / count;
+
+    average
+}
+
+#[cfg(test)]
+#[test]
+fn test_overall_score() {
+    const MISSING: Option<QueryMatchScore> = None;
+    const EXACT_MATCH: Option<QueryMatchScore> = Some(1.);
+
+    // Max score for exact matches.
+    assert_eq!(overall_score(&[EXACT_MATCH; 1]), 1.);
+    assert_eq!(overall_score(&[EXACT_MATCH; 2]), 1.);
+    assert_eq!(overall_score(&[EXACT_MATCH; 3]), 1.);
+    assert_eq!(overall_score(&[EXACT_MATCH; 4]), 1.);
+
+    // Lowest score for missing matches.
+    assert_eq!(overall_score(&[MISSING; 1]), 0.);
+    assert_eq!(overall_score(&[MISSING; 2]), 0.);
+    assert_eq!(overall_score(&[MISSING; 3]), 0.);
+    assert_eq!(overall_score(&[MISSING; 4]), 0.);
+
+    // Auto-complete > fuzzy matching (not always, but in most cases).
+    assert!(overall_score(&[Some(prefix_score(5, 4))]) > overall_score(&[Some(typo_score(1, 10))]));
+
+    // Missing one term.
+    assert_eq!(overall_score(&[MISSING, EXACT_MATCH]), 1. / 2.);
+    assert_eq!(overall_score(&[MISSING, EXACT_MATCH, EXACT_MATCH]), 2. / 3.);
+    // Missing one term is better than missing all terms.
+    assert!(overall_score(&[MISSING, EXACT_MATCH]) > overall_score(&[MISSING]));
+
+    // Term order has no meaning.
+    assert_eq!(
+        overall_score(&[
+            EXACT_MATCH,
+            Some(prefix_score(2, 3)),
+            Some(typo_score(2, 7))
+        ]),
+        overall_score(&[
+            Some(typo_score(2, 7)),
+            Some(prefix_score(2, 3)),
+            EXACT_MATCH
+        ])
+    );
+
+    // All typos in one term is like the same total across multiple terms.
+    // NOTE: This is not a requirement, it’s just a non-regression test.
+    assert_eq!(
+        overall_score(&[Some(typo_score(1, 7)); 2]),
+        overall_score(&[Some(typo_score(2, 7)), EXACT_MATCH])
+    );
+    assert_eq!(
+        overall_score(&[Some(typo_score(1, 7)); 3]),
+        overall_score(&[Some(typo_score(3, 7)), EXACT_MATCH, EXACT_MATCH])
+    );
+
+    // Examples for “The brown fox jumps over the lazy dog”:
+    // “brown fox jumps”
+    assert_eq!(overall_score(&[EXACT_MATCH, EXACT_MATCH, EXACT_MATCH]), 1.);
+    // “brown fox jum”
+    assert_eq!(
+        overall_score(&[EXACT_MATCH, EXACT_MATCH, Some(prefix_score(2, 3))]),
+        0.9892473
+    );
+    // “bron fox jum”
+    assert_eq!(
+        overall_score(&[
+            Some(typo_score(1, 5)),
+            EXACT_MATCH,
+            Some(prefix_score(2, 3))
+        ]),
+        0.92258066
+    );
+    // “brown fox”
+    assert_eq!(overall_score(&[EXACT_MATCH, EXACT_MATCH,]), 1.);
+    // “brown fox eats”
+    assert_eq!(
+        overall_score(&[EXACT_MATCH, EXACT_MATCH, MISSING]),
+        0.6666667
+    ); // 2/3
+}
+
+fn document_frequency(term_hash: StoreTermHash, kv_action: &KvStoreActionReadOnly<'_>) -> u64 {
+    kv_action
+        .get_term_to_iids(term_hash)
+        .inspect_err(|err| tracing::error!("{err:?}"))
+        .unwrap_or(None)
+        .map_or(0, |iids| iids.len()) as u64
+}
+
+fn bm25_lite_idf(document_count: u64, document_frequency: u64) -> f32 {
+    debug_assert!(
+        document_frequency <= document_count,
+        "{document_frequency} > {document_count}"
+    );
+
+    let document_count = document_count.max(document_frequency) as f64;
+    let df = document_frequency as f64;
+
+    (1.0 + (document_count - df + 0.5) / (df + 0.5)).ln() as f32
+}
+
+#[allow(clippy::too_many_arguments)] // We’ll refactor this someday, and it’ not public anyway.
+fn merge_suggestions(
+    suggestions: impl Iterator<Item = (String, QueryMatchScore)>,
+    scoring_matrix: &mut IndexMap<StoreObjectIid, Vec<Option<QueryMatchScore>>>,
+    term: &str,
+    term_idx: usize,
+    term_count: usize,
+    kv_action: &KvStoreActionReadOnly<'_>,
+    alternates_try: &mut usize,
+    higher_limit: usize,
+    document_count: u64,
+    minimum_idf: f32,
+) {
+    'suggestions: for (suggested_word, base_score) in suggestions {
+        // Do not load base results twice for same term as base term
+        if suggested_word.eq(term) {
+            continue;
+        }
+
+        tracing::trace!(?term, ?suggested_word, "got completed word for term");
+
+        let suggested_term_hash = StoreTermHash::from(suggested_word.as_str());
+        let suggested_iids = match kv_action.get_term_to_iids(suggested_term_hash) {
+            Ok(Some(suggested_iids)) => suggested_iids,
+            Ok(None) => continue,
+            Err(_) => continue,
+        };
+
+        let document_frequency = document_frequency(suggested_term_hash, kv_action);
+
+        // Filter out minimum IDF.
+        // PERF: Filtering `minimum_idf > 0` to save some computation.
+        if minimum_idf > 0. {
+            let idf = (document_count as f32 / document_frequency as f32).ln();
+            if idf < minimum_idf {
+                tracing::debug!(
+                    "skipping term {suggested_word:?} because idf too low ({idf}<{minimum_idf})"
+                );
+                continue;
+            }
+        }
+
+        let bm25_score = bm25_lite_idf(document_count, document_frequency);
+
+        let suggestion_score = base_score * bm25_score;
+
+        for suggested_iid in suggested_iids.into_iter().take(*alternates_try) {
+            // SAFETY: We can reach at most `alternates_try`.
+            *alternates_try = unsafe { alternates_try.unchecked_sub(1) };
+
+            let inserted = update_score(
+                scoring_matrix,
+                suggested_iid,
+                suggestion_score,
+                term_idx,
+                term_count,
+            );
+
+            if inserted {
+                // Higher limit now reached?
+                // Stop acquiring new suggested IIDs now.
+                if scoring_matrix.len() >= higher_limit {
+                    tracing::trace!(?term, "got enough completed results for term");
+
+                    break 'suggestions;
+                }
+            }
+        }
+    }
+
+    tracing::trace!(
+        ?term,
+        "done completing results for term, now {} total results",
+        scoring_matrix.len()
+    );
+}
+
+fn update_score(
+    scoring_matrix: &mut IndexMap<StoreObjectIid, Vec<Option<QueryMatchScore>>>,
+    iid: StoreObjectIid,
+    score: QueryMatchScore,
+    term_idx: usize,
+    term_count: usize,
+) -> bool {
+    match scoring_matrix.entry(iid) {
+        // If entry already exists, use lowest score.
+        indexmap::map::Entry::Occupied(mut occupied_entry) => {
+            // SAFETY: We always initialize vecs with `term_count` entries.
+            let entry_score = unsafe { occupied_entry.get_mut().get_unchecked_mut(term_idx) };
+
+            let new_score = entry_score.map_or(score, |entry_score| score.min(entry_score));
+
+            tracing::trace!(entry_score, new_score, "Updating to min score");
+            *entry_score = Some(new_score);
+
+            false
+        }
+        // If entry does not exist, insert score.
+        indexmap::map::Entry::Vacant(vacant_entry) => {
+            let mut scores = vec![None; term_count];
+
+            tracing::trace!(new_score = score, "Inserting new score");
+            // SAFETY: `scores` has `term_count` elements.
+            unsafe { *scores.get_unchecked_mut(term_idx) = Some(score) };
+
+            vacant_entry.insert(scores);
+
+            true
+        }
+    }
+}

core/src/executor/suggest.rs

@@ -0,0 +1,45 @@
+// Sonic
+//
+// Fast, lightweight and schema-less search backend
+// Copyright: 2019, Valerian Saliou <valerian@valeriansaliou.name>
+// Copyright: 2026, Rémi Bardon <remi@remibardon.name>
+// License: Mozilla Public License v2.0 (MPL v2.0)
+
+use super::types::QuerySearchLimit;
+use crate::lexer::preprocessor::PreprocessorOutput;
+use crate::store::StoreItemPart;
+use crate::store::fst::typo_factor;
+
+impl super::Executor {
+    pub fn suggest(
+        &self,
+        collection: StoreItemPart,
+        bucket: StoreItemPart,
+        input: PreprocessorOutput,
+        limit: QuerySearchLimit,
+    ) -> Result<Option<impl ExactSizeIterator<Item = String> + DoubleEndedIterator>, ()> {
+        // Important: acquire graph access read lock, and reference it in context. This \
+        //   prevents the graph from being erased while using it in this block.
+        let _fst_read_guard = self.fst_pool.lock_read_access();
+
+        if let Ok(fst_store) = self.fst_pool.acquire(collection, bucket) {
+            let mut tokens = input.tokens();
+
+            if let (Some(token), None) = (tokens.next(), tokens.next()) {
+                let len = token.as_original().len();
+                let term = token.into_normalized();
+                let max_typo_factor =
+                    typo_factor(len, &self.app_conf.search.typo_factor_word_lengths);
+
+                tracing::debug!("running suggest on word: {term:?}");
+
+                return match fst_store.suggest_words(term, len, limit as usize, max_typo_factor) {
+                    Some(words) => Ok(Some(words.map(|(k, _)| k))),
+                    None => Ok(None),
+                };
+            }
+        }
+
+        Err(())
+    }
+}

core/src/store/fst/backup.rs

@@ -0,0 +1,242 @@
+// Sonic
+//
+// Fast, lightweight and schema-less search backend
+// Copyright: 2019, Valerian Saliou <valerian@valeriansaliou.name>
+// Copyright: 2026, Rémi Bardon <remi@remibardon.name>
+// License: Mozilla Public License v2.0 (MPL v2.0)
+
+use std::fs::File;
+use std::path::Path;
+use std::{fs, io};
+
+use fst::Streamer as _;
+
+use super::{FstStoreId, FstStorePathMode, FstStorePool};
+
+impl FstStorePool {
+    pub fn backup(&self, path: &Path) -> Result<(), io::Error> {
+        tracing::debug!("backing up all fst stores to path: {path:?}");
+
+        // Create backup directory (full path)
+        fs::create_dir_all(path)?;
+
+        // Proceed dump action (backup)
+        self.dump_action(
+            "backup",
+            FstStorePathMode::Permanent,
+            &self.fst_store_config.path,
+            path,
+            &Self::backup_item,
+        )
+    }
+
+    pub fn restore(&self, path: &Path) -> Result<(), io::Error> {
+        tracing::debug!("restoring all fst stores from path: {path:?}");
+
+        // Proceed dump action (restore)
+        self.dump_action(
+            "restore",
+            FstStorePathMode::Backup,
+            path,
+            &self.fst_store_config.path,
+            &Self::restore_item,
+        )
+    }
+
+    #[allow(clippy::type_complexity)]
+    fn dump_action(
+        &self,
+        action: &str,
+        path_mode: FstStorePathMode,
+        read_path: &Path,
+        write_path: &Path,
+        fn_item: &dyn Fn(&Self, &Path, &Path, &str, &str) -> Result<(), io::Error>,
+    ) -> Result<(), io::Error> {
+        let fst_extension = path_mode.extension();
+
+        // Iterate on FST collections.
+        'collections: for collection_entry in fs::read_dir(read_path)? {
+            let collection_entry = collection_entry?;
+
+            // Actual collection found?
+            let file_type = collection_entry.file_type()?;
+            if !file_type.is_dir() {
+                tracing::trace!(
+                    ?file_type,
+                    "Found non-directory entry in {read_path:?}, ignoring"
+                );
+                continue 'collections;
+            }
+
+            let file_name = collection_entry.file_name();
+            let Some(collection_hash) = file_name.to_str() else {
+                tracing::warn!(
+                    file_name_bytes = ?file_name.as_encoded_bytes(),
+                    "Found invalid entry in {read_path:?}, ignoring"
+                );
+                continue 'collections;
+            };
+
+            tracing::debug!("fst collection ongoing {action}: {collection_hash}");
+
+            // Create write folder for collection.
+            fs::create_dir_all(write_path.join(&collection_hash))?;
+
+            // Iterate on FST collection buckets.
+            let buckets_path = read_path.join(&collection_hash);
+            'buckets: for bucket_entry in fs::read_dir(&buckets_path)? {
+                let bucket_entry = bucket_entry?;
+
+                // Actual bucket found?
+                let file_type = bucket_entry.file_type()?;
+                if !file_type.is_file() {
+                    tracing::trace!(
+                        ?file_type,
+                        "Found non-file entry in {buckets_path:?}, ignoring"
+                    );
+                    continue 'buckets;
+                }
+
+                let file_name = bucket_entry.file_name();
+                let Some(bucket_file_name) = file_name.to_str() else {
+                    tracing::warn!(
+                        file_name_bytes = ?file_name.as_encoded_bytes(),
+                        "Found invalid entry in {buckets_path:?}, ignoring"
+                    );
+                    continue 'buckets;
+                };
+
+                let Some(bucket_hash) = bucket_file_name.strip_suffix(fst_extension) else {
+                    tracing::trace!(
+                        file_name_bytes = ?file_name.as_encoded_bytes(),
+                        "Ignoring {bucket_file_name:?}: wrong extension (expected: {fst_extension:?})"
+                    );
+                    continue 'buckets;
+                };
+
+                tracing::debug!("fst bucket ongoing {action}: {collection_hash}/{bucket_hash}");
+
+                fn_item(
+                    self,
+                    write_path,
+                    &bucket_entry.path(),
+                    collection_hash,
+                    bucket_hash,
+                )?;
+            }
+        }
+
+        Ok(())
+    }
+
+    fn backup_item(
+        &self,
+        backup_path: &Path,
+        _origin_path: &Path,
+        collection_hash: &str,
+        bucket_hash: &str,
+    ) -> Result<(), io::Error> {
+        use io::Write as _;
+
+        // Acquire access lock (in blocking write mode), and reference it in context
+        // Notice: this prevents store to be acquired from any context.
+        let _access = self.graph_access_lock.write().unwrap();
+
+        // Generate path to FST backup.
+        let fst_backup_path = {
+            let ext = FstStorePathMode::Backup.extension();
+            assert!(ext.starts_with("."));
+            backup_path
+                .join(collection_hash)
+                .join(format!("{bucket_hash}{ext}"))
+        };
+
+        let store_id = FstStoreId::try_from_hex(collection_hash, bucket_hash)?;
+
+        tracing::debug!("fst store {store_id} backing up to path: {fst_backup_path:?}");
+
+        // Erase any previously-existing FST backup.
+        fs::remove_file(&fst_backup_path).ok();
+
+        // Stream actual FST data to FST backup.
+        let backup_fst_file = File::create(&fst_backup_path)?;
+        let mut backup_fst_writer = io::BufWriter::new(backup_fst_file);
+
+        let origin_fst = (self.open(store_id))
+            .map_err(|error| io::Error::other(format!("Graph open failure: {error:?}")))?;
+
+        let mut origin_fst_stream = origin_fst.stream();
+
+        let mut count_words = 0;
+        while let Some(word) = origin_fst_stream.next() {
+            count_words += 1;
+
+            // Write word, and append a new line.
+            backup_fst_writer.write_all(word)?;
+            backup_fst_writer.write_all(b"\n")?;
+        }
+
+        tracing::info!(
+            "fst store {store_id} backed up to path: {fst_backup_path:?} ({count_words} words)"
+        );
+
+        Ok(())
+    }
+
+    fn restore_item(
+        &self,
+        _backup_path: &Path,
+        origin_path: &Path,
+        collection_name: &str,
+        bucket_name: &str,
+    ) -> Result<(), io::Error> {
+        use io::BufRead as _;
+
+        // Acquire access lock (in blocking write mode) to prevent store from
+        // being acquired from any context.
+        let _access = self.graph_access_lock.write().unwrap();
+
+        let store_id = FstStoreId::try_from_hex(collection_name, bucket_name)?;
+
+        tracing::debug!("fst store {store_id} restoring from path: {origin_path:?}");
+
+        // Force a FST store close.
+        self.close(store_id);
+
+        // Generate path to FST.
+        let fst_path = self
+            .fst_store_config
+            .store_path(store_id, FstStorePathMode::Permanent);
+
+        // Remove existing FST data?
+        if fst_path.exists() {
+            fs::remove_file(&fst_path)?;
+        }
+
+        // Stream backup words to restored FST.
+        let fst_writer = io::BufWriter::new(File::create(&fst_path)?);
+        let fst_backup_reader = io::BufReader::new(File::open(&origin_path)?);
+
+        let mut fst_builder = fst::SetBuilder::new(fst_writer).map_err(|error| {
+            io::Error::other(format!("Graph restore builder failure: {error:?}"))
+        })?;
+
+        for word in fst_backup_reader.lines() {
+            let word = word?;
+
+            fst_builder.insert(word).map_err(|error| {
+                io::Error::other(format!("Graph restore word insert failure: {error:?}"))
+            })?;
+        }
+
+        fst_builder.finish().map_err(|error| {
+            io::Error::other(format!("Graph restore finish failure: {error:?}"))
+        })?;
+
+        tracing::info!(
+            "fst store {store_id} restored to path: {fst_path:?} from backup: {origin_path:?}"
+        );
+
+        Ok(())
+    }
+}

core/src/store/fst/mod.rs

@@ -0,0 +1,501 @@
+// Sonic
+//
+// Fast, lightweight and schema-less search backend
+// Copyright: 2019, Valerian Saliou <valerian@valeriansaliou.name>
+// Copyright: 2026, Rémi Bardon <remi@remibardon.name>
+// License: Mozilla Public License v2.0 (MPL v2.0)
+
+mod backup;
+mod pool;
+mod util;
+
+use std::fmt;
+use std::sync::{Arc, RwLock};
+use std::time::SystemTime;
+
+use fst::{IntoStreamer as _, Streamer as _};
+use hashbrown::HashSet;
+use regex_syntax::escape as regex_escape;
+
+use crate::lexer::ranges::LexerRegexRange;
+
+use super::generic::*;
+
+pub use self::pool::{FstStoreId, FstStorePool};
+use self::util::*;
+
+pub struct FstStore {
+    graph: fst::Set,
+    target: FstStoreId,
+    pending: FstStorePending,
+    last_used: Arc<RwLock<SystemTime>>,
+    last_consolidated: Arc<RwLock<SystemTime>>,
+    graph_consolidate: Arc<RwLock<HashSet<FstStoreId>>>,
+    // NOTE: This shouldn’t be here, but until a big rewrite let’s not care.
+    action_config: FstStoreActionConfig,
+}
+
+#[derive(Default)]
+pub struct FstStorePending {
+    pop: Arc<RwLock<HashSet<Vec<u8>>>>,
+    push: Arc<RwLock<HashSet<Vec<u8>>>>,
+}
+
+pub struct FstStoreActionBuilder<'build> {
+    pub fst_store_config: &'build crate::config::FstStoreConfig,
+}
+
+type FstStoreAtom = u32;
+
+pub struct FstStoreMisc;
+
+#[derive(Copy, Clone)]
+enum FstStorePathMode {
+    Permanent,
+    Temporary,
+    Backup,
+}
+
+impl FstStorePathMode {
+    fn extension(&self) -> &'static str {
+        match self {
+            FstStorePathMode::Permanent => ".fst",
+            FstStorePathMode::Temporary => ".fst.tmp",
+            FstStorePathMode::Backup => ".fst.bck",
+        }
+    }
+}
+
+#[derive(Debug, Clone, Copy)]
+pub struct FstStoreActionConfig {
+    pub prefix_matching_enabled: bool,
+    pub fuzzy_matching_enabled: bool,
+}
+
+impl Default for FstStoreActionConfig {
+    fn default() -> Self {
+        Self {
+            prefix_matching_enabled: true,
+            fuzzy_matching_enabled: true,
+        }
+    }
+}
+
+const WORD_LIMIT_LENGTH: usize = 40;
+
+impl FstStore {
+    pub fn cardinality(&self) -> usize {
+        self.graph.len()
+    }
+
+    pub fn as_stream(&self) -> fst::set::Stream<'_> {
+        self.graph.into_stream()
+    }
+
+    pub fn lookup_begins_(&self, word: &str) -> Result<fst::set::Stream<'_, fst_regex::Regex>, ()> {
+        // NOTE: This regex maps over an unicode range, for speed reasons at scale.
+        //   We found out that the 'match any' syntax ('.*') was super-slow. Using the restrictive
+        //   syntax below divided the cost of e.g. a search query by 2. The regex below has been
+        //   found out to be nearly zero-cost to compile and execute, for whatever reason.
+        // Regex format: '{escaped_word}([{unicode_range}]*)'
+        let mut regex_str = regex_escape(word);
+
+        regex_str.push('(');
+
+        LexerRegexRange::from(word)
+            .unwrap_or_default()
+            .write_to(&mut regex_str)
+            // Regex write failed? (this should not happen)
+            .map_err(|error| tracing::error!(
+                "Could not lookup word in fst via 'begins': {word:?} because regex write failed: {error:?}"
+            ))?;
+
+        regex_str.push_str("*)");
+
+        // Proceed word lookup.
+        tracing::debug!("Looking-up word in fst via 'begins': {word:?} with regex: {regex_str:?}");
+
+        let regex = fst_regex::Regex::new(&regex_str).map_err(|_error| ())?;
+
+        Ok(self.graph.search(regex).into_stream())
+    }
+
+    pub fn lookup_typos_(
+        &self,
+        word: &str,
+        typo_factor: u32,
+    ) -> Result<fst::set::Stream<'_, fst_levenshtein::Levenshtein>, ()> {
+        tracing::debug!(
+            "Looking-up word in fst via 'typos': {word:?} with typo factor: {typo_factor:?}"
+        );
+
+        let fuzzy = fst_levenshtein::Levenshtein::new(word, typo_factor).map_err(|_error| ())?;
+
+        Ok(self.graph.search(fuzzy).into_stream())
+    }
+
+    pub fn should_consolidate(&self) {
+        let id = self.target;
+
+        // Check if not already scheduled.
+        if self.graph_consolidate.read().unwrap().contains(&id) {
+            tracing::debug!("Graph consolidation already scheduled on pool: {id}");
+            return;
+        };
+
+        // Schedule target for next consolidation tick (i.e. collection + bucket tuple).
+        self.graph_consolidate.write().unwrap().insert(id);
+
+        // Bump “last consolidated” time, effectively de-bouncing consolidation
+        // to a fixed and predictable tick time in the future.
+        let mut last_consolidated_value = self.last_consolidated.write().unwrap();
+
+        *last_consolidated_value = SystemTime::now();
+
+        // Perform an early drop of the lock (frees up write lock early).
+        drop(last_consolidated_value);
+
+        tracing::info!("Graph consolidation scheduled on pool: {id}");
+    }
+}
+
+impl StoreGeneric for FstStore {
+    fn ref_last_used(&self) -> &RwLock<SystemTime> {
+        &self.last_used
+    }
+}
+
+impl FstStore {
+    pub fn push_word(&self, word: &str, fst_store_config: &crate::config::FstStoreConfig) -> bool {
+        // Word over limit? (abort, the FST does not perform well over large words)
+        if Self::word_over_limit(word) {
+            return false;
+        }
+
+        let word_bytes = word.as_bytes();
+
+        // Nuke word from 'pop' set? (void a previous un-consolidated commit)
+        if self.pending.pop.read().unwrap().contains(word_bytes) {
+            self.pending.pop.write().unwrap().remove(word_bytes);
+        }
+
+        // Add word in 'push' set? (only if word is not in FST)
+        // NOTE: also check whether FST is over limits or not from there, to avoid
+        //   stacking words that could never be consolidated to final FST anyway.
+        let graph_fst = self.graph.as_fst();
+
+        if self.graph.contains(&word) {
+            return false;
+        }
+
+        if check_over_limits(graph_fst.size(), graph_fst.len(), &fst_store_config.graph) {
+            return false;
+        }
+
+        {
+            let pending_push_guard = self.pending.push.read().unwrap();
+
+            if pending_push_guard.contains(word_bytes)
+                || pending_push_guard.len() >= fst_store_config.graph.max_words
+            {
+                return false;
+            }
+        }
+
+        (self.pending.push.write().unwrap()).insert(word_bytes.to_vec());
+
+        self.should_consolidate();
+
+        true
+    }
+
+    pub fn pop_word(&self, word: &str) -> bool {
+        // Word over limit? (abort, the FST does not perform well over large words)
+        if Self::word_over_limit(word) {
+            return false;
+        }
+
+        let word_bytes = word.as_bytes();
+
+        // Nuke word from 'push' set? (void a previous un-consolidated commit)
+        if self.pending.push.read().unwrap().contains(word_bytes) {
+            self.pending.push.write().unwrap().remove(word_bytes);
+        }
+
+        if !self.graph.contains(word_bytes) {
+            return false;
+        }
+
+        // Add word in 'pop' set? (only if word is in FST)
+        if self.pending.pop.read().unwrap().contains(word_bytes) {
+            return false;
+        }
+
+        (self.pending.pop.write().unwrap()).insert(word_bytes.to_vec());
+
+        self.should_consolidate();
+
+        true
+    }
+
+    pub fn suggest_words(
+        &self,
+        from_word: &str,
+        // Length before stemming. Useful to apply fuzzy matching rules based
+        // on user input.
+        original_word_len: usize,
+        limit: usize,
+        max_typo_factor: u32,
+    ) -> Option<impl ExactSizeIterator<Item = (String, u16)> + DoubleEndedIterator + use<>> {
+        use indexmap::IndexMap;
+
+        // Word over limit? (abort, the FST does not perform well over large words)
+        if Self::word_over_limit(from_word) {
+            return None;
+        }
+
+        let mut found_words: IndexMap<String, u16> = IndexMap::with_capacity(limit);
+
+        if self.action_config.prefix_matching_enabled {
+            // Try to complete provided word
+            if let Some(stream) = self.lookup_begins(from_word, original_word_len) {
+                for (word, score) in stream {
+                    if found_words.contains_key(&word) {
+                        continue;
+                    }
+
+                    found_words.insert(word, score);
+
+                    // Requested limit reached? Stop there.
+                    if found_words.len() >= limit {
+                        break;
+                    }
+                }
+            }
+        }
+
+        // Try to fuzzy-suggest other words? (e.g. correct typos)
+        if self.action_config.fuzzy_matching_enabled && found_words.len() < limit {
+            let mut typo_factor = 1u32;
+
+            // TODO: Rework the Levenshtein query feature to avoid repeating
+            //   the same query over and over again. Maybe try to see if
+            //   `fst_levenshtein` can return distances in its response.
+            while found_words.len() < limit && typo_factor <= max_typo_factor {
+                let Some(stream) = self.lookup_typos(from_word, typo_factor) else {
+                    break;
+                };
+
+                for (word, score) in stream {
+                    if found_words.contains_key(&word) {
+                        continue;
+                    }
+
+                    found_words.insert(word, score);
+
+                    // Requested limit reached? Stop there.
+                    if found_words.len() >= limit {
+                        break;
+                    }
+                }
+
+                typo_factor += 1;
+            }
+        }
+
+        if !found_words.is_empty() {
+            Some(found_words.into_iter())
+        } else {
+            None
+        }
+    }
+
+    pub fn lookup_begins(
+        &self,
+        word: &str,
+        // Length before stemming. Useful to calculate correct score.
+        original_word_len: usize,
+    ) -> Option<impl Iterator<Item = (String, u16)>> {
+        // Word over limit? (abort, the FST does not perform well over large words)
+        if Self::word_over_limit(word) {
+            return None;
+        }
+
+        if !self.action_config.prefix_matching_enabled {
+            return None;
+        }
+
+        let Ok(stream) = self.lookup_begins_(word) else {
+            return None;
+        };
+
+        tracing::debug!(?word, "looking up for word in 'begins' fst stream");
+
+        Some(FstStreamIterator(stream).map(move |word| {
+            // WARN: Calculating distance to original word length might
+            //   yield weird results when combines with stemming.
+            let distance: usize = original_word_len.abs_diff(word.len());
+            let score = u16::try_from(distance).unwrap_or(u16::MAX);
+            (word, score)
+        }))
+    }
+
+    pub fn lookup_typos(
+        &self,
+        word: &str,
+        typo_factor: u32,
+    ) -> Option<impl Iterator<Item = (String, u16)>> {
+        if !self.action_config.fuzzy_matching_enabled {
+            return None;
+        }
+
+        let Ok(stream) = self.lookup_typos_(word, typo_factor) else {
+            return None;
+        };
+
+        tracing::debug!(
+            ?word,
+            typo_factor,
+            "looking up for word in 'typos' fst stream"
+        );
+
+        // NOTE: Returning the same score for every word works only
+        //   because we re-run `lookup_typos` for increasingly
+        //   larger typo factors and do not re-insert existing
+        //   values. As explained in previous TODO, we should try
+        //   to get the real distance back from `fst_levenshtein`.
+        let score = u16::try_from(typo_factor).unwrap_or(u16::MAX);
+
+        Some(FstStreamIterator(stream).map(move |word| (word, score)))
+    }
+
+    pub fn list_words(&self, limit: usize, offset: usize) -> Result<Vec<String>, ()> {
+        let stream = self.as_stream();
+
+        // Enumerate words from FST stream.
+        match stream
+            .into_strs()
+            .map(|words| words.into_iter().skip(offset).take(limit).collect())
+        {
+            Err(err) => {
+                tracing::debug!("conversion of stream failed: {err:?}");
+                Err(())
+            }
+            Ok(words) => Ok(words),
+        }
+    }
+
+    pub fn count_words(&self) -> usize {
+        self.cardinality()
+    }
+
+    fn word_over_limit(word: &str) -> bool {
+        if word.len() > WORD_LIMIT_LENGTH {
+            tracing::debug!("got over-limit fst word: {word:?}");
+
+            true
+        } else {
+            false
+        }
+    }
+}
+
+/// Allow more typos in word as the word gets longer, up to a maximum limit.
+///
+/// `typo_factor_word_lengths` lists the word lengths from which one more typo
+/// is allowed (see `search.typo_factor_word_lengths`).
+pub(crate) fn typo_factor(word_len: usize, typo_factor_word_lengths: &[usize]) -> u32 {
+    typo_factor_word_lengths
+        .iter()
+        .filter(|&&min_word_len| word_len >= min_word_len)
+        .count() as u32
+}
+
+// MARK: - Helpers
+
+#[repr(transparent)]
+struct FstStreamIterator<'a, A: fst::Automaton>(fst::set::Stream<'a, A>);
+
+impl<'a, A: fst::Automaton> Iterator for FstStreamIterator<'a, A> {
+    type Item = String;
+
+    fn next(&mut self) -> Option<Self::Item> {
+        match self.0.next() {
+            Some(bytes) => match str::from_utf8(bytes) {
+                Ok(str) => Some(str.to_owned()),
+                Err(_) => None,
+            },
+            None => None,
+        }
+    }
+}
+
+// MARK: - Tests
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    pub(in crate::store::fst) fn test_fst_pool() -> FstStorePool {
+        let fst_store_config = test_fst_store_config();
+
+        FstStorePool::new(fst_store_config, Default::default())
+    }
+
+    pub(in crate::store::fst) fn test_fst_store_config() -> Arc<crate::config::FstStoreConfig> {
+        Arc::new(
+            config::Config::builder()
+                .add_source(config::File::from_str(
+                    crate::config::tests::defaults_toml(),
+                    config::FileFormat::Toml,
+                ))
+                .build()
+                .unwrap()
+                .get::<crate::config::FstStoreConfig>("store.fst")
+                .unwrap(),
+        )
+    }
+}
+
+// MARK: - Boilerplate
+
+impl fmt::Debug for FstStore {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        use crate::util::fmt::AsPrettyRwLock;
+
+        // NOTE: Deconstructing to future-proof this function.
+        let Self {
+            graph,
+            target,
+            pending,
+            last_used,
+            last_consolidated,
+            graph_consolidate,
+            action_config,
+        } = self;
+
+        f.debug_struct("FstStore")
+            .field("graph", graph)
+            .field("target", target)
+            .field("pending", pending)
+            .field("last_used", &AsPrettyRwLock(last_used))
+            .field("last_consolidated", &AsPrettyRwLock(last_consolidated))
+            .field("graph_consolidate", &AsPrettyRwLock(graph_consolidate))
+            .field("action_config", action_config)
+            .finish()
+    }
+}
+
+impl fmt::Debug for FstStorePending {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        use crate::util::fmt::AsPrettyRwLock;
+
+        // NOTE: Deconstructing to future-proof this function.
+        let Self { pop, push } = self;
+
+        f.debug_struct("FstStorePending")
+            .field("pop", &AsPrettyRwLock(pop))
+            .field("push", &AsPrettyRwLock(push))
+            .finish()
+    }
+}

core/src/store/kv/backup.rs

@@ -0,0 +1,187 @@
+// Sonic
+//
+// Fast, lightweight and schema-less search backend
+// Copyright: 2019, Valerian Saliou <valerian@valeriansaliou.name>
+// Copyright: 2026, Rémi Bardon <remi@remibardon.name>
+// License: Mozilla Public License v2.0 (MPL v2.0)
+
+use std::path::Path;
+use std::sync::Arc;
+use std::{fs, io};
+
+use rocksdb::backup::{
+    BackupEngine as DBBackupEngine, BackupEngineOptions as DBBackupEngineOptions,
+    RestoreOptions as DBRestoreOptions,
+};
+
+use super::{KvStoreId, KvStorePool};
+
+impl KvStorePool {
+    pub fn backup(&self, path: &Path) -> Result<(), io::Error> {
+        tracing::debug!("backing up all kv stores to path: {path:?}");
+
+        // Create backup directory (full path)
+        fs::create_dir_all(path)?;
+
+        // NOTE: The KV store directory gets created when the first KV store
+        //   is opened, so there is nothing to back up until then.
+        if !self.kv_store_config.path.exists() {
+            return Ok(());
+        }
+
+        // Proceed dump action (backup)
+        self.dump_action(
+            "backup",
+            &self.kv_store_config.path,
+            path,
+            &Self::backup_item,
+        )
+    }
+
+    pub fn restore(&self, path: &Path) -> Result<(), io::Error> {
+        tracing::debug!("restoring all kv stores from path: {path:?}");
+
+        // Proceed dump action (restore)
+        self.dump_action(
+            "restore",
+            path,
+            &self.kv_store_config.path,
+            &Self::restore_item,
+        )
+    }
+
+    #[allow(clippy::type_complexity)]
+    fn dump_action(
+        &self,
+        action: &str,
+        read_path: &Path,
+        write_path: &Path,
+        fn_item: &dyn Fn(&Self, &Path, &Path, &str) -> Result<(), io::Error>,
+    ) -> Result<(), io::Error> {
+        // Iterate on KV collections.
+        for entry in fs::read_dir(read_path)? {
+            let Ok(collection) = entry else {
+                continue;
+            };
+
+            // Actual collection found?
+            if !collection.file_type().is_ok_and(|f| f.is_dir()) {
+                continue;
+            }
+
+            if let Some(collection_hash) = collection.file_name().to_str() {
+                tracing::debug!("kv collection ongoing {action}: {collection_hash}");
+
+                fn_item(self, write_path, &collection.path(), collection_hash)?;
+            }
+        }
+
+        Ok(())
+    }
+
+    fn backup_item(
+        &self,
+        backup_path: &Path,
+        _origin_path: &Path,
+        collection_hash: &str,
+    ) -> Result<(), io::Error> {
+        // Acquire access lock (in blocking write mode), and reference it in context
+        // Notice: this prevents store to be acquired from any context
+        let _access = self.store_access_lock.write().unwrap();
+
+        // Generate path to KV backup
+        let kv_backup_path = backup_path.join(collection_hash);
+
+        let store_id = KvStoreId::try_from_hex(collection_hash)?;
+
+        tracing::debug!("kv store {store_id} backing up to path: {kv_backup_path:?}");
+
+        // Erase any previously-existing KV backup
+        if kv_backup_path.exists() {
+            fs::remove_dir_all(&kv_backup_path)?;
+        }
+
+        // Create backup folder for collection
+        fs::create_dir_all(backup_path.join(collection_hash))?;
+
+        // Initialize KV database backup engine
+        let kv_backup_options = DBBackupEngineOptions::new(&kv_backup_path)
+            .map_err(|_| io::Error::other("backup engine options acquire failure"))?;
+        let kv_backup_environment = rocksdb::Env::new()
+            .map_err(|_| io::Error::other("backup engine environment acquire failure"))?;
+
+        let mut kv_backup_engine = DBBackupEngine::open(&kv_backup_options, &kv_backup_environment)
+            .map_err(|_| io::Error::other("backup engine failure"))?;
+
+        // Proceed actual KV database backup
+        // NOTE: RocksDB refuses to open a database which is already open in
+        //   this process (its `LOCK` file is held), so back up the pooled
+        //   store if there is one. Flush it first, as its memtables may hold
+        //   writes not persisted yet (e.g. if the write-ahead log is disabled).
+        let pooled_store = self.read().unwrap().get(&store_id).map(Arc::clone);
+
+        match pooled_store {
+            Some(store) => kv_backup_engine.create_new_backup_flush(&store.database, true),
+            None => {
+                let origin_kv = self
+                    .open(store_id, |_| {})
+                    .map_err(|_| io::Error::other("database open failure"))?;
+
+                kv_backup_engine.create_new_backup(&origin_kv)
+            }
+        }
+        .map_err(|_| io::Error::other("database backup failure"))?;
+
+        tracing::info!("kv store {store_id} backed up to path: {kv_backup_path:?}");
+
+        Ok(())
+    }
+
+    fn restore_item(
+        &self,
+        _backup_path: &Path,
+        origin_path: &Path,
+        collection_hash: &str,
+    ) -> Result<(), io::Error> {
+        // Acquire access lock (in blocking write mode), and reference it in context
+        // Notice: this prevents store to be acquired from any context
+        let _access = self.store_access_lock.write().unwrap();
+
+        let store_id = KvStoreId::try_from_hex(collection_hash)?;
+
+        tracing::debug!("kv store {store_id} restoring from path: {origin_path:?}");
+
+        // Force a KV store close
+        self.close(store_id, None);
+
+        // Generate path to KV
+        let kv_path = self.kv_store_config.store_path(store_id);
+
+        // Remove existing KV database data?
+        if kv_path.exists() {
+            fs::remove_dir_all(&kv_path)?;
+        }
+
+        // Create KV folder for collection
+        fs::create_dir_all(&kv_path)?;
+
+        // Initialize KV database backup engine
+        let kv_backup_options = DBBackupEngineOptions::new(&origin_path)
+            .map_err(|_| io::Error::other("backup engine options acquire failure"))?;
+        let kv_backup_environment = rocksdb::Env::new()
+            .map_err(|_| io::Error::other("backup engine environment acquire failure"))?;
+
+        let mut kv_backup_engine = DBBackupEngine::open(&kv_backup_options, &kv_backup_environment)
+            .map_err(|_| io::Error::other("backup engine failure"))?;
+
+        kv_backup_engine
+            .restore_from_latest_backup(&kv_path, &kv_path, &DBRestoreOptions::default())
+            .map_err(|_| io::Error::other("database restore failure"))?;
+
+        tracing::info!(
+            "kv store {store_id} restored to path: {kv_path:?} from backup: {origin_path:?}"
+        );
+
+        Ok(())
+    }
+}

core/tests/common/config.rs

@@ -0,0 +1,81 @@
+// Sonic
+//
+// Fast, lightweight and schema-less search backend
+// Copyright: 2026, Rémi Bardon <remi@remibardon.name>
+// License: Mozilla Public License v2.0 (MPL v2.0)
+
+use crate::common::util::unique_hex;
+
+pub fn make_config(toml: &str) -> sonic::Config {
+    let raw_config: config::Config = config::Config::builder()
+        .add_source(config::File::from_str(toml, config::FileFormat::Toml))
+        .build()
+        .expect("error reading config");
+
+    // Parse configuration.
+    let config = raw_config
+        .try_deserialize::<sonic::Config>()
+        .expect("syntax error in config");
+
+    // Validate configuration.
+    config.validate();
+
+    config
+}
+
+pub fn defaults_toml() -> String {
+    let test_id = unique_hex().unwrap();
+
+    let test_data_path = std::path::Path::new(env!("CARGO_TARGET_TMPDIR")).join(&test_id);
+    std::fs::create_dir_all(&test_data_path).unwrap();
+
+    let kv_store_path = test_data_path.join("kv-store");
+    std::fs::create_dir(&kv_store_path).unwrap();
+    let fst_store_path = test_data_path.join("fst-store");
+    std::fs::create_dir(&fst_store_path).unwrap();
+
+    format!(
+        r#"
+        [normalization]
+        unicode_normalization = "none"
+        diacritic_folding_enabled = true
+        stemming_enabled = false
+
+        [tokenization]
+        detect_special_patterns = true
+        compat_split_special_patterns = false
+
+        [stopwords]
+        allow = []
+        deny = []
+
+        [search]
+        query_limit_default = 10
+        query_limit_maximum = 100
+        query_alternates_try = 4
+        query_minimum_term_idf_default = 0.1
+        query_minimum_term_idf_minimum_object_count = 100
+        suggest_limit_default = 5
+        suggest_limit_maximum = 20
+        list_limit_default = 100
+        list_limit_maximum = 500
+        typo_factor_word_lengths = [4, 7, 10]
+
+        [store.kv]
+        path = {kv_store_path:?}
+        retain_word_objects = 1000
+        pool.inactive_after = 1800
+        database.flush_after = 900
+        database.compression_type = "zstd"
+        database.parallelism = 2
+        database.write_ahead_log = true
+
+        [store.fst]
+        path = {fst_store_path:?}
+        pool.inactive_after = 300
+        graph.consolidate_after = 180
+        graph.max_size = 2048
+        graph.max_words = 250000
+        "#,
+    )
+}

core/tests/common/mod.rs

@@ -0,0 +1,196 @@
+// Sonic
+//
+// Fast, lightweight and schema-less search backend
+// Copyright: 2026, Rémi Bardon <remi@remibardon.name>
+// License: Mozilla Public License v2.0 (MPL v2.0)
+
+#![allow(unused)]
+
+pub mod config;
+mod executor;
+mod logging;
+mod meta_tests;
+pub mod util;
+
+pub use self::executor::*;
+pub(crate) use self::item_ref::*;
+pub use self::logging::*;
+pub(crate) use self::meta_tests::*;
+pub(crate) use self::util::assert_contains;
+
+// NOTE: Using macros instead of functions so `unwrap`s point to the call site
+//   (helps debugging).
+pub(crate) mod item_ref {
+    macro_rules! collection_ref {
+        ($collection:expr) => {
+            sonic::store::StoreItemBuilder::from_depth_1($collection).unwrap()
+        };
+    }
+    pub(crate) use collection_ref;
+
+    macro_rules! bucket_ref {
+        ($collection:expr, $bucket:expr) => {
+            sonic::store::StoreItemBuilder::from_depth_2($collection, $bucket).unwrap()
+        };
+    }
+    pub(crate) use bucket_ref;
+
+    macro_rules! object_ref {
+        ($collection:expr, $bucket:expr, $object:expr) => {
+            sonic::store::StoreItemBuilder::from_depth_3($collection, $bucket, $object).unwrap()
+        };
+    }
+    pub(crate) use object_ref;
+}
+
+macro_rules! exec {
+    ($executor:ident -> PUSH $collection:tt $bucket:tt $oid:tt $text:tt $(LANG($lang:expr))? NEW) => {
+        exec!(internal_ $executor -> PUSH $collection $bucket $oid $text true $(LANG($lang))?)
+    };
+    ($executor:ident -> PUSH $collection:tt $bucket:tt $oid:tt $text:tt $(LANG($lang:expr))?) => {
+        exec!(internal_ $executor -> PUSH $collection $bucket $oid $text false $(LANG($lang))?)
+    };
+    (internal_ $executor:ident -> PUSH $collection:tt $bucket:tt $oid:tt $text:tt $assume_new:ident $(LANG($lang:expr))?) => {{
+        #[rustfmt::skip]
+        $executor.log(format!(
+            "PUSH {:?} {:?} {:?} {:?}{}{}",
+            $collection, $bucket, $oid, $text, exec!(internal_ lang_txt $($lang)?), if $assume_new { " NEW" } else { "" }
+        ));
+        let (c, b, o) = crate::common::object_ref!($collection, $bucket, $oid);
+        let preprocessor = sonic::lexer::preprocessor::Preprocessor::new(
+            $executor.app_conf.tokenization,
+            $executor.app_conf.normalization,
+            $executor.app_conf.stopwords.clone(),
+            true,
+            true,
+        );
+        $executor
+            .push(
+                c, b, o,
+                preprocessor.preprocess($text, exec!(internal_ lang $($lang)?)),
+                $assume_new,
+            )
+            .unwrap()
+    }};
+
+    ($executor:ident -> TRIGGER consolidate) => {{
+        $executor.log(format!("TRIGGER consolidate"));
+        $executor.fst_pool.consolidate(true, |_| true)
+    }};
+
+    ($executor:ident -> COUNT $collection:tt) => {{
+        $executor.log(format!("COUNT {:?}", $collection));
+        let c = collection_ref!($collection);
+        $executor.countc(c)
+    }};
+
+    ($executor:ident -> COUNT $collection:tt $bucket:tt) => {{
+        $executor.log(format!("COUNT {:?} {:?}", $collection, $bucket));
+        let (c, b) = bucket_ref!($collection, $bucket);
+        $executor.legacy_countb(c, b)
+    }};
+
+    ($executor:ident -> COUNTB $collection:tt $bucket:tt) => {{
+        $executor.log(format!("COUNTB {:?} {:?}", $collection, $bucket));
+        let (c, b) = bucket_ref!($collection, $bucket);
+        $executor.countb(c, b)
+    }};
+
+    ($executor:ident -> COUNT $collection:tt $bucket:tt $oid:tt) => {{
+        $executor.log(format!("COUNT {:?} {:?} {:?}", $collection, $bucket, $oid));
+        let (c, b, o) = object_ref!($collection, $bucket, $oid);
+        $executor.counto(c, b, o)
+    }};
+
+    ($executor:ident -> QUERY $collection:tt $bucket:tt $term:tt $(LANG($lang:expr))? $(LIMIT($limit:expr))?) => {{
+        #[rustfmt::skip]
+        $executor.log(format!(
+            "QUERY {:?} {:?} {:?}{}{}",
+            $collection, $bucket, $term, exec!(internal_ lang_txt $($lang)?), exec!(internal_ limit_txt $($limit)?)
+        ));
+        let (c, b) = crate::common::bucket_ref!($collection, $bucket);
+        let preprocessor = sonic::lexer::preprocessor::Preprocessor::new(
+            $executor.app_conf.tokenization,
+            $executor.app_conf.normalization,
+            $executor.app_conf.stopwords.clone(),
+            true,
+            true,
+        );
+        $executor
+            .search(
+                c, b,
+                preprocessor.preprocess($term, exec!(internal_ lang $($lang)?)),
+                exec!(internal_ limit $($limit)?),
+                0,
+            )
+            .expect("QUERY should succeed")
+    }};
+
+    ($executor:ident -> SUGGEST $collection:tt $bucket:tt $word:tt $(LANG($lang:expr))?) => {{
+        #[rustfmt::skip]
+        $executor.log(format!(
+            "SUGGEST {:?} {:?} {:?}{}",
+            $collection, $bucket, $word, exec!(internal_ lang_txt $($lang)?)
+        ));
+        let (c, b) = crate::common::bucket_ref!($collection, $bucket);
+        let preprocessor = sonic::lexer::preprocessor::Preprocessor::new(
+            $executor.app_conf.tokenization,
+            $executor.app_conf.normalization,
+            $executor.app_conf.stopwords.clone(),
+            false,
+            false,
+        );
+        $executor
+            .suggest(
+                c, b,
+                preprocessor.preprocess($word, exec!(internal_ lang $($lang)?)),
+                $executor.app_conf.search.suggest_limit_default,
+            )
+            .expect("SUGGEST should succeed")
+            .map(|words| words.collect::<Vec<String>>())
+            .unwrap_or_default()
+    }};
+
+    ($executor:ident -> LIST $collection:tt $bucket:tt $(LIMIT($limit:expr))?) => {{
+        $executor.log(format!("LIST {:?} {:?}", $collection, $bucket));
+        let (c, b) = crate::common::bucket_ref!($collection, $bucket);
+        $executor
+            .list(
+                c, b,
+                exec!(internal_ limit $($limit)?),
+                0,
+            )
+            .expect("LIST should succeed")
+    }};
+
+    ($executor:ident -> FLUSHC $collection:tt) => {{
+        $executor.log(format!("FLUSHC {:?}", $collection));
+        let c = crate::common::collection_ref!($collection);
+        $executor.flushc(c).unwrap()
+    }};
+
+    ($executor:ident -> FLUSHB $collection:tt $bucket:tt) => {{
+        $executor.log(format!("FLUSHB {:?} {:?}", $collection, $bucket));
+        let (c, b) = crate::common::bucket_ref!($collection, $bucket);
+        $executor.flushb(c, b).unwrap()
+    }};
+
+    ($executor:ident -> FLUSHO $collection:tt $bucket:tt $oid:tt) => {{
+        $executor.log(format!("FLUSHO {:?} {:?} {:?}", $collection, $bucket, $oid));
+        let (c, b, o) = crate::common::object_ref!($collection, $bucket, $oid);
+        $executor.flusho(c, b, o).unwrap()
+    }};
+
+    (internal_ lang) => { None };
+    (internal_ lang $lang:expr) => { whatlang::Lang::from_code($lang) };
+
+    (internal_ lang_txt) => { "" };
+    (internal_ lang_txt $lang:expr) => { format!(" LANG({})", $lang) };
+
+    (internal_ limit) => { sonic::executor::QuerySearchLimit::MAX };
+    (internal_ limit $limit:expr) => { $limit };
+
+    (internal_ limit_txt) => { "" };
+    (internal_ limit_txt $limit:expr) => { format!(" LLIMIT({})", $limit) };
+}
+pub(crate) use exec;

core/tests/data_backup.rs

@@ -0,0 +1,46 @@
+// Sonic
+//
+// Fast, lightweight and schema-less search backend
+// Copyright: 2026, DualFroz <me@dualfroz.com>
+// License: Mozilla Public License v2.0 (MPL v2.0)
+
+//! Feature: Data backup
+
+mod common;
+
+use crate::common::util::unique_hex;
+use crate::common::*;
+
+/// Consolidated FST stores are backed up, and can be restored.
+#[test]
+fn test_fst_backup_and_restore() {
+    init_logging();
+    let executor = make_test_executor(|_| {});
+
+    exec!(executor -> PUSH "messages" "user:1" "chat:1" "Hello world");
+    exec!(executor -> TRIGGER consolidate);
+
+    let backup_path = std::path::Path::new(env!("CARGO_TARGET_TMPDIR"))
+        .join(unique_hex().unwrap())
+        .join("fst-backup");
+
+    executor.fst_pool.backup(&backup_path).unwrap();
+
+    let backup_files = std::fs::read_dir(&backup_path)
+        .unwrap()
+        .flat_map(|collection| std::fs::read_dir(collection.unwrap().path()).unwrap())
+        .map(|bucket| bucket.unwrap().path())
+        .collect::<Vec<_>>();
+    assert_eq!(backup_files.len(), 1, "got: {backup_files:?}");
+    assert!(
+        backup_files[0].to_string_lossy().ends_with(".fst.bck"),
+        "got: {backup_files:?}"
+    );
+
+    let restored_executor = make_test_executor(|_| {});
+
+    restored_executor.fst_pool.restore(&backup_path).unwrap();
+
+    let response = exec!(restored_executor -> LIST "messages" "user:1");
+    assert_contains!(response, ["hello", "world"]);
+}

core/tests/typo_correction.rs

@@ -0,0 +1,149 @@
+// Sonic
+//
+// Fast, lightweight and schema-less search backend
+// Copyright: 2026, Rémi Bardon <remi@remibardon.name>
+// License: Mozilla Public License v2.0 (MPL v2.0)
+
+//! Feature: Typo correction
+//!
+//! A “typo factor” is chosen based on the token’s length. Those tests ensure
+//! typo correction works, and ensure no regression in the quality of the
+//! results.
+//! The mapping is configured by `search.typo_factor_word_lengths`. With the
+//! default value (`[4, 7, 10]`), it is the following:
+//!
+//! ```
+//! let mut typo_factor = match word.len() {
+//!     1..=3 => 0,
+//!     4..=6 => 1,
+//!     7..=9 => 2,
+//!     _ => 3,
+//! };
+//! ```
+
+mod common;
+
+use crate::common::*;
+
+/// This test sentence contains words that are 3–10 characters-long, while
+/// not being stopwords. It’s not realistic but not having stopwords avoids
+/// false positives in tests.
+const ASTRONOMY_WORDS: &str = "sun moon comet nebula pulsars asteroid satellite spacecraft";
+
+/// Search should allow a certain number of typos (depending on token length).
+///
+/// This covers missing letters, inverted letters and additional letters.
+/// The underlying algorithm used is the Levenshtein distance, so this test is
+/// based on that knowledge.
+#[test]
+fn test_search_allows_typos() {
+    // NOTE: Need to make language explicit because of
+    //   <https://github.com/valeriansaliou/sonic/issues/322#issuecomment-4638688602>.
+    //   Will be fixed separately.
+    #[rustfmt::skip]
+    test_ingest_then_query!(push: ASTRONOMY_WORDS [ensure_no_stopword] LANG("eng"), query: [
+        // 3-letter word, distance = 1.
+        ("sum", false),
+        // 4-letter word, distance = 1.
+        ("ssun", true), // sun
+        ("noon", true), // moon
+        // 4-letter word, distance = 2.
+        ("commit", false),
+        // 6-letter word, distance = 1.
+        ("nzbula", true), // nebula
+        // 6-letter word, distance = 2.
+        ("nzbala", false),
+        // 7-letter word, distance = 2.
+        ("plusars", true), // pulsars
+        // 9-letter word, distance = 2.
+        ("saetllite", true), // satellite
+        // 9-letter word, distance = 3.
+        ("satemmote", false),
+        // 10-letter word, distance = 3.
+        ("sapcecrzft", true), // spacecraft
+        // 10-letter word, distance = 4.
+        ("sapcecarft", false),
+    ] LANG("eng"));
+}
+
+/// The word length -> typo factor mapping can be configured.
+///
+/// See <https://github.com/valeriansaliou/sonic/issues/394>.
+#[test]
+fn test_config_typo_factor_word_lengths() {
+    init_logging();
+    let executor = make_test_executor(|app_conf| {
+        // 1 typo from 3 letters, 2 typos from 5 letters, never 3 typos.
+        app_conf.search.typo_factor_word_lengths = vec![3, 5];
+    });
+
+    exec!(executor -> PUSH "messages" "user:1" "chat:1" ASTRONOMY_WORDS LANG("eng"));
+    exec!(executor -> TRIGGER consolidate);
+
+    #[rustfmt::skip]
+    let examples: [(&str, &[&str]); 3] = [
+        // 3-letter word, distance = 1.
+        ("sum", &["sun"]),
+        // 6-letter word, distance = 2.
+        ("nzbala", &["nebula"]),
+        // 10-letter word, distance = 3.
+        ("sapcecrzft", &[]),
+    ];
+
+    for (needle, expected_suggestions) in examples {
+        let expected_response: &[&str] = if expected_suggestions.is_empty() {
+            &[]
+        } else {
+            &["chat:1"]
+        };
+
+        let response = exec!(executor -> QUERY "messages" "user:1" needle LANG("eng"));
+        assert_eq!(response, expected_response, "QUERY {needle:?}");
+
+        let suggestions = exec!(executor -> SUGGEST "messages" "user:1" needle LANG("eng"));
+        assert_eq!(suggestions, expected_suggestions, "SUGGEST {needle:?}");
+    }
+}
+
+/// Typo correction can be disabled by configuring no word length.
+///
+/// See <https://github.com/valeriansaliou/sonic/issues/394>.
+#[test]
+fn test_config_typo_correction_disabled() {
+    init_logging();
+    let executor = make_test_executor(|app_conf| {
+        app_conf.search.typo_factor_word_lengths = vec![];
+    });
+
+    exec!(executor -> PUSH "messages" "user:1" "chat:1" ASTRONOMY_WORDS LANG("eng"));
+    exec!(executor -> TRIGGER consolidate);
+
+    // NOTE: All of those match with the default configuration
+    //   (see `test_search_allows_typos`).
+    for needle in ["ssun", "nzbula", "plusars", "sapcecrzft"] {
+        let response = exec!(executor -> QUERY "messages" "user:1" needle LANG("eng"));
+        assert_eq!(response, [] as [&str; 0], "QUERY {needle:?}");
+
+        let suggestions = exec!(executor -> SUGGEST "messages" "user:1" needle LANG("eng"));
+        assert_eq!(suggestions, [] as [&str; 0], "SUGGEST {needle:?}");
+    }
+}
+
+/// Ensures the order of words in search queries is insignificant.
+#[test]
+#[ignore = "Not supported yet (FIXME)"]
+fn test_search_term_order_insignificant() {
+    #[rustfmt::skip]
+    test_ingest_then_query!(push: ASTRONOMY_WORDS [ensure_no_stopword], query: [
+        ("satellite pulsars nebula", true),
+        (&format!("missing {ASTRONOMY_WORDS}"), true),
+    ]);
+}
+
+#[test]
+#[ignore = "Not supported yet"]
+fn test_chinese_typo_correction() {
+    // We should make sure languages with shorter graphemes get typo correction
+    // too.
+    unimplemented!()
+}

docs/search-results-ranking.md

@@ -0,0 +1,145 @@
+# Sonic search results ranking
+
+In search engines like Sonic, it’s common for results to be ranked using
+[tf-idf] or its “evolution” [BM25]. However, to keep a very small index Sonic
+doesn’t store some information that’s necessary to compute [term frequency], a
+core element of both [tf-idf] and [BM25].
+
+Out of simplicity, before `v1.7.0` Sonic used to return results that contained
+all search terms (with some fuzzy matching), sorted by reverse ingestion order
+(i.e. most recent documents first). While this was good enough for some use
+cases, discrepancies in ingestion and query languages would silently affect
+which words were considered stopwords and prevent some documents[^doc] from
+showing up in results (because of the implicit `AND`)[^1].
+
+[^doc]: Since this document is about information retrieval as a general concept, this document will use the term “document” to refer to Sonic objects as it’s the strandard term.
+[^1]: For example, see [issue #262](https://github.com/valeriansaliou/sonic/issues/262).
+
+In `v1.7.0`, we reworked Sonic’s ranking algorithm to yield better results.
+
+## Ranking algorithm
+
+For each search term, Sonic looks for documents containing exactly that search
+term, then it performs fuzzy matching until enough results are found. For each
+term, Sonic keeps track of the [Levenshtein distance] of the closest word found
+in each document. Once that done, a documents’s score is calculated by summing
+up its scores for each term, and documents are finally sorted by score. When
+ties occur, results are sorted by reverse ingestion order[^rio].
+
+[^rio]: This happens at no cost and is invisible in the code, but happens because it’s the order in which results are yielded by the index (on purpose).
+
+Here is some more detailed and accurate pseudo-code:
+
+```txt
+MISSING_MATCH_SCORE = 100
+
+fn query(query, app_config) -> [doc] {
+  limit = app_config.store.kv.retain_word_objects
+  mut alternates_try = app_config.search.query_alternates_try
+
+  // Updates the document’s score for a given term.
+  // Returns whether the document was inserted or not.
+  fn update_score(mut results_matrix, doc, score, term_idx, term_count) -> bool {
+    if results_matrix contains doc:
+      // Update score to keep minimum between existing and new score.
+      results_matrix[doc][term_idx] = min(results_matrix[doc][term_idx], score)
+      return false
+    else:
+      // By default, assign a high score to each term.
+      mut scores = [MISSING_MATCH_SCORE; term_count]
+      scores[term_idx] = score
+      results_matrix[doc] = scores
+      return true
+  }
+
+  mut results_matrix = []
+
+  // Look for exact matches.
+  for (index, term) in normalized(query):
+    for doc in find_exact_matches(term):
+      if len(results_matrix) < limit:
+        score = 0
+        update_score(results_matrix, doc, score, index, len(normalized(query)))
+
+  // Look for words containing `term` as prefix.
+  if len(results_matrix) < limit && alternates_try > 0:
+    for (index, term) in normalized(query):
+      for suggested_term in lookup_begins(term):
+        for doc in limit(find_exact_matches(term), alternates_try):
+          if len(results_matrix) < limit:
+            // Compute Levenshtein distance.
+            score = len(suggested_term) - len(term)
+            if update_score(results_matrix, doc, score, index, len(normalized(query))):
+              alternates_try -= 1
+
+  // Thresholds are configured by `search.typo_factor_word_lengths`
+  // (shown with the default value: `[4, 7, 10]`).
+  fn typo_factor(word_len) -> u32 {
+    if      word_len <= 3: return 0
+    else if word_len <= 6: return 1
+    else if word_len <= 9: return 2
+    else                 : return 3
+  }
+
+  // Look for words like `term` (fuzzy matching).
+  if len(results_matrix) < limit && alternates_try > 0:
+    for (index, term) in normalized(query):
+      max_typo_factor = typo_factor(len(term))
+      mut typo_factor = 1
+
+      // Iterate on increasingly larger typo factors (Levenshtein distances).
+      while len(results_matrix) < limit && typo_factor <= max_typo_factor:
+        for (suggested_term, lev_distance) in lookup_typos(term, typo_factor):
+          for doc in limit(find_exact_matches(term), alternates_try):
+            if len(results_matrix) < limit:
+              // Save Levenshtein distance as score.
+              if update_score(results_matrix, doc, lev_distance, index, len(normalized(query))):
+                alternates_try -= 1
+        typo_factor += 1
+
+  // Computes final score for each document, summing individual term scores.
+  // NOTE: Takes missing matches into account thanks to `MISSING_MATCH_SCORE`.
+  fn flatten_matrix(results_matrix) -> [(doc, score)] {
+    mut list = []
+
+    for (doc, scores) in results_matrix:
+      list += (doc, sum(scores))
+
+    return list
+  }
+
+  // Creates a B-Tree of `[doc]` keyed by score, then flattens it.
+  fn sort_by_score(results_map) -> [doc] {
+    btree = btree_by_score(results_map)
+    return flatten(btree)
+  }
+
+  return sort_by_score(flatten_matrix(results_matrix))
+}
+```
+
+### Algorithm properties
+
+- At most `store.kv.retain_word_objects` are returned.
+- Fuzzy matching yields at most `search.query_alternates_try` results.
+- Memory usage is linear, proportional to the number of results.
+- Although the pseudo-code doesn’t show it, most operations are lazy, meaning
+  only useful computations are executed.
+- Scoring is calculated using integer values only, without a single floating
+  point computation, division, square root or any other expensive operation.
+
+### Possible future improvements
+
+- Assign a weight to each term, based on [inverse document frequency], to give
+  more precedence to rare terms.
+  - This would imply making some floating point operations, but they’d only be
+    `O(len(results))`, with idf computation being `O(1)`.
+  - This could be done in parallel from lookups, as it is independent.
+- Parallelize some operations (e.g. each term scoring)?
+  - Overhead might be higher than gain, we’d have to benchmark.
+
+[tf-idf]: https://en.wikipedia.org/wiki/Tf%E2%80%93idf
+[BM25]: https://en.wikipedia.org/wiki/Okapi_BM25 "“Okapi BM25” on Wikipedia"
+[term frequency]: https://en.wikipedia.org/wiki/Tf%E2%80%93idf#Term_frequency "“Term frequency” in “tf-idf” on Wikipedia"
+[inverse document frequency]: https://en.wikipedia.org/wiki/Tf%E2%80%93idf#Inverse_document_frequency "“Inverse document frequency” in “tf-idf” on Wikipedia"
+[Levenshtein distance]: https://en.wikipedia.org/wiki/Levenshtein_distance "“Levenshtein distance” on Wikipedia"

server/src/config.rs

@@ -0,0 +1,326 @@
+// Sonic
+//
+// Fast, lightweight and schema-less search backend
+// Copyright: 2019, Valerian Saliou <valerian@valeriansaliou.name>
+// Copyright: 2026, Rémi Bardon <remi@remibardon.name>
+// License: Mozilla Public License v2.0 (MPL v2.0)
+
+use std::sync::Arc;
+
+use crate::DEFAULT_CONFIG_FILE_PATHS;
+
+pub fn defaults_toml() -> &'static str {
+    r#"
+    [server]
+    log_level = "warn"
+
+    [channel]
+    inet = "[::1]:1491"
+    tcp_timeout = 300
+
+    [normalization]
+    # TODO(major): Enable `"nfkc"` by default.
+    unicode_normalization = "none"
+    # TODO(major): Enable by default.
+    diacritic_folding_enabled = false
+    stemming_enabled = false
+
+    [tokenization]
+    detect_special_patterns = true
+    # TODO(major): Disable by default.
+    compat_split_special_patterns = true
+
+    [stopwords]
+    allow = []
+    deny = []
+
+    [search]
+    query_limit_default = 10
+    query_limit_maximum = 100
+    query_alternates_try = 4
+    query_minimum_term_idf_default = 0.1
+    query_minimum_term_idf_minimum_object_count = 100
+    suggest_limit_default = 5
+    suggest_limit_maximum = 20
+    list_limit_default = 100
+    list_limit_maximum = 500
+    typo_factor_word_lengths = [4, 7, 10]
+
+    [store.kv]
+    path = "./data/store/kv/"
+    retain_word_objects = 1000
+    pool.inactive_after = 1800
+    database.flush_after = 900
+    database.compression_type = "zstd"
+    database.parallelism = 2
+    # database.write_buffer_size = 16384 # Default handled via serde
+    database.write_ahead_log = true
+
+    [store.fst]
+    path = "./data/store/fst/"
+    pool.inactive_after = 300
+    graph.consolidate_after = 180
+    graph.max_size = 2048
+    graph.max_words = 250000
+    "#
+}
+
+impl Config {
+    pub fn parse(source: impl ::config::Source + Send + Sync + 'static) -> Self {
+        // Read configuration.
+        let raw_config: config::Config = config::Config::builder()
+            // Start from defaults.
+            .add_source(config::File::from_str(
+                defaults_toml(),
+                config::FileFormat::Toml,
+            ))
+            // Merge static configuration (from file).
+            .add_source(source)
+            // Merge environment overrides.
+            .add_source(
+                config::Environment::with_prefix("SONIC")
+                    .separator("__")
+                    .prefix_separator("_"),
+            )
+            .build()
+            .expect("error reading config");
+
+        // `#[serde(flatten)]` breaks type coercion from the `config` crate,
+        // so we can’t have `crate::Config` define
+        // `#[serde(flatten)] sonic: sonic::Config`. It’s a bit dirty but at
+        // least we don’t have to manually add custom deserialization logic to
+        // all non-string fields (in the core!).
+        #[derive(serde::Deserialize)]
+        pub struct ServerConfigTemp {
+            pub channel: ChannelConfig,
+            pub server: ServerConfig,
+        }
+
+        // Parse configuration.
+        let mut server_config = raw_config
+            .clone()
+            .try_deserialize::<ServerConfigTemp>()
+            .expect("syntax error in config");
+        let mut core_config = raw_config
+            .try_deserialize::<sonic::Config>()
+            .expect("syntax error in config");
+
+        back_compat::migrate_channel_search(&mut server_config.channel, &mut core_config);
+
+        // Validate configuration.
+        core_config.validate();
+
+        Config {
+            channel: server_config.channel,
+            server: server_config.server,
+            sonic: Arc::new(core_config),
+        }
+    }
+}
+
+pub fn read_config(custom_path: Option<&str>) -> Config {
+    if let Some(custom_path) = custom_path {
+        // Abort if the user specified a config path that does not exist.
+        if !std::path::Path::new(custom_path).exists() {
+            panic!("Cannot find config file at {custom_path:?}");
+        }
+
+        let path = std::path::absolute(custom_path).unwrap();
+        tracing::debug!("reading config file: {path:?}");
+
+        Config::parse(config::File::from(path).format(config::FileFormat::Toml))
+    } else {
+        for path in DEFAULT_CONFIG_FILE_PATHS {
+            let path = std::path::absolute(path).unwrap();
+            tracing::debug!("looking for config file at {path:?}");
+
+            if path.exists() {
+                tracing::debug!("reading config file: {path:?}");
+
+                return Config::parse(config::File::from(path).format(config::FileFormat::Toml));
+            }
+        }
+
+        tracing::debug!("no config file found in known locations, using defaults only");
+
+        Config::parse(config::File::from_str("", config::FileFormat::Toml))
+    }
+}
+
+pub struct Config {
+    pub channel: ChannelConfig,
+
+    pub server: ServerConfig,
+
+    pub sonic: Arc<sonic::Config>,
+}
+
+#[allow(deprecated)]
+#[derive(serde::Deserialize)]
+pub struct ChannelConfig {
+    #[serde(deserialize_with = "sonic::util::serde::env_var::socket_addr")]
+    pub inet: std::net::SocketAddr,
+
+    pub tcp_timeout: u64,
+
+    #[serde(default, deserialize_with = "sonic::util::serde::env_var::opt_str")]
+    pub auth_password: Option<String>,
+
+    #[deprecated(since = "1.6.0", note = "Use `search` instead of `channel.search`")]
+    #[serde(default)]
+    pub search: Option<back_compat::ChannelSearchConfig>,
+}
+
+#[derive(serde::Deserialize)]
+pub struct ServerConfig {
+    #[serde(deserialize_with = "sonic::util::serde::env_var::str")]
+    pub log_level: String,
+}
+
+#[allow(deprecated)]
+mod back_compat {
+    #[deprecated(since = "1.6.0", note = "Use `search` instead of `channel.search`")]
+    #[derive(serde::Deserialize)]
+    pub struct ChannelSearchConfig {
+        #[serde(default)]
+        pub query_limit_default: Option<u16>,
+
+        #[serde(default)]
+        pub query_limit_maximum: Option<u16>,
+
+        #[serde(default)]
+        pub query_alternates_try: Option<usize>,
+
+        #[serde(default)]
+        pub suggest_limit_default: Option<u16>,
+
+        #[serde(default)]
+        pub suggest_limit_maximum: Option<u16>,
+
+        #[serde(default)]
+        pub list_limit_default: Option<u16>,
+
+        #[serde(default)]
+        pub list_limit_maximum: Option<u16>,
+    }
+
+    // This is dirty, but AFAIK (@RemiBardon) the `config` crate doesn’t
+    // provide a better API and hopefully we won’t have to do this again.
+    pub fn migrate_channel_search(
+        channel: &mut crate::config::ChannelConfig,
+        sonic: &mut sonic::Config,
+    ) {
+        if let Some(search) = channel.search.take() {
+            tracing::warn!(
+                "You’re still using the deprecated `channel.search` key. \
+                Please use `search` instead. \
+                For this run, we will override `search` with `channel.search`."
+            );
+
+            let ChannelSearchConfig {
+                query_limit_default,
+                query_limit_maximum,
+                query_alternates_try,
+                suggest_limit_default,
+                suggest_limit_maximum,
+                list_limit_default,
+                list_limit_maximum,
+            } = search;
+
+            if let Some(query_limit_default) = query_limit_default {
+                sonic.search.query_limit_default = query_limit_default;
+            }
+            if let Some(query_limit_maximum) = query_limit_maximum {
+                sonic.search.query_limit_maximum = query_limit_maximum;
+            }
+            if let Some(query_alternates_try) = query_alternates_try {
+                sonic.search.query_alternates_try = query_alternates_try;
+            }
+            if let Some(suggest_limit_default) = suggest_limit_default {
+                sonic.search.suggest_limit_default = suggest_limit_default;
+            }
+            if let Some(suggest_limit_maximum) = suggest_limit_maximum {
+                sonic.search.suggest_limit_maximum = suggest_limit_maximum;
+            }
+            if let Some(list_limit_default) = list_limit_default {
+                sonic.search.list_limit_default = list_limit_default;
+            }
+            if let Some(list_limit_maximum) = list_limit_maximum {
+                sonic.search.list_limit_maximum = list_limit_maximum;
+            }
+        }
+    }
+
+    #[cfg(test)]
+    #[test]
+    fn test_channel_search_migration() {
+        let old_config_toml = r#"
+        [channel]
+        search.query_limit_default = 42
+        search.query_alternates_try = 42
+        search.query_minimum_term_idf_default = 42
+        search.query_minimum_term_idf_minimum_object_count = 42
+        search.suggest_limit_default = 42
+        search.suggest_limit_maximum = 42
+        search.list_limit_default = 42
+        search.list_limit_maximum = 42
+        "#;
+
+        let config = crate::Config::parse(config::File::from_str(
+            old_config_toml,
+            config::FileFormat::Toml,
+        ));
+
+        assert!(
+            config.channel.search.is_none(),
+            "`channel.search` not emptied"
+        );
+        assert_eq!(
+            config.sonic.search.list_limit_default, 42,
+            "should be overriden"
+        );
+        assert_ne!(
+            config.sonic.search.query_limit_maximum, 42,
+            "should not be overriden"
+        );
+    }
+
+    #[cfg(test)]
+    #[test]
+    fn test_channel_search_migration_mixed() {
+        let old_config_toml = r#"
+        [channel.search]
+        query_limit_default = 42
+        # query_limit_maximum = 42
+        query_alternates_try = 42
+        query_minimum_term_idf_default = 42
+        query_minimum_term_idf_minimum_object_count = 42
+        suggest_limit_default = 42
+        suggest_limit_maximum = 42
+        list_limit_default = 42
+        list_limit_maximum = 42
+
+        [search]
+        query_limit_default = 8
+        query_limit_maximum = 8
+        "#;
+
+        let config = crate::Config::parse(config::File::from_str(
+            old_config_toml,
+            config::FileFormat::Toml,
+        ));
+
+        assert!(
+            config.channel.search.is_none(),
+            "`channel.search` not emptied"
+        );
+        assert_eq!(
+            config.sonic.search.list_limit_default, 42,
+            "should be overriden"
+        );
+        assert_eq!(
+            config.sonic.search.query_limit_maximum, 8,
+            "should not be overriden"
+        );
+    }
+}

server/tests/control_trigger.rs

@@ -0,0 +1,96 @@
+// Sonic
+//
+// Fast, lightweight and schema-less search backend
+// Copyright: 2026, Rémi Bardon <remi@remibardon.name>
+// License: Mozilla Public License v2.0 (MPL v2.0)
+
+mod common;
+
+use crate::common::prelude::*;
+
+#[test]
+fn trigger_consolidate() {
+    let ctx = start_empty(|command| {
+        command
+            .env("SONIC_STORE__FST__GRAPH__CONSOLIDATE_AFTER", "3600")
+            .env("SONIC_STORE__FST__POOL__INACTIVE_AFTER", "3700")
+            .env("SONIC_STORE__KV__DATABASE__FLUSH_AFTER", "3600")
+            .env("SONIC_STORE__KV__POOL__INACTIVE_AFTER", "3700")
+    });
+
+    let multiplexer = SonicMultiplexer::new().unwrap();
+
+    let ingest =
+        SonicChannelIngestBlocking::connect(ctx.addr, "SecretPassword", &multiplexer).unwrap();
+    let control =
+        SonicChannelControlBlocking::connect(ctx.addr, "SecretPassword", &multiplexer).unwrap();
+    let search =
+        SonicChannelSearchBlocking::connect(ctx.addr, "SecretPassword", &multiplexer).unwrap();
+
+    ingest
+        .push("collection", "bucket", "object", "foo bar")
+        .unwrap();
+
+    let terms = search.list("collection", "bucket").unwrap();
+    assert_eq!(terms.len(), 0);
+
+    () = control.trigger_consolidate().unwrap();
+
+    let terms = search.list("collection", "bucket").unwrap();
+    assert_eq!(terms.len(), 2);
+}
+
+#[test]
+fn trigger_backup() {
+    let ctx = start_empty(|command| {
+        command
+            .env("SONIC_STORE__FST__GRAPH__CONSOLIDATE_AFTER", "3600")
+            .env("SONIC_STORE__FST__POOL__INACTIVE_AFTER", "3700")
+            .env("SONIC_STORE__KV__DATABASE__FLUSH_AFTER", "3600")
+            .env("SONIC_STORE__KV__POOL__INACTIVE_AFTER", "3700")
+    });
+
+    let multiplexer = SonicMultiplexer::new().unwrap();
+
+    let ingest =
+        SonicChannelIngestBlocking::connect(ctx.addr, "SecretPassword", &multiplexer).unwrap();
+    let control =
+        SonicChannelControlBlocking::connect(ctx.addr, "SecretPassword", &multiplexer).unwrap();
+
+    let backup_path = std::path::Path::new(env!("CARGO_TARGET_TMPDIR"))
+        .join("test-backups")
+        .join(ctx.id.to_string());
+    if backup_path.exists() {
+        std::fs::remove_dir_all(&backup_path).unwrap();
+    }
+
+    // NOTE: Nothing was pushed yet, so the store directories do not exist.
+    () = control
+        .trigger_backup(backup_path.to_str().unwrap())
+        .unwrap();
+
+    // NOTE: Pushing opens the KV store, which then stays open in the pool
+    //   while the backup runs, and leaves the FST changes pending.
+    ingest
+        .push("collection", "bucket", "object", "foo bar")
+        .unwrap();
+
+    () = control
+        .trigger_backup(backup_path.to_str().unwrap())
+        .unwrap();
+
+    let kv_backups = std::fs::read_dir(backup_path.join("kv"))
+        .unwrap()
+        .map(|entry| entry.unwrap().path())
+        .collect::<Vec<_>>();
+    assert_eq!(kv_backups.len(), 1);
+    assert!(std::fs::read_dir(&kv_backups[0]).unwrap().next().is_some());
+
+    std::fs::remove_dir_all(&backup_path).unwrap();
+}
+
+#[test]
+#[ignore = "Not supported by sonic_client yet (FIXME)"]
+fn trigger_restore() {
+    todo!()
+}