reset-replication: return {"elapsed_ms": u64} JSON in response body

Before: handler returned 200 with empty body. Operators had to grep
libsql logs to see how long the reset took.

After: handler returns 200 with {"elapsed_ms": N}. The streamer (and
any other caller) can parse this into a StatsD histogram for dashboard
latency tracking, without server-side log scraping.

Backwards compatible: older clients that ignore the body still work.
New clients that expect the body handle an empty body gracefully.

New turmoil test: reset_replication_response_includes_elapsed_ms
verifies the field is present and within sanity bounds.

Author: sle-c

libsql-server/src/http/admin/mod.rs

@@ -572,12 +572,18 @@ async fn handle_checkpoint<C>(
 /// - metastore config (jwt_key, block_writes, etc.) is preserved
 ///
 /// Other namespaces on this pod are completely unaffected.
+#[derive(serde::Serialize)]
+struct ResetReplicationResp {
+    /// Wall-clock duration of the reset, for operator-visible metrics.
+    elapsed_ms: u64,
+}
+
 async fn handle_reset_replication<C>(
     State(app_state): State<Arc<AppState<C>>>,
     Path(namespace): Path<NamespaceName>,
-) -> crate::Result<()> {
-    app_state.namespaces.reset_replication(namespace).await?;
-    Ok(())
+) -> crate::Result<axum::Json<ResetReplicationResp>> {
+    let elapsed_ms = app_state.namespaces.reset_replication(namespace).await?;
+    Ok(axum::Json(ResetReplicationResp { elapsed_ms }))
 }
 
 #[derive(serde::Deserialize, Default)]

libsql-server/src/namespace/store.rs

@@ -262,7 +262,7 @@ impl NamespaceStore {
     /// Brief unavailability window: from the moment we take the write lock
     /// until `make_namespace` returns. Other namespaces on the pod are
     /// completely unaffected.
-    pub async fn reset_replication(&self, namespace: NamespaceName) -> crate::Result<()> {
+    pub async fn reset_replication(&self, namespace: NamespaceName) -> crate::Result<u64> {
         if self.inner.has_shutdown.load(Ordering::Relaxed) {
             return Err(Error::NamespaceStoreShutdown);
         }
@@ -393,11 +393,11 @@ impl NamespaceStore {
             .await?;
         lock.replace(ns);
 
-        let elapsed_ms = start.elapsed().as_millis();
+        let elapsed_ms = start.elapsed().as_millis() as u64;
         tracing::info!(
             "reset_replication: rebuilt replication log for namespace {namespace} in {elapsed_ms}ms"
         );
-        Ok(())
+        Ok(elapsed_ms)
     }
 
     // This is only called on replica

libsql-server/tests/namespaces/mod.rs

@@ -347,6 +347,54 @@ fn integrity_check_defaults_to_quick_when_full_omitted() {
     sim.run().unwrap();
 }
 
+#[test]
+fn reset_replication_response_includes_elapsed_ms() {
+    // Operators (and ops dashboards) want to see how long reset took
+    // without having to grep logs. The JSON response body includes
+    // elapsed_ms, which must be a positive integer for any successful
+    // reset.
+    let mut sim = Builder::new()
+        .simulation_duration(Duration::from_secs(1000))
+        .build();
+    let tmp = tempdir().unwrap();
+    make_primary(&mut sim, tmp.path().to_path_buf());
+
+    sim.client("client", async {
+        let client = Client::new();
+        client
+            .post("http://primary:9090/v1/namespaces/elapsed/create", json!({}))
+            .await?;
+
+        let db = Database::open_remote_with_connector(
+            "http://elapsed.primary:8080",
+            "",
+            TurmoilConnector,
+        )?;
+        let conn = db.connect()?;
+        conn.execute("create table t(v text)", ()).await?;
+        conn.execute("insert into t values ('x')", ()).await?;
+
+        let resp = client
+            .post(
+                "http://primary:9090/v1/namespaces/elapsed/reset-replication",
+                json!({}),
+            )
+            .await?;
+        assert_eq!(resp.status(), hyper::http::StatusCode::OK);
+        let v = resp.json_value().await?;
+        let elapsed = v["elapsed_ms"].as_u64().expect("elapsed_ms must be u64");
+        // We can't assert a specific value (depends on hardware) but we
+        // can assert it's present and non-negative. In practice this is
+        // typically single-digit milliseconds for an empty/small table.
+        // Max bound guards against a runaway regression.
+        assert!(elapsed < 30_000, "elapsed_ms={elapsed} exceeds 30s sanity bound");
+
+        Ok(())
+    });
+
+    sim.run().unwrap();
+}
+
 #[test]
 fn reset_replication_is_idempotent() {
     // An operator (or the streamer's retry-after-reset path) may call