362 lines
12 KiB
Go
362 lines
12 KiB
Go
package outbox
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"encoding/json"
|
|
"errors"
|
|
"net/http"
|
|
"net/http/httptest"
|
|
"os"
|
|
"strings"
|
|
"sync"
|
|
"sync/atomic"
|
|
"testing"
|
|
"time"
|
|
"unicode"
|
|
|
|
orbit "git.yuntuo.hk/k2safe/orbit/sdk/go"
|
|
"github.com/go-sql-driver/mysql"
|
|
)
|
|
|
|
func integrationDB(t *testing.T) (*sql.DB, string) {
|
|
t.Helper()
|
|
dsn := os.Getenv("ORBIT_TEST_SENDER_DSN")
|
|
if dsn == "" {
|
|
t.Skip("requires isolated sender database")
|
|
}
|
|
cfg, err := mysql.ParseDSN(dsn)
|
|
if err != nil || cfg.DBName != "orbit_sender_test" || !strings.HasPrefix(cfg.Addr, "127.0.0.1:") {
|
|
t.Fatal("sender tests require disposable loopback database")
|
|
}
|
|
ctx, stop := context.WithTimeout(context.Background(), 5*time.Second)
|
|
defer stop()
|
|
db, err := Open(ctx, dsn)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
t.Cleanup(func() { db.Close() })
|
|
return db, dsn
|
|
}
|
|
func testStore(t *testing.T, db *sql.DB) *Store {
|
|
t.Helper()
|
|
s, err := New(db, testID(t), testID(t))
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
return s
|
|
}
|
|
func enqueue(t *testing.T, s *Store) string {
|
|
t.Helper()
|
|
id := testID(t)
|
|
if _, err := s.Enqueue(context.Background(), id, "objects/test", testRequest(s.queue)); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
return id
|
|
}
|
|
func status(t *testing.T, s *Store, id string) Record {
|
|
t.Helper()
|
|
r, err := s.Status(context.Background(), id)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
return r
|
|
}
|
|
func execSQL(t *testing.T, db *sql.DB, q string, args ...any) {
|
|
t.Helper()
|
|
ctx, stop := context.WithTimeout(context.Background(), 5*time.Second)
|
|
defer stop()
|
|
if _, err := db.ExecContext(ctx, q, args...); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
func countRows(t *testing.T, s *Store, table string) int {
|
|
t.Helper()
|
|
if table != "sender_jobs" && table != "sender_outbox" {
|
|
t.Fatal("invalid test table")
|
|
}
|
|
var n int
|
|
query := `SELECT COUNT(*) FROM sender_outbox WHERE namespace_id=? AND queue_id=?`
|
|
if table == "sender_jobs" {
|
|
query = `SELECT COUNT(*) FROM sender_jobs j JOIN sender_outbox o ON o.namespace_id=j.namespace_id AND o.operation_id=j.operation_id WHERE o.namespace_id=? AND o.queue_id=?`
|
|
}
|
|
if err := s.db.QueryRow(query, s.namespace, s.queue).Scan(&n); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
return n
|
|
}
|
|
func fakeReceipt(t *testing.T, s *Store) orbit.SubmitTaskResult {
|
|
return orbit.SubmitTaskResult{Task: orbit.Task{ID: testID(t), NamespaceID: s.namespace, QueueID: s.queue, Revision: 1, State: orbit.TaskWaiting}}
|
|
}
|
|
|
|
func TestIntegrationSenderSchema(t *testing.T) {
|
|
db, _ := integrationDB(t)
|
|
var fk int
|
|
if err := db.QueryRow(`SELECT COUNT(*) FROM information_schema.REFERENTIAL_CONSTRAINTS WHERE CONSTRAINT_SCHEMA=DATABASE()`).Scan(&fk); err != nil || fk != 0 {
|
|
t.Fatal(fk, err)
|
|
}
|
|
rows, err := db.Query(`SELECT COLUMN_COMMENT FROM information_schema.COLUMNS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME IN ('sender_outbox','sender_jobs')`)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer rows.Close()
|
|
n := 0
|
|
for rows.Next() {
|
|
var c string
|
|
if rows.Scan(&c) != nil || !strings.ContainsFunc(c, func(r rune) bool { return unicode.Is(unicode.Han, r) }) {
|
|
t.Fatal("missing Chinese column comment")
|
|
}
|
|
n++
|
|
}
|
|
if rows.Err() != nil || n != 22 {
|
|
t.Fatal(n, rows.Err())
|
|
}
|
|
rows.Close()
|
|
s := testStore(t, db)
|
|
id := enqueue(t, s)
|
|
for _, assignment := range []string{"state='ACCEPTED'", "send_count=101,lease_epoch=101", "request_json='not json'", "send_deadline_at=TIMESTAMPADD(HOUR,48,created_at)"} {
|
|
_, err = db.Exec(`UPDATE sender_outbox SET `+assignment+` WHERE namespace_id=? AND operation_id=?`, s.namespace, id)
|
|
var rejected *mysql.MySQLError
|
|
if !errors.As(err, &rejected) || rejected.Number != 3819 {
|
|
t.Fatal("invalid persisted state did not violate CHECK", assignment, err)
|
|
}
|
|
}
|
|
var plan string
|
|
if err = db.QueryRow(`EXPLAIN FORMAT=JSON SELECT operation_id FROM sender_outbox FORCE INDEX(ix_sender_due) WHERE namespace_id=? AND queue_id=? AND wake_at<=UTC_TIMESTAMP(6) ORDER BY wake_at,operation_id LIMIT 1 FOR UPDATE SKIP LOCKED`, s.namespace, s.queue).Scan(&plan); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if !strings.Contains(plan, `"key": "ix_sender_due"`) || strings.Contains(plan, `"using_filesort": true`) {
|
|
t.Fatal("sender due query lost ordered index", plan)
|
|
}
|
|
t.Log("two business tables, 22 Chinese-commented columns, zero foreign keys")
|
|
}
|
|
func TestIntegrationSenderConcurrentEnqueueAndConflict(t *testing.T) {
|
|
db, _ := integrationDB(t)
|
|
s := testStore(t, db)
|
|
id := testID(t)
|
|
in := testRequest(s.queue)
|
|
invalid := in
|
|
invalid.BusinessKey = "operation-\xff"
|
|
if _, err := s.Enqueue(context.Background(), id, "objects/test", invalid); !errors.Is(err, ErrInvalid) || countRows(t, s, "sender_jobs") != 0 || countRows(t, s, "sender_outbox") != 0 {
|
|
t.Fatal("invalid UTF-8 changed business state", err)
|
|
}
|
|
type outcome struct {
|
|
duplicate bool
|
|
err error
|
|
}
|
|
results := make(chan outcome, 8)
|
|
var wg sync.WaitGroup
|
|
for range 8 {
|
|
wg.Go(func() {
|
|
dup, err := s.Enqueue(context.Background(), id, "objects/test", in)
|
|
results <- outcome{dup, err}
|
|
})
|
|
}
|
|
wg.Wait()
|
|
close(results)
|
|
duplicates := 0
|
|
for r := range results {
|
|
if r.err != nil {
|
|
t.Fatal(r.err)
|
|
}
|
|
if r.duplicate {
|
|
duplicates++
|
|
}
|
|
}
|
|
if duplicates != 7 || countRows(t, s, "sender_jobs") != 1 || countRows(t, s, "sender_outbox") != 1 {
|
|
t.Fatal("duplicate business transaction", duplicates)
|
|
}
|
|
if _, err := s.Enqueue(context.Background(), id, "objects/other", in); !errors.Is(err, ErrConflict) {
|
|
t.Fatal("business identity conflict lost", err)
|
|
}
|
|
in.Payload = json.RawMessage(`{"sequence":9007199254740992}`)
|
|
if _, err := s.Enqueue(context.Background(), id, "objects/test", in); !errors.Is(err, ErrConflict) {
|
|
t.Fatal("request content conflict lost", err)
|
|
}
|
|
other, _ := New(db, s.namespace, testID(t))
|
|
if _, err := other.Status(context.Background(), id); !errors.Is(err, ErrNotFound) {
|
|
t.Fatal("cross queue record returned", err)
|
|
}
|
|
if _, err := other.Enqueue(context.Background(), id, "objects/test", testRequest(other.queue)); !errors.Is(err, ErrConflict) {
|
|
t.Fatal("operation was moved to another queue", err)
|
|
}
|
|
if !strings.Contains(string(status(t, s, id).RequestJSON), "9007199254740993") {
|
|
t.Fatal("large integer rounded")
|
|
}
|
|
}
|
|
func TestIntegrationSenderBusinessRollbackAndConfirmationAtomicity(t *testing.T) {
|
|
db, _ := integrationDB(t)
|
|
s := testStore(t, db)
|
|
id := testID(t)
|
|
execSQL(t, db, `CREATE TRIGGER sender_insert_fail BEFORE INSERT ON sender_jobs FOR EACH ROW SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='injected business insert failure'`)
|
|
t.Cleanup(func() {
|
|
db.Exec(`DROP TRIGGER IF EXISTS sender_insert_fail`)
|
|
db.Exec(`DROP TRIGGER IF EXISTS sender_update_fail`)
|
|
})
|
|
if _, err := s.Enqueue(context.Background(), id, "objects/test", testRequest(s.queue)); !errors.Is(err, ErrStorage) {
|
|
t.Fatal(err)
|
|
}
|
|
if countRows(t, s, "sender_outbox") != 0 || countRows(t, s, "sender_jobs") != 0 {
|
|
t.Fatal("business rollback left a request")
|
|
}
|
|
execSQL(t, db, `DROP TRIGGER sender_insert_fail`)
|
|
if _, err := s.Enqueue(context.Background(), id, "objects/test", testRequest(s.queue)); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
claim, err := s.claim(context.Background())
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
receipt := fakeReceipt(t, s)
|
|
execSQL(t, db, `CREATE TRIGGER sender_update_fail BEFORE UPDATE ON sender_jobs FOR EACH ROW SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='injected business confirmation failure'`)
|
|
if err = s.finish(context.Background(), claim, "ACCEPTED", "ACCEPTED", 0, &receipt); !errors.Is(err, ErrStorage) {
|
|
t.Fatal(err)
|
|
}
|
|
if r := status(t, s, id); r.State != "SENDING" || r.TaskID != nil {
|
|
t.Fatal("partial confirmation committed")
|
|
}
|
|
execSQL(t, db, `DROP TRIGGER sender_update_fail`)
|
|
if err = s.finish(context.Background(), claim, "ACCEPTED", "ACCEPTED", 0, &receipt); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
var task string
|
|
if err = db.QueryRow(`SELECT task_id FROM sender_jobs WHERE namespace_id=? AND operation_id=?`, s.namespace, id).Scan(&task); err != nil || task != receipt.Task.ID {
|
|
t.Fatal("business confirmation missing", err)
|
|
}
|
|
}
|
|
func TestIntegrationSenderConcurrentClaimsAndStaleGeneration(t *testing.T) {
|
|
db, _ := integrationDB(t)
|
|
s := testStore(t, db)
|
|
for range 20 {
|
|
enqueue(t, s)
|
|
}
|
|
claims := make(chan Record, 20)
|
|
errs := make(chan error, 8)
|
|
var wg sync.WaitGroup
|
|
for range 8 {
|
|
wg.Go(func() {
|
|
for {
|
|
r, err := s.claim(context.Background())
|
|
if errors.Is(err, ErrIdle) {
|
|
return
|
|
}
|
|
if err != nil {
|
|
errs <- err
|
|
return
|
|
}
|
|
claims <- r
|
|
}
|
|
})
|
|
}
|
|
wg.Wait()
|
|
close(claims)
|
|
close(errs)
|
|
for err := range errs {
|
|
t.Fatal(err)
|
|
}
|
|
seen := map[string]bool{}
|
|
var first Record
|
|
for r := range claims {
|
|
if seen[r.OperationID] || r.SendCount != 1 {
|
|
t.Fatal("duplicate claim")
|
|
}
|
|
seen[r.OperationID] = true
|
|
first = r
|
|
}
|
|
if len(seen) != 20 {
|
|
t.Fatal("unclaimed records", len(seen))
|
|
}
|
|
execSQL(t, db, `UPDATE sender_outbox SET wake_at=TIMESTAMPADD(SECOND,-1,UTC_TIMESTAMP(6)) WHERE namespace_id=? AND operation_id=?`, s.namespace, first.OperationID)
|
|
replacement, err := s.claim(context.Background())
|
|
if err != nil || replacement.LeaseEpoch != 2 || replacement.IdempotencyKey != first.IdempotencyKey {
|
|
t.Fatal(replacement, err)
|
|
}
|
|
receipt := fakeReceipt(t, s)
|
|
if err = s.finish(context.Background(), first, "ACCEPTED", "ACCEPTED", 0, &receipt); !errors.Is(err, ErrLeaseLost) {
|
|
t.Fatal("stale sender overwrote result", err)
|
|
}
|
|
if err = s.finish(context.Background(), replacement, "ACCEPTED", "ACCEPTED", 0, &receipt); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
func TestIntegrationSenderHTTPClassificationAndBudgets(t *testing.T) {
|
|
db, _ := integrationDB(t)
|
|
for _, mode := range []string{"success", "backpressure", "forbidden", "conflict", "unknown", "wrong_target", "invalid_receipt", "corrupt_request", "exhausted", "last_attempt", "deadline"} {
|
|
t.Run(mode, func(t *testing.T) {
|
|
s := testStore(t, db)
|
|
id := enqueue(t, s)
|
|
var posts atomic.Int32
|
|
if mode == "corrupt_request" {
|
|
execSQL(t, db, `UPDATE sender_outbox SET request_json='{}' WHERE namespace_id=?`, s.namespace)
|
|
}
|
|
if mode == "exhausted" {
|
|
execSQL(t, db, `UPDATE sender_outbox SET send_count=100,lease_epoch=100 WHERE namespace_id=?`, s.namespace)
|
|
}
|
|
if mode == "last_attempt" {
|
|
execSQL(t, db, `UPDATE sender_outbox SET send_count=99,lease_epoch=99 WHERE namespace_id=?`, s.namespace)
|
|
}
|
|
if mode == "deadline" {
|
|
execSQL(t, db, `UPDATE sender_outbox SET send_deadline_at=TIMESTAMPADD(MICROSECOND,1,created_at) WHERE namespace_id=?`, s.namespace)
|
|
}
|
|
receipt := fakeReceipt(t, s)
|
|
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
|
if r.Method == "GET" {
|
|
ns := s.namespace
|
|
if mode == "wrong_target" {
|
|
ns = testID(t)
|
|
}
|
|
json.NewEncoder(w).Encode(map[string]any{"request_id": id, "data": orbit.Queue{ID: s.queue, NamespaceID: ns}})
|
|
return
|
|
}
|
|
posts.Add(1)
|
|
if r.Header.Get("Idempotency-Key") != "outbox-task-"+id {
|
|
t.Error("wrong submission identity")
|
|
}
|
|
switch mode {
|
|
case "backpressure":
|
|
w.Header().Set("Retry-After", "30")
|
|
w.WriteHeader(429)
|
|
json.NewEncoder(w).Encode(map[string]any{"request_id": id, "error": map[string]string{"code": "QUEUE_FULL", "message": "full"}})
|
|
case "forbidden", "conflict":
|
|
code, status := "PERMISSION_DENIED", 403
|
|
if mode == "conflict" {
|
|
code, status = "IDEMPOTENCY_CONFLICT", 409
|
|
}
|
|
w.WriteHeader(status)
|
|
json.NewEncoder(w).Encode(map[string]any{"request_id": id, "error": map[string]string{"code": code, "message": "rejected"}})
|
|
case "unknown", "last_attempt":
|
|
w.WriteHeader(502)
|
|
default:
|
|
reply := receipt
|
|
if mode == "invalid_receipt" {
|
|
reply.Task.NamespaceID = testID(t)
|
|
}
|
|
json.NewEncoder(w).Encode(map[string]any{"request_id": id, "data": reply})
|
|
}
|
|
}))
|
|
defer server.Close()
|
|
api, _ := orbit.NewClient(server.URL, orbit.Options{Token: "test-token"})
|
|
r, err := s.SendOnce(context.Background(), api)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
want := "REVIEW"
|
|
if mode == "success" {
|
|
want = "ACCEPTED"
|
|
}
|
|
if mode == "backpressure" || mode == "unknown" {
|
|
want = "PENDING"
|
|
}
|
|
if r.State != want {
|
|
t.Fatal(mode, r.State, r.LastCode)
|
|
}
|
|
if (mode == "wrong_target" || mode == "corrupt_request" || mode == "exhausted" || mode == "deadline") && posts.Load() != 0 {
|
|
t.Fatal("unsafe HTTP sent")
|
|
}
|
|
if mode == "backpressure" && (r.WakeAt == nil || time.Until(*r.WakeAt) < 29*time.Second) {
|
|
t.Fatal("Retry-After was ignored")
|
|
}
|
|
})
|
|
}
|
|
}
|