Ckpt ts per partition

docs/07-durability-and-recovery.md

@@ -123,9 +123,11 @@ Flow:
    1. `CopyBaseToArchive` (MVCC only) — copy kv base rows about to be overwritten into the archive table;
    2. `PutAll` — write base rows;
    3. `PutArchivesAll` (MVCC only) — write in-memory archive versions;
-   4. `PersistKV` if `store_hd_->NeedPersistKV()` (e.g. EloqStore) — batched fsync-equivalent;
-   5. `UpdateCceCkptTsCc` per shard — stamp `cce->SetCkptTs(commit_ts)` on every flushed entry (only when `need_update_ckpt_ts_`);
-   6. `WaitableCc` → `CcShard::OnDirtyDataFlushed()` — re-arm kickout requests blocked on dirty data.
+   4. `PersistKV` if `store_hd_->NeedPersistKV()` — required by RocksDB-backed DSS handlers because checkpoint writes skip the WAL and remain in a memtable until `FlushData`; EloqStore's completed batch-write callback is already its durability boundary and its `FlushData` is a no-op;
+   5. publish ckpt ts at the backend's full durability boundary. A merged flush buffer can straddle a node-group term transition; each datastore phase (`CopyBaseToArchive`, `PutAll`, and `PutArchivesAll`) finds the highest task term represented for each node group across the entire merged batch and does not issue reads or writes for that node group's lower-term tasks, including tasks in a different table bucket. Terms from different node groups are independent. For non-MVCC EloqStore, one pre-armed `UpdateCceCkptTsCc` publishes each retained partition's cc entries as soon as it lands; `PutAll` does not return until that fan-in completes. RocksDB-backed stores publish after `PersistKV` succeeds, aggregating all retained entries of one table and node group into one `UpdateCceCkptTsCc`; lower-term entries are not published because their datastore writes were discarded. MVCC flushes use the same deferred aggregation, including on EloqStore, because `PutArchivesAll` follows the base writes and must succeed before an entry can be marked clean;
+   6. every `UpdateCceCkptTsCc` slice ends with `CcShard::OnDirtyDataFlushed()`, which resets that shard's eviction cursor and wakes its cleaner when requests are parked. A wake that arrives while `ShardCleanCc` is already in use is sticky, so its give-up branch re-runs rather than stranding the wait list.
+
+   For the progressive EloqStore path, each completed partition frees the key/payload buffers of the `FlushRecord`s it carried (the interleaving vectors themselves live until the flush task ends) and `SyncPutAllData` reports the cumulative `FlushSize()` bytes freed — the same unit `DataSyncScan` charged. `FlushDataImpl` releases exactly that watermark (capped by the task's charge); the remainder — vector storage, lower-term discards, archives, and deferred or failed shares — is released only after `FlushDataImpl` frees those buffers at the end of the round, before slice post-processing. Data-sync admission thus advances with durable partitions while resident flush memory never exceeds what the quota claims, on every backend. MVCC and persist-needing stores retain the full quota until their later durability boundary.
 4. **Completion & truncation** — `DataSyncTask::SetFinish/SetError` (`tx_service/src/data_sync_task.cpp:113-198`) maintain `truncate_log_ts_ = min(data_sync_ts_)` over the round's tasks. The last task to finish (or `Ckpt()` itself) 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.**
 
 ### 3.5 ckpt_ts on entries, eviction, dirty-memory trigger
@@ -254,6 +256,7 @@ processed by the `replay_notify` thread. `ProcessRecoverTxTask` (`log_replay_ser
 
 - **ckpt_ts never passes an uncommitted write.** `ActiveTxMinTs` returns `min(wlock_ts) − 1`; a tx publishes its write-lock ts before receiving a commit ts. Corollary: a stuck tx (orphan lock) blocks log truncation for the whole ng — which is exactly why `CkptTsCc` piggybacks `CheckRecoverTx`.
 - **Truncate only what is durable.** `UpdateCheckpointTs` is sent with `truncate_log_ts_` = min `data_sync_ts_` over the round's tasks, and only when no task errored and nothing was skipped. Skipped entries (buffered commands → `LOG_NOT_TRUNCATABLE`) or task-limiter dedup (`SetNoTruncateLog`) silently turn the round into flush-without-truncate.
+- **Entry ckpt ts uses the complete durability boundary.** Non-MVCC EloqStore may publish a base partition on batch completion. RocksDB-backed handlers must wait for `PersistKV`; MVCC must also wait for archive writes. Publishing earlier makes a failed, unpersisted entry appear clean, eligible for eviction, and absent from the next retry.
 - **`truncate_log_ts_` may exceed the round's `ckpt_ts`** when a queued task's ts was adjusted upward; `Ckpt()` deliberately truncates with `truncate_log_ts_`, not `ckpt_ts` (`checkpointer.cpp:397-424`).
 - **Replay is idempotent.** Data replay skips `commit_ts <= cce.CommitTs()` and `commit_ts < schema_ts_`; catalog replay refuses to downgrade an existing catalog version; bucket-ownership filters drop foreign keys. Re-streaming after an error is safe.
 - **Term discipline.** Replay runs under the *candidate* term; `LeaderTerm` flips only after **all** log groups finish (`FinishLogGroupReplay`). A `WriteLogRequest` whose `node_terms` mismatch the log service's view is rejected, so a tx that straddles a participant failover cannot commit; `FillDataLogRequest` already aborts locally on intra-tx term divergence (`NG_TERM_CHANGED`).

docs/09-store-handler.md

@@ -71,7 +71,9 @@ Shard-owner caching: `dss_shards_[shard_id]` (atomic index) points into `dss_nod
 
 Retry semantics (`ReadClosure::Run` et al.): on `REQUESTED_NODE_NOT_OWNER` the server attaches `new_key_sharding` (new primary node + `shard_version`) to `CommonResult`; the client's `HandleShardingError` spins on `UpgradeShardVersion` (CAS on the node slot, `bthread_usleep(10ms)` backoff) and re-issues the request, up to `retry_limit_ = 2` retries per closure. A `NodeGroupChanged` sharding error is currently `LOG(FATAL)` (topology change handling is a TODO at `data_store_service_client.cpp:4471`). `SetupConfig` (registered as the DSS `update_config_listener_`) applies pushed topology updates guarded by `dss_topology_version_`.
 
-`PutAll` (`PutAllImpl`) groups `FlushRecord`s by kv partition, builds ≤64 MB `BatchWriteRecords` batches (`MAX_WRITE_BATCH_SIZE`), and flushes **partitions concurrently but each partition serially** — `PartitionBatchCallback` chains the next batch of a partition only after the previous one completes; `SyncPutAllData`/`SyncConcurrentRequest` (max 32 in-flight) coordinate completion and support coroutine yield/resume. All checkpoint batches are written with `skip_wal=true`; durability comes from the subsequent `PersistKV` → `FlushData` across **all** shards. Synchronous helpers (`FetchTable`, `UpsertDatabase`, ...) use `SyncCallbackData` (bthread mutex/condvar, or yield/resume when provided). `UpsertTable` runs on a dedicated 1-thread `upsert_table_worker_` after pinning the node group and checking the tx term.
+`PutAll` (`PutAllImpl`) groups `FlushRecord`s by kv partition, builds ≤64 MB `BatchWriteRecords` batches (`MAX_WRITE_BATCH_SIZE`), and flushes **partitions concurrently but each partition serially** — `PartitionBatchCallback` chains the next batch only after the previous one completes; `SyncPutAllData`/`SyncConcurrentRequest` (max 32 in-flight) coordinate completion and support coroutine yield/resume. A merged flush buffer can contain `DataSyncTask`s from both sides of a node-group term transition. `CopyBaseToArchive`, `PutAll`, and `PutArchivesAll` each find the highest represented task term independently for every node group across the entire merged batch and skip all datastore work for that node group's lower-term tasks, even when the newer task belongs to another table bucket; a lower numeric term from another node group remains valid. The retained records in each kv partition share one term and one `UpdateCceCkptTsCc`.
+
+All checkpoint batches use `skip_wal=true`, but the completion durability contract is backend-specific. EloqStore reports a batch only after it is durable and its DSS `FlushData` is a no-op, so a non-MVCC flush publishes each completed partition's ckpt ts immediately, frees the key/payload buffers of the records that partition carried, and reports their charged flush-memory bytes so quota release matches memory actually freed. RocksDB/RocksDB-cloud writes remain in a WAL-disabled memtable: `NeedPersistKV()` is true, ckpt-ts publication stays deferred, and `PersistKV` → `FlushData` across all shards is the durability boundary. MVCC also defers publication on EloqStore until `PutArchivesAll` completes. The deferred path filters with the same batch-wide newest term used by the datastore and aggregates the retained entries into one `UpdateCceCkptTsCc` per table and node group. Synchronous helpers (`FetchTable`, `UpsertDatabase`, ...) use `SyncCallbackData` (bthread mutex/condvar, or yield/resume when provided). `UpsertTable` runs on a dedicated 1-thread `upsert_table_worker_` after pinning the node group and checking the tx term.
 
 Scans: `DataStoreServiceScanner` / `SinglePartitionScanner` (`store_handler/data_store_service_scanner.h`) implement `store::DataStoreScanner` (`tx_service/include/store/data_store_scanner.h`: `Current/MoveNext/End`) by fanning `ScanNext` RPCs over partitions and merge-sorting with the heap helpers in `store_handler/kv_store.h` (`ScanHeapTuple`, `CacheCompare`). Server-side scan sessions are identified by `session_id`.
 
@@ -148,9 +150,14 @@ per partition, serially:  BatchWriteRecords(..., skip_wal=true)
 PartitionBatchCallback ─► next batch of that partition, or
 SyncPutAllData::OnPartitionCompleted ─► resume_fn / cv when all done
   │
-  ▼ (caller, after all tables flushed)
-store_hd_->PersistKV(kv_table_names)              local_cc_shards.cpp:5964
-  └─► FlushData RPC/local on EVERY data shard  → only now is data durable
+  ├─► non-MVCC EloqStore: enqueue one UpdateCceCkptTsCc for this
+  │   partition; report completion/progress after its shard fan-in
+  └─► RocksDB-backed or MVCC: report PutAll completion without publishing
+  ▼ (caller, after archives where applicable)
+store_hd_->PersistKV(kv_table_names)              RocksDB-backed only
+  └─► FlushData RPC/local on EVERY data shard
+  ▼
+deferred UpdateCceCkptTsCc per retained table/ng    RocksDB-backed + all MVCC
 ```
 
 ### Cache-miss read (`FetchRecord`)
@@ -178,7 +185,7 @@ fetch_cc->SetFinish(0)  → re-enqueued on the owning CcShard          [03]
 ## 8. Gotchas and Invariants
 
 - **`is_range_partition` must match the table type.** Hash and range partition ids are mapped to buckets by different functions; the same integer routes to different DSS shards depending on the flag (`GetShardIdByPartitionId`). Several call sites derive it from `table_name.IsHashPartitioned()` — keep that pattern.
-- **`PutAll` alone is not durable.** Checkpoint batches set `skip_wal=true` on the DSS side; data is durable only after `PersistKV`/`FlushData` succeeds on every shard. The checkpointer must not advance ckpt-ts before `PersistKV` returns (see [07](07-durability-and-recovery.md)).
+- **`PutAll` durability depends on the backend.** EloqStore batch completion is durable; RocksDB-backed checkpoint batches set `skip_wal=true` and require `PersistKV`/`FlushData`. MVCC additionally requires archive writes. The checkpointer must not advance entry ckpt ts before the complete applicable boundary (see [07](07-durability-and-recovery.md)).
 - **Per-partition write ordering.** `PutAllImpl` allows at most one in-flight batch per kv partition; cross-partition writes are concurrent. Code that adds write paths must preserve per-partition ordering (last-writer-wins keyed by `records_ts`).
 - **Retries are bounded and not transparent.** `retry_limit_ = 2` per closure; after that the error surfaces to the caller (`PutAll` returns false; `FetchRecord` finishes the cc request with an error). `NodeGroupChanged` sharding errors crash the process today.
 - **`IsSharedStorage()` is correctness-critical**, not a hint: on shared storage a standby trusts the leader's checkpoint-ts when deciding whether an evicted entry is persistent (`cc_entry.cpp:64`); claiming shared storage on a local-disk backend would let standbys evict unpersisted data. Note the colocated DSS client returns true for EloqStore and `IsCloudMode()` for RocksDB variants — plain `ELOQDSS_ROCKSDB` colocated is *not* shared.