710 lines
27 KiB
Go
710 lines
27 KiB
Go
package mysql
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"encoding/json"
|
|
"fmt"
|
|
"os"
|
|
"strings"
|
|
"testing"
|
|
"time"
|
|
|
|
core "git.yuntuo.hk/k2safe/orbit/model"
|
|
"git.yuntuo.hk/k2safe/orbit/platform/auth"
|
|
"git.yuntuo.hk/k2safe/orbit/platform/events"
|
|
platform "git.yuntuo.hk/k2safe/orbit/platform/model"
|
|
pulse "git.yuntuo.hk/k2safe/orbit/pulse/model"
|
|
"git.yuntuo.hk/k2safe/orbit/pulse/service"
|
|
"git.yuntuo.hk/k2safe/orbit/pulse/worker"
|
|
)
|
|
|
|
type eventFixture struct {
|
|
*executionFixture
|
|
events *events.Service
|
|
store *EventStore
|
|
}
|
|
|
|
func newEventFixture(t *testing.T, ctx context.Context, db *sql.DB) *eventFixture {
|
|
f := newExecutionFixture(t, ctx, db)
|
|
var p auth.Permissions
|
|
var raw []byte
|
|
if err := db.QueryRowContext(ctx, `SELECT permissions FROM orbit_credentials WHERE namespace_id=? AND id=?`, f.token.NamespaceID, f.token.CredentialID).Scan(&raw); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := json.Unmarshal(raw, &p); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
p.Actions = append(p.Actions, "events:manage", "events:read", "events:consume")
|
|
raw, _ = json.Marshal(p)
|
|
f.exec(`UPDATE orbit_credentials SET permissions=? WHERE namespace_id=? AND id=?`, string(raw), f.token.NamespaceID, f.token.CredentialID)
|
|
store := NewEventStore(db)
|
|
return &eventFixture{f, events.New(store), store}
|
|
}
|
|
func (f *eventFixture) create(in events.CreateInput) events.Subscription {
|
|
f.t.Helper()
|
|
if in.Name == "" {
|
|
in.Name = "result-" + string(executionID(f.t))
|
|
}
|
|
if in.QueueID == "" {
|
|
in.QueueID = f.resources.QueueID
|
|
}
|
|
out, err := f.events.Create(f.ctx, f.token, string(executionID(f.t)), in)
|
|
if err != nil {
|
|
f.t.Fatal(err)
|
|
}
|
|
return out.Subscription
|
|
}
|
|
func (f *eventFixture) cancelResult() service.TaskView {
|
|
f.t.Helper()
|
|
task := f.submit(1)
|
|
out, err := f.tasks.CancelTask(f.ctx, f.token, task.ID, string(executionID(f.t)), service.CancelInput{ExpectedRevision: task.Revision, Reason: "create result notification"})
|
|
if err != nil {
|
|
f.t.Fatal(err)
|
|
}
|
|
return out.Task
|
|
}
|
|
func (f *eventFixture) grant(sub core.ID) *events.Grant {
|
|
f.t.Helper()
|
|
v, err := f.events.Claim(f.ctx, f.token, sub, events.ClaimInput{ConnectionID: executionID(f.t), Capacity: 1})
|
|
if err != nil || v == nil {
|
|
f.t.Fatalf("grant: %v", err)
|
|
}
|
|
return v
|
|
}
|
|
func (f *eventFixture) ack(sub core.ID, g *events.Grant) events.AckResult {
|
|
f.t.Helper()
|
|
v, err := f.events.Ack(f.ctx, f.token, sub, events.AckInput{DeliveryID: g.Delivery.ID, Epoch: g.Delivery.LeaseEpoch, LeaseSecret: g.LeaseSecret})
|
|
if err != nil {
|
|
f.t.Fatal(err)
|
|
}
|
|
return v
|
|
}
|
|
func (f *eventFixture) subscription(id core.ID) events.Subscription {
|
|
f.t.Helper()
|
|
v, err := f.events.Get(f.ctx, f.token, id)
|
|
if err != nil {
|
|
f.t.Fatal(err)
|
|
}
|
|
return v
|
|
}
|
|
func (f *eventFixture) expireDelivery(g *events.Grant) {
|
|
f.exec(`UPDATE orbit_deliveries SET lease_until=DATE_SUB(UTC_TIMESTAMP(6),INTERVAL 1 SECOND) WHERE namespace_id=? AND id=?`, f.token.NamespaceID, g.Delivery.ID)
|
|
}
|
|
func noEventGrant(t *testing.T, f *eventFixture, sub core.ID) {
|
|
t.Helper()
|
|
g, err := f.events.Claim(f.ctx, f.token, sub, events.ClaimInput{ConnectionID: executionID(t), Capacity: 1})
|
|
if err != nil || g != nil {
|
|
t.Fatalf("unexpected grant: %v %v", g, err)
|
|
}
|
|
}
|
|
|
|
func TestIntegrationResultSubscriptions(t *testing.T) {
|
|
dsn := os.Getenv("ORBIT_TEST_MYSQL_DSN")
|
|
if dsn == "" {
|
|
t.Skip("requires disposable MySQL")
|
|
}
|
|
ctx, cancel := context.WithTimeout(context.Background(), 90*time.Second)
|
|
defer cancel()
|
|
db, err := Open(ctx, dsn, 32, 8)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer db.Close()
|
|
if err = Migrate(ctx, db); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
zero := uint64(0)
|
|
t.Run("explicit_boundary_filters_and_frozen_result_precision", func(t *testing.T) {
|
|
f := newEventFixture(t, ctx, db)
|
|
task := f.submit(1)
|
|
ref := retryStart(t, f.executionFixture)
|
|
raw := json.RawMessage(`{"n":9007199254740993,"text":"<&>"}`)
|
|
in := worker.FinishInput{Ref: ref, Outcome: "SUCCEEDED", Result: raw}
|
|
message := executionID(t)
|
|
if _, err := f.a.Finish(ctx, f.token, f.binding, message, in); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, err := f.a.Finish(ctx, f.token, f.binding, message, in); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if f.count("orbit_result_events", "1=1") != 1 {
|
|
t.Fatal("duplicate result journal entry")
|
|
}
|
|
recent := f.create(events.CreateInput{})
|
|
if recent.CursorSequence != 1 {
|
|
t.Fatal("new subscription did not capture committed head")
|
|
}
|
|
noEventGrant(t, f, recent.ID)
|
|
history := f.create(events.CreateInput{FromSequence: &zero, Types: []string{"task.succeeded"}})
|
|
g := f.grant(history.ID)
|
|
if g.Event.ID != g.Delivery.EventID || g.Event.Sequence != 1 || g.Event.Snapshot.TaskID != task.ID || string(g.Event.Snapshot.Result) != string(raw) || g.Event.Snapshot.CurrentAttemptID == nil || *g.Event.Snapshot.CurrentAttemptID != ref.AttemptID {
|
|
t.Fatalf("bad snapshot: %+v", g.Event)
|
|
}
|
|
f.exec(`UPDATE pulse_tasks SET result='{"changed":true}' WHERE namespace_id=? AND id=?`, f.token.NamespaceID, task.ID)
|
|
f.ack(history.ID, g)
|
|
replay := f.create(events.CreateInput{FromSequence: &zero})
|
|
g = f.grant(replay.ID)
|
|
if string(g.Event.Snapshot.Result) != string(raw) {
|
|
t.Fatal("snapshot read mutable task result")
|
|
}
|
|
f.ack(replay.ID, g)
|
|
f.cancelResult()
|
|
noEventGrant(t, f, history.ID)
|
|
if f.subscription(history.ID).CursorSequence != 2 {
|
|
t.Fatal("filtered event did not advance cursor")
|
|
}
|
|
g = f.grant(recent.ID)
|
|
if g.Event.Type != "task.cancelled" || g.Event.Sequence != 2 {
|
|
t.Fatal("new subscription missed future result")
|
|
}
|
|
_, err := db.ExecContext(ctx, `UPDATE orbit_result_events SET snapshot='{}' WHERE namespace_id=? AND id=?`, f.token.NamespaceID, g.Event.ID)
|
|
if err == nil {
|
|
t.Fatal("journal is mutable")
|
|
}
|
|
})
|
|
t.Run("create_idempotency_queue_ownership_and_permission_isolation", func(t *testing.T) {
|
|
f := newEventFixture(t, ctx, db)
|
|
other := newEventFixture(t, ctx, db)
|
|
in := events.CreateInput{Name: "stable-result", QueueID: f.resources.QueueID, FromSequence: &zero}
|
|
a, err := f.events.Create(ctx, f.token, "stable", in)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
b, err := f.events.Create(ctx, f.token, "stable", in)
|
|
if err != nil || !b.Deduplicated || a.Subscription.ID != b.Subscription.ID {
|
|
t.Fatal("unstable create receipt", err)
|
|
}
|
|
in.MaxPending = 2
|
|
_, err = f.events.Create(ctx, f.token, "stable", in)
|
|
expectWorkerCode(t, err, "IDEMPOTENCY_CONFLICT")
|
|
_, err = f.events.Get(ctx, other.token, a.Subscription.ID)
|
|
expectWorkerCode(t, err, "NOT_FOUND")
|
|
_, err = f.events.Create(ctx, f.token, "cross-queue", events.CreateInput{Name: "cross", QueueID: other.resources.QueueID})
|
|
expectWorkerCode(t, err, "NOT_FOUND")
|
|
secondID := executionID(t)
|
|
secret, digest, err := auth.Issue(f.token.NamespaceID, secondID)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
p, _ := json.Marshal(auth.Permissions{Actions: []string{"events:read", "events:consume", "events:manage"}, Queues: []string{"*"}})
|
|
f.exec(`INSERT INTO orbit_credentials(namespace_id,id,name,secret_digest,permissions) VALUES (?,?,?, ?,?)`, f.token.NamespaceID, secondID, "second", digest, string(p))
|
|
second, err := auth.Parse(secret)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
_, err = f.events.Get(ctx, second, a.Subscription.ID)
|
|
expectWorkerCode(t, err, "NOT_FOUND")
|
|
own, err := f.events.Create(ctx, second, "stable", events.CreateInput{Name: "second", QueueID: f.resources.QueueID, FromSequence: &zero})
|
|
if err != nil || own.Subscription.ID == a.Subscription.ID {
|
|
t.Fatal("create receipt leaked across owners", err)
|
|
}
|
|
p, _ = json.Marshal(auth.Permissions{Actions: []string{"events:read", "events:consume", "events:manage"}, Queues: []string{string(executionID(t))}})
|
|
f.exec(`UPDATE orbit_credentials SET permissions=? WHERE namespace_id=? AND id=?`, string(p), f.token.NamespaceID, secondID)
|
|
_, err = f.events.Get(ctx, second, own.Subscription.ID)
|
|
expectWorkerCode(t, err, "NOT_FOUND")
|
|
p, _ = json.Marshal(auth.Permissions{Actions: []string{"tasks:read"}, Queues: []string{"*"}})
|
|
f.exec(`UPDATE orbit_credentials SET permissions=? WHERE namespace_id=? AND id=?`, string(p), f.token.NamespaceID, f.token.CredentialID)
|
|
_, err = f.events.Claim(ctx, f.token, a.Subscription.ID, events.ClaimInput{ConnectionID: executionID(t), Capacity: 1})
|
|
expectWorkerCode(t, err, "PERMISSION_DENIED")
|
|
f.exec(`UPDATE orbit_credentials SET revoked_at=UTC_TIMESTAMP(6) WHERE namespace_id=? AND id=?`, f.token.NamespaceID, f.token.CredentialID)
|
|
_, err = f.events.Get(ctx, f.token, a.Subscription.ID)
|
|
expectWorkerCode(t, err, "UNAUTHENTICATED")
|
|
})
|
|
t.Run("backpressure_does_not_block_result_commit_or_skip_history", func(t *testing.T) {
|
|
f := newEventFixture(t, ctx, db)
|
|
sub := f.create(events.CreateInput{MaxPending: 1})
|
|
f.cancelResult()
|
|
first := f.grant(sub.ID)
|
|
for range 3 {
|
|
f.cancelResult()
|
|
}
|
|
noEventGrant(t, f, sub.ID)
|
|
got := f.subscription(sub.ID)
|
|
if got.PendingCount != 1 || got.CursorSequence != 1 || f.count("orbit_result_events", "1=1") != 4 {
|
|
t.Fatal("window violated or producer blocked")
|
|
}
|
|
f.ack(sub.ID, first)
|
|
for seq := uint64(2); seq <= 4; seq++ {
|
|
g := f.grant(sub.ID)
|
|
if g.Event.Sequence != seq {
|
|
t.Fatal("backlog skipped")
|
|
}
|
|
f.ack(sub.ID, g)
|
|
}
|
|
if got = f.subscription(sub.ID); got.PendingCount != 0 || got.CursorSequence != 4 || got.Revision != 1 {
|
|
t.Fatal("invalid drained window or configuration revision changed")
|
|
}
|
|
})
|
|
t.Run("competing_consumers_and_duplicate_acks", func(t *testing.T) {
|
|
f := newEventFixture(t, ctx, db)
|
|
sub := f.create(events.CreateInput{})
|
|
f.cancelResult()
|
|
type answer struct {
|
|
g *events.Grant
|
|
err error
|
|
}
|
|
results := make(chan answer, 12)
|
|
for range 12 {
|
|
connection := executionID(t)
|
|
go func() {
|
|
g, err := f.events.Claim(ctx, f.token, sub.ID, events.ClaimInput{ConnectionID: connection, Capacity: 1})
|
|
results <- answer{g, err}
|
|
}()
|
|
}
|
|
var winner *events.Grant
|
|
for range 12 {
|
|
r := <-results
|
|
if r.err != nil {
|
|
t.Fatal(r.err)
|
|
}
|
|
if r.g != nil {
|
|
if winner != nil {
|
|
t.Fatal("two grants for one unexpired delivery")
|
|
}
|
|
winner = r.g
|
|
}
|
|
}
|
|
if winner == nil {
|
|
t.Fatal("no consumer won")
|
|
}
|
|
acks := make(chan events.AckResult, 12)
|
|
errs := make(chan error, 12)
|
|
for range 12 {
|
|
go func() {
|
|
v, err := f.events.Ack(ctx, f.token, sub.ID, events.AckInput{DeliveryID: winner.Delivery.ID, Epoch: winner.Delivery.LeaseEpoch, LeaseSecret: winner.LeaseSecret})
|
|
acks <- v
|
|
errs <- err
|
|
}()
|
|
}
|
|
fresh := 0
|
|
var ackTime time.Time
|
|
for range 12 {
|
|
v := <-acks
|
|
if err := <-errs; err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if !v.Deduplicated {
|
|
fresh++
|
|
}
|
|
if ackTime.IsZero() {
|
|
ackTime = v.AcknowledgedAt
|
|
} else if !ackTime.Equal(v.AcknowledgedAt) {
|
|
t.Fatal("ack receipt changed")
|
|
}
|
|
}
|
|
if fresh != 1 || f.subscription(sub.ID).PendingCount != 0 || f.count("orbit_audit_events", "resource_id=? AND action='delivery.ack'", winner.Delivery.ID) != 1 {
|
|
t.Fatal("duplicate ack side effects")
|
|
}
|
|
})
|
|
t.Run("lease_expiry_backoff_latest_proof_and_node_restart", func(t *testing.T) {
|
|
f := newEventFixture(t, ctx, db)
|
|
sub := f.create(events.CreateInput{MaxAttempts: 2})
|
|
f.cancelResult()
|
|
first := f.grant(sub.ID)
|
|
f.expireDelivery(first)
|
|
// 新建存储实例模拟所有连接内存丢失；权威交付依然来自数据库。
|
|
f.events = events.New(NewEventStore(db))
|
|
noEventGrant(t, f, sub.ID)
|
|
v, err := f.events.Delivery(ctx, f.token, sub.ID, first.Delivery.ID)
|
|
if err != nil || v.State != platform.DeliveryPending || v.Attempts != 1 {
|
|
t.Fatal("lease was not recovered", err)
|
|
}
|
|
f.exec(`UPDATE orbit_deliveries SET available_at=UTC_TIMESTAMP(6) WHERE namespace_id=? AND id=?`, f.token.NamespaceID, v.ID)
|
|
second := f.grant(sub.ID)
|
|
if second.Event.ID != first.Event.ID || second.Delivery.ID != first.Delivery.ID || second.Delivery.LeaseEpoch != 2 || second.Delivery.Attempts != 2 || second.LeaseSecret == first.LeaseSecret {
|
|
t.Fatal("redelivery identity/proof incorrect")
|
|
}
|
|
_, err = f.events.Ack(ctx, f.token, sub.ID, events.AckInput{DeliveryID: first.Delivery.ID, Epoch: first.Delivery.LeaseEpoch, LeaseSecret: first.LeaseSecret})
|
|
expectWorkerCode(t, err, "DELIVERY_LEASE_LOST")
|
|
f.expireDelivery(second)
|
|
noEventGrant(t, f, sub.ID)
|
|
v, err = f.events.Delivery(ctx, f.token, sub.ID, second.Delivery.ID)
|
|
if err != nil || v.State != platform.DeliveryExpired || f.subscription(sub.ID).PendingCount != 1 {
|
|
t.Fatal("exhaustion discarded event", err)
|
|
}
|
|
// 确认本身不是新的执行授权；最近证明在未被新代次替代时允许迟到确认。
|
|
f.ack(sub.ID, second)
|
|
if !f.ack(sub.ID, second).Deduplicated {
|
|
t.Fatal("late ack replay lost")
|
|
}
|
|
})
|
|
t.Run("manual_retry_adds_budget_without_resetting_attempt_or_epoch", func(t *testing.T) {
|
|
f := newEventFixture(t, ctx, db)
|
|
sub := f.create(events.CreateInput{MaxAttempts: 1})
|
|
f.cancelResult()
|
|
first := f.grant(sub.ID)
|
|
f.expireDelivery(first)
|
|
noEventGrant(t, f, sub.ID)
|
|
v, err := f.events.Delivery(ctx, f.token, sub.ID, first.Delivery.ID)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
in := events.RetryInput{ExpectedRevision: v.Revision, Reason: "receiver fixed", AdditionalAttempts: 2}
|
|
out, err := f.events.Retry(ctx, f.token, sub.ID, v.ID, "retry", in)
|
|
if err != nil || out.Delivery.Attempts != 1 || out.Delivery.AttemptBudget != 3 || out.Delivery.LeaseEpoch != 1 {
|
|
t.Fatal("budget reset", err)
|
|
}
|
|
second := f.grant(sub.ID)
|
|
if second.Delivery.Attempts != 2 || second.Delivery.LeaseEpoch != 2 {
|
|
t.Fatal("retry reused attempt")
|
|
}
|
|
again, err := f.events.Retry(ctx, f.token, sub.ID, v.ID, "retry", in)
|
|
if err != nil || !again.Deduplicated || again.Delivery.Revision != out.Delivery.Revision {
|
|
t.Fatal("retry replay changed receipt", err)
|
|
}
|
|
_, err = f.events.Retry(ctx, f.token, sub.ID, v.ID, "another", in)
|
|
expectWorkerCode(t, err, "STATE_CONFLICT")
|
|
_, err = f.events.Ack(ctx, f.token, sub.ID, events.AckInput{DeliveryID: first.Delivery.ID, Epoch: 1, LeaseSecret: first.LeaseSecret})
|
|
expectWorkerCode(t, err, "DELIVERY_LEASE_LOST")
|
|
f.ack(sub.ID, second)
|
|
if f.count("orbit_audit_events", "resource_id=? AND action='delivery.retry'", v.ID) != 1 {
|
|
t.Fatal("manual retry audit duplicated")
|
|
}
|
|
})
|
|
t.Run("pause_blocks_new_delivery_but_allows_inflight_ack", func(t *testing.T) {
|
|
f := newEventFixture(t, ctx, db)
|
|
sub := f.create(events.CreateInput{})
|
|
f.cancelResult()
|
|
g := f.grant(sub.ID)
|
|
paused, err := f.events.Control(ctx, f.token, sub.ID, "pause", events.ControlInput{ExpectedRevision: 1, Enabled: false, Reason: "receiver maintenance"})
|
|
if err != nil || paused.Subscription.Enabled {
|
|
t.Fatal(err)
|
|
}
|
|
f.cancelResult()
|
|
noEventGrant(t, f, sub.ID)
|
|
f.ack(sub.ID, g)
|
|
_, err = f.events.Control(ctx, f.token, sub.ID, "resume", events.ControlInput{ExpectedRevision: 2, Enabled: true, Reason: "receiver ready"})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
resumed := f.grant(sub.ID)
|
|
if resumed.Event.Sequence != 2 {
|
|
t.Fatal("resume lost result")
|
|
}
|
|
})
|
|
t.Run("ack_and_materialization_are_transactional_with_audit", func(t *testing.T) {
|
|
f := newEventFixture(t, ctx, db)
|
|
sub := f.create(events.CreateInput{})
|
|
f.cancelResult()
|
|
f.exec(`CREATE TRIGGER test_delivery_reject BEFORE INSERT ON orbit_deliveries FOR EACH ROW SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='injected delivery failure'`)
|
|
t.Cleanup(func() {
|
|
f.exec(`DROP TRIGGER IF EXISTS test_delivery_reject`)
|
|
f.exec(`DROP TRIGGER IF EXISTS test_delivery_ack_reject`)
|
|
})
|
|
_, err := f.events.Claim(ctx, f.token, sub.ID, events.ClaimInput{ConnectionID: executionID(t), Capacity: 1})
|
|
if err == nil {
|
|
t.Fatal("expected delivery insertion failure")
|
|
}
|
|
f.exec(`DROP TRIGGER test_delivery_reject`)
|
|
v := f.subscription(sub.ID)
|
|
if v.CursorSequence != 0 || v.PendingCount != 0 || f.count("orbit_deliveries", "subscription_id=?", sub.ID) != 0 {
|
|
t.Fatal("partial materialization escaped rollback")
|
|
}
|
|
g := f.grant(sub.ID)
|
|
f.exec(`CREATE TRIGGER test_delivery_ack_reject BEFORE INSERT ON orbit_audit_events FOR EACH ROW SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='injected audit failure'`)
|
|
_, err = f.events.Ack(ctx, f.token, sub.ID, events.AckInput{DeliveryID: g.Delivery.ID, Epoch: g.Delivery.LeaseEpoch, LeaseSecret: g.LeaseSecret})
|
|
if err == nil {
|
|
t.Fatal("ack escaped audit failure")
|
|
}
|
|
f.exec(`DROP TRIGGER test_delivery_ack_reject`)
|
|
d, err := f.events.Delivery(ctx, f.token, sub.ID, g.Delivery.ID)
|
|
if err != nil || d.State != platform.DeliveryLeased || d.AcknowledgedAt != nil || f.subscription(sub.ID).PendingCount != 1 {
|
|
t.Fatal("ack partial commit", err)
|
|
}
|
|
f.ack(sub.ID, g)
|
|
})
|
|
t.Run("producer_journal_failure_rolls_back_task_event_and_quota", func(t *testing.T) {
|
|
f := newEventFixture(t, ctx, db)
|
|
task := f.submit(1)
|
|
f.exec(`CREATE TRIGGER test_result_reject BEFORE INSERT ON orbit_result_events FOR EACH ROW SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='injected journal failure'`)
|
|
t.Cleanup(func() { f.exec(`DROP TRIGGER IF EXISTS test_result_reject`) })
|
|
_, err := f.tasks.CancelTask(ctx, f.token, task.ID, "cancel", service.CancelInput{ExpectedRevision: task.Revision, Reason: "verify atomic result"})
|
|
if err == nil {
|
|
t.Fatal("expected journal failure")
|
|
}
|
|
f.exec(`DROP TRIGGER test_result_reject`)
|
|
if f.get(task.ID).State != pulse.TaskWaiting || f.count("orbit_events", "aggregate_id=?", task.ID) != 1 || f.count("orbit_event_streams", "1=1") != 0 {
|
|
t.Fatal("result partial commit")
|
|
}
|
|
assertApplicationCounts(t, f.executionFixture, 1, 0, 0)
|
|
if _, err = f.tasks.CancelTask(ctx, f.token, task.ID, "cancel", service.CancelInput{ExpectedRevision: task.Revision, Reason: "verify atomic result"}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
})
|
|
t.Run("all_terminal_and_unknown_paths_emit_but_safe_requeue_does_not", func(t *testing.T) {
|
|
f := newEventFixture(t, ctx, db)
|
|
sub := f.create(events.CreateInput{})
|
|
f.submit(2)
|
|
first := retryStart(t, f.executionFixture)
|
|
retryFinish(t, f.executionFixture, first, "FAILED", true)
|
|
if f.count("orbit_result_events", "1=1") != 0 {
|
|
t.Fatal("safe retry published final result")
|
|
}
|
|
makeRetryDue(f.executionFixture, first.TaskID)
|
|
second := retryStart(t, f.executionFixture)
|
|
retryFinish(t, f.executionFixture, second, "FAILED", false)
|
|
g := f.grant(sub.ID)
|
|
if g.Event.Type != "task.failed" {
|
|
t.Fatal("missing failure result")
|
|
}
|
|
f.ack(sub.ID, g)
|
|
f.submit(1)
|
|
unknown := retryStart(t, f.executionFixture)
|
|
retryFinish(t, f.executionFixture, unknown, "UNKNOWN", false)
|
|
g = f.grant(sub.ID)
|
|
if g.Event.Type != "task.needs_review" || !g.Event.Snapshot.EffectUncertain {
|
|
t.Fatal("missing unknown result")
|
|
}
|
|
f.ack(sub.ID, g)
|
|
f.submit(1)
|
|
lost := retryStart(t, f.executionFixture)
|
|
f.expire(lost)
|
|
if _, err := f.storeB.Recover(ctx, 100); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
g = f.grant(sub.ID)
|
|
if g.Event.Type != "task.needs_review" {
|
|
t.Fatal("recovery failed to publish result")
|
|
}
|
|
})
|
|
t.Run("namespace_limit_future_cursor_and_history_gap_are_explicit", func(t *testing.T) {
|
|
f := newEventFixture(t, ctx, db)
|
|
future := uint64(1)
|
|
_, err := f.events.Create(ctx, f.token, "future", events.CreateInput{Name: "future", QueueID: f.resources.QueueID, FromSequence: &future})
|
|
expectWorkerCode(t, err, "INVALID_ARGUMENT")
|
|
sub := f.create(events.CreateInput{})
|
|
f.cancelResult()
|
|
f.exec(`UPDATE orbit_event_streams SET retained_from_sequence=2 WHERE namespace_id=?`, f.token.NamespaceID)
|
|
_, err = f.events.Claim(ctx, f.token, sub.ID, events.ClaimInput{ConnectionID: executionID(t), Capacity: 1})
|
|
expectWorkerCode(t, err, "EVENT_HISTORY_EXPIRED")
|
|
_, err = f.events.Create(ctx, f.token, "expired", events.CreateInput{Name: "expired", QueueID: f.resources.QueueID, FromSequence: &zero})
|
|
expectWorkerCode(t, err, "EVENT_HISTORY_EXPIRED")
|
|
for i := 1; i < events.MaxSubscriptions; i++ {
|
|
f.create(events.CreateInput{Name: fmt.Sprintf("limit-%d", i)})
|
|
}
|
|
_, err = f.events.Create(ctx, f.token, "overflow", events.CreateInput{Name: "overflow", QueueID: f.resources.QueueID})
|
|
expectWorkerCode(t, err, "SUBSCRIPTION_LIMIT")
|
|
})
|
|
|
|
t.Run("locked_subscriber_cannot_block_task_result_commit", func(t *testing.T) {
|
|
f := newEventFixture(t, ctx, db)
|
|
sub := f.create(events.CreateInput{})
|
|
task := f.submit(1)
|
|
tx, err := db.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelReadCommitted})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer tx.Rollback()
|
|
var id core.ID
|
|
if err = tx.QueryRowContext(ctx, `SELECT id FROM orbit_subscriptions WHERE namespace_id=? AND id=? FOR UPDATE`, f.token.NamespaceID, sub.ID).Scan(&id); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
bounded, stop := context.WithTimeout(ctx, 2*time.Second)
|
|
defer stop()
|
|
if _, err = f.tasks.CancelTask(bounded, f.token, task.ID, "while-locked", service.CancelInput{ExpectedRevision: task.Revision, Reason: "subscriber is slow"}); err != nil {
|
|
t.Fatal("producer waited for subscriber lock", err)
|
|
}
|
|
if err = tx.Rollback(); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
g := f.grant(sub.ID)
|
|
if g.Event.Snapshot.TaskID != task.ID {
|
|
t.Fatal("result lost while subscriber blocked")
|
|
}
|
|
})
|
|
t.Run("connection_capacity_is_database_enforced", func(t *testing.T) {
|
|
f := newEventFixture(t, ctx, db)
|
|
sub := f.create(events.CreateInput{})
|
|
f.cancelResult()
|
|
f.cancelResult()
|
|
claim := events.ClaimInput{ConnectionID: executionID(t), Capacity: 1}
|
|
first, err := f.events.Claim(ctx, f.token, sub.ID, claim)
|
|
if err != nil || first == nil {
|
|
t.Fatal(err)
|
|
}
|
|
if g, err := f.events.Claim(ctx, f.token, sub.ID, claim); err != nil || g != nil {
|
|
t.Fatal("connection capacity exceeded", err)
|
|
}
|
|
f.ack(sub.ID, first)
|
|
next, err := f.events.Claim(ctx, f.token, sub.ID, claim)
|
|
if err != nil || next == nil || next.Event.ID == first.Event.ID {
|
|
t.Fatal("capacity not released on confirmation", err)
|
|
}
|
|
})
|
|
t.Run("late_ack_races_new_delivery_without_double_confirmation", func(t *testing.T) {
|
|
f := newEventFixture(t, ctx, db)
|
|
sub := f.create(events.CreateInput{MaxPending: 1, MaxAttempts: 2})
|
|
for range 12 {
|
|
f.cancelResult()
|
|
first := f.grant(sub.ID)
|
|
f.expireDelivery(first)
|
|
noEventGrant(t, f, sub.ID)
|
|
f.exec(`UPDATE orbit_deliveries SET available_at=UTC_TIMESTAMP(6) WHERE namespace_id=? AND id=?`, f.token.NamespaceID, first.Delivery.ID)
|
|
type claimResult struct {
|
|
grant *events.Grant
|
|
err error
|
|
}
|
|
claims := make(chan claimResult, 1)
|
|
acks := make(chan error, 1)
|
|
start := make(chan struct{})
|
|
connection := executionID(t)
|
|
go func() {
|
|
<-start
|
|
g, err := f.events.Claim(ctx, f.token, sub.ID, events.ClaimInput{ConnectionID: connection, Capacity: 1})
|
|
claims <- claimResult{g, err}
|
|
}()
|
|
go func() {
|
|
<-start
|
|
_, err := f.events.Ack(ctx, f.token, sub.ID, events.AckInput{DeliveryID: first.Delivery.ID, Epoch: first.Delivery.LeaseEpoch, LeaseSecret: first.LeaseSecret})
|
|
acks <- err
|
|
}()
|
|
close(start)
|
|
c := <-claims
|
|
ackErr := <-acks
|
|
if c.err != nil {
|
|
t.Fatal(c.err)
|
|
}
|
|
if c.grant == nil {
|
|
if ackErr != nil {
|
|
t.Fatal("neither ack nor redelivery won", ackErr)
|
|
}
|
|
} else {
|
|
expectWorkerCode(t, ackErr, "DELIVERY_LEASE_LOST")
|
|
f.ack(sub.ID, c.grant)
|
|
}
|
|
if f.subscription(sub.ID).PendingCount != 0 || f.count("orbit_audit_events", "resource_id=? AND action='delivery.ack'", first.Delivery.ID) != 1 {
|
|
t.Fatal("race duplicated confirmation or leaked window")
|
|
}
|
|
}
|
|
})
|
|
t.Run("proof_is_not_stored_in_audit_or_plaintext_column", func(t *testing.T) {
|
|
f := newEventFixture(t, ctx, db)
|
|
sub := f.create(events.CreateInput{})
|
|
f.cancelResult()
|
|
g := f.grant(sub.ID)
|
|
f.ack(sub.ID, g)
|
|
var before, after string
|
|
if err := db.QueryRowContext(ctx, "SELECT `before`,`after` FROM orbit_audit_events WHERE namespace_id=? AND resource_id=? AND action='delivery.ack'", f.token.NamespaceID, g.Delivery.ID).Scan(&before, &after); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if strings.Contains(before+after, g.LeaseSecret) {
|
|
t.Fatal("delivery proof leaked to audit")
|
|
}
|
|
})
|
|
}
|
|
|
|
func TestIntegrationResultCommitOrder(t *testing.T) {
|
|
dsn := os.Getenv("ORBIT_TEST_MYSQL_DSN")
|
|
if dsn == "" {
|
|
t.Skip("requires disposable MySQL")
|
|
}
|
|
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
|
defer cancel()
|
|
db, err := Open(ctx, dsn, 16, 4)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer db.Close()
|
|
if err = Migrate(ctx, db); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
for _, rollback := range []bool{false, true} {
|
|
t.Run(fmt.Sprintf("first_rollback_%t", rollback), func(t *testing.T) {
|
|
f := newEventFixture(t, ctx, db)
|
|
sub := f.create(events.CreateInput{})
|
|
first, second := f.submit(1), f.submit(1)
|
|
finish := func(tx *sql.Tx, task service.TaskView) error {
|
|
if _, err := tx.ExecContext(ctx, `UPDATE pulse_tasks SET state='CANCELLED',completed_at=UTC_TIMESTAMP(6),revision=revision+1 WHERE namespace_id=? AND id=?`, f.token.NamespaceID, task.ID); err != nil {
|
|
return err
|
|
}
|
|
return emitEvent(ctx, tx, f.token.NamespaceID, task.ID, task.Revision+1, "task", "task.cancelled", map[string]any{"state": "CANCELLED"})
|
|
}
|
|
tx, err := db.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelReadCommitted})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer tx.Rollback()
|
|
if err = finish(tx, first); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
started := make(chan struct{})
|
|
result := make(chan error, 1)
|
|
go func() {
|
|
result <- WithTx(ctx, db, func(other *sql.Tx) error { close(started); return finish(other, second) })
|
|
}()
|
|
<-started
|
|
select {
|
|
case err := <-result:
|
|
t.Fatalf("later transaction crossed uncommitted stream: %v", err)
|
|
case <-time.After(100 * time.Millisecond):
|
|
}
|
|
// 已提交读在第一事务未提交时看不到较大的序号；消费也不等待生产者流锁。
|
|
noEventGrant(t, f, sub.ID)
|
|
if rollback {
|
|
err = tx.Rollback()
|
|
} else {
|
|
err = tx.Commit()
|
|
}
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err = <-result; err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
want := uint64(2)
|
|
if rollback {
|
|
want = 1
|
|
}
|
|
var head uint64
|
|
if err = db.QueryRowContext(ctx, `SELECT last_sequence FROM orbit_event_streams WHERE namespace_id=?`, f.token.NamespaceID).Scan(&head); err != nil || head != want {
|
|
t.Fatal("nontransactional sequence", err)
|
|
}
|
|
for seq := uint64(1); seq <= want; seq++ {
|
|
g := f.grant(sub.ID)
|
|
if g.Event.Sequence != seq {
|
|
t.Fatal("commit order skipped result")
|
|
}
|
|
f.ack(sub.ID, g)
|
|
}
|
|
})
|
|
}
|
|
t.Run("create_new_boundary_waits_for_earlier_producer", func(t *testing.T) {
|
|
f := newEventFixture(t, ctx, db)
|
|
task := f.submit(1)
|
|
tx, err := db.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelReadCommitted})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer tx.Rollback()
|
|
if _, err = tx.ExecContext(ctx, `UPDATE pulse_tasks SET state='CANCELLED',completed_at=UTC_TIMESTAMP(6),revision=2 WHERE namespace_id=? AND id=?`, f.token.NamespaceID, task.ID); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err = emitEvent(ctx, tx, f.token.NamespaceID, task.ID, 2, "task", "task.cancelled", map[string]any{}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
type result struct {
|
|
v events.SubscriptionResult
|
|
err error
|
|
}
|
|
ch := make(chan result, 1)
|
|
go func() {
|
|
v, err := f.events.Create(ctx, f.token, "boundary", events.CreateInput{Name: "boundary", QueueID: f.resources.QueueID})
|
|
ch <- result{v, err}
|
|
}()
|
|
select {
|
|
case <-ch:
|
|
t.Fatal("subscription creation crossed stream boundary")
|
|
case <-time.After(100 * time.Millisecond):
|
|
}
|
|
if err = tx.Commit(); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
r := <-ch
|
|
if r.err != nil || r.v.Subscription.CursorSequence != 1 {
|
|
t.Fatal("boundary did not use committed head", r.err)
|
|
}
|
|
noEventGrant(t, f, r.v.Subscription.ID)
|
|
})
|
|
}
|