# Scheduler per-keyspace pending write quota plan

## Problem
`SchedulerInner.running_write_bytes` is a global counter. A single keyspace can
accumulate large pending writes and exhaust the global limit, which rejects
writes from other keyspaces.

## Goals
- Prevent one keyspace from starving others due to pending write bytes.
- Preserve the global hard cap to protect memory.
- Keep the change localized to the scheduler and avoid heavy refactors.

## Non-goals
- Full resource-control integration for write throttling (future work).
- Per-region fairness (handled separately by flow control).

## Design overview
Add a per-keyspace pending-write counter and enforce a per-keyspace threshold
in `too_busy` before rejecting a request. Keep the existing global threshold as
an upper bound. Use the keyspace id from the request context before the
too-busy check; the latest clients set it correctly, so no pre-built lock is
required.

## Data structures
Update `SchedulerInner` (`src/storage/txn/scheduler.rs`):
- Keep `running_write_bytes: CachePadded<AtomicUsize>` (global).
- Add `running_write_bytes_by_keyspace: papaya::HashMap<u32, CachePadded<AtomicUsize>>`.
- Optional: track active keyspace count if we use auto fair-share thresholds.

Helper methods:
- `inc_running_write_bytes(keyspace_id, bytes) -> (global, per_ks)`
- `dec_running_write_bytes(keyspace_id, bytes) -> (global, per_ks)`
  - After decrement, attempt to remove the entry if the value hits zero to avoid
    unbounded map growth. If another thread increments concurrently, removal is
    best-effort and can be ignored.

## Keyspace id source
Update `run_cmd` in `src/storage/txn/scheduler.rs`:
- Use `cmd.ctx().get_keyspace_id()` for per-keyspace quota checks.
- Do not pre-build `Lock` solely to derive keyspace id.

Rationale: the latest clients set the keyspace id correctly in the request
context, so relying on `ctx.get_keyspace_id()` is acceptable and keeps the fast
path simple.

## Busy check logic
Change `SchedulerInner::too_busy` to accept keyspace id:
- Signature: `fn too_busy(&self, keyspace_id: u32, region_id: u64) -> bool`
- Logic:
  - If global pending bytes >= global threshold -> busy.
  - Else if per-keyspace pending bytes >= per-keyspace threshold -> busy.
  - Else if `flow_controller.should_drop(region_id)` -> busy.

Keep existing `cmd.need_flow_control()` guard so high-priority/system commands
skip the check.

## Per-keyspace threshold policy
Use a fixed per-keyspace threshold derived from the global threshold:
- `per_keyspace_threshold = global_threshold * 0.3`
- With this ratio, up to three write-heavy keyspaces can be active before the
  global cap is approached.

No additional config knobs are introduced for this plan.

## Accounting changes
Update `SchedulerInner::new_task_context` and `dequeue_task_context`:
- On enqueue: increment both global and per-keyspace counters.
- On dequeue: decrement both.
- Continue to update `SCHED_WRITING_BYTES_GAUGE` with the global value.

## Metrics
Avoid high-cardinality per-keyspace gauges in Prometheus by default.
Instead:
- Add a counter for "too busy due to keyspace limit" vs "global limit" to
  aid debugging.
- Optionally add a debug-only log (or tracing event) when a keyspace hits
  the per-keyspace limit with the keyspace id and pending bytes.

## Tests
Add unit tests in `src/storage/txn/scheduler.rs`:
1. Two keyspaces:
   - Set global threshold to a known value and rely on the fixed 30% ratio.
   - Ensure heavy writes in keyspace A exceed the per-keyspace threshold but
     stay below the global cap, so only A hits `SchedTooBusy` while B passes.
2. Global cap still enforced:
   - Use multiple keyspaces each below the 30% threshold.
   - Ensure the combined pending bytes exceed the global threshold and trigger
     `SchedTooBusy` due to the global cap.
3. Keyspace id source:
   - Verify per-keyspace quota checks use the keyspace id from request context.

## Rollout and compatibility
- Default behavior remains safe: global threshold still enforced.
- With a fixed ratio, per-keyspace behavior is predictable and stable across
  workloads.

## Implementation steps
1. Add per-keyspace counters to `SchedulerInner` and helper methods.
2. Update `run_cmd` to use the request context keyspace id and run the
   per-keyspace busy check without pre-generating locks.
3. Wire per-keyspace accounting into task enqueue/dequeue.
4. Derive the per-keyspace threshold from the global threshold (30% ratio).
5. Add tests for fairness and correctness.
6. Add metrics/counters to distinguish per-keyspace vs global busy rejections.