284 lines
11 KiB
Go
284 lines
11 KiB
Go
package orbit
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"net/http"
|
|
"net/http/httptest"
|
|
"strings"
|
|
"sync/atomic"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/coder/websocket"
|
|
)
|
|
|
|
const eventTestSub = "55555555555555555555555555555555"
|
|
const eventTestID = "66666666666666666666666666666666"
|
|
const eventTestDelivery = "77777777777777777777777777777777"
|
|
|
|
func eventTestGrant() EventGrant {
|
|
now := time.Now().UTC()
|
|
until := now.Add(30 * time.Second)
|
|
return EventGrant{Delivery: EventDelivery{ID: eventTestDelivery, SubscriptionID: eventTestSub, EventID: eventTestID, Sequence: 9007199254740993, Revision: 2, State: DeliveryLeased, CreatedAt: now, AvailableAt: now, LeaseEpoch: 1, LeaseUntil: &until, Attempts: 1, AttemptBudget: 20}, Event: ResultEvent{ID: eventTestID, Sequence: 9007199254740993, Type: "task.succeeded", AggregateKind: "task", AggregateID: workerTestTask, AggregateRevision: 9007199254740993, OccurredAt: now, Details: json.RawMessage(`{}`), Snapshot: TaskResultSnapshot{SchemaVersion: 1, TaskID: workerTestTask, QueueID: workerTestQueue, State: TaskSucceeded, Revision: 9007199254740993, Result: json.RawMessage(`{"n":9007199254740993,"text":"<&>"}`)}}, LeaseSecret: strings.Repeat("A", 43)}
|
|
}
|
|
func eventTestRead(ctx context.Context, c *websocket.Conn) (eventFrame, error) {
|
|
kind, raw, err := c.Read(ctx)
|
|
var f eventFrame
|
|
if err != nil {
|
|
return f, err
|
|
}
|
|
if kind != websocket.MessageText || json.Unmarshal(raw, &f) != nil {
|
|
return f, ErrEventProtocol
|
|
}
|
|
return f, nil
|
|
}
|
|
func eventTestWrite(ctx context.Context, c *websocket.Conn, f eventFrame) error {
|
|
f.Version = "1"
|
|
f.SessionID = workerTestSession
|
|
f.MessageID, _ = NewEventMessageID()
|
|
raw, err := marshalWire(f)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
return c.Write(ctx, websocket.MessageText, raw)
|
|
}
|
|
func eventBody(in any) json.RawMessage { b, _ := marshalWire(in); return b }
|
|
func eventMock(t *testing.T, serve func(context.Context, *websocket.Conn, EventRegistration, int)) *Client {
|
|
t.Helper()
|
|
ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
|
|
var count atomic.Int64
|
|
server := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
|
if r.URL.Path != "/v1/events/connect" || r.URL.RawQuery != "" || r.Header.Get("Authorization") != "Bearer event-test-token" {
|
|
t.Error("unexpected event handshake")
|
|
}
|
|
c, err := websocket.Accept(w, r, &websocket.AcceptOptions{Subprotocols: []string{eventSubprotocol}})
|
|
if err != nil {
|
|
return
|
|
}
|
|
defer c.CloseNow()
|
|
f, err := eventTestRead(ctx, c)
|
|
var in EventRegistration
|
|
if err != nil || f.Type != "events.hello" || f.SessionID != "" || !validID(f.MessageID) || json.Unmarshal(f.Body, &in) != nil {
|
|
t.Error("invalid event hello")
|
|
return
|
|
}
|
|
welcome := EventWelcome{SessionID: workerTestSession, SubscriptionID: in.SubscriptionID, Capacity: in.Capacity, LeaseDurationMS: 30000, HeartbeatIntervalMS: 15000, ServerTime: time.Now().UTC(), Enabled: true}
|
|
if eventTestWrite(ctx, c, eventFrame{Type: "events.welcome", ReplyTo: f.MessageID, Body: eventBody(welcome)}) != nil {
|
|
return
|
|
}
|
|
serve(ctx, c, in, int(count.Add(1)))
|
|
}))
|
|
t.Cleanup(func() { cancel(); server.Close() })
|
|
client, err := NewClient(server.URL, Options{Token: "event-test-token", HTTPClient: server.Client()})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
return client
|
|
}
|
|
func eventTestDial(t *testing.T, client *Client, capacity uint32) *EventConnection {
|
|
t.Helper()
|
|
ctx, stop := context.WithTimeout(context.Background(), 3*time.Second)
|
|
defer stop()
|
|
c, err := client.DialEvents(ctx, EventRegistration{SubscriptionID: eventTestSub, Capacity: capacity})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
t.Cleanup(func() { c.Close() })
|
|
return c
|
|
}
|
|
func eventAckFor(g EventGrant) EventAckResult {
|
|
return EventAckResult{DeliveryID: g.Delivery.ID, EventID: g.Event.ID, Epoch: g.Delivery.LeaseEpoch, AcknowledgedAt: time.Now().UTC()}
|
|
}
|
|
|
|
func TestEventConnectionMultiplexAndPrecision(t *testing.T) {
|
|
g := eventTestGrant()
|
|
client := eventMock(t, func(ctx context.Context, c *websocket.Conn, _ EventRegistration, _ int) {
|
|
if eventTestWrite(ctx, c, eventFrame{Type: "event.deliver", Body: eventBody(g)}) != nil {
|
|
return
|
|
}
|
|
first, err := eventTestRead(ctx, c)
|
|
if err != nil {
|
|
return
|
|
}
|
|
second, err := eventTestRead(ctx, c)
|
|
if err != nil {
|
|
return
|
|
}
|
|
for _, f := range []eventFrame{second, first} {
|
|
var body any = EventPong{ServerTime: time.Now().UTC(), Enabled: true}
|
|
if f.Type == "event.ack" {
|
|
body = eventAckFor(g)
|
|
}
|
|
if eventTestWrite(ctx, c, eventFrame{Type: "ack", ReplyTo: f.MessageID, Body: eventBody(body)}) != nil {
|
|
return
|
|
}
|
|
}
|
|
_, _ = eventTestRead(ctx, c)
|
|
})
|
|
c := eventTestDial(t, client, 1)
|
|
ctx, stop := context.WithTimeout(context.Background(), 3*time.Second)
|
|
defer stop()
|
|
got, err := c.Next(ctx)
|
|
if err != nil || got.Event.Sequence != 9007199254740993 || string(got.Event.Snapshot.Result) != string(g.Event.Snapshot.Result) {
|
|
t.Fatal("event snapshot changed", err)
|
|
}
|
|
errs := make(chan error, 2)
|
|
go func() { _, err := c.Ping(ctx, workerTestID(t)); errs <- err }()
|
|
go func() { _, err := c.Ack(ctx, workerTestID(t), got.Proof()); errs <- err }()
|
|
for range 2 {
|
|
if err := <-errs; err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
if c.Err() != nil {
|
|
t.Fatal(c.Err())
|
|
}
|
|
for _, value := range []any{g, g.Proof(), c, &EventError{Code: "UNAVAILABLE", Message: "secret-text"}} {
|
|
for _, format := range []string{"%v", "%+v", "%#v"} {
|
|
printed := fmt.Sprintf(format, value)
|
|
if strings.Contains(printed, g.LeaseSecret) || strings.Contains(printed, "secret-text") || strings.Contains(printed, "9007199254740993") {
|
|
t.Fatal("event formatter exposed secrets or business data")
|
|
}
|
|
}
|
|
}
|
|
}
|
|
func TestEventConnectionWaitCancellationAndAckUncertainty(t *testing.T) {
|
|
received := make(chan struct{}, 1)
|
|
client := eventMock(t, func(ctx context.Context, c *websocket.Conn, _ EventRegistration, _ int) {
|
|
f, err := eventTestRead(ctx, c)
|
|
if err != nil {
|
|
return
|
|
}
|
|
if f.Type != "event.ack" {
|
|
t.Error("unexpected control")
|
|
}
|
|
received <- struct{}{}
|
|
_, _ = eventTestRead(ctx, c)
|
|
})
|
|
c := eventTestDial(t, client, 1)
|
|
short, stop := context.WithTimeout(context.Background(), 20*time.Millisecond)
|
|
defer stop()
|
|
if _, err := c.Next(short); !errors.Is(err, context.DeadlineExceeded) || c.Err() != nil {
|
|
t.Fatal("Next timeout closed connection", err)
|
|
}
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
done := make(chan error, 1)
|
|
go func() { _, err := c.Ack(ctx, workerTestID(t), eventTestGrant().Proof()); done <- err }()
|
|
<-received
|
|
cancel()
|
|
if err := <-done; !errors.Is(err, context.Canceled) {
|
|
t.Fatal(err)
|
|
}
|
|
select {
|
|
case <-c.Done():
|
|
case <-time.After(time.Second):
|
|
t.Fatal("uncertain ACK did not close connection")
|
|
}
|
|
}
|
|
func TestEventConnectionBoundedControlsAndDeliveries(t *testing.T) {
|
|
t.Run("controls", func(t *testing.T) {
|
|
received := make(chan struct{}, 40)
|
|
client := eventMock(t, func(ctx context.Context, c *websocket.Conn, _ EventRegistration, _ int) {
|
|
for {
|
|
if _, err := eventTestRead(ctx, c); err != nil {
|
|
return
|
|
}
|
|
received <- struct{}{}
|
|
}
|
|
})
|
|
c := eventTestDial(t, client, 1)
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
defer cancel()
|
|
done := make(chan error, 32)
|
|
for range 32 {
|
|
go func() { _, err := c.Ping(ctx, workerTestID(t)); done <- err }()
|
|
<-received
|
|
}
|
|
if _, err := c.Ping(ctx, workerTestID(t)); !errors.Is(err, ErrEventBusy) {
|
|
t.Fatal("unbounded pending controls", err)
|
|
}
|
|
cancel()
|
|
for range 32 {
|
|
<-done
|
|
}
|
|
})
|
|
t.Run("deliveries", func(t *testing.T) {
|
|
client := eventMock(t, func(ctx context.Context, c *websocket.Conn, _ EventRegistration, _ int) {
|
|
for range 2 {
|
|
_ = eventTestWrite(ctx, c, eventFrame{Type: "event.deliver", Body: eventBody(eventTestGrant())})
|
|
}
|
|
_, _ = eventTestRead(ctx, c)
|
|
})
|
|
c := eventTestDial(t, client, 1)
|
|
select {
|
|
case <-c.Done():
|
|
case <-time.After(time.Second):
|
|
t.Fatal("unbounded deliveries")
|
|
}
|
|
if !errors.Is(c.Err(), ErrEventBackpressure) {
|
|
t.Fatal(c.Err())
|
|
}
|
|
})
|
|
}
|
|
func TestEventConnectionRejectsUntrustedProtocol(t *testing.T) {
|
|
g := eventTestGrant()
|
|
valid := eventFrame{Version: "1", Type: "event.deliver", MessageID: workerTestQueue, SessionID: workerTestSession, Body: eventBody(g)}
|
|
raw, _ := marshalWire(valid)
|
|
cases := map[string][]byte{
|
|
"duplicate": []byte(strings.Replace(string(raw), `"protocol_version":"1"`, `"protocol_version":"1","protocol_version":"1"`, 1)),
|
|
"alias": []byte(strings.Replace(string(raw), `"lease_secret"`, `"Lease_Secret"`, 1)),
|
|
"noncanonical": []byte(strings.Replace(string(raw), `"sequence":"9007199254740993"`, `"sequence":"09007199254740993"`, 1)),
|
|
"null_bool": []byte(strings.Replace(string(raw), `"effect_uncertain":false`, `"effect_uncertain":null`, 1)),
|
|
"null_error": []byte(strings.Replace(string(raw), `"body":`, `"error":null,"body":`, 1)),
|
|
"mismatched_event": []byte(strings.Replace(string(raw), `"event_id":"`+eventTestID+`"`, `"event_id":"`+workerTestTask+`"`, 1)),
|
|
"missing_created": []byte(strings.Replace(string(raw), `"created_at":`+string(eventBody(g.Delivery.CreatedAt))+`,`, "", 1)),
|
|
"wrong_state": []byte(strings.Replace(string(raw), `"state":"SUCCEEDED"`, `"state":"FAILED"`, 1)),
|
|
"unknown_body_field": []byte(strings.Replace(string(raw), `"schema_version":1`, `"extra":true,"schema_version":1`, 1)),
|
|
}
|
|
for name, bad := range cases {
|
|
t.Run(name, func(t *testing.T) {
|
|
client := eventMock(t, func(ctx context.Context, c *websocket.Conn, _ EventRegistration, _ int) {
|
|
_ = c.Write(ctx, websocket.MessageText, bad)
|
|
_, _ = eventTestRead(ctx, c)
|
|
})
|
|
c := eventTestDial(t, client, 1)
|
|
ctx, stop := context.WithTimeout(context.Background(), time.Second)
|
|
defer stop()
|
|
if _, err := c.Next(ctx); !errors.Is(err, ErrEventProtocol) {
|
|
t.Fatal("invalid response accepted", err)
|
|
}
|
|
})
|
|
}
|
|
}
|
|
func TestEventConnectionRejectsMismatchedAck(t *testing.T) {
|
|
client := eventMock(t, func(ctx context.Context, c *websocket.Conn, _ EventRegistration, _ int) {
|
|
f, err := eventTestRead(ctx, c)
|
|
if err != nil {
|
|
return
|
|
}
|
|
out := eventAckFor(eventTestGrant())
|
|
out.Epoch++
|
|
_ = eventTestWrite(ctx, c, eventFrame{Type: "ack", ReplyTo: f.MessageID, Body: eventBody(out)})
|
|
_, _ = eventTestRead(ctx, c)
|
|
})
|
|
c := eventTestDial(t, client, 1)
|
|
if _, err := c.Ack(context.Background(), workerTestID(t), eventTestGrant().Proof()); !errors.Is(err, ErrEventProtocol) {
|
|
t.Fatal("wrong receipt accepted", err)
|
|
}
|
|
}
|
|
|
|
func TestEventConnectionPolicyClosureIsPermanent(t *testing.T) {
|
|
client := eventMock(t, func(ctx context.Context, c *websocket.Conn, _ EventRegistration, _ int) {
|
|
_ = c.Close(websocket.StatusPolicyViolation, "private peer reason")
|
|
})
|
|
c := eventTestDial(t, client, 1)
|
|
ctx, stop := context.WithTimeout(context.Background(), time.Second)
|
|
defer stop()
|
|
if _, err := c.Next(ctx); !errors.Is(err, ErrEventProtocol) || eventRetryable(err) {
|
|
t.Fatal("policy violation classified as retryable", err)
|
|
}
|
|
}
|