fix: release consumed range scan results on source shards

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.

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.

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;

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();

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<DataSyncTask> data_sync_task, size_t worker_idx)
 {
@@ -4346,6 +4356,9 @@ void LocalCcShards::DataSyncForRangePartition(
                        << " with error code: "
                        << static_cast<uint32_t>(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,

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<RangePartitionDataSyncScanCc,
+                        decltype(release_scan_batch)>
+            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)

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