From d22ed8c7764ebb8c4477a238c24a015222862531 Mon Sep 17 00:00:00 2001 From: Jerry Zhao Date: Wed, 8 Jul 2026 04:50:41 +0000 Subject: [PATCH 1/3] fix(cloud): archive-tag reopen falls back to listing older-term archives MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Archives are named archive___ with the term of the process that created them, and are never re-termed — unlike the live manifest, which GetManifest promotes to the current term on first access. RefreshManifest's archive path, however, built only the current-term name (selected_term was hard-coded to process_term) and returned the download error directly. After any failover, every pre-existing archive tag — including every tag GlobalListArchiveTags returns — therefore resolved to NotFound, which InstallExternalSnapshot's missing-remote-state branch turned into a partition wipe: restore-to-tag silently became drop-partition. The promote branch below the lookup (selected_term != process_term) was dead code, evidence the cross-term case was intended but never wired. Mirror the tagless path: on NotFound, list the manifest prefix and select the newest archive of the active branch carrying the requested tag (term <= process_term), then download it — which also brings the existing promote-rename branch to life for the older-term result. The listing loop is shared with the tagless fallback via a new ListManifestObjects helper. Regression test: archive under term 1, restart with term 2 on an empty local cache, reopen with the tag — must restore the archived snapshot. Fails without this fix (the partition is wiped instead) and passes with it. --- include/async_io_manager.h | 11 ++ src/async_io_manager.cpp | 214 ++++++++++++++++++++++++------------- tests/cloud.cpp | 65 +++++++++++ 3 files changed, 214 insertions(+), 76 deletions(-) diff --git a/include/async_io_manager.h b/include/async_io_manager.h index d230165d..a9e6746e 100644 --- a/include/async_io_manager.h +++ b/include/async_io_manager.h @@ -9,6 +9,7 @@ #include #include #include +#include #include #include #include @@ -1326,6 +1327,16 @@ class CloudStoreMgr final : public IouringMgr private: void WaitForCloudTasksToDrain(); + /** + * @brief Lists every cloud object under @p tbl_id's manifest prefix + * (paginated, no delimiter), invoking @p on_object for each name as the + * pages stream in — nothing is accumulated across pages. Shared by the + * tagless and archive-tag fallbacks of RefreshManifest. + */ + KvError ListManifestObjects( + const TableIdent &tbl_id, + const std::function &on_object); + private: int CreateFile(LruFD::Ref dir_fd, TypedFileId file_id, diff --git a/src/async_io_manager.cpp b/src/async_io_manager.cpp index a1f2b0cf..a45a3d9c 100644 --- a/src/async_io_manager.cpp +++ b/src/async_io_manager.cpp @@ -4673,6 +4673,59 @@ class BufferManifest final : public ManifestFile }; } // namespace +KvError CloudStoreMgr::ListManifestObjects( + const TableIdent &tbl_id, + const std::function &on_object) +{ + std::string remote_path = + tbl_id.ToString() + "/" + FileNameManifest + FileNameSeparator; + std::string continuation_token; + KvTask *current_task = ThdTask(); + do + { + ObjectStore::ListTask list_task(remote_path, false); + // Flat manifest keys; no delimiter so the response cannot carry + // CommonPrefixes entries (see GetManifest). + list_task.SetRecursive(true); + list_task.SetContinuationToken(continuation_token); + list_task.SetKvTask(current_task); + AcquireCloudSlot(current_task); + obj_store_.SubmitTask(&list_task, shard); + current_task->WaitIo(); + + if (list_task.error_ != KvError::NoError) + { + LOG(ERROR) << "CloudStoreMgr::ListManifestObjects: list objects " + "failed for " + << tbl_id << " : " << ErrorString(list_task.error_); + return list_task.error_; + } + + std::vector batch_files; + std::string next_token; + if (!obj_store_.ParseListObjectsResponse( + list_task.response_data_.view(), + list_task.json_data_, + &batch_files, + nullptr, + &next_token)) + { + LOG(ERROR) << "CloudStoreMgr::ListManifestObjects: parse list " + "response failed for table " + << tbl_id; + return KvError::Corrupted; + } + + for (const std::string &name : batch_files) + { + on_object(name); + } + continuation_token = std::move(next_token); + } while (!continuation_token.empty()); + + return KvError::NoError; +} + std::pair CloudStoreMgr::RefreshManifest( const TableIdent &tbl_id, std::string_view archive_tag) { @@ -4715,84 +4768,40 @@ std::pair CloudStoreMgr::RefreshManifest( { uint64_t best_term = 0; bool found = false; - std::vector cloud_files; - std::string remote_path = - tbl_id.ToString() + "/" + FileNameManifest + FileNameSeparator; - - std::string continuation_token; - KvTask *current_task = ThdTask(); - do - { - ObjectStore::ListTask list_task(remote_path, false); - // Flat manifest keys; no delimiter so the response cannot - // carry CommonPrefixes entries (see GetManifest). - list_task.SetRecursive(true); - list_task.SetContinuationToken(continuation_token); - list_task.SetKvTask(current_task); - AcquireCloudSlot(current_task); - obj_store_.SubmitTask(&list_task, shard); - current_task->WaitIo(); - - if (list_task.error_ != KvError::NoError) + KvError list_err = ListManifestObjects( + tbl_id, + [&](const std::string &name) { - LOG(ERROR) - << "CloudStoreMgr::RefreshManifest: list objects " - "failed for " - << tbl_id << " : " << ErrorString(list_task.error_); - return {nullptr, list_task.error_}; - } - - std::vector batch_files; - std::string next_token; - if (!obj_store_.ParseListObjectsResponse( - list_task.response_data_.view(), - list_task.json_data_, - &batch_files, - nullptr, - &next_token)) - { - LOG(ERROR) << "CloudStoreMgr::RefreshManifest: parse list " - "response failed for table " - << tbl_id; - return {nullptr, KvError::Corrupted}; - } - - cloud_files.insert(cloud_files.end(), - std::make_move_iterator(batch_files.begin()), - std::make_move_iterator(batch_files.end())); - continuation_token = std::move(next_token); - } while (!continuation_token.empty()); - - if (cloud_files.empty() || (cloud_files.size() == 1 && - cloud_files[0] == CurrentTermFileName)) - { - return {nullptr, KvError::NotFound}; - } - - for (const std::string &name : cloud_files) + if (name == CurrentTermFileName) + { + return; + } + uint64_t term = 0; + std::string_view branch_name; + std::optional tag; + if (!ParseManifestFileSuffix(name, branch_name, term, tag)) + { + // Transient directory-style entry from a racing + // drop; see CloudStoreMgr::GetManifest for details. + LOG(WARNING) << "CloudStoreMgr::RefreshManifest: skip " + "unrecognized entry under manifest " + "prefix of table " + << tbl_id << ": " << name; + return; + } + if (tag.has_value()) + { + return; + } + if (term >= best_term) + { + found = true; + best_term = term; + } + }); + if (list_err != KvError::NoError) { - uint64_t term = 0; - std::string_view branch_name; - std::optional tag; - if (!ParseManifestFileSuffix(name, branch_name, term, tag)) - { - // Transient directory-style entry from a racing drop; - // see CloudStoreMgr::GetManifest for details. - LOG(WARNING) - << "CloudStoreMgr::RefreshManifest: skip unrecognized " - "entry under manifest prefix of table " - << tbl_id << ": " << name; - continue; - } - if (tag.has_value()) - { - continue; - } - if (term >= best_term) - { - found = true; - best_term = term; - } + return {nullptr, list_err}; } if (!found) @@ -4900,6 +4909,59 @@ std::pair CloudStoreMgr::RefreshManifest( KvError dl_err = download_to_buffer(selected_filename); + if (dl_err == KvError::NotFound) + { + // Archives are named with the term of the process that created them + // and are never re-termed, so after a failover every pre-existing + // tag lives under an earlier term and the current-term name above + // misses. Mirror the tagless path: stream the manifest-prefix + // listing and select the newest archive of the active branch + // carrying this tag. + uint64_t best_term = 0; + bool found = false; + KvError list_err = ListManifestObjects( + tbl_id, + [&](const std::string &name) + { + uint64_t term = 0; + std::string_view branch_name; + std::optional tag; + if (!ParseManifestFileSuffix(name, branch_name, term, tag)) + { + // Transient directory-style entry from a racing drop; + // see CloudStoreMgr::GetManifest for details. + LOG(WARNING) + << "CloudStoreMgr::RefreshManifest: skip unrecognized " + "entry under manifest prefix of table " + << tbl_id << ": " << name; + return; + } + if (!tag.has_value() || *tag != archive_tag || + branch_name != GetActiveBranch() || term > process_term) + { + return; + } + if (term >= best_term) + { + found = true; + best_term = term; + } + }); + if (list_err != KvError::NoError) + { + return {nullptr, list_err}; + } + + if (!found) + { + return {nullptr, KvError::NotFound}; + } + selected_term = best_term; + selected_filename = + BranchArchiveName(GetActiveBranch(), selected_term, archive_tag); + dl_err = download_to_buffer(selected_filename); + } + if (dl_err != KvError::NoError) { return {nullptr, dl_err}; diff --git a/tests/cloud.cpp b/tests/cloud.cpp index 4a533ace..565664c5 100644 --- a/tests/cloud.cpp +++ b/tests/cloud.cpp @@ -2216,3 +2216,68 @@ TEST_CASE("archive triggers with cloud-only partitions", "[cloud][archive]") store->Stop(); CleanupStore(options); } + +TEST_CASE("reopen to archive tag survives term bump across restart", + "[cloud][reopen][archive]") +{ + eloqstore::KvOptions options = cloud_archive_opts; + options.store_path = {"/tmp/test-data-archive-term"}; + options.cloud_store_path.push_back('/'); + options.cloud_store_path += "archive-term"; + + CleanupStore(options); + + const eloqstore::TableIdent tbl_id{"arch_term", 0}; + const std::string tag = "restore-point"; + std::map v1_dataset; + + // First life (term 1): write v1, archive it under `tag`, add v2 on top. + { + auto store = std::make_unique(options); + REQUIRE(store->Start(eloqstore::MainBranchName, /*term=*/1) == + eloqstore::KvError::NoError); + MapVerifier verifier(tbl_id, store.get(), false); + verifier.Upsert(0, 50); + v1_dataset = verifier.DataSet(); + REQUIRE_FALSE(v1_dataset.empty()); + + eloqstore::ArchiveRequest archive_req; + archive_req.SetTableId(tbl_id); + archive_req.SetTag(tag); + REQUIRE(store->ExecAsyn(&archive_req)); + archive_req.Wait(); + REQUIRE(archive_req.Error() == eloqstore::KvError::NoError); + + verifier.Upsert(100, 120); + verifier.SetAutoClean(false); + store->Stop(); + } + + // Second life at a bumped term (the embedder supplies a higher term on + // failover), while the archive object keeps the term of the process + // that created it. The tagged reopen must fall back to listing and + // find the older-term archive — before the fix it resolved to NotFound + // and wiped the partition instead of restoring it. Wipe the local + // cache first (a failed-over node starts with an empty disk); cloud + // state is preserved. + CleanupLocalStore(options); + { + auto store = std::make_unique(options); + REQUIRE(store->Start(eloqstore::MainBranchName, /*term=*/2) == + eloqstore::KvError::NoError); + + eloqstore::ReopenRequest reopen_req; + reopen_req.SetArgs(tbl_id); + reopen_req.SetTag(tag); + store->ExecSync(&reopen_req); + REQUIRE(reopen_req.Error() == eloqstore::KvError::NoError); + + MapVerifier verifier(tbl_id, store.get(), false); + verifier.SwitchDataSet(v1_dataset); + verifier.Validate(); + + verifier.SetAutoClean(false); + store->Stop(); + } + CleanupStore(options); +} From f06b1a85a7a2549dd1332b0348607331434c8539 Mon Sep 17 00:00:00 2001 From: Jerry Zhao Date: Wed, 8 Jul 2026 07:47:51 +0000 Subject: [PATCH 2/3] fix(scripts): use ports.ubuntu.com for arm64 libtinfo5 and bump deb revision cn.ports.ubuntu.com is unreachable from some networks and the libtinfo5 6.3-2ubuntu0.1 deb has been superseded by 0.2 on the mirrors, so the first-run clang-format bootstrap failed on arm64. Also add the missing trailing newline. --- scripts/format.sh | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/scripts/format.sh b/scripts/format.sh index e028c04f..ed644a22 100755 --- a/scripts/format.sh +++ b/scripts/format.sh @@ -24,7 +24,7 @@ install_clang_format_18_1_8() { TINFO_PKG_URL="https://security.ubuntu.com/ubuntu/pool/universe/n/ncurses" ;; aarch64|arm64) - TINFO_PKG_URL="https://cn.ports.ubuntu.com/pool/universe/n/ncurses" + TINFO_PKG_URL="https://ports.ubuntu.com/pool/universe/n/ncurses" ;; *) echo "[ERROR] Unsupported arch: $ARCH" @@ -33,7 +33,7 @@ install_clang_format_18_1_8() { esac $SUDO apt update - local TINFO_DEB="libtinfo5_6.3-2ubuntu0.1_${DPKG}.deb" + local TINFO_DEB="libtinfo5_6.3-2ubuntu0.2_${DPKG}.deb" local TINFO_URL="${TINFO_PKG_URL}/${TINFO_DEB}" echo "[INFO] Downloading ${TINFO_DEB} from ${TINFO_URL}" @@ -97,4 +97,4 @@ git ls-files -z '*.c' '*.cc' '*.cpp' '*.h' '*.hpp' \ | awk -v RS='\0' -v ORS='\0' '!/^(third_party|vendor|build|external)\//' \ | xargs -0 -r clang-format-18.1.8 -i --style="file" -echo "[OK] Format completed" \ No newline at end of file +echo "[OK] Format completed" From 6ba12527c6856d761cbb2295bed307de7162b9e8 Mon Sep 17 00:00:00 2001 From: Jerry Zhao Date: Wed, 8 Jul 2026 08:02:45 +0000 Subject: [PATCH 3/3] add header --- tests/cloud.cpp | 1 + 1 file changed, 1 insertion(+) diff --git a/tests/cloud.cpp b/tests/cloud.cpp index 565664c5..cbea4ba4 100644 --- a/tests/cloud.cpp +++ b/tests/cloud.cpp @@ -7,6 +7,7 @@ #include #include #include +#include #include #include #include