fix(cloud): archive-tag reopen falls back to listing older-term archives

include/async_io_manager.h

@@ -9,6 +9,7 @@
 #include <cstdint>
 #include <cstdlib>
 #include <deque>
+#include <functional>
 #include <limits>
 #include <memory>
 #include <optional>
@@ -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<void(const std::string &)> &on_object);
+
 private:
     int CreateFile(LruFD::Ref dir_fd,
                    TypedFileId file_id,

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"
+echo "[OK] Format completed"

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<void(const std::string &)> &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<std::string> 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<ManifestFilePtr, KvError> CloudStoreMgr::RefreshManifest(
     const TableIdent &tbl_id, std::string_view archive_tag)
 {
@@ -4715,84 +4768,40 @@ std::pair<ManifestFilePtr, KvError> CloudStoreMgr::RefreshManifest(
         {
             uint64_t best_term = 0;
             bool found = false;
-            std::vector<std::string> 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<std::string> 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<std::string> 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<std::string> 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<ManifestFilePtr, KvError> 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<std::string> 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};

tests/cloud.cpp

@@ -7,6 +7,7 @@
 #include <functional>
 #include <map>
 #include <memory>
+#include <string>
 #include <thread>
 #include <unordered_set>
 #include <vector>
@@ -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<std::string, eloqstore::KvEntry> v1_dataset;
+
+    // First life (term 1): write v1, archive it under `tag`, add v2 on top.
+    {
+        auto store = std::make_unique<eloqstore::EloqStore>(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<eloqstore::EloqStore>(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);
+}