golang

Создаём репозиторий в Go через менеджер транзакций

  • среда, 17 мая 2023 г. в 00:01:44
https://habr.com/ru/companies/avito/articles/727168/

Всем привет! Я Илья Сергунин, веб-разработчик из продуктовой команды Авито. Мы пишем на Go сервис для выкупа мобильных телефонов, про MLP которого уже писали в блоге. В качестве примеров я буду использовать всем знакомый интернет-магазин, чтобы показать, как создать репозиторий через менеджер транзакций.

Важное замечание: в Go нужно явно прописывать обработку каждой ошибки. В некоторых примерах я скрываю эту часть кода, чтобы они легче читались.

Для чего нам нужен менеджер транзакций

Паттерн Repository впервые был предложен Мартином Фаулером в книге «Шаблоны корпоративных приложений». Я буду использовать альтернативное понимание с методом сохранения данных. Его предложил Эрик Эванс в книге «Предметно-ориентированное проектирование (DDD). Структуризация сложных программных систем».

Repository конвертирует модель в представление для базы данных для сохранения, а при получении данных — обратно. Каждая реализация паттерна работает только с одной моделью. Repository в большинстве случаев не содержит внутри себя валидацию бизнес-правил.

По умолчанию у него очень простой интерфейс с двумя методами: для получения и сохранения данных.

Допустим, у нас есть репозиторий для пользователя:

User может включать в себя никнейм, пароль, имя, фамилию, настройки уведомлений и т.д. Все эти данные принадлежат нашему пользователю, однако могут быть представлены в разном виде.

Имя и фамилия могут быть вынесены в отдельную таблицу, если в системе пользователями могут быть люди и компании. Настройки уведомлений могут быть представлены в виде json-объекта. Их нужно конвертировать в структуры. Repository позволяет сконцентрировать эти преобразования в одном месте и сосредоточиться на модели с бизнес-правилами, а не на работе с базой данных.

Давайте посмотрим на примеры:

// Получение
func (r *userRepo) GetByID(id domain.UserID) (*domain.User, error) {
	query := `SELECT * FROM "user" WHERE id = ?;`
	uRow := userRow{}

	err := r.db.Get(&uRow, r.db.Rebind(query), id)

	return r.toModel(uRow), nil
}
// Сохранение
func (r *userRepo) Save(u *domain.User) error {
	query := `INSERT INTO "user" (username, password, notification)
VALUES (:username, :password, :notification)
ON CONFLICT (id)
    DO UPDATE SET excluded.username     = :username,
                  excluded.password     = :password,
                  excluded.notification = :notification
RETURNING id;`

	rows, err := r.db.Query(query, r.toRow(u))

	defer rows.Close()
	if !rows.Next() {
		return rows.Err()
	}

	err = rows.Scan(&u.ID)

	return err
}

В методах toModel и toRow находится маппинг данных из представления для базы данных в модель и обратно.

Мы хотим, чтобы при любых изменениях и доработках продукта интерфейс паттерна оставался таким же простым. 

Шаг 1 — регистрируем пользователя 

Каждого нового пользователя нужно зарегистрировать:

func (u *Register) Handle(in In) (user *domain.User, err error) {
	user, err = domain.NewUser(in.Username, in.Password)

	err = u.userRepo.Save(user)

	return user, nil
}

Мы хотим сообщить пользователю о том, что он зарегистрировался, поэтому добавляем отправку уведомления через очередь сообщений:

func (u *Register) Handle(in In) (user *domain.User, err error) {
	user, err = domain.NewUser(in.Username, in.Password)

	err = u.userRepo.Save(user)

    // НОВЫЙ КОД
	err = u.queue.Publish(domain.Registered{ID: user.ID})

	return user, nil
}

Мы предполагаем, что в очереди сообщений может произойти сбой, и тогда уведомление о регистрации не дойдет до пользователя. Чтобы этого избежать, добавим транзакцию и явно передадим её в репозиторий и в очередь:

func (u *Register) Handle(in In) (user *domain.User, err error) {
    // НОВЫЙ КОД
    tr, err = u.db.Beginx()

	user, err = domain.NewUser(in.Username, in.Password)

	err = u.userRepo.Save(tr, user)
	err = u.queue.Publish(tr, domain.Registered{ID: user.ID})

    // НОВЫЙ КОД
	err = tr.Commit() // tr.Rollback()

	return user, nil
}

Для очереди можно использовать паттерн transaction outbox, чтобы отправлять события атомарно с сохранением.

И также нужно дописать функцию сохранения, чтобы репозиторий знал о существовании транзакции:

// Сохранение
func (r *userRepo) Save(tr domain.Tr, u *domain.User) error {
	query := `INSERT INTO "user" ...;`

    // НОВЫЙ КОД
	if tr != nil {
		tr = r.db
	}

	rows, err := sqlx.NamedQuery(tr, query, r.toRow(u))

	defer rows.Close()
	if !rows.Next() {
		return rows.Err()
	}

	err = rows.Scan(&u.ID)

	return err
}

Мы получили транзакционность операций, но взамен усложнили интерфейс репозитория: 

Шаг 2 — создаём заказ и покупку в один клик для пользователя 

У нашего пользователя есть возможность сделать заказ. Для этого сценария нам нужен второй репозиторий:

Прописываем сценарий заказа и сразу предусматриваем транзакционность:

func (u *Purchase) Handle(tx *sqlx.Tx, in In) (order *domain.Order, err error) {
    tr, err = u.db.Beginx();

	order, err = domain.NewOrder(in.UserID, in.ProductID, in.Quantity)

	err = u.orderRepo.Save(tr, order)

	u.queue.Publish(tr, domain.Purchased{ID: order.ID})

	err = tr.Commit() // tr.Rollback()

	return order, nil
}

Для увеличения продаж дадим пользователю возможность совершить покупку одновременно с регистрацией:

func (u *FastPurchase) Handle(in In) (out Out, err error) {
	out.User, err = u.register.Handle(tx, register.In{
		Username: in.Register.Username,
	})

	out.Order, err = u.purchase.Handle(tx, purchase.In{
		UserID:    out.User.ID,
		ProductID: in.Purchase.ProductID,
		Quantity:  in.Purchase.Quantity,
	})

	return out, nil
}

Register и Purchase являются вложенными сценариями в FastPurchase. Давайте сделаем быструю покупку в одной транзакции:

func (u *FastPurchase) Handle(in In) (out Out, err error) {
	tx, err := u.db.Beginx()

	out.User, err = u.register.Handle(tx, register.In{
		Username: in.Register.Username,
	})

	out.Order, err = u.purchase.Handle(tx, purchase.In{
		UserID:    out.User.ID,
		ProductID: in.Purchase.ProductID,
		Quantity:  in.Purchase.Quantity,
	})

	err = tx.Commit() // tr.Rollback()

	return out, nil
}

Напомню, что в сценариях регистрации пользователя и покупки тоже нужно учесть транзакции. Доработаем их, чтобы транзакция работала с вложенными сценариями корректно. 

Регистрация пользователя:

// Было
func (u *Register) Handle(in In) (user *domain.User, err error) {
    tr, err = u.db.Beginx()

	user, err = domain.NewUser(in.Username, in.Password)

	err = u.userRepo.Save(tr, user)
	err = u.queue.Publish(tr, domain.Registered{ID: user.ID})

	err = tr.Commit() // tr.Rollback()

	return user, nil
}
// Стало
func (u *Register) Handle(tx *sqlx.Tx, in In) (user *domain.User, err error) {
	hasExternalTx := true
	tr := domain.Tr(tx)
	if tr == nil {
		tr, err = u.db.Beginx()
		hasExternalTx = false
	}

	defer func() {
		if !hasExternalTx && err != nil {
			tx.Rollback()
		}
	}()

	user, err = domain.NewUser(in.Username, in.Password)

	err = u.userRepo.Save(tr, user); err != nil {
	err = u.queue.Publish(tr, domain.Registered{ID: user.ID})

	if !hasExternalTx {
		return nil, tx.Commit()
	}

	return user, nil
}

Что в итоге получилось: у нас есть транзакционность и вложенные сценарии. Но при этом ещё и много дополнительного кода. Если мы захотим перейти на другую базу данных, сделать это будет сложно.

Шаг 3 — замыкаем выполнение сценариев в функцию

Чтобы улучшить решение, добавим замыкание в репозитории пользователя: 

func (r *userRepo) FastOrder(fn func() (*User, *Order, error)) error {
   tr, err := r.db.Begin()

   user, order, err := fn() // выполнения сценария

   err = r.Save(tr, user)
   err = r.orderRepo.Save(tr, order)

   err = tr.Commit() // или tr.Rollback()

   return nil
}

Но теперь пользовательский репозиторий знает о заказах. Нам это не нужно: со временем продукт улучшается и растёт, моделей и репозиториев становится всё больше. Сценарии тоже расширяются и усложняются, и репозиторий пользователя узнаёт о других моделях. Знание усложняет работу с репозиторием и его модификацию. В итоге он потеряет свою изначальную функцию — работа с данными пользователя. 

Решим проблему просто: вынесем замыкание в отдельную функцию на более высокий уровень. Получим новую функцию WithTransaction: 

func WithTransaction(tx *sqlx.Tx, fn func(*sqlx.Tx) error) error {
   hasExternalTx := true
   if tx == nil {
       tx, err = DB.Begin()
       hasExternalTx = false
   }

   err := fn(tx) // выполнение сценария

   if !hasExternalTx {
       err = tx.Commit() // tx.Rollback()
   }

   return nil
}

Благодаря этому мы можем сократить сценарий регистрации пользователя, чтобы была видна его суть, а не инфраструктура: 

// Было
func (u *Register) Handle(tx *sqlx.Tx, in In) (user *domain.User, err error) {
	hasExternalTx := true
	tr := domain.Tr(tx)
	if tr == nil {
		tr, err = u.db.Beginx()
		hasExternalTx = false
	}

	defer func() {
		if !hasExternalTx && err != nil {
			tx.Rollback()
		}
	}()

	// Тело сценария

	if !hasExternalTx {
		return nil, tx.Commit()
	}

	return user, nil
}
// Стало
func (u *Register) Handle(tx *sqlx.Tx, in In) (user *domain.User, err error) {
    user, err = domain.NewUser(in.Username, in.Password)

    err := WithTransaction(tx, func(tx *sqlx.Tx) error {
    	err = u.userRepo.Save(tx, user); err != nil {
    	err = u.queue.Publish(tx, domain.Registered{ID: user.ID})
    })

   return user, err
}

Что получается при таком решении: мы спрятали работу с транзакциями в отдельную функцию, но все еще передаём их явно в интерфейсе репозитория. 

Где хранить транзакции в Go

У нас получилось довольно неплохое решение, но это пока не менеджер транзакций. Транзакция как аргумент привязывает нас к определенной базе данных на разных уровнях приложения. Давайте попробуем подсмотреть решение в других языках — как они избавляются от прорастания базы данных во все части приложения. В разных языках эта проблема решается по-разному: 

  • в Python (SQLAlchemy) транзакция передаётся явно, как аргумент, но нам хочется от этого отказаться;

  • в Python (Django) и однопоточных языках PHP и JavaScript используется глобальная переменная для хранения транзакции;

  • в C#, Java транзакцию можно привязать к потоку. 

Среди этих решений нам может подойти последнее: в Go есть Goroutine, которые немного похожи на потоки. Но это не одно и то же, поэтому решение нужно модифицировать.

Варианты хранения транзакции в Go:

  1. Передать транзакцию как аргумент, что мы и делаем. Но хотим отказаться от этого подхода.

  2. Привязать транзакцию к Goroutine по id, но это нестабильное решение.

  3. Сохранить в local-storage, но его реализация привязана к конкретным версиям Go, не развивается и поэтому не подходит нам.

  4. Передать как context. Внутри этот пакет устроен как список, а значит, у него высокая сложность поиска O(N). 

Получается, идеального решения нет, но мы всё-таки остановимся на context. Это единственный способ хранить транзакцию в Go и не опасаться, что решение сломается при обновлении языка на новую версию. Транзакцию можно хранить в context, если они создаются и завершаются в рамках одного запроса.

Как довести решение до идеала

У нас есть пять пожеланий к решению: 

  1. простой интерфейс репозитория,

  2. транзакционность операций,

  3. поддержка вложенных сценариев,

  4. скрытая работа транзакций,

  5. поддержка разных баз данных. 

В решении, на котором мы остановились, не закрыты только два из них: 

  1. простой интерфейс репозитория;

  2. поддержка разных баз данных.

При этом у нас уже есть два готовых элемента: функция WithTransaction и хранилище на основе context.

Расширим функцию WithTransaction до полноценного менеджера транзакций:

type Manager interface {
   Do(context.Context, func(context.Context) error) error

   DoWithSettings(
     context.Context,
     Settings,
     func(context.Context) error,
   ) error
}

В менеджере добавим интерфейс с возможностью донастройки текущей транзакции. Общие настройки включают параметры, которые подходят для разных баз данных. Также можно добавить специализированные — для конкретной базы.

type Settings interface {
   // Объединяет настройки.
   EnrichBy(external Settings) Setings
   // Ключ для поиска текущей транзакции.
   CtxKey() CtxKey
   // Настройки запуска транзакции.
   Propagation() Propagaion
   // Отменяет ли внешняя транзакция вложенные в неё.
   Cancelable() bool
   // Время на выполнение транзакции.
   TimeoutOrNil() *time.Duration
}

Теперь мы можем убрать из сценария регистрации пользователя существенную часть инфраструктуры: 

// Было
func (u *Register) Handle(tx *sqlx.Tx, in In) (user *domain.User, err error) {
	hasExternalTx := true
	tr := domain.Tr(tx)
	if tr == nil {
		tr, err = u.db.Beginx()
		hasExternalTx = false
	}

	defer func() {
		if !hasExternalTx && err != nil {
			tx.Rollback()
		}
	}()

	// Тело сценария

	if !hasExternalTx {
		return nil, tx.Commit()
	}

	return user, nil
}
// Стало
func (u *Register) Handle(ctx context.Context, in In) (user *domain.User, err error) {
    user, err = domain.NewUser(in.Username, in.Password)

    err := u.trm.Do(ctx, func(ctx context.Context) error {
    	err = u.userRepo.Save(ctx, user); err != nil {
    	err = u.queue.Publish(ctx, domain.Registered{ID: user.ID})
    })

   return user, err
}

Новую версию кода намного легче читать. 

Также нужно переписать с использованием context сценарий создания заказа: 

func (u *Purchase) Handle(ctx context.Context, in In) (order *domain.Order, err error) {
	order, err = domain.NewOrder(in.UserID, in.ProductID, in.Quantity)

	err = u.trm.Do(ctx, func(ctx context.Context) error {
		err = u.orderRepo.Save(ctx, order)
		return u.queue.Publish(ctx, domain.Purchased{ID: order.ID})
	})

	return order, nil
}

И быструю покупку:

func (u *FastPurchase) Handle(ctx context.Context, in In) (out Out, err error) {
	err = u.trm.Do(ctx, func(ctx context.Context) error {
		out.User, err = u.register.Handle(ctx, register.In{
			Username: in.Register.Username,
			Password: in.Register.Password,
		})

		out.Order, err = u.purchase.Handle(ctx, purchase.In{
			UserID:    out.User.ID,
			ProductID: in.Purchase.ProductID,
			Quantity:  in.Purchase.Quantity,
		})

		return err
	})

	return out, nil
}

Теперь займёмся хранилищем транзакций и посмотрим, поможет ли оно сделать репозиторий идеальным.

Добавим интерфейс для хранилища транзакций, чтобы получать транзакцию из контекста по ключу или ключу по умолчанию. Наличие метода получения по ключу нужно, чтобы поддерживать подключение к нескольким базам данных.

type CtxManager interface {
   Default(context.Context) Ir
   ByKey(context.Context, CtxKey) Tr
}

type Tr interface{
   // *sqlx.DB
   // *sqlx.Tx
}

Теперь можем доработать функцию сохранения в репозиторий. Теперь функцию проще читать и в ней меньше кода:

// Было
func (r *userRepo) Save(tr domain.Tr, u *domain.User) error {
    query := `INSERT INTO "user" ...;`
    
    if tr != nil {
        tr = r.db
    }
    
    rows, err := sqlx.NamedQuery(tr, query, r.toRow(u))
    
    defer rows.Close()
    if !rows.Next() {
        return rows.Err()
    }
    
    err = rows.Scan(&u.ID)
    
    return err
}
// Стало
func (r *userRepo) Save(ctx context.Context, u *User) error {
    query := `INSERT INTO "user" ...;`

    rows, err := sqlx.NamedQueryContext(
      ctx, r.getter.DefaultTrOrDB(ctx, r.db), query, r.toRow(u))
    
    defer rows.Close()
    if !rows.Next() {
        return rows.Err()
    }
    
    err = rows.Scan(&u.ID)
    
    return err
}

Еще хочется, чтобы решение легко мигрировало с одной базы данных на другую. Для этого создадим абстракцию над транзакцией на основе Spring в Java и System Transaction из C#: 

type Transaction interface {
   IsActive() bool // определяет активность транзакции
   Commit(context.Context) error // применяет изменения
   Rollback(context.Context) error // отменяет изменения
   Transaction() interface{} // возвращает реальную транзакцию
}

// Создаёт транзакцию
type TrFactory func(Settings) Transaction

// Создаёт вложенную транзакцию
// для БД с их поддержкой
type NestedTrFactory func(Setting) Transaction

Наконец, научим хранилище транзакций работать с абстракцией и делать локальный менеджер транзакций для конкретного драйвера:

type CtxManager interface {
   Default(context.Context) Transaction
   ByKey(context.Context, CtxKey) Transaction
}

type SQLtxManager interface {/* *sql.DB *sql.TX */}

type SQLtxManager interface {
  DefaultTrOrDB(context.Context, Tr) Tr
  TrOrDB(context.Context, CtxKey, Tr) Tr
}

В итоге мы выполнили все пожелания:

Как менеджер транзакций влияет на производительность 

Чтобы проверить влияние менеджера на скорость работы кода, я написал бенчмарк на sqlmock, который мы используем для тестирования. В результате разрыв в производительности с менеджером транзакций и без него получился всего около 3,5%.

Но при этом меня удивило, что результаты теста оказались в миллисекундах, поэтому я переписал бенчмарк на SQLite для базы данных в памяти. Новый тест показал, что с менеджером транзакций производительность падает на 18%. В реальном времени это всего 5 микросекунд. 

Я решил протестировать кейс, который встречается чаще: когда база данных лежит в файле. На этот раз разницы практически не было. Это связано с большими издержками при работе файловой системы. 

Ознакомиться с бенчмарками можно по ссылке.

Какие недостатки есть у менеджера транзакций

Работает только на Go 1.13 и выше. При этом Go обновляется минорными версиями, поэтому перейти на нужную достаточно просто. 

Мало адаптеров для драйверов, баз данных и ORM. Пока есть только database/sql, jmoiron/sqlx, mongo-go-driver, gorm, go-redis). Но вы можете написать свой адаптер, на это понадобится всего около 60 строк кода.

Снижает производительность. Но нужно учесть, что этот разрыв заметен в ситуации, когда база данных хранится в памяти. В реальности база обычно хранится в файле или вообще на отдельном сервере. В этом случае нужно добавлять издержки на работу файловой системы и сетевое взаимодействие. Поэтому в большинстве реальных задач разница в скорости работы с файловым менеджером и без него будет практически незаметной. 

Нужно постоянно передавать в функции context. Правда, если вы пишете веб-приложение на Go, то скорее всего, уже носите его с собой. Это полезная штука, которая часто используется. Например, чтобы не делать лишние вычисления, если пользователь отменил запрос. Также context может хранить данные авторизованного пользователя или другие данные для логирования. Можно возразить, что транзакция — слишком сложный объект для контекста. Однако ряд статей (1, 2) из wiki GoLang и описание самого пакета context разрешает хранить транзакцию, если она открывается и закрывается в рамках одного запроса.

Не поддерживает долгие транзакции. Если у вас есть сетевой вызов к отдельному серверу, он не будет работать. Например, проверку по номеру телефона нужно будет вынести из транзакции. Пожалуй, это самый большой минус, который лечится с помощью паттерна UnitOfWork, но это отдельная тема. 

Нужен ли вам менеджер транзакций

Для нас менеджер транзакций оказался очень полезен. Раньше репозитории были жёстко привязаны к конкретной базе данных через транзакцию, а теперь мы передаем её в context. То есть можем переключаться на другую базу с минимальным количеством изменений в коде сценариев. Дополнительно скрыли дублирования работы с транзакциями до унифицированного интерфейса.

// Было
type UserRepo interface {
   GetByID(*sqlx.Tx, UserID) (*User, error)
   Save(*sqlx.Tx, *User) error
}
// Стало 
type UserRepo interface {
   GetByID(context.Context, UserID) (*User, error)
   Save(context.Context, *User) error
}

Код сценариев стал понятнее, потому что мы убрали из них инфраструктурные элементы.

Нужен ли вам менеджер транзакций в вашем проекте? Это можно понять по количеству вложенных сценариев. Это важно, потому что если их хотя бы больше двух, то количество строк кода начинает быстро расти.

Сам менеджер транзакции добавляет к коду всего 5 строк на инициализацию, и ещё несколько — на каждое использование. Без менеджера нам нужно каждый раз писать по 8 строчек кода.

Кстати, ещё я советую учесть когнитивную нагрузку на разработчика, который создаёт и поддерживает код с множеством вложенных сценариев. Чем кода больше, тем выше вероятность ошибки. При таких условиях менеджер транзакций очень помогает справляться с поддержкой лучше. 

Весь код решения можно посмотреть на Github, как и примеры. И приходите в комментарии — обсудить, стоит ли внедрять это решение, чтобы упростить код и уменьшить вероятность ошибок.

Предыдущая статья: Ультимативный гайд по HTTP. Структура запроса и ответа