# Replication Worker Maintainer Guide

## Scope and assumptions
- This document describes replication_worker internals that are not obvious from
  reading the code directly.
- It assumes you already know distributed KV serving, TiCDC, and PD at a high level.
- It focuses on design intent, ordering guarantees, and failure handling.
- It avoids restating trivial structs, configs, or API schemas.

## Why replication_worker exists
- It provides a CDC-compatible event stream without running a full storage-node Raft stack.
- It reconstructs committed state by replaying WAL from upstream stores into a
  local merged_engine instance.
- It exposes a TiCDC-like gRPC endpoint and proxies TiCDC HTTP API requests.
- It manages keyspaces and changefeeds, including GC safepoints.

## Mental model
- Think of replication_worker as a CDC capture that owns a synthetic store.
- The synthetic store is the merged_engine, which replays Raft logs gathered
  from multiple upstream stores.
- CDC clients (TiCDC) connect to replication_worker directly and never talk to
  upstream CSE nodes.
- The worker also fakes enough PD state (store and region heartbeats) for the
  replication PD (rep-pd) to accept region metadata.

## Core modules and responsibilities
- `worker.rs` drives the main loop, keyspace orchestration, CDC register state
  machines, WAL syncing, and resolved-ts emission.
- `unsafe_recover.rs` contains unsafe-recover clear validation, execution, and
  CDC cleanup paths for regions/keyspaces.
- `apply_observer.rs` converts apply write batches into CDC events and lock
  tracking updates.
- `delegate.rs` keeps per-region request state, buffered events, and the region
  resolver used for resolved-ts.
- `wal.rs` defines WAL target selection (`WalTargetFetcher`) and WAL config.
- `wal_fetcher.rs` fetches WAL bytes for targets, spools results, and emits
  `FetchedWalsBatch` to the worker.
- `wal_spool.rs` manages per-store WAL spool/cache directories under the worker
  dir (and best-effort cleanup on startup).
- `safepoint.rs` synchronizes changefeed checkpoints with PD service safepoints.
- `scheduler.rs` is the HTTP front door that forwards to TiCDC and schedules
  internal tasks.
- `kube.rs` and `provisioned.rs` create keyspace PD/CDC services.
- `util.rs` contains TiCDC request helpers, keyspace key range helpers, and
  resolved-ts stats helpers.

## Startup and recovery
- `ReplicationWorker::new` bootstraps merged_engine and loads its keyspace
  states from the manifest.
- When `replication_worker.lazy_load.enabled` is enabled, it initializes
  merged_engine with `MergedEngineInitOpts { load_shards_mode: LoadShardsMode::Lazy { pinned_policy: Arc<dyn PinnedShardPolicy> } }`
  so kvengine eagerly loads a bounded set of pinned shards (e.g. TiDB meta
  shards) on startup while keeping the rest unloaded; non-pinned shards are
  loaded on-demand when TiCDC registers regions.
- In lazy shard mode, it also pins and loads TiDB meta shards (the `m` prefix)
  per enabled keyspace so that TiCDC schema/meta reads via `kv_get`/`kv_scan`
  do not depend on CDC `Register`.
  - Pinned shard loads are event-driven:
    - Topology changes are observed by a `CoprocessorHost` observer
      (`PinnedRegionObserver`) which schedules loads only for pinned regions.
    - `kv_get` / `kv_scan` schedule a best-effort pinned load when a meta read
      observes a missing shard.
      - For `kv_scan`, the request range is first clamped to the target
        region bounds from rfengine before applying the pinned policy. This
        avoids broad/empty scan ranges misclassifying non-meta regions.
    - Failed pinned loads are not retried automatically; the next eligible
      trigger (a future `kv_get` / `kv_scan` or a region change) will schedule a
      new best-effort load attempt.
- For each keyspace with persisted state, it recreates the keyspace service and
  starts reporting regions to the keyspace PD.
- It starts a local GC runner that cleans up local files created by the merged
  engine.
- It cleans up stale WAL spool/cache directories from previous runs (best
  effort).
- It creates a gRPC server exposing `ChangeData` and `Tikv` services.
- It initializes `CdcApplyObserver` and wires it into the merged_engine apply
  pipeline.
- It constructs `ServiceSafepointManager` with existing keyspaces and lets it do
  an immediate TiCDC sync to reconstruct safepoint state.
- It intentionally delays starting the WAL fetch pipeline (including target
  fetching) until `run()` so that the runtime is ready.

## Main run loop summary
- The loop is message-driven and time-driven.
- It drains `CdcMsg` from a channel and processes them synchronously.
- It periodically evaluates `should_sync()` based on `sync_interval`.
- When there are new events or a sync tick, it emits resolved-ts and attempts
  to advance merged_engine.
- When a WAL target completes, it updates `last_update_ts` and notifies
  `WalFetcher` (via `appended_target_ts` in a fetch request) so it can advance
  to the next target.

## Threading and runtime model
- The main loop is synchronous, but it spawns async tasks for IO and scans.
- Incremental scans run on the tokio runtime and are rate-limited by a
  semaphore.
- Lock scans run on a blocking thread to avoid starving async tasks.
- WAL chunk spooling writes are offloaded to a blocking thread (`spawn_blocking`)
  so large downloads do not block tokio worker threads.
- HTTP requests to TiCDC and stores are performed by async tasks; failures are
  reported back through callbacks.
- The apply observer defers old-value reads onto the runtime to keep apply
  latency low.

## Keyspace services and their lifecycle
### Provisioned vs Kubernetes mode
- Provisioned mode expects explicit `pd_url` and `cdc_addr` per keyspace.
- Kubernetes mode derives PD/CDC addresses from StatefulSet templates.
- In Kubernetes mode, `KeyspaceKubeService` overwrites persisted PD/CDC fields
  to follow the latest deployment layout.
- Both modes run `bootstrap` against the keyspace PD to ensure the cluster is
  initialized with the merged store ID.

### LocalProvider test harness
- `provisioned.rs::local_provider::LocalProvider` is a separate test-only helper
  behind `testexport`; it is used by random replication tests to run local
  `rep-pd`, `rep-cdc`, and `rep-tidb` binaries.
- Each helper process is supervised by exactly one watcher thread that owns the
  `Child`. The shared watcher implementation now lives in
  `tikv_util::sys::process`; `LocalProvider` only builds process-specific spawn
  specs and keeps synchronous supervisor handles.
- Any unexpected helper-process exit is fatal, even if the exit status is `0`.
- `destroy()` and `restart_local_pd()` use bounded stop/restart paths:
  - they snapshot liveness with `try_wait()` before `SIGTERM`
  - they reject any non-zero intentional stop status
  - they fall back to `SIGKILL` after a fixed timeout and still report failure
- TiCDC exit code `1` is not special-cased anymore; intentional stop is success
  only for all three helper processes.
- The random replication test must install `tikv_util::set_panic_hook(false, ...)`
  before starting `LocalProvider`, otherwise a watcher-thread panic would not
  terminate the whole test process reliably during the early startup window.
- On Linux, child spawn uses `PR_SET_PDEATHSIG=SIGKILL` so a fail-fast parent
  exit does not leave local helper processes orphaned. Other Unix targets skip
  this best-effort hook because `prctl` is unavailable there.

### Add keyspace flow (high level)
- The scheduler receives `POST /cdc/keyspace` and sends `CdcMsg::AddKeyspace`.
- The worker starts the keyspace service asynchronously on the runtime.
- It validates that the keyspace TiCDC endpoint is ready before finishing the
  add flow.
- In eager mode (default):
  - It pauses meta pack compaction to prevent prepared shard files from being
    compacted mid-load.
  - It prepares shard metas twice to cover concurrent raft updates.
  - It loads shards into merged_engine and persists keyspace states.
- In lazy shard loading mode (`replication_worker.lazy_load.enabled = true`):
  - It persists keyspace states and starts reporting regions without preparing
    or loading kvengine shards for the full keyspace.
  - Shards are materialized per-region on CDC register.
- It starts a periodic region-report loop to rep-pd for that keyspace.
- It registers the keyspace with the safepoint manager.

### Load shard metas, why two passes
- During keyspace add, rfengine continues to append WAL.
- The first pass prepares a consistent set of `ShardMeta` from the snapshot.
- The second pass reduces the chance of a missed update between preparation and
  load.
- This avoids subtle holes where a shard exists in rfengine but not in
  kvengine, which would later break CDC reads and apply.
- In lazy shard loading mode, this keyspace-wide preparation is intentionally
  skipped to keep metadata/file prefetch bounded; shard metas are prepared on
  demand for registered regions.

### Remove keyspace flow
- The scheduler receives `DELETE /cdc/keyspace` and sends `CdcMsg::RemoveKeyspace`.
- The scheduler queries TiCDC with `GET /api/v2/changefeeds` before
  scheduling removal.
  - If TiCDC still reports changefeeds, removal is rejected unless `force=true`.
  - If the TiCDC query itself fails, removal is rejected unless `force=true`.
- The TiCDC check is a best-effort guard only; a concurrent add-changefeed can
  still race after the list call and before local teardown, which is acceptable
  for now.
- It unregisters the keyspace from safepoint management.
- It removes all regions from merged_engine and drops the keyspace manifest
  state.
- It destroys PD/CDC services (in Kubernetes mode) asynchronously.

### Keyspace state persistence
- Keyspace states are stored in the merged_engine manifest as opaque bytes.
- Persisted state only carries keyspace service discovery/configuration data
  (`pd_url`, `cdc_addr`, StatefulSet names).
- Changefeed existence is **not** persisted locally; TiCDC is the source of
  truth and safepoint state is rebuilt from TiCDC on startup.
- The code treats missing keyspaces as a clean slate and ignores dangling
  regions that are not in the manifest.

## CDC connections and requests
### gRPC event feed sessions
- The gRPC `ChangeData::event_feed` handler creates a per-connection CDC
  channel with a memory quota.
- The memory quota is the primary backpressure mechanism for CDC message
  delivery.
- Each connection is tracked in `conns` and its subscribed regions are tracked
  in `conn_regions`.
- Both the request stream and event sink trigger a `Deregister::Conn` on close.

### Deregister paths and cleanup
- `Deregister::Conn` removes all requests for the connection across regions.
- `Deregister::Request` removes a request from every region it subscribed to.
- `Deregister::Region` removes a request for a single region only.
- When a region has no remaining requests, its delegate is dropped to free
  resolver state and buffered events.

### Register request handling goals
- Ensure no duplicate or out-of-order events reach TiCDC during initialization.
- Ensure resolved-ts only advances when all locks are accounted for.
- Avoid scanning or applying data while the region delegate is stale.
- Provide idempotent handling for repeated register requests.

### Register state machine (per region, per request)
- States: `Blocked`, `Initializing`, `Initialized`.
- New requests are `Blocked` if `checkpoint_ts` is ahead of the register gate
  timestamp.
- The register gate timestamp prefers merged_engine `safe_target_ts`; if it is
  unavailable (region not synced), it falls back to resolver `resolved_ts`.
- `Initializing` buffers events until the initial scan finishes.
- `Initialized` sends events directly to the sink.
- Each initialization is tagged by `InitId` so stale tasks can be ignored.

### Register flow in detail
- Validate region existence and epoch.
  - Eager mode validates via the loaded kvengine shard metadata.
  - Lazy mode validates via rfengine region state; if the kvengine shard is
    missing, the request is queued and resumed after on-demand shard loading.
- If the delegate exists but the region version does not match, fail fast with
  `EpochNotMatch` to force TiCDC to retry.
- Decide whether to scan locks and whether to run incremental scan.
- Flush the apply observer for the region before scanning to avoid duplicates.
- Spawn a lock scan if the delegate does not yet have a resolver.
- Spawn an incremental scan handler if this request is not a duplicate.
- The incremental scan runs on the async runtime and respects a global
  semaphore to cap concurrency.

### Lazy shard loading (on CDC register)
- If `replication_worker.lazy_load.enabled = true` and the kvengine shard is
  missing at register:
  - The worker validates the region via rfengine and ensures the keyspace is
    active.
  - The request is recorded as pending and a per-region shard load is
    scheduled (deduplicated by region).
  - Shard prepare (remote file prefetch) runs in the background; load+recover
    runs synchronously on the worker loop.
  - If shard meta is still unavailable during prepare (`ShardMetaNotReady`),
    the attempt is failed immediately and pending requests get retry-oriented
    `server_is_busy`.
    Tradeoff: this keeps worker-side control flow simple, but can increase
    TiCDC register retry churn during temporary meta lag windows.
  - Load concurrency, timeouts, and pending request counts are bounded by
    config to keep the system stable under client retries and DFS issues.
  - `tikv_cdc_rep_shard_load_total{result=*}` distinguishes `ok`, `timeout`,
    `error`, `canceled`, and `destroyed` outcomes for better operability.
- Internally, lazy-only runtime state (pinned policy, enabled keyspaces, shard
  load semaphore, pending registers, and in-flight loads) is stored in a
  worker mode enum (`Eager` vs `Lazy`) so eager mode does not maintain lazy
  bookkeeping.
- Separately from CDC register-driven loads, the worker also pins TiDB meta
  shards per keyspace and schedules the same two-stage shard load so `kv_get`
  / `kv_scan` can serve TiDB schema/meta reads even when most table shards
  remain lazy.
- Merge-source delegate cleanup is event-driven: merged_engine emits a
  `RegionChangeEvent::Destroy` when `commit_merge` is accepted, and the worker
  handles `CdcMsg::RegionDestroyed` to fail pending lazy registers and remove
  stale delegates with `region_not_found`.

### Resuming blocked registrations
- When merged_engine `safe_target_ts` advances past a blocked request's
  `checkpoint_ts`, the worker resumes initialization using
  `CdcMsg::ResumeRegister`.
- Resume is evaluated in `send_resolved_ts`, so it is delayed if the region is
  not synced/present in merged_engine or if lock scanning has not completed
  (resolver is still `Pending`).
- Resume uses the same snapshot version as the delegate to avoid mixing epochs.
- The apply observer is flushed again before resume to avoid duplicate events.
- If the connection is already closed, the resume is safely ignored.

### Why we flush the apply observer before scanning
- The apply observer may have pending write batches for the region.
- If we scan locks or perform an incremental scan without flushing, we can
  duplicate events or miss lock tracking transitions.
- Flushing ensures a clean boundary between historical scan results and
  subsequent apply-driven events.

### Incremental scan handler details
- Uses `new_delta_write_iterator_async(checkpoint_ts)` over the write CF.
- Skips index keys and deletes, and ignores versions <= `checkpoint_ts`.
- Computes `old_value` by reading at `commit_ts - 1` to match TiCDC semantics.
- Batches rows to avoid huge events and uses a barrier to enforce ordering.
- Emits a final `Initialized` event and then sends `RegisterResult`.

### Why the barrier is required
- TiCDC expects resolved-ts to advance only after the scan is fully delivered.
- The barrier ensures the sink has flushed all scan events before we send the
  initialization completion signal.
- Without it, resolved-ts could overtake the scan and violate TiCDC ordering
  assumptions.

## Lock scanning and region resolver
### Initial lock scan
- The lock scan iterates the region lock CF and collects lock keys and start_ts.
- It ignores index keys and non-Put/Delete locks to reduce noise.
- It reports results back to the worker with the snapshot version used.

### Pending resolver state
- Before the scan completes, the resolver is in `Pending` mode.
- Lock track/untrack events from the apply observer are queued in memory.
- The pending list is converted to a real `Resolver` once scan results arrive.
- This avoids losing lock updates that occur during the scan window.

### Resolver upgrade and safety
- The scan result is applied only if the snapshot version matches the pending
  resolver's expected version.
- Stale scan results are discarded to avoid corrupting lock state.
- On scan error, the resolver is removed and errors are broadcast to the
  region's requests.

## Resolved-ts computation and emission
### Preconditions for resolved-ts
- `last_update_ts` must be non-zero (merged_engine has advanced).
- The region must be present in merged_engine and in a synced state.
- The resolver must have completed lock scanning.

### send_resolved_ts flow
- For each region delegate, the resolver advances using
  `min(last_update_ts, safe_target_ts)`.
- Resolved-ts is recorded in the request state, and only advanced forward.
- Blocked requests are resumed once their `checkpoint_ts` is <= `safe_target_ts`.
- Resolved-ts events are grouped by `(ts, request)` so multiple regions can be
  sent in one message.
- This grouping is a throughput optimization and preserves per-request ordering.

### Why we check `region_is_synced`
- Merged_engine can track regions that have not yet applied WAL data.
- Sending resolved-ts for those regions could move TiCDC ahead of actual data.
- The `region_is_synced` gate avoids this mis-ordering.

## Apply observer pipeline
### What the observer captures
- Applied writes from the write CF as `EventRow` entries.
- Lock CF changes as tracked lock updates for the resolver.
- Admin commands that affect region topology (split/merge) to trigger
  delegate cleanup and PD reporting.

### Asynchronous old value fetch
- The observer defers fetching `old_value` to a runtime task.
- This keeps the apply path lightweight while still delivering TiCDC-required
  old values for committed rows.
- A per-region task list ensures all queued batches for a region are flushed
  together and preserve ordering.

### Flushing behavior
- `flush_region` drains only one region and is used during register handling
  and admin events.
- `flush` drains all regions and is used for global synchronization.
- Flushing emits a single `CdcMsg::Applied` per region containing all events
  and lock tracking updates for that batch.

## Admin commands and region topology
### Applied admin handling
- On split/merge/rollback, the observer flushes the region and sends an
  `AppliedAdmin` message.
- The worker reports region metadata to the keyspace PD so TiCDC can locate
  region ranges.
- For splits, it also reports the new regions created by the split requests.

### Merge handling and delegate cleanup
- `commit_merge` implies the source region is gone.
- Source cleanup is driven by merged_engine destroy signals
  (`CdcMsg::RegionDestroyed`), not by `AppliedAdmin`.
- On `RegionDestroyed`, the worker sends `region_not_found` to source region
  requests and removes the source delegate (and lazy pending-register state).
- For the target region, an `epoch_not_match` error is sent to force TiCDC to
  re-register with the new epoch.

## WAL progress fetching and target selection
### Target queue
- `WalTargetFetcher` runs in the background and appends targets to a shared
  channel.
- Each target is `(target_ts, StoreWalProgresses)` logically.
- The queue is bounded (`MAX_PENDING_TARGETS`) to avoid unlimited lag.

### Choosing rfengine vs backup
- Normally the worker tracks WAL progress from rfengine for each store.
- If the last synced target is too old, targets are fetched from backups.
- This prevents missing stores that may have joined and left between targets.
- The min/max time span bounds are a safety mechanism, not a rate limiter.

### Skipping stores
- `include_only_store_addr_keywords` is a whitelist; when it is non-empty, only
  stores whose address contains any include keyword are kept.
- `skip_store_addr_keywords` is only used when
  `include_only_store_addr_keywords` is empty.
- It is applied during rfengine fetches and during backup target construction.
- This is used to ignore stores that should not be part of replication.
- The same skip/include logic is also reused by the admin-only
  `POST /cdc/unsafe_recover/clear` precheck. A region can only be cleared after all
  of its current upstream peers are already outside the active replay store set.

### Tolerated store errors
- Both rfengine fetch and backup selection can tolerate a limited number of
  missing stores.
- The tolerance is configurable and must match backup metadata semantics.
- If the number of missing stores exceeds tolerance, replication stops.

## WAL fetching and apply pipeline
### Worker / WalFetcher handshake
- The worker drives `WalFetcher` by sending `WalFetcherRequest::Fetch`.
- The request includes the current merged store progresses and an optional
  `appended_target_ts` to acknowledge that a target has been appended.
- `WalFetcher` turns the active `WalTarget` into a `FetchedWalsBatch` and sends
  it back to the worker (with an optional `advance_to` timestamp).
- After appending a batch, the worker sends the next fetch request immediately
  (before `sync_merged`) so fetching can overlap with apply.

### Ordering stores for update
- `get_decreasing_stores_lag` sorts stores by lag (epoch, then offset).
- `WalFetcher` selects a subset of stores per batch in round-robin order so no
  single store can starve others.
- Lag is computed against merged_engine's stored progress, not PD.
- If a store appears in a fetched target but does not yet exist in the merged_engine
  manifest (e.g. scale-out), the worker initializes its progress to `(epoch=1,
  offset=0)` on first apply so `MergedEngine::update_wal` has a baseline.

### Global WAL size limit
- `wal.update_stores_wal_size_limit` bounds the expected total WAL bytes fetched
  in a batch.
- `WalFetcher` derives fetch concurrency from
  `wal.update_stores_wal_size_limit / wal.tikv_rfengine_wal_size` and clamps it
  by `wal.max_fetch_concurrency`.
- This bounds in-flight WAL and prevents long single-cycle stalls.

### Fetching WAL from a store
- For each store and epoch, the WAL fetcher requests WAL chunks via the store HTTP
  endpoint `/rfengine/wal_chunk`.
- The request includes `start_off` and `end_off`, where `end_off` may be zero
  (read to end) depending on epoch.
- Store WAL downloads can be throughput-limited via `wal.store-fetch-max-throughput`
  (per-store independent limiters). This protects store backends when
  replication_worker runs with high fetch concurrency.
- Store responses are streamed directly into temporary spool files under the
  worker directory (spool writes run on a blocking thread to avoid blocking the
  async runtime) to avoid large in-memory buffers.
- If the store responds `GONE`, the WAL fetcher falls back to DFS for that epoch.

### Fetching WAL from DFS
- DFS retrieval is used when the store is unavailable, the epoch is too old, or
  the store explicitly returns `GONE`.
- DFS WAL downloads can be throughput-limited via `dfs.max-read-throughput`.
  - The limiter is per-DFS-client-instance (shared by all callers using the same
    `Arc<dyn Dfs>`).
  - In cloud_worker, subsystems typically share a single DFS instance, so this
    behaves like a process-wide cap in the common deployment.
- The WAL fetcher assembles a full epoch into an in-memory `WalCache` and then
  slices ranges using `range_reader`.
- `WalCache` reduces repeated downloads when multiple ranges are needed.
- The cache is cleared once the epoch is known complete and rotated.

### Rotation rules
- Rotation is triggered when the end offset reaches the end of an epoch.
- The worker calls `merged_engine.rotate_wal` only when its store progress
  exactly matches the end offset.
- This prevents skipping data or double-applying WAL bytes.

## sync_merged and last_update_ts
- After WAL updates, `merged_engine.sync_merged` applies committed entries to
  the kv engine and advances region progress.
- The apply observer is flushed to ensure all events are visible to CDC before
  updating `last_update_ts`.
- `last_update_ts` advances only when the target is fully applied.
- This is why `send_resolved_ts` relies on `last_update_ts` rather than the
  current time.

## Changefeed creation and idempotency
### start_ts handling
- If the user passes `start_ts=0`, the worker sets it to `last_update_ts`.
- This guarantees the changefeed starts from data already materialized in the
  merged engine.
- If the user passes a non-zero `start_ts` greater than `last_update_ts`, the
  request is rejected to prevent gaps.

### Create request handling after local feed persistence removal
- `replication_worker` still validates/adjusts `start_ts` and ensures the
  short-lived per-changefeed PD safepoint before dispatching create requests.
- The worker does **not** persist the request body or changefeed ID locally.
- After the TiCDC create request finishes, the worker triggers a best-effort
  safepoint refresh for that keyspace so TiCDC-backed state is observed quickly.

### Changefeed deletion
- The scheduler forwards the delete request to TiCDC and then sends
  `CdcMsg::RemoveTask`.
- `RemoveTask` no longer mutates manifest state; it triggers a best-effort
  safepoint refresh for the keyspace.
- Repeated deletes are still treated as success by TiCDC/worker flow.

## HTTP scheduler behavior
- Requests under `/cdc/` are translated to TiCDC paths by stripping the
  prefix and forwarding with the original method and headers.
- `/cdc/keyspace` requests are intercepted to add or remove keyspaces.
- `/cdc/api/v2/changefeeds` is intercepted to inject keyspace-specific
  validation and to run start_ts safety checks.
- `/cdc/keyspace` GET returns the current set of keyspace IDs from the worker.
- `/cdc/unsafe_recover/clear` is intercepted locally and never forwarded to
  TiCDC.
  It accepts exactly one selector (`region_id` or `keyspace_id`) plus optional
  `dry_run`.
- The scheduler always returns JSON with `success` and `dry_run`. Success
  responses return the resolved target region IDs; failures return a `reason`.
- The request is scheduled onto the worker main loop as
  `CdcMsg::UnsafeRecoverClear`, so it runs in the same serialized mutation path
  as ordinary merged-engine updates.

## Unsafe recover clear flow
- This API is for deleting stale local replay state from merged_engine after
  replication_worker has intentionally stopped replaying the upstream region.
- The flow is synchronous and single-threaded:
  resolve local target regions, precheck upstream skipped stores, validate
  merged-engine dependencies, optionally stop at `dry_run`, flush the apply
  observer, execute the destroy path, then remove CDC delegates/caches.
- The upstream skipped-store precheck still runs on the worker main loop, but
  it is bounded by `unsafe_recover_clear_pd_precheck_timeout`. A timeout is
  treated as an infrastructure failure for the admin request rather than as an
  operator-input rejection.
- `region_id` only succeeds for a live local replay region. `keyspace_id`
  resolves all live local replay regions in the keyspace and sorts them for a
  deterministic operator-visible result.
- The response reports only the explicit requested region set. Internal cleanup
  may also destroy delayed tombstone parents when the final dependency edge is
  removed; that side effect is logged but not surfaced in the HTTP payload.
- The merged-engine execution step keeps the operator-visible request order
  instead of reordering dependents-first. That admin flow is expected to be
  rare, and restart can rebuild dependency edges from replay/bootstrap state if
  the process exits mid-clear.

## Safepoint management
### Ensuring start_ts safety
- During changefeed creation, the worker updates a per-changefeed PD service
  safepoint with a short TTL.
- This prevents GC from removing data between request submission and feed
  initialization.
- The safepoint is not removed later, trading cleanup for operational safety.

### Periodic TiCDC sync
- A timer task fetches changefeed lists from each keyspace TiCDC using
  `GET /api/v2/changefeeds`.
- The safepoint manager also schedules an immediate sync on startup and a
  best-effort per-keyspace refresh after create/delete operations.
- For active feeds, it updates the keyspace service safepoint to the minimal
  checkpoint_ts and extends the TTL.
- If TiCDC reports no GC-blocking feeds, the keyspace safepoint is removed.
- If TiCDC cannot be queried, the existing keyspace safepoint is left unchanged.

### Expired changefeeds
- Feeds whose checkpoint_ts is older than GC TTL are considered expired.
- The worker updates the replication PD GC safepoint to the max expired
  checkpoint_ts, which triggers TiCDC to mark those feeds failed.
- This mirrors TiCDC's GC blocking logic and prevents silent data loss.

### Keyspace removal
- When a keyspace is removed, its safepoint entry is removed as well.
- This avoids holding GC back for a keyspace that no longer exists.

## PD reporting loops
- The worker periodically re-reports region metadata to rep-pd.
- This handles rep-pd restarts that lose in-memory region leaders.
- Reporting uses trimmed keyspace prefixes so rep-pd sees per-keyspace ranges.
- The store heartbeat is also periodically sent to keep the merged store alive.

## Read RPCs (kv_get / kv_scan)
- The gRPC `Tikv` service serves `kv_get`/`kv_scan` from the local kvengine
  snapshot materialized by merged_engine.
- It validates `context.region_id` and `context.region_epoch.version` against
  loaded kvengine shard metadata and returns a retry-oriented `region_error`
  when the shard is missing or stale.
- Request keys/ranges are "inner" keys (without keyspace prefixes); the service
  prepends keyspace prefixes based on the loaded shard and trims them from
  reply keys.
- For `kv_scan`, an empty `end_key` means "scan to this region's end" (not
  "scan across regions").
- `kv_scan` / `kv_scan_lock` clamp request ranges to shard bounds before
  iterating storage so out-of-bound request keys cannot scan across shards.
- **Consistency caveat:** these read RPCs query the local materialized state at
  the requested `version`. Callers must not assume the results are consistent
  with the CDC event stream unless they ensure the read ts is not ahead of the
  worker's replication progress.
- With `replication_worker.lazy_load.enabled = true`, reads require the target
  shard to be loaded.
  TiDB meta shards are pinned and loaded proactively so schema/meta reads work
  without CDC `Register`, while non-meta shards remain lazy.

## Backpressure and resource control
- CDC channels are memory-quotad to prevent unbounded buffering.
- Incremental scans are capped by a global semaphore.
- WAL targets are capped by a bounded channel (`MAX_PENDING_TARGETS`).
- WAL fetches are capped by `wal.max_fetch_concurrency` and the
  `wal.update_stores_wal_size_limit`-derived batch sizing.
- These limits prevent cascade failures when TiCDC or upstream stores are slow.

## Error handling patterns and invariants
- Many operations return `server_is_busy` to force client retries instead of
  accepting inconsistent state. These constructions increment the shared
  `tikv_server_is_busy_total{type="replication_worker"}` counter.
- Delegate version mismatches are treated as stale and trigger retries.
- `InitId` prevents a stale scan from completing a newer registration.
- Store progress mismatches are treated as fatal because they imply WAL
  corruption or reordering.
- Safepoint errors are logged and may not be retried immediately to avoid
  blocking the main loop.

## Operational debugging cues
- If CDC stalls but merged_engine keeps syncing, check `send_resolved_ts` paths
  and resolver initialization (lock scan completion).
- If add keyspace hangs, verify CDC status and meta pack compaction pause.
- If WAL sync stalls with `StoreProgressMismatch`, check for skipped or reordered
  WAL chunk responses from stores.
- If changefeed creation fails with `start_ts too large`, check `last_update_ts`
  advancement and WAL target completion.

## Common pitfalls for maintainers
- Do not remove the apply observer flush in register or resume flows.
- Do not advance `last_update_ts` unless the target is fully applied.
- Do not emit resolved-ts for regions that are not yet synced in merged_engine.
- Do not remove the meta pack compaction pause around shard preparation and
  loading.
- Do not assume kv_get/kv_scan are lock-aware; they are not.

## Extension points and safe changes
- Adding new CDC event types should go through the apply observer and maintain
  the pending-event buffer semantics for initializing requests.
- Changing WAL target selection should preserve min/max time span semantics or
  update backup logic accordingly.
- Adding new keyspace lifecycle steps should update keyspace states in the
  manifest to survive restarts.