refactor(mcp)!: drive Engine transactions with the RAII guard, and flatten client::Error

MIGRATING-0.3.md

@@ -19,7 +19,7 @@ The public `hyperdb_api::Error` type was redesigned into a flat enum per the [Mi
 | `Error::new(msg)`                    | Constructor deleted. Use a specific variant or `Error::internal(msg)` (see below). |
 | `Error::with_cause(msg, e)`          | Constructor deleted. For an `io::Error` cause use `Error::connection_with_io(msg, e)`; otherwise fold the cause into a message string. |
 | `Error::kind() -> Option<ErrorKind>` | Method deleted. Match directly on the enum. |
-| `pub use ... ErrorKind` from `hyperdb_api`              | Re-export removed. The `ErrorKind` type is internal to `hyperdb-api-core` and not part of `hyperdb-api`'s public surface. |
+| `pub use ... ErrorKind` from `hyperdb_api`              | Re-export removed, and the type itself is gone: `hyperdb-api-core`'s internal `client::Error` was later flattened the same way, so there is no `ErrorKind` anywhere in the workspace. |
 
 ### What's new
 
@@ -184,7 +184,8 @@ if let Error::Server { sqlstate: Some(code), detail, hint, .. } = &err {
 
 ### Notes for downstream crate authors
 
-- The `From<hyperdb_api_core::client::Error> for hyperdb_api::Error` impl is exhaustive over `client::ErrorKind`. Adding a kind to `client::Error` will break this build until a mapping is added. This is intended.
+- The `From<hyperdb_api_core::client::Error> for hyperdb_api::Error` impl is a variant-to-variant match: `client::Error` is a flat enum too ([#75](https://github.com/tableau/hyper-api-rust/issues/75)). Unlike the public enum it feeds, `client::Error` is deliberately **not** `#[non_exhaustive]`, so that match has no wildcard arm and stays exhaustive.
+  A variant added upstream therefore breaks this build until it is given a deliberate public mapping, rather than silently degrading to `Error::Internal`. That check is the reason for the asymmetry: `client::Error` is internal and has exactly one consumer, shipped from the same workspace, so it gains nothing from forward-compatibility and would lose the compiler's help where it matters most.
 - `Error::Connection { source }` carries an `Option<std::io::Error>`. The wire-protocol layer in `hyperdb-api-core` does not preserve typed causes through its boundary, so `source` is `None` for errors that originated there. Direct callers in `hyperdb-api` who construct `Error::connection_with_io` *do* preserve the typed source.
 - The `Error::Internal { .. }` variant is a deliberate catch-all for invariant violations. New code should reach for a domain variant first.
 
@@ -252,7 +253,13 @@ Callers that hold a pooled connection (`deadpool::managed::Object<ConnectionMana
 
 ### MCP follow-up
 
-The MCP server's `Engine::execute_in_transaction` helper takes `&self` and so cannot use the RAII guard. It retains the deprecated raw methods with a function-level `#[allow(deprecated, reason = "...")]` annotation. Migrating it requires reshaping `Engine`'s locking model. Two structural paths and an acceptance-criteria checklist are written up in [issue #72](https://github.com/tableau/hyper-api-rust/issues/72).
+The MCP server's `Engine::execute_in_transaction` helper originally took `&self` and so could not use the RAII guard. [Issue #72](https://github.com/tableau/hyper-api-rust/issues/72) closed that gap: the helper now takes `&mut self`, holds a `Transaction`, and hands its closure an `EngineTransaction` view instead of `&Engine`.
+
+Two in-tree holdouts remain, both by design rather than oversight — they are the `&self` helpers the `*_unguarded` methods were added for:
+
+- **`KvStore`** (`kv_store.rs`, in `pop`, `set_batch`, `set_batch_if_absent`) holds `connection: &'conn Connection`, a *shared* reference. `Connection::transaction()` needs `&mut self`, so adopting the guard would mean threading `&mut` out through `Connection::kv_store()` — a public breaking change that would also stop callers from opening two stores at once.
+  These paths do pair every begin with a commit or a best-effort rollback, so the obligation is discharged on the `Ok` and `Err` paths; a **panic** between them would leak an open transaction.
+- **`AsyncKvStore`** (`async_kv_store.rs`, same three operations) has the same shape and additionally cannot be rescued by a guard at all: Rust has no async `Drop`, which is why `AsyncTransaction`'s own `Drop` only warns. A future cancelled between the begin and the commit leaves the transaction open until the next command on that connection — the cancellation hazard described above, unmitigated.
 
 ---
 

docs/TRANSACTIONS.md

@@ -169,11 +169,22 @@ if result.is_ok() {
 Pairing is entirely on the caller, on **every** path including panics and
 cancelled futures. An unmatched begin wedges the session: later statements fail
 with "transaction already in progress" on a connection that is otherwise
-healthy, so reconnect logic won't clear it. The MCP server's
-`engine.rs::execute_in_transaction` is exactly this case and wraps its closure
-in `catch_unwind` to roll back before resuming an unwind; it stays on the
-unguarded methods until [issue #72](https://github.com/tableau/hyper-api-rust/issues/72)
-restructures `Engine`'s lock model.
+healthy, so reconnect logic won't clear it. That burden is why the MCP server's
+`engine.rs::execute_in_transaction` moved off these methods in
+[issue #72](https://github.com/tableau/hyper-api-rust/issues/72) — it now holds
+a `Transaction` and lets `Drop` discharge the obligation.
+
+Two in-tree callers remain, and they are the reason these methods exist:
+`KvStore::{pop, set_batch, set_batch_if_absent}` and their `AsyncKvStore`
+counterparts. Both hold `connection: &'conn Connection` — a shared reference —
+so the guard's `&mut self` is unavailable without making
+`Connection::kv_store()` take `&mut self`, which would break the public API and
+forbid two open stores on one connection. Both pair every begin with a commit
+or a best-effort rollback, so the `Ok` and `Err` paths are covered; what is
+*not* covered is a panic (sync) or a cancellation (async) landing between the
+two. The async three have no remedy even in principle — no async `Drop` — so
+treat them as the worked example of the hazard this section describes, not as
+code awaiting a mechanical fix.
 
 ## Test Inventory
 

hyperdb-api-core/CHANGELOG.md

@@ -29,6 +29,52 @@ and this project adheres to [Semantic Versioning](https://semver.org/).
 
 ### Changed
 
+- **BREAKING:** `client::Error` is now a flat enum — one
+  variant per failure mode, matched directly — rather than a struct with a
+  `kind` discriminator and a `Box<dyn StdError>` cause channel
+  ([#75](https://github.com/tableau/hyper-api-rust/issues/75)). This applies
+  the shape `hyperdb_api::Error` already adopted in
+  [#70](https://github.com/tableau/hyper-api-rust/issues/70) per the
+  [Microsoft Pragmatic Rust Guidelines](https://microsoft.github.io/rust-guidelines/)
+  (M-ERRORS-CANONICAL-STRUCTS, M-ERRORS-AVOID-WRAPPING-AND-AS-DYN).
+
+  **No effect on `hyperdb-api`'s public API** — `client::Error` and
+  `client::ErrorKind` were never re-exported from it, and this crate is
+  documented above as internal.
+
+  - `client::ErrorKind` is **removed**, along with `Error::kind()`,
+    `Error::new()`, `Error::with_cause()`, and `Error::new_with_details()`.
+  - Every variant has a snake_case constructor taking `impl Into<String>`:
+    `connection`, `authentication`, `query`, `protocol`, `io`, `config`,
+    `timeout`, `cancelled`, `closed`, `conversion`, `feature_not_supported`,
+    `other`. `Error::closed()` and `Error::timeout()` previously took no
+    arguments and supplied a canned message; they now take one.
+  - `Error::io(err: io::Error)` became `Error::from_io(err: io::Error)`, so
+    the name `io` could be the message-taking constructor like its peers.
+    `impl From<io::Error>` is unchanged.
+  - `Connection`, `Cancelled`, and `Closed` carry `{ message, sqlstate }`;
+    `Query` carries `{ message, sqlstate, detail, hint }`. Those are exactly
+    the variants that could hold a SQLSTATE before, so `sqlstate()`,
+    `detail()`, `hint()`, and `message()` return what they used to **on
+    those variants**.
+  - The enum is deliberately **not** `#[non_exhaustive]`, unlike the public
+    `hyperdb_api::Error`. This type is internal with a single in-workspace
+    consumer, so it gains nothing from forward-compatible matching, while
+    staying exhaustive keeps the compile-time check on
+    `From<client::Error> for hyperdb_api::Error` — a new variant must be
+    given a deliberate public mapping instead of silently degrading to
+    `Error::Internal`.
+  - **Narrower diagnostics on four variants.** `Authentication`,
+    `FeatureNotSupported`, `Timeout`, and `Other` are single-string, so the
+    gRPC error path folds any server `detail` into the message and
+    **discards `hint` and `sqlstate`** on those four; the old
+    `new_with_details` stored all three regardless of kind. No caller
+    observes the loss — the public `hyperdb_api::Error` mapping already
+    discarded `hint` and `sqlstate` on the arms these feed — but the stored
+    data is genuinely narrower, not merely reshaped.
+  - `Error::with_cause` had no call sites, so dropping the `Box<dyn>` channel
+    loses no information. `source()` now returns `None` for I/O errors; the
+    public `hyperdb_api::Error` mapping already discarded that cause.
 - The all-binary `Bind` path now sends a **single** parameter format code
   rather than one per parameter. The PostgreSQL protocol broadcasts a lone
   format code across every parameter, so this is wire-compatible and drops the
@@ -56,6 +102,17 @@ and this project adheres to [Semantic Versioning](https://semver.org/).
 
 ### Fixed
 
+- An `Error` built from an `io::Error` no longer renders its message twice.
+  `Error::io` stored the same text as both `message` and `cause`, and
+  `Display` printed both, so an I/O failure surfaced as `"refused: refused"`.
+  Dropping the cause channel removed the duplication.
+
+  This **changes the `Display` text** of I/O-origin connection errors, and
+  those reach the public `hyperdb_api::Error` (its `Connection` variant is
+  built from this message). The new text is the intended one, but anything
+  matching on the error *string* rather than the variant will see
+  `"refused"` where it saw `"refused: refused"`.
+
 - **`AuthenticatedGrpcClient::get_table_labels` and `get_column_labels` now
   report Arrow failures instead of returning a partial map.** Both iterated
   record batches with `if let Ok(batch) = batch_result`, so a decode failure

hyperdb-api-core/Cargo.toml

@@ -23,6 +23,7 @@ salesforce-auth = ["dep:hyperdb-api-salesforce", "dep:arrow"]
 bytes = { workspace = true }
 byteorder = { workspace = true }
 memchr = { workspace = true }
+thiserror = { workspace = true }
 tracing = { workspace = true }
 socket2 = { workspace = true }
 

hyperdb-api-core/src/client/async_client.rs

@@ -439,7 +439,7 @@ impl AsyncClient {
                         error = %e,
                         "query-cancel-send-failed"
                     );
-                    Error::io(e)
+                    Error::from_io(e)
                 })?;
             }
             #[cfg(unix)]
@@ -466,7 +466,7 @@ impl AsyncClient {
                         error = %e,
                         "query-cancel-send-failed"
                     );
-                    Error::io(e)
+                    Error::from_io(e)
                 })?;
             }
             #[cfg(windows)]
@@ -499,10 +499,10 @@ impl AsyncClient {
                         error = %e,
                         "query-cancel-send-failed"
                     );
-                    Error::io(e)
+                    Error::from_io(e)
                 })?;
 
-                file.flush().map_err(Error::io)?;
+                file.flush().map_err(Error::from_io)?;
             }
         }
 
@@ -608,8 +608,8 @@ impl AsyncClient {
                 let mut buf = BytesMut::with_capacity(16);
                 frontend::cancel_request(self.process_id, self.secret_key, &mut buf);
 
-                stream.write_all(&buf).map_err(Error::io)?;
-                stream.flush().map_err(Error::io)?;
+                stream.write_all(&buf).map_err(Error::from_io)?;
+                stream.flush().map_err(Error::from_io)?;
             }
             #[cfg(unix)]
             ConnectionEndpoint::DomainSocket { directory, name } => {
@@ -630,8 +630,8 @@ impl AsyncClient {
                 let mut buf = BytesMut::with_capacity(16);
                 frontend::cancel_request(self.process_id, self.secret_key, &mut buf);
 
-                stream.write_all(&buf).map_err(Error::io)?;
-                stream.flush().map_err(Error::io)?;
+                stream.write_all(&buf).map_err(Error::from_io)?;
+                stream.flush().map_err(Error::from_io)?;
             }
             #[cfg(windows)]
             ConnectionEndpoint::NamedPipe { host, name } => {
@@ -655,8 +655,8 @@ impl AsyncClient {
                 let mut buf = BytesMut::with_capacity(16);
                 frontend::cancel_request(self.process_id, self.secret_key, &mut buf);
 
-                file.write_all(&buf).map_err(Error::io)?;
-                file.flush().map_err(Error::io)?;
+                file.write_all(&buf).map_err(Error::from_io)?;
+                file.flush().map_err(Error::from_io)?;
             }
         }
 

hyperdb-api-core/src/client/async_connection.rs

@@ -102,8 +102,7 @@ where
     /// connection before any bytes hit the wire.
     pub(crate) fn ensure_healthy(&self) -> Result<()> {
         if self.desynchronized {
-            return Err(Error::new(
-                super::error::ErrorKind::Connection,
+            return Err(Error::connection(
                 "connection is desynchronized from the server and cannot be reused; \
                  discard it and open a new one",
             ));
@@ -526,7 +525,7 @@ where
     ///   (server closed the connection).
     pub async fn read_message(&mut self) -> Result<Message> {
         loop {
-            if let Some(msg) = Message::parse(&mut self.read_buf).map_err(Error::io)? {
+            if let Some(msg) = Message::parse(&mut self.read_buf).map_err(Error::from_io)? {
                 return Ok(msg);
             }
 
@@ -542,7 +541,7 @@ where
             if n == 0 {
                 self.read_buf.truncate(prev_len);
                 warn!(target: "hyperdb_api", "connection-closed");
-                return Err(Error::closed());
+                return Err(Error::closed("connection closed"));
             }
             self.read_buf.truncate(prev_len + n);
         }

hyperdb-api-core/src/client/client.rs

@@ -70,7 +70,7 @@ use super::cancel::Cancellable;
 use super::config::Config;
 use super::connection::{RawConnection, parse_error_response};
 use super::endpoint::ConnectionEndpoint;
-use super::error::{Error, ErrorKind, Result};
+use super::error::{Error, Result};
 use super::prepare;
 use super::row::{Row, StreamRow};
 use super::statement::ParamFormat;
@@ -572,10 +572,10 @@ impl Client {
                         error = %e,
                         "query-cancel-send-failed"
                     );
-                    Error::io(e)
+                    Error::from_io(e)
                 })?;
 
-                stream.flush().map_err(Error::io)?;
+                stream.flush().map_err(Error::from_io)?;
             }
             #[cfg(unix)]
             ConnectionEndpoint::DomainSocket { directory, name } => {
@@ -602,10 +602,10 @@ impl Client {
                         error = %e,
                         "query-cancel-send-failed"
                     );
-                    Error::io(e)
+                    Error::from_io(e)
                 })?;
 
-                stream.flush().map_err(Error::io)?;
+                stream.flush().map_err(Error::from_io)?;
             }
             #[cfg(windows)]
             ConnectionEndpoint::NamedPipe { host, name } => {
@@ -636,10 +636,10 @@ impl Client {
                         error = %e,
                         "query-cancel-send-failed"
                     );
-                    Error::io(e)
+                    Error::from_io(e)
                 })?;
 
-                file.flush().map_err(Error::io)?;
+                file.flush().map_err(Error::from_io)?;
             }
         }
 
@@ -1185,7 +1185,7 @@ impl Client {
     ///
     /// # Errors
     ///
-    /// - Returns [`ErrorKind::Query`] if `query` (trimmed) does not
+    /// - Returns [`Error::Query`] if `query` (trimmed) does not
     ///   start with `COPY` (defense-in-depth check against non-COPY
     ///   statements).
     /// - Returns [`Error`] (connection) if the connection mutex is
@@ -1195,8 +1195,7 @@ impl Client {
     pub fn copy_in_raw(&self, query: &str) -> Result<CopyInWriter<'_>> {
         // Defense-in-depth: reject queries that don't look like COPY statements
         if !query.trim_start().to_ascii_uppercase().starts_with("COPY") {
-            return Err(Error::new(
-                ErrorKind::Query,
+            return Err(Error::query(
                 "copy_in_raw() requires a COPY statement. \
                  The query must start with 'COPY'.",
             ));

hyperdb-api-core/src/client/connection.rs

@@ -144,7 +144,7 @@ where
         !self.desynchronized
     }
 
-    /// Fast-fails with an explicit [`ErrorKind::Connection`] error if the
+    /// Fast-fails with an explicit [`Error::Connection`] error if the
     /// wire has fallen out of sync with the server, before any bytes are
     /// written to the stream. Called from the entry point of every public
     /// method that initiates a new server request — simple queries,
@@ -155,8 +155,7 @@ where
     /// *next* unrelated operation.
     pub(crate) fn ensure_healthy(&self) -> Result<()> {
         if self.desynchronized {
-            return Err(Error::new(
-                crate::client::error::ErrorKind::Connection,
+            return Err(Error::connection(
                 "connection is desynchronized from the server and cannot be reused; \
                  discard it and open a new one",
             ));
@@ -579,7 +578,7 @@ where
     ///   (server closed the connection).
     pub fn read_message(&mut self) -> Result<Message> {
         loop {
-            if let Some(msg) = Message::parse(&mut self.read_buf).map_err(Error::io)? {
+            if let Some(msg) = Message::parse(&mut self.read_buf).map_err(Error::from_io)? {
                 return Ok(msg);
             }
 
@@ -604,7 +603,7 @@ where
             if n == 0 {
                 self.read_buf.truncate(prev_len);
                 warn!(target: "hyperdb_api", "connection-closed");
-                return Err(Error::closed());
+                return Err(Error::closed("connection closed"));
             }
             self.read_buf.truncate(prev_len + n);
         }
@@ -1180,12 +1179,9 @@ where
                 }
                 Message::CopyData(body) if in_copy_out => {
                     let chunk = body.data();
-                    writer.write_all(chunk).map_err(|e| {
-                        Error::new(
-                            super::error::ErrorKind::Io,
-                            format!("Failed to write COPY data: {e}"),
-                        )
-                    })?;
+                    writer
+                        .write_all(chunk)
+                        .map_err(|e| Error::io(format!("Failed to write COPY data: {e}")))?;
                     total_bytes += chunk.len() as u64;
                 }
                 Message::CopyDone => {
@@ -1280,7 +1276,7 @@ mod tests {
         conn.desynchronized = true;
         assert!(!conn.is_healthy());
         let err = conn.ensure_healthy().expect_err("must fail-fast");
-        assert_eq!(err.kind(), crate::client::error::ErrorKind::Connection);
+        assert!(matches!(err, Error::Connection { .. }));
         assert!(
             err.to_string().to_lowercase().contains("desynchron"),
             "error message should mention desynchronization; got: {err}",
@@ -1319,7 +1315,7 @@ mod tests {
         let Err(err) = conn.simple_query("SELECT 1") else {
             panic!("desynced simple_query must fail-fast")
         };
-        assert_eq!(err.kind(), crate::client::error::ErrorKind::Connection);
+        assert!(matches!(err, Error::Connection { .. }));
         assert!(err.to_string().to_lowercase().contains("desynchron"));
     }
 }