weichenandCursor
f34bec40f5
Cross-pool idempotency is same key plus same digest; a stale in-memory head from the seeder hid the one-winner four-session lock. Co-authored-by: Cursor <cursoragent@cursor.com>
249 lines
8.3 KiB
Go
249 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)
|
|
}
|
|
|
|
// Same idempotency key + same digest must replay across pools.
|
|
// A different digest with the same key is ErrIdempotencyConflict (covered in unit tests).
|
|
winnerIdx := 1
|
|
if errs[0] == nil {
|
|
winnerIdx = 0
|
|
}
|
|
winnerKey := fmt.Sprintf("2p-%s-%d", suffix, winnerIdx)
|
|
replay := docrevision.HumanRevisionRequest{
|
|
DocID: docID, Markdown: fmt.Sprintf("# base\n\npool %d\n", winnerIdx),
|
|
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
|
|
fresh := docrevision.NewService(poolA)
|
|
st2, err := fresh.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)
|
|
}
|
|
}
|