feat(index): CreateIndex/DropIndex/ListIndexes, persisted and re-armed at boot

Completes v2.9.0 indexing: the storage layer from c4e0c72 is now reachable and
durable.

- Three RPCs plus Client::createIndex/dropIndex/listIndexes. Creating an index
  backfills it over existing rows, so it is usable immediately, and is
  idempotent. Dropping one is idempotent too.
- The declaration persists in CollectionCfg (_collection_meta), an ordinary
  collection and therefore WAL'd and snapshotted for free - the versioningEnabled
  precedent. DatabaseService::applyIndexDeclarations() re-arms it at boot.
  That step is load-bearing: without it a restart leaves an index recorded but
  unmaintained while the planner still consults it, and queries return stale
  rows with nothing logged. The e2e restarts the process and asserts both that
  the declaration survives AND that a row written afterwards is indexed.
- Ordering within each handler is deliberate. CreateIndex backfills BEFORE
  declaring, so the planner never sees a half-built index. DropIndex undeclares
  BEFORE dropping the sub-db, so no write can repopulate a half-dropped one.
- A masked field cannot be indexed and its index is not listed. An index over
  values a principal cannot read is an inference channel, the same reason
  filtering on a masked field is refused.

Fixed on the way: CollectionConfigManager::loadFromStore truncated at 100 - a
bare Query, and Query::limit defaults to 100. Third occurrence after ViewManager
and PolicyManager. This had to be fixed before index declarations could live
there, because a silently lost declaration is worse than no index at all.
Also index_stats().entries no longer counts the v2.4.4 identity sentinel, which
is a reserved key rather than a posting.

Client API is METHODS, never new struct members: all six public struct sizes
verified byte-identical against the installed 2.8.1 headers.

Test note: the first e2e run failed four assertions and all four were the
test's fault - upsert takes the id as a PARAMETER, and an "_id" inside the body
is ignored, so it was addressing rows that did not exist. The fifth failure was
real (the sentinel counted as a posting).

ctest 20/20, index e2e 25/25 with a real restart, namespacing 44/44, views +
policy + TLS/auth green.

VERSION

@@ -1 +1 @@
-2.8.1
+2.9.0

client/include/smartbotic/database/client.hpp

@@ -626,6 +626,46 @@ public:
      */
     bool dropView(const std::string& name);
 
+    /**
+     * Secondary index on one field of a collection. v2.9.0.
+     *
+     * Declaration is explicit because an index is not free: it costs write
+     * throughput, and on a low-cardinality field it is slower than a scan (on a
+     * real collection, status="completed" matched 66% of rows, where reading the
+     * index and then fetching two thirds of the rows by id loses to scanning).
+     * The server's planner therefore re-decides per query value and will decline
+     * an index that is not selective, so declaring one cannot make a query
+     * slower - but declaring the right ones is still up to the caller.
+     *
+     * Creating an index backfills it over the rows already present, so it is
+     * usable immediately. Idempotent. `rowsIndexed` is set to how many rows the
+     * backfill covered.
+     *
+     * Only equality filters are served from an index today. Ranges and CONTAINS
+     * still scan.
+     *
+     * These are METHODS rather than fields on a config struct on purpose: adding
+     * a member to a public struct changes its size, and with an unchanged soname
+     * an already-installed consumer binds to the new library with the old layout
+     * and crashes. That is not hypothetical here - it took the local webserver
+     * down in a restart loop once already.
+     */
+    bool createIndex(const std::string& collection, const std::string& field);
+    bool createIndex(const std::string& collection, const std::string& field,
+                     uint64_t& rowsIndexed);
+    bool dropIndex(const std::string& collection, const std::string& field);
+
+    /** One declared index, as reported by listIndexes(). */
+    struct IndexDefinition {
+        std::string field;
+        /// Distinct values held. Few distinct values over many rows means the
+        /// planner will usually decline this index.
+        uint64_t distinctValues = 0;
+        /// Total postings, i.e. indexed rows.
+        uint64_t entries = 0;
+    };
+    [[nodiscard]] std::vector<IndexDefinition> listIndexes(const std::string& collection);
+
     /**
      * List all views.
      */

client/src/client.cpp

@@ -1169,6 +1169,69 @@ public:
         return true;
     }
 
+    // v2.9.0 — secondary indexes.
+    bool createIndex(const std::string& collection, const std::string& field,
+                     uint64_t* rowsIndexed) {
+        smartbotic::databasepb::CreateIndexRequest request;
+        request.set_collection(qualify(collection));
+        request.set_field(field);
+        smartbotic::databasepb::CreateIndexResponse response;
+        grpc::ClientContext context;
+        setDeadline(context);
+
+        auto status = stub_->CreateIndex(&context, request, &response);
+        if (!status.ok()) {
+            spdlog::error("Client::createIndex failed: {}", status.error_message());
+            return false;
+        }
+        if (!response.success()) {
+            spdlog::error("Client::createIndex rejected: {}", response.error());
+            return false;
+        }
+        if (rowsIndexed != nullptr) *rowsIndexed = response.rows_indexed();
+        return true;
+    }
+
+    bool dropIndex(const std::string& collection, const std::string& field) {
+        smartbotic::databasepb::DropIndexRequest request;
+        request.set_collection(qualify(collection));
+        request.set_field(field);
+        smartbotic::databasepb::DropIndexResponse response;
+        grpc::ClientContext context;
+        setDeadline(context);
+
+        auto status = stub_->DropIndex(&context, request, &response);
+        if (!status.ok()) {
+            spdlog::error("Client::dropIndex failed: {}", status.error_message());
+            return false;
+        }
+        return response.success();
+    }
+
+    std::vector<Client::IndexDefinition> listIndexes(const std::string& collection) {
+        smartbotic::databasepb::ListIndexesRequest request;
+        request.set_collection(qualify(collection));
+        smartbotic::databasepb::ListIndexesResponse response;
+        grpc::ClientContext context;
+        setDeadline(context);
+
+        std::vector<Client::IndexDefinition> out;
+        auto status = stub_->ListIndexes(&context, request, &response);
+        if (!status.ok()) {
+            spdlog::error("Client::listIndexes failed: {}", status.error_message());
+            return out;
+        }
+        out.reserve(response.indexes_size());
+        for (const auto& i : response.indexes()) {
+            Client::IndexDefinition d;
+            d.field = i.field();
+            d.distinctValues = i.distinct_values();
+            d.entries = i.entries();
+            out.push_back(std::move(d));
+        }
+        return out;
+    }
+
     bool dropView(const std::string& name) {
         smartbotic::databasepb::DropViewRequest request;
         request.set_name(qualify(name));
@@ -2079,6 +2142,23 @@ bool Client::dropView(const std::string& name) {
     return impl_->dropView(name);
 }
 
+bool Client::createIndex(const std::string& collection, const std::string& field) {
+    return impl_->createIndex(collection, field, nullptr);
+}
+
+bool Client::createIndex(const std::string& collection, const std::string& field,
+                         uint64_t& rowsIndexed) {
+    return impl_->createIndex(collection, field, &rowsIndexed);
+}
+
+bool Client::dropIndex(const std::string& collection, const std::string& field) {
+    return impl_->dropIndex(collection, field);
+}
+
+std::vector<Client::IndexDefinition> Client::listIndexes(const std::string& collection) {
+    return impl_->listIndexes(collection);
+}
+
 std::vector<Client::ViewDefinition> Client::listViews() {
     return impl_->listViews();
 }

docs/ROADMAP.md

@@ -1,6 +1,6 @@
 # Smartbotic Database - Status and Roadmap
 
-**Current version: 2.8.1** (see `VERSION`). Last reviewed: 2026-08-09.
+**Current version: 2.9.0** (see `VERSION`). Last reviewed: 2026-08-09.
 
 This is the single authoritative statement of what exists and what does not.
 If any other document in this repository disagrees with this one, this one is
@@ -53,7 +53,7 @@ Consequences of that substitution, which trip up readers:
 
 ## Shipped
 
-Every item below is in the installed product as of 2.8.0. `CLAUDE.md` has the
+Every item below is in the installed product as of 2.9.0. `CLAUDE.md` has the
 detail and the failure modes.
 
 - JSON document store: collections, version history, field-level encryption, TTL
@@ -67,6 +67,8 @@ detail and the failure modes.
 - Durable snapshots with tiered recovery and read-only lockout
 - Replication, events/subscribe, migrations, set operations
 - Paging fast path (no filter, no sort) and the two-pass filtered/sorted scan
+- Secondary indexes on declared fields, equality filters, with a measured
+  selectivity guard so declaring one cannot make a query slower
 
 ---
 
@@ -75,47 +77,21 @@ detail and the failure modes.
 Ordered by what a reader is most likely to need next. Nothing here has a
 committed date.
 
-### 1. Indexing (storage layer done, NOT yet reachable by clients)
-
-**What exists and works**, with tests, in `LmdbDocumentStore`:
-
-- `_idx_<collection>#<field>` sub-dbs, `MDB_DUPSORT`, key = order-preserving
-  encoded value, data = document id (`storage/secondary_index.{hpp,cpp}`).
-- Maintenance inside the **same write transaction** as the document, so an
-  aborted write cannot leave a stale index. Unindexed collections pay nothing.
-- `build_index` backfills existing rows, idempotently; `drop_index` removes one.
-- A **selectivity guard**: an index is used only while matches are under
-  `rows / 10`. Derived from measurement, not taste - a full `decode_document`
-  costs ~8.6x a yyjson parse of the same row, so an index stops paying above
-  ~11.6% selectivity. Declaring an index therefore cannot pessimise a query.
-- `index_plan_stats()` reports indexed vs full scans vs declined-as-unselective.
-
-Measured on a copy of the live `smartbotic-automation` `executions` collection
-(10,089 rows / 505 MB):
-
-| query | before | after |
-|---|---|---|
-| EQ on `workflowId` (1 match, 0.01%) | 372 ms | **<1 ms** (~1000x) |
-| EQ on `status="completed"` (6,665 matches, 66%) | 347 ms | 344 ms, index **declined** |
-
-Both returned identical results to the unindexed plan.
-
-**What is missing - it cannot be used yet:**
-
-- No RPC. No `CreateIndex` / `DropIndex` / `ListIndexes`.
-- No persistence of the declaration. `set_indexed_fields()` is in-memory, so
-  nothing survives a restart. The declaration belongs in the `_collection_meta`
-  system collection (`CollectionCfg`), which is an ordinary collection and so is
-  WAL'd and snapshotted for free - the same reasoning that put
-  `versioningEnabled` there.
-- No client API. It must be **methods** (`createIndex`/`dropIndex`/
-  `listIndexes`), never new members on a public struct, since changing a struct's
-  size under an unchanged soname crashes installed consumers.
-- Only `EQ` is served. Ranges need the numeric encoding unified first: integral
-  and non-integral numbers currently sit under different type tags, which is
-  correct for equality but means they order independently.
-- `CONTAINS` (array membership) would need a different index shape - one posting
-  per element rather than per value.
+### 1. Indexing - equality only (shipped v2.9.0)
+
+Equality filters are served from secondary indexes. What remains:
+
+- **Only `EQ`.** Ranges (`GT`/`GTE`/`LT`/`LTE`) still scan. The blocker is the key
+  encoding: integral and non-integral numbers sit under different type tags, which
+  is correct for equality but means they order independently, so a range spanning
+  both cannot walk the index in one pass. Unify the numeric encoding first.
+- **`CONTAINS` (array membership)** needs a different index shape - one posting per
+  element rather than per value.
+- **Multi-predicate intersection.** The planner picks the single most selective
+  indexed EQ and applies the rest as filters. Intersecting two posting lists would
+  help queries that are selective only in combination.
+- **No CLI surface.** `smartbotic-db-cli` has no index commands; management is via
+  the client API or RPC only.
 
 ### 2. `encode_document` still serialises with nlohmann
 

proto/database.proto

@@ -72,6 +72,14 @@ service DatabaseService {
 
     // Collection configuration
     rpc ConfigureCollection(ConfigureCollectionRequest) returns (ConfigureCollectionResponse);
+
+    // v2.9.0 — secondary indexes. Declaration is explicit per collection: an
+    // index costs write throughput, and indexing every field would build useless
+    // ones (a low-cardinality field like a status enum is slower through an
+    // index than a scan).
+    rpc CreateIndex(CreateIndexRequest) returns (CreateIndexResponse);
+    rpc DropIndex(DropIndexRequest) returns (DropIndexResponse);
+    rpc ListIndexes(ListIndexesRequest) returns (ListIndexesResponse);
     rpc GetCollectionConfig(GetCollectionConfigRequest) returns (GetCollectionConfigResponse);
     rpc MigrateCollectionTimestamps(MigrateCollectionTimestampsRequest) returns (MigrateCollectionTimestampsResponse);
 
@@ -910,6 +918,57 @@ message ConfigureCollectionResponse {
     string error = 2;
 }
 
+// v2.9.0 — secondary index management.
+message CreateIndexRequest {
+    // Project-qualified collection name, as every other collection-carrying
+    // request takes.
+    string collection = 1;
+    // Field to index. Dotted paths address nested fields ("nest.deep"), and the
+    // document metadata fields (_id, _created_at, _updated_at, _version) are
+    // addressable too.
+    string field = 2;
+}
+
+message CreateIndexResponse {
+    bool success = 1;
+    string error = 2;
+    // Rows indexed by the backfill. Creating an index over an existing
+    // collection populates it, so it is usable immediately rather than only for
+    // rows written afterwards.
+    uint64 rows_indexed = 3;
+    // True when the index already existed - creation is idempotent.
+    bool already_existed = 4;
+}
+
+message DropIndexRequest {
+    string collection = 1;
+    string field = 2;
+}
+
+message DropIndexResponse {
+    bool success = 1;
+    string error = 2;
+}
+
+message ListIndexesRequest {
+    string collection = 1;
+}
+
+message IndexInfo {
+    string field = 1;
+    // Distinct values held, and total postings. Both come from the index itself
+    // rather than a scan, so an operator can judge selectivity - a field whose
+    // postings are concentrated in few values will not be used by the planner.
+    uint64 distinct_values = 2;
+    uint64 entries = 3;
+}
+
+message ListIndexesResponse {
+    bool success = 1;
+    string error = 2;
+    repeated IndexInfo indexes = 3;
+}
+
 message GetCollectionConfigRequest {
     string collection = 1;
 }

service/src/config/collection_config_manager.cpp

@@ -12,7 +12,8 @@ namespace {
 nlohmann::json toJson(const CollectionCfg& cfg) {
     return {
         {"timestamp_precision", cfg.timestampPrecision},
-        {"versioning_enabled", cfg.versioningEnabled}
+        {"versioning_enabled", cfg.versioningEnabled},
+        {"indexed_fields", cfg.indexedFields}
     };
 }
 
@@ -23,6 +24,8 @@ CollectionCfg fromJson(const nlohmann::json& j) {
     // v2.4.5 — absent on records written before 2.4.5, which all had history
     // on, so the default must be true.
     c.versioningEnabled = j.value("versioning_enabled", true);
+    // Absent means no indexes, which is what every pre-v2.9.0 config record has.
+    c.indexedFields = j.value("indexed_fields", std::vector<std::string>{});
     return c;
 }
 
@@ -38,17 +41,44 @@ void CollectionConfigManager::loadFromStore() {
     CollectionOptions opts;
     store_.createCollection(SYSTEM_COLLECTION, opts);
 
-    // Load all config documents (docId == collection name)
-    Query q;
-    auto result = store_.find(SYSTEM_COLLECTION, q);
-    for (const auto& doc : result.documents) {
-        if (doc.id.empty()) continue;
-        cache_[doc.id] = fromJson(doc.data());
+    // Load ALL config documents (docId == collection name), paging explicitly.
+    //
+    // A bare `Query` here was a truncation bug: Query::limit defaults to 100, so
+    // on an installation with more than 100 collections every config past the
+    // hundredth silently did not exist and that collection reverted to defaults
+    // on restart - losing its timestamp precision, its versioning setting, and
+    // (since v2.9.0) its index declarations. A declared index that stops being
+    // maintained is worse than no index: queries keep using it while writes stop
+    // updating it, so it returns stale rows rather than an error.
+    //
+    // Third occurrence of this trap, after ViewManager and PolicyManager. Note
+    // limit=0 returns NOTHING rather than everything (find computes
+    // end = min(offset + limit, size)), so paging is the only correct form.
+    constexpr uint32_t kPage = 500;
+    uint32_t offset = 0;
+    while (true) {
+        Query q;
+        q.limit = kPage;
+        q.offset = offset;
+        auto result = store_.find(SYSTEM_COLLECTION, q);
+        if (result.documents.empty()) break;
+        for (const auto& doc : result.documents) {
+            if (doc.id.empty()) continue;
+            cache_[doc.id] = fromJson(doc.data());
+        }
+        if (result.documents.size() < kPage) break;
+        offset += kPage;
     }
     spdlog::info("CollectionConfigManager: loaded {} configs from {}",
                  cache_.size(), SYSTEM_COLLECTION);
 }
 
+std::unordered_map<std::string, CollectionCfg>
+CollectionConfigManager::allConfigs() const {
+    std::shared_lock<std::shared_mutex> lock(cacheMutex_);
+    return cache_;
+}
+
 bool CollectionConfigManager::setConfig(const std::string& collection,
                                          const CollectionCfg& cfg,
                                          std::string& errorOut) {

service/src/config/collection_config_manager.hpp

@@ -37,6 +37,18 @@ struct CollectionCfg {
     //
     // NOT the same as maxVersions == 0, which means "unlimited history".
     bool versioningEnabled = true;
+
+    // v2.9.0 — fields of this collection carrying a secondary index.
+    //
+    // Lives here for the same reason versioningEnabled does: `_collection_meta`
+    // is an ordinary collection, so it is WAL'd and snapshotted for free. Putting
+    // it in CollectionOptions would have needed a new WAL op to survive a restart
+    // between snapshots.
+    //
+    // The service applies this to each project's LmdbDocumentStore at boot via
+    // set_indexed_fields(). If that did not happen, writes would stop maintaining
+    // an index that queries still consult - stale rows returned as if current.
+    std::vector<std::string> indexedFields;
 };
 
 /**
@@ -77,6 +89,11 @@ public:
      */
     bool hasExplicitConfig(const std::string& collection) const;
 
+    // Every stored config, keyed by project-qualified collection name. The
+    // service needs this at boot to re-apply index declarations to each
+    // project's document store.
+    std::unordered_map<std::string, CollectionCfg> allConfigs() const;
+
 private:
     MemoryStore& store_;
     mutable std::shared_mutex cacheMutex_;

service/src/database_grpc_impl.cpp

@@ -3,6 +3,7 @@
 #include "project_addressing.hpp"
 #include "storage/cosine_simd.hpp"
 #include "storage/document_store.hpp"
+#include "storage/document_store_lmdb.hpp"
 #include "json_parse.hpp"
 #include "persistence/wal.hpp"
 #include "views/projection.hpp"
@@ -2713,6 +2714,200 @@ grpc::Status DatabaseReplicationGrpcImpl::GetNodeState(
 
 // ===== Collection Config Operations =====
 
+
+// -------------------------------------------------------------------------
+// v2.9.0 — secondary index management.
+//
+// Declaration is explicit rather than automatic. An index costs write
+// throughput, and indexing every field would build ones the planner never uses:
+// on the live `executions` collection status="completed" matches 66% of rows,
+// where reading an index and then fetching two thirds of the collection by id is
+// slower than scanning.
+//
+// The declaration is persisted in CollectionCfg (`_collection_meta`), which is an
+// ordinary collection and therefore WAL'd and snapshotted for free. It MUST be
+// durable: if a restart lost it, writes would stop maintaining an index that
+// queries still consult, and the index would return stale rows rather than an
+// error.
+// -------------------------------------------------------------------------
+
+grpc::Status DatabaseGrpcImpl::CreateIndex(
+    grpc::ServerContext* context,
+    const pb::CreateIndexRequest* request,
+    pb::CreateIndexResponse* response
+) {
+    // An index is derived from stored data and changes write behaviour, so it
+    // gates as a write - same reasoning as ConfigureCollection.
+    smartbotic::database::Decision dec;
+    if (auto st = gate(context, request->collection(),
+                       smartbotic::database::Access::Write, dec); !st.ok()) {
+        return st;
+    }
+    if (service_.isReadOnly()) {
+        response->set_success(false);
+        response->set_error("database is in read-only mode: " + service_.readOnlyReason());
+        return grpc::Status::OK;
+    }
+    if (request->field().empty()) {
+        response->set_success(false);
+        response->set_error("field is required");
+        return grpc::Status::OK;
+    }
+    // A masked field must not be indexable: the index would answer questions
+    // about values the principal is not allowed to see, which is the same
+    // inference channel that makes filtering on a masked field refused.
+    if (!dec.mask.empty() &&
+        std::find(dec.mask.begin(), dec.mask.end(), request->field()) != dec.mask.end()) {
+        response->set_success(false);
+        response->set_error("cannot index a masked field");
+        return grpc::Status::OK;
+    }
+
+    try {
+        const auto rc = smartbotic::database::resolveCollection(request->collection());
+        auto* ds = service_.docStore(rc.project);
+        if (ds == nullptr) {
+            response->set_success(false);
+            response->set_error("no storage for project '" + rc.project + "'");
+            return grpc::Status::OK;
+        }
+        auto* lmdb = dynamic_cast<smartbotic::db::storage::LmdbDocumentStore*>(ds);
+        if (lmdb == nullptr) {
+            response->set_success(false);
+            response->set_error("secondary indexes require the LMDB substrate");
+            return grpc::Status::OK;
+        }
+
+        CollectionCfg cfg = config_manager_.configFor(request->collection());
+        const bool existed =
+            std::find(cfg.indexedFields.begin(), cfg.indexedFields.end(),
+                      request->field()) != cfg.indexedFields.end();
+
+        // Backfill BEFORE declaring. Between the declaration and the backfill an
+        // index is incomplete, and the planner would happily serve a query from
+        // it and omit rows. Building first means it is only ever consulted once
+        // complete.
+        const uint64_t rows = lmdb->build_index(rc.collection, request->field());
+
+        if (!existed) {
+            cfg.indexedFields.push_back(request->field());
+            std::string err;
+            if (!config_manager_.setConfig(request->collection(), cfg, err)) {
+                response->set_success(false);
+                response->set_error("index built but could not be recorded: " + err);
+                return grpc::Status::OK;
+            }
+        }
+        lmdb->set_indexed_fields(rc.collection, cfg.indexedFields);
+
+        response->set_success(true);
+        response->set_rows_indexed(rows);
+        response->set_already_existed(existed);
+        spdlog::info("v2.9 index created coll={} field={} rows={} (already_existed={})",
+                     request->collection(), request->field(), rows, existed);
+        return grpc::Status::OK;
+    } catch (const std::invalid_argument& e) {
+        return grpc::Status(grpc::StatusCode::INVALID_ARGUMENT, e.what());
+    } catch (const std::exception& e) {
+        return grpc::Status(grpc::StatusCode::INTERNAL, e.what());
+    }
+}
+
+grpc::Status DatabaseGrpcImpl::DropIndex(
+    grpc::ServerContext* context,
+    const pb::DropIndexRequest* request,
+    pb::DropIndexResponse* response
+) {
+    smartbotic::database::Decision dec;
+    if (auto st = gate(context, request->collection(),
+                       smartbotic::database::Access::Write, dec); !st.ok()) {
+        return st;
+    }
+    if (service_.isReadOnly()) {
+        response->set_success(false);
+        response->set_error("database is in read-only mode: " + service_.readOnlyReason());
+        return grpc::Status::OK;
+    }
+    try {
+        const auto rc = smartbotic::database::resolveCollection(request->collection());
+        auto* ds = service_.docStore(rc.project);
+        auto* lmdb = dynamic_cast<smartbotic::db::storage::LmdbDocumentStore*>(ds);
+        if (lmdb == nullptr) {
+            response->set_success(false);
+            response->set_error("secondary indexes require the LMDB substrate");
+            return grpc::Status::OK;
+        }
+
+        // Undeclare FIRST. While the declaration stands, writes maintain the
+        // index; dropping the sub-db first would leave a window where writes
+        // recreate entries in a half-dropped index.
+        CollectionCfg cfg = config_manager_.configFor(request->collection());
+        auto it = std::find(cfg.indexedFields.begin(), cfg.indexedFields.end(),
+                            request->field());
+        if (it != cfg.indexedFields.end()) {
+            cfg.indexedFields.erase(it);
+            std::string err;
+            if (!config_manager_.setConfig(request->collection(), cfg, err)) {
+                response->set_success(false);
+                response->set_error(err);
+                return grpc::Status::OK;
+            }
+        }
+        lmdb->set_indexed_fields(rc.collection, cfg.indexedFields);
+        lmdb->drop_index(rc.collection, request->field());
+
+        // Idempotent: dropping an index that is not there is a success, the
+        // requested end state having been reached either way.
+        response->set_success(true);
+        return grpc::Status::OK;
+    } catch (const std::invalid_argument& e) {
+        return grpc::Status(grpc::StatusCode::INVALID_ARGUMENT, e.what());
+    } catch (const std::exception& e) {
+        return grpc::Status(grpc::StatusCode::INTERNAL, e.what());
+    }
+}
+
+grpc::Status DatabaseGrpcImpl::ListIndexes(
+    grpc::ServerContext* context,
+    const pb::ListIndexesRequest* request,
+    pb::ListIndexesResponse* response
+) {
+    smartbotic::database::Decision dec;
+    if (auto st = gate(context, request->collection(),
+                       smartbotic::database::Access::Read, dec); !st.ok()) {
+        return st;
+    }
+    try {
+        const auto rc = smartbotic::database::resolveCollection(request->collection());
+        auto* ds = service_.docStore(rc.project);
+        auto* lmdb = dynamic_cast<smartbotic::db::storage::LmdbDocumentStore*>(ds);
+
+        const CollectionCfg cfg = config_manager_.configFor(request->collection());
+        for (const auto& f : cfg.indexedFields) {
+            // A masked field's index is not listed: its existence and cardinality
+            // are information about values the principal cannot read.
+            if (!dec.mask.empty() &&
+                std::find(dec.mask.begin(), dec.mask.end(), f) != dec.mask.end()) {
+                continue;
+            }
+            auto* info = response->add_indexes();
+            info->set_field(f);
+            if (lmdb != nullptr) {
+                if (auto st = lmdb->index_stats(rc.collection, f)) {
+                    info->set_distinct_values(st->distinct_values);
+                    info->set_entries(st->entries);
+                }
+            }
+        }
+        response->set_success(true);
+        return grpc::Status::OK;
+    } catch (const std::invalid_argument& e) {
+        return grpc::Status(grpc::StatusCode::INVALID_ARGUMENT, e.what());
+    } catch (const std::exception& e) {
+        return grpc::Status(grpc::StatusCode::INTERNAL, e.what());
+    }
+}
+
 grpc::Status DatabaseGrpcImpl::ConfigureCollection(
     grpc::ServerContext* context,
     const pb::ConfigureCollectionRequest* request,

service/src/database_grpc_impl.hpp

@@ -264,6 +264,25 @@ public:
 
     // ===== Collection Config Operations =====
 
+    // v2.9.0 — secondary index management.
+    grpc::Status CreateIndex(
+        grpc::ServerContext* context,
+        const pb::CreateIndexRequest* request,
+        pb::CreateIndexResponse* response
+    ) override;
+
+    grpc::Status DropIndex(
+        grpc::ServerContext* context,
+        const pb::DropIndexRequest* request,
+        pb::DropIndexResponse* response
+    ) override;
+
+    grpc::Status ListIndexes(
+        grpc::ServerContext* context,
+        const pb::ListIndexesRequest* request,
+        pb::ListIndexesResponse* response
+    ) override;
+
     grpc::Status ConfigureCollection(
         grpc::ServerContext* context,
         const pb::ConfigureCollectionRequest* request,

service/src/database_service.cpp

@@ -196,6 +196,15 @@ bool DatabaseService::initialize() {
         config_manager_->loadFromStore();
         policy_manager_->loadFromStore();
 
+        // v2.9.0 — re-apply persisted index declarations to each project's LMDB
+        // store. This is load-bearing, not bookkeeping: the declaration is what
+        // makes the write path maintain an index, and the planner consults an
+        // index purely on the declaration's word. If this step were skipped, a
+        // restart would leave indexes recorded but unmaintained, and queries
+        // would be served from a frozen index - stale rows returned as current,
+        // with nothing logged.
+        applyIndexDeclarations();
+
         // Run migrations if enabled
         if (config_.migrations.enabled && !config_.migrations.directory.empty()) {
             if (!runMigrations()) {
@@ -327,6 +336,33 @@ bool DatabaseService::runMigrations() {
     return migrationRunner_->runMigrations();
 }
 
+void DatabaseService::applyIndexDeclarations() {
+    size_t applied = 0;
+    for (const auto& [qualified, cfg] : config_manager_->allConfigs()) {
+        if (cfg.indexedFields.empty()) continue;
+        try {
+            const auto rc = resolveCollection(qualified);
+            auto* ds = docStore(rc.project);
+            auto* lmdb =
+                dynamic_cast<smartbotic::db::storage::LmdbDocumentStore*>(ds);
+            if (lmdb == nullptr) continue;
+            lmdb->set_indexed_fields(rc.collection, cfg.indexedFields);
+            ++applied;
+            spdlog::info("v2.9 index: {} field(s) active on {}",
+                         cfg.indexedFields.size(), qualified);
+        } catch (const std::exception& e) {
+            // Advisory per collection: one unparseable name must not stop the
+            // rest from being armed. Loud, because a missing declaration means
+            // that collection's index silently stops being maintained.
+            spdlog::error("v2.9 index: could not apply declarations for {}: {}",
+                          qualified, e.what());
+        }
+    }
+    if (applied > 0) {
+        spdlog::info("v2.9 index: applied declarations for {} collection(s)", applied);
+    }
+}
+
 void DatabaseService::auditSubdbPlacement() {
     if (!projects_) return;
 

service/src/database_service.hpp

@@ -318,6 +318,10 @@ private:
     // affected project and leaves a summary; never fails startup.
     void auditSubdbPlacement();
 
+    // v2.9.0 — arm the write path and query planner with the index declarations
+    // persisted in _collection_meta. Must run at boot, after config load.
+    void applyIndexDeclarations();
+
     // v2.3 Stage C — atomic rename of <dataDir>/env/ into
     // <dataDir>/projects/default/env/ when the v2.2 layout is detected
     // and the new layout doesn't yet exist. Idempotent. Refuses to start

service/src/storage/document_store_lmdb.cpp

@@ -680,6 +680,41 @@ void LmdbDocumentStore::reset_index_plan_stats() {
     declined_unselective_.store(0, std::memory_order_relaxed);
 }
 
+std::optional<LmdbDocumentStore::IndexStats>
+LmdbDocumentStore::index_stats(std::string_view collection,
+                                const std::string& field) {
+    ReadTxn rtxn(env_);
+    auto dbi_opt = try_open_for_read(rtxn, index_subdb_name(collection, field));
+    if (!dbi_opt) return std::nullopt;
+
+    IndexStats out;
+    MDB_stat st{};
+    mdb_check(mdb_stat(rtxn.raw(), *dbi_opt, &st), "stat (index)");
+    // On a DUPSORT sub-db ms_entries counts data items, i.e. every posting - plus
+    // the v2.4.4 identity sentinel, which is a reserved key and not a posting.
+    // Reporting it would overstate every index by exactly one, the same
+    // off-by-one count() already subtracts for document sub-dbs.
+    out.entries = st.ms_entries;
+    if (!read_subdb_identity(rtxn, *dbi_opt).empty() && out.entries > 0) {
+        --out.entries;
+    }
+
+    // Distinct keys need a walk: MDB_NEXT_NODUP skips a key's remaining
+    // duplicates, so this costs one step per distinct value, not per posting.
+    MDB_cursor* cur = nullptr;
+    mdb_check(mdb_cursor_open(rtxn.raw(), *dbi_opt, &cur), "cursor_open (index stat)");
+    struct G { MDB_cursor* c; ~G() { if (c) mdb_cursor_close(c); } } g{cur};
+    MDB_val k{0, nullptr};
+    MDB_val v{0, nullptr};
+    int rc = mdb_cursor_get(cur, &k, &v, MDB_FIRST);
+    while (rc == MDB_SUCCESS) {
+        if (!is_identity_key(to_sv(k))) ++out.distinct_values;
+        rc = mdb_cursor_get(cur, &k, &v, MDB_NEXT_NODUP);
+    }
+    if (rc != MDB_NOTFOUND) throw_mdb(rc, "cursor next_nodup (index stat)");
+    return out;
+}
+
 uint64_t LmdbDocumentStore::build_index(std::string_view collection,
                                          const std::string& field) {
     // Walk the collection once and index every row. Idempotent thanks to

service/src/storage/document_store_lmdb.hpp

@@ -85,6 +85,16 @@ public:
     IndexPlanStats index_plan_stats() const;
     void reset_index_plan_stats();
 
+    // Size of one index, read from the index itself rather than by scanning.
+    // distinct_values lets an operator judge selectivity: postings concentrated
+    // in few values mean the planner will usually decline the index.
+    struct IndexStats {
+        uint64_t distinct_values = 0;
+        uint64_t entries = 0;
+    };
+    std::optional<IndexStats> index_stats(std::string_view collection,
+                                          const std::string& field);
+
     // Populate an index over the rows already present. Returns rows indexed.
     uint64_t build_index(std::string_view collection, const std::string& field);
 

tests/load_test/test_indexes.cpp

@@ -0,0 +1,175 @@
+// v2.9.0 — secondary indexes end to end, over gRPC, through the real client.
+//
+// The unit tests cover key encoding and the storage layer. What only an e2e can
+// establish is the part that spans process boundaries:
+//
+//   1. A declaration PERSISTS across a restart. This is the load-bearing one. The
+//      declaration is what makes the write path maintain an index, and the planner
+//      trusts it. If a restart lost it, writes would stop updating an index that
+//      queries still consult and the answers would go stale silently.
+//   2. The client qualifies collection names, so an index declared by a client
+//      with a non-default project lands on that project's collection. Two
+//      namespacing bugs (v2.4.2 createView, v2.4.5 createCollection) both slipped
+//      through unit tests for exactly this reason.
+//   3. Queries return the same rows whether or not an index exists.
+//
+// Usage: test_indexes <address> <project> <phase>
+//   phase=setup   - create data + index, verify queries
+//   phase=verify  - AFTER a restart: the index must still be declared and used
+
+#include <smartbotic/database/client.hpp>
+
+#include <algorithm>
+#include <iostream>
+#include <string>
+#include <vector>
+
+using namespace smartbotic::database;
+
+namespace {
+
+int g_pass = 0;
+int g_fail = 0;
+
+void check(bool cond, const std::string& msg) {
+    if (cond) {
+        ++g_pass;
+        std::cout << "  ok   " << msg << "\n";
+    } else {
+        ++g_fail;
+        std::cout << "  FAIL " << msg << "\n";
+    }
+}
+
+constexpr const char* kColl = "idx_execs";
+constexpr int kRows = 300;
+
+std::vector<std::string> idsMatching(Client& c, const std::string& field,
+                                      const nlohmann::json& value) {
+    Client::QueryOptions o;
+    o.limit = 1000;
+    o.filters.push_back(Client::Filter::eq(field, value));
+    std::vector<std::string> ids;
+    for (const auto& d : c.find(kColl, o)) {
+        ids.push_back(d.value("_id", std::string{}));
+    }
+    std::sort(ids.begin(), ids.end());
+    return ids;
+}
+
+void setup(Client& c) {
+    std::cout << "-- setup --\n";
+    c.createCollection(kColl, {});
+
+    for (int i = 0; i < kRows; ++i) {
+        // NOTE: upsert takes the id as a PARAMETER. An "_id" inside the body is
+        // ignored and the server generates one instead - which silently made an
+        // earlier version of this test address rows that did not exist.
+        nlohmann::json d{
+            {"wf", "wf-" + std::to_string(i % 30)},      // 30 groups of 10 = 3.3%
+            {"state", (i % 2 == 0) ? "done" : "pending"},  // 50%, unselective
+        };
+        c.upsert(kColl, d, "r" + std::to_string(1000 + i));
+    }
+
+    // Results BEFORE any index exists - the reference answer.
+    const auto before = idsMatching(c, "wf", "wf-7");
+    check(before.size() == 10, "10 rows match wf-7 before indexing");
+
+    uint64_t rows = 0;
+    check(c.createIndex(kColl, "wf", rows), "createIndex succeeds");
+    check(rows == kRows, "the backfill covered every existing row (" +
+                          std::to_string(rows) + ")");
+
+    // Same question, now with an index available.
+    const auto after = idsMatching(c, "wf", "wf-7");
+    check(after == before,
+          "an indexed query returns EXACTLY the rows the unindexed one did");
+
+    // Idempotent.
+    check(c.createIndex(kColl, "wf"), "creating the same index again succeeds");
+    check(idsMatching(c, "wf", "wf-7") == before,
+          "and does not duplicate postings - still 10 rows");
+
+    auto listed = c.listIndexes(kColl);
+    check(listed.size() == 1, "listIndexes reports one index");
+    check(!listed.empty() && listed[0].field == "wf", "on the right field");
+    check(!listed.empty() && listed[0].entries == kRows,
+          "with a posting per row");
+    check(!listed.empty() && listed[0].distinctValues == 30,
+          "and 30 distinct values, which is how an operator judges selectivity");
+
+    // A write must keep the index in step: move a row to a new group.
+    c.upsert(kColl, nlohmann::json{{"wf", "wf-moved"}, {"state", "done"}}, "r1007");
+    check(idsMatching(c, "wf", "wf-7").size() == 9,
+          "updating an indexed field removes the row from its old group");
+    check(idsMatching(c, "wf", "wf-moved").size() == 1,
+          "and adds it to the new one");
+
+    // A delete must remove its postings.
+    check(c.remove(kColl, "r1037"), "deleted a row");
+    check(idsMatching(c, "wf", "wf-7").size() == 8,
+          "a deleted row leaves no posting behind");
+}
+
+void verify(Client& c) {
+    std::cout << "-- verify after restart --\n";
+
+    // THE test. The declaration lives in _collection_meta, and the service
+    // re-applies it at boot. If either half were missing, this list would be
+    // empty and the index would silently stop being maintained.
+    auto listed = c.listIndexes(kColl);
+    check(listed.size() == 1,
+          "the index declaration SURVIVED the restart - without this, writes stop "
+          "maintaining an index the planner still trusts");
+    check(!listed.empty() && listed[0].field == "wf", "on the same field");
+
+    // And it is still correct after the restart.
+    check(idsMatching(c, "wf", "wf-7").size() == 8,
+          "queries still return the right rows (8 after the earlier move+delete)");
+    check(idsMatching(c, "wf", "wf-moved").size() == 1, "and the moved row");
+
+    // A write made AFTER the restart must be indexed - proving maintenance is
+    // armed, not just that the old entries persisted.
+    c.upsert(kColl, nlohmann::json{{"wf", "wf-postboot"}, {"state", "done"}}, "r9999");
+    check(idsMatching(c, "wf", "wf-postboot").size() == 1,
+          "a row written after the restart IS indexed - the declaration is armed, "
+          "not merely remembered");
+
+    // The unselective field: results must be right whether or not the planner
+    // uses an index for it.
+    uint64_t n = 0;
+    check(c.createIndex(kColl, "state", n), "an index on a 50%-selective field is allowed");
+    const auto done = idsMatching(c, "state", "done");
+    check(done.size() > 100,
+          "and the query is still correct (" + std::to_string(done.size()) +
+          " rows) - the planner declines to USE it, which is not the same as "
+          "refusing to create it");
+
+    check(c.dropIndex(kColl, "state"), "dropIndex succeeds");
+    check(c.listIndexes(kColl).size() == 1, "and it is gone from the listing");
+    check(idsMatching(c, "state", "done") == done,
+          "dropping an index does not change any answer");
+
+    check(c.dropIndex(kColl, "nosuch"),
+          "dropping an index that does not exist is idempotent success");
+}
+
+}  // namespace
+
+int main(int argc, char** argv) {
+    if (argc < 4) {
+        std::cerr << "usage: test_indexes <address> <project> <setup|verify>\n";
+        return 2;
+    }
+    Client c({.address = argv[1], .project = argv[2]});
+    c.connect();
+
+    const std::string phase = argv[3];
+    if (phase == "setup") setup(c);
+    else if (phase == "verify") verify(c);
+    else { std::cerr << "unknown phase\n"; return 2; }
+
+    std::cout << "\npassed=" << g_pass << " failed=" << g_fail << "\n";
+    return g_fail == 0 ? 0 : 1;
+}

tests/load_test/test_indexes.sh

@@ -0,0 +1,94 @@
+#!/usr/bin/env bash
+# v2.9.0 — secondary indexes end to end, including a real restart.
+#
+# The restart is the point of this script. An index declaration lives in the
+# _collection_meta system collection and the service re-applies it at boot; if
+# either half were missing, writes would stop maintaining an index that the query
+# planner still consults, and queries would quietly return stale rows. Only
+# stopping and starting the process can show that.
+#
+# Usage: tests/load_test/test_indexes.sh [build_dir]
+
+set -uo pipefail
+
+BUILD="${1:-build}"
+REPO="$(cd "$(dirname "${BASH_SOURCE[0]}")/../.." && pwd)"
+PORT=19411
+PROJECT=idxproj
+WORK="$(mktemp -d)"
+DRIVER="$WORK/test_indexes"
+SERVER_PID=""
+
+cleanup() {
+    [[ -n "$SERVER_PID" ]] && kill "$SERVER_PID" 2>/dev/null
+    wait "$SERVER_PID" 2>/dev/null
+    rm -rf "$WORK"
+}
+trap cleanup EXIT
+
+fail() { echo "FAIL: $*" >&2; exit 1; }
+
+echo "=== building the driver ==="
+g++ -std=c++20 -O1 -o "$DRIVER" "$REPO/tests/load_test/test_indexes.cpp" \
+    -I"$REPO/client/include" -I"$REPO/$BUILD/client" \
+    -L"$REPO/$BUILD/client" -lsmartbotic-db-client -lspdlog -lfmt \
+    || fail "driver did not compile"
+
+mkdir -p "$WORK/data"
+cat > "$WORK/config.json" <<EOF
+{
+  "storage": {
+    "data_directory": "$WORK/data",
+    "bind_address": "127.0.0.1",
+    "rpc_port": $PORT,
+    "encryption": { "enabled": false, "key_file": "$WORK/data/storage.key" },
+    "memory": { "max_memory_mb": 512 }
+  },
+  "migrations": { "enabled": false }
+}
+EOF
+
+start_server() {
+    local log="$1"
+    "$REPO/$BUILD/service/smartbotic-database" --config "$WORK/config.json" > "$log" 2>&1 &
+    SERVER_PID=$!
+    for _ in $(seq 1 60); do
+        grep -q "READY" "$log" 2>/dev/null && return 0
+        kill -0 "$SERVER_PID" 2>/dev/null || { cat "$log"; fail "server exited during startup"; }
+        sleep 1
+    done
+    cat "$log"
+    fail "server never became ready"
+}
+
+stop_server() {
+    [[ -n "$SERVER_PID" ]] || return 0
+    kill "$SERVER_PID" 2>/dev/null
+    wait "$SERVER_PID" 2>/dev/null
+    SERVER_PID=""
+}
+
+export LD_LIBRARY_PATH="$REPO/$BUILD/client:${LD_LIBRARY_PATH:-}"
+
+echo "=== phase 1: create data and an index ==="
+start_server "$WORK/boot1.log"
+"$DRIVER" "127.0.0.1:$PORT" "$PROJECT" setup || fail "setup phase"
+
+echo
+echo "=== restarting the service ==="
+stop_server
+start_server "$WORK/boot2.log"
+
+# The service must SAY it re-armed the declarations. A silent boot here would
+# mean the index is recorded but unmaintained.
+grep -q "v2.9 index" "$WORK/boot2.log" \
+    || { grep -iE "index|error" "$WORK/boot2.log" | tail -20
+         fail "the service did not re-apply index declarations at boot"; }
+echo "  boot log: $(grep 'v2.9 index' "$WORK/boot2.log" | tail -1 | sed 's/.*\] //')"
+
+echo
+echo "=== phase 2: verify the index survived and is still maintained ==="
+"$DRIVER" "127.0.0.1:$PORT" "$PROJECT" verify || fail "verify phase"
+
+echo
+echo "ALL INDEX E2E CHECKS PASSED"