From f4e14446b1695d08608a55e0d237c2fd256cc829 Mon Sep 17 00:00:00 2001 From: dajiaohuang <108231307+dajiaohuang@users.noreply.github.com> Date: Thu, 1 Oct 2026 22:32:07 +0800 Subject: [PATCH 1/6] Firestore: preserve batch IDs after queue restart --- Firestore/CHANGELOG.md | 2 ++ .../core/src/local/leveldb_index_manager.cc | 20 +++++++++++++ .../core/src/local/leveldb_index_manager.h | 3 ++ .../core/src/local/leveldb_mutation_queue.cc | 19 ++++++++++++- .../core/src/local/leveldb_mutation_queue.h | 8 +++--- .../unit/local/leveldb_mutation_queue_test.cc | 28 +++++++++++++++++++ 6 files changed, 75 insertions(+), 5 deletions(-) diff --git a/Firestore/CHANGELOG.md b/Firestore/CHANGELOG.md index ff26de65273..202f908a2b5 100644 --- a/Firestore/CHANGELOG.md +++ b/Firestore/CHANGELOG.md @@ -1,4 +1,6 @@ # 13.0.0 +- [fixed] Preserve pending document overlays for indexed queries after the + mutation queue becomes empty and is recreated. (#16742) - [changed] Dropped C++14 support; Firestore now requires C++17. - [changed] Update gRPC dependency to 1.83.1. - [changed] Update Abseil dependency to 1.20250512.2. diff --git a/Firestore/core/src/local/leveldb_index_manager.cc b/Firestore/core/src/local/leveldb_index_manager.cc index e62a0da7bd7..81740618d19 100644 --- a/Firestore/core/src/local/leveldb_index_manager.cc +++ b/Firestore/core/src/local/leveldb_index_manager.cc @@ -440,6 +440,26 @@ std::vector LevelDbIndexManager::GetFieldIndexes() const { return result; } +model::BatchId LevelDbIndexManager::GetHighestBatchIdAcrossUsers() const { + model::BatchId highest_batch_id = model::IndexOffset::InitialLargestBatchId(); + const auto state_key_prefix = LevelDbIndexStateKey::KeyPrefix(); + auto state_iter = db_->current_transaction()->NewIterator(); + LevelDbIndexStateKey state_key; + for (state_iter->Seek(state_key_prefix); + state_iter->Valid() && + absl::StartsWith(state_iter->key(), state_key_prefix); + state_iter->Next()) { + if (!state_key.Decode(state_iter->key())) { + break; + } + highest_batch_id = std::max( + highest_batch_id, DecodeIndexState(state_iter->value()) + .index_offset() + .largest_batch_id()); + } + return highest_batch_id; +} + std::optional LevelDbIndexManager::GetFieldIndex( const core::Target& target) const { HARD_ASSERT(started_, "IndexManager not started"); diff --git a/Firestore/core/src/local/leveldb_index_manager.h b/Firestore/core/src/local/leveldb_index_manager.h index 0b9843707fc..b9a0cac78f6 100644 --- a/Firestore/core/src/local/leveldb_index_manager.h +++ b/Firestore/core/src/local/leveldb_index_manager.h @@ -69,6 +69,9 @@ class LevelDbIndexManager : public IndexManager { std::vector GetFieldIndexes() const override; + /** Returns the greatest persisted index batch ID across all users. */ + model::BatchId GetHighestBatchIdAcrossUsers() const; + void DeleteAllFieldIndexes() override; void CreateTargetIndexes(const core::Target& target) override; diff --git a/Firestore/core/src/local/leveldb_mutation_queue.cc b/Firestore/core/src/local/leveldb_mutation_queue.cc index 01e36f415ae..683e5c997ae 100644 --- a/Firestore/core/src/local/leveldb_mutation_queue.cc +++ b/Firestore/core/src/local/leveldb_mutation_queue.cc @@ -16,6 +16,8 @@ #include "Firestore/core/src/local/leveldb_mutation_queue.h" +#include +#include #include #include #include @@ -32,6 +34,7 @@ #include "Firestore/core/src/model/resource_path.h" #include "Firestore/core/src/nanopb/nanopb_util.h" #include "Firestore/core/src/nanopb/reader.h" +#include "Firestore/core/src/util/hard_assert.h" #include "Firestore/core/src/util/string_util.h" #include "Firestore/core/src/util/to_string.h" #include "absl/strings/match.h" @@ -122,12 +125,14 @@ BatchId LoadNextBatchIdFromDb(DB* db) { } } + HARD_ASSERT(max_batch_id < std::numeric_limits::max(), + "Mutation batch ID space exhausted"); return max_batch_id + 1; } LevelDbMutationQueue::LevelDbMutationQueue(const User& user, LevelDbPersistence* db, - IndexManager* index_manager, + LevelDbIndexManager* index_manager, LocalSerializer* serializer) : db_(NOT_NULL(db)), index_manager_(NOT_NULL(index_manager)), @@ -137,6 +142,16 @@ LevelDbMutationQueue::LevelDbMutationQueue(const User& user, void LevelDbMutationQueue::Start() { next_batch_id_ = LoadNextBatchIdFromDb(db_->ptr()); + // Batch IDs are global. Index offsets are persisted per user and can outlive + // all acknowledged batches in the mutation queue, so include their global + // high-water mark when restoring the next ID. + const BatchId highest_indexed_batch_id = + index_manager_->GetHighestBatchIdAcrossUsers(); + if (highest_indexed_batch_id >= 0) { + HARD_ASSERT(highest_indexed_batch_id < std::numeric_limits::max(), + "Mutation batch ID space exhausted"); + next_batch_id_ = std::max(next_batch_id_, highest_indexed_batch_id + 1); + } metadata_ = MetadataForKey(mutation_queue_key()); } @@ -162,6 +177,8 @@ MutationBatch LevelDbMutationQueue::AddMutationBatch( const Timestamp& local_write_time, std::vector&& base_mutations, std::vector&& mutations) { + HARD_ASSERT(next_batch_id_ < std::numeric_limits::max(), + "Mutation batch ID space exhausted"); BatchId batch_id = next_batch_id_; next_batch_id_++; diff --git a/Firestore/core/src/local/leveldb_mutation_queue.h b/Firestore/core/src/local/leveldb_mutation_queue.h index 3c835b6122e..5256fea5765 100644 --- a/Firestore/core/src/local/leveldb_mutation_queue.h +++ b/Firestore/core/src/local/leveldb_mutation_queue.h @@ -45,8 +45,8 @@ class LevelDbPersistence; class LocalSerializer; /** - * Returns one larger than the largest batch ID that has been stored. If there - * are no mutations returns 0. Note that batch IDs are global. + * Returns one larger than the largest batch ID stored in the mutation table. + * Returns 1 if there are no mutations. Batch IDs are global. */ model::BatchId LoadNextBatchIdFromDb(leveldb::DB* db); @@ -54,7 +54,7 @@ class LevelDbMutationQueue : public MutationQueue { public: LevelDbMutationQueue(const credentials::User& user, LevelDbPersistence* db, - IndexManager* index_manager, + LevelDbIndexManager* index_manager, LocalSerializer* serializer); void Start() override; @@ -116,7 +116,7 @@ class LevelDbMutationQueue : public MutationQueue { // The LevelDbMutationQueue instance is owned by LevelDbPersistence. LevelDbPersistence* db_; - IndexManager* index_manager_; + LevelDbIndexManager* index_manager_; // Owned by LevelDbPersistence. LocalSerializer* serializer_ = nullptr; diff --git a/Firestore/core/test/unit/local/leveldb_mutation_queue_test.cc b/Firestore/core/test/unit/local/leveldb_mutation_queue_test.cc index 1605b00c151..ade05e1d3ec 100644 --- a/Firestore/core/test/unit/local/leveldb_mutation_queue_test.cc +++ b/Firestore/core/test/unit/local/leveldb_mutation_queue_test.cc @@ -25,6 +25,7 @@ #include "Firestore/core/src/local/leveldb_key.h" #include "Firestore/core/src/local/leveldb_persistence.h" #include "Firestore/core/src/local/reference_set.h" +#include "Firestore/core/src/model/field_index.h" #include "Firestore/core/src/nanopb/byte_string.h" #include "Firestore/core/src/nanopb/message.h" #include "Firestore/core/src/nanopb/reader.h" @@ -33,6 +34,7 @@ #include "Firestore/core/test/unit/local/mutation_queue_test.h" #include "Firestore/core/test/unit/local/persistence_testing.h" #include "Firestore/core/test/unit/testutil/status_testing.h" +#include "Firestore/core/test/unit/testutil/testutil.h" #include "absl/strings/string_view.h" #include "gtest/gtest.h" #include "leveldb/db.h" @@ -52,6 +54,7 @@ using nanopb::ByteString; using nanopb::Message; using nanopb::StringReader; using nanopb::StringWriter; +using testutil::MakeFieldIndex; using util::OrderedCode; // A dummy mutation value, useful for testing code that's known to examine only @@ -158,6 +161,31 @@ TEST_F(LevelDbMutationQueueTest, LoadNextBatchIdOnlyFindsMutations) { ASSERT_EQ(LoadNextBatchIdFromDb(db_), 4); } +TEST_F(LevelDbMutationQueueTest, + LoadNextBatchIdIncludesLargestIndexedBatchIdAcrossUsers) { + persistence_->Run("AddIndexAndRestartMutationQueue", [&] { + auto* index_manager = persistence_->GetIndexManager(User("user")); + index_manager->Start(); + index_manager->AddFieldIndex( + MakeFieldIndex("coll", "a", model::Segment::Kind::kAscending)); + index_manager->UpdateCollectionGroup( + "coll", model::IndexOffset(model::SnapshotVersion::None(), + model::DocumentKey::Empty(), 20)); + + auto* other_user_index_manager = + persistence_->GetIndexManager(User("other-user")); + other_user_index_manager->Start(); + other_user_index_manager->UpdateCollectionGroup( + "coll", model::IndexOffset(model::SnapshotVersion::None(), + model::DocumentKey::Empty(), 40)); + + // Re-starting models recreating the queue after the database has been + // reopened with no outstanding mutation batches. + mutation_queue_->Start(); + ASSERT_EQ(AddMutationBatch().batch_id(), 41); + }); +} + TEST_F(LevelDbMutationQueueTest, EmptyProtoCanBeUpgraded) { // An empty protocol buffer serializes to a zero-length byte buffer. google_protobuf_Empty empty{}; From 6641b7b89e4f7af1224f3372a3f10b8e9521184d Mon Sep 17 00:00:00 2001 From: dajiaohuang <108231307+dajiaohuang@users.noreply.github.com> Date: Fri, 2 Oct 2026 15:20:21 +0800 Subject: [PATCH 2/6] Format Firestore index manager --- Firestore/core/src/local/leveldb_index_manager.cc | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/Firestore/core/src/local/leveldb_index_manager.cc b/Firestore/core/src/local/leveldb_index_manager.cc index 81740618d19..b8eb0d36ff8 100644 --- a/Firestore/core/src/local/leveldb_index_manager.cc +++ b/Firestore/core/src/local/leveldb_index_manager.cc @@ -452,10 +452,10 @@ model::BatchId LevelDbIndexManager::GetHighestBatchIdAcrossUsers() const { if (!state_key.Decode(state_iter->key())) { break; } - highest_batch_id = std::max( - highest_batch_id, DecodeIndexState(state_iter->value()) - .index_offset() - .largest_batch_id()); + highest_batch_id = + std::max(highest_batch_id, DecodeIndexState(state_iter->value()) + .index_offset() + .largest_batch_id()); } return highest_batch_id; } From 7f708d43f351b761a67e70fd33b15520903224a7 Mon Sep 17 00:00:00 2001 From: dajiaohuang <108231307+dajiaohuang@users.noreply.github.com> Date: Sat, 3 Oct 2026 13:03:45 +0800 Subject: [PATCH 3/6] Address index batch restart review and add LocalStore regression --- Firestore/CHANGELOG.md | 6 ++- .../core/src/local/leveldb_mutation_queue.cc | 11 ++--- .../unit/local/leveldb_local_store_test.cc | 49 +++++++++++++++++++ .../unit/local/leveldb_mutation_queue_test.cc | 4 +- 4 files changed, 60 insertions(+), 10 deletions(-) diff --git a/Firestore/CHANGELOG.md b/Firestore/CHANGELOG.md index 202f908a2b5..4c0cace71bb 100644 --- a/Firestore/CHANGELOG.md +++ b/Firestore/CHANGELOG.md @@ -1,6 +1,8 @@ -# 13.0.0 +# Unreleased - [fixed] Preserve pending document overlays for indexed queries after the - mutation queue becomes empty and is recreated. (#16742) + mutation queue becomes empty and is recreated. (#16751) + +# 13.0.0 - [changed] Dropped C++14 support; Firestore now requires C++17. - [changed] Update gRPC dependency to 1.83.1. - [changed] Update Abseil dependency to 1.20250512.2. diff --git a/Firestore/core/src/local/leveldb_mutation_queue.cc b/Firestore/core/src/local/leveldb_mutation_queue.cc index 683e5c997ae..15cce3a05ab 100644 --- a/Firestore/core/src/local/leveldb_mutation_queue.cc +++ b/Firestore/core/src/local/leveldb_mutation_queue.cc @@ -125,8 +125,6 @@ BatchId LoadNextBatchIdFromDb(DB* db) { } } - HARD_ASSERT(max_batch_id < std::numeric_limits::max(), - "Mutation batch ID space exhausted"); return max_batch_id + 1; } @@ -148,9 +146,10 @@ void LevelDbMutationQueue::Start() { const BatchId highest_indexed_batch_id = index_manager_->GetHighestBatchIdAcrossUsers(); if (highest_indexed_batch_id >= 0) { - HARD_ASSERT(highest_indexed_batch_id < std::numeric_limits::max(), - "Mutation batch ID space exhausted"); - next_batch_id_ = std::max(next_batch_id_, highest_indexed_batch_id + 1); + // Do not fail startup on an unexpectedly large persisted index offset. + const BatchId indexed_batch_id = std::min( + highest_indexed_batch_id, std::numeric_limits::max() - 1); + next_batch_id_ = std::max(next_batch_id_, indexed_batch_id + 1); } metadata_ = MetadataForKey(mutation_queue_key()); } @@ -177,8 +176,6 @@ MutationBatch LevelDbMutationQueue::AddMutationBatch( const Timestamp& local_write_time, std::vector&& base_mutations, std::vector&& mutations) { - HARD_ASSERT(next_batch_id_ < std::numeric_limits::max(), - "Mutation batch ID space exhausted"); BatchId batch_id = next_batch_id_; next_batch_id_++; diff --git a/Firestore/core/test/unit/local/leveldb_local_store_test.cc b/Firestore/core/test/unit/local/leveldb_local_store_test.cc index bd6210f44e6..33b6c3ba7b2 100644 --- a/Firestore/core/test/unit/local/leveldb_local_store_test.cc +++ b/Firestore/core/test/unit/local/leveldb_local_store_test.cc @@ -18,8 +18,11 @@ #include "Firestore/core/src/core/filter.h" #include "Firestore/core/src/core/query.h" #include "Firestore/core/src/local/leveldb_persistence.h" +#include "Firestore/core/src/local/local_write_result.h" +#include "Firestore/core/src/local/target_data.h" #include "Firestore/core/src/model/delete_mutation.h" #include "Firestore/core/src/model/field_index.h" +#include "Firestore/core/src/model/mutation_batch_result.h" #include "Firestore/core/src/model/set_mutation.h" #include "Firestore/core/src/remote/remote_event.h" #include "Firestore/core/test/unit/local/local_store_test.h" @@ -221,6 +224,52 @@ TEST_F(LevelDbLocalStoreTest, UsesIndexes) { FSTAssertQueryReturned("coll/a"); } +TEST(LevelDbLocalStoreRestartTest, IndexedCacheQueryIncludesNewPendingWrite) { + const auto directory = LevelDbDir(); + const auto user = credentials::User::Unauthenticated(); + const core::Query query = + testutil::Query("coll").AddingFilter(Filter("matches", "==", true)); + + { + auto persistence = LevelDbPersistenceForTesting(directory); + CountingQueryEngine query_engine; + LocalStore store(persistence.get(), &query_engine, user); + store.Start(); + store.ConfigureFieldIndexes( + {MakeFieldIndex("coll", 0, FieldIndex::InitialState(), "matches", + model::Segment::Kind::kAscending)}); + store.AllocateTarget(core::QueryOrPipeline(query).ToTargetOrPipeline()); + + auto mutation = SetMutation("coll/a", Map("matches", true)); + const auto write = store.WriteLocally({mutation}); + store.Backfill(); + ASSERT_GT(store.GetFieldIndexes()[0] + .index_state() + .index_offset() + .largest_batch_id(), + 0); + + model::MutationBatch batch(write.batch_id(), Timestamp::Now(), {}, + {std::move(mutation)}); + std::vector results; + results.emplace_back(Version(10), Array()); + store.AcknowledgeBatch( + model::MutationBatchResult(batch, Version(10), std::move(results), {})); + } + + // Reopen the same database after acknowledging every mutation. The persisted + // index offset must not hide the first pending write from the new queue. + auto persistence = LevelDbPersistenceForTesting(directory); + CountingQueryEngine query_engine; + LocalStore store(persistence.get(), &query_engine, user); + store.Start(); + store.WriteLocally({SetMutation("coll/b", Map("matches", true))}); + const auto result = store.ExecuteQuery(core::QueryOrPipeline(query), false); + ASSERT_EQ(result.documents().size(), 2); + ASSERT_NE(result.documents().find(Key("coll/b")), result.documents().end()); + ASSERT_EQ(query_engine.documents_read_by_query(), 0); +} + TEST_F(LevelDbLocalStoreTest, UsesPartiallyIndexedRemoteDocumentsWhenAvailable) { FieldIndex index = diff --git a/Firestore/core/test/unit/local/leveldb_mutation_queue_test.cc b/Firestore/core/test/unit/local/leveldb_mutation_queue_test.cc index ade05e1d3ec..5090a8be407 100644 --- a/Firestore/core/test/unit/local/leveldb_mutation_queue_test.cc +++ b/Firestore/core/test/unit/local/leveldb_mutation_queue_test.cc @@ -162,7 +162,7 @@ TEST_F(LevelDbMutationQueueTest, LoadNextBatchIdOnlyFindsMutations) { } TEST_F(LevelDbMutationQueueTest, - LoadNextBatchIdIncludesLargestIndexedBatchIdAcrossUsers) { + StartIncludesLargestIndexedBatchIdAcrossUsers) { persistence_->Run("AddIndexAndRestartMutationQueue", [&] { auto* index_manager = persistence_->GetIndexManager(User("user")); index_manager->Start(); @@ -175,6 +175,8 @@ TEST_F(LevelDbMutationQueueTest, auto* other_user_index_manager = persistence_->GetIndexManager(User("other-user")); other_user_index_manager->Start(); + other_user_index_manager->AddFieldIndex( + MakeFieldIndex("coll", "a", model::Segment::Kind::kAscending)); other_user_index_manager->UpdateCollectionGroup( "coll", model::IndexOffset(model::SnapshotVersion::None(), model::DocumentKey::Empty(), 40)); From 4745d97b2c1cc3bc56ace391eb3a76685336f9b0 Mon Sep 17 00:00:00 2001 From: dajiaohuang <108231307+dajiaohuang@users.noreply.github.com> Date: Sat, 3 Oct 2026 13:08:28 +0800 Subject: [PATCH 4/6] Backfill the acknowledged document before reopening the indexed store --- Firestore/core/test/unit/local/leveldb_local_store_test.cc | 3 +++ 1 file changed, 3 insertions(+) diff --git a/Firestore/core/test/unit/local/leveldb_local_store_test.cc b/Firestore/core/test/unit/local/leveldb_local_store_test.cc index 33b6c3ba7b2..13b835967f3 100644 --- a/Firestore/core/test/unit/local/leveldb_local_store_test.cc +++ b/Firestore/core/test/unit/local/leveldb_local_store_test.cc @@ -255,6 +255,9 @@ TEST(LevelDbLocalStoreRestartTest, IndexedCacheQueryIncludesNewPendingWrite) { results.emplace_back(Version(10), Array()); store.AcknowledgeBatch( model::MutationBatchResult(batch, Version(10), std::move(results), {})); + // Include the acknowledged remote document in the index as well. This + // advances its read-time offset without discarding the batch high-water. + store.Backfill(); } // Reopen the same database after acknowledging every mutation. The persisted From b34454374215c369d51f0eecd99ebb2347dc171f Mon Sep 17 00:00:00 2001 From: dajiaohuang <108231307+dajiaohuang@users.noreply.github.com> Date: Sat, 3 Oct 2026 13:21:24 +0800 Subject: [PATCH 5/6] Keep the existing shared index configuration in the restart fixture --- Firestore/core/test/unit/local/leveldb_mutation_queue_test.cc | 2 -- 1 file changed, 2 deletions(-) diff --git a/Firestore/core/test/unit/local/leveldb_mutation_queue_test.cc b/Firestore/core/test/unit/local/leveldb_mutation_queue_test.cc index 5090a8be407..a9eff85955f 100644 --- a/Firestore/core/test/unit/local/leveldb_mutation_queue_test.cc +++ b/Firestore/core/test/unit/local/leveldb_mutation_queue_test.cc @@ -175,8 +175,6 @@ TEST_F(LevelDbMutationQueueTest, auto* other_user_index_manager = persistence_->GetIndexManager(User("other-user")); other_user_index_manager->Start(); - other_user_index_manager->AddFieldIndex( - MakeFieldIndex("coll", "a", model::Segment::Kind::kAscending)); other_user_index_manager->UpdateCollectionGroup( "coll", model::IndexOffset(model::SnapshotVersion::None(), model::DocumentKey::Empty(), 40)); From c87ebae0bf32c98312d4825e35ad4beaa229bbcd Mon Sep 17 00:00:00 2001 From: dajiaohuang <108231307+dajiaohuang@users.noreply.github.com> Date: Sat, 3 Oct 2026 13:30:00 +0800 Subject: [PATCH 6/6] Read persisted index offsets through the public persistence interface --- .../core/test/unit/local/leveldb_local_store_test.cc | 12 +++++++----- 1 file changed, 7 insertions(+), 5 deletions(-) diff --git a/Firestore/core/test/unit/local/leveldb_local_store_test.cc b/Firestore/core/test/unit/local/leveldb_local_store_test.cc index 13b835967f3..11a38459a9e 100644 --- a/Firestore/core/test/unit/local/leveldb_local_store_test.cc +++ b/Firestore/core/test/unit/local/leveldb_local_store_test.cc @@ -17,6 +17,7 @@ #include "Firestore/core/include/firebase/firestore/geo_point.h" #include "Firestore/core/src/core/filter.h" #include "Firestore/core/src/core/query.h" +#include "Firestore/core/src/local/leveldb_index_manager.h" #include "Firestore/core/src/local/leveldb_persistence.h" #include "Firestore/core/src/local/local_write_result.h" #include "Firestore/core/src/local/target_data.h" @@ -243,11 +244,12 @@ TEST(LevelDbLocalStoreRestartTest, IndexedCacheQueryIncludesNewPendingWrite) { auto mutation = SetMutation("coll/a", Map("matches", true)); const auto write = store.WriteLocally({mutation}); store.Backfill(); - ASSERT_GT(store.GetFieldIndexes()[0] - .index_state() - .index_offset() - .largest_batch_id(), - 0); + const auto indexed_batch_id = persistence->Run("ReadIndexedBatchId", [&] { + return persistence->GetIndexManager(user) + ->GetMinOffset("coll") + .largest_batch_id(); + }); + ASSERT_GT(indexed_batch_id, 0); model::MutationBatch batch(write.batch_id(), Timestamp::Now(), {}, {std::move(mutation)});