weichenandCursor
7f8ad76d12
Occupy the integration slot for M1 live sign-off, cover two pgx pools plus four sessions, and add a Playwright spec that treats projection consistency as 200 rather than 404. Co-authored-by: Cursor <cursoragent@cursor.com>
248 lines
8.3 KiB
Go
248 lines
8.3 KiB
Go
package docrevision_test
|
||
|
||
import (
|
||
"context"
|
||
"fmt"
|
||
"strings"
|
||
"sync"
|
||
"testing"
|
||
"time"
|
||
|
||
"chat-agent-system/internal/app/docflow"
|
||
"chat-agent-system/internal/app/uow"
|
||
"chat-agent-system/internal/db"
|
||
"chat-agent-system/internal/docgov"
|
||
"chat-agent-system/internal/docrevision"
|
||
|
||
"github.com/jackc/pgx/v5/pgxpool"
|
||
)
|
||
|
||
func TestDocumentCommitConcurrencyPG(t *testing.T) {
|
||
if testing.Short() {
|
||
t.Skip("postgresql concurrency is not part of -short")
|
||
}
|
||
dsn := db.SkipIfNoTestDatabaseURL(t)
|
||
if strings.Contains(dsn, "5433") || strings.Contains(dsn, "agent_system_dev") ||
|
||
strings.Contains(dsn, "5434") && strings.Contains(dsn, "agent_system_test") ||
|
||
strings.Contains(dsn, "5435") {
|
||
t.Fatal("TEST_DATABASE_URL points at a shared env port; refuse agent_system_dev/test/main")
|
||
}
|
||
ctx := context.Background()
|
||
pool, err := pgxpool.New(ctx, dsn)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
t.Cleanup(pool.Close)
|
||
if err := db.RunMigrationsOnceForTest(ctx, pool, dsn); err != nil {
|
||
t.Fatalf("migrate: %v", err)
|
||
}
|
||
|
||
suffix := fmt.Sprintf("%d", time.Now().UnixNano())
|
||
var projectID int64
|
||
if err := pool.QueryRow(ctx, `INSERT INTO projects (code, name, status) VALUES ($1,$2,'active') RETURNING id`,
|
||
"docflow-"+suffix, "docflow").Scan(&projectID); err != nil {
|
||
t.Fatalf("project: %v", err)
|
||
}
|
||
t.Cleanup(func() {
|
||
_, _ = pool.Exec(context.Background(), `DELETE FROM document_change_events WHERE doc_id LIKE $1`, "docflow_"+suffix+"%")
|
||
_, _ = pool.Exec(context.Background(), `DELETE FROM document_versions WHERE doc_id LIKE $1`, "docflow_"+suffix+"%")
|
||
_, _ = pool.Exec(context.Background(), `DELETE FROM document_snapshots WHERE doc_id LIKE $1`, "docflow_"+suffix+"%")
|
||
_, _ = pool.Exec(context.Background(), `DELETE FROM document_records WHERE project_id=$1`, projectID)
|
||
_, _ = pool.Exec(context.Background(), `DELETE FROM projects WHERE id=$1`, projectID)
|
||
})
|
||
|
||
docID := "docflow_" + suffix
|
||
rev := docrevision.NewService(pool)
|
||
gov := docgov.NewGovernanceEngine(pool)
|
||
if _, err := gov.CreateOrGetDocument(ctx, docID, projectID, "intake", "t", "V0.5_feasibility", "ws/"+docID, 1, "A", 2, "R", 3, "P"); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
st, err := rev.OpenTemplateVersion(ctx, docID, "# base\n", docrevision.ActorSnapshot{}, projectID, "ws/"+docID)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
cmds := &docflow.Commands{UoW: uow.NewPG(pool), Gov: gov, Rev: rev}
|
||
|
||
var wg sync.WaitGroup
|
||
wins := 0
|
||
var mu sync.Mutex
|
||
for i := 0; i < 2; i++ {
|
||
wg.Add(1)
|
||
go func(i int) {
|
||
defer wg.Done()
|
||
_, err := cmds.HumanEdit(ctx, docflow.Principal{UserID: 1}, docflow.HumanEditCommand{
|
||
DocID: docID,
|
||
Request: docrevision.HumanRevisionRequest{
|
||
DocID: docID, Markdown: fmt.Sprintf("# base\n\nuser %d\n", i),
|
||
ExpectedHeadSnapshot: &st.HeadSnapshotID, ExpectedRevisionNo: &st.HeadRevisionNo,
|
||
IdempotencyKey: fmt.Sprintf("pg-%s-%d", suffix, i), ActorUserID: 1,
|
||
LifecycleState: "drafting", ProjectID: projectID, WorkDir: "ws/" + docID,
|
||
},
|
||
})
|
||
if err == nil {
|
||
mu.Lock()
|
||
wins++
|
||
mu.Unlock()
|
||
}
|
||
}(i)
|
||
}
|
||
wg.Wait()
|
||
if wins != 1 {
|
||
t.Fatalf("same etag concurrent edits: wins=%d", wins)
|
||
}
|
||
got, err := rev.GetRevisionState(ctx, docID)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if got.HeadRevisionNo != 1 {
|
||
t.Fatalf("head revision=%d", got.HeadRevisionNo)
|
||
}
|
||
}
|
||
|
||
func refuseSharedDSN(t *testing.T, dsn string) {
|
||
t.Helper()
|
||
if strings.Contains(dsn, "5433") || strings.Contains(dsn, "agent_system_dev") ||
|
||
(strings.Contains(dsn, "5434") && strings.Contains(dsn, "agent_system_test")) ||
|
||
strings.Contains(dsn, "5435") {
|
||
t.Fatal("TEST_DATABASE_URL points at a shared env port; refuse agent_system_dev/test/main")
|
||
}
|
||
}
|
||
|
||
func TestDocumentCommitTwoPoolsAndFourSessionsPG(t *testing.T) {
|
||
if testing.Short() {
|
||
t.Skip("postgresql concurrency is not part of -short")
|
||
}
|
||
dsn := db.SkipIfNoTestDatabaseURL(t)
|
||
refuseSharedDSN(t, dsn)
|
||
ctx := context.Background()
|
||
poolA, err := pgxpool.New(ctx, dsn)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
t.Cleanup(poolA.Close)
|
||
poolB, err := pgxpool.New(ctx, dsn)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
t.Cleanup(poolB.Close)
|
||
if err := db.RunMigrationsOnceForTest(ctx, poolA, dsn); err != nil {
|
||
t.Fatalf("migrate: %v", err)
|
||
}
|
||
|
||
suffix := fmt.Sprintf("%d", time.Now().UnixNano())
|
||
var projectID int64
|
||
if err := poolA.QueryRow(ctx, `INSERT INTO projects (code, name, status) VALUES ($1,$2,'active') RETURNING id`,
|
||
"docflow2p-"+suffix, "docflow2p").Scan(&projectID); err != nil {
|
||
t.Fatalf("project: %v", err)
|
||
}
|
||
t.Cleanup(func() {
|
||
_, _ = poolA.Exec(context.Background(), `DELETE FROM document_change_events WHERE doc_id LIKE $1`, "docflow2p_"+suffix+"%")
|
||
_, _ = poolA.Exec(context.Background(), `DELETE FROM document_versions WHERE doc_id LIKE $1`, "docflow2p_"+suffix+"%")
|
||
_, _ = poolA.Exec(context.Background(), `DELETE FROM document_snapshots WHERE doc_id LIKE $1`, "docflow2p_"+suffix+"%")
|
||
_, _ = poolA.Exec(context.Background(), `DELETE FROM document_records WHERE project_id=$1`, projectID)
|
||
_, _ = poolA.Exec(context.Background(), `DELETE FROM projects WHERE id=$1`, projectID)
|
||
})
|
||
|
||
docID := "docflow2p_" + suffix
|
||
seed := docrevision.NewService(poolA)
|
||
gov := docgov.NewGovernanceEngine(poolA)
|
||
if _, err := gov.CreateOrGetDocument(ctx, docID, projectID, "intake", "t", "V0.5_feasibility", "ws/"+docID, 1, "A", 2, "R", 3, "P"); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
st, err := seed.OpenTemplateVersion(ctx, docID, "# base\n", docrevision.ActorSnapshot{}, projectID, "ws/"+docID)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
|
||
a := &docflow.Commands{UoW: uow.NewPG(poolA), Gov: docgov.NewGovernanceEngine(poolA), Rev: docrevision.NewService(poolA)}
|
||
b := &docflow.Commands{UoW: uow.NewPG(poolB), Gov: docgov.NewGovernanceEngine(poolB), Rev: docrevision.NewService(poolB)}
|
||
edit := func(cmds *docflow.Commands, i int) error {
|
||
_, err := cmds.HumanEdit(ctx, docflow.Principal{UserID: 1}, docflow.HumanEditCommand{
|
||
DocID: docID,
|
||
Request: docrevision.HumanRevisionRequest{
|
||
DocID: docID, Markdown: fmt.Sprintf("# base\n\npool %d\n", i),
|
||
ExpectedHeadSnapshot: &st.HeadSnapshotID, ExpectedRevisionNo: &st.HeadRevisionNo,
|
||
IdempotencyKey: fmt.Sprintf("2p-%s-%d", suffix, i), ActorUserID: 1,
|
||
LifecycleState: "drafting", ProjectID: projectID, WorkDir: "ws/" + docID,
|
||
},
|
||
})
|
||
return err
|
||
}
|
||
|
||
var wg sync.WaitGroup
|
||
errs := make([]error, 2)
|
||
wg.Add(2)
|
||
go func() { defer wg.Done(); errs[0] = edit(a, 0) }()
|
||
go func() { defer wg.Done(); errs[1] = edit(b, 1) }()
|
||
wg.Wait()
|
||
wins := 0
|
||
for _, err := range errs {
|
||
if err == nil {
|
||
wins++
|
||
}
|
||
}
|
||
if wins != 1 {
|
||
t.Fatalf("two pools same etag: wins=%d errs=%v", wins, errs)
|
||
}
|
||
|
||
// Four sessions: two pools × two sequential idempotent retries of the winner key must replay.
|
||
winnerKey := ""
|
||
if errs[0] == nil {
|
||
winnerKey = fmt.Sprintf("2p-%s-0", suffix)
|
||
} else {
|
||
winnerKey = fmt.Sprintf("2p-%s-1", suffix)
|
||
}
|
||
replay := docrevision.HumanRevisionRequest{
|
||
DocID: docID, Markdown: "# ignored-if-same-key\n",
|
||
IdempotencyKey: winnerKey, ActorUserID: 1, LifecycleState: "drafting",
|
||
ProjectID: projectID, WorkDir: "ws/" + docID,
|
||
}
|
||
first, err := a.HumanEdit(ctx, docflow.Principal{UserID: 1}, docflow.HumanEditCommand{DocID: docID, Request: replay})
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
second, err := b.HumanEdit(ctx, docflow.Principal{UserID: 1}, docflow.HumanEditCommand{DocID: docID, Request: replay})
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if first.State.HeadSnapshotID != second.State.HeadSnapshotID {
|
||
t.Fatalf("idempotent replay across pools diverged")
|
||
}
|
||
|
||
var extra sync.WaitGroup
|
||
extraWins := 0
|
||
var extraMu sync.Mutex
|
||
st2, err := seed.GetRevisionState(ctx, docID)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
for i := 0; i < 4; i++ {
|
||
extra.Add(1)
|
||
go func(i int) {
|
||
defer extra.Done()
|
||
cmds := a
|
||
if i%2 == 1 {
|
||
cmds = b
|
||
}
|
||
_, err := cmds.HumanEdit(ctx, docflow.Principal{UserID: 1}, docflow.HumanEditCommand{
|
||
DocID: docID,
|
||
Request: docrevision.HumanRevisionRequest{
|
||
DocID: docID, Markdown: fmt.Sprintf("# base\n\nfour %d\n", i),
|
||
ExpectedHeadSnapshot: &st2.HeadSnapshotID, ExpectedRevisionNo: &st2.HeadRevisionNo,
|
||
IdempotencyKey: fmt.Sprintf("4s-%s-%d", suffix, i), ActorUserID: 1,
|
||
LifecycleState: "drafting", ProjectID: projectID, WorkDir: "ws/" + docID,
|
||
},
|
||
})
|
||
if err == nil {
|
||
extraMu.Lock()
|
||
extraWins++
|
||
extraMu.Unlock()
|
||
}
|
||
}(i)
|
||
}
|
||
extra.Wait()
|
||
if extraWins != 1 {
|
||
t.Fatalf("four sessions same etag: wins=%d", extraWins)
|
||
}
|
||
}
|