Add defaultTxnLockTTLSeconds constant and use it

Author: bootjp

kv/coordinator.go

@@ -121,13 +121,16 @@ func (c *Coordinate) toRawRequest(req *Elem[OP]) *pb.Request {
 	panic("unreachable")
 }
 
+const defaultTxnLockTTLSeconds = uint64(30)
+
 func (c *Coordinate) toTxnRequests(req *Elem[OP]) []*pb.Request {
 	switch req.Op {
 	case Put:
 		return []*pb.Request{
 			{
 				IsTxn: true,
 				Phase: pb.Phase_PREPARE,
+				Ts:    defaultTxnLockTTLSeconds,
 				Mutations: []*pb.Mutation{
 					{
 						Key:   req.Key,
@@ -138,6 +141,7 @@ func (c *Coordinate) toTxnRequests(req *Elem[OP]) []*pb.Request {
 			{
 				IsTxn: true,
 				Phase: pb.Phase_COMMIT,
+				Ts:    defaultTxnLockTTLSeconds,
 				Mutations: []*pb.Mutation{
 					{
 						Key:   req.Key,
@@ -152,6 +156,7 @@ func (c *Coordinate) toTxnRequests(req *Elem[OP]) []*pb.Request {
 			{
 				IsTxn: true,
 				Phase: pb.Phase_PREPARE,
+				Ts:    defaultTxnLockTTLSeconds,
 				Mutations: []*pb.Mutation{
 					{
 						Key: req.Key,
@@ -161,6 +166,7 @@ func (c *Coordinate) toTxnRequests(req *Elem[OP]) []*pb.Request {
 			{
 				IsTxn: true,
 				Phase: pb.Phase_COMMIT,
+				Ts:    defaultTxnLockTTLSeconds,
 				Mutations: []*pb.Mutation{
 					{
 						Key: req.Key,

kv/fsm.go

@@ -209,7 +209,13 @@ func (f *kvFSM) handleCommitRequest(ctx context.Context, r *pb.Request) error {
 				}
 
 				if !ok {
-					return errors.WithStack(ErrKeyNotLocked)
+					// Lock missing (e.g., expired). Try to reacquire to make progress.
+					err := f.lockStore.TxnWithTTL(ctx, func(ctx context.Context, ttlTxn store.TTLTxn) error {
+						return f.lock(ttlTxn, mut.Key, r.Ts)
+					})
+					if err != nil {
+						return errors.WithStack(err)
+					}
 				}
 
 				err = f.commit(ctx, txn, mut)