fix(txn): TXN.COMMIT aborts instead of applying a partial transaction (#499)

CHANGELOG.md

@@ -37,6 +37,20 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   protocol version and selects between two static byte strings (`io::static_responses::null_bulk`,
   which also gives the previously caller-less `NULL_BULK` constant a home). `+OK` and `-WRONGTYPE`,
   the only other bytes this path frames, are spelled identically in both protocols.
+- **`TXN.COMMIT` no longer answers `+OK` for a transaction that was applied in part** (#499).
+  A TXN body whose ops are rejected by a TXN guard — a cross-shard write at `--shards > 1`,
+  `MOVE`, `COPY ... DB`, `SWAPDB`, a cross-shard Cypher write — used to commit the *accepted*
+  subset and reply `+OK`. The per-op errors did go back to the client, but a driver inspects the
+  COMMIT reply, not the replies of the body commands (exactly as it inspects `EXEC` and not the
+  `QUEUED`s), so a routing mistake became silent partial application. A rejection now poisons the
+  transaction the way a queue-time error poisons `MULTI`: `TXN.COMMIT` rolls the whole transaction
+  back through the `TXN.ABORT` path and answers
+  `EXECABORT TXN.COMMIT discarded because of previous errors: N operation(s) rejected inside the
+  transaction (first: <CMD>) -- rolled back and NOT committed`, leaving the connection out of the
+  transaction. The wording stops at what the code can guarantee — the commit did not happen —
+  because rollback is the same best-effort `TXN.ABORT` path whose undo capture still has gaps
+  (#500). Nothing changes for a transaction whose every op was accepted. Both handlers (monoio
+  and sharded/tokio) carry the check, and both are covered end-to-end.
 
 ### Changed
 - **Active expiry runs on a deadline-ordered index instead of probabilistic sampling** (#541).

src/command/transaction.rs

@@ -36,6 +36,41 @@ pub const ERR_MULTI_TXN_CONFLICT: &[u8] = b"ERR cannot use MULTI while in TXN bl
 pub const ERR_TXN_CROSS_SHARD: &[u8] = b"ERR TXN does not support cross-shard writes \
       -- use hash tags {tag} to co-locate keys (e.g. SET {txn}:key value)";
 
+/// #499: the error `TXN.COMMIT` answers when the body contained rejected ops.
+///
+/// Every guard rejection inside a TXN body (cross-shard write, `MOVE`,
+/// `COPY ... DB`, `SWAPDB`, cross-shard Cypher write) poisons the
+/// transaction. Committing the accepted subset would turn a routing mistake
+/// into silent partial application — the caller inspects the COMMIT reply,
+/// not the replies of the individual body commands, exactly as a driver
+/// inspects `EXEC` and not the `QUEUED`s.
+///
+/// Semantics match Redis's `CLIENT_DIRTY_EXEC`: the whole transaction is
+/// rolled back and discarded, and the reply carries the `EXECABORT` code so
+/// drivers classify it as a transaction abort rather than a generic command
+/// error.
+///
+/// Wording note: the message says "rolled back and NOT committed", not
+/// "nothing was applied". Rollback runs the `TXN.ABORT` path, which is
+/// best-effort by construction — `MSET` and multi-key `DEL` bypass undo
+/// capture today (#500), so an absolute claim would be a promise this code
+/// cannot keep. What IS guaranteed: the commit did not happen, and the
+/// transaction is discarded.
+pub fn err_txn_commit_dirty(rejected: u32, first_cmd: Option<&[u8]>) -> Frame {
+    let mut msg = bytes::BytesMut::new();
+    use std::fmt::Write as _;
+    let _ = write!(
+        msg,
+        "EXECABORT TXN.COMMIT discarded because of previous errors: \
+         {rejected} operation(s) rejected inside the transaction"
+    );
+    if let Some(cmd) = first_cmd {
+        let _ = write!(msg, " (first: {})", String::from_utf8_lossy(cmd));
+    }
+    let _ = write!(msg, " -- rolled back and NOT committed");
+    Frame::Error(msg.freeze())
+}
+
 /// TXN.BEGIN - Start a new cross-store transaction.
 ///
 /// Returns: +OK on success, or error if already in transaction.
@@ -226,6 +261,33 @@ mod tests {
         assert!(is_txn_abort(b"TXN", &args));
     }
 
+    /// #499: the commit-time abort error must carry the `EXECABORT` code, the
+    /// rejected-op count, the first offending command, and say plainly that
+    /// nothing was applied.
+    #[test]
+    fn test_err_txn_commit_dirty_shape() {
+        let Frame::Error(msg) = err_txn_commit_dirty(3, Some(b"SET")) else {
+            panic!("err_txn_commit_dirty must return Frame::Error");
+        };
+        let msg = String::from_utf8_lossy(&msg).into_owned();
+        assert!(msg.starts_with("EXECABORT "), "{msg}");
+        assert!(msg.contains("3 operation(s) rejected"), "{msg}");
+        assert!(msg.contains("(first: SET)"), "{msg}");
+        assert!(msg.contains("rolled back and NOT committed"), "{msg}");
+        assert!(!msg.contains('\r') && !msg.contains('\n'), "{msg}");
+    }
+
+    /// A missing command name must not produce a dangling `(first: )`.
+    #[test]
+    fn test_err_txn_commit_dirty_without_cmd_name() {
+        let Frame::Error(msg) = err_txn_commit_dirty(1, None) else {
+            panic!("err_txn_commit_dirty must return Frame::Error");
+        };
+        let msg = String::from_utf8_lossy(&msg).into_owned();
+        assert!(msg.contains("1 operation(s) rejected"), "{msg}");
+        assert!(!msg.contains("first:"), "{msg}");
+    }
+
     #[test]
     fn test_err_txn_cross_shard_is_defined() {
         assert!(!ERR_TXN_CROSS_SHARD.is_empty());

src/server/conn/core.rs

@@ -516,6 +516,21 @@ impl ConnectionState {
         self.active_cross_txn.is_some()
     }
 
+    /// #499: mark the active cross-store transaction dirty because a TXN guard
+    /// rejected `cmd`.
+    ///
+    /// MUST be called at EVERY site that answers `ERR_TXN_CROSS_SHARD` (or any
+    /// future in-TXN rejection). Without it `TXN.COMMIT` would apply the
+    /// accepted subset and still reply `+OK`, converting a routing mistake into
+    /// silent partial application. No-op when no transaction is active, so
+    /// callers need not re-check `in_cross_txn()`.
+    #[inline]
+    pub fn mark_cross_txn_rejected(&mut self, cmd: &[u8]) {
+        if let Some(txn) = self.active_cross_txn.as_mut() {
+            txn.record_rejected_op(cmd);
+        }
+    }
+
     /// D4 (#438): whether this connection may migrate to another shard
     /// RIGHT NOW. `MigratedConnectionState` carries none of the state
     /// checked here (queued MULTI txn, cross-store txn, subscriptions,

src/server/conn/handler_monoio/dispatch.rs

@@ -1309,7 +1309,7 @@ pub(super) async fn try_handle_shutdown(
 pub(super) async fn try_handle_swapdb(
     cmd: &[u8],
     cmd_args: &[Frame],
-    conn: &crate::server::conn::core::ConnectionState,
+    conn: &mut crate::server::conn::core::ConnectionState,
     ctx: &ConnectionContext,
     responses: &mut Vec<Frame>,
 ) -> bool {
@@ -1326,6 +1326,8 @@ pub(super) async fn try_handle_swapdb(
     // TXN guard: SWAPDB rewrites entire DB contents and has no undo path —
     // reject during an active cross-store TXN so TXN.ABORT remains coherent.
     if conn.in_cross_txn() {
+        // #499: poison the txn so COMMIT cannot report OK.
+        conn.mark_cross_txn_rejected(cmd);
         responses.push(Frame::Error(Bytes::from_static(
             crate::command::transaction::ERR_TXN_CROSS_SHARD,
         )));

src/server/conn/handler_monoio/mod.rs

@@ -2082,7 +2082,7 @@ pub(crate) async fn handle_connection_sharded_monoio<
             // of every privileged intercept. Its old comment claimed exactly
             // the invariant the code above it violated.
             // --- SWAPDB: handler-layer intercept (needs async + multi-db access) ---
-            if dispatch::try_handle_swapdb(cmd, cmd_args, &conn, ctx, &mut responses).await {
+            if dispatch::try_handle_swapdb(cmd, cmd_args, &mut conn, ctx, &mut responses).await {
                 continue;
             }
             if dispatch::try_handle_client_admin(cmd, cmd_args, client_id, &conn, &mut responses) {
@@ -2435,6 +2435,8 @@ pub(crate) async fn handle_connection_sharded_monoio<
                     // Reject during an active cross-store TXN so TXN.ABORT can
                     // still roll back cleanly (matches handler_sharded policy).
                     if conn.in_cross_txn() {
+                        // #499: poison the txn so COMMIT cannot report OK.
+                        conn.mark_cross_txn_rejected(cmd);
                         responses.push(Frame::Error(bytes::Bytes::from_static(
                             crate::command::transaction::ERR_TXN_CROSS_SHARD,
                         )));
@@ -2518,6 +2520,8 @@ pub(crate) async fn handle_connection_sharded_monoio<
                         // Reject only when DB clause is present (cross-DB);
                         // same-DB COPY falls through to the normal write path.
                         if conn.in_cross_txn() {
+                            // #499: poison the txn so COMMIT cannot report OK.
+                            conn.mark_cross_txn_rejected(cmd);
                             responses.push(Frame::Error(bytes::Bytes::from_static(
                                 crate::command::transaction::ERR_TXN_CROSS_SHARD,
                             )));
@@ -3119,6 +3123,10 @@ pub(crate) async fn handle_connection_sharded_monoio<
             } else if let Some(target) = target_shard {
                 // TXN cross-shard guard: reject cross-shard writes in active TXN (no undo log).
                 if conn.in_cross_txn() && metadata::is_write(cmd) {
+                    // #499: poison the txn — the rejected write is NOT part of the
+                    // transaction, so TXN.COMMIT must refuse rather than commit the
+                    // accepted subset behind a `+OK`.
+                    conn.mark_cross_txn_rejected(cmd);
                     responses.push(Frame::Error(bytes::Bytes::from_static(
                         crate::command::transaction::ERR_TXN_CROSS_SHARD,
                     )));

src/server/conn/handler_monoio/txn.rs

@@ -59,6 +59,42 @@ pub(super) async fn try_handle_txn_commit(
     match txn_commit_validate(conn.in_cross_txn()) {
         Ok(()) => {
             if let Some(txn) = conn.active_cross_txn.take() {
+                // #499: a transaction whose body had ops REJECTED by a TXN guard
+                // may not commit. The accepted subset is real (already applied to
+                // the shard-local store) but the rejected ops are not, so a `+OK`
+                // here reports an atomic transaction that was in fact applied in
+                // part. Redis's MULTI has the same shape and answers EXECABORT;
+                // do the same here, and roll the accepted subset back through the
+                // TXN.ABORT path so the outcome is "nothing was applied".
+                //
+                // Checked BEFORE the killed-snapshot arm: rollback is the strictly
+                // stronger action and `txn_manager.abort()` retires a killed
+                // transaction just as `abort_killed` would.
+                if txn.is_dirty() {
+                    let rejected = txn.rejected_ops;
+                    tracing::warn!(
+                        txn_id = txn.txn_id,
+                        rejected,
+                        "TXN.COMMIT rejected: transaction contained rejected ops -- rolling back"
+                    );
+                    let err = crate::command::transaction::err_txn_commit_dirty(
+                        rejected,
+                        txn.first_rejected_cmd.as_deref(),
+                    );
+                    Box::pin(crate::transaction::abort::abort_cross_store_txn_routed(
+                        &ctx.shard_databases,
+                        ctx.shard_id,
+                        conn.selected_db,
+                        ctx.num_shards,
+                        &ctx.dispatch_tx,
+                        &ctx.spsc_notifiers,
+                        *txn,
+                    ))
+                    .await;
+                    responses.push(err);
+                    return true;
+                }
+
                 // MA2: reject commit if the snapshot was killed (by operator KILL SNAPSHOT
                 // or by the automatic old_snapshot_threshold sweep). A killed snapshot may
                 // have been excluded from oldest_snapshot, allowing prune_committed to

src/server/conn/handler_monoio/write.rs

@@ -1108,6 +1108,8 @@ pub(super) async fn try_handle_graph_command(
                     && cmd.eq_ignore_ascii_case(b"GRAPH.QUERY")
                     && crate::command::graph::is_cypher_write_query(cmd_args)
                 {
+                    // #499: poison the txn so COMMIT cannot report OK.
+                    conn.mark_cross_txn_rejected(cmd);
                     responses.push(Frame::Error(bytes::Bytes::from_static(
                         crate::command::transaction::ERR_TXN_CROSS_SHARD,
                     )));

src/server/conn/handler_sharded/dispatch.rs

@@ -595,7 +595,7 @@ pub(super) async fn try_handle_cross_shard_scan(
 pub(super) async fn try_handle_swapdb(
     cmd: &[u8],
     cmd_args: &[Frame],
-    conn: &ConnectionState,
+    conn: &mut ConnectionState,
     ctx: &ConnectionContext,
     responses: &mut Vec<Frame>,
 ) -> bool {
@@ -612,6 +612,8 @@ pub(super) async fn try_handle_swapdb(
     // TXN guard: SWAPDB rewrites entire DB contents and has no undo path —
     // reject during an active cross-store TXN so TXN.ABORT remains coherent.
     if conn.in_cross_txn() {
+        // #499: poison the txn so COMMIT cannot report OK.
+        conn.mark_cross_txn_rejected(cmd);
         responses.push(Frame::Error(Bytes::from_static(
             crate::command::transaction::ERR_TXN_CROSS_SHARD,
         )));

src/server/conn/handler_sharded/mod.rs

@@ -1570,7 +1570,9 @@ pub(crate) async fn handle_connection_sharded_inner<
                     }
 
                     // --- SWAPDB: handler-layer intercept (needs async + multi-db access) ---
-                    if dispatch::try_handle_swapdb(cmd, cmd_args, &conn, ctx, &mut responses).await {
+                    if dispatch::try_handle_swapdb(cmd, cmd_args, &mut conn, ctx, &mut responses)
+                        .await
+                    {
                         continue;
                     }
 
@@ -1768,6 +1770,8 @@ pub(crate) async fn handle_connection_sharded_inner<
                             // active cross-store TXN so TXN.ABORT can still roll back
                             // cleanly. (Same policy as cross-shard writes.)
                             if conn.in_cross_txn() {
+                                // #499: poison the txn so COMMIT cannot report OK.
+                                conn.mark_cross_txn_rejected(cmd);
                                 responses.push(Frame::Error(bytes::Bytes::from_static(
                                     crate::command::transaction::ERR_TXN_CROSS_SHARD,
                                 )));
@@ -1835,6 +1839,8 @@ pub(crate) async fn handle_connection_sharded_inner<
                                 // same-DB COPY falls through to the normal write path
                                 // which already participates in TXN.
                                 if conn.in_cross_txn() {
+                                    // #499: poison the txn so COMMIT cannot report OK.
+                                    conn.mark_cross_txn_rejected(cmd);
                                     responses.push(Frame::Error(bytes::Bytes::from_static(
                                         crate::command::transaction::ERR_TXN_CROSS_SHARD,
                                     )));
@@ -2382,6 +2388,10 @@ pub(crate) async fn handle_connection_sharded_inner<
                         // cannot be rolled back on TXN.ABORT. Return an explicit error instead
                         // of silently permitting writes that resist rollback.
                         if conn.in_cross_txn() && metadata::is_write(cmd) {
+                            // #499: poison the txn — the rejected write is NOT part of
+                            // the transaction, so TXN.COMMIT must refuse rather than
+                            // commit the accepted subset behind a `+OK`.
+                            conn.mark_cross_txn_rejected(cmd);
                             responses.push(Frame::Error(bytes::Bytes::from_static(
                                 crate::command::transaction::ERR_TXN_CROSS_SHARD,
                             )));

src/server/conn/handler_sharded/txn.rs

@@ -60,6 +60,42 @@ pub(super) async fn try_handle_txn_commit(
     match txn_commit_validate(conn.in_cross_txn()) {
         Ok(()) => {
             if let Some(txn) = conn.active_cross_txn.take() {
+                // #499: a transaction whose body had ops REJECTED by a TXN guard
+                // may not commit. The accepted subset is real (already applied to
+                // the shard-local store) but the rejected ops are not, so a `+OK`
+                // here reports an atomic transaction that was in fact applied in
+                // part. Redis's MULTI has the same shape and answers EXECABORT;
+                // do the same here, and roll the accepted subset back through the
+                // TXN.ABORT path so the outcome is "nothing was applied".
+                //
+                // Checked BEFORE the killed-snapshot arm: rollback is the strictly
+                // stronger action and `txn_manager.abort()` retires a killed
+                // transaction just as `abort_killed` would.
+                if txn.is_dirty() {
+                    let rejected = txn.rejected_ops;
+                    tracing::warn!(
+                        txn_id = txn.txn_id,
+                        rejected,
+                        "TXN.COMMIT rejected: transaction contained rejected ops -- rolling back"
+                    );
+                    let err = crate::command::transaction::err_txn_commit_dirty(
+                        rejected,
+                        txn.first_rejected_cmd.as_deref(),
+                    );
+                    Box::pin(crate::transaction::abort::abort_cross_store_txn_routed(
+                        &ctx.shard_databases,
+                        ctx.shard_id,
+                        conn.selected_db,
+                        ctx.num_shards,
+                        &ctx.dispatch_tx,
+                        &ctx.spsc_notifiers,
+                        *txn,
+                    ))
+                    .await;
+                    responses.push(err);
+                    return true;
+                }
+
                 // MA2: reject commit if the snapshot was killed (by operator KILL SNAPSHOT
                 // or by the automatic old_snapshot_threshold sweep). A killed snapshot may
                 // have been excluded from oldest_snapshot, allowing prune_committed to

src/server/conn/handler_sharded/write.rs

@@ -886,6 +886,8 @@ pub(super) async fn try_handle_graph_command(
                     && cmd.eq_ignore_ascii_case(b"GRAPH.QUERY")
                     && crate::command::graph::is_cypher_write_query(cmd_args)
                 {
+                    // #499: poison the txn so COMMIT cannot report OK.
+                    conn.mark_cross_txn_rejected(cmd);
                     responses.push(Frame::Error(bytes::Bytes::from_static(
                         crate::command::transaction::ERR_TXN_CROSS_SHARD,
                     )));