test(canvas): verify canonical concurrency and late migration rollback
This commit is contained in:
@@ -0,0 +1,300 @@
|
||||
package platform
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"crypto/rand"
|
||||
"crypto/sha256"
|
||||
"database/sql"
|
||||
"encoding/base64"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"os"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
type canonicalAtomicFixture struct {
|
||||
db *sql.DB
|
||||
mux *http.ServeMux
|
||||
project, account, workspace, root, token string
|
||||
legacy json.RawMessage
|
||||
}
|
||||
|
||||
func newCanonicalAtomicFixture(t *testing.T) *canonicalAtomicFixture {
|
||||
t.Helper()
|
||||
dsn := os.Getenv("PLATFORM_TEST_DATABASE_URL")
|
||||
if dsn == "" {
|
||||
t.Skip("requires an isolated integration database")
|
||||
}
|
||||
db, err := sql.Open("pgx", dsn)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() { db.Close() })
|
||||
var entropy [12]byte
|
||||
_, _ = rand.Read(entropy[:])
|
||||
f := &canonicalAtomicFixture{db: db, project: "canonical-atomic-" + hex.EncodeToString(entropy[:])}
|
||||
ctx := context.Background()
|
||||
t.Cleanup(func() {
|
||||
for _, table := range []string{"canvas_model_operations", "canvas_model_identity_map", "canvas_model_requests", "canvas_model_projects", "canvas_asset_bindings", "canvas_subgraphs", "workspaces", "project_accounts"} {
|
||||
_, _ = db.ExecContext(ctx, "DELETE FROM "+table+" WHERE project=$1", f.project)
|
||||
}
|
||||
})
|
||||
if err = db.QueryRowContext(ctx, `INSERT INTO project_accounts(project,id) VALUES($1,gen_random_uuid()) RETURNING id::text`, f.project).Scan(&f.account); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var tokenBytes [32]byte
|
||||
_, _ = rand.Read(tokenBytes[:])
|
||||
f.token = base64.RawURLEncoding.EncodeToString(tokenBytes[:])
|
||||
hash := sha256.Sum256([]byte(f.token))
|
||||
if _, err = db.ExecContext(ctx, `INSERT INTO project_sessions(token_hash,project,account_id,expires_at) VALUES($1,$2,$3,now()+interval '1 hour')`, base64.RawURLEncoding.EncodeToString(hash[:]), f.project, f.account); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err = db.QueryRowContext(ctx, `INSERT INTO workspaces(project,name,owner_id) VALUES($1,'Canonical atomic',$2) RETURNING id::text`, f.project, f.account).Scan(&f.workspace); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err = db.ExecContext(ctx, `INSERT INTO workspace_members(project,workspace_id,account_id,role) VALUES($1,$2,$3,'admin')`, f.project, f.workspace, f.account); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
f.legacy = json.RawMessage(`{"schema_version":1,"nodes":[` + modelLegacyFrame + `,{"id":"a","type":"plainText","position":{"x":150,"y":260},"width":240,"height":72,"data":{"kind":"plain-text","title":"a","text":"preserved","frameId":"f"}},{"id":"b","type":"plainText","position":{"x":180,"y":300},"width":240,"height":72,"data":{"kind":"plain-text","title":"b","text":"preserved","frameId":"f"}},{"id":"c","type":"plainText","position":{"x":0,"y":0},"width":240,"height":72,"data":{"kind":"plain-text","title":"c","text":"unrelated"}}],"edges":[{"id":"e","source":"a","target":"b"}],"viewport":{"x":0,"y":0,"zoom":1}}`)
|
||||
if err = db.QueryRowContext(ctx, `INSERT INTO materials(project,workspace_id,kind,title,created_by,canvas_document) VALUES($1,$2,'board','Root',$3,$4) RETURNING id::text`, f.project, f.workspace, f.account, f.legacy).Scan(&f.root); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
f.mux = http.NewServeMux()
|
||||
Register(f.mux, db, map[string]string{f.project: "https://app.example.com/auth/callback"}, "https://api.example.com", nil)
|
||||
return f
|
||||
}
|
||||
|
||||
func (f *canonicalAtomicFixture) path() string {
|
||||
return "/projects/" + f.project + "/workspaces/" + f.workspace + "/materials/" + f.root + "/canvas/model"
|
||||
}
|
||||
|
||||
func (f *canonicalAtomicFixture) request(ctx context.Context, method, path, key string, raw []byte) *httptest.ResponseRecorder {
|
||||
r := httptest.NewRequest(method, path, bytes.NewReader(raw)).WithContext(ctx)
|
||||
r.Header.Set("Authorization", "Bearer "+f.token)
|
||||
r.Header.Set("X-Expected-Account", f.account)
|
||||
r.Header.Set("Accept", "application/vnd.foundry.canvas+json;version=2")
|
||||
if key != "" {
|
||||
r.Header.Set("Idempotency-Key", key)
|
||||
}
|
||||
w := httptest.NewRecorder()
|
||||
f.mux.ServeHTTP(w, r)
|
||||
return w
|
||||
}
|
||||
|
||||
func canonicalAtomicResponse(t *testing.T, w *httptest.ResponseRecorder, status int) response {
|
||||
t.Helper()
|
||||
if w.Code != status {
|
||||
t.Fatalf("canonical response got %d %s want %d", w.Code, w.Body.String(), status)
|
||||
}
|
||||
var body response
|
||||
if err := json.Unmarshal(w.Body.Bytes(), &body); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return body
|
||||
}
|
||||
|
||||
func (f *canonicalAtomicFixture) read(t *testing.T) canvasModel {
|
||||
t.Helper()
|
||||
tx, err := f.db.BeginTx(context.Background(), nil)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer tx.Rollback()
|
||||
model, err := loadCanvasModel(context.Background(), tx, f.project, f.workspace, f.root)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return model
|
||||
}
|
||||
|
||||
func (f *canonicalAtomicFixture) concurrent(t *testing.T, path string, keys []string, bodies [][]byte) []*httptest.ResponseRecorder {
|
||||
t.Helper()
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
|
||||
defer cancel()
|
||||
type result struct {
|
||||
index int
|
||||
writer *httptest.ResponseRecorder
|
||||
}
|
||||
start, done := make(chan struct{}), make(chan result, len(keys))
|
||||
for i := range keys {
|
||||
go func(index int) {
|
||||
<-start
|
||||
done <- result{index, f.request(ctx, "POST", path, keys[index], bodies[index])}
|
||||
}(i)
|
||||
}
|
||||
close(start)
|
||||
out := make([]*httptest.ResponseRecorder, len(keys))
|
||||
for range keys {
|
||||
select {
|
||||
case value := <-done:
|
||||
out[value.index] = value.writer
|
||||
case <-ctx.Done():
|
||||
t.Fatal("canonical concurrent requests did not finish")
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func TestCanonicalConcurrentMigrationAndOwnershipWrites(t *testing.T) {
|
||||
f := newCanonicalAtomicFixture(t)
|
||||
ctx := context.Background()
|
||||
key := modelTestID(8000)
|
||||
input := response{"protocol_version": 2, "request_id": key, "expected_versions": map[string]int64{f.root: 1}}
|
||||
raw := modelJSON(input)
|
||||
replies := f.concurrent(t, f.path()+"/migrate", []string{key, key, key, key}, [][]byte{raw, raw, raw, raw})
|
||||
first := canonicalAtomicResponse(t, replies[0], 200)
|
||||
for _, reply := range replies[1:] {
|
||||
if !modelEqual(first, canonicalAtomicResponse(t, reply, 200)) {
|
||||
t.Fatal("concurrent migration allocated different identities or receipts")
|
||||
}
|
||||
}
|
||||
ids := map[string]string{}
|
||||
for _, value := range first["identity_map"].([]any) {
|
||||
mapping := value.(map[string]any)
|
||||
ids[mapping["sourceId"].(string)] = mapping["elementId"].(string)
|
||||
}
|
||||
before := f.read(t)
|
||||
child := modelString(before.Elements[ids["a"]]["ownerCanvasId"])
|
||||
if len(before.Canvases) != 2 || len(ids) != 5 || child == f.root || before.Versions[f.root] != 2 || before.Versions[child] != 1 {
|
||||
t.Fatal("concurrent migration duplicated the implicit partition or version barrier")
|
||||
}
|
||||
key = modelTestID(8001)
|
||||
edit := modelBatch{ProtocolVersion: 2, RequestID: key, Expected: map[string]int64{child: 1}, Operations: []modelOperation{{ID: modelTestID(8002), Changes: []modelChange{{Kind: "field", Target: modelTarget{ID: ids["a"], Incarnation: 1}, Field: "data.title", Before: modelJSON("a"), After: modelJSON("renamed")}}}}}
|
||||
raw = modelJSON(edit)
|
||||
replies = f.concurrent(t, f.path()+"/operations", []string{key, key, key, key}, [][]byte{raw, raw, raw, raw})
|
||||
first = canonicalAtomicResponse(t, replies[0], 200)
|
||||
for _, reply := range replies[1:] {
|
||||
if !modelEqual(first, canonicalAtomicResponse(t, reply, 200)) {
|
||||
t.Fatal("concurrent edit retry changed its original receipt")
|
||||
}
|
||||
}
|
||||
before = f.read(t)
|
||||
if before.Versions[child] != 2 || before.Versions[f.root] != 2 || modelString(before.Elements[ids["a"]]["data"].(map[string]any)["title"]) != "renamed" {
|
||||
t.Fatal("concurrent edit executed more than once")
|
||||
}
|
||||
var batches []modelBatch
|
||||
var keys []string
|
||||
var bodies [][]byte
|
||||
for index, node := range []string{ids["a"], ids["b"]} {
|
||||
key := modelTestID(8100 + index)
|
||||
batch := modelBatch{ProtocolVersion: 2, RequestID: key, Expected: map[string]int64{f.root: 2, child: 2}, Operations: []modelOperation{{ID: modelTestID(8200 + index), Changes: []modelChange{
|
||||
{Kind: "field", Target: modelTarget{ID: node, Incarnation: 1}, Field: "ownerCanvasId", Before: modelJSON(child), After: modelJSON(f.root)},
|
||||
{Kind: "field", Target: modelTarget{ID: ids["e"], Incarnation: 1}, Field: "ownerCanvasId", Before: modelJSON(child), After: modelJSON(f.root)},
|
||||
}}}}
|
||||
batches, keys, bodies = append(batches, batch), append(keys, key), append(bodies, modelJSON(batch))
|
||||
}
|
||||
replies = f.concurrent(t, f.path()+"/operations", keys, bodies)
|
||||
winner, successes := -1, 0
|
||||
for index, reply := range replies {
|
||||
status := 412
|
||||
if reply.Code == 200 {
|
||||
winner, successes, status = index, successes+1, 200
|
||||
}
|
||||
body := canonicalAtomicResponse(t, reply, status)
|
||||
queried := canonicalAtomicResponse(t, f.request(ctx, "GET", f.path()+"/operations/"+keys[index], "", nil), 200)
|
||||
if !modelEqual(body, queried) || status == 412 && (body["status"] != "not-applied" || body["terminal"] != true || body["error"] != "version_conflict") {
|
||||
t.Fatal("a competing write lost its queryable applied or zero-effect receipt")
|
||||
}
|
||||
}
|
||||
if successes != 1 {
|
||||
t.Fatal("competing conditional ownership writes did not have exactly one winner")
|
||||
}
|
||||
after := f.read(t)
|
||||
if err := validateCanvasModel(after); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
nodes := []string{ids["a"], ids["b"]}
|
||||
if after.Versions[f.root] != 3 || after.Versions[child] != 3 || modelString(after.Elements[nodes[winner]]["ownerCanvasId"]) != f.root || modelString(after.Elements[nodes[1-winner]]["ownerCanvasId"]) != child || modelString(after.Elements[ids["e"]]["ownerCanvasId"]) != f.root || !modelEqual(before.Elements[ids["c"]], after.Elements[ids["c"]]) || !modelEqual(before.Canvases, after.Canvases) {
|
||||
t.Fatal("competing ownership writes published a partial node/edge/tree or advanced versions twice")
|
||||
}
|
||||
var count int
|
||||
if err := f.db.QueryRowContext(ctx, `SELECT count(*) FROM canvas_model_operations WHERE project=$1`, f.project).Scan(&count); err != nil || count != 2 {
|
||||
t.Fatalf("logical operations: %d %v", count, err)
|
||||
}
|
||||
changed := batches[winner]
|
||||
changed.Operations[0].ID = modelTestID(8999)
|
||||
body := canonicalAtomicResponse(t, f.request(ctx, "POST", f.path()+"/operations", keys[winner], modelJSON(changed)), 409)
|
||||
if body["error"] != "idempotency_conflict" || !modelEqual(after, f.read(t)) {
|
||||
t.Fatal("same key with different bytes changed the canonical model")
|
||||
}
|
||||
}
|
||||
|
||||
func TestCanonicalMigrationLateCancellationRollsBack(t *testing.T) {
|
||||
f := newCanonicalAtomicFixture(t)
|
||||
ctx, stop := context.WithTimeout(context.Background(), 20*time.Second)
|
||||
defer stop()
|
||||
barrier, err := f.db.BeginTx(ctx, nil)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer barrier.Rollback()
|
||||
if _, err = barrier.ExecContext(ctx, `LOCK TABLE canvas_model_identity_map IN ACCESS EXCLUSIVE MODE`); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var blocker int
|
||||
if err = barrier.QueryRowContext(ctx, `SELECT pg_backend_pid()`).Scan(&blocker); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
key := modelTestID(8300)
|
||||
input := response{"protocol_version": 2, "request_id": key, "expected_versions": map[string]int64{f.root: 1}}
|
||||
requestContext, cancel := context.WithCancel(ctx)
|
||||
defer cancel()
|
||||
done := make(chan *httptest.ResponseRecorder, 1)
|
||||
go func() { done <- f.request(requestContext, "POST", f.path()+"/migrate", key, modelJSON(input)) }()
|
||||
// Observe the actual late INSERT blocked behind this transaction. A timer
|
||||
// alone cannot prove migration reached canonical/material/subgraph writes.
|
||||
ticker := time.NewTicker(10 * time.Millisecond)
|
||||
defer ticker.Stop()
|
||||
observed := false
|
||||
for !observed {
|
||||
if err = f.db.QueryRowContext(ctx, `SELECT EXISTS(SELECT 1 FROM pg_stat_activity WHERE usename=current_user AND query LIKE 'INSERT INTO canvas_model_identity_map%' AND $1=ANY(pg_blocking_pids(pid)))`, blocker).Scan(&observed); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if observed {
|
||||
break
|
||||
}
|
||||
select {
|
||||
case reply := <-done:
|
||||
t.Fatalf("migration did not reach its late write: %d %s", reply.Code, reply.Body.String())
|
||||
case <-ctx.Done():
|
||||
t.Fatal("migration never reached its late identity-map INSERT")
|
||||
case <-ticker.C:
|
||||
}
|
||||
}
|
||||
cancel()
|
||||
select {
|
||||
case reply := <-done:
|
||||
canonicalAtomicResponse(t, reply, 503)
|
||||
case <-ctx.Done():
|
||||
t.Fatal("cancelled migration did not finish")
|
||||
}
|
||||
if err = barrier.Rollback(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for _, table := range []string{"canvas_model_projects", "canvas_model_canvases", "canvas_model_elements", "canvas_model_identity_map", "canvas_model_requests", "canvas_model_operations", "canvas_subgraphs"} {
|
||||
var count int
|
||||
if err = f.db.QueryRowContext(ctx, "SELECT count(*) FROM "+table+" WHERE project=$1", f.project).Scan(&count); err != nil || count != 0 {
|
||||
t.Fatalf("late cancellation left %s rows: %d %v", table, count, err)
|
||||
}
|
||||
}
|
||||
var count int
|
||||
var version int64
|
||||
var source json.RawMessage
|
||||
if err = f.db.QueryRowContext(ctx, `SELECT count(*) FROM materials WHERE project=$1`, f.project).Scan(&count); err != nil || count != 1 {
|
||||
t.Fatalf("synthetic partition material survived rollback: %d %v", count, err)
|
||||
}
|
||||
if err = f.db.QueryRowContext(ctx, `SELECT canvas_version,canvas_document FROM materials WHERE project=$1 AND id=$2`, f.project, f.root).Scan(&version, &source); err != nil || version != 1 || !modelEqual(f.legacy, source) {
|
||||
t.Fatalf("late cancellation changed legacy source/version: %d %v", version, err)
|
||||
}
|
||||
// Infrastructure failure saved no result. The same original request can be
|
||||
// retried safely, and all subsequent retries must return its one identity map.
|
||||
first := canonicalAtomicResponse(t, f.request(ctx, "POST", f.path()+"/migrate", key, modelJSON(input)), 200)
|
||||
retry := canonicalAtomicResponse(t, f.request(ctx, "POST", f.path()+"/migrate", key, modelJSON(input)), 200)
|
||||
if !modelEqual(first, retry) || len(f.read(t).Canvases) != 2 {
|
||||
t.Fatal("retry after rollback duplicated identities or the implicit partition")
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user