dolt/integration-tests/transactions/concurrent_tx_test.go

// Copyright 2022 Dolthub, Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package transactions

import (
	"fmt"
	"sync"
	"testing"

	_ "github.com/go-sql-driver/mysql"
	"github.com/gocraft/dbr/v2"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
)

var defaultConfig = ServerConfig{
	database: "mysql",
	host:     "127.0.0.1",
	port:     3316,
	user:     "root",
	password: "toor",
}

func TestConcurrentTransactions(t *testing.T) {
	sequential := &sync.Mutex{}
	for _, test := range txTests {
		t.Run(test.name, func(t *testing.T) {
			sequential.Lock()
			defer sequential.Unlock()
			testConcurrentTx(t, test)
		})
	}
}

type ConcurrentTxTest struct {
	name    string
	queries []concurrentQuery
}

type concurrentQuery struct {
	conn      string
	query     string
	assertion selector
	expected  []testRow
}

type selector func(s *dbr.Session) *dbr.SelectStmt

type testRow struct {
	Pk, C0 int
}

const (
	one = "one"
	two = "two"
)

var txTests = []ConcurrentTxTest{
	{
		name: "smoke test",
		queries: []concurrentQuery{
			{
				conn: one,
				assertion: func(s *dbr.Session) *dbr.SelectStmt {
					return s.Select("*").From("tx.data")
				},
				expected: []testRow{
					{1, 1},
					{2, 2},
					{3, 3},
				},
			},
		},
	},
	{
		name: "concurrent transactions",
		queries: []concurrentQuery{
			{
				conn:  one,
				query: "BEGIN;",
			},
			{
				conn:  two,
				query: "BEGIN;",
			},
			{
				conn: two,
				assertion: func(s *dbr.Session) *dbr.SelectStmt {
					return s.Select("*").From("tx.data")
				},
				expected: []testRow{
					{1, 1}, {2, 2}, {3, 3},
				},
			},
			{
				conn:  one,
				query: "INSERT INTO tx.data VALUES (4,4)",
			},
			{
				conn: two,
				assertion: func(s *dbr.Session) *dbr.SelectStmt {
					return s.Select("*").From("tx.data")
				},
				expected: []testRow{
					{1, 1}, {2, 2}, {3, 3},
				},
			},
			{
				conn:  one,
				query: "COMMIT",
			},
			{
				conn: two,
				assertion: func(s *dbr.Session) *dbr.SelectStmt {
					return s.Select("*").From("tx.data")
				},
				expected: []testRow{
					{1, 1}, {2, 2}, {3, 3},
				},
			},
			{
				conn:  two,
				query: "COMMIT",
			},
			{
				conn: two,
				assertion: func(s *dbr.Session) *dbr.SelectStmt {
					return s.Select("*").From("tx.data")
				},
				expected: []testRow{
					{1, 1}, {2, 2}, {3, 3}, {4, 4},
				},
			},
		},
	},
	{
		name: "concurrent updates",
		queries: []concurrentQuery{
			{
				conn:  one,
				query: "BEGIN;",
			},
			{
				conn:  two,
				query: "BEGIN;",
			},
			{
				conn: two,
				assertion: func(s *dbr.Session) *dbr.SelectStmt {
					return s.Select("*").From("tx.data").Where("pk = 1")
				},
				expected: []testRow{
					{1, 1},
				},
			},
			{
				conn:  one,
				query: "UPDATE tx.data SET c0 = c0 + 10 WHERE pk = 1;",
			},
			{
				conn: two,
				assertion: func(s *dbr.Session) *dbr.SelectStmt {
					return s.Select("*").From("tx.data").Where("pk = 1")
				},
				expected: []testRow{
					{1, 1},
				},
			},
			{
				conn:  one,
				query: "COMMIT",
			},
			{
				conn: two,
				assertion: func(s *dbr.Session) *dbr.SelectStmt {
					return s.Select("*").From("tx.data").Where("pk = 1")
				},
				expected: []testRow{
					{1, 1},
				},
			},
			{
				conn:  two,
				query: "UPDATE tx.data SET c0 = c0 + 10 WHERE pk = 1;",
			},
			{
				conn: two,
				assertion: func(s *dbr.Session) *dbr.SelectStmt {
					return s.Select("*").From("tx.data").Where("pk = 1")
				},
				expected: []testRow{
					{1, 21},
				},
			},
			{
				conn:  two,
				query: "COMMIT",
			},
		},
	},
}

func setupCommon(sess *dbr.Session) (err error) {
	queries := []string{
		"DROP DATABASE IF EXISTS tx;",
		"CREATE DATABASE IF NOT EXISTS tx;",
		"USE tx;",
		"CREATE TABLE data (pk int primary key, c0 int);",
		"INSERT INTO data VALUES (1,1),(2,2),(3,3);",
	}

	for _, q := range queries {
		if _, err = sess.Exec(q); err != nil {
			return
		}
	}
	return
}

func testConcurrentTx(t *testing.T, test ConcurrentTxTest) {
	conns, err := createNamedConnections(defaultConfig, one, two)
	require.NoError(t, err)
	defer func() { require.NoError(t, closeNamedConnections(conns)) }()

	err = setupCommon(conns[one])
	defer func() { require.NoError(t, teardownCommon(conns[one])) }()

	for _, q := range test.queries {
		conn := conns[q.conn]
		if q.query != "" {
			_, err = conn.Exec(q.query)
			require.NoError(t, err)
		}

		if q.assertion == nil {
			continue
		}

		var actual []testRow
		_, err = q.assertion(conn).Load(&actual)
		require.NoError(t, err)
		assert.Equal(t, q.expected, actual)
	}
}

func teardownCommon(sess *dbr.Session) (err error) {
	_, err = sess.Exec("DROP DATABASE tx;")
	return
}

type ServerConfig struct {
	database string
	host     string
	port     int
	user     string
	password string
}

type namedConnections map[string]*dbr.Session

// ConnectionString returns a Data Source Name (DSN) to be used by go clients for connecting to a running server.
func ConnectionString(config ServerConfig) string {
	return fmt.Sprintf("%v:%v@tcp(%v:%v)/%s",
		config.user,
		config.password,
		config.host,
		config.port,
		config.database,
	)
}

func createNamedConnections(config ServerConfig, names ...string) (nc namedConnections, err error) {
	nc = make(namedConnections, len(names))
	for _, name := range names {
		var c *dbr.Connection
		if c, err = dbr.Open("mysql", ConnectionString(config), nil); err != nil {
			return nil, err
		}
		nc[name] = c.NewSession(nil)
	}
	return
}

func closeNamedConnections(nc namedConnections) (err error) {
	for _, conn := range nc {
		if err = conn.Close(); err != nil {
			return
		}
	}
	return
}