199 lines
8.0 KiB
Go
199 lines
8.0 KiB
Go
package orbit
|
||
|
||
import (
|
||
"context"
|
||
"encoding/json"
|
||
"net/http"
|
||
"strings"
|
||
"testing"
|
||
"time"
|
||
|
||
"github.com/coder/websocket"
|
||
)
|
||
|
||
// 该测试用原始WSS帧驱动交付，用于验证管理SDK；不是待交付的订阅连接SDK替代品。
|
||
type managementEventFrame struct {
|
||
Version string `json:"protocol_version"`
|
||
Type string `json:"type"`
|
||
MessageID string `json:"message_id"`
|
||
SessionID string `json:"session_id,omitempty"`
|
||
ReplyTo string `json:"reply_to,omitempty"`
|
||
Body json.RawMessage `json:"body,omitempty"`
|
||
Error *struct {
|
||
Code string `json:"code"`
|
||
} `json:"error,omitempty"`
|
||
}
|
||
|
||
func TestIntegrationSubscriptionManagementSDK(t *testing.T) {
|
||
api, _, _ := integrationWorkerClient(t)
|
||
ctx, stop := context.WithTimeout(context.Background(), 45*time.Second)
|
||
defer stop()
|
||
queue, err := api.CreateQueue(ctx, workerTestID(t), CreateQueueRequest{Name: "subscription-" + workerTestID(t), MaxConcurrency: 1, RatePerSecond: 10})
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
key := workerTestID(t)
|
||
in := CreateSubscriptionRequest{Name: "results-" + workerTestID(t), QueueID: queue.Queue.ID, MaxPending: 1, MaxAttempts: 1, Types: []string{"task.cancelled"}}
|
||
created, err := api.CreateSubscription(ctx, key, in)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
duplicate, err := api.CreateSubscription(ctx, key, in)
|
||
if err != nil || !duplicate.Deduplicated || duplicate.Subscription.ID != created.Subscription.ID {
|
||
t.Fatal("SDK subscription create replay", err)
|
||
}
|
||
sub := created.Subscription
|
||
current, err := api.GetSubscription(ctx, sub.ID)
|
||
if err != nil || current.ID != sub.ID || !current.Enabled {
|
||
t.Fatal("SDK subscription get", err)
|
||
}
|
||
page, err := api.ListSubscriptions(ctx, queue.Queue.ID, SubscriptionPageRequest{Limit: 1})
|
||
if err != nil || len(page.Items) != 1 || page.Items[0].ID != sub.ID {
|
||
t.Fatal("SDK subscription list", err)
|
||
}
|
||
paused, err := api.ConfigureSubscription(ctx, sub.ID, workerTestID(t), ConfigureSubscriptionRequest{ExpectedRevision: sub.Revision, Reason: "maintenance", Enabled: false})
|
||
if err != nil || paused.Subscription.Enabled {
|
||
t.Fatal("SDK pause", err)
|
||
}
|
||
resumed, err := api.ConfigureSubscription(ctx, sub.ID, workerTestID(t), ConfigureSubscriptionRequest{ExpectedRevision: paused.Subscription.Revision, Reason: "ready", Enabled: true})
|
||
if err != nil || !resumed.Subscription.Enabled {
|
||
t.Fatal("SDK resume", err)
|
||
}
|
||
task, err := api.SubmitTask(ctx, workerTestID(t), SubmitTaskRequest{QueueID: queue.Queue.ID, Handler: "object.prepare", HandlerVersion: "v1", PayloadVersion: "v1", Payload: json.RawMessage(`{}`)})
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if _, err = api.CancelTask(ctx, task.Task.ID, workerTestID(t), CancelTaskRequest{ExpectedRevision: task.Task.Revision, Reason: "create result"}); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
send := func(conn *websocket.Conn, session, kind string, body any) string {
|
||
t.Helper()
|
||
id := workerTestID(t)
|
||
raw, err := marshalWire(body)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
frame, err := marshalWire(managementEventFrame{Version: "1", Type: kind, MessageID: id, SessionID: session, Body: raw})
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if err = conn.Write(ctx, websocket.MessageText, frame); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
return id
|
||
}
|
||
read := func(conn *websocket.Conn) managementEventFrame {
|
||
t.Helper()
|
||
bounded, cancel := context.WithTimeout(ctx, 5*time.Second)
|
||
defer cancel()
|
||
_, raw, err := conn.Read(bounded)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
var frame managementEventFrame
|
||
if json.Unmarshal(raw, &frame) != nil || frame.Version != "1" || !validID(frame.MessageID) {
|
||
t.Fatal("invalid management test frame")
|
||
}
|
||
return frame
|
||
}
|
||
connect := func() (*websocket.Conn, string) {
|
||
t.Helper()
|
||
headers := http.Header{}
|
||
headers.Set("Authorization", "Bearer "+api.token)
|
||
endpoint := "ws" + strings.TrimPrefix(api.baseURL, "http") + "/v1/events/connect"
|
||
conn, _, err := websocket.Dial(ctx, endpoint, &websocket.DialOptions{HTTPClient: api.http, HTTPHeader: headers, Subprotocols: []string{"orbit.events.v1"}})
|
||
if err != nil {
|
||
t.Fatal("management test event dial failed")
|
||
}
|
||
t.Cleanup(func() { conn.CloseNow() })
|
||
conn.SetReadLimit(512 << 10)
|
||
msg := send(conn, "", "events.hello", map[string]any{"subscription_id": sub.ID, "capacity": 1})
|
||
welcome := read(conn)
|
||
if welcome.Type != "events.welcome" || welcome.ReplyTo != msg || !validID(welcome.SessionID) {
|
||
t.Fatal("management test welcome missing")
|
||
}
|
||
return conn, welcome.SessionID
|
||
}
|
||
type grant struct {
|
||
Delivery EventDelivery `json:"delivery"`
|
||
Event ResultEvent `json:"event"`
|
||
Secret string `json:"lease_secret"`
|
||
}
|
||
a, _ := connect()
|
||
frame := read(a)
|
||
var first grant
|
||
if frame.Type != "event.deliver" || json.Unmarshal(frame.Body, &first) != nil || first.Event.Snapshot.TaskID != task.Task.ID {
|
||
t.Fatal("management test delivery missing")
|
||
}
|
||
a.CloseNow()
|
||
b, session := connect()
|
||
// 使用实际30秒租约，不改数据库时钟或手工更新租约时间。
|
||
var expired EventDelivery
|
||
for {
|
||
deliveries, err := api.ListEventDeliveries(ctx, sub.ID, DeliveryPageRequest{State: DeliveryExpired, Limit: 1})
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if len(deliveries.Items) == 1 {
|
||
expired = deliveries.Items[0]
|
||
break
|
||
}
|
||
select {
|
||
case <-ctx.Done():
|
||
t.Fatal("delivery did not expire through real WSS recovery")
|
||
case <-time.After(150 * time.Millisecond):
|
||
}
|
||
}
|
||
detail, err := api.GetEventDelivery(ctx, sub.ID, expired.ID)
|
||
if err != nil || detail.Event.ID != first.Event.ID || detail.Delivery.Attempts != 1 || detail.Event.Snapshot.State != TaskCancelled {
|
||
t.Fatal("SDK delivery detail", err)
|
||
}
|
||
retryKey := workerTestID(t)
|
||
retry := RetryDeliveryRequest{ExpectedRevision: expired.Revision, Reason: "receiver fixed", AdditionalAttempts: 1}
|
||
restored, err := api.RetryEventDelivery(ctx, sub.ID, expired.ID, retryKey, retry)
|
||
if err != nil || restored.Delivery.Attempts != 1 || restored.Delivery.AttemptBudget != 2 {
|
||
t.Fatal("SDK retry reset history", err)
|
||
}
|
||
replay, err := api.RetryEventDelivery(ctx, sub.ID, expired.ID, retryKey, retry)
|
||
if err != nil || !replay.Deduplicated || replay.Delivery.Revision != restored.Delivery.Revision {
|
||
t.Fatal("SDK retry replay", err)
|
||
}
|
||
frame = read(b)
|
||
var second grant
|
||
if frame.Type != "event.deliver" || json.Unmarshal(frame.Body, &second) != nil || second.Event.ID != first.Event.ID || second.Delivery.LeaseEpoch != 2 {
|
||
t.Fatal("manual retry failed to deliver")
|
||
}
|
||
ackBody := map[string]any{"delivery_id": second.Delivery.ID, "epoch": "2", "lease_secret": second.Secret}
|
||
msg := send(b, session, "event.ack", ackBody)
|
||
ack := read(b)
|
||
if ack.Type != "ack" || ack.ReplyTo != msg {
|
||
t.Fatal("management test ACK failed")
|
||
}
|
||
delivered, err := api.ListEventDeliveries(ctx, sub.ID, DeliveryPageRequest{State: DeliveryAcked})
|
||
if err != nil || len(delivered.Items) != 1 || delivered.Items[0].ID != expired.ID {
|
||
t.Fatal("SDK acknowledged history", err)
|
||
}
|
||
current, err = api.GetSubscription(ctx, sub.ID)
|
||
if err != nil || current.PendingCount != 0 || current.Revision != resumed.Subscription.Revision {
|
||
t.Fatal("SDK pending count or config version", err)
|
||
}
|
||
retireKey := workerTestID(t)
|
||
retireInput := RetireResourceRequest{ExpectedRevision: current.Revision, Reason: "completed SDK subscription regression"}
|
||
retired, err := api.RetireSubscription(ctx, sub.ID, retireKey, retireInput)
|
||
if err != nil || retired.Subscription.RetiredAt == nil {
|
||
t.Fatal("SDK retire subscription", err)
|
||
}
|
||
repeated, err := api.RetireSubscription(ctx, sub.ID, retireKey, retireInput)
|
||
if err != nil || !repeated.Deduplicated {
|
||
t.Fatal("SDK retire replay", err)
|
||
}
|
||
if _, err = api.GetEventDelivery(ctx, sub.ID, expired.ID); err != nil {
|
||
t.Fatal("retired delivery history", err)
|
||
}
|
||
queueRetired, err := api.RetireQueue(ctx, queue.Queue.ID, workerTestID(t), RetireResourceRequest{ExpectedRevision: queue.Queue.Revision, Reason: "completed SDK queue regression"})
|
||
if err != nil || queueRetired.Queue.RetiredAt == nil {
|
||
t.Fatal("SDK retire queue", err)
|
||
}
|
||
t.Log("management SDK exercised create/get/list/control/detail/expired retry with a real 30-second delivery lease")
|
||
}
|