Address review feedback on TSO waker PR

Signed-off-by: Michael Ingley <mingley@linkedin.com>

Author: Michael Ingley

benches/tso_waker_policy.rs

@@ -7,7 +7,9 @@ use criterion::{criterion_group, criterion_main, BenchmarkId, Criterion};
 use futures::task::AtomicWaker;
 
 const MAX_PENDING_COUNT: usize = 1 << 16;
+// Every `FULL_EVERY` iterations, the synthetic queue is reset to "full".
 const FULL_EVERY: u64 = 1024;
+// Within each `FULL_EVERY` window, keep queue full for `FULL_WINDOW` iterations.
 const FULL_WINDOW: u64 = 16;
 
 struct NoopWake;
@@ -30,6 +32,7 @@ fn response_policy_old(iterations: u64) -> Duration {
         }
         black_box(pending_len >= MAX_PENDING_COUNT);
         pending_len = pending_len.saturating_sub(1);
+        // Old behavior: wake unconditionally for every response batch.
         atomic_waker.wake();
     }
     start.elapsed()
@@ -49,6 +52,7 @@ fn response_policy_new(iterations: u64) -> Duration {
         let was_full = pending_len >= MAX_PENDING_COUNT;
         pending_len = pending_len.saturating_sub(1);
         let should_wake = was_full && pending_len < MAX_PENDING_COUNT;
+        // New behavior: wake only for full -> non-full transition.
         if black_box(should_wake) {
             atomic_waker.wake();
         }

doc/tso_waker_criterion.md

@@ -11,6 +11,10 @@ Rust toolchain: 1.84.1
 Quantify the latency impact of reducing TSO stream wake/registration churn in
 `src/pd/timestamp.rs`.
 
+These numbers are a point-in-time snapshot from this branch. Re-run the
+benchmark after meaningful changes to either `src/pd/timestamp.rs` or
+`benches/tso_waker_policy.rs`.
+
 ## Method
 
 Benchmark framework:
@@ -34,6 +38,10 @@ The benchmark compares old vs new policies in two isolated hot paths:
 - `response/*`: wake policy when processing responses
 - `register/*`: self-waker registration policy in no-request branch
 
+Note: the old/new response benchmarks intentionally do asymmetric work
+(always wake vs transition-only wake), so the speedup reflects the amortized
+benefit of skipping redundant wake calls under this simulation pattern.
+
 ## Results (Absolute Latency)
 
 From Criterion output (`time` line):

src/pd/timestamp.rs

@@ -22,6 +22,7 @@ use futures::task::Context;
 use futures::task::Poll;
 use log::debug;
 use log::info;
+use log::warn;
 use pin_project::pin_project;
 use tokio::sync::mpsc;
 use tokio::sync::oneshot;
@@ -58,8 +59,13 @@ impl TimestampOracle {
         let pd_client = pd_client.clone();
         let (request_tx, request_rx) = mpsc::channel(MAX_BATCH_SIZE);
 
-        // Start a background thread to handle TSO requests and responses
-        tokio::spawn(run_tso(cluster_id, pd_client, request_rx));
+        // Start a background task to handle TSO requests and responses.
+        // If it exits with an error, log it explicitly so root cause is preserved.
+        tokio::spawn(async move {
+            if let Err(err) = run_tso(cluster_id, pd_client, request_rx).await {
+                warn!("TSO background task exited with error: {:?}", err);
+            }
+        });
 
         Ok(TimestampOracle { request_tx })
     }