Files
RenewCMS/test/localization_concurrency_persistence_test.go
T

313 lines
10 KiB
Go

package test
import (
domainArticle "RenewCMS/internal/domain/article"
"RenewCMS/internal/domain/localization"
"RenewCMS/internal/domain/site"
"RenewCMS/internal/domain/transaction"
"RenewCMS/internal/infrastructure/persistence/database"
entity "RenewCMS/internal/infrastructure/persistence/models"
persistence "RenewCMS/internal/infrastructure/persistence/repositories"
"RenewCMS/internal/infrastructure/useCases"
"context"
"errors"
"fmt"
"sync"
"testing"
"time"
"gorm.io/gorm"
"gorm.io/gorm/clause"
)
const (
nowaitRefusalBudget = 3 * time.Second
setupRaceRounds = 4
setupRaceAttempts = 80
setupRacePause = 5 * time.Millisecond
mysqlShortLockWait = "SET SESSION innodb_lock_wait_timeout = 1"
postgresShortLockWait = "SET LOCAL lock_timeout = '300ms'"
pendingSiteDefaultLang = "en"
)
type localizationCase struct {
name string
run func(t *testing.T, db *gorm.DB)
}
type vanishingSites struct {
*persistence.SiteRepository
db *gorm.DB
}
func (v *vanishingSites) GetByKey(key string) (site.Site, error) {
found, err := v.SiteRepository.GetByKey(key)
if err == nil {
err = v.db.Delete(&entity.Site{}, found.ID).Error
}
return found, err
}
func pendingSite(t *testing.T, db *gorm.DB, key string) site.Site {
t.Helper()
return seedLocalizedSite(t, db, key, []entity.SiteLanguage{{Lang: pendingSiteDefaultLang, IsDefault: true, Enabled: true}}, nil)
}
func setupChoicesFor(keys ...string) []localization.SetupChoice {
choices := []localization.SetupChoice{{Key: "default", Lang: pendingSiteDefaultLang}}
for _, key := range keys {
choices = append(choices, localization.SetupChoice{Key: key, Lang: pendingSiteDefaultLang})
}
return choices
}
func createOnSite(db *gorm.DB, key string, title string) (domainArticle.Article, error) {
return createThroughUseCase(db).CreateArticle(context.Background(), useCases.CreateArticleCommand{Title: title, Site: &key, Caller: ownerCaller(1)})
}
func retryWhile(attempts int, retryable func(error) bool, work func() error) error {
var err error
for attempt := 0; attempt < attempts; attempt++ {
if err = work(); err == nil || !retryable(err) {
return err
}
time.Sleep(setupRacePause)
}
return err
}
func isContention(err error) bool {
return errors.Is(err, transaction.ErrLockContention) || errors.Is(err, localization.ErrSetupInProgress)
}
func articlesOnSite(t *testing.T, db *gorm.DB, siteID uint32) []entity.Article {
t.Helper()
var rows []entity.Article
if err := db.Unscoped().Where("site_id = ?", siteID).Order("id ASC").Find(&rows).Error; err != nil {
t.Fatal(err)
}
return rows
}
func holdSiteLock(t *testing.T, db *gorm.DB, siteID uint32, then func(transaction.Repositories) error) (func() error, error) {
t.Helper()
held := make(chan struct{})
release := make(chan struct{})
finished := make(chan error, 1)
go func() {
finished <- persistence.NewTransactor(db).Run(context.Background(), func(tx transaction.Repositories) error {
if err := tx.Sites.LockSite(siteID); err != nil {
return err
}
close(held)
<-release
return then(tx)
})
}()
var once sync.Once
finish := func() error {
once.Do(func() { close(release) })
return <-finished
}
select {
case <-held:
t.Cleanup(func() { once.Do(func() { close(release) }) })
return finish, nil
case err := <-finished:
return nil, err
}
}
func lockSiteRow(tx *gorm.DB, siteID uint32) error {
var locked entity.Site
return tx.Clauses(clause.Locking{Strength: "UPDATE"}).First(&locked, siteID).Error
}
func shortLockWait(db *gorm.DB) string {
if db.Name() == database.DriverMySQL {
return mysqlShortLockWait
}
return postgresShortLockWait
}
func skipWithoutRowLocks(t *testing.T, db *gorm.DB) {
t.Helper()
if db.Name() == database.DriverSQLite {
t.Skip("SQLite ignores FOR UPDATE; its single writer is covered by the race cases")
}
}
func localizationConcurrencyCases() []localizationCase {
return []localizationCase{
{"An article create whose site is deleted after the lookup is refused with target site deleted", func(t *testing.T, db *gorm.DB) {
gone := pendingSite(t, db, "vanishing")
create := useCases.NewCreateArticleUseCase(persistence.NewTransactor(db), &fakeBodyRenderer{}, discardRecorder, &vanishingSites{persistence.NewSiteRepository(db), db}, persistence.NewAuthorRepository(db))
key := gone.Key
_, err := create.CreateArticle(context.Background(), useCases.CreateArticleCommand{Title: "Vanishing site", Site: &key, Caller: ownerCaller(1)})
if !errors.Is(err, domainArticle.ErrTargetSiteDeleted) {
t.Fatalf("expected ErrTargetSiteDeleted, got %v", err)
}
if rows := articlesOnSite(t, db, gone.ID); len(rows) != 0 {
t.Errorf("an article was stored on the deleted site: %+v", rows)
}
}},
{"Duplicate addresses and languages rejected by the server become domain errors", func(t *testing.T, db *gorm.DB) {
owner := seedLocalizedSite(t, db, "unique", nil, nil)
articles := persistence.NewArticleRepository(db)
first, err := articles.Create(newStorageArticle(owner.ID, "en", "same-address", 0))
if err != nil {
t.Fatal(err)
}
_, slugErr := articles.Create(newStorageArticle(owner.ID, "en", "same-address", 0))
_, languageErr := articles.Create(newStorageArticle(owner.ID, "en", "other-address", first.StoryID))
duplicate := entity.Article{Title: "Raw", SiteID: owner.ID, Lang: "en", Slug: "same-address", StoryID: fixtureStory()}
rawErr := db.Create(&duplicate).Error
if !errors.Is(slugErr, domainArticle.ErrSlugTaken) {
t.Errorf("duplicate slug: %v", slugErr)
}
if !errors.Is(languageErr, domainArticle.ErrLanguageTaken) {
t.Errorf("duplicate story language: %v", languageErr)
}
if !database.IsUniqueViolation(db, rawErr) || database.IsLockContention(db, rawErr) {
t.Errorf("the raw duplicate must be a unique violation: %v", rawErr)
}
if rows := articlesOnSite(t, db, owner.ID); len(rows) != 1 {
t.Errorf("rejected writes left %d rows", len(rows))
}
}},
{"Language setup does not wait for a site held by an article create", func(t *testing.T, db *gorm.DB) {
skipWithoutRowLocks(t, db)
held := pendingSite(t, db, "held")
apply := useCases.NewApplyLanguageSetupUseCase(persistence.NewTransactor(db), discardRecorder)
finish, err := holdSiteLock(t, db, held.ID, func(tx transaction.Repositories) error {
_, err := tx.Articles.Create(newStorageArticle(held.ID, "", "held-create", 0))
return err
})
if err != nil {
t.Fatalf("hold: %v", err)
}
started := time.Now()
_, refused := apply.Apply(context.Background(), setupChoicesFor(held.Key))
waited := time.Since(started)
holdErr := finish()
if !errors.Is(refused, localization.ErrSetupInProgress) || !errors.Is(refused, transaction.ErrLockContention) {
t.Fatalf("expected a NOWAIT refusal, got %v", refused)
}
if waited > nowaitRefusalBudget {
t.Errorf("the setup waited %v for the lock", waited)
}
if holdErr != nil {
t.Fatalf("the held create failed: %v", holdErr)
}
result, err := apply.Apply(context.Background(), setupChoicesFor(held.Key))
if err != nil || result.Assigned != 1 {
t.Fatalf("retry = %+v, %v", result, err)
}
for _, row := range articlesOnSite(t, db, held.ID) {
if row.Lang != pendingSiteDefaultLang {
t.Errorf("article %d kept lang %q", row.ID, row.Lang)
}
}
}},
{"Language setup racing an article create never strands an article", func(t *testing.T, db *gorm.DB) {
apply := useCases.NewApplyLanguageSetupUseCase(persistence.NewTransactor(db), discardRecorder)
for round := 0; round < setupRaceRounds; round++ {
racing := pendingSite(t, db, fmt.Sprintf("racing%d", round))
start := make(chan struct{})
var group sync.WaitGroup
var createErr, setupErr error
group.Add(2)
go func() {
defer group.Done()
<-start
createErr = retryWhile(setupRaceAttempts, func(error) bool { return true }, func() error {
_, err := createOnSite(db, racing.Key, fmt.Sprintf("Racing article %d", round))
return err
})
}()
go func() {
defer group.Done()
<-start
setupErr = retryWhile(setupRaceAttempts, isContention, func() error {
_, err := apply.Apply(context.Background(), setupChoicesFor(racing.Key))
return err
})
}()
close(start)
group.Wait()
if createErr != nil || setupErr != nil {
t.Fatalf("round %d: create %v, setup %v", round, createErr, setupErr)
}
rows := articlesOnSite(t, db, racing.ID)
if len(rows) != 1 || rows[0].Lang != pendingSiteDefaultLang {
t.Fatalf("round %d: articles %+v", round, rows)
}
}
}},
{"A lock wait timeout from the server is lock contention", func(t *testing.T, db *gorm.DB) {
skipWithoutRowLocks(t, db)
held := pendingSite(t, db, "waiting")
finish, err := holdSiteLock(t, db, held.ID, func(transaction.Repositories) error { return nil })
if err != nil {
t.Fatal(err)
}
waitErr := db.Transaction(func(tx *gorm.DB) error {
if err := tx.Exec(shortLockWait(db)).Error; err != nil {
return err
}
return lockSiteRow(tx, held.ID)
})
if err := finish(); err != nil {
t.Fatal(err)
}
if waitErr == nil || !database.IsLockContention(db, waitErr) {
t.Errorf("expected lock contention, got %v", waitErr)
}
}},
{"A deadlock from the server is lock contention", func(t *testing.T, db *gorm.DB) {
skipWithoutRowLocks(t, db)
first := pendingSite(t, db, "deadfirst")
second := pendingSite(t, db, "deadsecond")
bothLocked := sync.WaitGroup{}
bothLocked.Add(2)
failures := make(chan error, 2)
crossLock := func(own uint32, other uint32) {
failures <- db.Transaction(func(tx *gorm.DB) error {
err := lockSiteRow(tx, own)
bothLocked.Done()
if err != nil {
return err
}
bothLocked.Wait()
return lockSiteRow(tx, other)
})
}
go crossLock(first.ID, second.ID)
go crossLock(second.ID, first.ID)
contended := 0
for range 2 {
if err := <-failures; err != nil {
if !database.IsLockContention(db, err) {
t.Errorf("unexpected error: %v", err)
}
contended++
}
}
if contended == 0 {
t.Errorf("one of the crossed transactions must be chosen as the deadlock victim")
}
}},
}
}