fix: bound default lock wait timeout

pkg/common/moerr/error.go

@@ -280,6 +280,8 @@ const (
 	ErrCannotCommitOnInvalidCN uint16 = 20708
 	// ErrRemoteLockWaitTimeout remote lock owner-side wait timeout
 	ErrRemoteLockWaitTimeout uint16 = 20709
+	// ErrLockWaitTimeout a lock waiter exceeded its configured wait budget
+	ErrLockWaitTimeout uint16 = 20710
 
 	// Group 8: partition
 	ErrPartitionFunctionIsNotAllowed       uint16 = 20801
@@ -542,6 +544,7 @@ var errorMsgRefer = map[uint16]moErrorMsgItem{
 	ErrLockNeedUpgrade:         {ER_UNKNOWN_ERROR, []string{MySQLDefaultSqlState}, "row level lock is too large that need upgrade to table level lock"},
 	ErrCannotCommitOnInvalidCN: {ER_UNKNOWN_ERROR, []string{MySQLDefaultSqlState}, "cannot commit a orphan transaction on invalid cn"},
 	ErrRemoteLockWaitTimeout:   {ER_UNKNOWN_ERROR, []string{MySQLDefaultSqlState}, "remote lock wait timeout"},
+	ErrLockWaitTimeout:         {ER_LOCK_WAIT_TIMEOUT, []string{MySQLDefaultSqlState}, "Lock wait timeout exceeded; try restarting transaction"},
 
 	// Group 8: partition
 	ErrPartitionFunctionIsNotAllowed:       {ER_PARTITION_FUNCTION_IS_NOT_ALLOWED, []string{MySQLDefaultSqlState}, "This partition function is not allowed"},
@@ -1491,6 +1494,10 @@ func NewRemoteLockWaitTimeout(ctx context.Context) *Error {
 	return newError(ctx, ErrRemoteLockWaitTimeout)
 }
 
+func NewLockWaitTimeout(ctx context.Context) *Error {
+	return newError(ctx, ErrLockWaitTimeout)
+}
+
 func NewPartitionFunctionIsNotAllowed(ctx context.Context) *Error {
 	return newError(ctx, ErrPartitionFunctionIsNotAllowed)
 }

pkg/common/moerr/error_no_ctx.go

@@ -434,6 +434,10 @@ func NewRemoteLockWaitTimeoutNoCtx() *Error {
 	return newError(Context(), ErrRemoteLockWaitTimeout)
 }
 
+func NewLockWaitTimeoutNoCtx() *Error {
+	return newError(Context(), ErrLockWaitTimeout)
+}
+
 func NewLockNeedUpgradeNoCtx() *Error {
 	return newError(Context(), ErrLockNeedUpgrade)
 }

pkg/common/moerr/error_test.go

@@ -91,6 +91,18 @@ func TestNew_MyErrorCode(t *testing.T) {
 	require.Equal(t, ER_DATA_OUT_OF_RANGE, err.MySQLCode())
 }
 
+func TestLockWaitTimeoutMySQLError(t *testing.T) {
+	err := NewLockWaitTimeout(context.Background())
+	require.Equal(t, ErrLockWaitTimeout, err.ErrorCode())
+	require.Equal(t, ER_LOCK_WAIT_TIMEOUT, err.MySQLCode())
+	require.Equal(t, MySQLDefaultSqlState, err.SqlState())
+	require.Equal(t, "Lock wait timeout exceeded; try restarting transaction", err.Error())
+
+	noCtxErr := NewLockWaitTimeoutNoCtx()
+	require.Equal(t, ErrLockWaitTimeout, noCtxErr.ErrorCode())
+	require.Equal(t, ER_LOCK_WAIT_TIMEOUT, noCtxErr.MySQLCode())
+}
+
 func TestIsMoErrCode(t *testing.T) {
 	err := NewDivByZero(context.TODO())
 	require.True(t, IsMoErrCode(err, ErrDivByZero))

pkg/frontend/util.go

@@ -1285,6 +1285,7 @@ var errCodeRollbackWholeTxn = map[uint16]bool{
 	moerr.ErrDeadlockCheckBusy:        false,
 	moerr.ErrLockConflict:             false,
 	moerr.ErrRemoteLockWaitTimeout:    false,
+	moerr.ErrLockWaitTimeout:          false,
 	moerr.ErrTxnUnknown:               false,
 	moerr.ErrBackendClosed:            false,
 	moerr.ErrNoAvailableBackend:       false,
@@ -1307,13 +1308,19 @@ func isErrorRollbackWholeTxn(inputErr error) bool {
 }
 
 func getRandomErrorRollbackWholeTxn() error {
-	rand.NewSource(time.Now().UnixNano())
 	x := rand.Intn(len(errCodeRollbackWholeTxn))
 	arr := make([]uint16, 0, len(errCodeRollbackWholeTxn))
 	for k := range errCodeRollbackWholeTxn {
 		arr = append(arr, k)
 	}
-	switch arr[x] {
+	return newErrorRollbackWholeTxn(arr[x])
+}
+
+// newErrorRollbackWholeTxn keeps the test error factory in sync with
+// errCodeRollbackWholeTxn. Its deterministic input lets tests cover every map
+// entry instead of relying on getRandomErrorRollbackWholeTxn to select it.
+func newErrorRollbackWholeTxn(code uint16) error {
+	switch code {
 	case moerr.ErrRetryForCNRollingRestart:
 		return moerr.NewRetryForCNRollingRestart()
 	case moerr.ErrDeadLockDetected:
@@ -1328,6 +1335,8 @@ func getRandomErrorRollbackWholeTxn() error {
 		return moerr.NewLockConflictNoCtx()
 	case moerr.ErrRemoteLockWaitTimeout:
 		return moerr.NewRemoteLockWaitTimeoutNoCtx()
+	case moerr.ErrLockWaitTimeout:
+		return moerr.NewLockWaitTimeoutNoCtx()
 	case moerr.ErrTxnUnknown:
 		return moerr.NewTxnUnknown(context.Background(), "test")
 	case moerr.ErrBackendClosed:
@@ -1337,7 +1346,7 @@ func getRandomErrorRollbackWholeTxn() error {
 	case moerr.ErrBackendCannotConnect:
 		return moerr.NewBackendCannotConnectNoCtx("test")
 	default:
-		panic(fmt.Sprintf("usp error code %d", arr[x]))
+		panic(fmt.Sprintf("unsupported error code %d", code))
 	}
 }
 

pkg/frontend/util_test.go

@@ -1057,6 +1057,7 @@ func (t testErr) Error() string {
 func Test_isErrorRollbackWholeTxn(t *testing.T) {
 	assert.Equal(t, false, isErrorRollbackWholeTxn(nil))
 	assert.Equal(t, false, isErrorRollbackWholeTxn(&testError{}))
+	assert.Equal(t, true, isErrorRollbackWholeTxn(moerr.NewLockWaitTimeoutNoCtx()))
 	assert.Equal(t, true, isErrorRollbackWholeTxn(moerr.NewRetryForCNRollingRestart()))
 	assert.Equal(t, true, isErrorRollbackWholeTxn(moerr.NewDeadLockDetectedNoCtx()))
 	assert.Equal(t, true, isErrorRollbackWholeTxn(moerr.NewLockTableBindChangedNoCtx()))
@@ -1070,6 +1071,16 @@ func Test_isErrorRollbackWholeTxn(t *testing.T) {
 	assert.Equal(t, true, isErrorRollbackWholeTxn(moerr.NewBackendCannotConnectNoCtx("test")))
 }
 
+func TestNewErrorRollbackWholeTxnCoversEveryCode(t *testing.T) {
+	for code := range errCodeRollbackWholeTxn {
+		err := newErrorRollbackWholeTxn(code)
+		require.True(t, isErrorRollbackWholeTxn(err), "error code %d", code)
+		moErr, ok := err.(*moerr.Error)
+		require.True(t, ok, "error code %d returned %T", code, err)
+		require.Equal(t, code, moErr.ErrorCode())
+	}
+}
+
 func TestUserInput_getSqlSourceType(t *testing.T) {
 	type fields struct {
 		sql           string

pkg/frontend/variables.go

@@ -33,6 +33,10 @@ import (
 	"github.com/matrixorigin/matrixone/pkg/util/gpumode"
 )
 
+// defaultLockWaitTimeoutSeconds is the transitional frontend fallback. Long
+// internal jobs should supply a task-owned deadline instead of relying on it.
+const defaultLockWaitTimeoutSeconds int64 = 120
+
 var (
 	errorConvertToBoolFailed                   = moerr.NewInternalError(context.Background(), "convert to the system variable bool type failed")
 	errorConvertToIntFailed                    = moerr.NewInternalError(context.Background(), "convert to the system variable int type failed")
@@ -2166,7 +2170,9 @@ var gSysVarsDefs = map[string]SystemVariable{
 		Dynamic:           true,
 		SetVarHintApplies: true,
 		Type:              InitSystemVariableIntType("lock_wait_timeout", 1, 31536000, false),
-		Default:           int64(31536000),
+		// Keep the default bounded so a single abandoned or slow transaction
+		// cannot stall every waiter behind the same row lock for hours.
+		Default: defaultLockWaitTimeoutSeconds,
 	},
 	"locked_in_memory": {
 		Name:              "locked_in_memory",

pkg/frontend/variables_test.go

@@ -35,6 +35,18 @@ func TestEventSchedulerDefaultDisabled(t *testing.T) {
 	})
 }
 
+func TestLockWaitTimeoutDefaultIsBounded(t *testing.T) {
+	convey.Convey("lock_wait_timeout default should fail fast", t, func() {
+		sv, ok := gSysVarsDefs["lock_wait_timeout"]
+		convey.So(ok, convey.ShouldBeTrue)
+		convey.So(sv.Default, convey.ShouldEqual, defaultLockWaitTimeoutSeconds)
+
+		got, err := sv.Type.Convert(sv.Default)
+		convey.So(err, convey.ShouldBeNil)
+		convey.So(got, convey.ShouldEqual, defaultLockWaitTimeoutSeconds)
+	})
+}
+
 func TestScope(t *testing.T) {
 	convey.Convey("test scope", t, func() {
 		wanted := make(map[Scope]string)

pkg/lockservice/cfg.go

@@ -30,6 +30,7 @@ var (
 	defaultRemoteLockTimeout      = time.Minute * 10
 	defaultRemoteLockOwnerTimeout = time.Minute * 2
 	defaultRemoteTxnTimeout       = time.Second * 10
+	defaultMaxLockWaitDuration    = time.Hour
 )
 
 // Config lock service config
@@ -64,6 +65,12 @@ type Config struct {
 	// RemoteLockOwnerWaitTimeout is the owner-side wait cap for remote Lock RPC
 	// handling. A nil value uses the default. A non-nil zero duration disables it.
 	RemoteLockOwnerWaitTimeout *toml.Duration `toml:"remote-lock-owner-wait-timeout"`
+	// MaxLockWaitDuration is the lockservice safety ceiling for a waiter. It
+	// applies when the caller omits LockWaitTimeout and caps larger caller
+	// values. This keeps every lockservice wait bounded even if an internal
+	// execution path forgets to propagate a session or task deadline. Callers
+	// that retry across Lock calls still need to own and propagate a deadline.
+	MaxLockWaitDuration toml.Duration `toml:"max-lock-wait-duration"`
 	// MaxLockRowCount each time a lock is added, some LockRow is stored in the lockservice, if
 	// too many LockRows are put in each time, it will cause too much memory overhead, this value
 	// limits the maximum count of LocRow put into the LockService each time, beyond this value it
@@ -110,6 +117,12 @@ func (c *Config) Validate() {
 	if c.RemoteLockOwnerWaitTimeout == nil {
 		c.RemoteLockOwnerWaitTimeout = &toml.Duration{Duration: defaultRemoteLockOwnerTimeout}
 	}
+	if c.MaxLockWaitDuration.Duration < 0 {
+		panic("max-lock-wait-duration must not be negative")
+	}
+	if c.MaxLockWaitDuration.Duration == 0 {
+		c.MaxLockWaitDuration.Duration = defaultMaxLockWaitDuration
+	}
 	if c.KeepBindTimeout.Duration == 0 {
 		c.KeepBindTimeout.Duration = defaultKeepBindTimeout
 	}

pkg/lockservice/cfg_test.go

@@ -32,6 +32,7 @@ func TestAdjustConfig(t *testing.T) {
 	assert.NotEmpty(t, c.KeepRemoteLockDuration)
 	assert.NotNil(t, c.RemoteLockOwnerWaitTimeout)
 	assert.NotEmpty(t, c.RemoteLockOwnerWaitTimeout.Duration)
+	assert.Equal(t, defaultMaxLockWaitDuration, c.MaxLockWaitDuration.Duration)
 	assert.NotEmpty(t, c.MaxFixedSliceSize)
 }
 
@@ -44,3 +45,9 @@ func TestRemoteLockOwnerWaitTimeoutCanBeDisabled(t *testing.T) {
 	require.NotNil(t, c.RemoteLockOwnerWaitTimeout)
 	assert.Equal(t, time.Duration(0), c.RemoteLockOwnerWaitTimeout.Duration)
 }
+
+func TestAdjustConfigRejectsNegativeMaxLockWaitDuration(t *testing.T) {
+	c := Config{ServiceID: "s1"}
+	c.MaxLockWaitDuration.Duration = -1
+	assert.Panics(t, c.Validate)
+}

pkg/lockservice/lock_table_local.go

@@ -199,7 +199,7 @@ func (l *localLockTable) doLock(
 			if lockWaitTimeoutHit {
 				// lock_wait_timeout expired: return ErrLockTimeout directly
 				// (not errors.Join) so upper layers can recognize it via
-				// moerr.IsMoErrCode(err, moerr.ErrInvalidState).
+				// moerr.IsMoErrCode(err, moerr.ErrLockWaitTimeout).
 				v.err = ErrLockTimeout
 			}
 		}

pkg/lockservice/lock_table_remote.go

@@ -112,7 +112,9 @@ func (l *remoteLockTable) lock(
 	// When session-level lock_wait_timeout is set, bound the RPC by that
 	// timeout plus slack so the lock-table owner has enough time to observe
 	// and return ErrLockTimeout before the client-side RPC deadline fires.
-	// Without a session timeout, use the caller context as-is.
+	// Service entry points also use this field for the safety ceiling. A zero
+	// value is possible only for direct lock-table callers and tests, where the
+	// caller context remains the fallback.
 	var rpcCtx context.Context
 	var rpcCancel context.CancelFunc
 	if d := time.Duration(opts.LockWaitTimeout) * time.Second; d > 0 {

pkg/lockservice/service.go

@@ -37,6 +37,7 @@ import (
 	"github.com/matrixorigin/matrixone/pkg/util/list"
 	v2 "github.com/matrixorigin/matrixone/pkg/util/metric/v2"
 	"github.com/matrixorigin/matrixone/pkg/util/trace"
+	"go.uber.org/zap"
 )
 
 // WithWait setup wait func to wait some condition ready
@@ -47,19 +48,22 @@ func WithWait(wait func()) Option {
 }
 
 type service struct {
-	cfg                  Config
-	serviceID            string
-	tableGroups          *lockTableHolders
-	activeTxnHolder      activeTxnHolder
-	fsp                  *fixedSlicePool
-	deadlockDetector     *detector
-	events               *waiterEvents
-	clock                clock.Clock
-	stopper              *stopper.Stopper
-	stopOnce             sync.Once
-	bindChangeMu         sync.RWMutex
-	fetchWhoWaitingListC chan who
-	logger               *log.MOLogger
+	cfg              Config
+	serviceID        string
+	tableGroups      *lockTableHolders
+	activeTxnHolder  activeTxnHolder
+	fsp              *fixedSlicePool
+	deadlockDetector *detector
+	events           *waiterEvents
+	clock            clock.Clock
+	stopper          *stopper.Stopper
+	stopOnce         sync.Once
+	// lockWaitCeilingWarned prevents a large explicit timeout from logging on
+	// every lock operation. The metric still counts every clamped request.
+	lockWaitCeilingWarned atomic.Bool
+	bindChangeMu          sync.RWMutex
+	fetchWhoWaitingListC  chan who
+	logger                *log.MOLogger
 
 	remote struct {
 		client Client
@@ -132,6 +136,7 @@ func (s *service) Lock(
 	rows [][]byte,
 	txnID []byte,
 	options pb.LockOptions) (pb.Result, error) {
+	options = s.applyLockWaitTimeoutCeiling(options)
 
 	if !s.canLockOnServiceStatus(txnID, options, tableID, rows) {
 		return pb.Result{}, moerr.NewNewTxnInCNRollingRestart()
@@ -218,6 +223,62 @@ func (s *service) Lock(
 	return result, err
 }
 
+// applyLockWaitTimeoutCeiling bounds missing or oversized wait budgets and
+// puts the effective absolute deadline in the returned options. Carrying that
+// deadline keeps local-to-remote/forward hops on one budget. Lock receives
+// options by value, so callers that retry by invoking Lock again must propagate
+// their own deadline; this service-side safety net cannot update their copy.
+func (s *service) applyLockWaitTimeoutCeiling(options pb.LockOptions) pb.LockOptions {
+	ceiling := s.cfg.MaxLockWaitDuration.Duration
+	if ceiling <= 0 {
+		return options
+	}
+	// LockWaitTimeout is encoded as whole seconds. Round up so a positive
+	// sub-second ceiling or remaining budget never becomes an unbounded zero.
+	seconds := int64(ceiling / time.Second)
+	if ceiling%time.Second != 0 {
+		seconds++
+	}
+	if seconds <= 0 {
+		seconds = 1
+	}
+
+	now := time.Now()
+	requested := options.LockWaitTimeout
+	effectiveSeconds := requested
+	if effectiveSeconds <= 0 || effectiveSeconds > seconds {
+		effectiveSeconds = seconds
+	}
+	effectiveDeadline := now.Add(time.Duration(effectiveSeconds) * time.Second)
+	if options.LockWaitDeadline > 0 {
+		callerDeadline := time.Unix(0, options.LockWaitDeadline)
+		if callerDeadline.Before(effectiveDeadline) {
+			effectiveDeadline = callerDeadline
+			effectiveSeconds = int64(effectiveDeadline.Sub(now) / time.Second)
+			if effectiveDeadline.Sub(now)%time.Second != 0 {
+				effectiveSeconds++
+			}
+			if effectiveSeconds <= 0 {
+				effectiveSeconds = 1
+			}
+		}
+	}
+	options.LockWaitTimeout = effectiveSeconds
+	options.LockWaitDeadline = effectiveDeadline.UnixNano()
+
+	if requested > seconds {
+		v2.TxnLockWaitTimeoutCeilingClampedCounter.Inc()
+		if s.lockWaitCeilingWarned.CompareAndSwap(false, true) && s.logger != nil {
+			s.logger.Warn("lock wait timeout exceeds lockservice safety ceiling; request was clamped",
+				zap.Int64("requested-seconds", requested),
+				zap.Duration("max-lock-wait-duration", ceiling),
+				zap.Int64("effective-seconds", effectiveSeconds),
+				zap.Time("effective-deadline", effectiveDeadline))
+		}
+	}
+	return options
+}
+
 func (s *service) Unlock(
 	ctx context.Context,
 	txnID []byte,

pkg/lockservice/service_remote.go

@@ -188,6 +188,7 @@ func (s *service) handleRemoteLock(
 	req *pb.Request,
 	resp *pb.Response,
 	cs morpc.ClientSession) {
+	req.Lock.Options = s.applyLockWaitTimeoutCeiling(req.Lock.Options)
 	logFields := remoteLockResponseLogFields(req)
 	if !s.canLockOnServiceStatus(req.Lock.TxnID, req.Lock.Options, req.LockTable.Table, req.Lock.Rows) {
 		_ = writeResponseWithDeadline(s.logger, cancel, resp, moerr.NewRetryForCNRollingRestart(), cs, defaultRPCWriteTimeout, logFields)
@@ -286,6 +287,7 @@ func (s *service) handleForwardLock(
 	req *pb.Request,
 	resp *pb.Response,
 	cs morpc.ClientSession) {
+	req.Lock.Options = s.applyLockWaitTimeoutCeiling(req.Lock.Options)
 	logFields := remoteLockResponseLogFields(req)
 	if !s.canLockOnServiceStatus(req.Lock.TxnID, req.Lock.Options, req.LockTable.Table, req.Lock.Rows) {
 		_ = writeResponseWithDeadline(s.logger, cancel, resp, moerr.NewRetryForCNRollingRestart(), cs, defaultRPCWriteTimeout, logFields)

pkg/lockservice/service_remote_test.go

@@ -826,7 +826,7 @@ func TestRemoteLockWaitTimeout_PrecisionIndependentOfLazyCheck(t *testing.T) {
 			elapsed := time.Since(start)
 
 			require.Error(t, err)
-			require.True(t, moerr.IsMoErrCode(err, moerr.ErrInvalidState),
+			require.True(t, moerr.IsMoErrCode(err, moerr.ErrLockWaitTimeout),
 				"expected lock-timeout, got %v", err)
 			// Must fire well before the 10s coarse tick.
 			require.Less(t, elapsed, 3*time.Second,
@@ -1013,9 +1013,9 @@ func TestRemoteLockWaitTimeout_ReturnsLockTimeout(t *testing.T) {
 
 			require.Error(t, err)
 			// Must receive lock-timeout, not connectivity/backend error.
-			require.True(t, moerr.IsMoErrCode(err, moerr.ErrInvalidState),
-				"expected ErrLockTimeout (InvalidState), got %v", err)
-			require.Contains(t, err.Error(), "lock timeout",
+			require.True(t, moerr.IsMoErrCode(err, moerr.ErrLockWaitTimeout),
+				"expected ErrLockWaitTimeout, got %v", err)
+			require.Contains(t, err.Error(), "Lock wait timeout exceeded",
 				"expected lock timeout message, got %v", err)
 			require.GreaterOrEqual(t, elapsed, time.Second,
 				"should have waited at least LockWaitTimeout")