diff --git a/pkg/common/moerr/error.go b/pkg/common/moerr/error.go index 8fe1e4618014b..823de817db115 100644 --- a/pkg/common/moerr/error.go +++ b/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) } diff --git a/pkg/common/moerr/error_no_ctx.go b/pkg/common/moerr/error_no_ctx.go index 8cc3c9c344afe..b89c6f52c8e5f 100644 --- a/pkg/common/moerr/error_no_ctx.go +++ b/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) } diff --git a/pkg/common/moerr/error_test.go b/pkg/common/moerr/error_test.go index 99a2d08f4f64d..3d51ec557105a 100644 --- a/pkg/common/moerr/error_test.go +++ b/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)) diff --git a/pkg/frontend/util.go b/pkg/frontend/util.go index c4ecc26297270..b28780c67e810 100644 --- a/pkg/frontend/util.go +++ b/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)) } } diff --git a/pkg/frontend/util_test.go b/pkg/frontend/util_test.go index cb89b7b7c049c..444190a56a3cb 100644 --- a/pkg/frontend/util_test.go +++ b/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 diff --git a/pkg/frontend/variables.go b/pkg/frontend/variables.go index c8650be131c2d..467438e7f93ab 100644 --- a/pkg/frontend/variables.go +++ b/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", diff --git a/pkg/frontend/variables_test.go b/pkg/frontend/variables_test.go index 6b53e593d770a..7dd7297733465 100644 --- a/pkg/frontend/variables_test.go +++ b/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) diff --git a/pkg/lockservice/cfg.go b/pkg/lockservice/cfg.go index c07a97f221007..a7b1280164054 100644 --- a/pkg/lockservice/cfg.go +++ b/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 } diff --git a/pkg/lockservice/cfg_test.go b/pkg/lockservice/cfg_test.go index 43e6082e3b2f8..4dfe610b8d278 100644 --- a/pkg/lockservice/cfg_test.go +++ b/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) +} diff --git a/pkg/lockservice/lock_table_local.go b/pkg/lockservice/lock_table_local.go index f7de893ba26de..d2776d9d324e2 100644 --- a/pkg/lockservice/lock_table_local.go +++ b/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 } } diff --git a/pkg/lockservice/lock_table_remote.go b/pkg/lockservice/lock_table_remote.go index 5418bf937fa39..8b81c0fb7505a 100644 --- a/pkg/lockservice/lock_table_remote.go +++ b/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 { diff --git a/pkg/lockservice/service.go b/pkg/lockservice/service.go index 115db132c4a98..f16bb6cc3e696 100644 --- a/pkg/lockservice/service.go +++ b/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, diff --git a/pkg/lockservice/service_remote.go b/pkg/lockservice/service_remote.go index 90ef055c263c6..b5faa5b33c9fb 100644 --- a/pkg/lockservice/service_remote.go +++ b/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) diff --git a/pkg/lockservice/service_remote_test.go b/pkg/lockservice/service_remote_test.go index 40345f6151371..26194c196e69f 100644 --- a/pkg/lockservice/service_remote_test.go +++ b/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") diff --git a/pkg/lockservice/service_test.go b/pkg/lockservice/service_test.go index ff3202aad0b02..4e39748b2d7b3 100644 --- a/pkg/lockservice/service_test.go +++ b/pkg/lockservice/service_test.go @@ -32,6 +32,8 @@ import ( "github.com/matrixorigin/matrixone/pkg/common/runtime" pb "github.com/matrixorigin/matrixone/pkg/pb/lock" "github.com/matrixorigin/matrixone/pkg/pb/timestamp" + v2 "github.com/matrixorigin/matrixone/pkg/util/metric/v2" + "github.com/prometheus/client_golang/prometheus/testutil" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "go.uber.org/zap/zapcore" @@ -4548,16 +4550,86 @@ func TestLockWaitTimeout(t *testing.T) { // Should time out after ~1 second, not immediately and not indefinitely. require.Error(t, err) - require.True(t, moerr.IsMoErrCode(err, moerr.ErrInvalidState), - "expected lock-timeout error (ErrInvalidState), got %v", err) - require.Contains(t, err.Error(), "lock timeout") + require.True(t, moerr.IsMoErrCode(err, moerr.ErrLockWaitTimeout), + "expected lock-wait-timeout error, got %v", err) + require.Contains(t, err.Error(), "Lock wait timeout exceeded") require.GreaterOrEqual(t, elapsed, time.Second) require.Less(t, elapsed, 3*time.Second) }, ) } -func TestLockWaitTimeoutDefaultNoTimeout(t *testing.T) { +func TestLockWaitTimeoutCeilingBoundsMissingCallerTimeout(t *testing.T) { + runLockServiceTestsWithAdjustConfig( + t, + []string{"s1"}, + time.Second*10, + func(alloc *lockTableAllocator, s []*service) { + l := s[0] + option := pb.LockOptions{ + Granularity: pb.Granularity_Row, + Mode: pb.LockMode_Exclusive, + Policy: pb.WaitPolicy_Wait, + } + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + _, err := l.Lock(ctx, 0, [][]byte{{1}}, []byte("txn1"), option) + require.NoError(t, err) + + start := time.Now() + _, err = l.Lock(ctx, 0, [][]byte{{1}}, []byte("txn2"), option) + elapsed := time.Since(start) + + require.True(t, moerr.IsMoErrCode(err, moerr.ErrLockWaitTimeout), + "expected safety-ceiling lock-wait-timeout error, got %v", err) + require.GreaterOrEqual(t, elapsed, time.Second) + require.Less(t, elapsed, 3*time.Second) + }, + func(c *Config) { + c.MaxLockWaitDuration.Duration = time.Second + }) +} + +func TestApplyLockWaitTimeoutCeiling(t *testing.T) { + s := &service{} + s.cfg.MaxLockWaitDuration.Duration = 1500 * time.Millisecond + + metricBefore := testutil.ToFloat64(v2.TxnLockWaitTimeoutCeilingClampedCounter) + start := time.Now() + injected := s.applyLockWaitTimeoutCeiling(pb.LockOptions{}) + require.Equal(t, int64(2), injected.LockWaitTimeout) + require.WithinDuration(t, start.Add(2*time.Second), time.Unix(0, injected.LockWaitDeadline), 100*time.Millisecond) + require.Equal(t, metricBefore, testutil.ToFloat64(v2.TxnLockWaitTimeoutCeilingClampedCounter), + "injecting a missing timeout is the normal safety-net path") + reapplied := s.applyLockWaitTimeoutCeiling(injected) + require.Equal(t, injected.LockWaitDeadline, reapplied.LockWaitDeadline, + "remote/forward owner must keep the deadline injected at the first service entry") + require.LessOrEqual(t, reapplied.LockWaitTimeout, injected.LockWaitTimeout) + + start = time.Now() + shorter := s.applyLockWaitTimeoutCeiling(pb.LockOptions{LockWaitTimeout: 1}) + require.Equal(t, int64(1), shorter.LockWaitTimeout) + require.WithinDuration(t, start.Add(time.Second), time.Unix(0, shorter.LockWaitDeadline), 100*time.Millisecond) + + callerDeadline := time.Now().Add(500 * time.Millisecond).UnixNano() + withEarlierDeadline := s.applyLockWaitTimeoutCeiling(pb.LockOptions{ + LockWaitTimeout: 1, + LockWaitDeadline: callerDeadline, + }) + require.Equal(t, int64(1), withEarlierDeadline.LockWaitTimeout) + require.Equal(t, callerDeadline, withEarlierDeadline.LockWaitDeadline) + + start = time.Now() + clamped := s.applyLockWaitTimeoutCeiling(pb.LockOptions{LockWaitTimeout: 30}) + require.Equal(t, int64(2), clamped.LockWaitTimeout) + require.WithinDuration(t, start.Add(2*time.Second), time.Unix(0, clamped.LockWaitDeadline), 100*time.Millisecond) + require.True(t, s.lockWaitCeilingWarned.Load()) + require.Equal(t, metricBefore+1, testutil.ToFloat64(v2.TxnLockWaitTimeoutCeilingClampedCounter)) +} + +func TestLockWaitTimeoutCallerContextBeforeCeiling(t *testing.T) { runLockServiceTests( t, []string{"s1"}, @@ -4579,7 +4651,7 @@ func TestLockWaitTimeoutDefaultNoTimeout(t *testing.T) { require.NoError(t, err) // txn2 tries to lock the same row WITHOUT LockWaitTimeout. - // Should be blocked until ctx expires (no internal/default timeout interception). + // The caller context is earlier than the one-hour safety ceiling. option2 := option option2.LockWaitTimeout = 0 // no session/internal timeout; rely on caller context deadline start := time.Now() @@ -4636,7 +4708,7 @@ func TestLockWaitTimeoutSucceedsWhenHolderReleases(t *testing.T) { ) } -func TestLockWaitTimeoutZeroMeansFallbackToContext(t *testing.T) { +func TestLockWaitTimeoutZeroUsesEarlierCallerContext(t *testing.T) { runLockServiceTests( t, []string{"s1"}, @@ -4657,8 +4729,8 @@ func TestLockWaitTimeoutZeroMeansFallbackToContext(t *testing.T) { _, err := l.Lock(ctx, 0, [][]byte{{1}}, []byte("txn1"), option) require.NoError(t, err) - // txn2 with LockWaitTimeout=0 should wait for context expiry (500ms), - // NOT the default 5-minute configLockWaitTimeout. + // txn2 with LockWaitTimeout=0 should use the earlier context expiry + // instead of waiting for the one-hour safety ceiling. option2 := option option2.LockWaitTimeout = 0 start := time.Now() @@ -4803,6 +4875,7 @@ func maybeAddTestLockWithDeadlockWithWaitRetry( if moerr.IsMoErrCode(err, moerr.ErrDeadLockDetected) || moerr.IsMoErrCode(err, moerr.ErrTxnNotFound) || + moerr.IsMoErrCode(err, moerr.ErrLockWaitTimeout) || moerr.IsMoErrCode(err, moerr.ErrInvalidState) { return res } diff --git a/pkg/lockservice/types.go b/pkg/lockservice/types.go index cc293d30b843b..d9a45e460f600 100644 --- a/pkg/lockservice/types.go +++ b/pkg/lockservice/types.go @@ -43,7 +43,7 @@ var ( // ErrLockConflict lock option conflict ErrLockConflict = moerr.NewLockConflictNoCtx() // ErrLockTimeout lock table timeout - ErrLockTimeout = moerr.NewInvalidStateNoCtx("lock timeout") + ErrLockTimeout = moerr.NewLockWaitTimeoutNoCtx() // ErrRemoteLockWaitTimeout remote lock owner-side wait timeout ErrRemoteLockWaitTimeout = moerr.NewRemoteLockWaitTimeoutNoCtx() ) diff --git a/pkg/lockservice/waiter.go b/pkg/lockservice/waiter.go index e58e91e81f67e..25564641b2233 100644 --- a/pkg/lockservice/waiter.go +++ b/pkg/lockservice/waiter.go @@ -98,8 +98,8 @@ type waiter struct { enableChecker bool // lockWaitTimeout is the session-level SET lock_wait_timeout value. - // A zero value means no session-level timeout is enforced here; in that - // case waiting relies on the context or other external cancellation. + // A zero value means no caller timeout was attached at this raw lock-table + // layer; service entry points normally replace it with the safety ceiling. // Set in waiterEvents.add() from the lockContext and checked in // waiterEvents.check() to enforce timeouts on the async (remote) lock path. lockWaitTimeout time.Duration @@ -107,6 +107,9 @@ type waiter struct { lockWaitGranularity pb.Granularity lockWaitMode pb.LockMode lockWaitTimer atomic.Pointer[time.Timer] + // waitTooLongLogged is per waiter lifecycle and is reset before reuse. + // It suppresses repeated diagnostics without disabling orphan checks. + waitTooLongLogged atomic.Bool // just used for testing beforeSwapStatusAdjustFunc func() @@ -320,6 +323,7 @@ func (w *waiter) reset() { w.lockWaitTimeoutErr = nil w.lockWaitGranularity = pb.Granularity_Row w.lockWaitMode = pb.LockMode_Exclusive + w.waitTooLongLogged.Store(false) w.stopLockWaitTimer() } diff --git a/pkg/lockservice/waiter_events.go b/pkg/lockservice/waiter_events.go index 6114981d1cb09..1d086b12a81e9 100644 --- a/pkg/lockservice/waiter_events.go +++ b/pkg/lockservice/waiter_events.go @@ -386,7 +386,7 @@ func (mw *waiterEvents) checkOrphan(v checkOrphan) { return } - if v.wait >= waitTooLong { + if v.logWaitTooLong { lockDetail := "" v.lt.mu.RLock() lock, ok := v.lt.mu.store.Get(v.key) @@ -461,16 +461,22 @@ func (mw *waiterEvents) addToOrphanCheck( wait time.Duration, ) { ck := *w.conflictKey.Load() + logWaitTooLong := wait >= waitTooLong && w.waitTooLongLogged.CompareAndSwap(false, true) v := checkOrphan{ - wait: wait, - key: ck, - lt: w.lt.Load(), - txn: w.txn, + wait: wait, + key: ck, + lt: w.lt.Load(), + txn: w.txn, + logWaitTooLong: logWaitTooLong, } select { case mw.checkOrphanC <- v: default: + if logWaitTooLong { + // The warning was not queued. Let a later check retry it. + w.waitTooLongLogged.Store(false) + } } } @@ -479,4 +485,7 @@ type checkOrphan struct { key []byte lt *localLockTable txn pb.WaitTxn + // logWaitTooLong controls only the diagnostic; every event still performs + // the orphan check regardless of this flag. + logWaitTooLong bool } diff --git a/pkg/lockservice/waiter_test.go b/pkg/lockservice/waiter_test.go index ee71cb9fe63f0..3494d20251a49 100644 --- a/pkg/lockservice/waiter_test.go +++ b/pkg/lockservice/waiter_test.go @@ -97,6 +97,43 @@ func TestWaitMultiTimes(t *testing.T) { }) } +func TestWaitTooLongLoggedOncePerWaiterLifecycle(t *testing.T) { + reuse.RunReuseTests(func() { + w := acquireWaiter(pb.WaitTxn{TxnID: []byte("w")}, "", nil) + defer w.close("", nil) + key := []byte{1} + lt := &localLockTable{} + w.conflictKey.Store(&key) + w.lt.Store(lt) + events := &waiterEvents{checkOrphanC: make(chan checkOrphan, 2)} + + events.addToOrphanCheck(w, waitTooLong) + require.True(t, (<-events.checkOrphanC).logWaitTooLong) + require.True(t, w.waitTooLongLogged.Load()) + + events.addToOrphanCheck(w, waitTooLong+time.Second) + require.False(t, (<-events.checkOrphanC).logWaitTooLong) + + w.reset() + require.False(t, w.waitTooLongLogged.Load()) + w.conflictKey.Store(&key) + w.lt.Store(lt) + + for len(events.checkOrphanC) < cap(events.checkOrphanC) { + events.checkOrphanC <- checkOrphan{} + } + events.addToOrphanCheck(w, waitTooLong) + require.False(t, w.waitTooLongLogged.Load(), + "a full diagnostics queue must not suppress every later warning") + for len(events.checkOrphanC) > 0 { + <-events.checkOrphanC + } + + events.addToOrphanCheck(w, waitTooLong) + require.True(t, (<-events.checkOrphanC).logWaitTooLong) + }) +} + func TestNotifyAfterCompleted(t *testing.T) { reuse.RunReuseTests(func() { w := acquireWaiter(pb.WaitTxn{}, "", nil) diff --git a/pkg/pb/lock/lock.go b/pkg/pb/lock/lock.go index 12e35c0551f20..81de4aec1e048 100644 --- a/pkg/pb/lock/lock.go +++ b/pkg/pb/lock/lock.go @@ -136,8 +136,8 @@ func (m LockOptions) WithWaitPolicy(policy WaitPolicy) LockOptions { return m } -// WithLockWaitTimeout sets the lock wait timeout in seconds. 0 disables -// lock-wait-timeout enforcement and relies on the caller context instead. +// WithLockWaitTimeout sets the caller lock wait timeout in seconds. A zero +// value delegates to the caller context and the lockservice safety ceiling. func (m LockOptions) WithLockWaitTimeout(seconds int64) LockOptions { m.LockWaitTimeout = seconds return m diff --git a/pkg/sql/colexec/lockop/lock_op.go b/pkg/sql/colexec/lockop/lock_op.go index 006502b62bed6..4777b7937a59b 100644 --- a/pkg/sql/colexec/lockop/lock_op.go +++ b/pkg/sql/colexec/lockop/lock_op.go @@ -792,6 +792,22 @@ type lockRetryState struct { } func lockWaitTimeout(proc *process.Process, txnOp client.TxnOperator) time.Duration { + txnTimeout := client.LockWaitTimeoutFromTxn(txnOp) + // Background/internal execution may carry a per-execution value in the + // process or txn options while its resolver only exposes compiled global + // defaults. Prefer the caller-owned budget in that case. Frontend execution + // keeps resolver-first semantics so SET SESSION and statement overrides are + // observed even after a transaction has started. + if proc != nil && proc.Base != nil && !proc.Base.IsFrontend { + if proc.GetSessionInfo() != nil { + if seconds := proc.GetSessionInfo().LockWaitTimeout; seconds > 0 { + return time.Duration(seconds) * time.Second + } + } + if txnTimeout > 0 { + return txnTimeout + } + } if proc != nil && proc.GetResolveVariableFunc() != nil { if v, err := proc.GetResolveVariableFunc()("lock_wait_timeout", true, false); err == nil { switch n := v.(type) { @@ -815,7 +831,7 @@ func lockWaitTimeout(proc *process.Process, txnOp client.TxnOperator) time.Durat return time.Duration(seconds) * time.Second } } - return client.LockWaitTimeoutFromTxn(txnOp) + return txnTimeout } func refreshLockWaitOptions(options lock.LockOptions) (lock.LockOptions, error) { diff --git a/pkg/sql/colexec/lockop/lock_op_test.go b/pkg/sql/colexec/lockop/lock_op_test.go index 3f8ff396e1c09..c3f9c15397c1c 100644 --- a/pkg/sql/colexec/lockop/lock_op_test.go +++ b/pkg/sql/colexec/lockop/lock_op_test.go @@ -97,6 +97,7 @@ func TestLockWaitTimeoutUsesCurrentSessionValue(t *testing.T) { nil, nil, nil) + proc.Base.IsFrontend = true proc.SetResolveVariableFunc(func(varName string, isSystemVar, isGlobalVar bool) (interface{}, error) { require.Equal(t, "lock_wait_timeout", varName) require.True(t, isSystemVar) @@ -111,6 +112,16 @@ func TestLockWaitTimeoutUsesCurrentSessionValue(t *testing.T) { proc.GetSessionInfo().LockWaitTimeout = 0 require.Equal(t, 60*time.Second, lockWaitTimeout(proc, txnOp)) + + proc.Base.IsFrontend = false + proc.SetResolveVariableFunc(func(string, bool, bool) (interface{}, error) { + return int64(2), nil + }) + require.Equal(t, 60*time.Second, lockWaitTimeout(proc, txnOp), + "background per-execution txn option must override the default resolver") + proc.GetSessionInfo().LockWaitTimeout = 4 + require.Equal(t, 4*time.Second, lockWaitTimeout(proc, txnOp), + "background process-level per-execution option must have highest priority") } func TestLockOpHelpers(t *testing.T) { @@ -157,6 +168,10 @@ func TestRefreshLockWaitOptionsReturnsTimeoutAfterDeadline(t *testing.T) { require.ErrorIs(t, err, lockservice.ErrLockTimeout) } +func TestLockWaitTimeoutIsNotRetryable(t *testing.T) { + require.False(t, isRetryLockError(lockservice.ErrLockTimeout)) +} + func TestLockOpTargetHelpers(t *testing.T) { op := NewArgument() defer op.Release() diff --git a/pkg/sql/compile/compile_test.go b/pkg/sql/compile/compile_test.go index e4b65163b984c..a4a61db644b7c 100644 --- a/pkg/sql/compile/compile_test.go +++ b/pkg/sql/compile/compile_test.go @@ -55,6 +55,7 @@ import ( plan2 "github.com/matrixorigin/matrixone/pkg/sql/plan" "github.com/matrixorigin/matrixone/pkg/testutil" "github.com/matrixorigin/matrixone/pkg/testutil/testengine" + "github.com/matrixorigin/matrixone/pkg/util/executor" "github.com/matrixorigin/matrixone/pkg/util/fault" "github.com/matrixorigin/matrixone/pkg/vm" "github.com/matrixorigin/matrixone/pkg/vm/engine" @@ -70,6 +71,24 @@ type compileTestCase struct { txnClient client.TxnClient // Store txnClient for truncating table with real transaction } +func TestApplyExecutorLockWaitTimeout(t *testing.T) { + proc := process.NewTopProcess( + context.Background(), + mpool.MustNewZero(), + nil, + nil, + nil, + nil, + nil, + nil, + nil, + nil, + nil) + + applyExecutorLockWaitTimeout(proc, executor.Options{}.WithLockWaitTimeout(1500*time.Millisecond)) + require.Equal(t, int64(2), proc.Base.SessionInfo.LockWaitTimeout) +} + func testPrint(_ *batch.Batch, crs *perfcounter.CounterSet) error { return nil } diff --git a/pkg/sql/compile/sql_executor.go b/pkg/sql/compile/sql_executor.go index f1ed1ea51dd93..5f82e22fd00c9 100644 --- a/pkg/sql/compile/sql_executor.go +++ b/pkg/sql/compile/sql_executor.go @@ -391,6 +391,7 @@ func (exec *txnExecutor) Exec( // background paths set resolvers too (idxcron's task.Metadata, // ProcessInitSQL's executor.DefaultResolveVariable). proc.Base.IsFrontend = exec.opts.IsFrontend() + applyExecutorLockWaitTimeout(proc, exec.opts) prepared := false if statementOption.HasParams() { @@ -605,6 +606,7 @@ func (exec *txnExecutor) LockTable(table string) error { exec.s.taskservice, ) proc.Base.IsFrontend = exec.opts.IsFrontend() + applyExecutorLockWaitTimeout(proc, exec.opts) proc.Base.SessionInfo.TimeZone = exec.opts.GetTimeZone() proc.Base.SessionInfo.Buf = exec.s.buf defer func() { @@ -613,6 +615,25 @@ func (exec *txnExecutor) LockTable(table string) error { return doLockTable(exec.s.eng, proc, rel, false) } +// applyExecutorLockWaitTimeout copies a per-execution background budget into +// SessionInfo, whose background lock-timeout precedence is above the default +// variable resolver. SessionInfo stores whole seconds, so positive fractions +// are rounded up rather than silently becoming an unbounded zero. +func applyExecutorLockWaitTimeout(proc *process.Process, opts executor.Options) { + if proc == nil || !opts.HasLockWaitTimeout() { + return + } + timeout := opts.LockWaitTimeout() + seconds := int64(timeout / time.Second) + if timeout%time.Second != 0 { + seconds++ + } + if seconds <= 0 { + seconds = 1 + } + proc.Base.SessionInfo.LockWaitTimeout = seconds +} + func (exec *txnExecutor) Txn() client.TxnOperator { return exec.opts.Txn() } diff --git a/pkg/util/executor/options.go b/pkg/util/executor/options.go index 41c8bd644852b..a7cb3351a8856 100644 --- a/pkg/util/executor/options.go +++ b/pkg/util/executor/options.go @@ -243,6 +243,32 @@ func (opts Options) ExtraTxnOptions() []client.TxnOption { return opts.txnOpts } +// WithLockWaitTimeout sets a per-execution lock wait budget. It is propagated +// both to newly created transactions and to the process used by an existing +// transaction, so background execution can override the global default +// resolver without changing other sessions. +func (opts Options) WithLockWaitTimeout(timeout time.Duration) Options { + opts.lockWaitTimeout = timeout + opts.hasLockWaitTimeout = timeout > 0 + // Txn options are applied in append order. Always append, including for + // zero, so a later WithLockWaitTimeout(0) intentionally clears an earlier + // value instead of leaving the first option effective. + opts.txnOpts = append(opts.txnOpts, client.WithTxnLockWaitTimeout(timeout)) + return opts +} + +// HasLockWaitTimeout reports whether this execution has a positive explicit +// lock wait budget that should override background defaults. +func (opts Options) HasLockWaitTimeout() bool { + return opts.hasLockWaitTimeout +} + +// LockWaitTimeout returns the per-execution lock wait budget. Callers should +// check HasLockWaitTimeout before treating a zero value as an explicit budget. +func (opts Options) LockWaitTimeout() time.Duration { + return opts.lockWaitTimeout +} + func (opts Options) WithEnableTrace() Options { opts.enableTrace = true return opts diff --git a/pkg/util/executor/options_test.go b/pkg/util/executor/options_test.go index 606c7412fc957..33f8f6508c6f9 100644 --- a/pkg/util/executor/options_test.go +++ b/pkg/util/executor/options_test.go @@ -16,6 +16,7 @@ package executor import ( "testing" + "time" "github.com/stretchr/testify/require" ) @@ -31,3 +32,18 @@ func TestOptionsStreaming(t *testing.T) { require.True(t, streaming) require.True(t, err_chan == errors) } + +func TestOptionsLockWaitTimeout(t *testing.T) { + var opts Options + require.False(t, opts.HasLockWaitTimeout()) + + opts = opts.WithLockWaitTimeout(1500 * time.Millisecond) + require.True(t, opts.HasLockWaitTimeout()) + require.Equal(t, 1500*time.Millisecond, opts.LockWaitTimeout()) + require.Len(t, opts.ExtraTxnOptions(), 1) + + opts = opts.WithLockWaitTimeout(0) + require.False(t, opts.HasLockWaitTimeout()) + require.Zero(t, opts.LockWaitTimeout()) + require.Len(t, opts.ExtraTxnOptions(), 2) +} diff --git a/pkg/util/executor/types.go b/pkg/util/executor/types.go index f0829d1fed92c..cee28be6e9f51 100644 --- a/pkg/util/executor/types.go +++ b/pkg/util/executor/types.go @@ -71,6 +71,8 @@ type Options struct { resolveVariableFunc func(varName string, isSystemVar, isGlobalVar bool) (interface{}, error) adjustTableExtraFunc func(*api.SchemaExtra) error keepTxnAlive bool + lockWaitTimeout time.Duration + hasLockWaitTimeout bool // isFrontend records whether the caller is a frontend // session-bound invocation. Go zero value (false) means // background: every caller of the internal SQL executor is diff --git a/pkg/util/metric/v2/txn.go b/pkg/util/metric/v2/txn.go index 2882c672fab02..5f17d6d099497 100644 --- a/pkg/util/metric/v2/txn.go +++ b/pkg/util/metric/v2/txn.go @@ -91,6 +91,10 @@ var ( TxnLockTotalCounter = txnLockCounter.WithLabelValues("total") TxnLocalLockTotalCounter = txnLockCounter.WithLabelValues("local") TxnRemoteLockTotalCounter = txnLockCounter.WithLabelValues("remote") + // TxnLockWaitTimeoutCeilingClampedCounter counts positive caller budgets + // reduced by the lockservice safety ceiling; zero-value injection is normal + // fallback behavior and is intentionally excluded. + TxnLockWaitTimeoutCeilingClampedCounter = txnLockCounter.WithLabelValues("wait-timeout-ceiling-clamped") TxnDeadlockDetectorEnqueueCounter = prometheus.NewCounterVec( prometheus.CounterOpts{ diff --git a/test/distributed/cases/pessimistic_transaction/lock_wait_timeout_remote.result b/test/distributed/cases/pessimistic_transaction/lock_wait_timeout_remote.result index a3759c7b20619..413854f346b88 100644 --- a/test/distributed/cases/pessimistic_transaction/lock_wait_timeout_remote.result +++ b/test/distributed/cases/pessimistic_transaction/lock_wait_timeout_remote.result @@ -3,6 +3,9 @@ create database lock_wait_timeout_remote_db; use lock_wait_timeout_remote_db; create table t(a int primary key, b varchar(64)); insert into t values (1, 'ONLY_FULL_GROUP_BY'), (2, 'STRICT_TRANS_TABLES'), (3, 'x'); +select @@global.lock_wait_timeout as global_timeout, @@session.lock_wait_timeout as session_timeout; +global_timeout session_timeout +120 120 set @@sql_mode = 'ONLY_FULL_GROUP_BY'; begin; set session lock_wait_timeout = 1; @@ -13,8 +16,24 @@ select a, b from t where b = @@sql_mode for update; ➤ a[4,32,0] ¦ b[12,-1,0] 𝄀 1 ¦ ONLY_FULL_GROUP_BY select a, b from t where b = @@sql_mode for update; -invalid state lock timeout +Lock wait timeout exceeded; try restarting transaction rollback; rollback; +create table timeout_rollback(a int primary key, b varchar(64)); +insert into timeout_rollback values (1, 'original-1'), (2, 'original-2'); +set session lock_wait_timeout = 1; +begin; +update timeout_rollback set b = 'must-be-rolled-back' where a = 1; +begin; +update timeout_rollback set b = 'holder' where a = 2; +update timeout_rollback set b = 'waiter' where a = 2; +Lock wait timeout exceeded; try restarting transaction +begin; +commit; +rollback; +select * from timeout_rollback order by a; +a b +1 original-1 +2 original-2 set @@sql_mode = default; drop database lock_wait_timeout_remote_db; diff --git a/test/distributed/cases/pessimistic_transaction/lock_wait_timeout_remote.sql b/test/distributed/cases/pessimistic_transaction/lock_wait_timeout_remote.sql index 2d1fcbcef3c8b..80c5de056e98d 100644 --- a/test/distributed/cases/pessimistic_transaction/lock_wait_timeout_remote.sql +++ b/test/distributed/cases/pessimistic_transaction/lock_wait_timeout_remote.sql @@ -10,6 +10,8 @@ use lock_wait_timeout_remote_db; create table t(a int primary key, b varchar(64)); insert into t values (1, 'ONLY_FULL_GROUP_BY'), (2, 'STRICT_TRANS_TABLES'), (3, 'x'); +select @@global.lock_wait_timeout as global_timeout, @@session.lock_wait_timeout as session_timeout; + set @@sql_mode = 'ONLY_FULL_GROUP_BY'; begin; @@ -22,7 +24,7 @@ begin; select a, b from t where b = @@sql_mode for update; -- @session} --- @regex("(?s)(invalid state lock timeout|lock timeout|context deadline exceeded)",true) +-- @regex("(?s)Lock wait timeout exceeded; try restarting transaction",true) select a, b from t where b = @@sql_mode for update; rollback; @@ -30,5 +32,32 @@ rollback; rollback; -- @session} +-- A lock wait timeout must roll back the whole explicit transaction. In +-- particular, a following BEGIN must not commit writes completed before the +-- timed-out statement. +create table timeout_rollback(a int primary key, b varchar(64)); +insert into timeout_rollback values (1, 'original-1'), (2, 'original-2'); +set session lock_wait_timeout = 1; + +begin; +update timeout_rollback set b = 'must-be-rolled-back' where a = 1; + +-- @session:id=1{ +begin; +update timeout_rollback set b = 'holder' where a = 2; +-- @session} + +-- @regex("(?s)Lock wait timeout exceeded; try restarting transaction",true) +update timeout_rollback set b = 'waiter' where a = 2; + +begin; +commit; + +-- @session:id=1{ +rollback; +-- @session} + +select * from timeout_rollback order by a; + set @@sql_mode = default; drop database lock_wait_timeout_remote_db;