MM-19548: Add a deadlock retry function for SaveChannel (#14857)

* MM-19548: Add a deadlock retry function for SaveChannel

A deadlock has been seen to occur in the upsertPublicChannelT method
during bulk import.

Here is a brief excerpt:

*** (1) TRANSACTION:
TRANSACTION 3141, ACTIVE 1 sec inserting
INSERT INTO
                            PublicChannels(Id, DeleteAt, TeamId, DisplayName, Name, Header, Purpose)
                        VALUES
                            (?, ?, ?, ?, ?, ?, ?)
                        ON DUPLICATE KEY UPDATE
                            DeleteAt = ?,
                            TeamId = ?,
                            DisplayName = ?,
                            Name = ?,
                            Header = ?,
                            Purpose = ?
*** (1) WAITING FOR THIS LOCK TO BE GRANTED:
RECORD LOCKS space id 76 page no 4 n bits 104 index Name of table `mydb`.`PublicChannels` trx id 3141 lock_mode X locks gap before rec insert intention waiting
** (2) TRANSACTION:
TRANSACTION 3140, ACTIVE 1 sec inserting
mysql tables in use 1, locked 1
5 lock struct(s), heap size 1136, 3 row lock(s), undo log entries 2
MySQL thread id 50, OS thread handle 140641523848960, query id 3226 172.17.0.1 mmuser update
INSERT INTO
                            PublicChannels(Id, DeleteAt, TeamId, DisplayName, Name, Header, Purpose)
                        VALUES
                            (?, ?, ?, ?, ?, ?, ?)
                        ON DUPLICATE KEY UPDATE
                            DeleteAt = ?,
                            TeamId = ?,
                            DisplayName = ?,
                            Name = ?,
                            Header = ?,
                            Purpose = ?
*** (2) HOLDS THE LOCK(S):
RECORD LOCKS space id 76 page no 4 n bits 104 index Name of table `mydb`.`PublicChannels` trx id 3140 lock_mode X locks gap before rec

*** (2) WAITING FOR THIS LOCK TO BE GRANTED:
RECORD LOCKS space id 76 page no 4 n bits 104 index Name of table `mydb`.`PublicChannels` trx id 3140 lock_mode X locks gap before rec insert intention waiting

*** WE ROLL BACK TRANSACTION (1)

Following is my analysis:

From the deadlock output, it can be seen that it's due to a gap lock.
And that's clear because the index is Name which is a multi-column index using Name and TeamId.
But interestingly, both transactions seem to be inserting the same data, which is what is puzzling me.
The multi-column index on Name and TeamId will guarantee that they are always unique. And from looking at the code,
it does not seem possible to me that it will try to insert the same data from 2 different transactions.
But even if they do, why does tx 2 try to acquire the same lock again when it already has that ?

Here is what I think the order of events happening

Tx 2 gets a gap lock.
Tx 1 tries to get the same gap lock.
Tx 2 tries to again get the same gap lock ?

The last step is what is puzzling me. Why does an UPSERT statement acquire 2 gap locks ? From my reading of https://dev.mysql.com/doc/refman/8.0/en/innodb-locks-set.html:

> INSERT ... ON DUPLICATE KEY UPDATE differs from a simple INSERT in that an exclusive lock rather than a shared lock is placed on the row to be updated when a duplicate-key error occurs. An exclusive index-record lock is taken for a duplicate primary key value. An exclusive next-key lock is taken for a duplicate unique key value.

From what I understand, the expectation is that there will be one X lock and one gap lock is taken.
But that's not what the deadlock output seems to say.

The general advice on the internet seems to be that deadlocks will happen and not all of them can be understood.

For now, we add a generic deadlock retry function at the store package which can be reused by other queries too.

P.S.: This is a verbatim copy of my investigation posted at https://dba.stackexchange.com/questions/268652/mysql-deadlock-upsert-query-acquiring-gap-lock-twice

Testing:

This is ofcourse hard to test because it is impossible to reproduce this. I have tested this by manually returning an error
and confirming that it indeed retries.

WARN[2020-06-19T11:18:24.9585676+05:30] A deadlock happened. Retrying.                caller="sqlstore/channel_store.go:568" error="Error 1213: mydeadlock"
WARN[2020-06-19T11:18:24.959158+05:30] A deadlock happened. Retrying.                caller="sqlstore/channel_store.go:568" error="Error 1213: mydeadlock"
WARN[2020-06-19T11:18:24.9595072+05:30] A deadlock happened. Retrying.                caller="sqlstore/channel_store.go:568" error="Error 1213: mydeadlock"
WARN[2020-06-19T11:18:24.9595451+05:30] Deadlock happened 3 times. Giving up          caller="sqlstore/channel_store.go:579"
ERRO[2020-06-19T11:18:24.9596426+05:30] Unable to save channel.                       caller="mlog/log.go:175" err_details="Error 1213: mydeadlock" err_where=CreateChannel http_code=500 ip_addr="::1" method=POST path=/api/v4/channels request_id=745bsj13b7f6mnmsbn3t97grbw user_id=xcof1ipipbrfxpfjf6x4p6kx9e

* Fix tests

* Address review comments

* Address review comments

* Add forgotten test

Co-authored-by: Mattermod <mattermod@users.noreply.github.com>

app/channel.go

@@ -229,7 +229,7 @@ func (a *App) CreateChannel(channel *model.Channel, addMember bool) (*model.Chan
 				return nil, model.NewAppError("CreateChannel", "store.sql_channel.save_channel.existing.app_error", nil, "id="+invErr.Value.(string), http.StatusBadRequest)
 			}
 		case errors.As(nErr, &cErr):
-			return channel, model.NewAppError("CreateChannel", store.CHANNEL_EXISTS_ERROR, nil, cErr.Error(), http.StatusBadRequest)
+			return sc, model.NewAppError("CreateChannel", store.CHANNEL_EXISTS_ERROR, nil, cErr.Error(), http.StatusBadRequest)
 		case errors.As(nErr, &ltErr):
 			return nil, model.NewAppError("CreateChannel", "store.sql_channel.save_channel.limit.app_error", nil, ltErr.Error(), http.StatusBadRequest)
 		case errors.As(nErr, &appErr): // in case we haven't converted to plain error.

store/sqlstore/channel_store.go

@@ -518,27 +518,32 @@ func (s SqlChannelStore) Save(channel *model.Channel, maxChannelsPerTeam int64)
 		return nil, store.NewErrInvalidInput("Channel", "Type", channel.Type)
 	}
 
-	transaction, err := s.GetMaster().Begin()
-	if err != nil {
-		return nil, errors.Wrapf(err, "begin_transaction: ")
-	}
-	defer finalizeTransaction(transaction)
+	var newChannel *model.Channel
+	err := store.WithDeadlockRetry(func() error {
+		transaction, err := s.GetMaster().Begin()
+		if err != nil {
+			return errors.Wrap(err, "begin_transaction")
+		}
+		defer finalizeTransaction(transaction)
 
-	newChannel, err := s.saveChannelT(transaction, channel, maxChannelsPerTeam)
-	if err != nil {
-		return newChannel, err
-	}
+		newChannel, err = s.saveChannelT(transaction, channel, maxChannelsPerTeam)
+		if err != nil {
+			return err
+		}
 
-	// Additionally propagate the write to the PublicChannels table.
-	if err := s.upsertPublicChannelT(transaction, newChannel); err != nil {
-		return nil, errors.Wrapf(err, "upsert_public_channel: ")
-	}
+		// Additionally propagate the write to the PublicChannels table.
+		if err := s.upsertPublicChannelT(transaction, newChannel); err != nil {
+			return errors.Wrap(err, "upsert_public_channel")
+		}
 
-	if err := transaction.Commit(); err != nil {
-		return nil, errors.Wrapf(err, "commit_transaction: ")
-	}
-
-	return newChannel, nil
+		if err := transaction.Commit(); err != nil {
+			return errors.Wrap(err, "commit_transaction")
+		}
+		return nil
+	})
+	// There are cases when in case of conflict, the original channel value is returned.
+	// So we return both and let the caller do the checks.
+	return newChannel, err
 }
 
 func (s SqlChannelStore) CreateDirectChannel(user *model.User, otherUser *model.User) (*model.Channel, error) {

store/store.go

@@ -9,7 +9,11 @@ import (
 	"context"
 	"time"
 
+	"github.com/mattermost/mattermost-server/v5/mlog"
 	"github.com/mattermost/mattermost-server/v5/model"
+
+	"github.com/go-sql-driver/mysql"
+	"github.com/pkg/errors"
 )
 
 type StoreResult struct {
@@ -770,3 +774,31 @@ type IntegrityCheckResult struct {
 	Data interface{}
 	Err  error
 }
+
+const mySQLDeadlockCode = uint16(1213)
+
+// WithDeadlockRetry retries a given f if it throws a deadlock error.
+// It breaks after a threshold and propagates the error upwards.
+// TODO: This can be a separate retry layer in itself where transaction retries
+// are automatically applied.
+func WithDeadlockRetry(f func() error) error {
+	var err error
+	for i := 0; i < 3; i++ {
+		err = f()
+		if err == nil {
+			// No error, return nil.
+			return nil
+		}
+		// XXX: Possibly add check for postgres deadlocks later.
+		// But deadlocks are very rarely seen in postgres.
+		var mysqlErr *mysql.MySQLError
+		if errors.As(err, &mysqlErr) && mysqlErr.Number == mySQLDeadlockCode {
+			mlog.Warn("A deadlock happened. Retrying.", mlog.Err(err))
+			// This is a deadlock, retry.
+			continue
+		}
+		// Some other error, return as-is.
+		return err
+	}
+	return errors.Wrap(err, "giving up after 3 consecutive deadlocks")
+}

store/store_test.go

@@ -0,0 +1,56 @@
+// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
+// See LICENSE.txt for license information.
+
+package store
+
+import (
+	"testing"
+
+	"github.com/go-sql-driver/mysql"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+type deadlock struct {
+	numRetries int
+	hasRetried int
+}
+
+func newDeadlock(numRetries int) *deadlock {
+	return &deadlock{
+		numRetries: numRetries,
+	}
+}
+
+func (d *deadlock) f() error {
+	if d.numRetries == d.hasRetried {
+		return nil
+	}
+	d.hasRetried++
+	return &mysql.MySQLError{
+		Number: mySQLDeadlockCode,
+	}
+}
+
+func TestDeadlockRetry(t *testing.T) {
+	t.Run("NoDeadlock", func(t *testing.T) {
+		d := newDeadlock(0)
+		err := WithDeadlockRetry(d.f)
+		require.NoError(t, err)
+		assert.Equal(t, 0, d.hasRetried)
+	})
+
+	t.Run("1Deadlock", func(t *testing.T) {
+		d := newDeadlock(1)
+		err := WithDeadlockRetry(d.f)
+		require.NoError(t, err)
+		assert.Equal(t, 1, d.hasRetried)
+	})
+
+	t.Run("AlwaysDeadlock", func(t *testing.T) {
+		d := newDeadlock(4)
+		err := WithDeadlockRetry(d.f)
+		require.Error(t, err)
+		assert.Equal(t, 3, d.hasRetried)
+	})
+}