Merge pull request #16325 from serathius/reader-writer

Separate Writer interface from BatchTx interfaces

etcdutl/snapshot/v3_snapshot.go

@@ -353,7 +353,7 @@ func (s *v3Manager) modifyLatestRevision(bumpAmount uint64) error {
 	return nil
 }
 
-func (s *v3Manager) unsafeBumpRevision(tx backend.BatchTx, latest revision, amount int64) revision {
+func (s *v3Manager) unsafeBumpRevision(tx backend.UnsafeWriter, latest revision, amount int64) revision {
 	s.lg.Info(
 		"bumping latest revision",
 		zap.Int64("latest-revision", latest.main),
@@ -370,7 +370,7 @@ func (s *v3Manager) unsafeBumpRevision(tx backend.BatchTx, latest revision, amou
 	return latest
 }
 
-func (s *v3Manager) unsafeMarkRevisionCompacted(tx backend.BatchTx, latest revision) {
+func (s *v3Manager) unsafeMarkRevisionCompacted(tx backend.UnsafeWriter, latest revision) {
 	s.lg.Info(
 		"marking revision compacted",
 		zap.Int64("revision", latest.main),
@@ -379,7 +379,7 @@ func (s *v3Manager) unsafeMarkRevisionCompacted(tx backend.BatchTx, latest revis
 	mvcc.UnsafeSetScheduledCompact(tx, latest.main)
 }
 
-func (s *v3Manager) unsafeGetLatestRevision(tx backend.BatchTx) (revision, error) {
+func (s *v3Manager) unsafeGetLatestRevision(tx backend.UnsafeReader) (revision, error) {
 	var latest revision
 	err := tx.UnsafeForEach(schema.Key, func(k, _ []byte) (err error) {
 		rev := bytesToRev(k)

server/auth/store.go

@@ -205,18 +205,6 @@ type AuthBackend interface {
 	GetAllRoles() []*authpb.Role
 }
 
-type AuthBatchTx interface {
-	Lock()
-	Unlock()
-	UnsafeAuthReader
-	UnsafeSaveAuthEnabled(enabled bool)
-	UnsafeSaveAuthRevision(rev uint64)
-	UnsafePutUser(*authpb.User)
-	UnsafeDeleteUser(string)
-	UnsafePutRole(*authpb.Role)
-	UnsafeDeleteRole(string)
-}
-
 type AuthReadTx interface {
 	RLock()
 	RUnlock()
@@ -232,6 +220,26 @@ type UnsafeAuthReader interface {
 	UnsafeGetAllRoles() []*authpb.Role
 }
 
+type AuthBatchTx interface {
+	Lock()
+	Unlock()
+	UnsafeAuthReadWriter
+}
+
+type UnsafeAuthReadWriter interface {
+	UnsafeAuthReader
+	UnsafeAuthWriter
+}
+
+type UnsafeAuthWriter interface {
+	UnsafeSaveAuthEnabled(enabled bool)
+	UnsafeSaveAuthRevision(rev uint64)
+	UnsafePutUser(*authpb.User)
+	UnsafeDeleteUser(string)
+	UnsafePutRole(*authpb.Role)
+	UnsafeDeleteRole(string)
+}
+
 type authStore struct {
 	// atomic operations; need 64-bit align, or 32-bit tests will crash
 	revision uint64
@@ -990,7 +998,7 @@ func hasRootRole(u *authpb.User) bool {
 	return idx != len(u.Roles) && u.Roles[idx] == rootRole
 }
 
-func (as *authStore) commitRevision(tx AuthBatchTx) {
+func (as *authStore) commitRevision(tx UnsafeAuthWriter) {
 	atomic.AddUint64(&as.revision, 1)
 	tx.UnsafeSaveAuthRevision(as.Revision())
 }

server/etcdserver/cindex/cindex.go

@@ -46,7 +46,7 @@ type ConsistentIndexer interface {
 
 	// UnsafeSave must be called holding the lock on the tx.
 	// It saves consistentIndex to the underlying stable storage.
-	UnsafeSave(tx backend.BatchTx)
+	UnsafeSave(tx backend.UnsafeReadWriter)
 
 	// SetBackend set the available backend.BatchTx for ConsistentIndexer.
 	SetBackend(be Backend)
@@ -115,7 +115,7 @@ func (ci *consistentIndex) SetConsistentIndex(v uint64, term uint64) {
 	atomic.StoreUint64(&ci.term, term)
 }
 
-func (ci *consistentIndex) UnsafeSave(tx backend.BatchTx) {
+func (ci *consistentIndex) UnsafeSave(tx backend.UnsafeReadWriter) {
 	index := atomic.LoadUint64(&ci.consistentIndex)
 	term := atomic.LoadUint64(&ci.term)
 	schema.UnsafeUpdateConsistentIndex(tx, index, term)
@@ -166,8 +166,8 @@ func (f *fakeConsistentIndex) SetConsistentApplyingIndex(index uint64, term uint
 	atomic.StoreUint64(&f.term, term)
 }
 
-func (f *fakeConsistentIndex) UnsafeSave(_ backend.BatchTx) {}
-func (f *fakeConsistentIndex) SetBackend(_ Backend)         {}
+func (f *fakeConsistentIndex) UnsafeSave(_ backend.UnsafeReadWriter) {}
+func (f *fakeConsistentIndex) SetBackend(_ Backend)                  {}
 
 func UpdateConsistentIndexForce(tx backend.BatchTx, index uint64, term uint64) {
 	tx.LockOutsideApply()

server/storage/backend/batch_tx.go

@@ -46,18 +46,26 @@ type Bucket interface {
 type BatchTx interface {
 	Lock()
 	Unlock()
-	UnsafeReader
-	UnsafeCreateBucket(bucket Bucket)
-	UnsafeDeleteBucket(bucket Bucket)
-	UnsafePut(bucket Bucket, key []byte, value []byte)
-	UnsafeSeqPut(bucket Bucket, key []byte, value []byte)
-	UnsafeDelete(bucket Bucket, key []byte)
 	// Commit commits a previous tx and begins a new writable one.
 	Commit()
 	// CommitAndStop commits the previous tx and does not create a new one.
 	CommitAndStop()
 	LockInsideApply()
 	LockOutsideApply()
+	UnsafeReadWriter
+}
+
+type UnsafeReadWriter interface {
+	UnsafeReader
+	UnsafeWriter
+}
+
+type UnsafeWriter interface {
+	UnsafeCreateBucket(bucket Bucket)
+	UnsafeDeleteBucket(bucket Bucket)
+	UnsafePut(bucket Bucket, key []byte, value []byte)
+	UnsafeSeqPut(bucket Bucket, key []byte, value []byte)
+	UnsafeDelete(bucket Bucket, key []byte)
 }
 
 type batchTx struct {

server/storage/backend/hooks.go

@@ -14,20 +14,20 @@
 
 package backend
 
-type HookFunc func(tx BatchTx)
+type HookFunc func(tx UnsafeReadWriter)
 
 // Hooks allow to add additional logic executed during transaction lifetime.
 type Hooks interface {
 	// OnPreCommitUnsafe is executed before Commit of transactions.
 	// The given transaction is already locked.
-	OnPreCommitUnsafe(tx BatchTx)
+	OnPreCommitUnsafe(tx UnsafeReadWriter)
 }
 
 type hooks struct {
 	onPreCommitUnsafe HookFunc
 }
 
-func (h hooks) OnPreCommitUnsafe(tx BatchTx) {
+func (h hooks) OnPreCommitUnsafe(tx UnsafeReadWriter) {
 	h.onPreCommitUnsafe(tx)
 }
 

server/storage/backend/hooks_test.go

@@ -113,7 +113,7 @@ func prepareBuckenAndKey(tx backend.BatchTx) {
 
 func newTestHooksBackend(t testing.TB, baseConfig backend.BackendConfig) backend.Backend {
 	cfg := baseConfig
-	cfg.Hooks = backend.NewHooks(func(tx backend.BatchTx) {
+	cfg.Hooks = backend.NewHooks(func(tx backend.UnsafeReadWriter) {
 		k, v := tx.UnsafeRange(bucket, key, nil, 1)
 		t.Logf("OnPreCommit executed: %v %v", string(k[0]), string(v[0]))
 		assert.Len(t, k, 1)

server/storage/hooks.go

@@ -41,7 +41,7 @@ func NewBackendHooks(lg *zap.Logger, indexer cindex.ConsistentIndexer) *BackendH
 	return &BackendHooks{lg: lg, indexer: indexer}
 }
 
-func (bh *BackendHooks) OnPreCommitUnsafe(tx backend.BatchTx) {
+func (bh *BackendHooks) OnPreCommitUnsafe(tx backend.UnsafeReadWriter) {
 	bh.indexer.UnsafeSave(tx)
 	bh.confStateLock.Lock()
 	defer bh.confStateLock.Unlock()

server/storage/mvcc/store.go

@@ -41,7 +41,7 @@ func SetScheduledCompact(tx backend.BatchTx, value int64) {
 	UnsafeSetScheduledCompact(tx, value)
 }
 
-func UnsafeSetScheduledCompact(tx backend.BatchTx, value int64) {
+func UnsafeSetScheduledCompact(tx backend.UnsafeWriter, value int64) {
 	rbytes := newRevBytes()
 	revToBytes(revision{main: value}, rbytes)
 	tx.UnsafePut(schema.Meta, schema.ScheduledCompactKeyName, rbytes)
@@ -53,7 +53,7 @@ func SetFinishedCompact(tx backend.BatchTx, value int64) {
 	UnsafeSetFinishedCompact(tx, value)
 }
 
-func UnsafeSetFinishedCompact(tx backend.BatchTx, value int64) {
+func UnsafeSetFinishedCompact(tx backend.UnsafeWriter, value int64) {
 	rbytes := newRevBytes()
 	revToBytes(revision{main: value}, rbytes)
 	tx.UnsafePut(schema.Meta, schema.FinishedCompactKeyName, rbytes)

server/storage/schema/actions.go

@@ -23,7 +23,7 @@ import (
 type action interface {
 	// unsafeDo executes the action and returns revert action, when executed
 	// should restore the state from before.
-	unsafeDo(tx backend.BatchTx) (revert action, err error)
+	unsafeDo(tx backend.UnsafeReadWriter) (revert action, err error)
 }
 
 type setKeyAction struct {
@@ -32,7 +32,7 @@ type setKeyAction struct {
 	FieldValue []byte
 }
 
-func (a setKeyAction) unsafeDo(tx backend.BatchTx) (action, error) {
+func (a setKeyAction) unsafeDo(tx backend.UnsafeReadWriter) (action, error) {
 	revert := restoreFieldValueAction(tx, a.Bucket, a.FieldName)
 	tx.UnsafePut(a.Bucket, a.FieldName, a.FieldValue)
 	return revert, nil
@@ -43,13 +43,13 @@ type deleteKeyAction struct {
 	FieldName []byte
 }
 
-func (a deleteKeyAction) unsafeDo(tx backend.BatchTx) (action, error) {
+func (a deleteKeyAction) unsafeDo(tx backend.UnsafeReadWriter) (action, error) {
 	revert := restoreFieldValueAction(tx, a.Bucket, a.FieldName)
 	tx.UnsafeDelete(a.Bucket, a.FieldName)
 	return revert, nil
 }
 
-func restoreFieldValueAction(tx backend.BatchTx, bucket backend.Bucket, fieldName []byte) action {
+func restoreFieldValueAction(tx backend.UnsafeReader, bucket backend.Bucket, fieldName []byte) action {
 	_, vs := tx.UnsafeRange(bucket, fieldName, nil, 1)
 	if len(vs) == 1 {
 		return &setKeyAction{
@@ -68,7 +68,7 @@ type ActionList []action
 
 // unsafeExecute executes actions one by one. If one of actions returns error,
 // it will revert them.
-func (as ActionList) unsafeExecute(lg *zap.Logger, tx backend.BatchTx) error {
+func (as ActionList) unsafeExecute(lg *zap.Logger, tx backend.UnsafeReadWriter) error {
 	var revertActions = make(ActionList, 0, len(as))
 	for _, a := range as {
 		revert, err := a.unsafeDo(tx)
@@ -84,7 +84,7 @@ func (as ActionList) unsafeExecute(lg *zap.Logger, tx backend.BatchTx) error {
 
 // unsafeExecuteInReversedOrder executes actions in revered order. Will panic on
 // action error. Should be used when reverting.
-func (as ActionList) unsafeExecuteInReversedOrder(lg *zap.Logger, tx backend.BatchTx) {
+func (as ActionList) unsafeExecuteInReversedOrder(lg *zap.Logger, tx backend.UnsafeReadWriter) {
 	for j := len(as) - 1; j >= 0; j-- {
 		_, err := as[j].unsafeDo(tx)
 		if err != nil {

server/storage/schema/actions_test.go

@@ -147,17 +147,17 @@ type brokenAction struct{}
 
 var errBrokenAction = fmt.Errorf("broken action error")
 
-func (c brokenAction) unsafeDo(tx backend.BatchTx) (action, error) {
+func (c brokenAction) unsafeDo(tx backend.UnsafeReadWriter) (action, error) {
 	return nil, errBrokenAction
 }
 
-func putKeyValues(tx backend.BatchTx, bucket backend.Bucket, kvs map[string]string) {
+func putKeyValues(tx backend.UnsafeWriter, bucket backend.Bucket, kvs map[string]string) {
 	for k, v := range kvs {
 		tx.UnsafePut(bucket, []byte(k), []byte(v))
 	}
 }
 
-func assertBucketState(t *testing.T, tx backend.BatchTx, bucket backend.Bucket, expect map[string]string) {
+func assertBucketState(t *testing.T, tx backend.UnsafeReadWriter, bucket backend.Bucket, expect map[string]string) {
 	t.Helper()
 	got := map[string]string{}
 	ks, vs := tx.UnsafeRange(bucket, []byte("\x00"), []byte("\xff"), 0)

server/storage/schema/alarm.go

@@ -47,7 +47,7 @@ func (s *alarmBackend) MustPutAlarm(alarm *etcdserverpb.AlarmMember) {
 	s.mustUnsafePutAlarm(tx, alarm)
 }
 
-func (s *alarmBackend) mustUnsafePutAlarm(tx backend.BatchTx, alarm *etcdserverpb.AlarmMember) {
+func (s *alarmBackend) mustUnsafePutAlarm(tx backend.UnsafeWriter, alarm *etcdserverpb.AlarmMember) {
 	v, err := alarm.Marshal()
 	if err != nil {
 		s.lg.Panic("failed to marshal alarm member", zap.Error(err))
@@ -63,7 +63,7 @@ func (s *alarmBackend) MustDeleteAlarm(alarm *etcdserverpb.AlarmMember) {
 	s.mustUnsafeDeleteAlarm(tx, alarm)
 }
 
-func (s *alarmBackend) mustUnsafeDeleteAlarm(tx backend.BatchTx, alarm *etcdserverpb.AlarmMember) {
+func (s *alarmBackend) mustUnsafeDeleteAlarm(tx backend.UnsafeWriter, alarm *etcdserverpb.AlarmMember) {
 	v, err := alarm.Marshal()
 	if err != nil {
 		s.lg.Panic("failed to marshal alarm member", zap.Error(err))

server/storage/schema/auth_roles.go

@@ -21,7 +21,7 @@ import (
 	"go.etcd.io/etcd/server/v3/storage/backend"
 )
 
-func UnsafeCreateAuthRolesBucket(tx backend.BatchTx) {
+func UnsafeCreateAuthRolesBucket(tx backend.UnsafeWriter) {
 	tx.UnsafeCreateBucket(AuthRoles)
 }
 

server/storage/schema/auth_roles_test.go

@@ -30,17 +30,17 @@ import (
 func TestGetAllRoles(t *testing.T) {
 	tcs := []struct {
 		name  string
-		setup func(tx auth.AuthBatchTx)
+		setup func(tx auth.UnsafeAuthWriter)
 		want  []*authpb.Role
 	}{
 		{
 			name:  "Empty by default",
-			setup: func(tx auth.AuthBatchTx) {},
+			setup: func(tx auth.UnsafeAuthWriter) {},
 			want:  nil,
 		},
 		{
 			name: "Returns data put before",
-			setup: func(tx auth.AuthBatchTx) {
+			setup: func(tx auth.UnsafeAuthWriter) {
 				tx.UnsafePutRole(&authpb.Role{
 					Name: []byte("readKey"),
 					KeyPermission: []*authpb.Permission{
@@ -67,7 +67,7 @@ func TestGetAllRoles(t *testing.T) {
 		},
 		{
 			name: "Skips deleted",
-			setup: func(tx auth.AuthBatchTx) {
+			setup: func(tx auth.UnsafeAuthWriter) {
 				tx.UnsafePutRole(&authpb.Role{
 					Name: []byte("role1"),
 				})
@@ -80,7 +80,7 @@ func TestGetAllRoles(t *testing.T) {
 		},
 		{
 			name: "Returns data overriden by put",
-			setup: func(tx auth.AuthBatchTx) {
+			setup: func(tx auth.UnsafeAuthWriter) {
 				tx.UnsafePutRole(&authpb.Role{
 					Name: []byte("role1"),
 					KeyPermission: []*authpb.Permission{
@@ -135,17 +135,17 @@ func TestGetAllRoles(t *testing.T) {
 func TestGetRole(t *testing.T) {
 	tcs := []struct {
 		name  string
-		setup func(tx auth.AuthBatchTx)
+		setup func(tx auth.UnsafeAuthWriter)
 		want  *authpb.Role
 	}{
 		{
 			name:  "Returns nil for missing",
-			setup: func(tx auth.AuthBatchTx) {},
+			setup: func(tx auth.UnsafeAuthWriter) {},
 			want:  nil,
 		},
 		{
 			name: "Returns data put before",
-			setup: func(tx auth.AuthBatchTx) {
+			setup: func(tx auth.UnsafeAuthWriter) {
 				tx.UnsafePutRole(&authpb.Role{
 					Name: []byte("role1"),
 					KeyPermission: []*authpb.Permission{
@@ -170,7 +170,7 @@ func TestGetRole(t *testing.T) {
 		},
 		{
 			name: "Return nil for deleted",
-			setup: func(tx auth.AuthBatchTx) {
+			setup: func(tx auth.UnsafeAuthWriter) {
 				tx.UnsafePutRole(&authpb.Role{
 					Name: []byte("role1"),
 				})
@@ -180,7 +180,7 @@ func TestGetRole(t *testing.T) {
 		},
 		{
 			name: "Returns data overriden by put",
-			setup: func(tx auth.AuthBatchTx) {
+			setup: func(tx auth.UnsafeAuthWriter) {
 				tx.UnsafePutRole(&authpb.Role{
 					Name: []byte("role1"),
 					KeyPermission: []*authpb.Permission{