95 lines
3.4 KiB
Go
95 lines
3.4 KiB
Go
package tasks
|
|
|
|
import (
|
|
"fmt"
|
|
"os"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/stretchr/testify/require"
|
|
"gorm.io/driver/postgres"
|
|
"gorm.io/gorm"
|
|
"senlinai-agent/backend/internal/models"
|
|
)
|
|
|
|
func TestPostgresMovePreventsOldProjectShareFromBeingInsertedConcurrently(t *testing.T) {
|
|
dsn := os.Getenv("DATABASE_URL")
|
|
if dsn == "" {
|
|
t.Skip("DATABASE_URL is not configured; skipping PostgreSQL row-lock concurrency test")
|
|
}
|
|
database, err := gorm.Open(postgres.Open(dsn), &gorm.Config{TranslateError: true})
|
|
require.NoError(t, err)
|
|
require.NoError(t, models.AutoMigrate(database))
|
|
suffix := fmt.Sprint(time.Now().UnixNano())
|
|
owner := models.SenlinAgentUser{Email: "lock-" + suffix + "@example.com", DisplayName: "Lock Owner", PasswordHash: "hash"}
|
|
require.NoError(t, database.Create(&owner).Error)
|
|
first := models.SenlinAgentProject{OwnerID: owner.ID, Name: "First", Identifier: "FIRST-" + suffix}
|
|
second := models.SenlinAgentProject{OwnerID: owner.ID, Name: "Second", Identifier: "SECOND-" + suffix}
|
|
require.NoError(t, database.Create(&first).Error)
|
|
require.NoError(t, database.Create(&second).Error)
|
|
task := models.SenlinAgentTask{ProjectID: first.ID, CreatedBy: owner.ID, Title: "Move", Status: "open"}
|
|
note := models.SenlinAgentNote{ProjectID: first.ID, CreatedBy: owner.ID, Title: "Old context", Markdown: "private"}
|
|
require.NoError(t, database.Create(&task).Error)
|
|
require.NoError(t, database.Create(¬e).Error)
|
|
t.Cleanup(func() {
|
|
database.Where("task_id = ?", task.ID).Delete(&models.SenlinAgentTaskShare{})
|
|
database.Where("entity_type = ? AND entity_id = ?", "task", task.ID).Delete(&models.SenlinAgentProjectEvent{})
|
|
database.Delete(¬e)
|
|
database.Delete(&task)
|
|
database.Delete(&first)
|
|
database.Delete(&second)
|
|
database.Delete(&owner)
|
|
})
|
|
|
|
locked := make(chan struct{})
|
|
release := make(chan struct{})
|
|
var once sync.Once
|
|
var releaseOnce sync.Once
|
|
releaseLock := func() { releaseOnce.Do(func() { close(release) }) }
|
|
t.Cleanup(releaseLock)
|
|
callbackName := "test:pause_first_task_lock_" + suffix
|
|
require.NoError(t, database.Callback().Query().After("gorm:query").Register(callbackName, func(tx *gorm.DB) {
|
|
if tx.Statement.Schema == nil || tx.Statement.Schema.Table != (models.SenlinAgentTask{}).TableName() {
|
|
return
|
|
}
|
|
if _, ok := tx.Statement.Clauses["FOR"]; !ok {
|
|
return
|
|
}
|
|
once.Do(func() {
|
|
close(locked)
|
|
<-release
|
|
})
|
|
}))
|
|
t.Cleanup(func() { database.Callback().Query().Remove(callbackName) })
|
|
|
|
service := NewService(database)
|
|
moveResult := make(chan error, 1)
|
|
go func() {
|
|
_, err := service.Update(owner.ID, first.Identity, task.Identity, UpdateTaskInput{
|
|
Title: task.Title, NextProjectIdentity: second.Identity,
|
|
})
|
|
moveResult <- err
|
|
}()
|
|
select {
|
|
case <-locked:
|
|
case <-time.After(5 * time.Second):
|
|
t.Fatal("timed out waiting for Update to acquire the task row lock")
|
|
}
|
|
|
|
shareResult := make(chan error, 1)
|
|
go func() { shareResult <- service.ShareObject(task.ID, "note", note.ID) }()
|
|
select {
|
|
case err := <-shareResult:
|
|
t.Fatalf("ShareObject returned before the moving transaction released its task lock: %v", err)
|
|
case <-time.After(150 * time.Millisecond):
|
|
}
|
|
releaseLock()
|
|
require.NoError(t, <-moveResult)
|
|
require.ErrorContains(t, <-shareResult, "shared object not found in task project")
|
|
|
|
var shareCount int64
|
|
require.NoError(t, database.Model(&models.SenlinAgentTaskShare{}).Where("task_id = ?", task.ID).Count(&shareCount).Error)
|
|
require.Zero(t, shareCount)
|
|
}
|