313 lines
10 KiB
Go
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.Languages.LockSiteLanguages(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")
|
|
}
|
|
}},
|
|
}
|
|
}
|