diff --git a/docs/07-durability-and-recovery.md b/docs/07-durability-and-recovery.md index f03e1174..72ea1815 100644 --- a/docs/07-durability-and-recovery.md +++ b/docs/07-durability-and-recovery.md @@ -128,6 +128,8 @@ Flow: 6. `WaitableCc` → `CcShard::OnDirtyDataFlushed()` — re-arm kickout requests blocked on dirty data. 4. **Completion & truncation** — `DataSyncTask::SetFinish/SetError` maintain `truncate_log_ts_ = min(data_sync_ts_)` over the operation's tasks. Once `all_task_started_` is true, the last task to finish (or `Ckpt()` for a zero-task round) performs exactly-once outcome finalization. Only `Origin::Checkpoint` can affect checkpoint failure metrics. Deduplicated/skipped success is neutral, a remaining task error is still a failure, and `NG_TERM_CHANGED` / `REQUESTED_NODE_NOT_LEADER` is cancellation. A successful truncatable operation performs `UpdateNodeGroupCkptTs` + `UpdateCheckpointTs` + `BrocastPrimaryCkptTs` (`tx_service/src/standby.cpp:107`, the `UpdateStandbyCkptTs` RPC). **Truncation contract: never report a ckpt ts unless every entry with `commit_ts <= ts` of this ng is durable in the kv store.** +**Range scan result ownership.** `RangePartitionDataSyncScanCc::Reset()` re-arms the request and restores the vector's constructed slots; it is not a release operation. Data-sync transfers payload ownership to its flush task and retains the normal non-full scan buffer for reuse. A full heap, an empty output batch (which may still retain keys from the previous batch), scan failure, and final completion release the remaining scan references through `LocalCcShards::ReleaseScanResultsAndWait`, which dispatches `ReleaseDataSyncScanHeapCc` to the source CC shard and waits for completion. Call this interface from the consumer after the scan and all result accesses have finished; it does not implicitly reset the scan. This keeps allocator accounting current and prevents a zero-row/full batch from repeatedly retrying while retaining its own scan allocations. Release does not clear entries' being-checkpointed state or replace successful flush completion; payloads already moved to flush tasks remain valid. Index generation releases every consumed PK batch before upload backpressure; see [08-range-and-bucket-management.md](08-range-and-bucket-management.md). + ### 3.5 ckpt_ts on entries, eviction, dirty-memory trigger - `CkptTs()` / monotonic `SetCkptTs()` live in `VersionedLruEntry`'s entry info (`tx_service/include/cc/cc_entry.h:580-662`). `IsDirty()` = `CommitTs > CkptTs` (versioned) or flush-bit unset (non-versioned); `IsFree()` (no locks ∧ not dirty) gates eviction — **only checkpointed entries can be kicked out** (`LocalCcShards::KickoutPage`, `local_cc_shards.h:1566`, additionally consults range `last_sync_ts`/dirty-range version for range tables). When eviction finds nothing free, the tx processor calls `ckpter_->Notify()` — memory pressure drives checkpointing. diff --git a/docs/08-range-and-bucket-management.md b/docs/08-range-and-bucket-management.md index 8b2d1fc5..0899c4e3 100644 --- a/docs/08-range-and-bucket-management.md +++ b/docs/08-range-and-bucket-management.md @@ -127,7 +127,7 @@ Per batch, `DataMigrationOp` (`tx_operation.h:1395`) stages — each log write c `SkGenerator` (`sk_generator.h:150`): 1. Registers with the NG's `GenerateSkStatus` (`local_cc_shards.h:129`): `StartGenerateSk(tx_term)` rejects stale terms and terminates/waits out an older run; the scan loop polls `CheckTxTermStatus()` and calls `TerminateGenerateSk()` if a newer term took over. -2. Scans the PK cc map range in batches (`RangePartitionDataSyncScanCc`, batch = `DATA_SYNC_SCAN_BATCH_SIZE` 3072, `src/sk_generator.cpp:325`), and for each visible record runs every new index's `SkEncoder::AppendPackedSk` to emit packed SK `WriteEntry`s (multikey detection and `PackSkError` propagate back in the RPC response / tx result). +2. Scans the PK cc map range in batches (`RangePartitionDataSyncScanCc`, batch = `DATA_SYNC_SCAN_BATCH_SIZE` 3072, `src/sk_generator.cpp:325`), and for each visible record runs every new index's `SkEncoder::AppendPackedSk` to emit packed SK `WriteEntry`s (multikey detection and `PackSkError` propagate back in the RPC response / tx result). After all encoders have consumed a batch, the generator calls `LocalCcShards::ReleaseScanResultsAndWait`, which uses `ReleaseDataSyncScanHeapCc` to destroy its cloned PK keys and shared payload references on the **source CC shard**. The interface waits for release before the generator calls `Reset()` or waits for an upload slot. A completed-batch scope guard also releases partial scan results on errors and term-change exits; retryable scan errors release before the retry sleep. Packed SK entries and the resume key own their data independently. `Reset()` reconstructs the indexed scan slots after release even if the heap was not full; it does not itself release a live batch. Retaining a consumed batch can keep the shared scan heap full, causing `ExportForCkpt` to return zero before it can overwrite those slots. 3. Hands batches to `UploadIndexContext` (5 background upload workers; 2 in debug). `UploadEncodedIndex` acquires **range read locks** on the SK table's ranges (bucket+range, `AcquireRangeReadLocks`) to get a stable range→NG mapping, buckets the entries into per-(NG, sk-range) sets, and sends them: locally as pooled `UploadBatchCc` requests (`UploadBatchType::SkIndexData`), remotely as `UploadBatch` RPCs of kind `SK_DATA`, batch size 128. Uploaded entries land in the SK cc maps (and are later flushed by `flush_all_old_tuples_sk_op_`). `has_dml_since_ddl_` (`range_slice.h:852`): set on a `StoreRange` when the index-build scan observes keys whose version exceeds the dirty schema version (concurrent DML during DDL). It is preserved across range splits (`SplitTableRange`) and shipped in `UploadRangeSlicesCc`, and lets recovery decide whether old-tuple SK data can be trusted as complete. diff --git a/tx_service/include/cc/cc_request.h b/tx_service/include/cc/cc_request.h index 8f5424a3..29c642ba 100644 --- a/tx_service/include/cc/cc_request.h +++ b/tx_service/include/cc/cc_request.h @@ -4090,13 +4090,13 @@ struct RangePartitionDataSyncScanCc : public CcRequestBase accumulated_scan_cnt_ = 0; accumulated_flush_data_size_ = 0; - if (scan_heap_is_full_ == 1) - { - // vec has been cleared during ReleaseDataSyncScanHeapCc, - // resize to prepared size - data_sync_vec_.resize(scan_batch_size_); - scan_heap_is_full_ = 0; - } + // Consumers may release a completed batch before waiting for downstream + // capacity even when the scan heap was not full. ExportForCkpt writes + // existing elements with operator[], so reconstruct those slots after + // release. Keeping capacity alone is insufficient. Reset still does not + // release live records; consumers must do that on the source shard. + data_sync_vec_.resize(scan_batch_size_); + scan_heap_is_full_ = 0; if (export_base_table_item_) { curr_slice_index_ = 0; diff --git a/tx_service/include/cc/local_cc_shards.h b/tx_service/include/cc/local_cc_shards.h index b6283c9f..6a60d2e7 100644 --- a/tx_service/include/cc/local_cc_shards.h +++ b/tx_service/include/cc/local_cc_shards.h @@ -51,6 +51,7 @@ #include "catalog_key_record.h" #include "cc_entry.h" #include "cc_page_clean_guard.h" +#include "cc_request.h" #include "cc_shard.h" #include "data_sync_task.h" #include "eloq_basic_catalog_factory.h" @@ -441,6 +442,17 @@ class LocalCcShards ccs->EnqueueLowPriorityCcRequest(req); } + /** + * Release scan result elements on the source shard and wait for completion. + * The scan must have completed and its consumers must have finished using + * the results. source_core must be the shard that exported this batch. + * Call from a consumer context, never from a CC request, and keep scan + * alive until this returns. Does not Reset the scan or release payloads + * already transferred to flush tasks. + */ + void ReleaseScanResultsAndWait(uint16_t source_core, + RangePartitionDataSyncScanCc &scan); + static uint64_t ClockTs(); static uint64_t ClockTsInMillseconds(); uint64_t TsBase(); diff --git a/tx_service/src/cc/local_cc_shards.cpp b/tx_service/src/cc/local_cc_shards.cpp index bba8fd47..6477ac63 100644 --- a/tx_service/src/cc/local_cc_shards.cpp +++ b/tx_service/src/cc/local_cc_shards.cpp @@ -3714,6 +3714,16 @@ void LocalCcShards::PostProcessRangePartitionDataSyncTask( } } +void LocalCcShards::ReleaseScanResultsAndWait( + uint16_t source_core, RangePartitionDataSyncScanCc &scan) +{ + ReleaseDataSyncScanHeapCc release_cc(&scan.DataSyncVec(), + &scan.ArchiveVec()); + EnqueueLowPriorityCcRequestToShard(source_core, &release_cc); + // Keep the stack request alive through every incremental release round. + release_cc.Wait(); +} + void LocalCcShards::DataSyncForRangePartition( std::shared_ptr data_sync_task, size_t worker_idx) { @@ -4346,6 +4356,9 @@ void LocalCcShards::DataSyncForRangePartition( << " with error code: " << static_cast(scan_cc.ErrorCode()); + // A failed scan may already have exported part of a batch. Return + // its references before completing the task and releasing its pins. + ReleaseScanResultsAndWait(dest_core, scan_cc); PostProcessRangePartitionDataSyncTask( std::move(data_sync_task), data_sync_txm, @@ -4463,6 +4476,13 @@ void LocalCcShards::DataSyncForRangePartition( if (data_sync_vec->empty()) { LOG(WARNING) << "data_sync_vec is empty."; + // A full scan heap can stop this batch before its first export, + // leaving keys from the preceding batch in scan_cc. Release + // them before Reset clears the Full flag and we retry the same + // cursor. Any archive payloads already moved above stay owned + // by archive_vec; only the scan request's remaining refs are + // freed. + ReleaseScanResultsAndWait(dest_core, scan_cc); // Reset scan_cc.Reset(); // Return the quota to flush data memory usage pool since the @@ -4540,15 +4560,7 @@ void LocalCcShards::DataSyncForRangePartition( if (scan_cc.scan_heap_is_full_ == 1) { - // Clear the FlushRecords' memory of scan cc since the - // DataSyncScan heap is full. - auto &data_sync_vec_ref = scan_cc.DataSyncVec(); - auto &archive_vec_ref = scan_cc.ArchiveVec(); - ReleaseDataSyncScanHeapCc release_scan_heap_cc( - &data_sync_vec_ref, &archive_vec_ref); - EnqueueLowPriorityCcRequestToShard(dest_core, - &release_scan_heap_cc); - release_scan_heap_cc.Wait(); + ReleaseScanResultsAndWait(dest_core, scan_cc); } // Reset scan_cc.Reset(); @@ -4563,12 +4575,7 @@ void LocalCcShards::DataSyncForRangePartition( } // Release scan heap memory after scan finish. - auto &data_sync_vec_ref = scan_cc.DataSyncVec(); - auto &archive_vec_ref = scan_cc.ArchiveVec(); - ReleaseDataSyncScanHeapCc release_scan_heap_cc(&data_sync_vec_ref, - &archive_vec_ref); - EnqueueLowPriorityCcRequestToShard(dest_core, &release_scan_heap_cc); - release_scan_heap_cc.Wait(); + ReleaseScanResultsAndWait(dest_core, scan_cc); PostProcessRangePartitionDataSyncTask(std::move(data_sync_task), data_sync_txm, diff --git a/tx_service/src/sk_generator.cpp b/tx_service/src/sk_generator.cpp index 5000e7c6..0c9c9a6e 100644 --- a/tx_service/src/sk_generator.cpp +++ b/tx_service/src/sk_generator.cpp @@ -355,6 +355,19 @@ void SkGenerator::ScanAndEncodeIndex(const TxKey *start_key, cc_shards->EnqueueToCcShard(dest_core, &scan_req); scan_req.Wait(); + // The completed scan owns cloned keys and PK payload references. Keep + // them through every encoder, then return them on their source shard + // before upload backpressure or a retry sleep. The guard also covers + // partial scan errors, encoding errors and term-change returns; it does + // not own the stack request itself. Never run it while a scan is + // active. + auto release_scan_batch = + [cc_shards, dest_core](RangePartitionDataSyncScanCc *scan) + { cc_shards->ReleaseScanResultsAndWait(dest_core, *scan); }; + std::unique_ptr + completed_batch(&scan_req, release_scan_batch); + if (scan_req.IsError()) { scan_res = scan_req.ErrorCode(); @@ -372,7 +385,6 @@ void SkGenerator::ScanAndEncodeIndex(const TxKey *start_key, else if (scan_res == CcErrorCode::OUT_OF_MEMORY || scan_res == CcErrorCode::DATA_STORE_ERR) { - std::this_thread::sleep_for(std::chrono::seconds(30)); // Reset the paused key. const TxKey &paused_key = scan_req.PausePos().first; if (!scan_req.IsDrained()) @@ -383,9 +395,11 @@ void SkGenerator::ScanAndEncodeIndex(const TxKey *start_key, assert(paused_key.IsOwner()); paused_key.Copy(last_finished_pos); } + completed_batch.reset(); scan_req.Reset(); scan_pk_finished = false; scan_res = CcErrorCode::NO_ERROR; + std::this_thread::sleep_for(std::chrono::seconds(30)); continue; } else @@ -481,6 +495,12 @@ void SkGenerator::ScanAndEncodeIndex(const TxKey *start_key, } /* End of foreach new_indexes_name */ scan_pk_finished = scan_data_drained; + // Encoded SK entries and last_finished_pos own their data. Drop aliases + // to the consumed PK batch before it is released and before Enqueue can + // wait for an upload slot. Release also handles a zero-row/full batch. + target_key = TxKey(); + target_rec = nullptr; + completed_batch.reset(); scan_req.Reset(); scanned_items_count_ += batch_tuples; if (batch_tuples > 0) diff --git a/tx_service/tests/CMakeLists.txt b/tx_service/tests/CMakeLists.txt index 66898a88..4ff8b840 100644 --- a/tx_service/tests/CMakeLists.txt +++ b/tx_service/tests/CMakeLists.txt @@ -102,6 +102,16 @@ foreach(name ${OWN_MAIN_TESTS}) add_substrate_test(${name} Catch2::Catch2) endforeach() +# Standalone component regression: real scan export/heap release on a TestNode. +# A timeout bounds a regression in the source-shard completion handshake. +add_executable(RangeScanMemory-Test ${TESTS_DIR}/RangeScanMemory-Test.cpp) +target_link_libraries(RangeScanMemory-Test PRIVATE test_harness) +set_property(TARGET RangeScanMemory-Test PROPERTY CXX_STANDARD 20) +set_property(TARGET RangeScanMemory-Test PROPERTY CXX_EXTENSIONS OFF) +add_test(NAME RangeScanMemory-Test COMMAND RangeScanMemory-Test) +set_tests_properties(RangeScanMemory-Test PROPERTIES + LABELS "RangeScanMemory-Test" TIMEOUT 90) + # --- Phase 2 cross-NG cluster test (own main; links cluster_harness) --- # Drives a real 2-node out-of-process cluster, so it links cluster_harness (the # TestCluster driver + generated WorkloadService stub) and must be built after diff --git a/tx_service/tests/RangeScanMemory-Test.cpp b/tx_service/tests/RangeScanMemory-Test.cpp new file mode 100644 index 00000000..201d2742 --- /dev/null +++ b/tx_service/tests/RangeScanMemory-Test.cpp @@ -0,0 +1,310 @@ +/** + * Copyright (C) 2026 EloqData Inc. + * + * This program is free software: you can redistribute it and/or modify + * it under either of the following two licenses: + * 1. GNU Affero General Public License, version 3, as published by the Free + * Software Foundation. + * 2. GNU General Public License as published by the Free Software + * Foundation; version 2 of the License. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License or GNU General Public License for more + * details. + * + * You should have received a copy of the GNU Affero General Public License + * and GNU General Public License V2 along with this program. If not, see + * . + * + */ + +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "cc/cc_entry.h" +#include "cc/cc_req_misc.h" +#include "cc/cc_request.h" +#include "cc/cc_shard.h" +#include "cc/local_cc_shards.h" +#include "cc/template_cc_map.h" +#include "harness/test_node.h" +#include "sharder.h" +#include "tx_key.h" +#include "tx_record.h" + +namespace txservice +{ +namespace +{ +void Check(bool value, const char *message) +{ + if (!value) + { + std::cerr << "FAIL: " << message << std::endl; + std::exit(1); + } +} + +/** + * ReleaseDataSyncScanHeapCc leaves size zero while retaining capacity. Both + * a partial/full batch and a normal consumed batch must be reusable afterward: + * ExportForCkpt writes existing FlushRecord objects with operator[]. + */ +void ResetReleasedBatch(bool heap_full) +{ + const TableName table(std::string_view("scan_memory_test"), + TableType::Primary, + TableEngine::EloqDoc); + TxKey start = CompositeKey::NegativeInfinity()->CloneTxKey(); + TxKey end = CompositeKey::PositiveInfinity()->CloneTxKey(); + constexpr size_t kBatchSize = 64; + RangePartitionDataSyncScanCc scan( + table, 20, 0, 1, kBatchSize, 1, &start, &end, 0, true, true); + auto payload = std::make_shared>(7); + std::weak_ptr weak = payload; + scan.DataSyncVec()[0].SetVersionedPayload(std::move(payload)); + scan.accumulated_scan_cnt_ = 1; + scan.scan_heap_is_full_ = heap_full; + scan.PausePos().first = TxKey(std::make_unique>(9)); + // Isolate the Reset postcondition from scheduling: this is the state left + // after the source-shard release request has destroyed all elements. + scan.DataSyncVec().clear(); + Check(weak.expired(), "consumed payload reference was not released"); + scan.Reset(); + Check(scan.DataSyncVec().size() == kBatchSize, + "Reset did not reconstruct released scan slots without Full"); + Check(scan.accumulated_scan_cnt_ == 0, "Reset kept the old count"); + Check(scan.scan_heap_is_full_ == 0, "Reset kept the Full flag"); + Check(!scan.IsDrained(), "Reset incorrectly marked the range drained"); + Check(*scan.PausePos().first.GetKey>() == + CompositeKey(9), + "Reset changed the resume key"); + + CcEntry, CompositeRecord, true, true> entry; + entry.payload_.cur_payload_ = std::make_unique>(11); + entry.SetCommitTsPayloadStatus(10, RecordStatus::Normal); + entry.SetCkptTs(10); + uint64_t flush_size = 0; + const size_t exported = entry.ExportForCkpt(CompositeKey(9), + scan.DataSyncVec(), + scan.ArchiveVec(), + scan.MoveBaseIdxVec(), + 20, + 1, + true, + scan.accumulated_scan_cnt_, + true, + true, + false, + flush_size); + Check(exported == 1 && scan.accumulated_scan_cnt_ == 1, + "the next batch did not export its resume record"); + Check(scan.DataSyncVec()[0].Payload() != nullptr, + "the next batch lost its payload"); +} + +/** + * Exercise real scan-heap accounting and incremental release on a running CC + * shard. This is the export/release boundary, not a full index-build fixture. + */ +void ReleaseOnSourceShard() +{ + test::TestNode node; + LocalCcShards *shards = Sharder::Instance().GetLocalCcShards(); + // Use a nonzero source so release must honor the supplied shard index. + constexpr uint16_t kSourceShard = 1; + constexpr size_t kBatchSize = LocalCcShards::DATA_SYNC_SCAN_BATCH_SIZE; + using Key = CompositeKey; + using Record = CompositeRecord; + using Entry = CcEntry; + using Map = TemplateCcMap; + const TableName table(std::string_view("scan_memory_test"), + TableType::Primary, + TableEngine::EloqDoc); + TxKey start = Key::NegativeInfinity()->CloneTxKey(); + TxKey end = Key::PositiveInfinity()->CloneTxKey(); + RangePartitionDataSyncScanCc scan( + table, 20, 0, 1, kBatchSize, 1, &start, &end, 0, true, true); + std::weak_ptr pk_payload; + int64_t allocated_before = 0; + int64_t allocated_after = 0; + + auto run_on_source = [&](std::function task) + { + WaitableCc request(std::move(task)); + shards->EnqueueToCcShard(kSourceShard, &request); + request.Wait(); + Check(!request.IsError(), "source-shard request failed"); + }; + auto export_entry = [&](Map &map, Entry &entry, const Key &key) + { + return map.ExportForCkpt(&entry, + key, + scan.DataSyncVec(), + scan.ArchiveVec(), + scan.MoveBaseIdxVec(), + 20, + 1, + true, + scan.accumulated_scan_cnt_, + true, + true, + false, + scan.accumulated_flush_data_size_); + }; + + run_on_source( + [&](CcShard &shard) + { + Map map(&shard, 0, table, 1); + Entry entry; + entry.payload_.cur_payload_ = std::make_unique(7); + entry.SetCommitTsPayloadStatus(10, RecordStatus::Normal); + entry.SetCkptTs(10); + pk_payload = entry.payload_.VersionedCurrentPayload(); + // Large cloned keys reach the real heap limit with a small number + // of exports; the PK payload itself is still shared from the source + // entry. + Key key(std::string(512 * 1024, 'k')); + // The old libstdc++ ABI uses copy-on-write strings. Taking a + // mutable element reference makes this source unshareable, so each + // scan key clone really allocates its bytes on the scan heap in + // either ABI. + std::get<0>(key.Tuple())[0] = 'k'; + bool full = false; + while (scan.accumulated_scan_cnt_ < kBatchSize) + { + const auto result = export_entry(map, entry, key); + if (result.second) + { + full = true; + break; + } + Check(result.first == 1, "pressure fixture failed to export"); + } + std::cout << "Pressure fixture: rows=" << scan.accumulated_scan_cnt_ + << " full=" << full << " limit=" + << shard.GetShardDataSyncScanHeap()->MemoryLimit() + << std::endl; + Check(full && scan.accumulated_scan_cnt_ > 0, + "pressure fixture did not fill the scan heap"); + scan.scan_heap_is_full_ = 1; + + // Exercise both vectors across more than one release request round. + CcShardHeap *heap = shard.GetShardDataSyncScanHeap(); + mi_heap_t *previous = heap->SetAsDefaultHeap(); +#if defined(WITH_JEMALLOC) + auto previous_arena = heap->SetAsDefaultArena(); +#endif + const Key archive_key(std::string("archive")); + for (size_t i = 0; + i < 2 * ReleaseDataSyncScanHeapCc::VEC_ERASE_BATCH_SIZE + 1; + ++i) + { + scan.ArchiveVec().emplace_back(); + // Small archive keys are enough to exercise incremental + // deletion. + scan.ArchiveVec().back().CloneOrCopyKey(TxKey(&archive_key)); + } + Check(heap->Full(&allocated_before), + "scan heap unexpectedly shrank"); + mi_heap_set_default(previous); +#if defined(WITH_JEMALLOC) + JemallocArenaSwitcher::SwitchToArena(previous_arena); +#endif + return true; + }); + + // The previous Reset-only pattern retains the consumed batch. The next + // export stops at the heap check before it can overwrite any old slot. + scan.Reset(); + Check(!pk_payload.expired(), "Reset unexpectedly released the PK batch"); + run_on_source( + [&](CcShard &shard) + { + Map map(&shard, 0, table, 1); + Entry entry; + const auto result = + export_entry(map, entry, Key(std::string("resume"))); + Check(result.first == 0 && result.second, + "Reset-only did not reproduce the zero-progress heap gate"); + return true; + }); + + shards->ReleaseScanResultsAndWait(kSourceShard, scan); + Check(scan.DataSyncVec().empty() && scan.ArchiveVec().empty(), + "incremental source-shard release left scan records behind"); + Check(pk_payload.expired(), "source-shard release retained the PK payload"); + scan.Reset(); + Check(scan.DataSyncVec().size() == kBatchSize, + "released zero-row batch cannot be reused"); + + run_on_source( + [&](CcShard &shard) + { + CcShardHeap *heap = shard.GetShardDataSyncScanHeap(); + mi_heap_t *previous = heap->SetAsDefaultHeap(); +#if defined(WITH_JEMALLOC) + auto previous_arena = heap->SetAsDefaultArena(); +#endif + Check(!heap->Full(&allocated_after), + "released scan heap is still full"); + mi_heap_set_default(previous); +#if defined(WITH_JEMALLOC) + JemallocArenaSwitcher::SwitchToArena(previous_arena); +#endif + Map map(&shard, 0, table, 1); + Entry entry; + entry.payload_.cur_payload_ = std::make_unique(11); + entry.SetCommitTsPayloadStatus(10, RecordStatus::Normal); + entry.SetCkptTs(10); + const auto result = + export_entry(map, entry, Key(std::string("resume"))); + Check(result.first == 1 && !result.second, + "scan did not resume after source-shard release"); + return true; + }); + // DataSync transfers this reference to its flush task before releasing the + // scan buffer. Releasing the scan must not destroy a transferred payload. + TxKey flush_key = scan.DataSyncVec()[0].Key().Clone(); + auto flush_payload = scan.DataSyncVec()[0].ReleaseVersionedPayload(); + std::weak_ptr transferred_payload = flush_payload; + shards->ReleaseScanResultsAndWait(kSourceShard, scan); + Check(flush_payload != nullptr, + "release lost the transferred flush payload"); + Check(std::get<0>(static_cast(*flush_payload).Tuple()) == 11 && + *flush_key.GetKey() == Key(std::string("resume")), + "release invalidated the flush task's key or payload"); + flush_payload.reset(); + Check(transferred_payload.expired(), + "scan kept another reference after flush ownership was released"); + Check(allocated_after < allocated_before, + "source-shard release did not update heap accounting"); + std::cout << "PASS: source-shard release resumed export; scan heap bytes " + << allocated_before << " -> " << allocated_after << std::endl; +} +} // namespace +} // namespace txservice + +int main(int argc, char **argv) +{ + GFLAGS_NAMESPACE::ParseCommandLineFlags(&argc, &argv, true); + txservice::ResetReleasedBatch(true); + txservice::ResetReleasedBatch(false); + std::cout << "PASS: released full and non-full scan batches resume safely" + << std::endl; + txservice::ReleaseOnSourceShard(); +}