mirror of
https://github.com/multica-ai/multica.git
synced 2026-07-24 11:10:25 +02:00
121 lines
3.5 KiB
Go
121 lines
3.5 KiB
Go
package scheduler
|
|
|
|
import (
|
|
"context"
|
|
"sync"
|
|
"testing"
|
|
)
|
|
|
|
// TestConcurrentClaimsSingleWinner covers RFC §14:
|
|
//
|
|
// "pg_cron 并跑 | app scheduler 与 pg_cron 同时调用函数,
|
|
// 无重复窗口写入"
|
|
//
|
|
// The legacy `pg_cron` tick and the in-process scheduler both call
|
|
// `rollup_task_usage_hourly()`, so the SQL function's advisory lock
|
|
// 4246 prevents double-writes of the rollup itself. The scheduler adds
|
|
// a second layer via `sys_cron_executions`: even if multiple ticks (a
|
|
// scheduler in a second replica, a manual SQL call, a leftover
|
|
// `pg_cron` job) arrive at the same plan_time, only one row exists per
|
|
// (job, scope, plan_time) and only one runner gets Won=true. The rest
|
|
// fall through the conflict path and no-op.
|
|
//
|
|
// We simulate this by firing N concurrent claims at the same plan_time
|
|
// from distinct runner ids and asserting the table contract:
|
|
//
|
|
// - Exactly ONE caller observes Won=true.
|
|
// - Every other caller observes Conflicted=true (no Won, no Stole).
|
|
// - sys_cron_executions has exactly one row for the plan.
|
|
// - The row's runner_id matches the winner.
|
|
//
|
|
// This is the same single-winner property the SQL advisory lock 4246
|
|
// gives at the function-execution layer; the `sys_cron_executions`
|
|
// uniqueness key gives it at the scheduler layer, so a `pg_cron` tick
|
|
// running alongside the in-process scheduler cannot produce a duplicate
|
|
// SUCCESS audit row.
|
|
func TestConcurrentClaimsSingleWinner(t *testing.T) {
|
|
pool := integrationPool(t)
|
|
job := newTestJobSpec(uniqueJobName(t, "concurrent_claim"))
|
|
t.Cleanup(func() { cleanupExecutions(t, pool, job.Name) })
|
|
|
|
ctx := context.Background()
|
|
now, err := dbNow(ctx, pool)
|
|
if err != nil {
|
|
t.Fatalf("dbNow: %v", err)
|
|
}
|
|
planTime := FloorPlan(now, job.Cadence)
|
|
|
|
const contenders = 8
|
|
type result struct {
|
|
runnerID string
|
|
c claim
|
|
err error
|
|
}
|
|
results := make([]result, contenders)
|
|
var wg sync.WaitGroup
|
|
start := make(chan struct{})
|
|
for i := range contenders {
|
|
i := i
|
|
runnerID := "runner-" + string(rune('A'+i))
|
|
wg.Add(1)
|
|
go func() {
|
|
defer wg.Done()
|
|
<-start
|
|
c, err := tryClaim(ctx, pool, job, ScopeGlobal, planTime, now, runnerID)
|
|
results[i] = result{runnerID: runnerID, c: c, err: err}
|
|
}()
|
|
}
|
|
close(start)
|
|
wg.Wait()
|
|
|
|
wins := 0
|
|
conflicts := 0
|
|
steals := 0
|
|
var winner string
|
|
for _, r := range results {
|
|
if r.err != nil {
|
|
t.Fatalf("contender %s: %v", r.runnerID, r.err)
|
|
}
|
|
switch {
|
|
case r.c.Won:
|
|
wins++
|
|
winner = r.runnerID
|
|
case r.c.Stole:
|
|
steals++
|
|
case r.c.Conflicted:
|
|
conflicts++
|
|
}
|
|
}
|
|
if wins != 1 {
|
|
t.Fatalf("expected exactly 1 fresh winner, got %d (conflicts=%d steals=%d)",
|
|
wins, conflicts, steals)
|
|
}
|
|
if steals != 0 {
|
|
t.Fatalf("a fresh insert race must not produce a stale steal, got %d steals", steals)
|
|
}
|
|
if conflicts != contenders-1 {
|
|
t.Fatalf("expected %d conflicts, got %d", contenders-1, conflicts)
|
|
}
|
|
|
|
// Database-side proof: exactly one row, runner_id matches.
|
|
var rowCount int
|
|
if err := pool.QueryRow(ctx, `
|
|
SELECT COUNT(*) FROM sys_cron_executions WHERE job_name = $1 AND plan_time = $2
|
|
`, job.Name, planTime).Scan(&rowCount); err != nil {
|
|
t.Fatalf("count rows: %v", err)
|
|
}
|
|
if rowCount != 1 {
|
|
t.Fatalf("expected exactly 1 row in sys_cron_executions, got %d", rowCount)
|
|
}
|
|
|
|
var dbRunner string
|
|
if err := pool.QueryRow(ctx, `
|
|
SELECT runner_id FROM sys_cron_executions WHERE job_name = $1 AND plan_time = $2
|
|
`, job.Name, planTime).Scan(&dbRunner); err != nil {
|
|
t.Fatalf("scan winner: %v", err)
|
|
}
|
|
if dbRunner != winner {
|
|
t.Fatalf("DB winner %q != local winner %q", dbRunner, winner)
|
|
}
|
|
}
|