Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 20 additions & 1 deletion src/async_io_manager.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -6070,6 +6070,9 @@ KvError CloudStoreMgr::DownloadFile(const TableIdent &tbl_id,

std::string tmp_filename = filename + ".tmp";

// True once the existing file has been moved aside to tmp_filename (as
// opposed to not existing yet), so a failed write can restore it.
bool moved_existing = false;
if (download_to_exist)
{
// Rename the existing file away before overwriting, so readers see
Expand All @@ -6082,13 +6085,25 @@ KvError CloudStoreMgr::DownloadFile(const TableIdent &tbl_id,
ReleaseCloudBuffer(std::move(download_task.response_data_));
return ToKvError(res);
}
moved_existing = (res == 0);
}

uint64_t flags = O_WRONLY | O_CREAT | O_DIRECT | O_NOATIME;
KvError err = WriteFile(
tbl_id, tmp_filename, download_task.response_data_, flags, offset);
ReleaseCloudBuffer(std::move(download_task.response_data_));
CHECK_KV_ERR(err);
if (err != KvError::NoError)
{
// WriteFile only touches bytes >= offset (the not-yet-installed tail no
// live snapshot maps); the reader-visible prefix is untouched. Restore
// the file we moved aside so committed reads keep working -- a later
// reopen re-syncs the tail -- instead of stranding it as .tmp.
if (moved_existing)
{
Rename(dir_fd.FdPair(), tmp_filename.c_str(), filename.c_str());
}
return err;
}

int res = Rename(dir_fd.FdPair(), tmp_filename.c_str(), filename.c_str());
if (res < 0)
Expand Down Expand Up @@ -7237,6 +7252,10 @@ KvError CloudStoreMgr::WriteFile(const TableIdent &tbl_id,
uint64_t flags,
uint64_t offset)
{
// Test seam: fail the local write so a test can drive DownloadFile's
// restore-on-failure path (e.g. cache-disk ENOSPC during a tail sync).
TEST_FAIL_POINT_RETURN("CloudWriteFile", KvError::OutOfSpace);

auto [dir_fd, dir_err] =
OpenOrCreateFD(tbl_id, LruFD::kDirectory, false, true, "", 0);
if (dir_err != KvError::NoError)
Expand Down
50 changes: 50 additions & 0 deletions tests/cloud.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@

#include "async_io_manager.h"
#include "common.h"
#include "fail_point.h"
#include "kv_options.h"
#include "storage/shard.h"
#include "test_utils.h"
Expand Down Expand Up @@ -1418,6 +1419,55 @@ TEST_CASE("cloud reopen refreshes manifest via archive swap", "[cloud][reopen]")
CleanupStore(options);
}

// A reopen syncs the tail data file via DownloadFile(download_to_exist=true),
// which renames the existing file aside before rewriting it. If the local
// write then fails (e.g. cache-disk ENOSPC), the committed pages in that file
// must not be lost: DownloadFile restores the file it moved aside. Without the
// restore the file is stranded as .tmp and its committed pages become
// unreadable.
TEST_CASE("cloud reopen tail-sync restores the data file on write failure",
"[cloud][reopen]")
{
eloqstore::KvOptions options = cloud_options;
options.store_path = {"/tmp/test-data-tailsync"};
options.cloud_store_path += "/tailsync-restore";
CleanupStore(options);

eloqstore::TableIdent tbl_id{"tailsync", 0};
eloqstore::EloqStore *store = InitStore(options);
MapVerifier verifier(tbl_id, store, false);
verifier.Upsert(0, 50);
verifier.Validate(); // baseline: committed data is readable

// Fail the tail-sync's local write so DownloadFile hits its restore path
// (the tail file has already been moved aside to .tmp at that point).
eloqstore::FailPoint::GetInstance().ArmOnce("CloudWriteFile");
eloqstore::ReopenRequest reopen_req;
reopen_req.SetArgs(tbl_id);
store->ExecSync(&reopen_req);
eloqstore::FailPoint::GetInstance().Disarm();
REQUIRE(reopen_req.Error() != eloqstore::KvError::NoError);

// The data file the tail sync moved aside must be restored, not left
// stranded as .tmp. (Reads alone don't show the bug: the pages are still
// in the buffer-pool cache, and a disk fault would auto-reopen and heal.)
namespace fs = std::filesystem;
const std::string partition_dir =
std::string(options.store_path[0]) + "/" + tbl_id.ToString();
size_t stranded_tmp = 0;
for (const auto &entry : fs::recursive_directory_iterator(partition_dir))
{
if (entry.is_regular_file() && entry.path().extension() == ".tmp")
{
++stranded_tmp;
}
}
REQUIRE(stranded_tmp == 0);

store->Stop();
CleanupStore(options);
}

TEST_CASE("cloud reopen refreshes local manifest from remote",
"[cloud][reopen]")
{
Expand Down
Loading