weichen
3a08cd4260
Why: Both approved Backend slices share extension surfaces and migration numbers, and current main added account audit v31. Evidence: The isolated probe resolves additive interfaces and v31-v40 ordering; focused and full Go tests passed without PostgreSQL. Impact: Produces a local integration candidate for isolated PostgreSQL and race verification only; shipping branches remain unchanged. Refs: feature/rdm-two-trees-latest-templates W6 W9
611 lines
23 KiB
Go
611 lines
23 KiB
Go
package store
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"encoding/json"
|
|
"fmt"
|
|
"os"
|
|
"sort"
|
|
"strings"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
|
|
agentcommand "agent-command"
|
|
)
|
|
|
|
func openProjectAuditPG(t *testing.T) *Store {
|
|
t.Helper()
|
|
if os.Getenv("AGENT_PG_TEST_DSN") == "" {
|
|
t.Skip("AGENT_PG_TEST_DSN unset: isolated PostgreSQL check NOT_RUN")
|
|
}
|
|
st := openTestStore(t)
|
|
if st.Dialect() != "postgres" {
|
|
t.Fatal("expected PostgreSQL")
|
|
}
|
|
return st
|
|
}
|
|
|
|
func TestProjectAuditAllWritersRejectUnsafeOnlineChanges(t *testing.T) {
|
|
st := openProjectAuditPG(t)
|
|
if err := st.AppendAudit("alice", agentcommand.AuditEnvelope{Time: time.Now(), Actor: "alice", CommandID: "first", State: agentcommand.StateCommitted}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
row := `INSERT INTO command_audit(id,time,command_id,source,entrance,state,operation_id,conversation_id,owner,payload)
|
|
VALUES ($1,'2026-10-14T00:00:00Z','explicit','','','committed','','','alice','{}')`
|
|
if _, err := st.db.Exec(row, 100); err == nil {
|
|
t.Fatal("online explicit ID above reserved sequence position was accepted")
|
|
}
|
|
if _, err := st.db.Exec(`UPDATE command_audit SET command_id='changed' WHERE id=1`); err == nil {
|
|
t.Fatal("online audit UPDATE was accepted")
|
|
}
|
|
if _, err := st.db.Exec(`DELETE FROM command_audit WHERE id=1`); err == nil {
|
|
t.Fatal("online audit DELETE was accepted")
|
|
}
|
|
for _, level := range []sql.IsolationLevel{sql.LevelRepeatableRead, sql.LevelSerializable} {
|
|
tx, err := st.db.BeginTx(context.Background(), &sql.TxOptions{Isolation: level})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
_, err = tx.Exec(`INSERT INTO command_audit(time,command_id,source,entrance,state,operation_id,conversation_id,owner,payload)
|
|
VALUES ('2026-10-14T00:00:00Z','stale-snapshot','','','committed','','','alice','{}')`)
|
|
tx.Rollback()
|
|
if err == nil || !strings.Contains(err.Error(), "read committed") {
|
|
t.Fatalf("isolation %v did not fail closed: %v", level, err)
|
|
}
|
|
}
|
|
seq := commandAuditSequenceName(t, st)
|
|
if _, err := st.db.Exec(`SELECT nextval($1::regclass)`, seq); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, err := st.db.Exec(`INSERT INTO command_audit(id,time,command_id,source,entrance,state,operation_id,conversation_id,owner,payload)
|
|
VALUES (4,'2026-10-14T00:00:00Z','first-of-batch','','','committed','','','alice','{}'),
|
|
(1,'2026-10-14T00:00:00Z','decreasing','','','committed','','','alice','{}')`); err == nil {
|
|
t.Fatal("unordered multirow audit INSERT was accepted")
|
|
}
|
|
if _, err := st.db.Exec(row+` ON CONFLICT DO NOTHING`, 1); err == nil {
|
|
t.Fatal("online replay bypassed the INSERT guard through ON CONFLICT")
|
|
}
|
|
var count int
|
|
if err := st.db.QueryRow(`SELECT COUNT(*) FROM command_audit`).Scan(&count); err != nil || count != 1 {
|
|
t.Fatalf("unsafe writes persisted: count=%d err=%v", count, err)
|
|
}
|
|
}
|
|
|
|
func TestProjectAuditAllWritersGuardDriftFailsClosed(t *testing.T) {
|
|
st := openProjectAuditPG(t)
|
|
for _, trigger := range []string{"project_audit_insert_guard_v1", "project_audit_mutation_guard_v1"} {
|
|
if _, err := st.db.Exec(`ALTER TABLE command_audit DISABLE TRIGGER ` + trigger); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, _, _, err := st.ListProjectCommandAudits("alice", "project", 0, 0, 10); err == nil {
|
|
t.Fatalf("read accepted disabled trigger %s", trigger)
|
|
}
|
|
if err := st.AppendAudit("alice", agentcommand.AuditEnvelope{Time: time.Now(), Actor: "alice", CommandID: "drift"}); err == nil {
|
|
t.Fatalf("append accepted disabled trigger %s", trigger)
|
|
}
|
|
if _, err := st.db.Exec(`ALTER TABLE command_audit ENABLE ALWAYS TRIGGER ` + trigger); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
if _, err := st.db.Exec(`CREATE OR REPLACE TRIGGER project_audit_insert_guard_v1 BEFORE INSERT ON command_audit FOR EACH ROW EXECUTE FUNCTION project_audit_mutation_guard_v1()`); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, err := st.db.Exec(`ALTER TABLE command_audit ENABLE ALWAYS TRIGGER project_audit_insert_guard_v1`); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, _, _, err := st.ListProjectCommandAudits("alice", "project", 0, 0, 10); err == nil {
|
|
t.Fatal("read accepted rebound insert trigger")
|
|
}
|
|
if _, err := st.db.Exec(`CREATE OR REPLACE TRIGGER project_audit_insert_guard_v1 BEFORE INSERT ON command_audit FOR EACH ROW EXECUTE FUNCTION project_audit_insert_guard_v1()`); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, err := st.db.Exec(`ALTER TABLE command_audit ENABLE ALWAYS TRIGGER project_audit_insert_guard_v1`); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := st.AppendAudit("alice", agentcommand.AuditEnvelope{Time: time.Now(), Actor: "alice", CommandID: "restored"}); err != nil {
|
|
t.Fatalf("restored guard: %v", err)
|
|
}
|
|
}
|
|
|
|
func commandAuditSequenceName(t *testing.T, st *Store) string {
|
|
t.Helper()
|
|
var name string
|
|
if err := st.db.QueryRow(`SELECT pg_get_serial_sequence('command_audit','id')`).Scan(&name); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if name == "" {
|
|
t.Fatal("command_audit has no owned sequence")
|
|
}
|
|
return name
|
|
}
|
|
|
|
func TestProjectAuditSequenceMigrationRejectsCacheTen(t *testing.T) {
|
|
st := openProjectAuditPG(t)
|
|
if err := st.MigrateDown(38); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
seq := commandAuditSequenceName(t, st)
|
|
if _, err := st.db.Exec("ALTER SEQUENCE " + seq + " CACHE 10"); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := st.migrate(); err == nil {
|
|
t.Fatal("migration accepted CACHE 10, which permits out-of-order IDs")
|
|
}
|
|
if _, err := st.db.Exec("ALTER SEQUENCE " + seq + " CACHE 1"); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := st.migrate(); err != nil {
|
|
t.Fatalf("safe CACHE 1 migration: %v", err)
|
|
}
|
|
}
|
|
|
|
func TestProjectAuditSequenceRuntimeRejectsConfigurationDrift(t *testing.T) {
|
|
st := openProjectAuditPG(t)
|
|
seq := commandAuditSequenceName(t, st)
|
|
if _, err := st.db.Exec("ALTER SEQUENCE " + seq + " CACHE 10"); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, _, _, err := st.ListProjectCommandAudits("alice", "project", 0, 0, 10); err == nil {
|
|
t.Fatal("project audit reported a cutoff with CACHE 10")
|
|
}
|
|
if err := st.AppendAudit("alice", agentcommand.AuditEnvelope{Time: time.Now(), Actor: "alice", CommandID: "unsafe-cache"}); err == nil {
|
|
t.Fatal("append allocated an ID with CACHE 10")
|
|
}
|
|
if _, err := st.db.Exec("ALTER SEQUENCE " + seq + " CACHE 1"); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := st.AppendAudit("alice", agentcommand.AuditEnvelope{Time: time.Now(), Actor: "alice", CommandID: "safe-cache"}); err != nil {
|
|
t.Fatalf("safe append after reset: %v", err)
|
|
}
|
|
if _, err := st.db.Exec("ALTER SEQUENCE " + seq + " CYCLE"); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, _, _, err := st.ListProjectCommandAudits("alice", "project", 0, 0, 10); err == nil {
|
|
t.Fatal("read accepted cyclic audit IDs")
|
|
}
|
|
if err := st.AppendAudit("alice", agentcommand.AuditEnvelope{Time: time.Now(), Actor: "alice", CommandID: "cyclic"}); err == nil {
|
|
t.Fatal("append accepted cyclic audit IDs")
|
|
}
|
|
if _, err := st.db.Exec("ALTER SEQUENCE " + seq + " NO CYCLE"); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, err := st.db.Exec("ALTER SEQUENCE " + seq + " MINVALUE -10"); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, _, _, err := st.ListProjectCommandAudits("alice", "project", 0, 0, 10); err == nil {
|
|
t.Fatal("read accepted a nonpositive audit ID range")
|
|
}
|
|
if _, err := st.db.Exec("ALTER SEQUENCE " + seq + " MINVALUE 1"); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, err := st.db.Exec(`ALTER TABLE command_audit ALTER COLUMN id SET DEFAULT nextval(pg_get_serial_sequence('command_audit','id')::regclass)`); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, _, _, err := st.ListProjectCommandAudits("alice", "project", 0, 0, 10); err == nil {
|
|
t.Fatal("read accepted an unguarded sequence default")
|
|
}
|
|
if err := st.AppendAudit("alice", agentcommand.AuditEnvelope{Time: time.Now(), Actor: "alice", CommandID: "unguarded-default"}); err == nil {
|
|
t.Fatal("append accepted an unguarded sequence default")
|
|
}
|
|
if _, err := st.db.Exec(`ALTER TABLE command_audit ALTER COLUMN id SET DEFAULT CASE WHEN false THEN next_command_audit_id('command_audit'::regclass) ELSE nextval(pg_get_serial_sequence('command_audit','id')::regclass) END`); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, _, _, err := st.ListProjectCommandAudits("alice", "project", 0, 0, 10); err == nil {
|
|
t.Fatal("read accepted a replaced default that merely mentions the guarded function")
|
|
}
|
|
if _, err := st.db.Exec(`ALTER TABLE command_audit ALTER COLUMN id SET DEFAULT next_command_audit_id('command_audit'::regclass)`); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
|
|
func TestProjectAuditSequenceRejectsRewindIntoUnusedGap(t *testing.T) {
|
|
st := openProjectAuditPG(t)
|
|
p, err := st.CreateProject("alice", "A")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
c, err := st.CreateConversation("alice", string(p.ID), "work", "")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
appendAudit := func(id string) error {
|
|
return st.AppendAudit("alice", agentcommand.AuditEnvelope{Time: time.Now(), Actor: "alice", CommandID: id, ConversationID: string(c.ID), State: agentcommand.StateCommitted})
|
|
}
|
|
if err := appendAudit("first"); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
seq := commandAuditSequenceName(t, st)
|
|
if _, err := st.db.Exec("SELECT nextval('" + seq + "'::regclass)"); err != nil {
|
|
t.Fatal(err)
|
|
} // consume one ID without a row
|
|
if err := appendAudit("third"); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, err := st.db.Exec("SELECT setval('" + seq + "'::regclass,1)"); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := appendAudit("rewound-into-gap"); err == nil {
|
|
t.Fatal("rewound sequence inserted a low ID into a committed-prefix gap")
|
|
}
|
|
if _, _, _, err := st.ListProjectCommandAudits("alice", string(p.ID), 0, 0, 10); err == nil {
|
|
t.Fatal("read claimed a safe cutoff after sequence rewind")
|
|
}
|
|
if _, err := st.db.Exec("SELECT setval('" + seq + "'::regclass,3)"); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := appendAudit("recovered"); err != nil {
|
|
t.Fatalf("append after restoring sequence: %v", err)
|
|
}
|
|
}
|
|
|
|
// Advisory locking alone does not fix cached sequence allocation: session A
|
|
// can hold 2..10 while session B commits 11, then A later commits 2.
|
|
func TestProjectAuditSequenceCacheTenDefeatsAdvisoryLock(t *testing.T) {
|
|
st := openProjectAuditPG(t)
|
|
for _, query := range []string{
|
|
`CREATE TABLE cache_probe(id BIGSERIAL PRIMARY KEY)`,
|
|
`ALTER SEQUENCE cache_probe_id_seq CACHE 10`,
|
|
`CREATE FUNCTION cache_probe_next_id(target regclass) RETURNS bigint LANGUAGE plpgsql AS $$
|
|
BEGIN PERFORM pg_advisory_xact_lock(715328491,target::oid::int4);
|
|
RETURN nextval(pg_get_serial_sequence(target::text,'id')::regclass); END $$`,
|
|
`ALTER TABLE cache_probe ALTER COLUMN id SET DEFAULT cache_probe_next_id('cache_probe'::regclass)`,
|
|
} {
|
|
if _, err := st.db.Exec(query); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
ctx := context.Background()
|
|
a, err := st.db.Conn(ctx)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer a.Close()
|
|
b, err := st.db.Conn(ctx)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer b.Close()
|
|
var first, other, late int64
|
|
if err := a.QueryRowContext(ctx, `INSERT INTO cache_probe DEFAULT VALUES RETURNING id`).Scan(&first); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := b.QueryRowContext(ctx, `INSERT INTO cache_probe DEFAULT VALUES RETURNING id`).Scan(&other); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := a.QueryRowContext(ctx, `INSERT INTO cache_probe DEFAULT VALUES RETURNING id`).Scan(&late); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if first != 1 || other != 11 || late != 2 {
|
|
t.Fatalf("CACHE 10 counterexample changed: A=%d B=%d A-late=%d", first, other, late)
|
|
}
|
|
}
|
|
|
|
func TestProjectAuditSequenceNeverCommitsPreUpgradeCachedLowID(t *testing.T) {
|
|
st := openProjectAuditPG(t)
|
|
if err := st.MigrateDown(38); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
seq := commandAuditSequenceName(t, st)
|
|
if _, err := st.db.Exec("ALTER SEQUENCE " + seq + " CACHE 10"); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
ctx := context.Background()
|
|
a, err := st.db.Conn(ctx)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer a.Close()
|
|
b, err := st.db.Conn(ctx)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer b.Close()
|
|
insert := `INSERT INTO command_audit(time,command_id,source,entrance,state,operation_id,conversation_id,owner,payload)
|
|
VALUES ('2026-10-14T00:00:00Z','legacy','','','committed','','','alice','{}') RETURNING id`
|
|
var first, high, late int64
|
|
if err := a.QueryRowContext(ctx, insert).Scan(&first); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := b.QueryRowContext(ctx, insert).Scan(&high); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if first != 1 || high != 11 {
|
|
t.Fatalf("old cached allocation: first=%d high=%d", first, high)
|
|
}
|
|
if _, err := st.db.Exec("ALTER SEQUENCE " + seq + " CACHE 1"); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := st.migrate(); err != nil {
|
|
t.Fatalf("upgrade with idle cached connections: %v", err)
|
|
}
|
|
if err := a.QueryRowContext(ctx, insert).Scan(&late); err == nil && late <= high {
|
|
t.Fatalf("old connection committed late low ID %d below %d", late, high)
|
|
}
|
|
var count int
|
|
if err := st.db.QueryRow(`SELECT COUNT(*) FROM command_audit WHERE id=2`).Scan(&count); err != nil || count != 0 {
|
|
t.Fatalf("late cached row persisted: count=%d err=%v", count, err)
|
|
}
|
|
}
|
|
|
|
func TestProjectAuditSequenceMigrationRoundTripPreservesLinks(t *testing.T) {
|
|
st := openProjectAuditPG(t)
|
|
if err := st.EnsureUser("alice", "Alice"); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
p, err := st.CreateProject("alice", "A")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
c, err := st.CreateConversation("alice", string(p.ID), "work", "")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := st.AppendAudit("alice", agentcommand.AuditEnvelope{Time: time.Now(), Actor: "alice", CommandID: "rdm.confirm", OperationID: "op-1", ConversationID: string(c.ID), State: agentcommand.StateCommitted}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
rows, _, _, err := st.ListProjectCommandAudits("alice", string(p.ID), 0, 0, 10)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
offset := rows[0].Offset
|
|
link := ProjectAuditLink{HostAuditOffset: offset, ProjectID: string(p.ID), Owner: "alice", SourceKind: "domain-commit", SourceID: "commit-1", OperationID: "op-1", ResourceKind: "document", ResourceID: "doc-1", DisclosureClass: "metadata", IdempotencyKey: "outbox-1"}
|
|
if _, err := st.SaveProjectAuditLink(link); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := st.MigrateDown(38); err != nil {
|
|
t.Fatalf("down to v38: %v", err)
|
|
}
|
|
if _, _, _, err := st.ListProjectCommandAudits("alice", string(p.ID), 0, 0, 10); err == nil {
|
|
t.Fatal("old default exposed a strong cutoff after down")
|
|
}
|
|
if _, row, err := st.ProjectAuditLinkedReceipt(string(p.ID), "domain-commit", "commit-1", offset); err != nil || row.Offset != offset {
|
|
t.Fatalf("link changed during down: %+v %v", row, err)
|
|
}
|
|
if err := st.migrate(); err != nil {
|
|
t.Fatalf("up to v39: %v", err)
|
|
}
|
|
rows, _, _, err = st.ListProjectCommandAudits("alice", string(p.ID), 0, 0, 10)
|
|
if err != nil || len(rows) != 1 || rows[0].Offset != offset {
|
|
t.Fatalf("restored audit: %+v %v", rows, err)
|
|
}
|
|
if err := st.AppendAudit("alice", agentcommand.AuditEnvelope{Time: time.Now(), Actor: "alice", CommandID: "after-up", ConversationID: string(c.ID), State: agentcommand.StateCommitted}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
rows, _, _, err = st.ListProjectCommandAudits("alice", string(p.ID), 0, 0, 10)
|
|
if err != nil || len(rows) != 2 || rows[0].Offset <= offset {
|
|
t.Fatalf("sequence not advanced after up: %+v %v", rows, err)
|
|
}
|
|
}
|
|
|
|
func TestProjectAuditSequenceLockTimeoutAndRollback(t *testing.T) {
|
|
st := openProjectAuditPG(t)
|
|
p, err := st.CreateProject("alice", "A")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
c, err := st.CreateConversation("alice", string(p.ID), "work", "")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
tx, err := st.db.Begin()
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer tx.Rollback()
|
|
insert := st.q(`INSERT INTO command_audit(time,command_id,source,entrance,state,operation_id,conversation_id,owner,payload,actor_name,project_id)
|
|
VALUES (?,?,'','','committed','',?,?,?,?,?) RETURNING id`)
|
|
args := []any{time.Now().UTC().Format(time.RFC3339Nano), "held", string(c.ID), "alice", `{}`, "Alice", string(p.ID)}
|
|
var heldID int64
|
|
if err := tx.QueryRow(insert, args...).Scan(&heldID); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
ctx := context.Background()
|
|
conn, err := st.db.Conn(ctx)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer conn.Close()
|
|
if _, err := conn.ExecContext(ctx, `SET lock_timeout='150ms'`); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
var blockedID int64
|
|
err = conn.QueryRowContext(ctx, insert, args...).Scan(&blockedID)
|
|
if err == nil || !strings.Contains(err.Error(), "lock timeout") {
|
|
t.Fatalf("second writer did not stop on advisory lock: id=%d err=%v", blockedID, err)
|
|
}
|
|
if _, err := conn.ExecContext(ctx, `RESET lock_timeout`); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
cancelCtx, cancel := context.WithTimeout(ctx, 150*time.Millisecond)
|
|
defer cancel()
|
|
var canceledID int64
|
|
if err := conn.QueryRowContext(cancelCtx, insert, args...).Scan(&canceledID); err == nil {
|
|
t.Fatalf("canceled writer passed advisory lock: id=%d", canceledID)
|
|
}
|
|
if err := tx.Rollback(); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := st.AppendAudit("alice", agentcommand.AuditEnvelope{Time: time.Now(), Actor: "alice", CommandID: "after-rollback", ConversationID: string(c.ID), State: agentcommand.StateCommitted}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
rows, cutoff, _, err := st.ListProjectCommandAudits("alice", string(p.ID), 0, 0, 10)
|
|
if err != nil || len(rows) != 1 || rows[0].Offset <= heldID || cutoff != rows[0].Offset {
|
|
t.Fatalf("rollback gap affected cutoff: rows=%+v cutoff=%d held=%d err=%v", rows, cutoff, heldID, err)
|
|
}
|
|
}
|
|
|
|
func TestProjectAuditSequenceConcurrentLoad(t *testing.T) {
|
|
st := openProjectAuditPG(t)
|
|
var serverVersion string
|
|
if err := st.db.QueryRow(`SHOW server_version`).Scan(&serverVersion); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := st.EnsureUser("alice", "Alice"); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
p, err := st.CreateProject("alice", "A")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
c, err := st.CreateConversation("alice", string(p.ID), "work", "")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
const writers, each = 8, 24
|
|
run := func(label string, write func(agentcommand.AuditEnvelope) error) (time.Duration, time.Duration) {
|
|
t.Helper()
|
|
durations := make([]time.Duration, writers*each)
|
|
errors := make(chan error, writers*each)
|
|
start := time.Now()
|
|
var wg sync.WaitGroup
|
|
for worker := 0; worker < writers; worker++ {
|
|
wg.Add(1)
|
|
go func(worker int) {
|
|
defer wg.Done()
|
|
for i := 0; i < each; i++ {
|
|
env := agentcommand.AuditEnvelope{Time: time.Now(), Actor: "alice", CommandID: fmt.Sprintf("%s-%d-%d", label, worker, i), ConversationID: string(c.ID), State: agentcommand.StateCommitted}
|
|
at := time.Now()
|
|
if err := write(env); err != nil {
|
|
errors <- err
|
|
return
|
|
}
|
|
durations[worker*each+i] = time.Since(at)
|
|
}
|
|
}(worker)
|
|
}
|
|
wg.Wait()
|
|
close(errors)
|
|
for err := range errors {
|
|
t.Fatalf("%s append: %v", label, err)
|
|
}
|
|
elapsed := time.Since(start)
|
|
sort.Slice(durations, func(i, j int) bool { return durations[i] < durations[j] })
|
|
return elapsed, durations[len(durations)*95/100]
|
|
}
|
|
if err := st.MigrateDown(38); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
oldQuery := st.q(`INSERT INTO command_audit(time,command_id,source,entrance,state,operation_id,conversation_id,owner,payload,actor_name,project_id)
|
|
VALUES (?,?,?,?,?,?,?,?,?,COALESCE((SELECT name FROM users WHERE owner=?),''),COALESCE((SELECT project_id FROM conversations WHERE id=? AND owner=?),''))`)
|
|
oldElapsed, oldP95 := run("old", func(env agentcommand.AuditEnvelope) error {
|
|
raw, err := json.Marshal(env)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
_, err = st.db.Exec(oldQuery, env.Time.UTC().Format(time.RFC3339Nano), env.CommandID, "", "", string(env.State), "", env.ConversationID, "alice", string(raw), env.Actor, env.ConversationID, "alice")
|
|
return err
|
|
})
|
|
if err := st.migrate(); err != nil {
|
|
t.Fatalf("upgrade after baseline: %v", err)
|
|
}
|
|
statusMode := "MAX-first unlocked status"
|
|
if os.Getenv("AGENT_AUDIT_BENCH_READ_LOCK") == "1" {
|
|
// A test-only wrapper reconstructs the earlier status-read lock so the
|
|
// isolated load receipt can compare its cost against the final design.
|
|
if _, err := st.db.Exec(`ALTER FUNCTION project_audit_cutoff_safe(regclass) RENAME TO project_audit_cutoff_unlocked`); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, err := st.db.Exec(`CREATE FUNCTION project_audit_cutoff_safe(target regclass) RETURNS boolean LANGUAGE plpgsql VOLATILE AS $$
|
|
BEGIN PERFORM pg_advisory_xact_lock(715328491,target::oid::int4);
|
|
RETURN project_audit_cutoff_unlocked(target); END $$`); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
statusMode = "test-only locked status wrapper"
|
|
}
|
|
newElapsed, newP95 := run("guarded", func(env agentcommand.AuditEnvelope) error { return st.AppendAudit("alice", env) })
|
|
t.Logf("isolated PostgreSQL %s audit append: %d writes per mode / %d workers; v38 direct INSERT=%s p95=%s (%.0f/s), v40 Store.AppendAudit with %s=%s p95=%s (%.0f/s); hardware/load dependent SELF only",
|
|
serverVersion, writers*each, writers, oldElapsed, oldP95, float64(writers*each)/oldElapsed.Seconds(), statusMode, newElapsed, newP95, float64(writers*each)/newElapsed.Seconds())
|
|
seen := map[string]bool{}
|
|
var before, cutoff int64
|
|
for {
|
|
page, fixed, next, err := st.ListProjectCommandAudits("alice", string(p.ID), before, cutoff, 100)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if cutoff == 0 {
|
|
cutoff = fixed
|
|
}
|
|
for _, row := range page {
|
|
if seen[row.Envelope.CommandID] {
|
|
t.Fatalf("duplicate paged command: %s", row.Envelope.CommandID)
|
|
}
|
|
seen[row.Envelope.CommandID] = true
|
|
}
|
|
if next == 0 {
|
|
break
|
|
}
|
|
before = next
|
|
}
|
|
if len(seen) != 2*writers*each {
|
|
t.Fatalf("paged commands=%d want=%d", len(seen), 2*writers*each)
|
|
}
|
|
}
|
|
|
|
func TestProjectAuditConcurrentReadWriteNeverReportsFalseDrift(t *testing.T) {
|
|
st := openProjectAuditPG(t)
|
|
p, err := st.CreateProject("alice", "A")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
c, err := st.CreateConversation("alice", string(p.ID), "work", "")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
for i := 0; i < 210; i++ {
|
|
if err := st.AppendAudit("alice", agentcommand.AuditEnvelope{Time: time.Now(), Actor: "alice", CommandID: fmt.Sprintf("seed-%d", i), ConversationID: string(c.ID), State: agentcommand.StateCommitted}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
const writers, writes, readers, reads = 6, 30, 4, 40
|
|
errs := make(chan error, writers*writes+readers*reads)
|
|
readDurations := make(chan time.Duration, readers*reads)
|
|
var wg sync.WaitGroup
|
|
for worker := 0; worker < writers; worker++ {
|
|
wg.Add(1)
|
|
go func(worker int) {
|
|
defer wg.Done()
|
|
for i := 0; i < writes; i++ {
|
|
err := st.AppendAudit("alice", agentcommand.AuditEnvelope{Time: time.Now(), Actor: "alice", CommandID: fmt.Sprintf("write-%d-%d", worker, i), ConversationID: string(c.ID), State: agentcommand.StateCommitted})
|
|
if err != nil {
|
|
errs <- fmt.Errorf("append: %w", err)
|
|
return
|
|
}
|
|
}
|
|
}(worker)
|
|
}
|
|
for reader := 0; reader < readers; reader++ {
|
|
wg.Add(1)
|
|
go func() {
|
|
defer wg.Done()
|
|
for i := 0; i < reads; i++ {
|
|
started := time.Now()
|
|
page, _, _, err := st.ListProjectCommandAudits("alice", string(p.ID), 0, 0, 200)
|
|
if err != nil {
|
|
errs <- fmt.Errorf("read: %w", err)
|
|
return
|
|
}
|
|
if len(page) != 200 {
|
|
errs <- fmt.Errorf("large page rows=%d want=200", len(page))
|
|
return
|
|
}
|
|
readDurations <- time.Since(started)
|
|
}
|
|
}()
|
|
}
|
|
wg.Wait()
|
|
close(errs)
|
|
for err := range errs {
|
|
t.Fatal(err)
|
|
}
|
|
close(readDurations)
|
|
var samples []time.Duration
|
|
for duration := range readDurations {
|
|
samples = append(samples, duration)
|
|
}
|
|
sort.Slice(samples, func(i, j int) bool { return samples[i] < samples[j] })
|
|
t.Logf("isolated PG mixed load: %d append writes / %d writers and %d 200-row pages / %d readers; page p95=%s, SELF cost signal only",
|
|
writers*writes, writers, readers*reads, readers, samples[len(samples)*95/100])
|
|
}
|