Merge pull request #444 from bootjp/feat/fix-abort

kv: suppress abort warnings for committed txns

Author: bootjp

kv/sharded_coordinator.go

@@ -36,6 +36,7 @@ type ShardedCoordinator struct {
 	defaultGroup uint64
 	clock        *HLC
 	store        store.MVCCStore
+	log          *slog.Logger
 }
 
 // NewShardedCoordinator builds a coordinator for the provided shard groups.
@@ -52,6 +53,7 @@ func NewShardedCoordinator(engine *distribution.Engine, groups map[uint64]*Shard
 		defaultGroup: defaultGroup,
 		clock:        clock,
 		store:        st,
+		log:          slog.Default(),
 	}
 }
 
@@ -211,7 +213,7 @@ func (c *ShardedCoordinator) commitSecondaryTxns(startTS uint64, primaryGid uint
 		}
 		r, err := commitSecondaryWithRetry(g, req)
 		if err != nil {
-			slog.Warn("txn secondary commit failed",
+			c.log.Warn("txn secondary commit failed",
 				slog.Uint64("gid", gid),
 				slog.String("primary_key", string(primaryKey)),
 				slog.Uint64("start_ts", startTS),
@@ -266,7 +268,10 @@ func (c *ShardedCoordinator) abortPreparedTxn(startTS uint64, primaryKey []byte,
 			Mutations: append([]*pb.Mutation{meta}, pg.keys...),
 		}
 		if _, err := g.Txn.Commit([]*pb.Request{req}); err != nil {
-			slog.Warn("txn abort failed; locks may remain until TTL expiry",
+			if errors.Is(err, ErrTxnAlreadyCommitted) {
+				continue
+			}
+			c.log.Warn("txn abort failed; locks may remain until TTL expiry",
 				slog.Uint64("gid", pg.gid),
 				slog.String("primary_key", string(primaryKey)),
 				slog.Uint64("start_ts", startTS),

kv/sharded_coordinator_abort_test.go

@@ -1,7 +1,9 @@
 package kv
 
 import (
+	"bytes"
 	"context"
+	"log/slog"
 	"testing"
 
 	"github.com/bootjp/elastickv/distribution"
@@ -82,3 +84,22 @@ func TestShardedAbortRollback_PrepareFailOnShard2_CleansShard1Locks(t *testing.T
 	_, err = s2.GetAt(ctx, []byte("x"), ^uint64(0))
 	require.ErrorIs(t, err, store.ErrKeyNotFound, "user data for key 'x' should not exist on shard2")
 }
+
+func TestAbortPreparedTxn_DoesNotWarnWhenTxnAlreadyCommitted(t *testing.T) {
+	var buf bytes.Buffer
+	logger := slog.New(slog.NewTextHandler(&buf, &slog.HandlerOptions{Level: slog.LevelWarn}))
+
+	coord := &ShardedCoordinator{
+		log: logger,
+		groups: map[uint64]*ShardGroup{
+			1: {Txn: &failingTransactional{err: errors.WithStack(ErrTxnAlreadyCommitted)}},
+		},
+	}
+
+	coord.abortPreparedTxn(10, []byte("pk"), []preparedGroup{{
+		gid:  1,
+		keys: []*pb.Mutation{{Op: pb.Op_PUT, Key: []byte("pk")}},
+	}}, 20)
+
+	require.NotContains(t, buf.String(), "txn abort failed; locks may remain until TTL expiry")
+}

kv/transaction.go

@@ -229,7 +229,13 @@ func (t *TransactionManager) commitSequential(reqs []*pb.Request) (*TransactionR
 
 func needsTxnCleanup(reqs []*pb.Request) bool {
 	for _, req := range reqs {
-		if req != nil && req.IsTxn && req.Phase != pb.Phase_NONE {
+		if req == nil || !req.IsTxn {
+			continue
+		}
+		// Be conservative: any transactional phase other than NONE/ABORT may have
+		// left intents that require cleanup, including unknown enum values that
+		// can appear during rolling upgrades.
+		if req.Phase != pb.Phase_NONE && req.Phase != pb.Phase_ABORT {
 			return true
 		}
 	}

kv/transaction_batch_test.go

@@ -192,3 +192,15 @@ func TestApplyRequestsDoesNotCountBusinessErrorAsProposalFailure(t *testing.T) {
 	require.ErrorIs(t, results[0], ErrInvalidRequest)
 	require.Zero(t, observer.FailureCount())
 }
+
+func TestNeedsTxnCleanupSkipsAbortRequests(t *testing.T) {
+	t.Parallel()
+
+	reqs := []*pb.Request{
+		{IsTxn: true, Phase: pb.Phase_ABORT},
+	}
+	require.False(t, needsTxnCleanup(reqs))
+	require.True(t, needsTxnCleanup([]*pb.Request{{IsTxn: true, Phase: pb.Phase_PREPARE}}))
+	require.True(t, needsTxnCleanup([]*pb.Request{{IsTxn: true, Phase: pb.Phase_COMMIT}}))
+	require.False(t, needsTxnCleanup([]*pb.Request{{IsTxn: true, Phase: pb.Phase_NONE}}))
+}