From 5fd0834d1d68db7c1951bdc0e0aa82ff90f656ac Mon Sep 17 00:00:00 2001 From: Jerry Zhao Date: Fri, 10 Jul 2026 06:32:10 +0000 Subject: [PATCH] fix: free the held cache page on write-path error returns A data page promoted into the buffer-pool cache (enable_data_page_cache) is detached and pinned only by the writing task's handle until a successful WritePageCallback swizzles it into the mapping. Several write-path error returns dropped that handle without freeing the page: Handle::~Handle only unpins, so a detached-but-unpinned page becomes invisible to eviction yet still counts toward the pool limit, permanently leaking one buffer-pool slot per occurrence. Under sustained I/O errors the usable page count ratchets down until AllocPage fails and foreground writes hit OutOfMem even after the fault clears, requiring a restart. Add WriteTask::ReleaseHeldPage, which frees the page only when releasing this handle leaves it detached and unpinned (the sole-owner promoted-page case); double-pinned index writes (the caller keeps a pin) and pages already linked into the active/free list are left alone. Call it from: - AppendWritePage's error returns (flush failure, OnDataFileSealed, AcquireWriteBuffer/TryReserve OutOfMem); - BatchWriteTask::Pop's five index-build error returns (prev_handle), guarded by an assert that an empty handle implies OutOfMem; - WriteTask::WritePage's non-append path -- IouringMgr::WritePage now takes the page by reference and consumes it only on success, so a synchronous OpenOrCreateFD failure no longer orphans it. Three persist regression tests reproduce each site by stuffing a small buffer pool via fault injection (SubmitMergedWrite, FlushIndexPage, WritePageBeforeSubmit) and asserting a later healthy write still commits; each fails with OutOfMem without the fix. --- include/async_io_manager.h | 9 +- include/tasks/write_task.h | 7 ++ src/async_io_manager.cpp | 13 ++- src/tasks/batch_write_task.cpp | 21 ++++ src/tasks/write_task.cpp | 49 ++++++++- tests/persist.cpp | 188 +++++++++++++++++++++++++++++++++ 6 files changed, 279 insertions(+), 8 deletions(-) diff --git a/include/async_io_manager.h b/include/async_io_manager.h index d230165d..3d72e254 100644 --- a/include/async_io_manager.h +++ b/include/async_io_manager.h @@ -121,8 +121,11 @@ class AsyncIoManager std::span page_ids, std::vector &pages) = 0; + // Writes a single page. @p page is consumed (moved out) ONLY on success; on + // an error return it is left untouched so the caller still owns it and can + // release the cache page it holds (see WriteTask::WritePage). virtual KvError WritePage(const TableIdent &tbl_id, - VarPage page, + VarPage &page, FilePageId file_page_id) = 0; virtual GlobalRegisteredMemory *GetGlobalRegisteredMemory() const @@ -561,7 +564,7 @@ class IouringMgr : public AsyncIoManager std::vector &pages) override; KvError WritePage(const TableIdent &tbl_id, - VarPage page, + VarPage &page, FilePageId file_page_id) override; KvError ReadSegments(const TableIdent &tbl_id, @@ -1590,7 +1593,7 @@ class MemStoreMgr : public AsyncIoManager std::vector &pages) override; KvError WritePage(const TableIdent &tbl_id, - VarPage page, + VarPage &page, FilePageId file_page_id) override; KvError SyncData(const TableIdent &tbl_id) override; KvError AbortWrite(const TableIdent &tbl_id) override; diff --git a/include/tasks/write_task.h b/include/tasks/write_task.h index c0de8f5a..53df2288 100644 --- a/include/tasks/write_task.h +++ b/include/tasks/write_task.h @@ -158,6 +158,13 @@ class WriteTask : public KvTask KvError WritePage(MemCachedPage::Handle &page, FilePageId file_page_id); KvError WritePage(VarPage page, FilePageId file_page_id); KvError AppendWritePage(VarPage page, FilePageId file_page_id); + // Release a cache page still held when a write path bails out on error + // (AppendWritePage's early returns, BatchWriteTask::Pop's index build). A + // freshly promoted/allocated page is detached and pinned only by this + // handle, so it must be freed back to the buffer pool or its slot leaks; + // data/overflow variants and cache pages the caller still pins (index + // writes keep a second IO pin) own their storage and are left alone. + void ReleaseHeldPage(VarPage page); void FlushAppendWrites(); // Build this task's CoW root and snapshot the branch-file-mapping tail in // one step: forwards to PageManager::MakeCowRoot, and on success captures diff --git a/src/async_io_manager.cpp b/src/async_io_manager.cpp index ecd3a574..f8239176 100644 --- a/src/async_io_manager.cpp +++ b/src/async_io_manager.cpp @@ -971,9 +971,13 @@ std::pair IouringMgr::GetManifest( } KvError IouringMgr::WritePage(const TableIdent &tbl_id, - VarPage page, + VarPage &page, FilePageId file_page_id) { + // Test seam: fail before `page` is consumed (models an OpenOrCreateFD + // failure) so a test can drive the caller's release of the held page. + TEST_FAIL_POINT_RETURN("WritePageBeforeSubmit", KvError::Corrupted); + auto [file_id, offset] = ConvFilePageId(file_page_id); uint64_t term = ProcessTerm(); std::string_view branch = GetActiveBranch(); @@ -1253,6 +1257,11 @@ KvError IouringMgr::SubmitMergedWrite(const TableIdent &tbl_id, std::vector &release_indices, bool use_fixed) { + // Test seam: force the append-mode merged write to fail so a test can drive + // FlushAppendWrites's error path (which must free the pages it holds) while + // AppendWritePage is still holding the current cache page. + TEST_FAIL_POINT_RETURN("SubmitMergedWrite", KvError::Corrupted); + const uint64_t term = ProcessTerm(); const std::string_view branch = GetActiveBranch(); // In append mode, offset 0 means this merged write targets a brand-new @@ -7083,7 +7092,7 @@ std::pair MemStoreMgr::GetManifest( } KvError MemStoreMgr::WritePage(const TableIdent &tbl_id, - VarPage page, + VarPage &page, FilePageId file_page_id) { auto it = store_.find(tbl_id); diff --git a/src/tasks/batch_write_task.cpp b/src/tasks/batch_write_task.cpp index d7093b51..917e18fa 100644 --- a/src/tasks/batch_write_task.cpp +++ b/src/tasks/batch_write_task.cpp @@ -11,6 +11,7 @@ #include "async_io_manager.h" #include "coding.h" #include "compression.h" +#include "fail_point.h" #include "storage/shard.h" #include "tasks/task.h" #include "utils.h" @@ -896,6 +897,11 @@ std::pair BatchWriteTask::Pop() KvError err = add_to_page(new_key, new_page_id); if (err != KvError::NoError) { + // prev_handle is empty only when FinishIndexPage OutOfMem'd at + // its own AllocPage (never assigned the page); any other error + // leaves it holding the page to free. + assert(prev_handle || err == KvError::OutOfMem); + ReleaseHeldPage(VarPage(std::move(prev_handle))); return {MemCachedPage::Handle(), err}; } } @@ -922,6 +928,8 @@ std::pair BatchWriteTask::Pop() KvError err = add_to_page(new_key, new_page_id); if (err != KvError::NoError) { + assert(prev_handle || err == KvError::OutOfMem); + ReleaseHeldPage(VarPage(std::move(prev_handle))); return {MemCachedPage::Handle(), err}; } AdvanceIndexPageIter(base_page_iter, is_base_iter_valid); @@ -936,6 +944,8 @@ std::pair BatchWriteTask::Pop() KvError err = add_to_page(new_key, new_page); if (err != KvError::NoError) { + assert(prev_handle || err == KvError::OutOfMem); + ReleaseHeldPage(VarPage(std::move(prev_handle))); return {MemCachedPage::Handle(), err}; } } @@ -961,12 +971,19 @@ std::pair BatchWriteTask::Pop() prev_handle, prev_key, prev_page_id, std::move(curr_page_key)); if (err != KvError::NoError) { + // Empty only when FinishIndexPage OutOfMem'd at its AllocPage. + assert(prev_handle || err == KvError::OutOfMem); + ReleaseHeldPage(VarPage(std::move(prev_handle))); return {MemCachedPage::Handle(), err}; } err = FlushIndexPage( prev_handle, std::move(prev_key), prev_page_id, splited); if (err != KvError::NoError) { + // FinishIndexPage above succeeded, so prev_handle always holds a + // page here (FlushIndexPage never clears it on failure). + assert(prev_handle); + ReleaseHeldPage(VarPage(std::move(prev_handle))); return {MemCachedPage::Handle(), err}; } if (!splited) @@ -1023,6 +1040,10 @@ KvError BatchWriteTask::FlushIndexPage(MemCachedPage::Handle &idx_page, PageId page_id, bool split) { + // Test seam: fail the flush while Pop still holds prev_handle, so a test + // can drive Pop's error returns (which must release the held index page). + TEST_FAIL_POINT_RETURN("FlushIndexPage", KvError::Corrupted); + // Flushes the built index page. idx_page->SetPageId(page_id); KvError err = WritePage(idx_page); diff --git a/src/tasks/write_task.cpp b/src/tasks/write_task.cpp index 5b6dd5a4..29307615 100644 --- a/src/tasks/write_task.cpp +++ b/src/tasks/write_task.cpp @@ -299,8 +299,15 @@ KvError WriteTask::WritePage(VarPage page, FilePageId file_page_id) return AppendWritePage(std::move(page), file_page_id); } - KvError err = IoMgr()->WritePage(tbl_ident_, std::move(page), file_page_id); - CHECK_KV_ERR(err); + // WritePage consumes `page` only on success; on error it leaves it with us + // so a synchronous failure (e.g. OpenOrCreateFD) does not orphan the cache + // page this handle holds. + KvError err = IoMgr()->WritePage(tbl_ident_, page, file_page_id); + if (err != KvError::NoError) + { + ReleaseHeldPage(std::move(page)); + return err; + } if (inflight_io_ >= opts->max_write_batch_pages) { // Avoid long running WriteTask block ReadTask/ScanTask @@ -345,6 +352,7 @@ KvError WriteTask::AppendWritePage(VarPage page, FilePageId file_page_id) FlushAppendWrites(); if (write_err_ != KvError::NoError) { + ReleaseHeldPage(std::move(page)); return write_err_; } // In cloud append mode, trigger immediate upload of sealed file @@ -353,12 +361,17 @@ KvError WriteTask::AppendWritePage(VarPage page, FilePageId file_page_id) { KvError err = IoMgr()->OnDataFileSealed( tbl_ident_, DataFileKey(sealed_file_id)); - CHECK_KV_ERR(err); + if (err != KvError::NoError) + { + ReleaseHeldPage(std::move(page)); + return err; + } } uint16_t buf_index = 0; char *buf = IoMgr()->AcquireWriteBuffer(buf_index); if (buf == nullptr) { + ReleaseHeldPage(std::move(page)); return KvError::OutOfMem; } bool use_fixed = IoMgr()->WriteBufferUseFixed(); @@ -369,6 +382,7 @@ KvError WriteTask::AppendWritePage(VarPage page, FilePageId file_page_id) char *dst = append_aggregator_.TryReserve(file_id, offset, page_size); if (dst == nullptr) { + ReleaseHeldPage(std::move(page)); return KvError::OutOfMem; } std::memcpy(dst, page_ptr, page_size); @@ -385,6 +399,35 @@ KvError WriteTask::AppendWritePage(VarPage page, FilePageId file_page_id) return KvError::NoError; } +void WriteTask::ReleaseHeldPage(VarPage page) +{ + if (VarPageType(page.index()) != VarPageType::MemCachedPage) + { + // Data/overflow pages carry their own buffer; destroying the VarPage + // releases it. Only promoted cache pages need explicit accounting. + return; + } + MemCachedPage::Handle &handle = std::get(page); + MemCachedPage *cache_page = handle.Get(); + if (cache_page == nullptr) + { + // Empty handle: nothing to release. This happens on an OutOfMem return + // where the allocation that would have populated the handle is exactly + // what failed (BatchWriteTask::Pop's index build via FinishIndexPage); + // the caller asserts the OutOfMem precondition. + return; + } + handle.Reset(); // drop this task's pin + // Free only when this handle was the sole owner (the promoted data-page + // case). A page still pinned by the caller -- index-page writes pass a + // second, temporary pin -- is the caller's to release, and a page already + // linked into the active/free list must not be freed here. + if (cache_page->IsDetached() && !cache_page->IsPinned()) + { + shard->IndexManager()->FreePage(cache_page); + } +} + void WriteTask::FlushAppendWrites() { if (!append_aggregator_.HasData()) diff --git a/tests/persist.cpp b/tests/persist.cpp index 32486cd1..c64fc602 100644 --- a/tests/persist.cpp +++ b/tests/persist.cpp @@ -510,6 +510,194 @@ TEST_CASE("write task abort rolls back branch file-id high-water (cloud)", REQUIRE(write(2, 64, 2) == eloqstore::KvError::NoError); } +// A write whose merged data flush fails must not leak the cache page it is +// holding. With enable_data_page_cache, each leaf data page is promoted into a +// fresh MemCachedPage before being written (write_task.cpp WritePage). When a +// subsequent page forces AppendWritePage to flush the buffered batch and that +// flush fails, the buggy error return drops the current promote page's only +// Handle -- which merely unpins a page that is detached (not on the free list +// nor the active LRU), permanently orphaning that buffer-pool slot. Under a +// tiny pool, repeated failures stuff every slot with orphans until no page can +// be allocated at all, so a later healthy write can never rebuild its index and +// fails with OutOfMem forever. +// +// The pool is sized to hold 8 cache pages (64KB * (1 - 0.5) / 4KB); one page +// per data file (shift 0) makes every leaf after the first target a new file, +// so each failing write flushes the prior leaf and leaks the current one. The +// SubmitMergedWrite fail point forces exactly one flush failure per armed +// write. +TEST_CASE("failed merged write must not leak the held cache page", + "[persist][append][abort]") +{ + eloqstore::KvOptions options = append_opts; // data_append_mode = true + options.pages_per_file_shift = 0; // one 4K page per data file + options.enable_data_page_cache = true; // promote leaves -> the leak site + options.data_page_size = 4 * eloqstore::KB; + // Write buffer holds 2 pages, so a buffered leaf does not auto-flush before + // the next (different-file) leaf forces the flush we want to fail. + options.write_buffer_size = 8 * eloqstore::KB; + options.write_buffer_ratio = 0.5; // 32KB write buffers, 32KB cache + options.buffer_pool_size = 64 * eloqstore::KB; // cache limit = 8 pages + options.auto_oom_retry_times = 0; // let OutOfMem surface immediately + + eloqstore::EloqStore *store = InitStore(options); + const eloqstore::TableIdent tbl_id{"merged-write-leak", 0}; + + auto write_batch = [&](uint64_t base, uint64_t count, uint64_t ts) + { + eloqstore::BatchWriteRequest req; + req.SetTableId(tbl_id); + for (uint64_t k = 0; k < count; ++k) + { + req.AddWrite(Key(base + k), + Value(base + k, 512), + ts, + eloqstore::WriteOp::Upsert); + } + store->ExecSync(&req); + return req.Error(); + }; + + // Each failing write spans several 4K leaf pages (>=2 files, shift 0): the + // first leaf is promoted and buffered, the second forces AppendWritePage to + // flush it, the injected failure aborts, and the buggy code leaks the + // second leaf's held promote page. After ~7 failures every cache slot but + // one is orphaned; from then on promotion just fails and the write degrades + // to the un-promoted path (no further leak), so the failures keep returning + // the injected error rather than OutOfMem. + for (int i = 0; i < 32; ++i) + { + eloqstore::FailPoint::GetInstance().ArmOnce("SubmitMergedWrite"); + eloqstore::KvError err = + write_batch(/*base=*/i * 100000, /*count=*/40, /*ts=*/i + 1); + eloqstore::FailPoint::GetInstance().Disarm(); + REQUIRE(err == eloqstore::KvError::Corrupted); + } + + // A healthy write whose B-tree is deep enough to buffer more than one index + // page at once (Pop flushes the previous index page -- still pinned in the + // write buffer -- then allocates the next). On fixed code the intact pool + // absorbs this and the write commits; on buggy code the orphaned slots + // leave no page to allocate for the second index page and it fails with + // OutOfMem. + REQUIRE(write_batch(/*base=*/90000000, /*count=*/4000, /*ts=*/1000) == + eloqstore::KvError::NoError); + + store->Stop(); +} + +// Sibling of the above for the index-build path: when Pop's FinishIndexPage / +// FlushIndexPage fails while prev_handle still holds the freshly allocated +// index page, the buggy code drops that handle without freeing it, orphaning +// the cache slot. The FlushIndexPage fail point forces exactly that: the data +// leaves flush fine, then the root index page's flush fails with prev_handle +// held. Each failure leaks one slot until the pool is exhausted and a later +// healthy write can no longer allocate an index page. +TEST_CASE("failed index-page flush must not leak the held cache page", + "[persist][append][abort]") +{ + eloqstore::KvOptions options = append_opts; // data_append_mode = true + options.pages_per_file_shift = 0; // one 4K page per data file + options.enable_data_page_cache = true; + options.data_page_size = 4 * eloqstore::KB; + options.write_buffer_size = 8 * eloqstore::KB; + options.write_buffer_ratio = 0.5; // 32KB write buffers + options.buffer_pool_size = 64 * eloqstore::KB; // cache limit = 8 pages + options.auto_oom_retry_times = 0; + + eloqstore::EloqStore *store = InitStore(options); + const eloqstore::TableIdent tbl_id{"index-flush-leak", 0}; + + auto write_batch = [&](uint64_t base, uint64_t count, uint64_t ts) + { + eloqstore::BatchWriteRequest req; + req.SetTableId(tbl_id); + for (uint64_t k = 0; k < count; ++k) + { + req.AddWrite(Key(base + k), + Value(base + k, 512), + ts, + eloqstore::WriteOp::Upsert); + } + store->ExecSync(&req); + return req.Error(); + }; + + // Each armed write fails at the index-page flush while Pop holds the index + // page; on buggy code that page leaks. Once the pool fills the writes start + // failing earlier with OutOfMem instead, so assert only that they fail. + for (int i = 0; i < 32; ++i) + { + eloqstore::FailPoint::GetInstance().ArmOnce("FlushIndexPage"); + eloqstore::KvError err = + write_batch(/*base=*/i * 100000, /*count=*/40, /*ts=*/i + 1); + eloqstore::FailPoint::GetInstance().Disarm(); + REQUIRE(err != eloqstore::KvError::NoError); + } + + // On fixed code no slot leaked, so this healthy write commits; on buggy + // code the orphaned slots leave no page for the index build and it + // OutOfMems. + REQUIRE(write_batch(/*base=*/90000000, /*count=*/40, /*ts=*/1000) == + eloqstore::KvError::NoError); + + store->Stop(); +} + +// Third leak site: the non-append write path (no write-buffer pool), where the +// promoted page is moved into IouringMgr::WritePage and a synchronous failure +// (modeled by the fail point, in reality an OpenOrCreateFD error) returned +// before the page is submitted. IouringMgr::WritePage now takes the page by +// reference and only consumes it on success, so WriteTask::WritePage can +// release it on error; the buggy code dropped the moved-in page and leaked it. +TEST_CASE("failed non-append write must not leak the held cache page", + "[persist][abort]") +{ + eloqstore::KvOptions options = default_opts; // data_append_mode = false + options.enable_data_page_cache = true; // promote leaves -> leak site + options.data_page_size = 4 * eloqstore::KB; + options.buffer_pool_size = 64 * eloqstore::KB; + options.auto_oom_retry_times = 0; + + eloqstore::EloqStore *store = InitStore(options); + const eloqstore::TableIdent tbl_id{"non-append-leak", 0}; + + auto write_batch = [&](uint64_t base, uint64_t count, uint64_t ts) + { + eloqstore::BatchWriteRequest req; + req.SetTableId(tbl_id); + for (uint64_t k = 0; k < count; ++k) + { + req.AddWrite(Key(base + k), + Value(base + k, 512), + ts, + eloqstore::WriteOp::Upsert); + } + store->ExecSync(&req); + return req.Error(); + }; + + // Each armed write fails at the first page's WritePage while a promoted + // page is held; on buggy code that page leaks. Once the pool is full + // promotion stops (the write degrades to the un-promoted path), so no + // further leak. + for (int i = 0; i < 32; ++i) + { + eloqstore::FailPoint::GetInstance().ArmOnce("WritePageBeforeSubmit"); + eloqstore::KvError err = + write_batch(/*base=*/i * 100000, /*count=*/40, /*ts=*/i + 1); + eloqstore::FailPoint::GetInstance().Disarm(); + REQUIRE(err != eloqstore::KvError::NoError); + } + + // On fixed code no slot leaked, so this healthy write commits; on buggy + // code the orphaned slots leave no page to allocate and it OutOfMems. + REQUIRE(write_batch(/*base=*/90000000, /*count=*/40, /*ts=*/1000) == + eloqstore::KvError::NoError); + + store->Stop(); +} + TEST_CASE("append mode survives compression toggles across restarts", "[persist]") {