191 lines
7.2 KiB
Go
191 lines
7.2 KiB
Go
package orbit
|
||
|
||
import (
|
||
"context"
|
||
"encoding/json"
|
||
"errors"
|
||
"sync/atomic"
|
||
"testing"
|
||
"time"
|
||
)
|
||
|
||
func integrationEventSubscription(t *testing.T, ctx context.Context, api *Client, maxAttempts uint32) Subscription {
|
||
t.Helper()
|
||
queue, err := api.CreateQueue(ctx, workerTestID(t), CreateQueueRequest{Name: "event-sdk-" + workerTestID(t), MaxConcurrency: 1, RatePerSecond: 10})
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
sub, err := api.CreateSubscription(ctx, workerTestID(t), CreateSubscriptionRequest{Name: "sdk-" + workerTestID(t), QueueID: queue.Queue.ID, Types: []string{"task.cancelled"}, MaxPending: 10, MaxAttempts: maxAttempts})
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
return sub.Subscription
|
||
}
|
||
func integrationCancelledEvent(t *testing.T, ctx context.Context, api *Client, sub Subscription) Task {
|
||
t.Helper()
|
||
task, err := api.SubmitTask(ctx, workerTestID(t), SubmitTaskRequest{QueueID: sub.QueueID, Handler: "object.prepare", HandlerVersion: "v1", PayloadVersion: "v1", Payload: json.RawMessage(`{}`)})
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
cancelled, err := api.CancelTask(ctx, task.Task.ID, workerTestID(t), CancelTaskRequest{ExpectedRevision: task.Task.Revision, Reason: "subscription SDK regression"})
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
return cancelled.Task
|
||
}
|
||
func eventDialIntegration(t *testing.T, ctx context.Context, api *Client, sub Subscription, capacity uint32) *EventConnection {
|
||
t.Helper()
|
||
c, err := api.DialEvents(ctx, EventRegistration{SubscriptionID: sub.ID, Capacity: capacity})
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
t.Cleanup(func() { c.Close() })
|
||
return c
|
||
}
|
||
func TestIntegrationEventConnectionSDK(t *testing.T) {
|
||
api, _, _ := integrationWorkerClient(t)
|
||
ctx, stop := context.WithTimeout(context.Background(), 15*time.Second)
|
||
defer stop()
|
||
sub := integrationEventSubscription(t, ctx, api, 3)
|
||
task := integrationCancelledEvent(t, ctx, api, sub)
|
||
a := eventDialIntegration(t, ctx, api, sub, 1)
|
||
grant, err := a.Next(ctx)
|
||
if err != nil || grant.Event.Snapshot.TaskID != task.ID || grant.Event.Snapshot.Revision != task.Revision || grant.Event.Type != "task.cancelled" {
|
||
t.Fatal("SDK snapshot mismatch", err)
|
||
}
|
||
paused, err := api.ConfigureSubscription(ctx, sub.ID, workerTestID(t), ConfigureSubscriptionRequest{ExpectedRevision: sub.Revision, Reason: "pause with active delivery", Enabled: false})
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
pong, err := a.Ping(ctx, workerTestID(t))
|
||
if err != nil || pong.Enabled {
|
||
t.Fatal("paused subscription heartbeat", err)
|
||
}
|
||
first, err := a.Ack(ctx, workerTestID(t), grant.Proof())
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
a.Close()
|
||
b := eventDialIntegration(t, ctx, api, sub, 1)
|
||
duplicate, err := b.Ack(ctx, workerTestID(t), grant.Proof())
|
||
if err != nil || !duplicate.Deduplicated || !duplicate.AcknowledgedAt.Equal(first.AcknowledgedAt) {
|
||
t.Fatal("ACK proof could not cross connections", err)
|
||
}
|
||
_ = integrationCancelledEvent(t, ctx, api, sub)
|
||
short, cancel := context.WithTimeout(ctx, 100*time.Millisecond)
|
||
_, err = b.Next(short)
|
||
cancel()
|
||
if !errors.Is(err, context.DeadlineExceeded) || b.Err() != nil {
|
||
t.Fatal("paused subscription delivered or Next closed connection", err)
|
||
}
|
||
if _, err = api.ConfigureSubscription(ctx, sub.ID, workerTestID(t), ConfigureSubscriptionRequest{ExpectedRevision: paused.Subscription.Revision, Reason: "resume", Enabled: true}); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
second, err := b.Next(ctx)
|
||
if err != nil || second.Event.ID == grant.Event.ID {
|
||
t.Fatal("resume failed", err)
|
||
}
|
||
if _, err = b.Ack(ctx, workerTestID(t), second.Proof()); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
current, err := api.GetSubscription(ctx, sub.ID)
|
||
if err != nil || current.PendingCount != 0 {
|
||
t.Fatal("SDK acknowledgement accounting", err)
|
||
}
|
||
}
|
||
func TestIntegrationEventConsumerHeartbeatAndLostAck(t *testing.T) {
|
||
api, _, token := integrationWorkerClient(t)
|
||
ctx, stop := context.WithTimeout(context.Background(), 25*time.Second)
|
||
defer stop()
|
||
sub := integrationEventSubscription(t, ctx, api, 3)
|
||
_ = integrationCancelledEvent(t, ctx, api, sub)
|
||
gateway := newProtocolFaultGateway(t, api.baseURL, "/v1/events/connect", eventSubprotocol)
|
||
gateway.drop.Store("event.ack")
|
||
client, err := NewClient(gateway.server.URL, Options{Token: token, HTTPClient: gateway.server.Client()})
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
var calls atomic.Int64
|
||
consumer, err := NewEventConsumer(client, EventConsumerOptions{SubscriptionID: sub.ID, Capacity: 1, ShutdownTimeout: time.Second, Handle: func(work context.Context, event ResultEvent) error {
|
||
calls.Add(1)
|
||
// 慢业务期间真实等待15秒自动心跳；确认计数来自上游服务端ACK，不是客户端写入。
|
||
ticker := time.NewTicker(20 * time.Millisecond)
|
||
defer ticker.Stop()
|
||
for gateway.confirmedEventPings.Load() == 0 {
|
||
select {
|
||
case <-work.Done():
|
||
return work.Err()
|
||
case <-ticker.C:
|
||
}
|
||
}
|
||
return nil
|
||
}})
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
run, cancel := context.WithCancel(ctx)
|
||
defer cancel()
|
||
done := make(chan error, 1)
|
||
go func() { done <- consumer.Run(run) }()
|
||
for consumer.Stats().Acknowledged == 0 {
|
||
select {
|
||
case <-ctx.Done():
|
||
t.Fatal("consumer did not reconcile committed ACK")
|
||
case err := <-done:
|
||
t.Fatal("consumer stopped", err)
|
||
case <-time.After(20 * time.Millisecond):
|
||
}
|
||
}
|
||
cancel()
|
||
if err := awaitConsumerStop(t, done); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if calls.Load() != 1 || consumer.Stats().Reconnects < 1 || gateway.confirmedEventPings.Load() < 1 {
|
||
t.Fatal("heartbeat or reconciliation missing", consumer.Stats())
|
||
}
|
||
page, err := api.ListEventDeliveries(ctx, sub.ID, DeliveryPageRequest{State: DeliveryAcked})
|
||
if err != nil || len(page.Items) != 1 || page.Items[0].Attempts != 1 {
|
||
t.Fatal("lost ACK created another delivery attempt", err)
|
||
}
|
||
t.Log("real 15-second heartbeat and committed ACK loss recovered; handler and delivery attempt each remained one")
|
||
}
|
||
func TestIntegrationEventConsumerFailureRedeliversNaturally(t *testing.T) {
|
||
api, _, _ := integrationWorkerClient(t)
|
||
ctx, stop := context.WithTimeout(context.Background(), 45*time.Second)
|
||
defer stop()
|
||
sub := integrationEventSubscription(t, ctx, api, 2)
|
||
_ = integrationCancelledEvent(t, ctx, api, sub)
|
||
var calls atomic.Int64
|
||
consumer, err := NewEventConsumer(api, EventConsumerOptions{SubscriptionID: sub.ID, Capacity: 1, ShutdownTimeout: time.Second, Handle: func(context.Context, ResultEvent) error {
|
||
if calls.Add(1) == 1 {
|
||
return errors.New("receiver transaction rolled back")
|
||
}
|
||
return nil
|
||
}})
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
run, cancel := context.WithCancel(ctx)
|
||
defer cancel()
|
||
done := make(chan error, 1)
|
||
go func() { done <- consumer.Run(run) }()
|
||
for consumer.Stats().Acknowledged == 0 {
|
||
select {
|
||
case <-ctx.Done():
|
||
t.Fatal("delivery did not retry after natural lease expiry")
|
||
case err := <-done:
|
||
t.Fatal("consumer stopped", err)
|
||
case <-time.After(50 * time.Millisecond):
|
||
}
|
||
}
|
||
cancel()
|
||
if err := awaitConsumerStop(t, done); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
page, err := api.ListEventDeliveries(ctx, sub.ID, DeliveryPageRequest{State: DeliveryAcked})
|
||
if err != nil || len(page.Items) != 1 || page.Items[0].Attempts != 2 || calls.Load() != 2 || consumer.Stats().HandlerFailures != 1 {
|
||
t.Fatal("handler error or retry budget mishandled", err)
|
||
}
|
||
t.Log("handler error caused no ACK; same delivery reappeared after its real 30-second lease and backoff, then acknowledged at attempt two")
|
||
}
|