diff --git a/CMakeLists.txt b/CMakeLists.txt
index 63f462f61..c79af03e9 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -190,6 +190,7 @@ target_link_libraries(clice-core PUBLIC
     kota::codec::toml
     kota::option
     simdjson::simdjson
+    lmdb
 )
 
 file(GLOB CLICE_DRIVER_SOURCES CONFIGURE_DEPENDS "${PROJECT_SOURCE_DIR}/src/driver/*.cc")
diff --git a/cmake/package.cmake b/cmake/package.cmake
index 00f77d2fc..b301488ce 100644
--- a/cmake/package.cmake
+++ b/cmake/package.cmake
@@ -44,4 +44,27 @@ set(KOTA_CODEC_ENABLE_FLATBUFFERS ON)
 set(KOTA_ENABLE_EXCEPTIONS OFF)
 set(KOTA_ENABLE_RTTI OFF)
 
-FetchContent_MakeAvailable(kotatsu spdlog croaring)
+# lmdb — index blob database backend (index::BlobDatabase). Upstream ships
+# no CMake; the two-file static library is defined below. Pinned to the
+# 0.9 stable line.
+FetchContent_Declare(
+    lmdb
+    GIT_REPOSITORY https://github.com/LMDB/lmdb.git
+    GIT_TAG LMDB_0.9.31
+    GIT_SHALLOW TRUE
+)
+
+FetchContent_MakeAvailable(kotatsu spdlog croaring lmdb)
+
+add_library(lmdb STATIC
+    ${lmdb_SOURCE_DIR}/libraries/liblmdb/mdb.c
+    ${lmdb_SOURCE_DIR}/libraries/liblmdb/midl.c)
+target_include_directories(lmdb SYSTEM PUBLIC ${lmdb_SOURCE_DIR}/libraries/liblmdb)
+# Third-party C, not held to the project's warning set.
+if(MSVC)
+    target_compile_options(lmdb PRIVATE /w)
+else()
+    target_compile_options(lmdb PRIVATE -w)
+endif()
+find_package(Threads REQUIRED)
+target_link_libraries(lmdb PUBLIC Threads::Threads)
diff --git a/src/driver/index.cc b/src/driver/index.cc
index a06568658..870ecd173 100644
--- a/src/driver/index.cc
+++ b/src/driver/index.cc
@@ -104,7 +104,7 @@ kota::task<> run_indexing_task(MasterServer& server, std::string root, int& exit
     // or an unreadable global blob disabled persistence) the run would
     // only warm this process's memory and a rerun would start from
     // nothing — fail instead of pretending.
-    if(!server.workspace.index_storage) {
+    if(!server.workspace.index_db) {
         LOG_ERROR("Cannot persist the index at {}; see the log for the cause and rerun",
                   std::string_view(server.workspace.config.project.cache_dir));
         exit_code = 1;
@@ -211,7 +211,7 @@ int run_stats_once(llvm::StringRef root, std::uint32_t top, bool allow_retry) {
     Workspace workspace;
     workspace.config = std::move(config);
     workspace.store.emplace(std::move(*store));
-    workspace.index_storage = index::make_fs_index_storage(*workspace.store);
+    workspace.index_db = index::open_database(*workspace.store, workspace.config.project.index_db);
     // A namespace whose directory scan failed looks empty while its blobs
     // exist — reporting "Index is empty" with exit code 0 would be a lie.
     if(auto ec = workspace.store->scan_error()) {
@@ -231,7 +231,7 @@ int run_stats_once(llvm::StringRef root, std::uint32_t top, bool allow_retry) {
     }
     // load() detaches the storage when the global blob exists but cannot
     // be read — a transient IO error, not an empty index.
-    if(workspace.index_storage == nullptr) {
+    if(workspace.index_db == nullptr) {
         LOG_ERROR("Failed to read the index cache at {}; the cache was left untouched",
                   std::string_view(workspace.config.project.cache_dir));
         return 1;
diff --git a/src/index/database.cpp b/src/index/database.cpp
new file mode 100644
index 000000000..957461794
--- /dev/null
+++ b/src/index/database.cpp
@@ -0,0 +1,778 @@
+#include "index/database.h"
+
+#include <atomic>
+#include <cassert>
+#include <cstring>
+#include <type_traits>
+
+#ifdef __linux__
+#include <sys/vfs.h>
+#endif
+
+#include "lmdb.h"
+#include "support/cache_store.h"
+#include "support/filesystem.h"
+#include "support/logging.h"
+
+#include "llvm/ADT/SmallString.h"
+#include "llvm/Support/FileSystem.h"
+#include "llvm/Support/Process.h"
+#include "llvm/Support/raw_ostream.h"
+
+#ifdef _WIN32
+#include <io.h>
+#include <windows.h>
+#endif
+
+namespace clice::index {
+
+namespace {
+
+constexpr llvm::StringLiteral index_lock_name = "index.lock";
+
+/// Cross-process writer lock for the index lineage, shared by both
+/// backends and always taken before the backend touches any of its files.
+/// Atomic per-blob replacement (or a database file) cannot serialize the
+/// mutable global/manifest lineage: two writers (an LSP server plus a
+/// batch `clice index`) derive the same next generation from the same
+/// loaded state, so a manifest written by one passes the other's
+/// generation pin with FileVersion ids allocated against a different
+/// table, loading rows under the wrong files. An OS advisory lock dies
+/// with its process, so a crash leaves nothing stale behind.
+std::optional<int> acquire_writer_lock(CacheStore& store) {
+    auto lock_path = path::join(store.base_dir(), index_lock_name);
+    int lock_fd = -1;
+    if(auto ec = llvm::sys::fs::openFileForReadWrite(lock_path,
+                                                     lock_fd,
+                                                     llvm::sys::fs::CD_OpenAlways,
+                                                     llvm::sys::fs::OF_None)) {
+        LOG_WARN("Failed to open the index writer lock {}: {}", lock_path, ec.message());
+        return std::nullopt;
+    }
+    if(llvm::sys::fs::tryLockFile(lock_fd)) {
+        LOG_WARN(
+            "Another clice process is writing the index cache at {}; "
+            "index persistence is disabled for this process",
+            store.base_dir());
+        llvm::sys::Process::SafelyCloseFileDescriptor(lock_fd);
+        return std::nullopt;
+    }
+    return lock_fd;
+}
+
+void release_writer_lock(int lock_fd) {
+    if(lock_fd != -1) {
+        llvm::sys::fs::unlockFile(lock_fd);
+        llvm::sys::Process::SafelyCloseFileDescriptor(lock_fd);
+    }
+}
+
+// ── Filesystem backend ──────────────────────────────────────────────
+
+llvm::StringRef namespace_of(IndexBlobKind kind) {
+    switch(kind) {
+        case IndexBlobKind::Shard: return "index";
+        case IndexBlobKind::Manifest: return "index-manifest";
+        case IndexBlobKind::Global: return "index-global";
+        case IndexBlobKind::CDB: return "index-cdb";
+    }
+    std::unreachable();
+}
+
+class FsDatabase final : public BlobDatabase {
+public:
+    FsDatabase(CacheStore& store, int lock_fd) : store(store), lock_fd(lock_fd) {
+        for(auto kind: {IndexBlobKind::Shard,
+                        IndexBlobKind::Manifest,
+                        IndexBlobKind::Global,
+                        IndexBlobKind::CDB}) {
+            store.register_namespace({
+                .name = std::string(namespace_of(kind)),
+                .extension = ".idx",
+                .policy = CachePolicy::Persistent,
+            });
+        }
+    }
+
+    ~FsDatabase() override {
+        release_writer_lock(lock_fd);
+    }
+
+    ReadBlob read(IndexBlobKind kind, llvm::StringRef key) override {
+        auto path = store.lookup(namespace_of(kind), key);
+        if(!path) {
+            return {};
+        }
+        auto buffer = llvm::MemoryBuffer::getFile(*path);
+        if(!buffer) {
+            return {};
+        }
+        return {.buffer = std::move(*buffer)};
+    }
+
+    bool contains(IndexBlobKind kind, llvm::StringRef key) override {
+        return store.lookup(namespace_of(kind), key).has_value();
+    }
+
+    llvm::SmallVector<std::size_t> write(llvm::ArrayRef<Blob> puts,
+                                         llvm::ArrayRef<BlobKey> removes) override {
+        // A failed batch keeps its removals too, mirroring the LMDB
+        // backend's all-or-nothing commit: a removal landing without the
+        // puts it was batched behind can delete a blob the surviving
+        // on-disk state still references. load() re-sweeps whatever the
+        // skip leaves behind.
+        auto failed = write_puts(puts);
+        if(!failed.empty()) {
+            return failed;
+        }
+        for(auto& [kind, key]: removes) {
+            store.invalidate(namespace_of(kind), key);
+        }
+        return {};
+    }
+
+    void for_each_key(IndexBlobKind kind, llvm::function_ref<void(llvm::StringRef)> fn) override {
+        store.for_each_key(namespace_of(kind), fn);
+    }
+
+    std::expected<std::uint64_t, std::string> advance_read_snapshot() override {
+        return 0;
+    }
+
+    void retire_old_snapshot() override {}
+
+    std::expected<bool, std::string> grow() override {
+        return false;
+    }
+
+private:
+    llvm::SmallVector<std::size_t> write_puts(llvm::ArrayRef<Blob> puts) {
+        // Batch order encodes dependency (shards → manifests → global →
+        // CDB snapshot), so the first failure fails the rest of the batch:
+        // continuing would publish an entry whose prerequisites never
+        // landed — e.g. a CDB snapshot vouching for a global that failed —
+        // and load paths only tolerate a committed prefix, the crash shape.
+        auto fail_from = [&](std::size_t i) {
+            llvm::SmallVector<std::size_t> failed;
+            for(; i < puts.size(); i += 1) {
+                failed.push_back(i);
+            }
+            return failed;
+        };
+        for(std::size_t i = 0; i < puts.size(); i += 1) {
+            auto& blob = puts[i];
+            auto ns = namespace_of(blob.kind);
+            auto pending = store.begin_store(ns, blob.key);
+            std::error_code ec;
+            llvm::raw_fd_ostream os(pending.tmp_path, ec);
+            if(ec) {
+                LOG_WARN("Failed to write index blob {}/{}: {}", ns, blob.key, ec.message());
+                return fail_from(i);
+            }
+            os.write(blob.bytes.data(), blob.bytes.size());
+            os.close();
+            // A truncated blob (disk full) must never be committed: the
+            // namespaces are Persistent, so it would be served forever.
+            if(os.has_error()) {
+                LOG_WARN("Failed to write index blob {}/{}: {}",
+                         ns,
+                         blob.key,
+                         os.error().message());
+                os.clear_error();
+                return fail_from(i);
+            }
+            if(auto committed = store.commit(std::move(pending)); !committed) {
+                LOG_WARN("Failed to commit index blob {}/{}: {}",
+                         ns,
+                         blob.key,
+                         committed.error().message());
+                return fail_from(i);
+            }
+        }
+        return {};
+    }
+
+    CacheStore& store;
+    int lock_fd;
+};
+
+// ── LMDB backend ────────────────────────────────────────────────────
+
+constexpr llvm::StringLiteral lmdb_file_name = "index.mdb";
+
+/// Virtual reservation; pages materialize on use. The file is created
+/// sparse on Windows (real allocation otherwise); when that fails the
+/// caller starts small and relies on grow().
+constexpr std::size_t lmdb_default_mapsize = 64ull << 30;
+constexpr std::size_t lmdb_small_mapsize = 256ull << 20;
+
+char kind_prefix(IndexBlobKind kind) {
+    switch(kind) {
+        case IndexBlobKind::Shard: return 'S';
+        case IndexBlobKind::Manifest: return 'M';
+        case IndexBlobKind::Global: return 'G';
+        case IndexBlobKind::CDB: return 'C';
+    }
+    std::unreachable();
+}
+
+llvm::SmallString<64> encode_key(IndexBlobKind kind, llvm::StringRef key) {
+    llvm::SmallString<64> encoded;
+    encoded.push_back(kind_prefix(kind));
+    encoded.append(key);
+    return encoded;
+}
+
+/// The environment's self-description, stored under a key no kind prefix
+/// can produce. LMDB files are not portable across word sizes or byte
+/// orders and carry no schema of ours — any mismatch discards the whole
+/// database (it is a rebuildable cache).
+constexpr llvm::StringLiteral meta_key = "\xffmeta";
+constexpr std::uint32_t lmdb_schema_version = 1;
+
+struct MetaRecord {
+    std::uint32_t schema = lmdb_schema_version;
+    std::uint32_t word_bits = sizeof(std::size_t) * 8;
+    std::uint32_t byte_order = 0x01020304;
+};
+
+// The record is compared with memcmp; padding or a surprising layout
+// would poison every comparison.
+static_assert(sizeof(MetaRecord) == 12 && std::has_unique_object_representations_v<MetaRecord>);
+
+MDB_val to_val(llvm::StringRef bytes) {
+    return {bytes.size(), const_cast<char*>(bytes.data())};
+}
+
+// MDB_PAGE_NOTFOUND is not a key miss: a page the B-tree references is
+// absent from the file, which LMDB documents as corruption.
+bool is_corruption(int rc) {
+    return rc == MDB_CORRUPTED || rc == MDB_INVALID || rc == MDB_VERSION_MISMATCH ||
+           rc == MDB_PAGE_NOTFOUND;
+}
+
+void remove_database_files(llvm::StringRef path) {
+    llvm::sys::fs::remove(path);
+    llvm::sys::fs::remove(path + "-lock");
+}
+
+class LmdbDatabase final : public BlobDatabase {
+public:
+    LmdbDatabase(MDB_env* env, MDB_dbi dbi, MDB_txn* txn, std::string path, int lock_fd) :
+        env(env), dbi(dbi), txn(txn), path(std::move(path)), lock_fd(lock_fd) {}
+
+    ~LmdbDatabase() override {
+        retire_old_snapshot();
+        if(txn) {
+            mdb_txn_abort(txn);
+        }
+        mdb_env_close(env);
+        // Condemned = corruption observed at read time; deleting under the
+        // writer lock lets the next open start from an empty database.
+        if(condemned) {
+            remove_database_files(path);
+        }
+        release_writer_lock(lock_fd);
+    }
+
+    ReadBlob read(IndexBlobKind kind, llvm::StringRef key) override {
+        if(!txn) {
+            return {};
+        }
+        auto encoded = encode_key(kind, key);
+        auto mk = to_val(encoded);
+        MDB_val value;
+        if(int rc = mdb_get(txn, dbi, &mk, &value)) {
+            note_error(rc);
+            return {};
+        }
+        llvm::StringRef bytes(static_cast<const char*>(value.mv_data), value.mv_size);
+        // Index blobs need an 8-aligned base (their largest scalar is
+        // u64 and the flatbuffers verifier checks offsets relative to
+        // it). LMDB only guarantees that for overflow-page values; small
+        // inline values are copied into an owned, allocator-aligned
+        // buffer and become immortal (generation 0).
+        if(reinterpret_cast<std::uintptr_t>(value.mv_data) % 8 != 0) {
+            return {.buffer = llvm::MemoryBuffer::getMemBufferCopy(bytes)};
+        }
+        return {.buffer = llvm::MemoryBuffer::getMemBuffer(bytes,
+                                                           "",
+                                                           /*RequiresNullTerminator=*/false),
+                .generation = generation};
+    }
+
+    bool contains(IndexBlobKind kind, llvm::StringRef key) override {
+        if(!txn) {
+            return false;
+        }
+        auto encoded = encode_key(kind, key);
+        auto mk = to_val(encoded);
+        MDB_val value;
+        // Any error other than "not found" counts as present-but-unreadable;
+        // the loader reads corrupted() to decide rebuild vs touch-nothing.
+        int rc = mdb_get(txn, dbi, &mk, &value);
+        note_error(rc);
+        return rc != MDB_NOTFOUND;
+    }
+
+    llvm::SmallVector<std::size_t> write(llvm::ArrayRef<Blob> puts,
+                                         llvm::ArrayRef<BlobKey> removes) override {
+        auto fail_all = [&](int rc, llvm::StringRef stage) {
+            if(rc == MDB_MAP_FULL) {
+                map_full.store(true, std::memory_order_relaxed);
+            }
+            note_error(rc);
+            LOG_WARN("Index database write failed ({}): {}", stage, mdb_strerror(rc));
+            llvm::SmallVector<std::size_t> failed;
+            for(std::size_t i = 0; i < puts.size(); i += 1) {
+                failed.push_back(i);
+            }
+            return failed;
+        };
+        MDB_txn* wtxn = nullptr;
+        if(int rc = mdb_txn_begin(env, nullptr, 0, &wtxn)) {
+            return fail_all(rc, "begin");
+        }
+        for(auto& blob: puts) {
+            auto encoded = encode_key(blob.kind, blob.key);
+            auto mk = to_val(encoded);
+            auto mv = to_val(blob.bytes);
+            if(int rc = mdb_put(wtxn, dbi, &mk, &mv, 0)) {
+                mdb_txn_abort(wtxn);
+                return fail_all(rc, "put");
+            }
+        }
+        for(auto& [kind, key]: removes) {
+            auto encoded = encode_key(kind, key);
+            auto mk = to_val(encoded);
+            if(int rc = mdb_del(wtxn, dbi, &mk, nullptr); rc != 0 && rc != MDB_NOTFOUND) {
+                mdb_txn_abort(wtxn);
+                return fail_all(rc, "del");
+            }
+        }
+        if(int rc = mdb_txn_commit(wtxn)) {
+            return fail_all(rc, "commit");
+        }
+        return {};
+    }
+
+    void for_each_key(IndexBlobKind kind, llvm::function_ref<void(llvm::StringRef)> fn) override {
+        if(!txn) {
+            return;
+        }
+        MDB_cursor* cursor = nullptr;
+        if(int rc = mdb_cursor_open(txn, dbi, &cursor)) {
+            note_error(rc);
+            return;
+        }
+        char prefix = kind_prefix(kind);
+        MDB_val key{1, &prefix};
+        MDB_val value;
+        int rc = mdb_cursor_get(cursor, &key, &value, MDB_SET_RANGE);
+        while(rc == 0) {
+            llvm::StringRef bytes(static_cast<const char*>(key.mv_data), key.mv_size);
+            if(bytes.empty() || bytes[0] != prefix) {
+                break;
+            }
+            fn(bytes.drop_front());
+            rc = mdb_cursor_get(cursor, &key, &value, MDB_NEXT);
+        }
+        if(rc != 0 && rc != MDB_NOTFOUND) {
+            note_error(rc);
+        }
+        mdb_cursor_close(cursor);
+    }
+
+    std::expected<std::uint64_t, std::string> advance_read_snapshot() override {
+        MDB_txn* fresh = nullptr;
+        if(int rc = mdb_txn_begin(env, nullptr, MDB_RDONLY, &fresh)) {
+            note_error(rc);
+            return std::unexpected(std::string(mdb_strerror(rc)));
+        }
+        // No predecessor to retire when a grow() resized the map but
+        // failed to reopen a snapshot.
+        if(txn) {
+            outstanding.push_back(txn);
+        }
+        txn = fresh;
+        generation += 1;
+        return generation;
+    }
+
+    void retire_old_snapshot() override {
+        for(auto* old: outstanding) {
+            mdb_txn_abort(old);
+        }
+        outstanding.clear();
+    }
+
+    std::expected<bool, std::string> grow() override {
+        if(!map_full.load(std::memory_order_relaxed)) {
+            return false;
+        }
+        // set_mapsize demands no live transaction in this process, so
+        // every snapshot dies here; the caller owns rebinding all
+        // borrowers before the next suspension point.
+        retire_old_snapshot();
+        if(txn) {
+            mdb_txn_abort(txn);
+            txn = nullptr;
+        }
+        MDB_envinfo info;
+        mdb_env_info(env, &info);
+        auto grown = std::max<std::size_t>(info.me_mapsize * 2, lmdb_small_mapsize);
+        std::string error;
+        if(int rc = mdb_env_set_mapsize(env, grown)) {
+            error = mdb_strerror(rc);
+        } else {
+            map_full.store(false, std::memory_order_relaxed);
+        }
+        MDB_txn* fresh = nullptr;
+        if(int rc = mdb_txn_begin(env, nullptr, MDB_RDONLY, &fresh)) {
+            note_error(rc);
+            // No snapshot at all: reads degrade to "missing" until a
+            // later grow() succeeds. Persistence stays attached — closing
+            // the environment would invalidate every borrowed buffer.
+            return std::unexpected(error.empty() ? std::string(mdb_strerror(rc)) : error);
+        }
+        txn = fresh;
+        generation += 1;
+        if(!error.empty()) {
+            return std::unexpected(error);
+        }
+        return true;
+    }
+
+    bool corrupted() const override {
+        return poisoned.load(std::memory_order_relaxed);
+    }
+
+    void condemn() override {
+        condemned = true;
+    }
+
+private:
+    /// Latch corruption-family read errors for the loader's rebuild
+    /// decision; everything else stays "missing or transiently unreadable".
+    void note_error(int rc) {
+        if(is_corruption(rc)) {
+            poisoned.store(true, std::memory_order_relaxed);
+        }
+    }
+
+    MDB_env* env;
+    MDB_dbi dbi;
+    /// Current read snapshot; owned by the opening (event-loop) thread.
+    MDB_txn* txn;
+    /// Pre-advance snapshots kept alive while their borrowers migrate; a
+    /// migration cancelled mid-way leaves entries for the next one.
+    llvm::SmallVector<MDB_txn*, 2> outstanding;
+    std::uint64_t generation = 1;
+    std::string path;
+    /// Set by write() on the pool thread, consumed by grow() on the loop.
+    std::atomic<bool> map_full = false;
+    /// See note_error()/corrupted(); written on both loop and pool threads.
+    std::atomic<bool> poisoned = false;
+    bool condemned = false;
+    int lock_fd;
+};
+
+#ifdef _WIN32
+/// Without the sparse attribute Windows backs the whole mapsize with real
+/// disk (CreateFileMapping allocates eagerly), so the file is marked
+/// sparse before LMDB first maps it. Returns false when the volume does
+/// not support sparse files — the caller falls back to a small mapsize.
+bool make_sparse(llvm::StringRef path) {
+    int fd = -1;
+    if(llvm::sys::fs::openFileForReadWrite(path,
+                                           fd,
+                                           llvm::sys::fs::CD_OpenAlways,
+                                           llvm::sys::fs::OF_None)) {
+        return false;
+    }
+    auto handle = reinterpret_cast<HANDLE>(_get_osfhandle(fd));
+    DWORD returned = 0;
+    bool ok =
+        DeviceIoControl(handle, FSCTL_SET_SPARSE, nullptr, 0, nullptr, 0, &returned, nullptr) != 0;
+    llvm::sys::Process::SafelyCloseFileDescriptor(fd);
+    return ok;
+}
+#endif
+
+enum class MetaCheck : std::uint8_t {
+    Ok,
+    /// Foreign word size/endianness/schema, or data without any meta —
+    /// the corruption-recovery shape (delete and rebuild when writable).
+    Mismatch,
+    /// Could not validate right now; touch nothing.
+    Transient,
+};
+
+/// Validates the meta record under the resident transaction; initializes
+/// it first on a fresh (provably empty) writable database.
+MetaCheck check_meta(MDB_env* env, MDB_dbi dbi, MDB_txn* txn, bool read_only) {
+    auto mk = to_val(meta_key);
+    MDB_val value;
+    int rc = mdb_get(txn, dbi, &mk, &value);
+    MetaRecord expected;
+    if(rc == 0) {
+        bool ok = value.mv_size == sizeof(MetaRecord) &&
+                  std::memcmp(value.mv_data, &expected, sizeof(MetaRecord)) == 0;
+        return ok ? MetaCheck::Ok : MetaCheck::Mismatch;
+    }
+    if(rc != MDB_NOTFOUND) {
+        return is_corruption(rc) ? MetaCheck::Mismatch : MetaCheck::Transient;
+    }
+    // Meta missing: only a provably EMPTY database may be initialized in
+    // place — populated bytes without our meta are a foreign layout, and
+    // stamping them would bypass the schema gate.
+    MDB_stat stat;
+    if(mdb_stat(txn, dbi, &stat) != 0) {
+        return MetaCheck::Transient;
+    }
+    if(stat.ms_entries != 0) {
+        return MetaCheck::Mismatch;
+    }
+    if(read_only) {
+        return MetaCheck::Ok;
+    }
+    MDB_txn* wtxn = nullptr;
+    if(mdb_txn_begin(env, nullptr, 0, &wtxn) != 0) {
+        return MetaCheck::Transient;
+    }
+    MDB_val mv{sizeof(MetaRecord), &expected};
+    if(mdb_put(wtxn, dbi, &mk, &mv, 0) != 0) {
+        mdb_txn_abort(wtxn);
+        return MetaCheck::Transient;
+    }
+    return mdb_txn_commit(wtxn) == 0 ? MetaCheck::Ok : MetaCheck::Transient;
+}
+
+std::unique_ptr<LmdbDatabase> open_lmdb_env(CacheStore& store,
+                                            int lock_fd,
+                                            std::size_t initial_mapsize) {
+    auto path = path::join(store.base_dir(), lmdb_file_name);
+    bool read_only = store.read_only();
+
+    auto mapsize = initial_mapsize != 0 ? initial_mapsize : lmdb_default_mapsize;
+
+    // One recovery retry: confirmed corruption (or a meta mismatch) is
+    // repaired by deleting the database — it is a rebuildable cache, and
+    // the writer lock guarantees no other live process is using it.
+    // Transient errors (permissions, fd/memory pressure) must NOT delete
+    // anything: persistence is disabled for this session instead, the
+    // same discipline the loader applies to an unreadable global blob.
+    for(int attempt = 0; attempt < 2; attempt += 1) {
+        // Giving up must not leave behind a file this attempt created
+        // (make_sparse and mdb_env_open both create on demand): read-only
+        // opens select the LMDB backend on bare existence, so an abandoned
+        // uninitialized placeholder would shadow the per-file blobs.
+        bool created = !read_only && !llvm::sys::fs::exists(path);
+        auto discard_created = [&] {
+            if(created) {
+                remove_database_files(path);
+            }
+        };
+#ifdef _WIN32
+        if(!read_only && !make_sparse(path) && initial_mapsize == 0) {
+            LOG_WARN("Index database at {} cannot be sparse; starting at {} bytes and growing",
+                     path,
+                     lmdb_small_mapsize);
+            mapsize = lmdb_small_mapsize;
+        }
+#endif
+        MDB_env* env = nullptr;
+        if(int rc = mdb_env_create(&env)) {
+            LOG_WARN("Failed to create the index database environment: {}", mdb_strerror(rc));
+            discard_created();
+            return nullptr;
+        }
+        mdb_env_set_mapsize(env, mapsize);
+        unsigned flags = MDB_NOSUBDIR | MDB_NOTLS | (read_only ? MDB_RDONLY : 0);
+        // True = the failure was repaired by deleting the database and the
+        // loop should retry; false = give up with persistence disabled.
+        auto fail = [&](int rc, llvm::StringRef stage) {
+            mdb_env_close(env);
+            if(!read_only && is_corruption(rc) && attempt == 0) {
+                LOG_WARN("Index database at {} is corrupt ({} failed: {}); rebuilding",
+                         path,
+                         stage,
+                         mdb_strerror(rc));
+                remove_database_files(path);
+                return true;
+            }
+            LOG_WARN(
+                "Cannot open the index database at {} ({} failed: {}); "
+                "index persistence is disabled for this session",
+                path,
+                stage,
+                mdb_strerror(rc));
+            discard_created();
+            return false;
+        };
+
+        if(int rc = mdb_env_open(env, path.c_str(), flags, 0644)) {
+            if(fail(rc, "open")) {
+                continue;
+            }
+            return nullptr;
+        }
+        if(!read_only) {
+            int dead = 0;
+            mdb_reader_check(env, &dead);
+            if(dead != 0) {
+                LOG_INFO("Cleared {} stale index database readers", dead);
+            }
+        }
+        MDB_txn* txn = nullptr;
+        int rc = mdb_txn_begin(env, nullptr, MDB_RDONLY, &txn);
+        if(rc == MDB_MAP_RESIZED) {
+            // Another process grew the map between open and here; adopt
+            // its size (no transaction is active yet) and retry once.
+            mdb_env_set_mapsize(env, 0);
+            rc = mdb_txn_begin(env, nullptr, MDB_RDONLY, &txn);
+        }
+        if(rc != 0) {
+            if(fail(rc, "snapshot")) {
+                continue;
+            }
+            return nullptr;
+        }
+        MDB_dbi dbi;
+        if(int dbi_rc = mdb_dbi_open(txn, nullptr, 0, &dbi)) {
+            mdb_txn_abort(txn);
+            if(fail(dbi_rc, "dbi")) {
+                continue;
+            }
+            return nullptr;
+        }
+        switch(check_meta(env, dbi, txn, read_only)) {
+            case MetaCheck::Ok: break;
+            case MetaCheck::Mismatch: {
+                mdb_txn_abort(txn);
+                if(fail(MDB_INVALID, "meta")) {
+                    continue;
+                }
+                return nullptr;
+            }
+            case MetaCheck::Transient: {
+                mdb_txn_abort(txn);
+                mdb_env_close(env);
+                LOG_WARN(
+                    "Cannot validate the index database at {}; "
+                    "index persistence is disabled for this session",
+                    path);
+                discard_created();
+                return nullptr;
+            }
+        }
+        return std::make_unique<LmdbDatabase>(env, dbi, txn, std::move(path), lock_fd);
+    }
+    return nullptr;
+}
+
+enum class FsLocality : std::uint8_t {
+    Local,
+    Remote,
+    /// FUSE fronts anything from a local overlay to sshfs; LMDB may or
+    /// may not survive there — warn and respect the configuration.
+    Unknown,
+};
+
+FsLocality filesystem_locality(llvm::StringRef dir) {
+#ifdef __linux__
+    // llvm's is_local only knows NFS/SMB/CIFS on Linux; 9p (WSL drvfs
+    // mounts) and FUSE pass as local, and those are exactly the mounts
+    // LMDB is documented to break on.
+    struct statfs sfs;
+    if(statfs(std::string(dir).c_str(), &sfs) == 0) {
+        // Through uint32 first: f_type is a signed word, and the SMB2/CIFS
+        // magics have the top bit set.
+        switch(static_cast<std::uint32_t>(sfs.f_type)) {
+            case 0x6969:      // NFS
+            case 0x517B:      // SMB
+            case 0xFE534D42:  // SMB2
+            case 0xFF534D42:  // CIFS
+            case 0x01021997:  // 9p
+                return FsLocality::Remote;
+            case 0x65735546:  // FUSE
+                return FsLocality::Unknown;
+            default: break;
+        }
+    }
+#endif
+    bool local = true;
+    if(auto ec = llvm::sys::fs::is_local(dir, local)) {
+        // Undetermined counts as local: an over-eager fallback would
+        // silently fork the index lineage.
+        LOG_WARN("Cannot determine whether {} is a local filesystem ({}); assuming local",
+                 dir,
+                 ec.message());
+        return FsLocality::Local;
+    }
+    return local ? FsLocality::Local : FsLocality::Remote;
+}
+
+}  // namespace
+
+std::unique_ptr<BlobDatabase> open_fs_database(CacheStore& store) {
+    int lock_fd = -1;
+    if(!store.read_only()) {
+        auto locked = acquire_writer_lock(store);
+        if(!locked) {
+            return nullptr;
+        }
+        lock_fd = *locked;
+    }
+    return std::make_unique<FsDatabase>(store, lock_fd);
+}
+
+std::unique_ptr<BlobDatabase> open_lmdb_database(CacheStore& store, std::size_t initial_mapsize) {
+    int lock_fd = -1;
+    if(!store.read_only()) {
+        auto locked = acquire_writer_lock(store);
+        if(!locked) {
+            return nullptr;
+        }
+        lock_fd = *locked;
+    }
+    auto db = open_lmdb_env(store, lock_fd, initial_mapsize);
+    if(!db) {
+        release_writer_lock(lock_fd);
+    }
+    return db;
+}
+
+std::unique_ptr<BlobDatabase> open_database(CacheStore& store, llvm::StringRef backend) {
+    if(backend == "files") {
+        return open_fs_database(store);
+    }
+    if(backend != "lmdb") {
+        LOG_WARN("Unknown index_db backend '{}'; using lmdb", backend);
+    }
+    switch(filesystem_locality(store.base_dir())) {
+        case FsLocality::Local: break;
+        case FsLocality::Remote: {
+            LOG_WARN(
+                "{} is on a remote filesystem, which LMDB does not support; "
+                "using per-file index storage",
+                store.base_dir());
+            return open_fs_database(store);
+        }
+        case FsLocality::Unknown: {
+            LOG_WARN(
+                "{} is on a FUSE filesystem; LMDB needs local-filesystem semantics — "
+                "set index_db = \"files\" if the index database misbehaves",
+                store.base_dir());
+            break;
+        }
+    }
+    // A reader before any LMDB writer ever ran (or after "files" runs)
+    // reads whatever the per-file backend left behind — including nothing.
+    if(store.read_only() && !llvm::sys::fs::exists(path::join(store.base_dir(), lmdb_file_name))) {
+        return open_fs_database(store);
+    }
+    return open_lmdb_database(store);
+}
+
+}  // namespace clice::index
diff --git a/src/index/database.h b/src/index/database.h
new file mode 100644
index 000000000..6e54a750c
--- /dev/null
+++ b/src/index/database.h
@@ -0,0 +1,179 @@
+#pragma once
+
+#include <cstddef>
+#include <cstdint>
+#include <expected>
+#include <memory>
+#include <string>
+
+#include "llvm/ADT/ArrayRef.h"
+#include "llvm/ADT/STLFunctionalExtras.h"
+#include "llvm/ADT/SmallVector.h"
+#include "llvm/ADT/StringRef.h"
+#include "llvm/Support/MemoryBuffer.h"
+
+namespace clice {
+
+class CacheStore;
+
+}
+
+namespace clice::index {
+
+/// The blob families the index persists.
+enum class IndexBlobKind : std::uint8_t {
+    /// Per-file row blobs (ShardBlob), keyed by a path hash.
+    Shard,
+    /// Per-TU manifests, keyed by a path hash.
+    Manifest,
+    /// The single global blob (FileVersion table + symbols), key "global".
+    Global,
+    /// The single CDB command-hash snapshot the persisted index was built
+    /// against, key "cdb" — how a cold start detects compile-command
+    /// changes that happened while no server was running.
+    CDB,
+};
+
+/// One blob read out of the database. The bytes' lifetime is
+/// backend-defined, and `generation` says which contract applies:
+///
+/// - 0: the buffer owns its bytes (filesystem backend, and small LMDB
+///   values copied out for alignment) — valid for the buffer's lifetime.
+/// - nonzero: the bytes are borrowed from the backend's read snapshot of
+///   that generation and die when it is retired (advance_read_snapshot()
+///   followed by retire_old_snapshot(), or grow()). A caller keeping such
+///   bytes across a save must re-read them after every snapshot advance.
+struct ReadBlob {
+    std::unique_ptr<llvm::MemoryBuffer> buffer;
+    std::uint64_t generation = 0;
+
+    explicit operator bool() const {
+        return buffer != nullptr;
+    }
+};
+
+/// Key of one persisted blob, for batch removals.
+struct BlobKey {
+    IndexBlobKind kind;
+    std::string key;
+};
+
+/// Keyed blob database of the persisted index.
+///
+/// Boundary with CacheStore: CacheStore manages the versioned cache
+/// directory and file-shaped artifacts (PCH/PCM/header contexts — clang
+/// consumes those by path, so they must be real files); BlobDatabase is
+/// the index's keyed blob store living inside that directory. The
+/// filesystem backend materializes each blob as one CacheStore-namespace
+/// file; the LMDB backend keeps them all in a single `index.mdb`.
+///
+/// `write` does the heavy IO and belongs off the event loop; every other
+/// method is cheap and event-loop-only (read snapshots are owned by the
+/// opening thread).
+class BlobDatabase {
+public:
+    virtual ~BlobDatabase() = default;
+
+    /// The blob's bytes, or a null ReadBlob when missing or unreadable.
+    /// See ReadBlob for the lifetime contract.
+    virtual ReadBlob read(IndexBlobKind kind, llvm::StringRef key) = 0;
+
+    /// Whether a blob exists under the key, even when unreadable — how the
+    /// loader tells a missing global blob (sweep everything) from a
+    /// transient read failure (touch nothing).
+    virtual bool contains(IndexBlobKind kind, llvm::StringRef key) = 0;
+
+    struct Blob {
+        IndexBlobKind kind;
+        std::string key;
+        std::string bytes;
+    };
+
+    /// Persist `puts` in order, then delete `removes`. Atomicity is
+    /// backend-defined but never weaker than a committed prefix of `puts`:
+    /// batch order encodes dependency (shards → manifests → global → CDB
+    /// snapshot), load paths tolerate exactly a prefix, and the LMDB
+    /// backend commits everything or nothing in one transaction, removes
+    /// included. The failed put indices are returned so the caller can
+    /// re-dirty them; removes are best-effort (load sweeps stale blobs by
+    /// their generation pins). A full LMDB map fails the whole batch and
+    /// latches the grow request `grow()` serves.
+    virtual llvm::SmallVector<std::size_t> write(llvm::ArrayRef<Blob> puts,
+                                                 llvm::ArrayRef<BlobKey> removes) = 0;
+
+    virtual void for_each_key(IndexBlobKind kind, llvm::function_ref<void(llvm::StringRef)> fn) = 0;
+
+    /// Open a new read snapshot and return its generation; reads from now
+    /// on come from it. The previous snapshot — and every borrowed buffer
+    /// read out of it — stays valid until retire_old_snapshot(), so the
+    /// caller can migrate long-lived borrowers incrementally, yielding
+    /// between batches. On failure the current snapshot keeps serving and
+    /// nothing is invalidated. No-op on backends without snapshots
+    /// (filesystem): returns the current generation, 0.
+    virtual std::expected<std::uint64_t, std::string> advance_read_snapshot() = 0;
+
+    /// Retire every snapshot advance_read_snapshot() has replaced,
+    /// invalidating the buffers still borrowed from them. A migration
+    /// cancelled between advance and retire leaves snapshots outstanding;
+    /// the next migration rebinds every borrower onto the newest snapshot
+    /// and retires them all. No-op when nothing is pending.
+    virtual void retire_old_snapshot() = 0;
+
+    /// Resize the map after write() failed on a full map, retiring EVERY
+    /// read snapshot first — all borrowed buffers die immediately — and
+    /// opening a fresh one. Returns true when the map was resized: the
+    /// caller must then rebind every borrower before yielding the event
+    /// loop. False when no growth was pending (including the filesystem
+    /// backend, where this is a no-op). An error return also means growth
+    /// was attempted, so every snapshot has already been retired: borrowed
+    /// buffers are dead and must be shed or re-read, and a replacement
+    /// snapshot may not have opened — reads then miss until a later grow()
+    /// succeeds.
+    virtual std::expected<bool, std::string> grow() = 0;
+
+    /// Whether the backend observed page-level corruption after open
+    /// (reads failing with LMDB's corruption family). The loader uses it
+    /// to tell "rebuild me" apart from a transient failure; backends
+    /// without the concept never latch it.
+    virtual bool corrupted() const {
+        return false;
+    }
+
+    /// Mark the database for deletion when it closes — the recovery for
+    /// corruption observed at read time. The files go away under the
+    /// writer lock, so the next open starts from an empty database.
+    virtual void condemn() {}
+};
+
+/// Filesystem backend over the cache store; registers the index
+/// namespaces on construction. On a writable store this takes an exclusive
+/// cross-process writer lock for the index (held until destruction) and
+/// returns nullptr when another clice process already holds it — the
+/// global/manifest blobs form one mutable lineage that tolerates no second
+/// writer. Read-only stores skip the lock.
+std::unique_ptr<BlobDatabase> open_fs_database(CacheStore& store);
+
+/// LMDB backend: a single `index.mdb` (plus its `-lock` file) in the
+/// store's version directory. Takes the same writer lock as the
+/// filesystem backend before touching the environment. Returns nullptr
+/// when the lock is held elsewhere or the environment cannot be opened
+/// safely — only confirmed corruption (or a meta mismatch) is repaired by
+/// deleting and rebuilding the database; transient errors disable index
+/// persistence for the session and touch nothing. A read-only open uses
+/// MDB_RDONLY, which still registers a reader slot in the `-lock` file —
+/// the one deviation from the cache store's read-only-touches-nothing
+/// contract. `initial_mapsize` overrides the default virtual map
+/// reservation (tests exercise the growth path with a tiny map); 0 keeps
+/// the default.
+std::unique_ptr<BlobDatabase> open_lmdb_database(CacheStore& store,
+                                                 std::size_t initial_mapsize = 0);
+
+/// Backend selection: `backend` is the `project.index_db` config value
+/// ("lmdb" or "files"). Remote filesystems (which LMDB cannot run on) fall
+/// back to the filesystem backend with a warning; an LMDB environment that
+/// cannot be opened safely disables persistence (nullptr) rather than
+/// falling back — a second lineage of filesystem blobs next to a live
+/// index.mdb would split the index's history.
+std::unique_ptr<BlobDatabase> open_database(CacheStore& store, llvm::StringRef backend);
+
+}  // namespace clice::index
diff --git a/src/index/shard.cpp b/src/index/shard.cpp
index 5a39975a1..81a649e99 100644
--- a/src/index/shard.cpp
+++ b/src/index/shard.cpp
@@ -551,6 +551,14 @@ Shard Shard::from_bytes(llvm::StringRef data) {
     return from_buffer(llvm::MemoryBuffer::getMemBuffer(data, "", false));
 }
 
+bool Shard::rebind(std::unique_ptr<llvm::MemoryBuffer> replacement) {
+    if(!buffer || !replacement || replacement->getBufferSize() != buffer->getBufferSize()) {
+        return false;
+    }
+    buffer = std::move(replacement);
+    return true;
+}
+
 Shard Shard::from_buffer(std::unique_ptr<llvm::MemoryBuffer> buffer) {
     if(!buffer) {
         return {};
diff --git a/src/index/shard.h b/src/index/shard.h
index a37dc03da..d874cc160 100644
--- a/src/index/shard.h
+++ b/src/index/shard.h
@@ -45,6 +45,16 @@ class Shard {
     /// disk".
     static Shard from_buffer(std::unique_ptr<llvm::MemoryBuffer> buffer);
 
+    /// Swap the backing buffer for byte-identical storage (read-snapshot
+    /// migration after a save). Verification, the live mask and the
+    /// line-start cache all describe the bytes, not the address, so they
+    /// carry over. Byte identity is the caller's contract (the exclusive
+    /// writer lock keeps every key the batch left alone byte-identical);
+    /// only the size is checked, so migrating a resident shard never
+    /// faults its content pages in. Returns false — and keeps the current
+    /// buffer — when the replacement is missing or its size differs.
+    bool rebind(std::unique_ptr<llvm::MemoryBuffer> replacement);
+
     /// Whether this shard holds a blob.
     bool loaded() const {
         return buffer != nullptr;
diff --git a/src/index/storage.cpp b/src/index/storage.cpp
deleted file mode 100644
index 4264414ee..000000000
--- a/src/index/storage.cpp
+++ /dev/null
@@ -1,155 +0,0 @@
-#include "index/storage.h"
-
-#include "support/cache_store.h"
-#include "support/filesystem.h"
-#include "support/logging.h"
-
-#include "llvm/Support/Process.h"
-#include "llvm/Support/raw_ostream.h"
-
-namespace clice::index {
-
-namespace {
-
-constexpr llvm::StringLiteral index_lock_name = "index.lock";
-
-llvm::StringRef namespace_of(IndexBlobKind kind) {
-    switch(kind) {
-        case IndexBlobKind::Shard: return "index";
-        case IndexBlobKind::Manifest: return "index-manifest";
-        case IndexBlobKind::Global: return "index-global";
-        case IndexBlobKind::Cdb: return "index-cdb";
-    }
-    std::unreachable();
-}
-
-class FsIndexStorage final : public IndexStorage {
-public:
-    FsIndexStorage(CacheStore& store, int lock_fd) : store(store), lock_fd(lock_fd) {
-        for(auto kind: {IndexBlobKind::Shard,
-                        IndexBlobKind::Manifest,
-                        IndexBlobKind::Global,
-                        IndexBlobKind::Cdb}) {
-            store.register_namespace({
-                .name = std::string(namespace_of(kind)),
-                .extension = ".idx",
-                .policy = CachePolicy::Persistent,
-            });
-        }
-    }
-
-    ~FsIndexStorage() override {
-        if(lock_fd != -1) {
-            llvm::sys::fs::unlockFile(lock_fd);
-            llvm::sys::Process::SafelyCloseFileDescriptor(lock_fd);
-        }
-    }
-
-    std::unique_ptr<llvm::MemoryBuffer> read(IndexBlobKind kind, llvm::StringRef key) override {
-        auto path = store.lookup(namespace_of(kind), key);
-        if(!path) {
-            return nullptr;
-        }
-        auto buffer = llvm::MemoryBuffer::getFile(*path);
-        if(!buffer) {
-            return nullptr;
-        }
-        return std::move(*buffer);
-    }
-
-    bool contains(IndexBlobKind kind, llvm::StringRef key) override {
-        return store.lookup(namespace_of(kind), key).has_value();
-    }
-
-    llvm::SmallVector<std::size_t> write(llvm::ArrayRef<Blob> batch) override {
-        // Batch order encodes dependency (shards → manifests → global →
-        // CDB snapshot), so the first failure fails the rest of the batch:
-        // continuing would publish an entry whose prerequisites never
-        // landed — e.g. a CDB snapshot vouching for a global that failed —
-        // and load paths only tolerate a committed prefix, the crash shape.
-        auto fail_from = [&](std::size_t i) {
-            llvm::SmallVector<std::size_t> failed;
-            for(; i < batch.size(); i += 1) {
-                failed.push_back(i);
-            }
-            return failed;
-        };
-        for(std::size_t i = 0; i < batch.size(); i += 1) {
-            auto& blob = batch[i];
-            auto ns = namespace_of(blob.kind);
-            auto pending = store.begin_store(ns, blob.key);
-            std::error_code ec;
-            llvm::raw_fd_ostream os(pending.tmp_path, ec);
-            if(ec) {
-                LOG_WARN("Failed to write index blob {}/{}: {}", ns, blob.key, ec.message());
-                return fail_from(i);
-            }
-            os.write(blob.bytes.data(), blob.bytes.size());
-            os.close();
-            // A truncated blob (disk full) must never be committed: the
-            // namespaces are Persistent, so it would be served forever.
-            if(os.has_error()) {
-                LOG_WARN("Failed to write index blob {}/{}: {}",
-                         ns,
-                         blob.key,
-                         os.error().message());
-                os.clear_error();
-                return fail_from(i);
-            }
-            if(auto committed = store.commit(std::move(pending)); !committed) {
-                LOG_WARN("Failed to commit index blob {}/{}: {}",
-                         ns,
-                         blob.key,
-                         committed.error().message());
-                return fail_from(i);
-            }
-        }
-        return {};
-    }
-
-    void remove(IndexBlobKind kind, llvm::StringRef key) override {
-        store.invalidate(namespace_of(kind), key);
-    }
-
-    void for_each_key(IndexBlobKind kind, llvm::function_ref<void(llvm::StringRef)> fn) override {
-        store.for_each_key(namespace_of(kind), fn);
-    }
-
-private:
-    CacheStore& store;
-    int lock_fd;
-};
-
-}  // namespace
-
-std::unique_ptr<IndexStorage> make_fs_index_storage(CacheStore& store) {
-    int lock_fd = -1;
-    if(!store.read_only()) {
-        // Atomic per-blob replacement cannot serialize the mutable
-        // global/manifest lineage: two writers (an LSP server plus a batch
-        // `clice index`) derive the same next generation from the same
-        // loaded state, so a manifest written by one passes the other's
-        // generation pin with FileVersion ids allocated against a different
-        // table, loading rows under the wrong files. An OS advisory lock
-        // dies with its process, so a crash leaves nothing stale behind.
-        auto lock_path = path::join(store.base_dir(), index_lock_name);
-        if(auto ec = llvm::sys::fs::openFileForReadWrite(lock_path,
-                                                         lock_fd,
-                                                         llvm::sys::fs::CD_OpenAlways,
-                                                         llvm::sys::fs::OF_None)) {
-            LOG_WARN("Failed to open the index writer lock {}: {}", lock_path, ec.message());
-            return nullptr;
-        }
-        if(llvm::sys::fs::tryLockFile(lock_fd)) {
-            LOG_WARN(
-                "Another clice process is writing the index cache at {}; "
-                "index persistence is disabled for this process",
-                store.base_dir());
-            llvm::sys::Process::SafelyCloseFileDescriptor(lock_fd);
-            return nullptr;
-        }
-    }
-    return std::make_unique<FsIndexStorage>(store, lock_fd);
-}
-
-}  // namespace clice::index
diff --git a/src/index/storage.h b/src/index/storage.h
deleted file mode 100644
index 18ed83d22..000000000
--- a/src/index/storage.h
+++ /dev/null
@@ -1,81 +0,0 @@
-#pragma once
-
-#include <cstddef>
-#include <cstdint>
-#include <memory>
-#include <string>
-
-#include "llvm/ADT/ArrayRef.h"
-#include "llvm/ADT/STLFunctionalExtras.h"
-#include "llvm/ADT/SmallVector.h"
-#include "llvm/ADT/StringRef.h"
-#include "llvm/Support/MemoryBuffer.h"
-
-namespace clice {
-
-class CacheStore;
-
-}
-
-namespace clice::index {
-
-/// The blob families the index persists.
-enum class IndexBlobKind : std::uint8_t {
-    /// Per-file row blobs (ShardBlob), keyed by a path hash.
-    Shard,
-    /// Per-TU manifests, keyed by a path hash.
-    Manifest,
-    /// The single global blob (FileVersion table + symbols), key "global".
-    Global,
-    /// The single CDB command-hash snapshot the persisted index was built
-    /// against, key "cdb" — how a cold start detects compile-command
-    /// changes that happened while no server was running.
-    Cdb,
-};
-
-/// Storage backend for index blobs. The filesystem implementation below is
-/// the default; a database-backed one plugs in behind the same interface.
-///
-/// All methods are thread-safe. `write` does the heavy IO (fsync) and
-/// belongs off the event loop; reads are cheap.
-class IndexStorage {
-public:
-    virtual ~IndexStorage() = default;
-
-    /// The blob's bytes (memory-mapped where the backend allows), or
-    /// nullptr when missing or unreadable.
-    virtual std::unique_ptr<llvm::MemoryBuffer> read(IndexBlobKind kind, llvm::StringRef key) = 0;
-
-    /// Whether a blob exists under the key, even when unreadable — how the
-    /// loader tells a missing global blob (sweep everything) from a
-    /// transient read failure (touch nothing).
-    virtual bool contains(IndexBlobKind kind, llvm::StringRef key) = 0;
-
-    struct Blob {
-        IndexBlobKind kind;
-        std::string key;
-        std::string bytes;
-    };
-
-    /// Persist a batch in order. Atomicity is per blob, not per batch — a
-    /// crash can land a prefix, and every load path treats a committed
-    /// prefix as stale data to rebuild. A failed entry therefore fails the
-    /// rest of the batch too: what lands is always a prefix, never a
-    /// suffix without its prerequisites. The failed indices are logged and
-    /// returned so the caller can re-dirty them for a later save.
-    virtual llvm::SmallVector<std::size_t> write(llvm::ArrayRef<Blob> batch) = 0;
-
-    virtual void remove(IndexBlobKind kind, llvm::StringRef key) = 0;
-
-    virtual void for_each_key(IndexBlobKind kind, llvm::function_ref<void(llvm::StringRef)> fn) = 0;
-};
-
-/// Filesystem implementation over the cache store; registers the index
-/// namespaces on construction. On a writable store this takes an exclusive
-/// cross-process writer lock for the index (held until destruction) and
-/// returns nullptr when another clice process already holds it — the
-/// global/manifest blobs form one mutable lineage that tolerates no second
-/// writer. Read-only stores skip the lock.
-std::unique_ptr<IndexStorage> make_fs_index_storage(CacheStore& store);
-
-}  // namespace clice::index
diff --git a/src/server/compiler/indexer.cpp b/src/server/compiler/indexer.cpp
index 735488bc7..1cd63993c 100644
--- a/src/server/compiler/indexer.cpp
+++ b/src/server/compiler/indexer.cpp
@@ -7,9 +7,9 @@
 #include <string>
 #include <vector>
 
+#include "index/database.h"
 #include "index/manifest.h"
 #include "index/shard.h"
-#include "index/storage.h"
 #include "index/tu_index.h"
 #include "server/compiler/context_resolver.h"
 #include "server/protocol/worker.h"
@@ -38,20 +38,20 @@ std::string blob_key(llvm::StringRef path) {
     return std::format("{:016x}", llvm::xxh3_64bits(path));
 }
 
-/// JSON layout of the persisted CDB snapshot (blob kind Cdb): per source
+/// JSON layout of the persisted CDB snapshot (blob kind CDB): per source
 /// file, the sorted canonical command hashes of its entries and a hash of
 /// its matched config rules when the index state was last saved. A
 /// standalone-indexed header gets an entry too (empty hashes): its own
 /// matched rules plus the host source whose command its rows borrowed.
-struct CdbSnapshotEntry {
+struct CDBSnapshotEntry {
     std::string file;
     std::vector<std::string> hashes;
     std::string rules;
     std::string host;
 };
 
-struct CdbSnapshot {
-    std::vector<CdbSnapshotEntry> entries;
+struct CDBSnapshot {
+    std::vector<CDBSnapshotEntry> entries;
 };
 
 /// clice.toml append/remove rules change the effective indexing command
@@ -77,10 +77,10 @@ std::string rules_hash(const Config& config, llvm::StringRef file) {
     return std::format("{:016x}", llvm::xxh3_64bits(joined));
 }
 
-CdbSnapshot build_cdb_snapshot(Workspace& workspace,
+CDBSnapshot build_cdb_snapshot(Workspace& workspace,
                                const llvm::DenseMap<std::uint32_t, std::uint32_t>& header_hosts,
                                llvm::ArrayRef<std::uint32_t> standalone_debt) {
-    CdbSnapshot snapshot;
+    CDBSnapshot snapshot;
     for(auto& [path_id, hashes]: workspace.cdb.command_hash_snapshot()) {
         auto file = workspace.cdb.resolve_path(path_id).str();
         auto rules = rules_hash(workspace.config, file);
@@ -117,7 +117,7 @@ CdbSnapshot build_cdb_snapshot(Workspace& workspace,
         add_standalone(tu);
     }
     // Deterministic bytes: save() decides "unchanged" by byte equality.
-    std::ranges::sort(snapshot.entries, {}, &CdbSnapshotEntry::file);
+    std::ranges::sort(snapshot.entries, {}, &CDBSnapshotEntry::file);
     return snapshot;
 }
 
@@ -398,9 +398,9 @@ kota::task<> Indexer::save() {
     // nothing, and the gauge must not keep exposing the previous round's
     // count as current.
     saved_shards = 0;
-    if(!workspace.index_storage)
+    if(!workspace.index_db)
         co_return;
-    auto& storage = *workspace.index_storage;
+    auto& db = *workspace.index_db;
     auto& project = workspace.project_index;
     ScopedTimer timer;
 
@@ -435,23 +435,18 @@ kota::task<> Indexer::save() {
     for(auto path_id: retired) {
         workspace.shards.erase(path_id);
         dirty_shards.erase(path_id);
-        auto it = project.contributions.find(path_id);
-        if(it == project.contributions.end()) {
-            continue;
-        }
-        for(auto tu: llvm::make_first_range(it->second)) {
-            enqueue(tu, ReindexReason::ContentChanged);
-        }
+        requeue_owners(path_id);
     }
 
     // Snapshot the dirty state on the loop: everything below serializes
     // from copies, so merges landing across the write await simply re-dirty
     // for the next save. The id vectors parallel the batch so a failed
     // entry can be re-dirtied by its batch index.
-    std::vector<index::IndexStorage::Blob> batch;
+    std::vector<index::BlobDatabase::Blob> batch;
     llvm::SmallVector<std::uint32_t> shard_ids;
     llvm::SmallVector<std::uint32_t> manifest_ids;
-    llvm::SmallVector<std::pair<index::IndexBlobKind, std::string>> removals;
+    llvm::SmallVector<index::BlobKey> removals = std::move(startup_removes);
+    startup_removes.clear();
     for(auto path_id: retired) {
         removals.push_back(
             {index::IndexBlobKind::Shard, blob_key(workspace.path_pool.resolve(path_id))});
@@ -520,7 +515,7 @@ kota::task<> Indexer::save() {
         cdb_bytes = serialize_cdb_snapshot(workspace, header_hosts, standalone_debt());
         if(!cdb_bytes.empty() && cdb_bytes != persisted_cdb_snapshot) {
             cdb_index = batch.size();
-            batch.push_back({index::IndexBlobKind::Cdb, "cdb", cdb_bytes});
+            batch.push_back({index::IndexBlobKind::CDB, "cdb", cdb_bytes});
         } else {
             cdb_dirty = false;
         }
@@ -530,6 +525,21 @@ kota::task<> Indexer::save() {
     dirty_manifests.clear();
     global_dirty = false;
 
+    // A deferred load-time sweep can name a key this very save re-writes:
+    // the swept TU was re-enqueued at load and has already re-indexed.
+    // Puts run before removes inside write(), so the stale removal would
+    // delete the fresh blob — the put wins.
+    if(!removals.empty()) {
+        llvm::DenseSet<std::pair<unsigned, llvm::StringRef>> putting;
+        for(auto& put: batch) {
+            putting.insert({static_cast<unsigned>(put.kind), llvm::StringRef(put.key)});
+        }
+        llvm::erase_if(removals, [&](const index::BlobKey& remove) {
+            return putting.contains(
+                {static_cast<unsigned>(remove.kind), llvm::StringRef(remove.key)});
+        });
+    }
+
     if(batch.empty() && removals.empty()) {
         co_return;
     }
@@ -541,12 +551,7 @@ kota::task<> Indexer::save() {
     // commit is still running (and saved_shards still holds its reset).
     saving_shards = shard_count;
     llvm::SmallVector<std::size_t> failed;
-    co_await kota::queue([&] {
-        failed = storage.write(batch);
-        for(auto& [kind, key]: removals) {
-            storage.remove(kind, key);
-        }
-    });
+    co_await kota::queue([&] { failed = db.write(batch, removals); });
     // An entry the storage failed to commit is re-dirtied so a later save
     // retries it; discarded, the cache would trail the in-memory index
     // until an unrelated merge happens to dirty the same entry or a
@@ -575,6 +580,21 @@ kota::task<> Indexer::save() {
     saved_shards = shard_count - failed_shards;
     saving_shards = 0;
 
+    // Corruption can surface first at write time (a damaged page only the
+    // write's tree descent reaches): heal like load-time corruption instead
+    // of writing into the damaged environment every save. The batch's
+    // shards — dirty at batch build, hence memory-backed — re-dirty before
+    // the recovery's shed so they survive it and re-persist wholesale.
+    if(db.corrupted()) {
+        for(auto path_id: shard_ids) {
+            dirty_shards.insert(path_id);
+        }
+        recover_corrupt_database();
+        co_return;
+    }
+
+    co_await migrate_shard_views();
+
     LOG_PERF("index",
              "phase=save shards={} manifests={} total={} elapsed_ms={}",
              shard_count,
@@ -583,10 +603,145 @@ kota::task<> Indexer::save() {
              timer.ms());
 }
 
+kota::task<> Indexer::migrate_shard_views() {
+    if(!workspace.index_db) {
+        co_return;
+    }
+    auto& db = *workspace.index_db;
+
+    // A full-map write left nothing committed (everything is dirty again);
+    // growing retires every snapshot at once, so the rebind below must run
+    // to completion before the first yield.
+    auto grown = db.grow();
+    if(!grown) {
+        // Degenerate (address space exhausted): borrowed views may already
+        // be dead, so everything borrowed is shed rather than left
+        // dangling. The shed shards' persisted bytes stay intact but their
+        // manifests still read fresh — only the owner requeue rebuilds
+        // their resident rows this session.
+        LOG_ERROR("Index database growth failed: {}", grown.error());
+        if(db.corrupted()) {
+            recover_corrupt_database();
+        } else {
+            shed_borrowed_shards();
+        }
+        co_return;
+    }
+    bool grew = *grown;
+    if(!grew) {
+        // The LMDB backend hands out pointers into its resident read
+        // snapshot; after a commit the resident shards migrate onto a
+        // fresh snapshot so the old one can be retired. Both snapshots
+        // stay valid across the yields and the blobs are byte-identical,
+        // so queries between batches may observe a mix of old and new
+        // pointers with identical meaning. Shards whose write just failed
+        // are dirty again by now (their bytes never landed) and stay
+        // memory-backed until a later save.
+        auto advanced = db.advance_read_snapshot();
+        if(!advanced) {
+            LOG_WARN("Index read-snapshot advance failed: {}", advanced.error());
+            if(db.corrupted()) {
+                recover_corrupt_database();
+            }
+            co_return;
+        }
+        if(*advanced == 0) {
+            // Filesystem backend: buffers are immortal, nothing to migrate.
+            co_return;
+        }
+    }
+
+    constexpr std::size_t rebind_batch = 512;
+    llvm::SmallVector<std::uint32_t> resident;
+    for(auto path_id: llvm::make_first_range(workspace.shards)) {
+        if(!dirty_shards.contains(path_id)) {
+            resident.push_back(path_id);
+        }
+    }
+    for(std::size_t i = 0; i < resident.size(); i += 1) {
+        if(!grew && i != 0 && i % rebind_batch == 0) {
+            co_await kota::sleep(std::chrono::milliseconds(0), loop);
+        }
+        auto path_id = resident[i];
+        auto it = workspace.shards.find(path_id);
+        if(it == workspace.shards.end() || dirty_shards.contains(path_id)) {
+            continue;
+        }
+        auto blob =
+            db.read(index::IndexBlobKind::Shard, blob_key(workspace.path_pool.resolve(path_id)));
+        if(!blob || !it->second.rebind(std::move(blob.buffer))) {
+            // Corruption can also surface first here (a damaged page only
+            // this re-read reaches); the recovery below sheds the whole
+            // resident set, nothing per-shard to do.
+            if(db.corrupted()) {
+                break;
+            }
+            // Unreachable under the writer lock; the shard is dropped and
+            // its owners requeued to rebuild the rows, while keeping the
+            // old view would dangle once the snapshot retires.
+            LOG_ERROR("Index shard for {} diverged during snapshot migration",
+                      workspace.path_pool.resolve(path_id));
+            assert(false && "persisted shard must survive snapshot migration");
+            workspace.shards.erase(path_id);
+            requeue_owners(path_id);
+        }
+    }
+    // Corruption observed by any read since the write-time check — this
+    // loop's, or a query's during its yields — condemns the database; the
+    // recovery sheds every borrowed view before the environment closes.
+    if(db.corrupted()) {
+        recover_corrupt_database();
+        co_return;
+    }
+    db.retire_old_snapshot();
+}
+
+void Indexer::requeue_owners(std::uint32_t path_id) {
+    auto it = workspace.project_index.contributions.find(path_id);
+    if(it == workspace.project_index.contributions.end()) {
+        return;
+    }
+    for(auto tu: llvm::make_first_range(it->second)) {
+        enqueue(tu, ReindexReason::ContentChanged);
+    }
+}
+
+void Indexer::shed_borrowed_shards() {
+    llvm::SmallVector<std::uint32_t> shed;
+    for(auto path_id: llvm::make_first_range(workspace.shards)) {
+        if(!dirty_shards.contains(path_id)) {
+            shed.push_back(path_id);
+        }
+    }
+    for(auto path_id: shed) {
+        workspace.shards.erase(path_id);
+        requeue_owners(path_id);
+    }
+}
+
+void Indexer::recover_corrupt_database() {
+    LOG_WARN("Index database is corrupt; discarding it and rebuilding from scratch");
+    saved_shards = 0;
+    shed_borrowed_shards();
+    for(auto tu_path_id: llvm::make_first_range(workspace.project_index.manifests)) {
+        dirty_manifests.insert(tu_path_id);
+    }
+    global_dirty = true;
+    cdb_dirty = true;
+    persisted_cdb_snapshot.clear();
+    reopen_fresh_database();
+}
+
+void Indexer::reopen_fresh_database() {
+    workspace.index_db->condemn();
+    workspace.index_db.reset();
+    workspace.index_db = index::open_database(*workspace.store, workspace.config.project.index_db);
+}
+
 bool Indexer::load(bool read_only) {
-    if(!workspace.index_storage)
+    if(!workspace.index_db)
         return true;
-    auto& storage = *workspace.index_storage;
+    auto& db = *workspace.index_db;
     auto& project = workspace.project_index;
     ScopedTimer timer;
 
@@ -595,15 +750,13 @@ bool Indexer::load(bool read_only) {
             return;
         }
         for(auto kind: {index::IndexBlobKind::Shard, index::IndexBlobKind::Manifest}) {
-            llvm::SmallVector<std::string> keys;
-            storage.for_each_key(kind, [&](llvm::StringRef key) { keys.push_back(key.str()); });
-            for(auto& key: keys) {
-                storage.remove(kind, key);
-            }
+            db.for_each_key(kind, [&](llvm::StringRef key) {
+                startup_removes.push_back({kind, key.str()});
+            });
         }
     };
 
-    auto global = storage.read(index::IndexBlobKind::Global, "global");
+    auto global = db.read(index::IndexBlobKind::Global, "global");
     if(!global) {
         // A global blob that exists but failed to open is a transient IO
         // error, not absence: sweeping would destroy an intact index and
@@ -611,9 +764,19 @@ bool Indexer::load(bool read_only) {
         // saving a fresh lineage over blobs whose anchor was never read
         // could alias their fv ids and generation stamps — and leave
         // everything for a healthier restart to load.
-        if(storage.contains(index::IndexBlobKind::Global, "global")) {
-            LOG_WARN("Index global blob unreadable; disabling index persistence this session");
-            workspace.index_storage.reset();
+        if(db.contains(index::IndexBlobKind::Global, "global")) {
+            // Confirmed page corruption heals through rebuildability: the
+            // condemned database deletes itself and a fresh empty one
+            // opens in its place, so this session's rebuild persists
+            // instead of being redone at the next start. Transient
+            // failures touch nothing.
+            if(!read_only && db.corrupted()) {
+                LOG_WARN("Index database is corrupt; discarding it and rebuilding from scratch");
+                reopen_fresh_database();
+            } else {
+                LOG_WARN("Index global blob unreadable; disabling index persistence this session");
+                workspace.index_db.reset();
+            }
             return true;
         }
         // No global table means no resolvable manifests: everything else
@@ -622,11 +785,11 @@ bool Indexer::load(bool read_only) {
         return true;
     }
     llvm::DenseMap<std::uint32_t, std::uint64_t> manifest_pins;
-    if(!project.load_global(global->getBuffer(), workspace.path_pool, manifest_pins)) {
+    if(!project.load_global(global.buffer->getBuffer(), workspace.path_pool, manifest_pins)) {
         LOG_INFO("Discarding old-format index global blob");
         sweep_all();
         if(!read_only) {
-            storage.remove(index::IndexBlobKind::Global, "global");
+            startup_removes.push_back({index::IndexBlobKind::Global, "global"});
         }
         return false;
     }
@@ -638,9 +801,9 @@ bool Indexer::load(bool read_only) {
     // and are swept, with their TUs re-enqueued where recoverable.
     llvm::DenseSet<std::uint32_t> adopted_pins;
     llvm::SmallVector<std::string> dead_manifests;
-    storage.for_each_key(index::IndexBlobKind::Manifest, [&](llvm::StringRef key) {
-        auto blob = storage.read(index::IndexBlobKind::Manifest, key);
-        auto manifest = blob ? index::deserialize_manifest(blob->getBuffer()) : std::nullopt;
+    db.for_each_key(index::IndexBlobKind::Manifest, [&](llvm::StringRef key) {
+        auto blob = db.read(index::IndexBlobKind::Manifest, key);
+        auto manifest = blob ? index::deserialize_manifest(blob.buffer->getBuffer()) : std::nullopt;
         auto pin = manifest ? manifest_pins.find(manifest->tu_fv) : manifest_pins.end();
         if(!manifest || pin == manifest_pins.end() || pin->second != manifest->global_gen ||
            !project.knows_file_versions(*manifest)) {
@@ -663,7 +826,7 @@ bool Indexer::load(bool read_only) {
     });
     if(!read_only) {
         for(auto& key: dead_manifests) {
-            storage.remove(index::IndexBlobKind::Manifest, key);
+            startup_removes.push_back({index::IndexBlobKind::Manifest, std::move(key)});
         }
     }
     // A pinned TU without an adopted manifest lost it to a failed write
@@ -705,7 +868,7 @@ bool Indexer::load(bool read_only) {
     llvm::SmallVector<std::uint32_t> unservable;
     for(auto& [path_id, entry]: project.contributions) {
         auto key = blob_key(workspace.path_pool.resolve(path_id));
-        auto shard = index::Shard::from_buffer(storage.read(index::IndexBlobKind::Shard, key));
+        auto shard = index::Shard::from_buffer(db.read(index::IndexBlobKind::Shard, key).buffer);
         // A blob can verify yet miss a contributed variant, or carry
         // another content generation than the contributions pin (crash or
         // failed write left a manifest newer than its shard); set_live
@@ -746,8 +909,8 @@ bool Indexer::load(bool read_only) {
             auto affected = project.remove_manifest(tu);
             mask_refresh.append(affected.begin(), affected.end());
             if(!read_only) {
-                storage.remove(index::IndexBlobKind::Manifest,
-                               blob_key(workspace.path_pool.resolve(tu)));
+                startup_removes.push_back(
+                    {index::IndexBlobKind::Manifest, blob_key(workspace.path_pool.resolve(tu))});
             }
             enqueue(tu, ReindexReason::ContentChanged);
         }
@@ -761,18 +924,42 @@ bool Indexer::load(bool read_only) {
 
     // Sweep shard blobs nothing references any more.
     if(!read_only) {
-        llvm::SmallVector<std::string> orphans;
-        storage.for_each_key(index::IndexBlobKind::Shard, [&](llvm::StringRef key) {
+        db.for_each_key(index::IndexBlobKind::Shard, [&](llvm::StringRef key) {
             if(!expected_keys.contains(key)) {
-                orphans.push_back(key.str());
+                startup_removes.push_back({index::IndexBlobKind::Shard, key.str()});
             }
         });
-        for(auto& key: orphans) {
-            storage.remove(index::IndexBlobKind::Shard, key);
-        }
         reconcile_cdb_snapshot();
     }
 
+    // The reads above touch every adopted manifest and shard, so page
+    // corruption anywhere in the database has latched by now; heal it like
+    // the unreadable-global case above instead of writing into a damaged
+    // tree every session. The adopted state unwinds wholesale: the shard
+    // views borrow from the condemned environment, and manifests kept
+    // without their views would read as fresh and gate the rebuild sweep
+    // off exactly the files whose rows were lost. Standalone-indexed TUs
+    // re-enqueue first — the CDB sweep that rebuilds everything else never
+    // covers them, and the condemned database is deleting their only
+    // persistent record. The dirtied snapshot carries them as debt from
+    // the fresh database's first save on, so even a crash before their
+    // rebuild lands cannot lose them a second time.
+    if(!read_only && db.corrupted()) {
+        LOG_WARN("Index database is corrupt; discarding it and rebuilding from scratch");
+        for(auto tu: llvm::make_first_range(project.manifests)) {
+            if(!workspace.cdb.has_entry(workspace.path_pool.resolve(tu))) {
+                enqueue(tu, ReindexReason::ContentChanged);
+            }
+        }
+        workspace.shards.clear();
+        project = index::ProjectIndex();
+        startup_removes.clear();
+        persisted_cdb_snapshot.clear();
+        cdb_dirty = true;
+        reopen_fresh_database();
+        return true;
+    }
+
     if(!workspace.shards.empty()) {
         LOG_INFO("Loaded {} index shards, {} manifests, {} symbols",
                  workspace.shards.size(),
@@ -808,9 +995,10 @@ llvm::SmallVector<std::uint32_t> Indexer::standalone_debt() {
 }
 
 void Indexer::reconcile_cdb_snapshot() {
-    auto blob = workspace.index_storage->read(index::IndexBlobKind::Cdb, "cdb");
-    CdbSnapshot persisted;
-    if(!blob || !kota::codec::json::from_string(std::string_view(blob->getBuffer()), persisted)) {
+    auto blob = workspace.index_db->read(index::IndexBlobKind::CDB, "cdb");
+    CDBSnapshot persisted;
+    if(!blob ||
+       !kota::codec::json::from_string(std::string_view(blob.buffer->getBuffer()), persisted)) {
         // Unknown baseline: nothing to diff against. Dirty the snapshot so
         // the next save recreates it even when it commits nothing else —
         // after a crash that lost only the CDB blob, waiting for an
@@ -819,9 +1007,9 @@ void Indexer::reconcile_cdb_snapshot() {
         cdb_dirty = true;
         return;
     }
-    persisted_cdb_snapshot = blob->getBuffer().str();
+    persisted_cdb_snapshot = blob.buffer->getBuffer().str();
 
-    llvm::StringMap<const CdbSnapshotEntry*> before;
+    llvm::StringMap<const CDBSnapshotEntry*> before;
     for(auto& entry: persisted.entries) {
         before[entry.file] = &entry;
     }
diff --git a/src/server/compiler/indexer.h b/src/server/compiler/indexer.h
index b09818459..728a5f9e2 100644
--- a/src/server/compiler/indexer.h
+++ b/src/server/compiler/indexer.h
@@ -335,6 +335,11 @@ class Indexer {
     llvm::DenseSet<std::uint32_t> dirty_manifests;
     bool global_dirty = false;
 
+    /// Blob removals discovered during load (stale manifests, orphan
+    /// shards, swept layouts), deferred into the first save so startup
+    /// never runs synchronous database commits on the event loop.
+    llvm::SmallVector<index::BlobKey> startup_removes;
+
     /// The persisted CDB snapshot blob's bytes as last read or written;
     /// empty when none exists. save() rewrites the blob whenever the live
     /// CDB serializes differently.
@@ -422,6 +427,38 @@ class Indexer {
         kota::event task_done{false};
     };
 
+    /// Confirmed corruption heals through rebuildability: condemn the
+    /// database (deleted on close) and continue on a freshly opened empty
+    /// one, so the session's rebuild persists instead of waiting for the
+    /// next start. A failed reopen (another process grabbed the writer
+    /// lock meanwhile) leaves persistence disabled for the session.
+    void reopen_fresh_database();
+
+    /// Re-enqueue every TU contributing to `path_id`'s shard. Used when the
+    /// file's resident rows are lost while its manifests still read fresh:
+    /// no in-process event would ever rebuild them, and for standalone
+    /// headers no restart sweep would either.
+    void requeue_owners(std::uint32_t path_id);
+
+    /// Drop every resident shard that may borrow database memory —
+    /// everything not dirty, since dirty shards own their bytes by
+    /// construction (merges install memory copies) — and requeue the
+    /// owners of the dropped rows.
+    void shed_borrowed_shards();
+
+    /// Runtime-corruption recovery, shared by the write-time and the
+    /// snapshot-migration detection points: nothing in the condemned
+    /// database survives, so borrowed shards are shed with their owners
+    /// requeued while every manifest, the global and the CDB snapshot
+    /// re-dirty to re-persist into the freshly opened database.
+    void recover_corrupt_database();
+
+    /// Migrate resident shards onto a fresh database read snapshot after a
+    /// save's commit (growing the map first when the write hit a full one),
+    /// then retire the previous snapshot. Filesystem-backed runs return
+    /// immediately: their buffers are immortal.
+    kota::task<> migrate_shard_views();
+
     kota::task<> run_background_indexing();
 
     /// The round's dispatch loop, spawned as a child of `workers` so that a
diff --git a/src/server/state/config.h b/src/server/state/config.h
index 275e53982..ee838735d 100644
--- a/src/server/state/config.h
+++ b/src/server/state/config.h
@@ -63,6 +63,12 @@ struct ProjectConfig {
     KOTATSU_ANNOTATE(defaulted = true, description = "Build the background index.")
     <bool> enable_indexing = true;
 
+    KOTATSU_ANNOTATE(defaulted = true,
+                     description =
+                         "Index persistence backend: \"lmdb\" (single database "
+                         "file) or \"files\" (one file per blob).")
+    <std::string> index_db = "lmdb";
+
     KOTATSU_ANNOTATE(defaulted = true,
                      description =
                          "Idle delay in milliseconds before background indexing "
diff --git a/src/server/state/workspace.h b/src/server/state/workspace.h
index edb58b6b1..a8d27b113 100644
--- a/src/server/state/workspace.h
+++ b/src/server/state/workspace.h
@@ -12,9 +12,9 @@
 #include "command/command.h"
 #include "command/toolchain.h"
 #include "compile/dep_file.h"
+#include "index/database.h"
 #include "index/project_index.h"
 #include "index/shard.h"
-#include "index/storage.h"
 #include "index/tu_index.h"
 #include "semantic/symbol.h"
 #include "server/compiler/compile_graph.h"
@@ -36,7 +36,7 @@ class ContextResolver;
 
 /// On-disk cache layout version (CacheStore root `cache/v{N}`).
 /// Bump to discard all cached artifacts after incompatible format changes.
-constexpr inline std::uint32_t cache_format_version = 6;
+constexpr inline std::uint32_t cache_format_version = 7;
 
 /// Sentinel for "no path": path pool ids start at 0, so 0 is a real file.
 constexpr inline std::uint32_t no_path_id = ~0u;
@@ -221,6 +221,12 @@ struct Workspace {
     /// recovery); validity metadata (deps snapshots) stays in cache.json.
     std::optional<CacheStore> store;
 
+    /// Index blob persistence, opened together with the cache store.
+    /// Declared right after `store` (both backends borrow it) and before
+    /// every index structure that borrows database bytes (`shards`), so
+    /// destruction runs shards → index_db → store.
+    std::unique_ptr<index::BlobDatabase> index_db;
+
     /// Include relationships between files on disk (#include edges).
     /// Built once at startup from CDB scan; updated incrementally on didSave.
     DependencyGraph dep_graph;
@@ -271,11 +277,6 @@ struct Workspace {
     /// position mapping, served zero-copy.
     llvm::DenseMap<std::uint32_t, index::Shard> shards;
 
-    /// Index blob persistence, opened together with the cache store.
-    /// Declared after `store`: the filesystem backend borrows it, so it
-    /// must be destroyed first.
-    std::unique_ptr<index::IndexStorage> index_storage;
-
     /// Monotonic generation of context-affecting workspace state (include
     /// graph, CDB, disk contents). Bumped on didSave; clice/queryContext
     /// stamps its results with it and clice/switchContext rejects requests
diff --git a/src/server/transport/master_server.cpp b/src/server/transport/master_server.cpp
index 418aa3faf..ea0b1bf51 100644
--- a/src/server/transport/master_server.cpp
+++ b/src/server/transport/master_server.cpp
@@ -491,8 +491,7 @@ void MasterServer::open_cache_store() {
     store->register_namespace(
         {.name = "header_context", .extension = ".h", .policy = CachePolicy::Scratch});
     workspace.store.emplace(std::move(*store));
-    // Registers the index namespaces itself.
-    workspace.index_storage = index::make_fs_index_storage(*workspace.store);
+    workspace.index_db = index::open_database(*workspace.store, cfg.index_db);
     LOG_INFO("Cache store: {}", workspace.store->base_dir());
 
     workspace.load_cache(contexts);
diff --git a/tests/integration/compilation/persistent_cache.test.ts b/tests/integration/compilation/persistent_cache.test.ts
index 306c48769..bbd3707a5 100644
--- a/tests/integration/compilation/persistent_cache.test.ts
+++ b/tests/integration/compilation/persistent_cache.test.ts
@@ -487,13 +487,18 @@ test("cache dirs created on startup", async ({ session }) => {
     const [uri] = await client.openAndWait("main.cpp");
     client.assertCleanCompile(uri);
 
-    for (const subdir of ["pch", "pcm", "index"]) {
+    for (const subdir of ["pch", "pcm"]) {
         expect(
             fs.existsSync(path.join(workspace.cacheRoot(), subdir)) &&
                 fs.statSync(path.join(workspace.cacheRoot(), subdir)).isDirectory(),
             `${subdir}/ should be created`,
         ).toBe(true);
     }
+    // The index persists into a single LMDB database, not a namespace dir.
+    expect(
+        fs.existsSync(path.join(workspace.cacheRoot(), "index.mdb")),
+        "index.mdb should be created",
+    ).toBe(true);
 });
 
 test("different flags different pch", async ({ session }) => {
@@ -688,6 +693,32 @@ test("corrupt pch idx retracted", async ({ session }) => {
     await c2.shutdown();
 });
 
+/// Best-effort recursive wipe: a live server keeps its LMDB index
+/// memory-mapped, and Windows refuses to delete mapped files (EPERM) —
+/// exactly what a user wiping the cache mid-session experiences there.
+/// Everything unlocked still goes.
+function wipeBestEffort(dir: string): void {
+    let entries: fs.Dirent[];
+    try {
+        entries = fs.readdirSync(dir, { withFileTypes: true });
+    } catch {
+        return;
+    }
+    for (const entry of entries) {
+        const child = path.join(dir, entry.name);
+        try {
+            if (entry.isDirectory()) {
+                wipeBestEffort(child);
+                fs.rmdirSync(child);
+            } else {
+                fs.rmSync(child, { force: true });
+            }
+        } catch {
+            // Locked by the live server; leave it.
+        }
+    }
+}
+
 test("cache wiped while running", async ({ session }) => {
     // Wiping the cache directory under a running server must not wedge
     // PCH builds forever: the store re-creates its directories on demand.
@@ -702,7 +733,7 @@ test("cache wiped while running", async ({ session }) => {
     client.assertCleanCompile(uri);
 
     // Simulate a user resetting state without restarting the server.
-    workspace.rm(path.join(".clice", "cache"));
+    wipeBestEffort(workspace.path(path.join(".clice", "cache")));
 
     // Change the preamble so a fresh PCH build is required.
     await sleep(1_100);
diff --git a/tests/integration/features/index_staleness.test.ts b/tests/integration/features/index_staleness.test.ts
index 23e15aaa7..8d7b92ecb 100644
--- a/tests/integration/features/index_staleness.test.ts
+++ b/tests/integration/features/index_staleness.test.ts
@@ -7,6 +7,10 @@ import { MTIME_GRANULARITY, sleep } from "@clice/tools/client";
 import { Workspace } from "@clice/tools/workspace";
 import { expect, test } from "../fixtures.ts";
 
+/// The probes below watch per-file blob mtimes, so the sessions pin the
+/// files backend (the default LMDB backend has no per-blob files).
+const NO_LMDB = { project: { index_db: "files" } };
+
 const HEADER = "#pragma once\ninline int alpha() { return 1; }\n";
 const CLOSED_TU = '#include "header.h"\nint use() { return alpha(); }\n';
 
@@ -49,7 +53,7 @@ test("touch header no reindex", async ({ session }) => {
 
     // Session 1: background-index the closed TU into a shard.
     const c1 = session.spawn(workspace);
-    await c1.initialize(workspace);
+    await c1.initialize(workspace, { initializationOptions: NO_LMDB });
     expect(await poll(() => shardMtimes(workspace).size > 0), "closed TU never indexed").toBe(true);
     await c1.shutdown();
 
@@ -66,7 +70,7 @@ test("touch header no reindex", async ({ session }) => {
     const before = shardMtimes(workspace);
     const globalBefore = globalMtime(workspace);
     const c2 = session.spawn(workspace);
-    await c2.initialize(workspace);
+    await c2.initialize(workspace, { initializationOptions: NO_LMDB });
     // The touch makes the header's stat mismatch its FileVersion stamp; the
     // staleness check re-hashes, proves a mere touch, and repairs the stamp
     // — which dirties the global blob, so its mtime moving proves both that
diff --git a/tests/unit/index/database_tests.cpp b/tests/unit/index/database_tests.cpp
new file mode 100644
index 000000000..14dfea551
--- /dev/null
+++ b/tests/unit/index/database_tests.cpp
@@ -0,0 +1,315 @@
+#include <cstdlib>
+#include <print>
+
+#include "test/temp_dir.h"
+#include "test/test.h"
+#include "index/database.h"
+#include "support/cache_store.h"
+#include "support/filesystem.h"
+
+#include "llvm/Support/FileSystem.h"
+#include "llvm/Support/raw_ostream.h"
+
+namespace clice::testing {
+namespace {
+
+constexpr std::uint32_t version = 1;
+constexpr llvm::StringLiteral backends[] = {"files", "lmdb"};
+
+/// Helper precondition check that survives NDEBUG builds; failures abort
+/// with a message instead of becoming UB on a bad expected access.
+void require(bool condition, const char* what) {
+    if(!condition) {
+        std::println(stderr, "database_tests: requirement failed: {}", what);
+        std::abort();
+    }
+}
+
+CacheStore open_store(TempDir& tmp, llvm::StringRef sub, bool read_only = false) {
+    auto store = CacheStore::open(tmp.path(sub), version, read_only);
+    require(store.has_value(), "CacheStore::open failed");
+    return std::move(*store);
+}
+
+index::BlobDatabase::Blob blob(index::IndexBlobKind kind, llvm::StringRef key, std::string bytes) {
+    return {kind, key.str(), std::move(bytes)};
+}
+
+/// Value sizes on the two LMDB layouts: large lands on overflow pages
+/// (16-aligned, served borrowed), small stays inline in the leaf
+/// (2-aligned at worst, served as an owned copy).
+std::string large_value(char fill) {
+    return std::string(8192, fill);
+}
+
+TEST_SUITE(IndexDatabase) {
+
+TEST_CASE(WriteReadRoundTrip) {
+    TempDir tmp;
+    for(auto backend: backends) {
+        auto store = open_store(tmp, backend);
+        auto db = index::open_database(store, backend);
+        ASSERT_TRUE(db != nullptr);
+
+        auto rejected = db->write({blob(index::IndexBlobKind::Shard, "a", large_value('a')),
+                                   blob(index::IndexBlobKind::Global, "global", "gg")},
+                                  {});
+        ASSERT_TRUE(rejected.empty());
+        // Reads serve the resident snapshot; committed writes become
+        // visible only after an advance (a no-op on the files backend).
+        ASSERT_TRUE(db->advance_read_snapshot().has_value());
+        db->retire_old_snapshot();
+
+        auto shard = db->read(index::IndexBlobKind::Shard, "a");
+        ASSERT_TRUE(bool(shard));
+        ASSERT_TRUE(shard.buffer->getBuffer() == large_value('a'));
+        ASSERT_TRUE(db->contains(index::IndexBlobKind::Global, "global"));
+        ASSERT_FALSE(db->contains(index::IndexBlobKind::Shard, "missing"));
+        ASSERT_FALSE(bool(db->read(index::IndexBlobKind::Shard, "missing")));
+    }
+}
+
+TEST_CASE(WriteRemoves) {
+    TempDir tmp;
+    for(auto backend: backends) {
+        auto store = open_store(tmp, backend);
+        auto db = index::open_database(store, backend);
+        ASSERT_TRUE(db != nullptr);
+
+        ASSERT_TRUE(db->write({blob(index::IndexBlobKind::Manifest, "m1", "one"),
+                               blob(index::IndexBlobKind::Manifest, "m2", "two")},
+                              {})
+                        .empty());
+        ASSERT_TRUE(db->write(
+                          {
+        },
+                          {{index::IndexBlobKind::Manifest, "m1"}})
+                        .empty());
+        ASSERT_TRUE(db->advance_read_snapshot().has_value());
+        db->retire_old_snapshot();
+
+        ASSERT_FALSE(db->contains(index::IndexBlobKind::Manifest, "m1"));
+        ASSERT_TRUE(db->contains(index::IndexBlobKind::Manifest, "m2"));
+    }
+}
+
+TEST_CASE(KindsAreIsolated) {
+    TempDir tmp;
+    for(auto backend: backends) {
+        auto store = open_store(tmp, backend);
+        auto db = index::open_database(store, backend);
+        ASSERT_TRUE(db != nullptr);
+
+        ASSERT_TRUE(db->write({blob(index::IndexBlobKind::Shard, "same", "shard"),
+                               blob(index::IndexBlobKind::Manifest, "same", "manifest")},
+                              {})
+                        .empty());
+        ASSERT_TRUE(db->advance_read_snapshot().has_value());
+        db->retire_old_snapshot();
+
+        llvm::SmallVector<std::string> shard_keys;
+        db->for_each_key(index::IndexBlobKind::Shard,
+                         [&](llvm::StringRef key) { shard_keys.push_back(key.str()); });
+        ASSERT_TRUE(shard_keys.size() == 1);
+        ASSERT_TRUE(shard_keys.front() == "same");
+        ASSERT_TRUE(db->read(index::IndexBlobKind::Manifest, "same").buffer->getBuffer() ==
+                    "manifest");
+        ASSERT_FALSE(db->contains(index::IndexBlobKind::Global, "same"));
+    }
+}
+
+TEST_CASE(SnapshotPinsUntilAdvance) {
+    TempDir tmp;
+    auto store = open_store(tmp, "lmdb");
+    auto db = index::open_database(store, "lmdb");
+    ASSERT_TRUE(db != nullptr);
+
+    ASSERT_TRUE(db->write({blob(index::IndexBlobKind::Shard, "k", large_value('1'))}, {}).empty());
+    ASSERT_TRUE(db->advance_read_snapshot().has_value());
+    db->retire_old_snapshot();
+
+    auto before = db->read(index::IndexBlobKind::Shard, "k");
+    ASSERT_TRUE(bool(before));
+
+    // Committed writes stay invisible to the resident snapshot until the
+    // next advance; the borrowed buffer keeps serving the old bytes.
+    ASSERT_TRUE(db->write({blob(index::IndexBlobKind::Shard, "k", large_value('2'))}, {}).empty());
+    ASSERT_TRUE(db->read(index::IndexBlobKind::Shard, "k").buffer->getBuffer() == large_value('1'));
+    ASSERT_TRUE(before.buffer->getBuffer() == large_value('1'));
+
+    auto advanced = db->advance_read_snapshot();
+    ASSERT_TRUE(advanced.has_value());
+    ASSERT_TRUE(*advanced != 0);
+    // Old and new snapshots serve their own bytes side by side until the
+    // old one retires — the migration window's core invariant.
+    ASSERT_TRUE(db->read(index::IndexBlobKind::Shard, "k").buffer->getBuffer() == large_value('2'));
+    ASSERT_TRUE(before.buffer->getBuffer() == large_value('1'));
+    db->retire_old_snapshot();
+}
+
+TEST_CASE(SmallValuesCopiedAligned) {
+    TempDir tmp;
+    auto store = open_store(tmp, "lmdb");
+    auto db = index::open_database(store, "lmdb");
+    ASSERT_TRUE(db != nullptr);
+
+    ASSERT_TRUE(db->write({blob(index::IndexBlobKind::Manifest, "small", "tiny"),
+                           blob(index::IndexBlobKind::Shard, "big", large_value('b'))},
+                          {})
+                    .empty());
+    ASSERT_TRUE(db->advance_read_snapshot().has_value());
+    db->retire_old_snapshot();
+
+    // Inline leaf values are only 2-aligned, so they must come back as an
+    // owned copy (generation 0); overflow-page values are borrowed.
+    auto small = db->read(index::IndexBlobKind::Manifest, "small");
+    ASSERT_TRUE(bool(small));
+    ASSERT_TRUE(small.generation == 0);
+    ASSERT_TRUE(reinterpret_cast<std::uintptr_t>(small.buffer->getBufferStart()) % 8 == 0);
+
+    auto big = db->read(index::IndexBlobKind::Shard, "big");
+    ASSERT_TRUE(bool(big));
+    ASSERT_TRUE(big.generation != 0);
+    ASSERT_TRUE(reinterpret_cast<std::uintptr_t>(big.buffer->getBufferStart()) % 8 == 0);
+}
+
+TEST_CASE(ReopenServesPersistedBlobs) {
+    TempDir tmp;
+    auto store = open_store(tmp, "lmdb");
+    {
+        auto db = index::open_database(store, "lmdb");
+        ASSERT_TRUE(db != nullptr);
+        ASSERT_TRUE(db->write({blob(index::IndexBlobKind::CDB, "cdb", "snapshot")}, {}).empty());
+    }
+    auto db = index::open_database(store, "lmdb");
+    ASSERT_TRUE(db != nullptr);
+    ASSERT_TRUE(db->read(index::IndexBlobKind::CDB, "cdb").buffer->getBuffer() == "snapshot");
+}
+
+TEST_CASE(CorruptDatabaseRebuilds) {
+    TempDir tmp;
+    auto store = open_store(tmp, "lmdb");
+    {
+        std::error_code ec;
+        llvm::raw_fd_ostream os(path::join(store.base_dir(), "index.mdb"), ec);
+        require(!ec, "writing garbage failed");
+        os << "this is not an lmdb file, not even close, but long enough to map";
+    }
+    auto db = index::open_database(store, "lmdb");
+    ASSERT_TRUE(db != nullptr);
+    ASSERT_FALSE(db->contains(index::IndexBlobKind::CDB, "cdb"));
+    ASSERT_TRUE(db->write({blob(index::IndexBlobKind::CDB, "cdb", "fresh")}, {}).empty());
+}
+
+TEST_CASE(FullMapFailsWholeBatchThenGrows) {
+    TempDir tmp;
+    auto store = open_store(tmp, "lmdb");
+    // Small enough that a handful of large values exhausts it.
+    auto db = index::open_lmdb_database(store, 256 * 1024);
+    ASSERT_TRUE(db != nullptr);
+
+    std::vector<index::BlobDatabase::Blob> puts;
+    for(int i = 0; i < 64; i += 1) {
+        puts.push_back(blob(index::IndexBlobKind::Shard, std::to_string(i), large_value('x')));
+    }
+    auto rejected = db->write(puts, {});
+    ASSERT_TRUE(rejected.size() == puts.size());
+    ASSERT_FALSE(db->contains(index::IndexBlobKind::Shard, "0"));
+
+    auto grown = db->grow();
+    ASSERT_TRUE(grown.has_value());
+    ASSERT_TRUE(*grown);
+    // grow() opened a fresh snapshot, so this proves the failed batch
+    // really committed nothing (the pre-grow check only saw the pinned
+    // old snapshot).
+    ASSERT_FALSE(db->contains(index::IndexBlobKind::Shard, "0"));
+    // A second grow without a latched full map is a no-op.
+    auto again = db->grow();
+    ASSERT_TRUE(again.has_value());
+    ASSERT_FALSE(*again);
+
+    ASSERT_TRUE(db->write(puts, {}).empty());
+    ASSERT_TRUE(db->advance_read_snapshot().has_value());
+    db->retire_old_snapshot();
+    ASSERT_TRUE(db->read(index::IndexBlobKind::Shard, "63").buffer->getBuffer() ==
+                large_value('x'));
+}
+
+TEST_CASE(ReadOnlyWithoutDatabaseFallsBack) {
+    TempDir tmp;
+    { auto store = open_store(tmp, "empty"); }
+    auto store = open_store(tmp, "empty", /*read_only=*/true);
+    // A reader before any LMDB writer ran gets the filesystem view (empty
+    // here) instead of failing on a missing index.mdb.
+    auto db = index::open_database(store, "lmdb");
+    ASSERT_TRUE(db != nullptr);
+    ASSERT_FALSE(db->contains(index::IndexBlobKind::Global, "global"));
+}
+
+TEST_CASE(ReadOnlyServesExistingDatabase) {
+    TempDir tmp;
+    {
+        auto store = open_store(tmp, "ws");
+        auto db = index::open_database(store, "lmdb");
+        ASSERT_TRUE(db != nullptr);
+        ASSERT_TRUE(db->write({blob(index::IndexBlobKind::Global, "global", "gg")}, {}).empty());
+    }
+    auto store = open_store(tmp, "ws", /*read_only=*/true);
+    auto db = index::open_database(store, "lmdb");
+    ASSERT_TRUE(db != nullptr);
+    ASSERT_TRUE(db->contains(index::IndexBlobKind::Global, "global"));
+    ASSERT_TRUE(db->read(index::IndexBlobKind::Global, "global").buffer->getBuffer() == "gg");
+}
+
+TEST_CASE(CondemnedDatabaseDeletesOnClose) {
+    TempDir tmp;
+    auto store = open_store(tmp, "lmdb");
+    {
+        auto db = index::open_database(store, "lmdb");
+        ASSERT_TRUE(db != nullptr);
+        ASSERT_TRUE(db->write({blob(index::IndexBlobKind::CDB, "cdb", "bytes")}, {}).empty());
+        db->condemn();
+    }
+    ASSERT_FALSE(llvm::sys::fs::exists(path::join(store.base_dir(), "index.mdb")));
+    auto db = index::open_database(store, "lmdb");
+    ASSERT_TRUE(db != nullptr);
+    ASSERT_FALSE(db->contains(index::IndexBlobKind::CDB, "cdb"));
+}
+
+TEST_CASE(OutstandingSnapshotsStack) {
+    TempDir tmp;
+    auto store = open_store(tmp, "lmdb");
+    auto db = index::open_database(store, "lmdb");
+    ASSERT_TRUE(db != nullptr);
+
+    ASSERT_TRUE(db->write({blob(index::IndexBlobKind::Shard, "k", large_value('1'))}, {}).empty());
+    ASSERT_TRUE(db->advance_read_snapshot().has_value());
+    db->retire_old_snapshot();
+    auto lease = db->read(index::IndexBlobKind::Shard, "k");
+    ASSERT_TRUE(bool(lease));
+
+    // A cancelled migration leaves its old snapshot outstanding and the
+    // next advance stacks another; every stacked snapshot keeps its
+    // borrowers alive until one retire clears them all.
+    ASSERT_TRUE(db->write({blob(index::IndexBlobKind::Shard, "k", large_value('2'))}, {}).empty());
+    ASSERT_TRUE(db->advance_read_snapshot().has_value());
+    ASSERT_TRUE(db->write({blob(index::IndexBlobKind::Shard, "k", large_value('3'))}, {}).empty());
+    ASSERT_TRUE(db->advance_read_snapshot().has_value());
+    ASSERT_TRUE(lease.buffer->getBuffer() == large_value('1'));
+    ASSERT_TRUE(db->read(index::IndexBlobKind::Shard, "k").buffer->getBuffer() == large_value('3'));
+    db->retire_old_snapshot();
+    ASSERT_TRUE(db->read(index::IndexBlobKind::Shard, "k").buffer->getBuffer() == large_value('3'));
+}
+
+TEST_CASE(UnknownBackendFallsBackToLmdb) {
+    TempDir tmp;
+    auto store = open_store(tmp, "lmdb");
+    auto db = index::open_database(store, "bogus");
+    ASSERT_TRUE(db != nullptr);
+    ASSERT_TRUE(llvm::sys::fs::exists(path::join(store.base_dir(), "index.mdb")));
+}
+
+};  // TEST_SUITE(IndexDatabase)
+
+}  // namespace
+}  // namespace clice::testing
diff --git a/tests/unit/index/shard_tests.cpp b/tests/unit/index/shard_tests.cpp
index a9e57e302..a1b85dfc4 100644
--- a/tests/unit/index/shard_tests.cpp
+++ b/tests/unit/index/shard_tests.cpp
@@ -1101,6 +1101,33 @@ TEST_CASE(CorruptRoaringMaskRejected) {
     ASSERT_FALSE(make_shard(bytes_of()).loaded());
 }
 
+TEST_CASE(RebindSwapsIdenticalBytes) {
+    build_index("int rebind_value() { return 1; }\n");
+    auto bytes = main_blob();
+    ASSERT_FALSE(bytes.empty());
+    auto shard = make_shard(bytes);
+    ASSERT_TRUE(shard.loaded());
+    auto hash_before = shard.content_hash();
+    const char* address_before = shard.bytes().data();
+
+    ASSERT_TRUE(shard.rebind(llvm::MemoryBuffer::getMemBufferCopy(bytes)));
+    ASSERT_TRUE(shard.bytes().data() != address_before);
+    ASSERT_EQ(shard.content_hash(), hash_before);
+
+    // Byte identity is the caller's contract: only the size gates the
+    // swap, so migration never touches the replacement's content pages.
+    std::string drift = bytes;
+    drift.back() = static_cast<char>(drift.back() ^ 1);
+    ASSERT_TRUE(shard.rebind(llvm::MemoryBuffer::getMemBufferCopy(drift)));
+
+    // A missing replacement or another size is rejected; the current
+    // buffer stays.
+    const char* kept = shard.bytes().data();
+    ASSERT_FALSE(shard.rebind(llvm::MemoryBuffer::getMemBufferCopy(bytes + "x")));
+    ASSERT_FALSE(shard.rebind(nullptr));
+    ASSERT_TRUE(shard.bytes().data() == kept);
+}
+
 };  // TEST_SUITE(Shard)
 
 }  // namespace
diff --git a/tests/unit/server/indexer_tests.cpp b/tests/unit/server/indexer_tests.cpp
index 2403b7589..8ed6919b8 100644
--- a/tests/unit/server/indexer_tests.cpp
+++ b/tests/unit/server/indexer_tests.cpp
@@ -1,3 +1,4 @@
+#include <expected>
 #include <format>
 #include <limits>
 #include <memory>
@@ -6,10 +7,10 @@
 #include "test/test.h"
 #include "command/argument_parser.h"
 #include "compile/compilation.h"
+#include "index/database.h"
 #include "index/manifest.h"
 #include "index/serialization.h"
 #include "index/shard.h"
-#include "index/storage.h"
 #include "index/tu_index.h"
 #include "server/compiler/context_resolver.h"
 #include "server/compiler/indexer.h"
@@ -239,7 +240,7 @@ void open_store(TempDir& tmp, Workspace& workspace) {
     auto store = CacheStore::open(tmp.path("cache"), 1);
     ASSERT_TRUE(store.has_value());
     workspace.store.emplace(std::move(*store));
-    workspace.index_storage = index::make_fs_index_storage(*workspace.store);
+    workspace.index_db = index::open_fs_database(*workspace.store);
 }
 
 /// The storage key of a file's shard or manifest blob (Indexer's naming).
@@ -332,6 +333,174 @@ TEST_CASE(SaveCommitsDirtyShard) {
     ASSERT_TRUE(workspace.project_index.contributions.lookup(path_id).contains(path_id));
 }
 
+TEST_CASE(SaveMigratesShardViews) {
+    TempDir tmp;
+    tmp.touch("main.cpp", "int migrate_value() { return 1; }\n");
+    auto src = tmp.path("main.cpp");
+
+    // The LMDB backend wrapped in a spy: save() must advance the read
+    // snapshot exactly once, rebind the resident shard onto it, and
+    // retire the old snapshot exactly once.
+    struct SnapshotSpy final : index::BlobDatabase {
+        std::unique_ptr<index::BlobDatabase> real;
+        int advances = 0;
+        int retires = 0;
+
+        index::ReadBlob read(index::IndexBlobKind kind, llvm::StringRef key) override {
+            return real->read(kind, key);
+        }
+
+        bool contains(index::IndexBlobKind kind, llvm::StringRef key) override {
+            return real->contains(kind, key);
+        }
+
+        llvm::SmallVector<std::size_t> write(llvm::ArrayRef<Blob> puts,
+                                             llvm::ArrayRef<index::BlobKey> removes) override {
+            return real->write(puts, removes);
+        }
+
+        void for_each_key(index::IndexBlobKind kind,
+                          llvm::function_ref<void(llvm::StringRef)> fn) override {
+            real->for_each_key(kind, fn);
+        }
+
+        std::expected<std::uint64_t, std::string> advance_read_snapshot() override {
+            advances += 1;
+            return real->advance_read_snapshot();
+        }
+
+        void retire_old_snapshot() override {
+            retires += 1;
+            real->retire_old_snapshot();
+        }
+
+        std::expected<bool, std::string> grow() override {
+            return real->grow();
+        }
+    };
+
+    auto store = CacheStore::open(tmp.path("cache"), 1);
+    ASSERT_TRUE(store.has_value());
+    workspace.store.emplace(std::move(*store));
+    auto spy = std::make_unique<SnapshotSpy>();
+    spy->real = index::open_lmdb_database(*workspace.store);
+    ASSERT_TRUE(spy->real != nullptr);
+    auto* probe = spy.get();
+    workspace.index_db = std::move(spy);
+
+    auto indexed = index_file(tmp, src);
+    ASSERT_FALSE(indexed.data.empty());
+    indexer.merge(indexed.data.data(), indexed.data.size());
+    auto path_id = workspace.path_pool.intern(indexed.tu_path);
+    auto before = workspace.shards.find(path_id);
+    ASSERT_TRUE(before != workspace.shards.end());
+    auto variants_before = before->second.variants();
+    const char* bytes_before = before->second.bytes().data();
+
+    auto save_body = [&]() -> kota::task<> {
+        co_await indexer.save();
+    };
+    auto task = save_body();
+    loop.schedule(task);
+    loop.run();
+
+    // Rebound: the shard now serves the database's snapshot view — the
+    // same bytes at a different address — with the verification state and
+    // variant set carried over.
+    ASSERT_EQ(probe->advances, 1);
+    ASSERT_EQ(probe->retires, 1);
+    auto it = workspace.shards.find(path_id);
+    ASSERT_TRUE(it != workspace.shards.end());
+    ASSERT_TRUE(it->second.loaded());
+    ASSERT_TRUE(it->second.bytes().data() != bytes_before);
+    ASSERT_EQ(it->second.content_hash(), llvm::xxh3_64bits("int migrate_value() { return 1; }\n"));
+    ASSERT_TRUE(it->second.variants() == variants_before);
+}
+
+TEST_CASE(GrowFailureShedsCleanShards) {
+    TempDir tmp;
+    tmp.touch("clean.cpp", "int clean_value() { return 1; }\n");
+    tmp.touch("dirty.cpp", "int dirty_value() { return 2; }\n");
+    open_store(tmp, workspace);
+
+    // grow() failing is backend-independent shed territory: every clean
+    // (non-dirty, possibly borrowed) shard must go with its owner requeued
+    // — the manifests still read fresh, so nothing else would rebuild the
+    // dropped rows — while dirty ones are owned by construction and stay.
+    // The spy also fails dirty.cpp's put so it is re-dirtied by the time
+    // the migration runs.
+    struct FailingGrow final : index::BlobDatabase {
+        std::unique_ptr<index::BlobDatabase> real;
+        std::string fail_key;
+
+        index::ReadBlob read(index::IndexBlobKind kind, llvm::StringRef key) override {
+            return real->read(kind, key);
+        }
+
+        bool contains(index::IndexBlobKind kind, llvm::StringRef key) override {
+            return real->contains(kind, key);
+        }
+
+        llvm::SmallVector<std::size_t> write(llvm::ArrayRef<Blob> puts,
+                                             llvm::ArrayRef<index::BlobKey> removes) override {
+            auto failed = real->write(puts, removes);
+            for(std::size_t i = 0; i < puts.size(); i += 1) {
+                if(puts[i].key == fail_key && !llvm::is_contained(failed, i)) {
+                    failed.push_back(i);
+                }
+            }
+            return failed;
+        }
+
+        void for_each_key(index::IndexBlobKind kind,
+                          llvm::function_ref<void(llvm::StringRef)> fn) override {
+            real->for_each_key(kind, fn);
+        }
+
+        std::expected<std::uint64_t, std::string> advance_read_snapshot() override {
+            return real->advance_read_snapshot();
+        }
+
+        void retire_old_snapshot() override {
+            real->retire_old_snapshot();
+        }
+
+        std::expected<bool, std::string> grow() override {
+            return std::unexpected(std::string("address space exhausted"));
+        }
+    };
+
+    auto indexed_clean = index_file(tmp, tmp.path("clean.cpp"));
+    auto indexed_dirty = index_file(tmp, tmp.path("dirty.cpp"));
+    ASSERT_FALSE(indexed_clean.data.empty());
+    ASSERT_FALSE(indexed_dirty.data.empty());
+
+    auto spy = std::make_unique<FailingGrow>();
+    spy->real = std::move(workspace.index_db);
+    // Through the pool: the indexer keys blobs by the pool-canonical path,
+    // which need not equal the raw temp path byte-for-byte (Windows 8.3
+    // names).
+    spy->fail_key =
+        blob_key(workspace.path_pool.resolve(workspace.path_pool.intern(indexed_dirty.tu_path)));
+    workspace.index_db = std::move(spy);
+
+    indexer.merge(indexed_clean.data.data(), indexed_clean.data.size());
+    indexer.merge(indexed_dirty.data.data(), indexed_dirty.data.size());
+    auto clean_id = workspace.path_pool.intern(indexed_clean.tu_path);
+    auto dirty_id = workspace.path_pool.intern(indexed_dirty.tu_path);
+
+    auto save_body = [&]() -> kota::task<> {
+        co_await indexer.save();
+    };
+    auto task = save_body();
+    loop.schedule(task);
+    loop.run();
+
+    ASSERT_FALSE(workspace.shards.contains(clean_id));
+    ASSERT_TRUE(workspace.shards.contains(dirty_id));
+    ASSERT_TRUE(indexer.pending_reason(clean_id) == ReindexReason::ContentChanged);
+}
+
 TEST_CASE(MidSaveMergeKept) {
     TempDir tmp;
     tmp.touch("main.cpp", "int first_value() { return 1; }\n");
@@ -539,8 +708,8 @@ TEST_CASE(SaveCompactsAndRetires) {
     ASSERT_FALSE(indexer.pending_reason(workspace.path_pool.intern(a2.tu_path)).has_value());
     bool on_disk = false;
     auto key = blob_key(workspace.path_pool.resolve(header_id));
-    workspace.index_storage->for_each_key(index::IndexBlobKind::Shard,
-                                          [&](llvm::StringRef k) { on_disk |= k == key; });
+    workspace.index_db->for_each_key(index::IndexBlobKind::Shard,
+                                     [&](llvm::StringRef k) { on_disk |= k == key; });
     ASSERT_FALSE(on_disk);
 }
 
@@ -591,8 +760,8 @@ TEST_CASE(SaveRetiresPinnedShard) {
     ASSERT_FALSE(workspace.shards.contains(header_id));
     bool on_disk = false;
     auto key = blob_key(workspace.path_pool.resolve(header_id));
-    workspace.index_storage->for_each_key(index::IndexBlobKind::Shard,
-                                          [&](llvm::StringRef k) { on_disk |= k == key; });
+    workspace.index_db->for_each_key(index::IndexBlobKind::Shard,
+                                     [&](llvm::StringRef k) { on_disk |= k == key; });
     ASSERT_FALSE(on_disk);
 
     // pb's manifest survives, still pinning rows the retirement made
@@ -715,30 +884,39 @@ TEST_CASE(FailedWriteNotCounted) {
     // A storage whose commits never land (disk full, permissions): the
     // gauge must report what was durably committed, not what the save
     // attempted.
-    struct FailingStorage final : index::IndexStorage {
-        std::unique_ptr<llvm::MemoryBuffer> read(index::IndexBlobKind, llvm::StringRef) override {
-            return nullptr;
+    struct FailingStorage final : index::BlobDatabase {
+        index::ReadBlob read(index::IndexBlobKind, llvm::StringRef) override {
+            return {};
         }
 
         bool contains(index::IndexBlobKind, llvm::StringRef) override {
             return false;
         }
 
-        llvm::SmallVector<std::size_t> write(llvm::ArrayRef<Blob> batch) override {
+        llvm::SmallVector<std::size_t> write(llvm::ArrayRef<Blob> puts,
+                                             llvm::ArrayRef<index::BlobKey>) override {
             llvm::SmallVector<std::size_t> failed;
-            for(std::size_t i = 0; i < batch.size(); i += 1) {
+            for(std::size_t i = 0; i < puts.size(); i += 1) {
                 failed.push_back(i);
             }
             return failed;
         }
 
-        void remove(index::IndexBlobKind, llvm::StringRef) override {}
-
         void for_each_key(index::IndexBlobKind,
                           llvm::function_ref<void(llvm::StringRef)>) override {}
+
+        std::expected<std::uint64_t, std::string> advance_read_snapshot() override {
+            return 0;
+        }
+
+        void retire_old_snapshot() override {}
+
+        std::expected<bool, std::string> grow() override {
+            return false;
+        }
     };
 
-    workspace.index_storage = std::make_unique<FailingStorage>();
+    workspace.index_db = std::make_unique<FailingStorage>();
 
     auto indexed = index_file(tmp, src);
     ASSERT_FALSE(indexed.data.empty());
@@ -766,28 +944,212 @@ TEST_CASE(FailedWriteNotCounted) {
     ASSERT_EQ(indexer.pending_shard_writes(), 0u);
 }
 
+TEST_CASE(WriteCorruptionRebuildsDatabase) {
+    TempDir tmp;
+    tmp.touch("clean.cpp", "int clean_value() { return 1; }\n");
+    tmp.touch("dirty.cpp", "int dirty_value() { return 2; }\n");
+    open_store(tmp, workspace);
+
+    // Corruption surfacing at write time (a damaged page only the write's
+    // tree descent reaches): the save must condemn the environment and
+    // continue on a fresh one instead of re-writing into it every save.
+    // Batch shards own their bytes and stay to re-persist; the clean
+    // resident view is shed and its owner re-enqueued.
+    struct CorruptOnWrite final : index::BlobDatabase {
+        bool* condemned;
+        bool fail = false;
+        bool poisoned = false;
+
+        index::ReadBlob read(index::IndexBlobKind, llvm::StringRef) override {
+            return {};
+        }
+
+        bool contains(index::IndexBlobKind, llvm::StringRef) override {
+            return false;
+        }
+
+        llvm::SmallVector<std::size_t> write(llvm::ArrayRef<Blob> puts,
+                                             llvm::ArrayRef<index::BlobKey>) override {
+            if(!fail) {
+                return {};
+            }
+            poisoned = true;
+            llvm::SmallVector<std::size_t> failed;
+            for(std::size_t i = 0; i < puts.size(); i += 1) {
+                failed.push_back(i);
+            }
+            return failed;
+        }
+
+        void for_each_key(index::IndexBlobKind,
+                          llvm::function_ref<void(llvm::StringRef)>) override {}
+
+        std::expected<std::uint64_t, std::string> advance_read_snapshot() override {
+            return 0;
+        }
+
+        void retire_old_snapshot() override {}
+
+        std::expected<bool, std::string> grow() override {
+            return false;
+        }
+
+        bool corrupted() const override {
+            return poisoned;
+        }
+
+        void condemn() override {
+            *condemned = true;
+        }
+    };
+
+    bool condemned = false;
+    auto spy = std::make_unique<CorruptOnWrite>();
+    spy->condemned = &condemned;
+    auto* probe = spy.get();
+    workspace.index_db = std::move(spy);
+
+    auto indexed_clean = index_file(tmp, tmp.path("clean.cpp"));
+    auto indexed_dirty = index_file(tmp, tmp.path("dirty.cpp"));
+    ASSERT_FALSE(indexed_clean.data.empty());
+    ASSERT_FALSE(indexed_dirty.data.empty());
+
+    auto save = [&] {
+        auto body = [&]() -> kota::task<> {
+            co_await indexer.save();
+        };
+        auto task = body();
+        loop.schedule(task);
+        loop.run();
+    };
+
+    indexer.merge(indexed_clean.data.data(), indexed_clean.data.size());
+    save();
+    indexer.merge(indexed_dirty.data.data(), indexed_dirty.data.size());
+    probe->fail = true;
+    save();
+
+    ASSERT_TRUE(condemned);
+    ASSERT_TRUE(workspace.index_db != nullptr);
+    auto clean_id = workspace.path_pool.intern(indexed_clean.tu_path);
+    auto dirty_id = workspace.path_pool.intern(indexed_dirty.tu_path);
+    ASSERT_FALSE(workspace.shards.contains(clean_id));
+    ASSERT_TRUE(workspace.shards.contains(dirty_id));
+    ASSERT_TRUE(indexer.pending_reason(clean_id) == ReindexReason::ContentChanged);
+    ASSERT_EQ(indexer.last_save_shards(), 0u);
+
+    // The next save re-persists everything servable into the fresh database.
+    save();
+    ASSERT_EQ(indexer.last_save_shards(), 1u);
+    ASSERT_FALSE(indexer.has_unsaved_state());
+}
+
+TEST_CASE(MigrationCorruptionRebuildsDatabase) {
+    TempDir tmp;
+    tmp.touch("main.cpp", "int migrate_value() { return 1; }\n");
+    open_store(tmp, workspace);
+
+    // Corruption surfacing first at migration time (a damaged page only the
+    // re-read from the advanced snapshot reaches, after the write-time
+    // check passed): same recovery as write-time corruption — the resident
+    // view is shed with its owner re-enqueued, the environment condemned
+    // and replaced by a fresh one.
+    struct CorruptOnRead final : index::BlobDatabase {
+        bool* condemned;
+        bool poisoned = false;
+
+        index::ReadBlob read(index::IndexBlobKind, llvm::StringRef) override {
+            poisoned = true;
+            return {};
+        }
+
+        bool contains(index::IndexBlobKind, llvm::StringRef) override {
+            return false;
+        }
+
+        llvm::SmallVector<std::size_t> write(llvm::ArrayRef<Blob>,
+                                             llvm::ArrayRef<index::BlobKey>) override {
+            return {};
+        }
+
+        void for_each_key(index::IndexBlobKind,
+                          llvm::function_ref<void(llvm::StringRef)>) override {}
+
+        std::expected<std::uint64_t, std::string> advance_read_snapshot() override {
+            return 2;
+        }
+
+        void retire_old_snapshot() override {}
+
+        std::expected<bool, std::string> grow() override {
+            return false;
+        }
+
+        bool corrupted() const override {
+            return poisoned;
+        }
+
+        void condemn() override {
+            *condemned = true;
+        }
+    };
+
+    bool condemned = false;
+    auto spy = std::make_unique<CorruptOnRead>();
+    spy->condemned = &condemned;
+    workspace.index_db = std::move(spy);
+
+    auto indexed = index_file(tmp, tmp.path("main.cpp"));
+    ASSERT_FALSE(indexed.data.empty());
+    indexer.merge(indexed.data.data(), indexed.data.size());
+
+    auto save = [&] {
+        auto body = [&]() -> kota::task<> {
+            co_await indexer.save();
+        };
+        auto task = body();
+        loop.schedule(task);
+        loop.run();
+    };
+    save();
+
+    auto path_id = workspace.path_pool.intern(indexed.tu_path);
+    ASSERT_TRUE(condemned);
+    ASSERT_TRUE(workspace.index_db != nullptr);
+    ASSERT_FALSE(workspace.shards.contains(path_id));
+    ASSERT_TRUE(indexer.pending_reason(path_id) == ReindexReason::ContentChanged);
+    ASSERT_EQ(indexer.last_save_shards(), 0u);
+
+    // The re-dirtied manifests, global and CDB snapshot re-persist into
+    // the fresh database.
+    save();
+    ASSERT_FALSE(indexer.has_unsaved_state());
+}
+
 TEST_CASE(WriteFailureStopsBatch) {
     TempDir tmp;
     open_store(tmp, workspace);
-    auto& storage = *workspace.index_storage;
+    auto& db = *workspace.index_db;
 
     // Wedge the manifest's destination with a non-empty directory so its
     // commit fails while the shard before it lands.
     tmp.touch("cache/cache/v1/index-manifest/k.idx/wedge");
 
-    auto failures = storage.write({
-        {index::IndexBlobKind::Shard,    "k",      "shard bytes"   },
-        {index::IndexBlobKind::Manifest, "k",      "manifest bytes"},
-        {index::IndexBlobKind::Global,   "global", "global bytes"  },
-    });
+    auto failures = db.write(
+        {
+            {index::IndexBlobKind::Shard,    "k",      "shard bytes"   },
+            {index::IndexBlobKind::Manifest, "k",      "manifest bytes"},
+            {index::IndexBlobKind::Global,   "global", "global bytes"  },
+    },
+        {});
 
     // The failure fails the rest of the batch: a global must never land
     // above a manifest that did not.
     ASSERT_EQ(failures.size(), 2u);
     ASSERT_EQ(failures[0], 1u);
     ASSERT_EQ(failures[1], 2u);
-    ASSERT_TRUE(storage.contains(index::IndexBlobKind::Shard, "k"));
-    ASSERT_FALSE(storage.contains(index::IndexBlobKind::Global, "global"));
+    ASSERT_TRUE(db.contains(index::IndexBlobKind::Shard, "k"));
+    ASSERT_FALSE(db.contains(index::IndexBlobKind::Global, "global"));
 }
 
 };  // TEST_SUITE(IndexerMerge)
@@ -943,8 +1305,17 @@ TEST_CASE(LoadHealsBrokenShard) {
     auto extra_it = f.workspace.shards.find(extra_id);
     ASSERT_TRUE(extra_it != f.workspace.shards.end());
     ASSERT_TRUE(extra_it->second.has_dead_variants());
+
+    // Load defers blob cleanup into the first save (no synchronous
+    // database commits on the startup event loop); the orphan dies there.
+    auto save_body = [&]() -> kota::task<> {
+        co_await f.indexer.save();
+    };
+    auto task = save_body();
+    f.loop.schedule(task);
+    f.loop.run();
     bool orphan_alive = false;
-    f.workspace.index_storage->for_each_key(index::IndexBlobKind::Shard, [&](llvm::StringRef key) {
+    f.workspace.index_db->for_each_key(index::IndexBlobKind::Shard, [&](llvm::StringRef key) {
         orphan_alive |= key == "deadbeefdeadbeef";
     });
     ASSERT_FALSE(orphan_alive);
@@ -986,13 +1357,13 @@ TEST_CASE(ReadOnlyLoadKeepsDisk) {
     // But every blob survives on disk — a server running concurrently may
     // still reference what this reader judged stale.
     bool header_alive = false, orphan_alive = false, manifest_alive = false;
-    f.workspace.index_storage->for_each_key(index::IndexBlobKind::Shard, [&](llvm::StringRef key) {
+    f.workspace.index_db->for_each_key(index::IndexBlobKind::Shard, [&](llvm::StringRef key) {
         header_alive |= key == header_key;
         orphan_alive |= key == "deadbeefdeadbeef";
     });
-    f.workspace.index_storage->for_each_key(
-        index::IndexBlobKind::Manifest,
-        [&](llvm::StringRef key) { manifest_alive |= key == manifest_key; });
+    f.workspace.index_db->for_each_key(index::IndexBlobKind::Manifest, [&](llvm::StringRef key) {
+        manifest_alive |= key == manifest_key;
+    });
     ASSERT_TRUE(header_alive);
     ASSERT_TRUE(orphan_alive);
     ASSERT_TRUE(manifest_alive);
@@ -1098,11 +1469,13 @@ TEST_CASE(LoadDropsNewerManifest) {
         index::serialize_manifest(raced, os);
         // Keyed by the interned (canonical) spelling, like save() itself:
         // on Windows the raw TempDir spelling hashes to a different key.
-        f.workspace.index_storage->write({
-            {index::IndexBlobKind::Manifest,
-             blob_key(f.workspace.path_pool.resolve(tu_id)),
-             std::move(bytes)}
-        });
+        f.workspace.index_db->write(
+            {
+                {index::IndexBlobKind::Manifest,
+                 blob_key(f.workspace.path_pool.resolve(tu_id)),
+                 std::move(bytes)}
+        },
+            {});
     }
 
     IndexerFixture f;
@@ -1140,11 +1513,13 @@ TEST_CASE(LoadDropsLostManifest) {
         std::string bytes;
         llvm::raw_string_ostream os(bytes);
         index::serialize_manifest(lost, os);
-        f.workspace.index_storage->write({
-            {index::IndexBlobKind::Manifest,
-             blob_key(f.workspace.path_pool.resolve(tu_id)),
-             std::move(bytes)}
-        });
+        f.workspace.index_db->write(
+            {
+                {index::IndexBlobKind::Manifest,
+                 blob_key(f.workspace.path_pool.resolve(tu_id)),
+                 std::move(bytes)}
+        },
+            {});
     }
 
     IndexerFixture f;
@@ -1191,24 +1566,33 @@ TEST_CASE(LoadRequeuesStaleManifest) {
         std::string bytes;
         llvm::raw_string_ostream os(bytes);
         index::serialize_manifest(stale, os);
-        f.workspace.index_storage->write({
-            {index::IndexBlobKind::Manifest, blob_key(header), std::move(bytes)}
-        });
+        f.workspace.index_db->write(
+            {
+                {index::IndexBlobKind::Manifest, blob_key(header), std::move(bytes)}
+        },
+            {});
     }
 
     IndexerFixture f;
     open_store(tmp, f.workspace);
     f.indexer.load();
 
-    // The unresolvable manifest is dropped from storage and its TU
-    // re-enqueued instead of losing its persisted index forever.
+    // The unresolvable manifest is dropped and its TU re-enqueued instead
+    // of losing its persisted index forever; the blob itself dies at the
+    // first save (load defers cleanup off the startup event loop).
     auto header_id = f.workspace.path_pool.intern(header);
     ASSERT_TRUE(f.indexer.pending_reason(header_id) == ReindexReason::ContentChanged);
     ASSERT_FALSE(f.workspace.project_index.manifests.contains(header_id));
+    auto save_body = [&]() -> kota::task<> {
+        co_await f.indexer.save();
+    };
+    auto task = save_body();
+    f.loop.schedule(task);
+    f.loop.run();
     bool stale_alive = false;
-    f.workspace.index_storage->for_each_key(
-        index::IndexBlobKind::Manifest,
-        [&](llvm::StringRef key) { stale_alive |= key == blob_key(header); });
+    f.workspace.index_db->for_each_key(index::IndexBlobKind::Manifest, [&](llvm::StringRef key) {
+        stale_alive |= key == blob_key(header);
+    });
     ASSERT_FALSE(stale_alive);
 
     // The TU whose manifest resolved is untouched.
@@ -1217,6 +1601,253 @@ TEST_CASE(LoadRequeuesStaleManifest) {
     ASSERT_FALSE(f.indexer.pending_reason(tu_id).has_value());
 }
 
+TEST_CASE(DeferredSweepYieldsToFreshWrite) {
+    TempDir tmp;
+    tmp.touch("main.cpp", "int sweep_target() { return 1; }\n");
+    auto src = tmp.path("main.cpp");
+
+    {
+        IndexerFixture f;
+        open_store(tmp, f.workspace);
+        auto indexed = index_file(tmp, src);
+        ASSERT_FALSE(indexed.data.empty());
+        f.indexer.merge(indexed.data.data(), indexed.data.size());
+        f.save();
+        // The indexer keys blobs by the pool-canonical path, which need
+        // not equal the raw temp path byte-for-byte (Windows 8.3 names).
+        auto key =
+            blob_key(f.workspace.path_pool.resolve(f.workspace.path_pool.intern(indexed.tu_path)));
+        // Replace the persisted manifest with an unresolvable one: the
+        // next load sweeps it — deferred into the first save — and the
+        // TU's shard turns orphan, deferred too.
+        index::TUManifest stale;
+        stale.tu_fv = 999999;
+        stale.nodes.push_back({.fv = 9999});
+        std::string bytes;
+        llvm::raw_string_ostream os(bytes);
+        index::serialize_manifest(stale, os);
+        f.workspace.index_db->write(
+            {
+                {index::IndexBlobKind::Manifest, key, std::move(bytes)}
+        },
+            {});
+    }
+
+    IndexerFixture f;
+    open_store(tmp, f.workspace);
+    ASSERT_TRUE(f.indexer.load());
+
+    // The swept TU re-indexes before the first save, so that save both
+    // re-writes and (deferred) removes the same keys — the fresh write
+    // must win or the file never persists again.
+    auto indexed = index_file(tmp, src);
+    ASSERT_FALSE(indexed.data.empty());
+    f.indexer.merge(indexed.data.data(), indexed.data.size());
+    f.save();
+
+    auto key =
+        blob_key(f.workspace.path_pool.resolve(f.workspace.path_pool.intern(indexed.tu_path)));
+    bool manifest_alive = false;
+    f.workspace.index_db->for_each_key(index::IndexBlobKind::Manifest,
+                                       [&](llvm::StringRef k) { manifest_alive |= k == key; });
+    bool shard_alive = false;
+    f.workspace.index_db->for_each_key(index::IndexBlobKind::Shard,
+                                       [&](llvm::StringRef k) { shard_alive |= k == key; });
+    ASSERT_TRUE(manifest_alive);
+    ASSERT_TRUE(shard_alive);
+}
+
+TEST_CASE(LmdbLoadServesAcrossSaves) {
+    TempDir tmp;
+    tmp.touch("main.cpp", "int lmdb_value() { return 1; }\n");
+    auto src = tmp.path("main.cpp");
+
+    auto open_lmdb = [&](Workspace& workspace) {
+        auto store = CacheStore::open(tmp.path("cache"), 1);
+        ASSERT_TRUE(store.has_value());
+        workspace.store.emplace(std::move(*store));
+        workspace.index_db = index::open_lmdb_database(*workspace.store);
+        ASSERT_TRUE(workspace.index_db != nullptr);
+    };
+
+    {
+        IndexerFixture f;
+        open_lmdb(f.workspace);
+        auto indexed = index_file(tmp, src);
+        ASSERT_FALSE(indexed.data.empty());
+        f.indexer.merge(indexed.data.data(), indexed.data.size());
+        f.save();
+    }
+
+    IndexerFixture f;
+    open_lmdb(f.workspace);
+    ASSERT_TRUE(f.indexer.load());
+    auto path_id = f.workspace.path_pool.intern(src);
+    ASSERT_TRUE(f.workspace.shards.contains(path_id));
+
+    // The loaded shard borrows the open-time snapshot. A save that commits
+    // anything advances and retires it — the shard must come out rebound
+    // onto the fresh snapshot, still serving.
+    tmp.touch("other.cpp", "int other_value() { return 2; }\n");
+    auto other = index_file(tmp, tmp.path("other.cpp"));
+    ASSERT_FALSE(other.data.empty());
+    f.indexer.merge(other.data.data(), other.data.size());
+    f.save();
+
+    auto it = f.workspace.shards.find(path_id);
+    ASSERT_TRUE(it != f.workspace.shards.end());
+    ASSERT_TRUE(it->second.loaded());
+    ASSERT_EQ(it->second.content_hash(), llvm::xxh3_64bits("int lmdb_value() { return 1; }\n"));
+    ASSERT_FALSE(it->second.bytes().empty());
+}
+
+TEST_CASE(CorruptGlobalCondemnsDatabase) {
+    // Page corruption under the global blob: read fails, contains() still
+    // says present, corrupted() confirms. load must condemn the database
+    // (deleted on close) and continue on a fresh empty one instead of
+    // parking in the disabled-persistence limbo forever.
+    struct CorruptGlobal final : index::BlobDatabase {
+        bool* condemned;
+
+        index::ReadBlob read(index::IndexBlobKind, llvm::StringRef) override {
+            return {};
+        }
+
+        bool contains(index::IndexBlobKind, llvm::StringRef) override {
+            return true;
+        }
+
+        llvm::SmallVector<std::size_t> write(llvm::ArrayRef<Blob>,
+                                             llvm::ArrayRef<index::BlobKey>) override {
+            return {};
+        }
+
+        void for_each_key(index::IndexBlobKind,
+                          llvm::function_ref<void(llvm::StringRef)>) override {}
+
+        std::expected<std::uint64_t, std::string> advance_read_snapshot() override {
+            return 0;
+        }
+
+        void retire_old_snapshot() override {}
+
+        std::expected<bool, std::string> grow() override {
+            return false;
+        }
+
+        bool corrupted() const override {
+            return true;
+        }
+
+        void condemn() override {
+            *condemned = true;
+        }
+    };
+
+    TempDir tmp;
+    IndexerFixture f;
+    open_store(tmp, f.workspace);
+    bool condemned = false;
+    auto spy = std::make_unique<CorruptGlobal>();
+    spy->condemned = &condemned;
+    f.workspace.index_db = std::move(spy);
+
+    ASSERT_TRUE(f.indexer.load());
+    ASSERT_TRUE(condemned);
+    ASSERT_TRUE(f.workspace.index_db != nullptr);
+}
+
+TEST_CASE(CorruptShardCondemnsDatabase) {
+    TempDir tmp;
+    tmp.touch("main.cpp", "int gone() { return 1; }\n");
+    auto src = tmp.path("main.cpp");
+
+    std::string tu_path;
+    {
+        IndexerFixture f;
+        open_store(tmp, f.workspace);
+        auto indexed = index_file(tmp, src);
+        ASSERT_FALSE(indexed.data.empty());
+        f.indexer.merge(indexed.data.data(), indexed.data.size());
+        tu_path = indexed.tu_path;
+        f.save();
+    }
+
+    // Corruption latched past the global anchor: the shard read poisons
+    // the latch while the global stays readable. load must condemn here
+    // too, and unwind everything it adopted — the views would otherwise
+    // borrow from the condemned environment.
+    struct CorruptShard final : index::BlobDatabase {
+        std::unique_ptr<index::BlobDatabase> real;
+        bool* condemned;
+        bool poisoned = false;
+
+        index::ReadBlob read(index::IndexBlobKind kind, llvm::StringRef key) override {
+            if(kind == index::IndexBlobKind::Shard) {
+                poisoned = true;
+                return {};
+            }
+            return real->read(kind, key);
+        }
+
+        bool contains(index::IndexBlobKind kind, llvm::StringRef key) override {
+            return real->contains(kind, key);
+        }
+
+        llvm::SmallVector<std::size_t> write(llvm::ArrayRef<Blob> puts,
+                                             llvm::ArrayRef<index::BlobKey> removes) override {
+            return real->write(puts, removes);
+        }
+
+        void for_each_key(index::IndexBlobKind kind,
+                          llvm::function_ref<void(llvm::StringRef)> fn) override {
+            real->for_each_key(kind, fn);
+        }
+
+        std::expected<std::uint64_t, std::string> advance_read_snapshot() override {
+            return 0;
+        }
+
+        void retire_old_snapshot() override {}
+
+        std::expected<bool, std::string> grow() override {
+            return false;
+        }
+
+        bool corrupted() const override {
+            return poisoned;
+        }
+
+        void condemn() override {
+            *condemned = true;
+        }
+    };
+
+    IndexerFixture f;
+    open_store(tmp, f.workspace);
+    bool condemned = false;
+    auto wrapper = std::make_unique<CorruptShard>();
+    wrapper->real = std::move(f.workspace.index_db);
+    wrapper->condemned = &condemned;
+    f.workspace.index_db = std::move(wrapper);
+
+    ASSERT_TRUE(f.indexer.load());
+    ASSERT_TRUE(condemned);
+    ASSERT_TRUE(f.workspace.shards.empty());
+    ASSERT_TRUE(f.workspace.project_index.symbols.empty());
+
+    // The TU has no CDB entry, so nothing else records the debt: it is
+    // re-enqueued before the adopted state unwinds, and the fresh
+    // database's first save persists it as standalone debt.
+    ASSERT_TRUE(f.indexer.pending_reason(f.workspace.path_pool.intern(tu_path)) ==
+                ReindexReason::ContentChanged);
+    ASSERT_TRUE(f.workspace.index_db != nullptr);
+    f.save();
+    auto snapshot = f.workspace.index_db->read(index::IndexBlobKind::CDB, "cdb");
+    ASSERT_TRUE(snapshot);
+    ASSERT_TRUE(snapshot.buffer->getBuffer().contains("main.cpp"));
+}
+
 TEST_CASE(UnreadableGlobalPreserved) {
     TempDir tmp;
     tmp.touch("main.cpp", "int keep() { return 1; }\n");
@@ -1236,41 +1867,47 @@ TEST_CASE(UnreadableGlobalPreserved) {
     // fresh lineage saved over the unread one could alias its fv ids and
     // generation stamps. The session must run memory-only and leave every
     // blob for the next start.
-    struct UnreadableGlobal final : index::IndexStorage {
-        std::unique_ptr<index::IndexStorage> real;
+    struct UnreadableGlobal final : index::BlobDatabase {
+        std::unique_ptr<index::BlobDatabase> real;
 
-        std::unique_ptr<llvm::MemoryBuffer> read(index::IndexBlobKind kind,
-                                                 llvm::StringRef key) override {
-            return kind == index::IndexBlobKind::Global ? nullptr : real->read(kind, key);
+        index::ReadBlob read(index::IndexBlobKind kind, llvm::StringRef key) override {
+            return kind == index::IndexBlobKind::Global ? index::ReadBlob{} : real->read(kind, key);
         }
 
         bool contains(index::IndexBlobKind kind, llvm::StringRef key) override {
             return real->contains(kind, key);
         }
 
-        llvm::SmallVector<std::size_t> write(llvm::ArrayRef<Blob> batch) override {
-            return real->write(batch);
-        }
-
-        void remove(index::IndexBlobKind kind, llvm::StringRef key) override {
-            real->remove(kind, key);
+        llvm::SmallVector<std::size_t> write(llvm::ArrayRef<Blob> puts,
+                                             llvm::ArrayRef<index::BlobKey> removes) override {
+            return real->write(puts, removes);
         }
 
         void for_each_key(index::IndexBlobKind kind,
                           llvm::function_ref<void(llvm::StringRef)> fn) override {
             real->for_each_key(kind, fn);
         }
+
+        std::expected<std::uint64_t, std::string> advance_read_snapshot() override {
+            return 0;
+        }
+
+        void retire_old_snapshot() override {}
+
+        std::expected<bool, std::string> grow() override {
+            return false;
+        }
     };
 
     {
         IndexerFixture f;
         open_store(tmp, f.workspace);
         auto wrapper = std::make_unique<UnreadableGlobal>();
-        wrapper->real = std::move(f.workspace.index_storage);
-        f.workspace.index_storage = std::move(wrapper);
+        wrapper->real = std::move(f.workspace.index_db);
+        f.workspace.index_db = std::move(wrapper);
         f.indexer.load();
         ASSERT_TRUE(f.workspace.project_index.manifests.empty());
-        ASSERT_TRUE(f.workspace.index_storage == nullptr);
+        ASSERT_TRUE(f.workspace.index_db == nullptr);
     }
 
     IndexerFixture f;
@@ -1590,20 +2227,19 @@ TEST_CASE(UnreachableHostRebuilds) {
     ASSERT_TRUE(f.indexer.pending_reason(header_id) == ReindexReason::ContentChanged);
 }
 
-TEST_CASE(CdbWriteFailureRetried) {
+TEST_CASE(CDBWriteFailureRetried) {
     TempDir tmp;
     tmp.touch("main.cpp", "int value() { return 1; }\n");
     auto src = tmp.path("main.cpp");
 
     // A storage that fails only the CDB snapshot blob, with everything
     // else landing normally.
-    struct CdbFailingStorage final : index::IndexStorage {
-        std::unique_ptr<index::IndexStorage> real;
+    struct CDBFailingStorage final : index::BlobDatabase {
+        std::unique_ptr<index::BlobDatabase> real;
         bool fail_cdb = true;
         llvm::SmallVector<index::IndexBlobKind> written;
 
-        std::unique_ptr<llvm::MemoryBuffer> read(index::IndexBlobKind kind,
-                                                 llvm::StringRef key) override {
+        index::ReadBlob read(index::IndexBlobKind kind, llvm::StringRef key) override {
             return real->read(kind, key);
         }
 
@@ -1611,36 +2247,44 @@ TEST_CASE(CdbWriteFailureRetried) {
             return real->contains(kind, key);
         }
 
-        llvm::SmallVector<std::size_t> write(llvm::ArrayRef<Blob> batch) override {
+        llvm::SmallVector<std::size_t> write(llvm::ArrayRef<Blob> puts,
+                                             llvm::ArrayRef<index::BlobKey> removes) override {
             llvm::SmallVector<std::size_t> failed;
-            for(std::size_t i = 0; i < batch.size(); i += 1) {
-                written.push_back(batch[i].kind);
-                if(fail_cdb && batch[i].kind == index::IndexBlobKind::Cdb) {
+            for(std::size_t i = 0; i < puts.size(); i += 1) {
+                written.push_back(puts[i].kind);
+                if(fail_cdb && puts[i].kind == index::IndexBlobKind::CDB) {
                     failed.push_back(i);
-                } else if(!real->write(llvm::ArrayRef(batch[i])).empty()) {
+                } else if(!real->write(llvm::ArrayRef(puts[i]), {}).empty()) {
                     failed.push_back(i);
                 }
             }
+            real->write({}, removes);
             return failed;
         }
 
-        void remove(index::IndexBlobKind kind, llvm::StringRef key) override {
-            real->remove(kind, key);
-        }
-
         void for_each_key(index::IndexBlobKind kind,
                           llvm::function_ref<void(llvm::StringRef)> fn) override {
             real->for_each_key(kind, fn);
         }
+
+        std::expected<std::uint64_t, std::string> advance_read_snapshot() override {
+            return 0;
+        }
+
+        void retire_old_snapshot() override {}
+
+        std::expected<bool, std::string> grow() override {
+            return false;
+        }
     };
 
     IndexerFixture f;
     open_store(tmp, f.workspace);
     f.workspace.cdb.add_command(tmp.root, src, llvm::StringRef("clang++ -c main.cpp"));
-    auto failing = std::make_unique<CdbFailingStorage>();
-    failing->real = std::move(f.workspace.index_storage);
+    auto failing = std::make_unique<CDBFailingStorage>();
+    failing->real = std::move(f.workspace.index_db);
     auto* storage = failing.get();
-    f.workspace.index_storage = std::move(failing);
+    f.workspace.index_db = std::move(failing);
 
     auto indexed = index_file(tmp, src);
     ASSERT_FALSE(indexed.data.empty());
@@ -1648,8 +2292,8 @@ TEST_CASE(CdbWriteFailureRetried) {
     f.save();
 
     // The snapshot rides the batch behind the index state it describes.
-    ASSERT_EQ(int(storage->written.back()), int(index::IndexBlobKind::Cdb));
-    ASSERT_FALSE(storage->real->contains(index::IndexBlobKind::Cdb, "cdb"));
+    ASSERT_EQ(int(storage->written.back()), int(index::IndexBlobKind::CDB));
+    ASSERT_FALSE(storage->real->contains(index::IndexBlobKind::CDB, "cdb"));
 
     // Nothing else is dirty any more, yet the failed snapshot alone must
     // drive the next save until it lands — and until it does, the state
@@ -1657,7 +2301,7 @@ TEST_CASE(CdbWriteFailureRetried) {
     ASSERT_TRUE(f.indexer.has_unsaved_state());
     storage->fail_cdb = false;
     f.save();
-    ASSERT_TRUE(storage->real->contains(index::IndexBlobKind::Cdb, "cdb"));
+    ASSERT_TRUE(storage->real->contains(index::IndexBlobKind::CDB, "cdb"));
     ASSERT_FALSE(f.indexer.has_unsaved_state());
 }
 
@@ -1676,7 +2320,10 @@ TEST_CASE(MissingSnapshotRewritten) {
         f.save();
         // The global landed but the final CDB write never did: the rest of
         // the index is intact.
-        f.workspace.index_storage->remove(index::IndexBlobKind::Cdb, "cdb");
+        f.workspace.index_db->write(
+            {
+        },
+            {{index::IndexBlobKind::CDB, "cdb"}});
     }
 
     // A rerun that dirties nothing must still recreate the baseline —
@@ -1687,7 +2334,7 @@ TEST_CASE(MissingSnapshotRewritten) {
     f.indexer.load();
     ASSERT_TRUE(f.indexer.has_unsaved_state());
     f.save();
-    ASSERT_TRUE(f.workspace.index_storage->contains(index::IndexBlobKind::Cdb, "cdb"));
+    ASSERT_TRUE(f.workspace.index_db->contains(index::IndexBlobKind::CDB, "cdb"));
     ASSERT_FALSE(f.indexer.has_unsaved_state());
 }
 
diff --git a/tools/bench/bench.ts b/tools/bench/bench.ts
index 165a8b4b1..4b629248c 100644
--- a/tools/bench/bench.ts
+++ b/tools/bench/bench.ts
@@ -238,7 +238,7 @@ function derivePosition(file: string): { line: number; character: number } {
 
 /// Locate the CDB the way clice does: workspace root first, then any
 /// first-level subdirectory (e.g. build/).
-function findCdb(workspace: string): string {
+function findCDB(workspace: string): string {
     const candidates = [workspace];
     for (const entry of fs.readdirSync(workspace, { withFileTypes: true })) {
         if (entry.isDirectory()) {
@@ -254,7 +254,7 @@ function findCdb(workspace: string): string {
     fail(`no compile_commands.json under ${workspace} or its direct subdirectories`);
 }
 
-function firstCdbEntry(cdbPath: string): string {
+function firstCDBEntry(cdbPath: string): string {
     const entries = JSON.parse(fs.readFileSync(cdbPath, "utf8")) as {
         file: string;
         directory?: string;
@@ -553,9 +553,9 @@ function printScenario(name: string, result: ScenarioResult): void {
 
 async function main(): Promise<void> {
     const opts = parseOptions();
-    const cdb = findCdb(opts.workspace);
+    const cdb = findCDB(opts.workspace);
     opts.cdbDir = path.dirname(cdb);
-    const file = opts.file !== null ? path.resolve(opts.workspace, opts.file) : firstCdbEntry(cdb);
+    const file = opts.file !== null ? path.resolve(opts.workspace, opts.file) : firstCDBEntry(cdb);
     const cpus = os.cpus();
 
     const result: BenchResult = {
diff --git a/tools/client/workspace.ts b/tools/client/workspace.ts
index 7edf1864c..d0f1533dd 100644
--- a/tools/client/workspace.ts
+++ b/tools/client/workspace.ts
@@ -10,7 +10,7 @@ import { generateCDB } from "../compile_commands.ts";
 
 /// Versioned root of the unified cache store; bump together with
 /// cache_format_version in src/server/state/workspace.h.
-const CACHE_ROOT = path.join(".clice", "cache", "v6");
+const CACHE_ROOT = path.join(".clice", "cache", "v7");
 
 /// The harness-wide canonical URI spelling: percent-decoded. vscode-uri
 /// encodes the drive colon (file:///c%3A/...) while the server emits it