diff --git a/etcdutl/etcdutl/backup_command.go b/etcdutl/etcdutl/backup_command.go index 49b6a9310..e0d53fc1b 100644 --- a/etcdutl/etcdutl/backup_command.go +++ b/etcdutl/etcdutl/backup_command.go @@ -322,7 +322,7 @@ func saveDB(lg *zap.Logger, destDB, srcDB string, idx uint64, term uint64, desir if !v3 { tx := be.BatchTx() - tx.Lock() + tx.LockOutsideApply() defer tx.Unlock() schema.UnsafeCreateMetaBucket(tx) schema.UnsafeUpdateConsistentIndex(tx, idx, term, false) diff --git a/etcdutl/etcdutl/migrate_command.go b/etcdutl/etcdutl/migrate_command.go index 195576e31..87b10664f 100644 --- a/etcdutl/etcdutl/migrate_command.go +++ b/etcdutl/etcdutl/migrate_command.go @@ -140,7 +140,7 @@ func migrateCommandFunc(c *migrateConfig) error { } func migrateForce(lg *zap.Logger, tx backend.BatchTx, target *semver.Version) { - tx.LockWithoutHook() + tx.LockOutsideApply() defer tx.Unlock() // Storage version is only supported since v3.6 if target.LessThan(schema.V3_6) { diff --git a/server/auth/store.go b/server/auth/store.go index 762caecd7..2d978a011 100644 --- a/server/auth/store.go +++ b/server/auth/store.go @@ -374,7 +374,7 @@ func (as *authStore) CheckPassword(username, password string) (uint64, error) { func (as *authStore) Recover(be AuthBackend) { as.be = be - tx := be.BatchTx() + tx := be.ReadTx() tx.Lock() enabled := tx.UnsafeReadAuthEnabled() @@ -939,7 +939,7 @@ func NewAuthStore(lg *zap.Logger, be AuthBackend, tp TokenProvider, bcryptCost i be.CreateAuthBuckets() tx := be.BatchTx() - // We should call LockWithoutHook here, but the txPostLockHoos isn't set + // We should call LockOutsideApply here, but the txPostLockHoos isn't set // to EtcdServer yet, so it's OK. tx.Lock() enabled := tx.UnsafeReadAuthEnabled() diff --git a/server/etcdserver/adapters.go b/server/etcdserver/adapters.go index 5f1bcfef1..d875cf14e 100644 --- a/server/etcdserver/adapters.go +++ b/server/etcdserver/adapters.go @@ -94,7 +94,7 @@ func (s *serverVersionAdapter) UpdateStorageVersion(target semver.Version) error defer s.bemu.RUnlock() tx := s.be.BatchTx() - tx.LockWithoutHook() + tx.LockOutsideApply() defer tx.Unlock() return schema.UnsafeMigrate(s.lg, tx, s.r.storage, target) } diff --git a/server/etcdserver/bootstrap.go b/server/etcdserver/bootstrap.go index 857e7afa6..43605be5e 100644 --- a/server/etcdserver/bootstrap.go +++ b/server/etcdserver/bootstrap.go @@ -231,7 +231,7 @@ func bootstrapBackend(cfg config.ServerConfig, haveWAL bool, st v2store.Store, s } } if beExist { - err = schema.Validate(cfg.Logger, be.BatchTx()) + err = schema.Validate(cfg.Logger, be.ReadTx()) if err != nil { cfg.Logger.Error("Failed to validate schema", zap.Error(err)) return nil, err diff --git a/server/etcdserver/cindex/cindex.go b/server/etcdserver/cindex/cindex.go index 6367967f8..91046cd03 100644 --- a/server/etcdserver/cindex/cindex.go +++ b/server/etcdserver/cindex/cindex.go @@ -82,8 +82,6 @@ func (ci *consistentIndex) ConsistentIndex() uint64 { return v } -// UnsafeConsistentIndex is similar to ConsistentIndex, -// but it shouldn't lock the transaction. func (ci *consistentIndex) UnsafeConsistentIndex() uint64 { if index := atomic.LoadUint64(&ci.consistentIndex); index > 0 { return index @@ -134,7 +132,7 @@ func (f *fakeConsistentIndex) UnsafeSave(_ backend.BatchTx) {} func (f *fakeConsistentIndex) SetBackend(_ Backend) {} func UpdateConsistentIndex(tx backend.BatchTx, index uint64, term uint64, onlyGrow bool) { - tx.LockWithoutHook() + tx.LockOutsideApply() defer tx.Unlock() schema.UnsafeUpdateConsistentIndex(tx, index, term, onlyGrow) } diff --git a/server/etcdserver/server.go b/server/etcdserver/server.go index 612454227..015bcaf6f 100644 --- a/server/etcdserver/server.go +++ b/server/etcdserver/server.go @@ -405,7 +405,7 @@ func NewServer(cfg config.ServerConfig) (srv *EtcdServer, err error) { // Set the hook after EtcdServer finishes the initialization to avoid // the hook being called during the initialization process. - srv.be.SetTxPostLockHook(srv.getTxPostLockHook()) + srv.be.SetTxPostLockInsideApplyHook(srv.getTxPostLockHook()) // TODO: move transport initialization near the definition of remote tr := &rafthttp.Transport{ @@ -984,7 +984,7 @@ func (s *EtcdServer) applySnapshot(ep *etcdProgress, apply *apply) { } s.consistIndex.SetBackend(newbe) - newbe.SetTxPostLockHook(s.getTxPostLockHook()) + newbe.SetTxPostLockInsideApplyHook(s.getTxPostLockHook()) lg.Info("restored mvcc store", zap.Uint64("consistent-index", s.consistIndex.ConsistentIndex())) diff --git a/server/lease/lessor.go b/server/lease/lessor.go index 4af816c76..931cb3d09 100644 --- a/server/lease/lessor.go +++ b/server/lease/lessor.go @@ -797,7 +797,7 @@ func (le *lessor) findDueScheduledCheckpoints(checkpointLimit int) []*pb.LeaseCh func (le *lessor) initAndRecover() { tx := le.b.BatchTx() - tx.LockWithoutHook() + tx.LockOutsideApply() schema.UnsafeCreateLeaseBucket(tx) lpbs := schema.MustUnsafeGetAllLeases(tx) tx.Unlock() @@ -845,7 +845,7 @@ func (l *Lease) expired() bool { func (l *Lease) persistTo(b backend.Backend) { lpb := leasepb.Lease{ID: int64(l.ID), TTL: l.ttl, RemainingTTL: l.remainingTTL} tx := b.BatchTx() - tx.Lock() + tx.LockInsideApply() defer tx.Unlock() schema.MustUnsafePutLease(tx, &lpb) } diff --git a/server/storage/backend/backend.go b/server/storage/backend/backend.go index f949f282b..ebb99ee2c 100644 --- a/server/storage/backend/backend.go +++ b/server/storage/backend/backend.go @@ -68,8 +68,8 @@ type Backend interface { ForceCommit() Close() error - // SetTxPostLockHook sets a txPostLockHook. - SetTxPostLockHook(func()) + // SetTxPostLockInsideApplyHook sets a txPostLockHook. + SetTxPostLockInsideApplyHook(func()) } type Snapshot interface { @@ -233,10 +233,10 @@ func (b *backend) BatchTx() BatchTx { return b.batchTx } -func (b *backend) SetTxPostLockHook(hook func()) { +func (b *backend) SetTxPostLockInsideApplyHook(hook func()) { // It needs to lock the batchTx, because the periodic commit // may be accessing the txPostLockHook at the moment. - b.batchTx.LockWithoutHook() + b.batchTx.lock() defer b.batchTx.Unlock() b.txPostLockHook = hook } @@ -452,7 +452,7 @@ func (b *backend) defrag() error { // TODO: make this non-blocking? // lock batchTx to ensure nobody is using previous tx, and then // close previous ongoing tx. - b.batchTx.LockWithoutHook() + b.batchTx.LockOutsideApply() defer b.batchTx.Unlock() // lock database after lock tx to avoid deadlock. diff --git a/server/storage/backend/batch_tx.go b/server/storage/backend/batch_tx.go index 8628d9aaa..7eca835fd 100644 --- a/server/storage/backend/batch_tx.go +++ b/server/storage/backend/batch_tx.go @@ -65,25 +65,31 @@ type batchTx struct { pending int } +// Lock is supposed to be called only by the unit test. func (t *batchTx) Lock() { - t.LockWithoutHook() - if t.backend.txPostLockHook != nil { - t.backend.txPostLockHook() - } + ValidateCalledInsideUnittest(t.backend.lg) + t.lock() } -func (t *batchTx) LockWithoutHook() { +func (t *batchTx) lock() { t.Mutex.Lock() } func (t *batchTx) LockInsideApply() { - ValidateCalledInsideApply(t.backend.lg) - t.Lock() + t.lock() + if t.backend.txPostLockHook != nil { + // The callers of some methods (i.e., (*RaftCluster).AddMember) + // can be coming from both InsideApply and OutsideApply, but the + // callers from OutsideApply will have a nil txPostLockHook. So we + // should check the txPostLockHook before validating the callstack. + ValidateCalledInsideApply(t.backend.lg) + t.backend.txPostLockHook() + } } func (t *batchTx) LockOutsideApply() { ValidateCalledOutSideApply(t.backend.lg) - t.Lock() + t.lock() } func (t *batchTx) Unlock() { @@ -233,14 +239,14 @@ func unsafeForEach(tx *bolt.Tx, bucket Bucket, visitor func(k, v []byte) error) // Commit commits a previous tx and begins a new writable one. func (t *batchTx) Commit() { - t.LockWithoutHook() + t.lock() t.commit(false) t.Unlock() } // CommitAndStop commits the previous tx and does not create a new one. func (t *batchTx) CommitAndStop() { - t.LockWithoutHook() + t.lock() t.commit(true) t.Unlock() } @@ -310,13 +316,13 @@ func (t *batchTxBuffered) Unlock() { } func (t *batchTxBuffered) Commit() { - t.LockWithoutHook() + t.lock() t.commit(false) t.Unlock() } func (t *batchTxBuffered) CommitAndStop() { - t.LockWithoutHook() + t.lock() t.commit(true) t.Unlock() } diff --git a/server/storage/backend/hooks_test.go b/server/storage/backend/hooks_test.go index 766464484..b77efbba4 100644 --- a/server/storage/backend/hooks_test.go +++ b/server/storage/backend/hooks_test.go @@ -41,8 +41,6 @@ func TestBackendPreCommitHook(t *testing.T) { // Empty commit. tx.Commit() - write(tx, []byte("foo"), []byte("bar")) - assert.Equal(t, ">cc", getCommitsKey(t, be), "expected 2 explict commits") tx.Commit() assert.Equal(t, ">ccc", getCommitsKey(t, be), "expected 3 explict commits") diff --git a/server/storage/backend/verify.go b/server/storage/backend/verify.go index 2f3dc0221..a6a0b8675 100644 --- a/server/storage/backend/verify.go +++ b/server/storage/backend/verify.go @@ -46,6 +46,15 @@ func ValidateCalledOutSideApply(lg *zap.Logger) { } } +func ValidateCalledInsideUnittest(lg *zap.Logger) { + if !verifyLockEnabled() { + return + } + if !insideUnittest() { + lg.Fatal("Lock called outside of unit test!", zap.Stack("stacktrace")) + } +} + func verifyLockEnabled() bool { return os.Getenv(ENV_VERIFY) == ENV_VERIFY_ALL_VALUE || os.Getenv(ENV_VERIFY) == ENV_VERIFY_LOCK } @@ -54,3 +63,8 @@ func insideApply() bool { stackTraceStr := string(debug.Stack()) return strings.Contains(stackTraceStr, ".applyEntries") } + +func insideUnittest() bool { + stackTraceStr := string(debug.Stack()) + return strings.Contains(stackTraceStr, "_test.go") && !strings.Contains(stackTraceStr, "tests/") +} diff --git a/server/storage/backend/verify_test.go b/server/storage/backend/verify_test.go index 08efb3921..2345f46b5 100644 --- a/server/storage/backend/verify_test.go +++ b/server/storage/backend/verify_test.go @@ -15,7 +15,6 @@ package backend_test import ( - "fmt" "os" "testing" "time" @@ -26,40 +25,60 @@ import ( func TestLockVerify(t *testing.T) { tcs := []struct { - insideApply bool - lock func(tx backend.BatchTx) - expectPanic bool + name string + insideApply bool + lock func(tx backend.BatchTx) + txPostLockHook func() + expectPanic bool }{ { + name: "call lockInsideApply from inside apply", insideApply: true, lock: lockInsideApply, expectPanic: false, }, { + name: "call lockInsideApply from outside apply (without txPostLockHook)", insideApply: false, lock: lockInsideApply, - expectPanic: true, + expectPanic: false, }, { + name: "call lockInsideApply from outside apply (with txPostLockHook)", + insideApply: false, + lock: lockInsideApply, + txPostLockHook: func() {}, + expectPanic: true, + }, + { + name: "call lockOutsideApply from outside apply", insideApply: false, lock: lockOutsideApply, expectPanic: false, }, { + name: "call lockOutsideApply from inside apply", insideApply: true, lock: lockOutsideApply, expectPanic: true, }, + { + name: "call Lock from unit test", + insideApply: false, + lock: lockFromUT, + expectPanic: false, + }, } env := os.Getenv("ETCD_VERIFY") os.Setenv("ETCD_VERIFY", "lock") defer func() { os.Setenv("ETCD_VERIFY", env) }() - for i, tc := range tcs { - t.Run(fmt.Sprintf("%d", i), func(t *testing.T) { + for _, tc := range tcs { + t.Run(tc.name, func(t *testing.T) { be, _ := betesting.NewTmpBackend(t, time.Hour, 10000) + be.SetTxPostLockInsideApplyHook(tc.txPostLockHook) hasPaniced := handlePanic(func() { if tc.insideApply { @@ -89,3 +108,4 @@ func applyEntries(be backend.Backend, f func(tx backend.BatchTx)) { func lockInsideApply(tx backend.BatchTx) { tx.LockInsideApply() } func lockOutsideApply(tx backend.BatchTx) { tx.LockOutsideApply() } +func lockFromUT(tx backend.BatchTx) { tx.Lock() } diff --git a/server/storage/mvcc/kvstore.go b/server/storage/mvcc/kvstore.go index 9b79c090a..074f1bea6 100644 --- a/server/storage/mvcc/kvstore.go +++ b/server/storage/mvcc/kvstore.go @@ -121,7 +121,7 @@ func NewStore(lg *zap.Logger, b backend.Backend, le lease.Lessor, cfg StoreConfi } tx := s.b.BatchTx() - tx.LockWithoutHook() + tx.LockOutsideApply() tx.UnsafeCreateBucket(schema.Key) schema.UnsafeCreateMetaBucket(tx) tx.Unlock() @@ -331,7 +331,7 @@ func (s *store) restore() error { // restore index tx := s.b.BatchTx() - tx.LockWithoutHook() + tx.LockOutsideApply() finishedCompact, found := UnsafeReadFinishedCompact(tx) if found { diff --git a/server/storage/mvcc/kvstore_compaction.go b/server/storage/mvcc/kvstore_compaction.go index 941f056a9..849f73b95 100644 --- a/server/storage/mvcc/kvstore_compaction.go +++ b/server/storage/mvcc/kvstore_compaction.go @@ -42,7 +42,7 @@ func (s *store) scheduleCompaction(compactMainRev int64, keep map[revision]struc start := time.Now() tx := s.b.BatchTx() - tx.LockWithoutHook() + tx.LockOutsideApply() keys, _ := tx.UnsafeRange(schema.Key, last, end, int64(batchNum)) for _, key := range keys { rev = bytesToRev(key) diff --git a/server/storage/mvcc/kvstore_test.go b/server/storage/mvcc/kvstore_test.go index 4a51e9c40..2779f10b7 100644 --- a/server/storage/mvcc/kvstore_test.go +++ b/server/storage/mvcc/kvstore_test.go @@ -881,7 +881,8 @@ type fakeBatchTx struct { rangeRespc chan rangeResp } -func (b *fakeBatchTx) LockWithoutHook() {} +func (b *fakeBatchTx) LockInsideApply() {} +func (b *fakeBatchTx) LockOutsideApply() {} func (b *fakeBatchTx) Lock() {} func (b *fakeBatchTx) Unlock() {} func (b *fakeBatchTx) RLock() {} @@ -905,10 +906,8 @@ func (b *fakeBatchTx) UnsafeDelete(bucket backend.Bucket, key []byte) { func (b *fakeBatchTx) UnsafeForEach(bucket backend.Bucket, visitor func(k, v []byte) error) error { return nil } -func (b *fakeBatchTx) Commit() {} -func (b *fakeBatchTx) CommitAndStop() {} -func (b *fakeBatchTx) LockInsideApply() {} -func (b *fakeBatchTx) LockOutsideApply() {} +func (b *fakeBatchTx) Commit() {} +func (b *fakeBatchTx) CommitAndStop() {} type fakeBackend struct { tx *fakeBatchTx @@ -925,7 +924,7 @@ func (b *fakeBackend) Snapshot() backend.Snapshot func (b *fakeBackend) ForceCommit() {} func (b *fakeBackend) Defrag() error { return nil } func (b *fakeBackend) Close() error { return nil } -func (b *fakeBackend) SetTxPostLockHook(func()) {} +func (b *fakeBackend) SetTxPostLockInsideApplyHook(func()) {} type indexGetResp struct { rev revision diff --git a/server/storage/mvcc/kvstore_txn.go b/server/storage/mvcc/kvstore_txn.go index fb7a9ca1f..604fac78c 100644 --- a/server/storage/mvcc/kvstore_txn.go +++ b/server/storage/mvcc/kvstore_txn.go @@ -133,7 +133,7 @@ type storeTxnWrite struct { func (s *store) Write(trace *traceutil.Trace) TxnWrite { s.mu.RLock() tx := s.b.BatchTx() - tx.Lock() + tx.LockInsideApply() tw := &storeTxnWrite{ storeTxnRead: storeTxnRead{s, tx, 0, 0, trace}, tx: tx, diff --git a/server/storage/mvcc/store.go b/server/storage/mvcc/store.go index e530c82f4..a002ada71 100644 --- a/server/storage/mvcc/store.go +++ b/server/storage/mvcc/store.go @@ -36,7 +36,7 @@ func UnsafeReadScheduledCompact(tx backend.ReadTx) (scheduledComact int64, found } func SetScheduledCompact(tx backend.BatchTx, value int64) { - tx.Lock() + tx.LockInsideApply() defer tx.Unlock() UnsafeSetScheduledCompact(tx, value) } @@ -48,7 +48,7 @@ func UnsafeSetScheduledCompact(tx backend.BatchTx, value int64) { } func SetFinishedCompact(tx backend.BatchTx, value int64) { - tx.Lock() + tx.LockInsideApply() defer tx.Unlock() UnsafeSetFinishedCompact(tx, value) } diff --git a/server/storage/schema/alarm.go b/server/storage/schema/alarm.go index 09a49994d..825a8dbe0 100644 --- a/server/storage/schema/alarm.go +++ b/server/storage/schema/alarm.go @@ -34,14 +34,14 @@ func NewAlarmBackend(lg *zap.Logger, be backend.Backend) *alarmBackend { func (s *alarmBackend) CreateAlarmBucket() { tx := s.be.BatchTx() - tx.LockWithoutHook() + tx.LockOutsideApply() defer tx.Unlock() tx.UnsafeCreateBucket(Alarm) } func (s *alarmBackend) MustPutAlarm(alarm *etcdserverpb.AlarmMember) { tx := s.be.BatchTx() - tx.Lock() + tx.LockInsideApply() defer tx.Unlock() s.mustUnsafePutAlarm(tx, alarm) } @@ -57,7 +57,7 @@ func (s *alarmBackend) mustUnsafePutAlarm(tx backend.BatchTx, alarm *etcdserverp func (s *alarmBackend) MustDeleteAlarm(alarm *etcdserverpb.AlarmMember) { tx := s.be.BatchTx() - tx.Lock() + tx.LockInsideApply() defer tx.Unlock() s.mustUnsafeDeleteAlarm(tx, alarm) } diff --git a/server/storage/schema/auth.go b/server/storage/schema/auth.go index 3956ca782..aa695bb1d 100644 --- a/server/storage/schema/auth.go +++ b/server/storage/schema/auth.go @@ -49,7 +49,7 @@ func NewAuthBackend(lg *zap.Logger, be backend.Backend) *authBackend { func (abe *authBackend) CreateAuthBuckets() { tx := abe.be.BatchTx() - tx.LockWithoutHook() + tx.LockOutsideApply() defer tx.Unlock() tx.UnsafeCreateBucket(Auth) tx.UnsafeCreateBucket(AuthUsers) @@ -106,7 +106,7 @@ func (atx *authBatchTx) UnsafeReadAuthRevision() uint64 { } func (atx *authBatchTx) Lock() { - atx.tx.Lock() + atx.tx.LockInsideApply() } func (atx *authBatchTx) Unlock() { diff --git a/server/storage/schema/cindex.go b/server/storage/schema/cindex.go index 38eea6f91..7d215bac6 100644 --- a/server/storage/schema/cindex.go +++ b/server/storage/schema/cindex.go @@ -26,7 +26,7 @@ func UnsafeCreateMetaBucket(tx backend.BatchTx) { // CreateMetaBucket creates the `meta` bucket (if it does not exists yet). func CreateMetaBucket(tx backend.BatchTx) { - tx.LockWithoutHook() + tx.LockOutsideApply() defer tx.Unlock() tx.UnsafeCreateBucket(Meta) } diff --git a/server/storage/schema/membership.go b/server/storage/schema/membership.go index 153699e69..a42353c41 100644 --- a/server/storage/schema/membership.go +++ b/server/storage/schema/membership.go @@ -52,7 +52,7 @@ func (s *membershipBackend) MustSaveMemberToBackend(m *membership.Member) { } tx := s.be.BatchTx() - tx.Lock() + tx.LockInsideApply() defer tx.Unlock() tx.UnsafePut(Members, mkey, mvalue) } @@ -61,7 +61,7 @@ func (s *membershipBackend) MustSaveMemberToBackend(m *membership.Member) { // from the v3 backend. func (s *membershipBackend) TrimClusterFromBackend() error { tx := s.be.BatchTx() - tx.LockWithoutHook() + tx.LockOutsideApply() defer tx.Unlock() tx.UnsafeDeleteBucket(Cluster) return nil @@ -71,7 +71,7 @@ func (s *membershipBackend) MustDeleteMemberFromBackend(id types.ID) { mkey := BackendMemberKey(id) tx := s.be.BatchTx() - tx.Lock() + tx.LockInsideApply() defer tx.Unlock() tx.UnsafeDelete(Members, mkey) tx.UnsafePut(MembersRemoved, mkey, []byte("removed")) @@ -121,7 +121,7 @@ func (s *membershipBackend) readMembersFromBackend() (map[types.ID]*membership.M func (s *membershipBackend) TrimMembershipFromBackend() error { s.lg.Info("Trimming membership information from the backend...") tx := s.be.BatchTx() - tx.LockWithoutHook() + tx.LockOutsideApply() defer tx.Unlock() err := tx.UnsafeForEach(Members, func(k, v []byte) error { tx.UnsafeDelete(Members, k) @@ -146,7 +146,7 @@ func (s *membershipBackend) MustSaveClusterVersionToBackend(ver *semver.Version) ckey := ClusterClusterVersionKeyName tx := s.be.BatchTx() - tx.Lock() + tx.LockInsideApply() defer tx.Unlock() tx.UnsafePut(Cluster, ckey, []byte(ver.String())) } @@ -160,14 +160,14 @@ func (s *membershipBackend) MustSaveDowngradeToBackend(downgrade *version.Downgr s.lg.Panic("failed to marshal downgrade information", zap.Error(err)) } tx := s.be.BatchTx() - tx.Lock() + tx.LockInsideApply() defer tx.Unlock() tx.UnsafePut(Cluster, dkey, dvalue) } func (s *membershipBackend) MustCreateBackendBuckets() { tx := s.be.BatchTx() - tx.LockWithoutHook() + tx.LockOutsideApply() defer tx.Unlock() tx.UnsafeCreateBucket(Members) tx.UnsafeCreateBucket(MembersRemoved) diff --git a/server/storage/schema/migration.go b/server/storage/schema/migration.go index e1e44dab5..61ea51bf2 100644 --- a/server/storage/schema/migration.go +++ b/server/storage/schema/migration.go @@ -49,7 +49,7 @@ func newPlan(lg *zap.Logger, current semver.Version, target semver.Version) (pla } func (p migrationPlan) Execute(lg *zap.Logger, tx backend.BatchTx) error { - tx.LockWithoutHook() + tx.LockOutsideApply() defer tx.Unlock() return p.unsafeExecute(lg, tx) } @@ -90,7 +90,7 @@ func newMigrationStep(v semver.Version, isUpgrade bool, changes []schemaChange) // execute runs actions required to migrate etcd storage between two minor versions. func (s migrationStep) execute(lg *zap.Logger, tx backend.BatchTx) error { - tx.LockWithoutHook() + tx.LockOutsideApply() defer tx.Unlock() return s.unsafeExecute(lg, tx) } diff --git a/server/storage/schema/schema.go b/server/storage/schema/schema.go index 2b4c15f29..68bb212d7 100644 --- a/server/storage/schema/schema.go +++ b/server/storage/schema/schema.go @@ -30,13 +30,13 @@ var ( ) // Validate checks provided backend to confirm that schema used is supported. -func Validate(lg *zap.Logger, tx backend.BatchTx) error { - tx.LockWithoutHook() +func Validate(lg *zap.Logger, tx backend.ReadTx) error { + tx.Lock() defer tx.Unlock() return unsafeValidate(lg, tx) } -func unsafeValidate(lg *zap.Logger, tx backend.BatchTx) error { +func unsafeValidate(lg *zap.Logger, tx backend.ReadTx) error { current, err := UnsafeDetectSchemaVersion(lg, tx) if err != nil { // v3.5 requires a wal snapshot to persist its fields, so we can assign it a schema version. @@ -60,7 +60,7 @@ type WALVersion interface { // Migrate updates storage schema to provided target version. // Downgrading requires that provided WAL doesn't contain unsupported entries. func Migrate(lg *zap.Logger, tx backend.BatchTx, w WALVersion, target semver.Version) error { - tx.LockWithoutHook() + tx.LockOutsideApply() defer tx.Unlock() return UnsafeMigrate(lg, tx, w, target) } diff --git a/server/storage/schema/schema_test.go b/server/storage/schema/schema_test.go index f3c0c4a7f..8dbd337b2 100644 --- a/server/storage/schema/schema_test.go +++ b/server/storage/schema/schema_test.go @@ -88,7 +88,7 @@ func TestValidate(t *testing.T) { b := backend.NewDefaultBackend(lg, dataPath) defer b.Close() - err := Validate(lg, b.BatchTx()) + err := Validate(lg, b.ReadTx()) if (err != nil) != tc.expectError { t.Errorf("Validate(lg, tx) = %+v, expected error: %v", err, tc.expectError) }