fix: correct Pebble iterator bounds and use pre-allocated txn prefix bytes

cmd/server/demo.go

@@ -305,6 +305,29 @@ func joinClusterWaitError(err error) error {
 	return err
 }
 
+func setupFSMStore(cfg config, cleanup *internalutil.CleanupStack) (store.MVCCStore, error) {
+	var st store.MVCCStore
+	var err error
+	if cfg.raftDataDir != "" {
+		st, err = store.NewPebbleStore(filepath.Join(cfg.raftDataDir, "fsm.db"))
+		if err != nil {
+			return nil, errors.WithStack(err)
+		}
+	} else {
+		fsmDir, tmpErr := os.MkdirTemp("", "elastickv-fsm-*")
+		if tmpErr != nil {
+			return nil, errors.WithStack(tmpErr)
+		}
+		cleanup.Add(func() { os.RemoveAll(fsmDir) })
+		st, err = store.NewPebbleStore(fsmDir)
+		if err != nil {
+			return nil, errors.WithStack(err)
+		}
+	}
+	cleanup.Add(func() { st.Close() })
+	return st, nil
+}
+
 func setupStorage(dir string) (raft.LogStore, raft.StableStore, raft.SnapshotStore, error) {
 	if dir == "" {
 		return raft.NewInmemStore(), raft.NewInmemStore(), raft.NewInmemSnapshotStore(), nil
@@ -376,24 +399,10 @@ func run(ctx context.Context, eg *errgroup.Group, cfg config) error {
 		return err
 	}
 
-	var st store.MVCCStore
-	if cfg.raftDataDir != "" {
-		st, err = store.NewPebbleStore(filepath.Join(cfg.raftDataDir, "fsm.db"))
-		if err != nil {
-			return errors.WithStack(err)
-		}
-	} else {
-		fsmDir, tmpErr := os.MkdirTemp("", "elastickv-fsm-*")
-		if tmpErr != nil {
-			return errors.WithStack(tmpErr)
-		}
-		cleanup.Add(func() { os.RemoveAll(fsmDir) })
-		st, err = store.NewPebbleStore(fsmDir)
-		if err != nil {
-			return errors.WithStack(err)
-		}
+	st, err := setupFSMStore(cfg, &cleanup)
+	if err != nil {
+		return err
 	}
-	cleanup.Add(func() { st.Close() })
 	fsm := kv.NewKvFSM(st)
 	readTracker := kv.NewActiveTimestampTracker()
 

kv/fsm.go

@@ -95,17 +95,20 @@ func decodeRaftRequests(data []byte) ([]*pb.Request, error) {
 		}
 		return cmd.Requests, nil
 	default:
-		// Legacy format (no prefix byte) — try both for backward compatibility.
-		cmd := &pb.RaftCommand{}
-		if err := proto.Unmarshal(data, cmd); err == nil && len(cmd.Requests) > 0 {
-			return cmd.Requests, nil
-		}
-		req := &pb.Request{}
-		if err := proto.Unmarshal(data, req); err != nil {
-			return nil, errors.WithStack(err)
-		}
-		return []*pb.Request{req}, nil
+		return decodeLegacyRaftRequest(data)
+	}
+}
+
+func decodeLegacyRaftRequest(data []byte) ([]*pb.Request, error) {
+	cmd := &pb.RaftCommand{}
+	if err := proto.Unmarshal(data, cmd); err == nil && len(cmd.Requests) > 0 {
+		return cmd.Requests, nil
+	}
+	req := &pb.Request{}
+	if err := proto.Unmarshal(data, req); err != nil {
+		return nil, errors.WithStack(err)
 	}
+	return []*pb.Request{req}, nil
 }
 
 func requestCommitTS(r *pb.Request) (uint64, error) {

kv/lock_resolver.go

@@ -16,6 +16,9 @@ const (
 	lockResolverInterval = 10 * time.Second
 	// lockResolverBatchSize limits the number of locks scanned per group per cycle.
 	lockResolverBatchSize = 100
+	// txnNamespaceSentinel is appended to a namespace prefix key to produce a
+	// key that sorts just after all valid keys in that namespace.
+	txnNamespaceSentinel byte = 0xFF
 )
 
 // LockResolver periodically scans for expired transaction locks and resolves
@@ -92,8 +95,8 @@ func (lr *LockResolver) resolveGroupLocks(ctx context.Context, gid uint64, g *Sh
 	}
 
 	// Scan lock key range: [!txn|lock| ... !txn|lock|<max>)
-	lockStart := txnLockKey(nil)                 // "!txn|lock|"
-	lockEnd := append(txnLockKey(nil), byte(0xFF)) // one past the lock namespace
+	lockStart := txnLockKey(nil)                             // "!txn|lock|"
+	lockEnd := append(txnLockKey(nil), txnNamespaceSentinel) // one past the lock namespace
 
 	lockKVs, err := g.Store.ScanAt(ctx, lockStart, lockEnd, lockResolverBatchSize, math.MaxUint64)
 	if err != nil {
@@ -103,9 +106,8 @@ func (lr *LockResolver) resolveGroupLocks(ctx context.Context, gid uint64, g *Sh
 	var resolved, skipped int
 	for _, kvp := range lockKVs {
 		if ctx.Err() != nil {
-			return ctx.Err()
+			return errors.WithStack(ctx.Err())
 		}
-
 		userKey, ok := txnUserKeyFromLockKey(kvp.Key)
 		if !ok {
 			continue

kv/sharded_coordinator_abort_test.go

@@ -4,8 +4,8 @@ import (
 	"context"
 	"testing"
 
-	pb "github.com/bootjp/elastickv/proto"
 	"github.com/bootjp/elastickv/distribution"
+	pb "github.com/bootjp/elastickv/proto"
 	"github.com/bootjp/elastickv/store"
 	"github.com/cockroachdb/errors"
 	"github.com/stretchr/testify/require"

kv/txn_codec.go

@@ -20,6 +20,9 @@ const (
 
 const txnLockFlagPrimary byte = 0x01
 
+// uint64FieldSize is the byte size of a serialized uint64 field.
+const uint64FieldSize = 8
+
 // TxnMeta is embedded into transactional raft log requests via a synthetic
 // mutation (key prefix "!txn|meta|"). It is not persisted in the MVCC store.
 type TxnMeta struct {
@@ -30,7 +33,7 @@ type TxnMeta struct {
 
 func EncodeTxnMeta(m TxnMeta) []byte {
 	// version(1) + LockTTLms(8) + CommitTS(8) + primaryLen(8) + primaryKey
-	size := 1 + 8 + 8 + 8 + len(m.PrimaryKey)
+	size := 1 + uint64FieldSize + uint64FieldSize + uint64FieldSize + len(m.PrimaryKey)
 	b := make([]byte, size)
 	b[0] = txnMetaVersion
 	binary.BigEndian.PutUint64(b[1:], m.LockTTLms)
@@ -79,7 +82,7 @@ type txnLock struct {
 
 func encodeTxnLock(l txnLock) []byte {
 	// version(1) + StartTS(8) + TTLExpireAt(8) + flags(1) + primaryLen(8) + primaryKey
-	size := 1 + 8 + 8 + 1 + 8 + len(l.PrimaryKey)
+	size := 1 + uint64FieldSize + uint64FieldSize + 1 + uint64FieldSize + len(l.PrimaryKey)
 	b := make([]byte, size)
 	b[0] = txnLockVersion
 	binary.BigEndian.PutUint64(b[1:], l.StartTS)
@@ -143,7 +146,7 @@ const (
 
 func encodeTxnIntent(i txnIntent) []byte {
 	// version(1) + StartTS(8) + Op(1) + valLen(8) + value
-	size := 1 + 8 + 1 + 8 + len(i.Value)
+	size := 1 + uint64FieldSize + 1 + uint64FieldSize + len(i.Value)
 	b := make([]byte, size)
 	b[0] = txnIntentVersion
 	binary.BigEndian.PutUint64(b[1:], i.StartTS)

kv/txn_keys.go

@@ -28,15 +28,15 @@ var (
 const txnStartTSSuffixLen = 8
 
 func txnLockKey(userKey []byte) []byte {
-	k := make([]byte, 0, len(txnLockPrefix)+len(userKey))
-	k = append(k, txnLockPrefix...)
+	k := make([]byte, 0, len(txnLockPrefixBytes)+len(userKey))
+	k = append(k, txnLockPrefixBytes...)
 	k = append(k, userKey...)
 	return k
 }
 
 func txnIntentKey(userKey []byte) []byte {
-	k := make([]byte, 0, len(txnIntentPrefix)+len(userKey))
-	k = append(k, txnIntentPrefix...)
+	k := make([]byte, 0, len(txnIntentPrefixBytes)+len(userKey))
+	k = append(k, txnIntentPrefixBytes...)
 	k = append(k, userKey...)
 	return k
 }