姚章浩
245 lines
9.8 KiB
Go
245 lines
9.8 KiB
Go
package qaopsorg_test
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"net/url"
|
|
"os"
|
|
"strings"
|
|
"sync"
|
|
"sync/atomic"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/jackc/pgx/v5/pgxpool"
|
|
"knowledge-governance/internal/jobs"
|
|
postgresstore "knowledge-governance/internal/platform/postgres"
|
|
"knowledge-governance/internal/qaopsorg"
|
|
)
|
|
|
|
type syncRunner func(context.Context) (qaopsorg.Import, error)
|
|
|
|
func (f syncRunner) Run(ctx context.Context) (qaopsorg.Import, error) { return f(ctx) }
|
|
func TestDirectorySyncQueueDurableReplayAndLeaseFencing(t *testing.T) {
|
|
dsn := os.Getenv("KG_QAOPS_ORG_TEST_DSN")
|
|
if dsn == "" {
|
|
t.Skip("explicit isolated organization database is not configured")
|
|
}
|
|
u, err := url.Parse(dsn)
|
|
if err != nil || u.Hostname() != "127.0.0.1" || !strings.HasPrefix(u.Path, "/kg_qaops_org_test_") || os.Getenv("KG_QAOPS_ORG_TEST_ISOLATED") != "1" {
|
|
t.Fatal("refusing non-isolated organization database")
|
|
}
|
|
ctx := context.Background()
|
|
pool, err := pgxpool.New(ctx, dsn)
|
|
if err != nil {
|
|
t.Fatal("isolated connection failed")
|
|
}
|
|
defer pool.Close()
|
|
if err = postgresstore.ApplyMigrations(ctx, pool); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
store := qaopsorg.NewStore(pool, "sync-fixture-tenant")
|
|
queue := postgresstore.NewStore(pool)
|
|
initial := qaopsorg.Import{Source: "weekly_report_live", Complete: true, SyncedAt: time.Now().Add(-time.Hour), Directory: qaopsorg.Directory{Departments: []qaopsorg.Department{{ID: "root", Name: "同步夹具"}}, People: []qaopsorg.Person{{ID: "fixture-person", DepartmentIDs: []string{"root"}}}}}
|
|
old, err := store.ReplaceDirectory(ctx, initial)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
saved, err := store.SaveRules(ctx, qaopsorg.SaveRequest{DirectoryID: old.DirectoryID, Rules: []qaopsorg.Rule{{DepartmentID: "root", Region: "APAC"}}}, "fixture-admin")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, _, err = store.EnqueueSync(ctx, "op-no-worker", "fixture-admin"); !errors.Is(err, qaopsorg.ErrSyncWorkerUnavailable) {
|
|
t.Fatalf("disconnected worker accepted: %v", err)
|
|
}
|
|
if _, err = queue.Claim(ctx, "fixture-sync-worker", []string{store.SyncJobType()}, time.Now(), time.Minute); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
var first qaopsorg.SyncReceipt
|
|
var created atomic.Int32
|
|
var wg sync.WaitGroup
|
|
failures := make(chan error, 12)
|
|
ids := make(chan string, 12)
|
|
for range 12 {
|
|
wg.Add(1)
|
|
go func() {
|
|
defer wg.Done()
|
|
r, replay, e := qaopsorg.NewStore(pool, "sync-fixture-tenant").EnqueueSync(ctx, "op-same-action", "fixture-admin")
|
|
if e != nil {
|
|
failures <- e
|
|
return
|
|
}
|
|
if !replay {
|
|
created.Add(1)
|
|
}
|
|
ids <- r.JobID
|
|
}()
|
|
}
|
|
wg.Wait()
|
|
close(failures)
|
|
for e := range failures {
|
|
t.Fatal(e)
|
|
}
|
|
close(ids)
|
|
for id := range ids {
|
|
if first.JobID == "" {
|
|
first.JobID = id
|
|
}
|
|
if id != first.JobID {
|
|
t.Fatal("duplicate effect")
|
|
}
|
|
}
|
|
if created.Load() != 1 {
|
|
t.Fatal("operation replay was not durable")
|
|
}
|
|
reused, _, err := store.EnqueueSync(ctx, "op-another-click", "fixture-admin")
|
|
if err != nil || !reused.Deduplicated || reused.JobID != first.JobID {
|
|
t.Fatal("in-flight clicks did not coalesce")
|
|
}
|
|
if _, _, err = store.EnqueueSync(ctx, "op-same-action", "another-admin"); !errors.Is(err, qaopsorg.ErrConflict) {
|
|
t.Fatal("actor fingerprint conflict ignored")
|
|
}
|
|
other, _ := qaopsorg.NewStore(pool, "other-sync-tenant").GetSyncStatus(ctx)
|
|
if other.Job != nil {
|
|
t.Fatal("job leaked across tenants")
|
|
}
|
|
var n int
|
|
if err = pool.QueryRow(ctx, "SELECT count(*) FROM jobs WHERE type=$1", store.SyncJobType()).Scan(&n); err != nil || n != 1 {
|
|
t.Fatal("job count must remain one")
|
|
}
|
|
job, err := queue.Claim(ctx, "worker-before-crash", []string{store.SyncJobType()}, time.Now(), time.Minute)
|
|
if err != nil || job == nil {
|
|
t.Fatal("job was not durable/claimable")
|
|
}
|
|
// A different process can reclaim expired work; the old owner must not activate data.
|
|
if _, err = pool.Exec(ctx, "UPDATE jobs SET lease_until=now()-interval '1 second' WHERE id=$1", job.ID); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
reclaimed, err := queue.Claim(ctx, "worker-after-restart", []string{store.SyncJobType()}, time.Now(), time.Minute)
|
|
if err != nil || reclaimed == nil || reclaimed.Attempt != 2 {
|
|
t.Fatal("expired lease did not resume")
|
|
}
|
|
newer := initial
|
|
newer.SyncedAt = time.Now()
|
|
newer.Directory.People = append(append([]qaopsorg.Person{}, initial.Directory.People...), qaopsorg.Person{ID: "fixture-person-2", DepartmentIDs: []string{"root"}})
|
|
var fetched atomic.Int32
|
|
run := syncRunner(func(context.Context) (qaopsorg.Import, error) { fetched.Add(1); return newer, nil })
|
|
if err = store.HandleDirectorySync(ctx, *job, run); err == nil {
|
|
t.Fatal("stale owner activated a snapshot")
|
|
}
|
|
still, _ := store.GetView(ctx)
|
|
if still.DirectoryID != old.DirectoryID {
|
|
t.Fatal("stale lease changed active data")
|
|
}
|
|
if err = store.HandleDirectorySync(ctx, *reclaimed, run); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
count := fetched.Load()
|
|
// Crash after committing the snapshot, before queue.Complete: do not fetch twice.
|
|
if err = store.HandleDirectorySync(ctx, *reclaimed, run); err != nil || fetched.Load() != count {
|
|
t.Fatal("committed source was fetched again")
|
|
}
|
|
if err = queue.Complete(ctx, reclaimed.ID, reclaimed.LeaseOwner, time.Now()); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
current, _ := store.GetView(ctx)
|
|
if current.View.TotalPeople != 2 || current.RevisionID != saved.RevisionID || current.View.Regions["APAC"] != 2 {
|
|
t.Fatal("refresh lost regional rules or failed to activate")
|
|
}
|
|
status, err := store.GetSyncStatus(ctx)
|
|
if err != nil || status.Job == nil || status.Job.State != "completed" || status.Job.Result == nil || status.Job.Result.People != 2 {
|
|
t.Fatal("safe status missing durable result")
|
|
}
|
|
raw, _ := json.Marshal(status)
|
|
if strings.Contains(string(raw), "fixture-person") || strings.Contains(string(raw), "lease_owner") {
|
|
t.Fatal("personnel/lease details leaked in status")
|
|
}
|
|
// Failure must preserve the previous complete directory and never store raw exceptions.
|
|
failed, _, err := store.EnqueueSync(ctx, "op-fails", "fixture-admin")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
bad, err := queue.Claim(ctx, "worker-failure", []string{store.SyncJobType()}, time.Now(), time.Minute)
|
|
if err != nil || bad == nil || bad.ID != failed.JobID {
|
|
t.Fatal("failure job was not enqueued")
|
|
}
|
|
failure := store.HandleDirectorySync(ctx, *bad, syncRunner(func(context.Context) (qaopsorg.Import, error) {
|
|
return qaopsorg.Import{}, errors.New("sensitive-fixture-must-not-escape")
|
|
}))
|
|
if failure == nil || strings.Contains(failure.Error(), "sensitive-fixture") {
|
|
t.Fatal("unsafe source exception")
|
|
}
|
|
if err = queue.Fail(ctx, bad.ID, bad.LeaseOwner, failure.Error(), "directory_sync_failed", time.Now(), time.Now().Add(time.Minute)); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
kept, _ := store.GetView(ctx)
|
|
if kept.DirectoryID != current.DirectoryID || kept.RevisionID != saved.RevisionID {
|
|
t.Fatal("failed refresh replaced valid state")
|
|
}
|
|
late := qaopsorg.NewStore(pool, "late-fence-tenant")
|
|
lateOld, err := late.ReplaceDirectory(ctx, initial)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, err = queue.Claim(ctx, "late-worker", []string{late.SyncJobType()}, time.Now(), time.Minute); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, _, err = late.EnqueueSync(ctx, "op-late-fence", "fixture-admin"); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
lateJob, err := queue.Claim(ctx, "late-worker", []string{late.SyncJobType()}, time.Now(), time.Minute)
|
|
if err != nil || lateJob == nil {
|
|
t.Fatal("late-fence job unavailable")
|
|
}
|
|
lateErr := late.HandleDirectorySync(ctx, *lateJob, syncRunner(func(context.Context) (qaopsorg.Import, error) {
|
|
_, failure := pool.Exec(ctx, "UPDATE jobs SET lease_owner='replacement-worker',attempt=attempt+1,lease_until=now()+interval '1 minute' WHERE id=$1", lateJob.ID)
|
|
return newer, failure
|
|
}))
|
|
if lateErr == nil {
|
|
t.Fatal("lease lost DURING source fetch was not fenced at commit")
|
|
}
|
|
lateKept, _ := late.GetView(ctx)
|
|
if lateKept.DirectoryID != lateOld.DirectoryID {
|
|
t.Fatal("late stale worker activated data")
|
|
}
|
|
if err = pool.QueryRow(ctx, "SELECT count(*) FROM qa_ops_directory_snapshots WHERE tenant_key=$1", "late-fence-tenant").Scan(&n); err != nil || n != 1 {
|
|
t.Fatal("rejected stale snapshot left durable rows")
|
|
}
|
|
if _, err = pool.Exec(ctx, "UPDATE jobs SET lease_until=now()-interval '1 second' WHERE id=$1", lateJob.ID); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
interrupted, err := late.GetSyncStatus(ctx)
|
|
if err != nil || interrupted.Job == nil || interrupted.Job.State != "retry" {
|
|
t.Fatal("interrupted lease must wait for takeover, not claim active source processing")
|
|
}
|
|
if _, err = pool.Exec(ctx, "UPDATE jobs SET attempt=max_attempts,lease_until=now()-interval '1 second' WHERE id=$1", lateJob.ID); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
exhausted, err := late.GetSyncStatus(ctx)
|
|
if err != nil || exhausted.Job == nil || exhausted.Job.State != "failed" {
|
|
t.Fatal("exhausted crashed worker reported running forever")
|
|
}
|
|
fresh, _, err := late.EnqueueSync(ctx, "op-after-exhausted", "fixture-admin")
|
|
if err != nil || fresh.JobID == lateJob.ID {
|
|
t.Fatal("explicit retry could not replace an exhausted crashed job")
|
|
}
|
|
ghost := qaopsorg.NewStore(pool, "unadmitted-tenant")
|
|
if _, err = ghost.ReplaceDirectory(ctx, initial); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
_, err = queue.Enqueue(ctx, jobs.EnqueueRequest{ID: "fixture-unadmitted-job", Type: ghost.SyncJobType(), Payload: json.RawMessage(`{}`), AvailableAt: time.Now(), MaxAttempts: 1, IdempotencyKey: "fixture-unadmitted-job"}, time.Now())
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
ghostJob, err := queue.Claim(ctx, "ghost-worker", []string{ghost.SyncJobType()}, time.Now(), time.Minute)
|
|
if err != nil || ghostJob == nil {
|
|
t.Fatal("forged fixture unavailable")
|
|
}
|
|
var ghostCalls int
|
|
ghostErr := ghost.HandleDirectorySync(ctx, *ghostJob, syncRunner(func(context.Context) (qaopsorg.Import, error) { ghostCalls++; return newer, nil }))
|
|
if ghostErr == nil || ghostCalls != 0 {
|
|
t.Fatal("unadmitted generic task consumed the directory source")
|
|
}
|
|
}
|