test(docnotify): prove notify shadow, live uniqueness, and rollback on isolated PG
Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
weichen
@@ -0,0 +1,349 @@
|
||||
package docnotify_test
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"os"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"chat-agent-system/internal/app/docnotify"
|
||||
"chat-agent-system/internal/app/uow"
|
||||
"chat-agent-system/internal/db"
|
||||
"chat-agent-system/internal/docgov"
|
||||
"chat-agent-system/internal/events"
|
||||
"chat-agent-system/internal/notification"
|
||||
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
)
|
||||
|
||||
func init() {
|
||||
if _, err := os.Stat("deploy/postgres/goose"); err == nil {
|
||||
return
|
||||
}
|
||||
for _, cand := range []string{"../../..", "../../../.."} {
|
||||
if fi, err := os.Stat(cand + "/deploy/postgres/goose"); err == nil && fi.IsDir() {
|
||||
_ = os.Chdir(cand)
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestIntendedNotifyDigestStable(t *testing.T) {
|
||||
a := docnotify.IntendedNotifyDigest("e1", notification.TypeDocApproved, 9)
|
||||
b := docnotify.IntendedNotifyDigest("e1", notification.TypeDocApproved, 9)
|
||||
if a == "" || a != b {
|
||||
t.Fatalf("%s %s", a, b)
|
||||
}
|
||||
want := events.VisibleOutcomeDigest("e1", notification.TypeDocApproved, 9)
|
||||
if a != want {
|
||||
t.Fatalf("normalizer drift: %s vs %s", a, want)
|
||||
}
|
||||
}
|
||||
|
||||
func TestE1A_NotifyShadowLiveRollback_PG(t *testing.T) {
|
||||
pool, ctx := e1aPool(t)
|
||||
suffix := fmt.Sprintf("%d", time.Now().UnixNano())
|
||||
userID, projectID := seedUserProject(t, ctx, pool, suffix)
|
||||
docID := "e1a_" + suffix
|
||||
seedDoc(t, ctx, pool, docID, projectID, userID)
|
||||
|
||||
env := approvedEnv(suffix, docID, projectID, 1)
|
||||
appendEnv(t, ctx, pool, env, []string{events.ConsumerNotify, events.ConsumerWorkflowAdvance})
|
||||
|
||||
d := newNotifyDispatcher(t, pool, "e1a-w")
|
||||
if err := events.SetMode(ctx, pool, "dev", events.ConsumerNotify, events.ModeShadow); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
_, revShadow, err := events.LoadMode(ctx, pool, "dev", events.ConsumerNotify)
|
||||
if err != nil || revShadow < 1 {
|
||||
t.Fatalf("shadow revision=%d err=%v", revShadow, err)
|
||||
}
|
||||
if n, err := d.RunOnce(ctx, events.ConsumerNotify); err != nil || n < 1 {
|
||||
t.Fatalf("shadow n=%d err=%v", n, err)
|
||||
}
|
||||
var notes, recon int
|
||||
if err := pool.QueryRow(ctx, `SELECT COUNT(*) FROM notifications WHERE source_event_id=$1`, env.EventID).Scan(¬es); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if notes != 0 {
|
||||
t.Fatalf("shadow wrote notifications=%d", notes)
|
||||
}
|
||||
var intended string
|
||||
if err := pool.QueryRow(ctx, `SELECT COUNT(*), MAX(intended_digest) FROM event_reconciliations WHERE event_id=$1 AND mode='shadow'`, env.EventID).Scan(&recon, &intended); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if recon < 1 {
|
||||
t.Fatal("missing shadow recon")
|
||||
}
|
||||
want := docnotify.IntendedNotifyDigest(env.EventID, notification.TypeDocApproved, userID)
|
||||
if intended != want {
|
||||
t.Fatalf("intended digest %s want %s", intended, want)
|
||||
}
|
||||
|
||||
if err := events.SetMode(ctx, pool, "dev", events.ConsumerNotify, events.ModeLive); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
_, revLive, err := events.LoadMode(ctx, pool, "dev", events.ConsumerNotify)
|
||||
if err != nil || revLive <= revShadow {
|
||||
t.Fatalf("live revision %d must exceed shadow %d", revLive, revShadow)
|
||||
}
|
||||
if _, err := d.RunOnce(ctx, events.ConsumerNotify); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := d.RunOnce(ctx, events.ConsumerNotify); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := pool.QueryRow(ctx, `SELECT COUNT(*) FROM notifications WHERE source_event_id=$1 AND user_id=$2 AND type=$3`,
|
||||
env.EventID, userID, notification.TypeDocApproved).Scan(¬es); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if notes != 1 {
|
||||
t.Fatalf("live visible notifications=%d", notes)
|
||||
}
|
||||
liveHash, err := hashCommitted(ctx, pool, env.EventID)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if liveHash != intended {
|
||||
t.Fatalf("live hash %s != shadow intended %s", liveHash, intended)
|
||||
}
|
||||
|
||||
var attempts, receipts int
|
||||
if err := pool.QueryRow(ctx, `
|
||||
SELECT
|
||||
(SELECT COUNT(*) FROM event_delivery_attempts WHERE event_id=$1 AND consumer=$2),
|
||||
(SELECT COUNT(*) FROM notifications WHERE source_event_id=$1)
|
||||
`, env.EventID, events.ConsumerNotify).Scan(&attempts, &receipts); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if attempts < 1 || receipts != 1 {
|
||||
t.Fatalf("correlation attempts=%d receipts=%d", attempts, receipts)
|
||||
}
|
||||
|
||||
if err := events.SetMode(ctx, pool, "dev", events.ConsumerNotify, events.ModeOff); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
n, err := d.RunOnce(ctx, events.ConsumerNotify)
|
||||
if err != nil || n != 0 {
|
||||
t.Fatalf("rollback off claimed=%d err=%v", n, err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestE1A_NotifyFixtureThirtyAuthzAndRollbackTx_PG(t *testing.T) {
|
||||
pool, ctx := e1aPool(t)
|
||||
suffix := fmt.Sprintf("%d", time.Now().UnixNano())
|
||||
userID, projectID := seedUserProject(t, ctx, pool, suffix)
|
||||
d := newNotifyDispatcher(t, pool, "e1a-n30")
|
||||
d.Batch = 64
|
||||
|
||||
const nFix = 30
|
||||
ids := make([]string, 0, nFix)
|
||||
for i := 0; i < nFix; i++ {
|
||||
docID := fmt.Sprintf("e1a30_%s_%d", suffix, i)
|
||||
seedDoc(t, ctx, pool, docID, projectID, userID)
|
||||
env := approvedEnv(fmt.Sprintf("%s-%d", suffix, i), docID, projectID, int64(i+1))
|
||||
appendEnv(t, ctx, pool, env, []string{events.ConsumerNotify})
|
||||
ids = append(ids, env.EventID)
|
||||
}
|
||||
|
||||
if err := events.SetMode(ctx, pool, "dev", events.ConsumerNotify, events.ModeShadow); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var scanned int
|
||||
for i := 0; i < 8; i++ {
|
||||
n, err := d.RunOnce(ctx, events.ConsumerNotify)
|
||||
if err != nil {
|
||||
t.Fatalf("shadow err=%v", err)
|
||||
}
|
||||
scanned += n
|
||||
var reconNow int
|
||||
if err := pool.QueryRow(ctx, `SELECT COUNT(*) FROM event_reconciliations WHERE mode='shadow' AND event_id LIKE $1`, "%"+suffix+"%").Scan(&reconNow); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if reconNow >= nFix {
|
||||
break
|
||||
}
|
||||
if n == 0 {
|
||||
t.Fatalf("shadow stalled at recon=%d scanned=%d", reconNow, scanned)
|
||||
}
|
||||
}
|
||||
var shadowNotes int
|
||||
if err := pool.QueryRow(ctx, `SELECT COUNT(*) FROM notifications WHERE source_event_id LIKE $1`, "%"+suffix+"%").Scan(&shadowNotes); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if shadowNotes != 0 {
|
||||
t.Fatalf("fixture shadow wrote %d notifications", shadowNotes)
|
||||
}
|
||||
var recon int
|
||||
if err := pool.QueryRow(ctx, `SELECT COUNT(*) FROM event_reconciliations WHERE mode='shadow' AND event_id LIKE $1`, "%"+suffix+"%").Scan(&recon); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if recon < nFix {
|
||||
t.Fatalf("fixture shadow recon=%d want >=%d", recon, nFix)
|
||||
}
|
||||
for _, id := range ids {
|
||||
want := docnotify.IntendedNotifyDigest(id, notification.TypeDocApproved, userID)
|
||||
var got string
|
||||
if err := pool.QueryRow(ctx, `SELECT intended_digest FROM event_reconciliations WHERE event_id=$1 AND mode='shadow'`, id).Scan(&got); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if got != want {
|
||||
t.Fatalf("event %s intended %s want %s", id, got, want)
|
||||
}
|
||||
}
|
||||
|
||||
d.Batch = 64
|
||||
if err := events.SetMode(ctx, pool, "dev", events.ConsumerNotify, events.ModeLive); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for i := 0; i < 8; i++ {
|
||||
n, err := d.RunOnce(ctx, events.ConsumerNotify)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var liveNow int
|
||||
if err := pool.QueryRow(ctx, `SELECT COUNT(*) FROM notifications WHERE source_event_id LIKE $1 AND type=$2`, "%"+suffix+"%", notification.TypeDocApproved).Scan(&liveNow); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if liveNow >= nFix {
|
||||
break
|
||||
}
|
||||
if n == 0 && i > 0 {
|
||||
t.Fatalf("live stalled at notifications=%d", liveNow)
|
||||
}
|
||||
}
|
||||
var liveNotes int
|
||||
if err := pool.QueryRow(ctx, `SELECT COUNT(*) FROM notifications WHERE source_event_id LIKE $1 AND type=$2`, "%"+suffix+"%", notification.TypeDocApproved).Scan(&liveNotes); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if liveNotes != nFix {
|
||||
t.Fatalf("live unique outcomes=%d want %d", liveNotes, nFix)
|
||||
}
|
||||
|
||||
// same-tx rollback: handle+ack then rollback leaves no extra row
|
||||
missingDoc := approvedEnv("missing-"+suffix, "no-such-doc-"+suffix, projectID, 99)
|
||||
appendEnv(t, ctx, pool, missingDoc, []string{events.ConsumerNotify})
|
||||
if _, err := d.RunOnce(ctx, events.ConsumerNotify); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var qStatus string
|
||||
if err := pool.QueryRow(ctx, `SELECT status FROM event_deliveries WHERE event_id=$1 AND consumer=$2`, missingDoc.EventID, events.ConsumerNotify).Scan(&qStatus); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if qStatus != "quarantine" {
|
||||
t.Fatalf("missing doc status=%s want quarantine", qStatus)
|
||||
}
|
||||
var qNotes int
|
||||
if err := pool.QueryRow(ctx, `SELECT COUNT(*) FROM notifications WHERE source_event_id=$1`, missingDoc.EventID).Scan(&qNotes); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if qNotes != 0 {
|
||||
t.Fatalf("quarantine leaked notifications=%d", qNotes)
|
||||
}
|
||||
|
||||
tx, err := pool.Begin(ctx)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := (&docnotify.EventConsumer{
|
||||
Notify: notification.NewService(pool), Docs: docgov.NewGovernanceEngine(pool), Env: "dev",
|
||||
}).HandleLease(ctx, events.Lease{
|
||||
EventID: ids[0], EventName: events.EventApproved, ProjectID: projectID, Environment: "dev",
|
||||
Envelope: events.Envelope{EventID: ids[0], EventName: events.EventApproved, ProjectID: projectID, AggregateID: "e1a30_" + suffix + "_0", Payload: []byte(fmt.Sprintf(`{"doc_id":%q,"project_id":%d}`, "e1a30_"+suffix+"_0", projectID))},
|
||||
}, tx); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
_ = tx.Rollback(ctx)
|
||||
if err := pool.QueryRow(ctx, `SELECT COUNT(*) FROM notifications WHERE source_event_id=$1`, ids[0]).Scan(&liveNotes); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if liveNotes != 1 {
|
||||
t.Fatalf("rollback of extra handle changed committed count=%d", liveNotes)
|
||||
}
|
||||
|
||||
if err := events.SetMode(ctx, pool, "dev", events.ConsumerNotify, events.ModeOff); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
func e1aPool(t *testing.T) (*pgxpool.Pool, context.Context) {
|
||||
t.Helper()
|
||||
if testing.Short() {
|
||||
t.Skip("e1a")
|
||||
}
|
||||
dsn := db.SkipIfNoTestDatabaseURL(t)
|
||||
db.RefuseSharedEnvDSN(t, dsn)
|
||||
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)
|
||||
}
|
||||
return pool, ctx
|
||||
}
|
||||
|
||||
func newNotifyDispatcher(t *testing.T, pool *pgxpool.Pool, owner string) *events.Dispatcher {
|
||||
t.Helper()
|
||||
d := events.NewDispatcher(pool, "dev", owner)
|
||||
if err := d.Register(events.ConsumerNotify, (&docnotify.EventConsumer{
|
||||
Notify: notification.NewService(pool), Docs: docgov.NewGovernanceEngine(pool), Env: "dev",
|
||||
}).HandleLease); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return d
|
||||
}
|
||||
|
||||
func seedUserProject(t *testing.T, ctx context.Context, pool *pgxpool.Pool, suffix string) (userID, projectID int64) {
|
||||
t.Helper()
|
||||
if err := pool.QueryRow(ctx, `INSERT INTO users (account_id, name, role) VALUES ($1,'A','engineer') RETURNING id`, "e1a-"+suffix).Scan(&userID); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := pool.QueryRow(ctx, `INSERT INTO projects (code, name, status) VALUES ($1,$2,'active') RETURNING id`, "e1a-"+suffix, "e1a").Scan(&projectID); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return userID, projectID
|
||||
}
|
||||
|
||||
func seedDoc(t *testing.T, ctx context.Context, pool *pgxpool.Pool, docID string, projectID, userID int64) {
|
||||
t.Helper()
|
||||
if _, err := pool.Exec(ctx, `
|
||||
INSERT INTO document_records (
|
||||
doc_id, project_id, task_key, title, stage, version, current_state, is_locked, is_template_only,
|
||||
author_id, author_name, created_at, updated_at
|
||||
) VALUES ($1,$2,'intake','t','V0.5',1,'approved',true,false,$3,'A',now(),now())
|
||||
`, docID, projectID, userID); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
func approvedEnv(id, docID string, projectID, ver int64) events.Envelope {
|
||||
body := []byte(fmt.Sprintf(`{"doc_id":%q,"project_id":%d,"task_key":"intake"}`, docID, projectID))
|
||||
return events.Envelope{
|
||||
EventID: "approved-" + id, EventName: events.EventApproved, SchemaVersion: events.EnvelopeSchema,
|
||||
Environment: "dev", ProjectID: projectID, AggregateType: "document_approval", AggregateID: docID,
|
||||
AggregateVersion: ver, ActorRef: "user:3", IdempotencyKey: "approved-" + id,
|
||||
PayloadDigest: events.Digest(body), Producer: "docflow", OccurredAt: time.Now().UTC(), Payload: body,
|
||||
}
|
||||
}
|
||||
|
||||
func appendEnv(t *testing.T, ctx context.Context, pool *pgxpool.Pool, env events.Envelope, consumers []string) {
|
||||
t.Helper()
|
||||
if err := uow.NewPG(pool).Within(ctx, func(scope uow.TxScope) error {
|
||||
return (events.PGSink{}).Append(uow.WithScope(ctx, scope), env, consumers)
|
||||
}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
func hashCommitted(ctx context.Context, pool *pgxpool.Pool, eventID string) (string, error) {
|
||||
tx, err := pool.Begin(ctx)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
defer func() { _ = tx.Rollback(ctx) }()
|
||||
return events.HashVisibleNotifications(ctx, tx, eventID)
|
||||
}
|
||||
Reference in New Issue
Block a user