package orbit import ( "context" "sync/atomic" "testing" "time" "github.com/coder/websocket" ) // TestEventConsumerReconnectStormUsesBoundedBackoff verifies that repeated transport // loss only reconnects with bounded jitter and that a later delivery is handled once. func TestEventConsumerReconnectStormUsesBoundedBackoff(t *testing.T) { grant := eventTestGrant() var calls atomic.Int64 var connected atomic.Int64 client := eventMock(t, func(ctx context.Context, c *websocket.Conn, _ EventRegistration, n int) { connected.Store(int64(n)) if n <= 4 { _ = c.Close(websocket.StatusInternalError, "controlled reconnect storm") return } if n != 5 { return } if err := eventTestWrite(ctx, c, eventFrame{Type: "event.deliver", Body: eventBody(grant)}); err != nil { return } frame, err := eventTestRead(ctx, c) if err != nil || frame.Type != "event.ack" { return } _ = eventTestWrite(ctx, c, eventFrame{Type: "ack", ReplyTo: frame.MessageID, Body: eventBody(eventAckFor(grant))}) <-ctx.Done() }) consumer, err := NewEventConsumer(client, EventConsumerOptions{ SubscriptionID: eventTestSub, Capacity: 1, Handle: func(context.Context, ResultEvent) error { calls.Add(1); return nil }, ReconnectMinDelay: 10 * time.Millisecond, ReconnectMaxDelay: 40 * time.Millisecond, ShutdownTimeout: time.Second, }) if err != nil { t.Fatal(err) } ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) defer cancel() done := make(chan error, 1) started := time.Now() go func() { done <- consumer.Run(ctx) }() awaitConsumerEvent(t, consumer, "acknowledged") cancel() if err := awaitConsumerStop(t, done); err != nil { t.Fatal(err) } stats := consumer.Stats() if connected.Load() != 5 || stats.Reconnects != 4 || stats.Acknowledged != 1 || calls.Load() != 1 { t.Fatalf("reconnect storm was not bounded or redelivered incorrectly: connections=%d stats=%+v calls=%d", connected.Load(), stats, calls.Load()) } if elapsed := time.Since(started); elapsed > 2*time.Second { t.Fatalf("bounded reconnect storm took too long: %s", elapsed) } }