feat: expose result subscription HTTP and WebSocket APIs
This commit is contained in:
@@ -67,7 +67,7 @@ func TestIntegrationApplicationResourcesUpgrade(t *testing.T) {
|
||||
f.exec(`DROP TABLE orbit_application_resources`)
|
||||
f.exec(`DROP TABLE orbit_result_events,orbit_event_streams`)
|
||||
f.exec(`ALTER TABLE orbit_subscriptions DROP CHECK ck_subscription_window, DROP COLUMN queue_id, DROP COLUMN cursor_sequence, DROP COLUMN pending_count, DROP COLUMN max_pending, DROP COLUMN max_attempts`)
|
||||
f.exec(`ALTER TABLE orbit_deliveries DROP INDEX ix_delivery_subscription_due, DROP INDEX ix_delivery_subscription_lease, DROP CHECK ck_delivery_budget, DROP COLUMN attempt_budget, DROP COLUMN token_digest`)
|
||||
f.exec(`ALTER TABLE orbit_deliveries DROP INDEX ix_delivery_history, DROP INDEX ix_delivery_state_history, DROP INDEX ix_delivery_subscription_due, DROP INDEX ix_delivery_subscription_lease, DROP CHECK ck_delivery_budget, DROP COLUMN attempt_budget, DROP COLUMN token_digest`)
|
||||
f.exec(`ALTER TABLE pulse_tasks DROP INDEX ix_task_priority_dispatch, DROP CHECK ck_task_priority, DROP COLUMN dispatch_at`)
|
||||
f.exec(`ALTER TABLE pulse_tasks DROP CHECK ck_task_concurrency_key, DROP COLUMN concurrency_key, DROP COLUMN concurrency_key_digest`)
|
||||
f.exec(`ALTER TABLE pulse_attempts DROP COLUMN concurrency_token`)
|
||||
|
||||
@@ -98,12 +98,12 @@ type storedDelivery struct {
|
||||
|
||||
func readDelivery(ctx context.Context, tx *sql.Tx, ns, sub, id core.ID, lock bool) (storedDelivery, error) {
|
||||
var out storedDelivery
|
||||
q := `SELECT id,subscription_id,event_id,revision,state,available_at,lease_epoch,lease_until,attempts,attempt_budget,acknowledged_at,token_digest FROM orbit_deliveries WHERE namespace_id=? AND subscription_id=? AND id=?`
|
||||
q := `SELECT id,subscription_id,event_id,revision,state,available_at,lease_epoch,lease_until,attempts,attempt_budget,acknowledged_at,token_digest,created_at FROM orbit_deliveries WHERE namespace_id=? AND subscription_id=? AND id=?`
|
||||
if lock {
|
||||
q += ` FOR UPDATE`
|
||||
}
|
||||
v := &out.view
|
||||
err := tx.QueryRowContext(ctx, q, ns, sub, id).Scan(&v.ID, &v.SubscriptionID, &v.EventID, &v.Revision, &v.State, &v.AvailableAt, &v.LeaseEpoch, &v.LeaseUntil, &v.Attempts, &v.AttemptBudget, &v.AcknowledgedAt, &out.digest)
|
||||
err := tx.QueryRowContext(ctx, q, ns, sub, id).Scan(&v.ID, &v.SubscriptionID, &v.EventID, &v.Revision, &v.State, &v.AvailableAt, &v.LeaseEpoch, &v.LeaseUntil, &v.Attempts, &v.AttemptBudget, &v.AcknowledgedAt, &out.digest, &v.CreatedAt)
|
||||
if errors.Is(err, sql.ErrNoRows) {
|
||||
return out, fault.NotFound
|
||||
}
|
||||
|
||||
179
internal/storage/mysql/event_queries.go
Normal file
179
internal/storage/mysql/event_queries.go
Normal file
@@ -0,0 +1,179 @@
|
||||
package mysql
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"database/sql"
|
||||
"encoding/base64"
|
||||
"encoding/json"
|
||||
"time"
|
||||
|
||||
"git.yuntuo.hk/k2safe/orbit/internal/fault"
|
||||
"git.yuntuo.hk/k2safe/orbit/internal/jsonvalue"
|
||||
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"
|
||||
)
|
||||
|
||||
type eventCursor struct {
|
||||
Version int `json:"v"`
|
||||
Kind string `json:"k"`
|
||||
Namespace core.ID `json:"n"`
|
||||
Credential core.ID `json:"c"`
|
||||
Parent core.ID `json:"p"`
|
||||
State platform.DeliveryState `json:"s"`
|
||||
At time.Time `json:"t"`
|
||||
ID core.ID `json:"i"`
|
||||
}
|
||||
|
||||
func parseEventCursor(raw string, t auth.Token, parent core.ID, kind string, state platform.DeliveryState) (eventCursor, error) {
|
||||
var c eventCursor
|
||||
b, err := base64.RawURLEncoding.Strict().DecodeString(raw)
|
||||
if err != nil || len(b) > 1024 {
|
||||
return c, fault.Invalid("invalid event cursor")
|
||||
}
|
||||
if _, err = jsonvalue.Canonical(b); err != nil {
|
||||
return c, fault.Invalid("invalid event cursor")
|
||||
}
|
||||
var fields map[string]json.RawMessage
|
||||
if json.Unmarshal(b, &fields) != nil || len(fields) != 8 {
|
||||
return c, fault.Invalid("invalid event cursor")
|
||||
}
|
||||
for _, k := range []string{"v", "k", "n", "c", "p", "s", "t", "i"} {
|
||||
if _, ok := fields[k]; !ok {
|
||||
return c, fault.Invalid("invalid event cursor")
|
||||
}
|
||||
}
|
||||
d := json.NewDecoder(bytes.NewReader(b))
|
||||
d.DisallowUnknownFields()
|
||||
if d.Decode(&c) != nil || c.Version != 1 || c.Kind != kind || c.Namespace != t.NamespaceID || c.Credential != t.CredentialID || c.Parent != parent || c.State != state || c.ID.Validate() != nil || c.At.Year() < 2000 || c.At.Year() > 9998 || c.At.Nanosecond()%1000 != 0 {
|
||||
return c, fault.Invalid("cursor does not match event query")
|
||||
}
|
||||
return c, nil
|
||||
}
|
||||
func makeEventCursor(t auth.Token, parent core.ID, kind string, state platform.DeliveryState, at time.Time, id core.ID) (string, error) {
|
||||
b, err := json.Marshal(eventCursor{1, kind, t.NamespaceID, t.CredentialID, parent, state, at, id})
|
||||
return base64.RawURLEncoding.EncodeToString(b), err
|
||||
}
|
||||
func (s *EventStore) AuthorizeEvents(ctx context.Context, t auth.Token) error {
|
||||
return s.transaction(ctx, t, "events:consume", func(*sql.Tx, auth.Permissions) error { return nil })
|
||||
}
|
||||
func (s *EventStore) BindConsumer(ctx context.Context, t auth.Token, id core.ID) (events.ConsumerBinding, error) {
|
||||
var out events.ConsumerBinding
|
||||
err := s.transaction(ctx, t, "events:consume", func(tx *sql.Tx, p auth.Permissions) error {
|
||||
sub, err := readSubscription(ctx, tx, t, p, id, false)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
out.SubscriptionID = sub.ID
|
||||
out.QueueID = sub.QueueID
|
||||
out.Enabled = sub.Enabled
|
||||
return tx.QueryRowContext(ctx, `SELECT application_id,UTC_TIMESTAMP(6) FROM orbit_namespaces WHERE id=?`, t.NamespaceID).Scan(&out.ApplicationID, &out.ServerTime)
|
||||
})
|
||||
return out, err
|
||||
}
|
||||
func (s *EventStore) ListSubscriptions(ctx context.Context, t auth.Token, queue core.ID, in events.PageInput) (events.SubscriptionPage, error) {
|
||||
out := events.SubscriptionPage{Items: []events.Subscription{}}
|
||||
err := s.transaction(ctx, t, "events:read", func(tx *sql.Tx, p auth.Permissions) error {
|
||||
if !p.AllowsQueue(queue) {
|
||||
return fault.Forbidden
|
||||
}
|
||||
q := `SELECT ` + subscriptionColumns + ` FROM orbit_subscriptions WHERE namespace_id=? AND credential_id=? AND queue_id=?`
|
||||
args := []any{t.NamespaceID, t.CredentialID, queue}
|
||||
if in.Cursor != "" {
|
||||
c, err := parseEventCursor(in.Cursor, t, queue, "subscriptions", "")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
q += ` AND (created_at<? OR (created_at=? AND id<?))`
|
||||
args = append(args, c.At, c.At, c.ID)
|
||||
}
|
||||
q += ` ORDER BY created_at DESC,id DESC LIMIT ?`
|
||||
args = append(args, in.Limit+1)
|
||||
rows, err := tx.QueryContext(ctx, q, args...)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer rows.Close()
|
||||
for rows.Next() {
|
||||
v, err := scanSubscription(rows)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
out.Items = append(out.Items, v)
|
||||
}
|
||||
if err = rows.Err(); err != nil {
|
||||
return err
|
||||
}
|
||||
if len(out.Items) > in.Limit {
|
||||
out.Items = out.Items[:in.Limit]
|
||||
last := out.Items[len(out.Items)-1]
|
||||
out.NextCursor, err = makeEventCursor(t, queue, "subscriptions", "", last.CreatedAt, last.ID)
|
||||
}
|
||||
return err
|
||||
})
|
||||
return out, err
|
||||
}
|
||||
func (s *EventStore) ListDeliveries(ctx context.Context, t auth.Token, id core.ID, in events.PageInput) (events.DeliveryPage, error) {
|
||||
out := events.DeliveryPage{Items: []events.Delivery{}}
|
||||
err := s.transaction(ctx, t, "events:read", func(tx *sql.Tx, p auth.Permissions) error {
|
||||
if _, err := readSubscription(ctx, tx, t, p, id, false); err != nil {
|
||||
return err
|
||||
}
|
||||
q := `SELECT d.id,d.subscription_id,d.event_id,r.sequence,d.revision,d.state,d.available_at,d.lease_epoch,d.lease_until,d.attempts,d.attempt_budget,d.acknowledged_at,d.created_at FROM orbit_deliveries d LEFT JOIN orbit_result_events r ON r.namespace_id=d.namespace_id AND r.id=d.event_id WHERE d.namespace_id=? AND d.subscription_id=?`
|
||||
args := []any{t.NamespaceID, id}
|
||||
if in.State != "" {
|
||||
q += ` AND d.state=?`
|
||||
args = append(args, in.State)
|
||||
}
|
||||
if in.Cursor != "" {
|
||||
c, err := parseEventCursor(in.Cursor, t, id, "deliveries", in.State)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
q += ` AND (d.created_at<? OR (d.created_at=? AND d.id<?))`
|
||||
args = append(args, c.At, c.At, c.ID)
|
||||
}
|
||||
q += ` ORDER BY d.created_at DESC,d.id DESC LIMIT ?`
|
||||
args = append(args, in.Limit+1)
|
||||
rows, err := tx.QueryContext(ctx, q, args...)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer rows.Close()
|
||||
for rows.Next() {
|
||||
var v events.Delivery
|
||||
if err = rows.Scan(&v.ID, &v.SubscriptionID, &v.EventID, &v.Sequence, &v.Revision, &v.State, &v.AvailableAt, &v.LeaseEpoch, &v.LeaseUntil, &v.Attempts, &v.AttemptBudget, &v.AcknowledgedAt, &v.CreatedAt); err != nil {
|
||||
return err
|
||||
}
|
||||
out.Items = append(out.Items, v)
|
||||
}
|
||||
if err = rows.Err(); err != nil {
|
||||
return err
|
||||
}
|
||||
if len(out.Items) > in.Limit {
|
||||
out.Items = out.Items[:in.Limit]
|
||||
last := out.Items[len(out.Items)-1]
|
||||
out.NextCursor, err = makeEventCursor(t, id, "deliveries", in.State, last.CreatedAt, last.ID)
|
||||
}
|
||||
return err
|
||||
})
|
||||
return out, err
|
||||
}
|
||||
func (s *EventStore) GetDeliveryDetail(ctx context.Context, t auth.Token, sub, id core.ID) (events.DeliveryDetail, error) {
|
||||
var out events.DeliveryDetail
|
||||
err := s.transaction(ctx, t, "events:read", func(tx *sql.Tx, p auth.Permissions) error {
|
||||
if _, err := readSubscription(ctx, tx, t, p, sub, false); err != nil {
|
||||
return err
|
||||
}
|
||||
d, err := readDelivery(ctx, tx, t.NamespaceID, sub, id, false)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
out.Delivery = d.view
|
||||
out.Event, err = readResultEvent(ctx, tx, t.NamespaceID, d.view.EventID)
|
||||
return err
|
||||
})
|
||||
return out, err
|
||||
}
|
||||
@@ -27,13 +27,11 @@ var _ events.Store = (*EventStore)(nil)
|
||||
const subscriptionColumns = `id,queue_id,credential_id,name,revision,enabled,filter,cursor_sequence,pending_count,max_pending,max_attempts,created_at,updated_at`
|
||||
|
||||
func readSubscription(ctx context.Context, tx *sql.Tx, t auth.Token, p auth.Permissions, id core.ID, lock bool) (events.Subscription, error) {
|
||||
var out events.Subscription
|
||||
var raw []byte
|
||||
q := `SELECT ` + subscriptionColumns + ` FROM orbit_subscriptions WHERE namespace_id=? AND id=? AND credential_id=? AND queue_id IS NOT NULL`
|
||||
if lock {
|
||||
q += ` FOR UPDATE`
|
||||
}
|
||||
err := tx.QueryRowContext(ctx, q, t.NamespaceID, id, t.CredentialID).Scan(&out.ID, &out.QueueID, &out.CredentialID, &out.Name, &out.Revision, &out.Enabled, &raw, &out.CursorSequence, &out.PendingCount, &out.MaxPending, &out.MaxAttempts, &out.CreatedAt, &out.UpdatedAt)
|
||||
out, err := scanSubscription(tx.QueryRowContext(ctx, q, t.NamespaceID, id, t.CredentialID))
|
||||
if errors.Is(err, sql.ErrNoRows) {
|
||||
return out, fault.NotFound
|
||||
}
|
||||
@@ -43,10 +41,17 @@ func readSubscription(ctx context.Context, tx *sql.Tx, t auth.Token, p auth.Perm
|
||||
if !p.AllowsQueue(out.QueueID) {
|
||||
return events.Subscription{}, fault.NotFound
|
||||
}
|
||||
if err = json.Unmarshal(raw, &out.Types); err != nil {
|
||||
return out, nil
|
||||
}
|
||||
func scanSubscription(row scanner) (events.Subscription, error) {
|
||||
var out events.Subscription
|
||||
var raw []byte
|
||||
err := row.Scan(&out.ID, &out.QueueID, &out.CredentialID, &out.Name, &out.Revision, &out.Enabled, &raw, &out.CursorSequence, &out.PendingCount, &out.MaxPending, &out.MaxAttempts, &out.CreatedAt, &out.UpdatedAt)
|
||||
if err != nil {
|
||||
return out, err
|
||||
}
|
||||
return out, nil
|
||||
err = json.Unmarshal(raw, &out.Types)
|
||||
return out, err
|
||||
}
|
||||
|
||||
func (s *EventStore) CreateSubscription(ctx context.Context, t auth.Token, key string, in events.CreateInput) (events.SubscriptionResult, error) {
|
||||
|
||||
502
internal/storage/mysql/event_transport_integration_test.go
Normal file
502
internal/storage/mysql/event_transport_integration_test.go
Normal file
@@ -0,0 +1,502 @@
|
||||
package mysql
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"database/sql"
|
||||
"encoding/json"
|
||||
"io"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"net/url"
|
||||
"os"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.yuntuo.hk/k2safe/orbit/internal/jsonwire"
|
||||
"git.yuntuo.hk/k2safe/orbit/internal/testcontract"
|
||||
"git.yuntuo.hk/k2safe/orbit/internal/transport/httpserver"
|
||||
core "git.yuntuo.hk/k2safe/orbit/model"
|
||||
"git.yuntuo.hk/k2safe/orbit/platform/events"
|
||||
platform "git.yuntuo.hk/k2safe/orbit/platform/model"
|
||||
"git.yuntuo.hk/k2safe/orbit/pulse/service"
|
||||
"git.yuntuo.hk/k2safe/orbit/pulse/worker"
|
||||
"github.com/coder/websocket"
|
||||
)
|
||||
|
||||
func startEventServer(t *testing.T, ctx context.Context, db *sql.DB) (*httptest.Server, *httpserver.EventHub) {
|
||||
t.Helper()
|
||||
svc := events.New(NewEventStore(db))
|
||||
hub := httpserver.NewEventHub(ctx, svc)
|
||||
server := httpserver.New("", func(context.Context) error { return nil }, service.New(NewTaskStore(db)))
|
||||
httpserver.RegisterEventSubscriptions(server, svc)
|
||||
httpserver.RegisterEventConsumers(server, hub)
|
||||
tls := httptest.NewTLSServer(server.Handler)
|
||||
t.Cleanup(func() {
|
||||
bounded, stop := context.WithTimeout(context.Background(), 3*time.Second)
|
||||
defer stop()
|
||||
if err := hub.Shutdown(bounded); err != nil {
|
||||
t.Error(err)
|
||||
}
|
||||
tls.Close()
|
||||
})
|
||||
return tls, hub
|
||||
}
|
||||
func eventHTTP(t *testing.T, ctx context.Context, server *httptest.Server, method, path, token, key string, body any, status int) json.RawMessage {
|
||||
t.Helper()
|
||||
var raw []byte
|
||||
var err error
|
||||
switch b := body.(type) {
|
||||
case nil:
|
||||
case string:
|
||||
raw = []byte(b)
|
||||
default:
|
||||
raw, err = jsonwire.Marshal(body)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
req, err := http.NewRequestWithContext(ctx, method, server.URL+path, bytes.NewReader(raw))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if token != "" {
|
||||
req.Header.Set("Authorization", "Bearer "+token)
|
||||
}
|
||||
if key != "" {
|
||||
req.Header.Set("Idempotency-Key", key)
|
||||
}
|
||||
if body != nil {
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
}
|
||||
resp, err := server.Client().Do(req)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
data, err := io.ReadAll(io.LimitReader(resp.Body, 1<<20))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var envelope struct {
|
||||
RequestID core.ID `json:"request_id"`
|
||||
Data json.RawMessage `json:"data"`
|
||||
Error *struct {
|
||||
Code string `json:"code"`
|
||||
} `json:"error"`
|
||||
}
|
||||
if err = json.Unmarshal(data, &envelope); err != nil {
|
||||
t.Fatalf("invalid event HTTP response: status %d", resp.StatusCode)
|
||||
}
|
||||
if resp.StatusCode != status {
|
||||
code := ""
|
||||
if envelope.Error != nil {
|
||||
code = envelope.Error.Code
|
||||
}
|
||||
t.Fatalf("expected HTTP %d, got %d/%s", status, resp.StatusCode, code)
|
||||
}
|
||||
if envelope.RequestID.Validate() != nil || resp.Header.Get("Cache-Control") != "no-store" {
|
||||
t.Fatal("missing event API envelope metadata")
|
||||
}
|
||||
if status >= 400 {
|
||||
if envelope.Error == nil || len(envelope.Data) != 0 {
|
||||
t.Fatal("invalid event error response")
|
||||
}
|
||||
return data
|
||||
}
|
||||
if envelope.Error != nil {
|
||||
return nil
|
||||
}
|
||||
return envelope.Data
|
||||
}
|
||||
|
||||
type testEventSocket struct {
|
||||
t *testing.T
|
||||
ctx context.Context
|
||||
conn *websocket.Conn
|
||||
session core.ID
|
||||
pending []testWorkerFrame
|
||||
}
|
||||
|
||||
func connectTestEvents(t *testing.T, ctx context.Context, server *httptest.Server, f *eventFixture, sub core.ID, capacity uint32) *testEventSocket {
|
||||
t.Helper()
|
||||
headers := http.Header{}
|
||||
headers.Set("Authorization", "Bearer "+f.resources.Token)
|
||||
conn, resp, err := websocket.Dial(ctx, strings.Replace(server.URL, "https://", "wss://", 1)+"/v1/events/connect", &websocket.DialOptions{HTTPClient: server.Client(), HTTPHeader: headers, Subprotocols: []string{httpserver.EventSubprotocol}})
|
||||
if err != nil {
|
||||
status := 0
|
||||
if resp != nil {
|
||||
status = resp.StatusCode
|
||||
}
|
||||
t.Fatalf("event dial failed HTTP %d", status)
|
||||
}
|
||||
t.Cleanup(func() { conn.CloseNow() })
|
||||
conn.SetReadLimit(512 << 10)
|
||||
c := &testEventSocket{t: t, ctx: ctx, conn: conn}
|
||||
hello := c.send("events.hello", map[string]any{"subscription_id": sub, "capacity": capacity}, "")
|
||||
frame := c.read()
|
||||
if frame.Type != "events.welcome" || frame.ReplyTo != hello || frame.SessionID.Validate() != nil {
|
||||
t.Fatal("missing event welcome")
|
||||
}
|
||||
var welcome struct {
|
||||
SessionID core.ID `json:"session_id"`
|
||||
SubscriptionID core.ID `json:"subscription_id"`
|
||||
Capacity uint32 `json:"capacity"`
|
||||
LeaseDurationMS int `json:"lease_duration_ms"`
|
||||
ServerTime time.Time `json:"server_time"`
|
||||
}
|
||||
if json.Unmarshal(frame.Body, &welcome) != nil || welcome.SessionID != frame.SessionID || welcome.SubscriptionID != sub || welcome.Capacity != capacity || welcome.LeaseDurationMS != 30000 || welcome.ServerTime.IsZero() {
|
||||
t.Fatal("invalid welcome binding")
|
||||
}
|
||||
c.session = frame.SessionID
|
||||
return c
|
||||
}
|
||||
func (c *testEventSocket) send(kind string, body any, id core.ID) core.ID {
|
||||
c.t.Helper()
|
||||
if id == "" {
|
||||
id = executionID(c.t)
|
||||
}
|
||||
payload, err := jsonwire.Marshal(body)
|
||||
if err != nil {
|
||||
c.t.Fatal(err)
|
||||
}
|
||||
raw, err := jsonwire.Marshal(testWorkerFrame{Version: "1", Type: kind, MessageID: id, SessionID: c.session, Body: payload})
|
||||
if err != nil {
|
||||
c.t.Fatal(err)
|
||||
}
|
||||
testcontract.EventFrame(c.t, "client", raw)
|
||||
ctx, stop := context.WithTimeout(c.ctx, 5*time.Second)
|
||||
defer stop()
|
||||
if err = c.conn.Write(ctx, websocket.MessageText, raw); err != nil {
|
||||
c.t.Fatal(err)
|
||||
}
|
||||
return id
|
||||
}
|
||||
func (c *testEventSocket) read() testWorkerFrame {
|
||||
c.t.Helper()
|
||||
ctx, stop := context.WithTimeout(c.ctx, 5*time.Second)
|
||||
defer stop()
|
||||
kind, raw, err := c.conn.Read(ctx)
|
||||
if err != nil {
|
||||
c.t.Fatal(err)
|
||||
}
|
||||
testcontract.EventFrame(c.t, "server", raw)
|
||||
var frame testWorkerFrame
|
||||
if kind != websocket.MessageText || json.Unmarshal(raw, &frame) != nil {
|
||||
c.t.Fatal("invalid event frame")
|
||||
}
|
||||
if c.session != "" && frame.SessionID != c.session {
|
||||
c.t.Fatal("event response changed session")
|
||||
}
|
||||
return frame
|
||||
}
|
||||
func (c *testEventSocket) reply(id core.ID, code string) json.RawMessage {
|
||||
c.t.Helper()
|
||||
for {
|
||||
frame := c.read()
|
||||
if frame.Type == "event.deliver" {
|
||||
c.pending = append(c.pending, frame)
|
||||
continue
|
||||
}
|
||||
if frame.ReplyTo != id {
|
||||
c.t.Fatal("uncorrelated event reply")
|
||||
}
|
||||
if code != "" {
|
||||
if frame.Type != "error" || frame.Error == nil || frame.Error.Code != code {
|
||||
c.t.Fatal("unexpected event error")
|
||||
}
|
||||
} else if frame.Type != "ack" || frame.Error != nil {
|
||||
c.t.Fatal("expected event acknowledgement")
|
||||
}
|
||||
return frame.Body
|
||||
}
|
||||
}
|
||||
func (c *testEventSocket) grant() *events.Grant {
|
||||
c.t.Helper()
|
||||
var frame testWorkerFrame
|
||||
if len(c.pending) > 0 {
|
||||
frame = c.pending[0]
|
||||
c.pending = c.pending[1:]
|
||||
} else {
|
||||
frame = c.read()
|
||||
}
|
||||
if frame.Type != "event.deliver" {
|
||||
c.t.Fatalf("expected event delivery, got %s", frame.Type)
|
||||
}
|
||||
var g events.Grant
|
||||
if json.Unmarshal(frame.Body, &g) != nil || g.Delivery.EventID != g.Event.ID || g.Delivery.Sequence != g.Event.Sequence || g.Delivery.State != platform.DeliveryLeased {
|
||||
c.t.Fatal("invalid delivered event")
|
||||
}
|
||||
return &g
|
||||
}
|
||||
func eventProof(g *events.Grant) events.AckInput {
|
||||
return events.AckInput{DeliveryID: g.Delivery.ID, Epoch: g.Delivery.LeaseEpoch, LeaseSecret: g.LeaseSecret}
|
||||
}
|
||||
|
||||
func TestIntegrationEventHTTPAndWebSocket(t *testing.T) {
|
||||
dsn := os.Getenv("ORBIT_TEST_MYSQL_DSN")
|
||||
if dsn == "" {
|
||||
t.Skip("requires disposable MySQL")
|
||||
}
|
||||
ctx, stop := context.WithTimeout(context.Background(), 90*time.Second)
|
||||
defer stop()
|
||||
db, err := Open(ctx, dsn, 20, 5)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer db.Close()
|
||||
if err = Migrate(ctx, db); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
serverA, hubA := startEventServer(t, ctx, db)
|
||||
serverB, _ := startEventServer(t, ctx, db)
|
||||
t.Run("HTTP_management_required_fields_and_scoped_pagination", func(t *testing.T) {
|
||||
f := newEventFixture(t, ctx, db)
|
||||
other := newEventFixture(t, ctx, db)
|
||||
in := events.CreateInput{Name: "managed-results", QueueID: f.resources.QueueID, MaxAttempts: 1}
|
||||
raw := eventHTTP(t, ctx, serverA, "POST", "/v1/subscriptions", f.resources.Token, "create", in, 201)
|
||||
var created events.SubscriptionResult
|
||||
if json.Unmarshal(raw, &created) != nil {
|
||||
t.Fatal("invalid subscription receipt")
|
||||
}
|
||||
id := created.Subscription.ID
|
||||
raw = eventHTTP(t, ctx, serverA, "POST", "/v1/subscriptions", f.resources.Token, "create", in, 200)
|
||||
var duplicate events.SubscriptionResult
|
||||
if json.Unmarshal(raw, &duplicate) != nil || !duplicate.Deduplicated || duplicate.Subscription.ID != id {
|
||||
t.Fatal("idempotency changed through HTTP")
|
||||
}
|
||||
eventHTTP(t, ctx, serverA, "GET", "/v1/subscriptions/"+string(id), other.resources.Token, "", nil, 404)
|
||||
eventHTTP(t, ctx, serverA, "POST", "/v1/subscriptions/"+string(id)+":configure", f.resources.Token, "invalid-control", `{"expected_revision":"1","reason":"ambiguous"}`, 400)
|
||||
if !f.subscription(id).Enabled {
|
||||
t.Fatal("missing enabled paused subscription")
|
||||
}
|
||||
eventHTTP(t, ctx, serverA, "POST", "/v1/subscriptions/"+string(id)+":configure", f.resources.Token, "pause", events.ControlInput{ExpectedRevision: 1, Enabled: false, Reason: "maintenance"}, 200)
|
||||
eventHTTP(t, ctx, serverA, "POST", "/v1/subscriptions/"+string(id)+":configure", f.resources.Token, "resume", events.ControlInput{ExpectedRevision: 2, Enabled: true, Reason: "ready"}, 200)
|
||||
for _, bad := range []string{`"00"`, `"+1"`, `1`, `false`, `"18446744073709551616"`} {
|
||||
eventHTTP(t, ctx, serverA, "POST", "/v1/subscriptions", f.resources.Token, "invalid-sequence", `{"name":"invalid","queue_id":"`+string(f.resources.QueueID)+`","from_sequence":`+bad+`}`, 400)
|
||||
}
|
||||
for _, path := range []string{"/v1/subscriptions?queue_id=" + string(f.resources.QueueID) + "&limit=01", "/v1/subscriptions?queue_id=" + string(f.resources.QueueID) + "&limit=101", "/v1/subscriptions?queue_id=" + string(f.resources.QueueID) + "&cursor=a&cursor=b", "/v1/subscriptions/" + string(id) + "?unknown=1"} {
|
||||
eventHTTP(t, ctx, serverA, "GET", path, f.resources.Token, "", nil, 400)
|
||||
}
|
||||
f.create(events.CreateInput{})
|
||||
f.create(events.CreateInput{})
|
||||
path := "/v1/subscriptions?queue_id=" + string(f.resources.QueueID) + "&limit=1"
|
||||
var page events.SubscriptionPage
|
||||
raw = eventHTTP(t, ctx, serverA, "GET", path, f.resources.Token, "", nil, 200)
|
||||
if json.Unmarshal(raw, &page) != nil || len(page.Items) != 1 || page.NextCursor == "" {
|
||||
t.Fatal("missing subscription page")
|
||||
}
|
||||
first := page.Items[0].ID
|
||||
raw = eventHTTP(t, ctx, serverA, "GET", path+"&cursor="+url.QueryEscape(page.NextCursor), f.resources.Token, "", nil, 200)
|
||||
if json.Unmarshal(raw, &page) != nil || page.Items[0].ID == first {
|
||||
t.Fatal("subscription pagination repeated row")
|
||||
}
|
||||
eventHTTP(t, ctx, serverA, "GET", "/v1/subscriptions?queue_id="+string(other.resources.QueueID)+"&cursor="+url.QueryEscape(page.NextCursor), f.resources.Token, "", nil, 400)
|
||||
for range 3 {
|
||||
f.cancelResult()
|
||||
}
|
||||
firstGrant := f.grant(id)
|
||||
f.ack(id, firstGrant)
|
||||
second := f.grant(id)
|
||||
f.expireDelivery(second)
|
||||
third := f.grant(id)
|
||||
path = "/v1/subscriptions/" + string(id) + "/deliveries"
|
||||
var deliveries events.DeliveryPage
|
||||
raw = eventHTTP(t, ctx, serverA, "GET", path+"?limit=1", f.resources.Token, "", nil, 200)
|
||||
if json.Unmarshal(raw, &deliveries) != nil || len(deliveries.Items) != 1 || deliveries.NextCursor == "" {
|
||||
t.Fatal("missing delivery page")
|
||||
}
|
||||
eventHTTP(t, ctx, serverA, "GET", path+"?state=EXPIRED&cursor="+url.QueryEscape(deliveries.NextCursor), f.resources.Token, "", nil, 400)
|
||||
nextRaw := eventHTTP(t, ctx, serverA, "GET", path+"?limit=1&cursor="+url.QueryEscape(deliveries.NextCursor), f.resources.Token, "", nil, 200)
|
||||
var nextPage events.DeliveryPage
|
||||
if json.Unmarshal(nextRaw, &nextPage) != nil || len(nextPage.Items) != 1 || nextPage.Items[0].ID == deliveries.Items[0].ID {
|
||||
t.Fatal("delivery page repeated a row")
|
||||
}
|
||||
eventHTTP(t, ctx, serverA, "GET", path+"/"+string(second.Delivery.ID), other.resources.Token, "", nil, 404)
|
||||
wrongSub := f.create(events.CreateInput{})
|
||||
eventHTTP(t, ctx, serverA, "GET", "/v1/subscriptions/"+string(wrongSub.ID)+"/deliveries/"+string(second.Delivery.ID), f.resources.Token, "", nil, 404)
|
||||
raw = eventHTTP(t, ctx, serverA, "GET", path+"?state=EXPIRED", f.resources.Token, "", nil, 200)
|
||||
if json.Unmarshal(raw, &deliveries) != nil || len(deliveries.Items) != 1 || deliveries.Items[0].ID != second.Delivery.ID {
|
||||
t.Fatal("expired delivery not queryable")
|
||||
}
|
||||
expired := deliveries.Items[0]
|
||||
raw = eventHTTP(t, ctx, serverA, "GET", path+"/"+string(expired.ID), f.resources.Token, "", nil, 200)
|
||||
var detail events.DeliveryDetail
|
||||
if json.Unmarshal(raw, &detail) != nil || detail.Event.ID != second.Event.ID || strings.Contains(string(raw), second.LeaseSecret) || strings.Contains(string(raw), "token_digest") {
|
||||
t.Fatal("unsafe delivery detail")
|
||||
}
|
||||
retry := events.RetryInput{ExpectedRevision: expired.Revision, Reason: "consumer fixed", AdditionalAttempts: 1}
|
||||
eventHTTP(t, ctx, serverA, "POST", path+"/"+string(expired.ID)+":retry", f.resources.Token, "retry", retry, 200)
|
||||
eventHTTP(t, ctx, serverA, "POST", path+"/"+string(expired.ID)+":retry", f.resources.Token, "retry", retry, 200)
|
||||
f.ack(id, third)
|
||||
})
|
||||
t.Run("handshake_authority_protocol_origin_and_URL", func(t *testing.T) {
|
||||
f := newEventFixture(t, ctx, db)
|
||||
plain := newExecutionFixture(t, ctx, db)
|
||||
for _, c := range []struct {
|
||||
token, protocol, path, origin string
|
||||
status int
|
||||
}{
|
||||
{"", httpserver.EventSubprotocol, "/v1/events/connect", "", 401},
|
||||
{plain.resources.Token, httpserver.EventSubprotocol, "/v1/events/connect", "", 403},
|
||||
{f.resources.Token, "pulse.worker.v1", "/v1/events/connect", "", 426},
|
||||
{f.resources.Token, httpserver.EventSubprotocol, "/v1/events/connect?token=forbidden", "", 400},
|
||||
{f.resources.Token, httpserver.EventSubprotocol, "/v1/events/connect", "https://foreign.example", 403},
|
||||
} {
|
||||
headers := http.Header{}
|
||||
if c.token != "" {
|
||||
headers.Set("Authorization", "Bearer "+c.token)
|
||||
}
|
||||
if c.origin != "" {
|
||||
headers.Set("Origin", c.origin)
|
||||
}
|
||||
conn, resp, err := websocket.Dial(ctx, strings.Replace(serverA.URL, "https://", "wss://", 1)+c.path, &websocket.DialOptions{HTTPClient: serverA.Client(), HTTPHeader: headers, Subprotocols: []string{c.protocol}})
|
||||
if conn != nil {
|
||||
conn.CloseNow()
|
||||
}
|
||||
if err == nil || resp == nil || resp.StatusCode != c.status {
|
||||
t.Fatalf("event handshake expected HTTP %d", c.status)
|
||||
}
|
||||
}
|
||||
})
|
||||
t.Run("WSS_capacity_ack_receipt_and_result_snapshot", func(t *testing.T) {
|
||||
f := newEventFixture(t, ctx, db)
|
||||
sub := f.create(events.CreateInput{})
|
||||
task := f.submit(1)
|
||||
ref := retryStart(t, f.executionFixture)
|
||||
result := json.RawMessage(`{"n":9007199254740993,"text":"` + strings.Repeat("<&>", 80000) + `"}`)
|
||||
if _, err := f.a.Finish(ctx, f.token, f.binding, executionID(t), worker.FinishInput{Ref: ref, Outcome: "SUCCEEDED", Result: result}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
f.cancelResult()
|
||||
c := connectTestEvents(t, ctx, serverA, f, sub.ID, 1)
|
||||
first := c.grant()
|
||||
if first.Event.Snapshot.TaskID != task.ID || string(first.Event.Snapshot.Result) != string(result) {
|
||||
t.Fatal("WSS snapshot changed large result")
|
||||
}
|
||||
c.reply(c.send("events.ping", struct{}{}, ""), "")
|
||||
var leased int
|
||||
if err := db.QueryRowContext(ctx, `SELECT COUNT(*) FROM orbit_deliveries WHERE namespace_id=? AND subscription_id=? AND state='LEASED'`, f.token.NamespaceID, sub.ID).Scan(&leased); err != nil || leased != 1 {
|
||||
t.Fatal("WSS capacity exceeded", err)
|
||||
}
|
||||
var ack events.AckResult
|
||||
message := c.send("event.ack", eventProof(first), "")
|
||||
if json.Unmarshal(c.reply(message, ""), &ack) != nil || ack.Deduplicated {
|
||||
t.Fatal("first ack not applied")
|
||||
}
|
||||
second := c.grant()
|
||||
if second.Event.ID == first.Event.ID {
|
||||
t.Fatal("confirmed result redelivered")
|
||||
}
|
||||
var repeated events.AckResult
|
||||
if json.Unmarshal(c.reply(c.send("event.ack", eventProof(first), message), ""), &repeated) != nil || !repeated.Deduplicated || !repeated.AcknowledgedAt.Equal(ack.AcknowledgedAt) {
|
||||
t.Fatal("duplicate ACK changed receipt")
|
||||
}
|
||||
c.reply(c.send("event.ack", eventProof(second), ""), "")
|
||||
})
|
||||
t.Run("cross_node_redelivery_fences_old_proof", func(t *testing.T) {
|
||||
f := newEventFixture(t, ctx, db)
|
||||
sub := f.create(events.CreateInput{MaxAttempts: 2})
|
||||
f.cancelResult()
|
||||
a := connectTestEvents(t, ctx, serverA, f, sub.ID, 1)
|
||||
first := a.grant()
|
||||
a.conn.CloseNow()
|
||||
f.expireDelivery(first)
|
||||
b := connectTestEvents(t, ctx, serverB, f, sub.ID, 1)
|
||||
second := b.grant()
|
||||
if second.Event.ID != first.Event.ID || second.Delivery.ID != first.Delivery.ID || second.Delivery.LeaseEpoch != first.Delivery.LeaseEpoch+1 {
|
||||
t.Fatal("cross-node redelivery lost durable identity")
|
||||
}
|
||||
b.reply(b.send("event.ack", eventProof(first), ""), "DELIVERY_LEASE_LOST")
|
||||
b.reply(b.send("event.ack", eventProof(second), ""), "")
|
||||
if f.subscription(sub.ID).PendingCount != 0 {
|
||||
t.Fatal("cross-node ack did not release window")
|
||||
}
|
||||
})
|
||||
t.Run("discarded_ack_response_reconciles_on_new_connection", func(t *testing.T) {
|
||||
f := newEventFixture(t, ctx, db)
|
||||
sub := f.create(events.CreateInput{})
|
||||
f.cancelResult()
|
||||
a := connectTestEvents(t, ctx, serverA, f, sub.ID, 1)
|
||||
g := a.grant()
|
||||
message := a.send("event.ack", eventProof(g), "")
|
||||
deadline := time.Now().Add(3 * time.Second)
|
||||
var original events.Delivery
|
||||
for time.Now().Before(deadline) {
|
||||
original, err = f.events.Delivery(ctx, f.token, sub.ID, g.Delivery.ID)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if original.State == platform.DeliveryAcked {
|
||||
break
|
||||
}
|
||||
time.Sleep(5 * time.Millisecond)
|
||||
}
|
||||
if original.State != platform.DeliveryAcked {
|
||||
t.Fatal("ACK did not commit")
|
||||
}
|
||||
a.conn.CloseNow()
|
||||
b := connectTestEvents(t, ctx, serverB, f, sub.ID, 1)
|
||||
var receipt events.AckResult
|
||||
if json.Unmarshal(b.reply(b.send("event.ack", eventProof(g), message), ""), &receipt) != nil || !receipt.Deduplicated || original.AcknowledgedAt == nil || !receipt.AcknowledgedAt.Equal(*original.AcknowledgedAt) {
|
||||
t.Fatal("lost ACK response could not be reconciled")
|
||||
}
|
||||
if f.count("orbit_audit_events", "resource_id=? AND action='delivery.ack'", g.Delivery.ID) != 1 {
|
||||
t.Fatal("replayed ACK duplicated audit")
|
||||
}
|
||||
})
|
||||
t.Run("revocation_and_session_spoofing_close_live_connection", func(t *testing.T) {
|
||||
f := newEventFixture(t, ctx, db)
|
||||
sub := f.create(events.CreateInput{})
|
||||
c := connectTestEvents(t, ctx, serverA, f, sub.ID, 1)
|
||||
f.exec(`UPDATE orbit_credentials SET revoked_at=UTC_TIMESTAMP(6) WHERE namespace_id=? AND id=?`, f.token.NamespaceID, f.token.CredentialID)
|
||||
// 主动心跳或下一次领取均重验权限；这里主动心跳固定相关响应。
|
||||
msg := c.send("events.ping", struct{}{}, "")
|
||||
frame := c.read()
|
||||
if frame.Type != "error" || frame.Error == nil || frame.Error.Code != "UNAUTHENTICATED" || (frame.ReplyTo != "" && frame.ReplyTo != msg) {
|
||||
t.Fatal("revoked connection retained authority")
|
||||
}
|
||||
g := newEventFixture(t, ctx, db)
|
||||
owned := g.create(events.CreateInput{})
|
||||
d := connectTestEvents(t, ctx, serverA, g, owned.ID, 1)
|
||||
raw := `{"protocol_version":"1","type":"events.ping","message_id":"` + string(executionID(t)) + `","session_id":"` + string(executionID(t)) + `","body":{}}`
|
||||
if err := d.conn.Write(ctx, websocket.MessageText, []byte(raw)); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
readCtx, cancel := context.WithTimeout(ctx, 2*time.Second)
|
||||
defer cancel()
|
||||
_, _, err := d.conn.Read(readCtx)
|
||||
if websocket.CloseStatus(err) != websocket.StatusPolicyViolation {
|
||||
t.Fatal("foreign connection session accepted")
|
||||
}
|
||||
})
|
||||
t.Run("shutdown_closes_socket_preserves_delivery_and_rejects_new_connection", func(t *testing.T) {
|
||||
f := newEventFixture(t, ctx, db)
|
||||
sub := f.create(events.CreateInput{})
|
||||
f.cancelResult()
|
||||
a := connectTestEvents(t, ctx, serverA, f, sub.ID, 1)
|
||||
g := a.grant()
|
||||
bounded, cancel := context.WithTimeout(ctx, 3*time.Second)
|
||||
defer cancel()
|
||||
if err := hubA.Shutdown(bounded); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
v, err := f.events.Delivery(ctx, f.token, sub.ID, g.Delivery.ID)
|
||||
if err != nil || v.State != platform.DeliveryLeased || v.LeaseEpoch != g.Delivery.LeaseEpoch {
|
||||
t.Fatal("shutdown erased authoritative lease", err)
|
||||
}
|
||||
eventHTTP(t, ctx, serverA, "GET", "/v1/events/connect", f.resources.Token, "", nil, 426)
|
||||
headers := http.Header{}
|
||||
headers.Set("Authorization", "Bearer "+f.resources.Token)
|
||||
conn, response, err := websocket.Dial(ctx, strings.Replace(serverA.URL, "https://", "wss://", 1)+"/v1/events/connect", &websocket.DialOptions{HTTPClient: serverA.Client(), HTTPHeader: headers, Subprotocols: []string{httpserver.EventSubprotocol}})
|
||||
if conn != nil {
|
||||
conn.CloseNow()
|
||||
}
|
||||
if err == nil || response == nil || response.StatusCode != 503 {
|
||||
t.Fatal("closing hub accepted new connection")
|
||||
}
|
||||
b := connectTestEvents(t, ctx, serverB, f, sub.ID, 1)
|
||||
b.reply(b.send("event.ack", eventProof(g), ""), "")
|
||||
})
|
||||
}
|
||||
@@ -0,0 +1,5 @@
|
||||
-- 面向已物化交付历史的稳定创建时间分页；确认和重新发送不会改变排序键。
|
||||
-- 不使用外键或级联删除；序号和事件归属继续由服务端按命名空间校验。
|
||||
ALTER TABLE orbit_deliveries
|
||||
ADD KEY ix_delivery_history(namespace_id,subscription_id,created_at,id),
|
||||
ADD KEY ix_delivery_state_history(namespace_id,subscription_id,state,created_at,id);
|
||||
@@ -88,7 +88,7 @@ func TestIntegrationNoForeignKeysUpgrade(t *testing.T) {
|
||||
}
|
||||
indexes := func() []string {
|
||||
t.Helper()
|
||||
rows, err := db.QueryContext(ctx, `SELECT CONCAT_WS(':',TABLE_NAME,INDEX_NAME,NON_UNIQUE,SEQ_IN_INDEX,COLUMN_NAME,COALESCE(SUB_PART,0),INDEX_TYPE,IS_VISIBLE) FROM information_schema.STATISTICS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME NOT IN ('orbit_application_resources','orbit_event_streams','orbit_result_events') AND INDEX_NAME NOT IN ('ix_task_priority_dispatch','ix_delivery_subscription_due','ix_delivery_subscription_lease') ORDER BY TABLE_NAME,INDEX_NAME,SEQ_IN_INDEX`)
|
||||
rows, err := db.QueryContext(ctx, `SELECT CONCAT_WS(':',TABLE_NAME,INDEX_NAME,NON_UNIQUE,SEQ_IN_INDEX,COLUMN_NAME,COALESCE(SUB_PART,0),INDEX_TYPE,IS_VISIBLE) FROM information_schema.STATISTICS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME NOT IN ('orbit_application_resources','orbit_event_streams','orbit_result_events') AND INDEX_NAME NOT IN ('ix_task_priority_dispatch','ix_delivery_subscription_due','ix_delivery_subscription_lease','ix_delivery_history','ix_delivery_state_history') ORDER BY TABLE_NAME,INDEX_NAME,SEQ_IN_INDEX`)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user