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/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" 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..cbea4ba4 100644 --- a/tests/cloud.cpp +++ b/tests/cloud.cpp @@ -7,6 +7,7 @@ #include #include #include +#include #include #include #include @@ -2216,3 +2217,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); +}