280 lines
10 KiB
Go
280 lines
10 KiB
Go
package orbit
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"sync/atomic"
|
|
"testing"
|
|
"time"
|
|
)
|
|
|
|
func TestIntegrationEventProcessLeaseRecovery(t *testing.T) {
|
|
a := startOrbitProcess(t)
|
|
_, _, token := integrationWorkerClient(t)
|
|
api := processClient(t, a, token)
|
|
ctx, stop := context.WithTimeout(context.Background(), 45*time.Second)
|
|
defer stop()
|
|
sub := integrationEventSubscription(t, ctx, api, 3)
|
|
_ = integrationCancelledEvent(t, ctx, api, sub)
|
|
firstConn := eventDialIntegration(t, ctx, api, sub, 1)
|
|
first, err := firstConn.Next(ctx)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
a.kill(t)
|
|
select {
|
|
case <-firstConn.Done():
|
|
case <-time.After(3 * time.Second):
|
|
t.Fatal("dead server connection remained live")
|
|
}
|
|
b := startOrbitProcess(t)
|
|
other := processClient(t, b, token)
|
|
secondConn := eventDialIntegration(t, ctx, other, sub, 1)
|
|
second, err := secondConn.Next(ctx)
|
|
if err != nil || first.Event.ID != second.Event.ID || first.Delivery.ID != second.Delivery.ID || second.Delivery.LeaseEpoch != 2 || second.Delivery.Attempts != 2 {
|
|
t.Fatal("node restart did not recover original delivery", err)
|
|
}
|
|
_, err = secondConn.Ack(ctx, workerTestID(t), first.Proof())
|
|
var remote *EventError
|
|
if !errors.As(err, &remote) || remote.Code != "DELIVERY_LEASE_LOST" {
|
|
t.Fatal("dead node proof acknowledged newer lease", err)
|
|
}
|
|
if _, err = secondConn.Ack(ctx, workerTestID(t), second.Proof()); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
current, err := other.GetSubscription(ctx, sub.ID)
|
|
if err != nil || current.PendingCount != 0 {
|
|
t.Fatal("recovered acknowledgement not persisted", err)
|
|
}
|
|
t.Log("SIGKILL confirmed by Wait; new process recovered original delivery after natural lease expiry, rejected old proof and committed new ACK")
|
|
}
|
|
func TestIntegrationEventProcessAckCommitResponseLoss(t *testing.T) {
|
|
a := startOrbitProcess(t)
|
|
_, _, token := integrationWorkerClient(t)
|
|
api := processClient(t, a, token)
|
|
ctx, stop := context.WithTimeout(context.Background(), 20*time.Second)
|
|
defer stop()
|
|
sub := integrationEventSubscription(t, ctx, api, 3)
|
|
_ = integrationCancelledEvent(t, ctx, api, sub)
|
|
gateway := newProtocolFaultGateway(t, a.url, "/v1/events/connect", eventSubprotocol)
|
|
gateway.drop.Store("event.ack")
|
|
gateway.dropOffline.Store(true)
|
|
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(context.Context, ResultEvent) error { calls.Add(1); 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) }()
|
|
awaitProcessDrop(t, ctx, gateway, "event.ack")
|
|
before, err := api.ListEventDeliveries(ctx, sub.ID, DeliveryPageRequest{State: DeliveryAcked})
|
|
if err != nil || len(before.Items) != 1 || before.Items[0].AcknowledgedAt == nil {
|
|
t.Fatal("fault occurred before durable ACK", err)
|
|
}
|
|
a.kill(t)
|
|
b := startOrbitProcess(t)
|
|
other := processClient(t, b, token)
|
|
gateway.target.Store(b.url)
|
|
gateway.offline.Store(false)
|
|
for consumer.Stats().Acknowledged == 0 {
|
|
select {
|
|
case <-ctx.Done():
|
|
t.Fatal("consumer failed to reconcile ACK on new process")
|
|
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)
|
|
}
|
|
after, err := other.ListEventDeliveries(ctx, sub.ID, DeliveryPageRequest{State: DeliveryAcked})
|
|
if err != nil || len(after.Items) != 1 || after.Items[0].Attempts != 1 || after.Items[0].AcknowledgedAt == nil || !after.Items[0].AcknowledgedAt.Equal(*before.Items[0].AcknowledgedAt) || calls.Load() != 1 {
|
|
t.Fatal("ACK receipt changed or committed handler reran", err)
|
|
}
|
|
t.Log("ACK commit verified before response loss and SIGKILL; replacement process returned original receipt without another handler or delivery attempt")
|
|
}
|
|
|
|
func TestIntegrationEventProcessSlowConsumerPeerProgress(t *testing.T) {
|
|
a := startOrbitProcess(t)
|
|
_, _, token := integrationWorkerClient(t)
|
|
apiA := processClient(t, a, token)
|
|
b := startOrbitProcess(t)
|
|
apiB := processClient(t, b, token)
|
|
if a.nodeID == b.nodeID {
|
|
t.Fatal("server processes reused node identity")
|
|
}
|
|
ctx, stop := context.WithTimeout(context.Background(), 45*time.Second)
|
|
defer stop()
|
|
sub := integrationEventSubscription(t, ctx, apiA, 3)
|
|
_ = integrationCancelledEvent(t, ctx, apiA, sub)
|
|
started := make(chan struct{})
|
|
release := make(chan struct{})
|
|
var callsA, callsB atomic.Int64
|
|
consumerA, err := NewEventConsumer(apiA, EventConsumerOptions{SubscriptionID: sub.ID, Capacity: 1, ShutdownTimeout: time.Second, Handle: func(work context.Context, _ ResultEvent) error {
|
|
if callsA.Add(1) == 1 {
|
|
close(started)
|
|
select {
|
|
case <-release:
|
|
case <-work.Done():
|
|
return work.Err()
|
|
}
|
|
}
|
|
return nil
|
|
}})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
run, cancel := context.WithCancel(ctx)
|
|
defer cancel()
|
|
doneA := make(chan error, 1)
|
|
go func() { doneA <- consumerA.Run(run) }()
|
|
select {
|
|
case <-started:
|
|
case <-ctx.Done():
|
|
t.Fatal("node A slow handler did not start")
|
|
case err := <-doneA:
|
|
t.Fatal("node A consumer stopped", err)
|
|
}
|
|
consumerB, err := NewEventConsumer(apiB, EventConsumerOptions{SubscriptionID: sub.ID, Capacity: 1, ShutdownTimeout: time.Second, Handle: func(context.Context, ResultEvent) error {
|
|
callsB.Add(1)
|
|
return nil
|
|
}})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
doneB := make(chan error, 1)
|
|
go func() { doneB <- consumerB.Run(run) }()
|
|
awaitConsumerEvent(t, consumerB, "connected")
|
|
_ = integrationCancelledEvent(t, ctx, apiA, sub)
|
|
deadline := time.Now().Add(5 * time.Second)
|
|
for consumerB.Stats().Acknowledged < 1 {
|
|
select {
|
|
case <-ctx.Done():
|
|
t.Fatal("node B did not progress while node A was slow")
|
|
case err := <-doneB:
|
|
t.Fatal("node B consumer stopped", err)
|
|
case <-time.After(20 * time.Millisecond):
|
|
}
|
|
if time.Now().After(deadline) {
|
|
t.Fatal("node B did not acknowledge peer delivery before deadline")
|
|
}
|
|
}
|
|
if callsA.Load() != 1 || callsB.Load() != 1 || consumerA.Stats().Acknowledged != 0 {
|
|
t.Fatalf("peer progress violated slow consumer isolation: callsA=%d callsB=%d statsA=%+v statsB=%+v", callsA.Load(), callsB.Load(), consumerA.Stats(), consumerB.Stats())
|
|
}
|
|
close(release)
|
|
for consumerA.Stats().Acknowledged < 1 {
|
|
select {
|
|
case <-ctx.Done():
|
|
t.Fatal("node A did not drain its held delivery")
|
|
case err := <-doneA:
|
|
t.Fatal("node A consumer stopped", err)
|
|
case <-time.After(20 * time.Millisecond):
|
|
}
|
|
}
|
|
cancel()
|
|
if err := awaitConsumerStop(t, doneA); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := awaitConsumerStop(t, doneB); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
page, err := apiB.ListEventDeliveries(ctx, sub.ID, DeliveryPageRequest{State: DeliveryAcked})
|
|
if err != nil || len(page.Items) != 2 || page.Items[0].Attempts != 1 || page.Items[1].Attempts != 1 {
|
|
t.Fatalf("peer delivery accounting failed: err=%v deliveries=%+v", err, page.Items)
|
|
}
|
|
t.Log("node B acknowledged a second delivery while node A's capacity-one handler was blocked; both nodes preserved one attempt per delivery")
|
|
}
|
|
|
|
// TestIntegrationEventProcessCrossNodeReconnectStorm exercises the failure mode
|
|
// where a consumer loses several WebSocket sessions while the gateway alternates
|
|
// between two Orbit nodes. The durable delivery must be acknowledged once and
|
|
// the handler must never run again during the reconnect storm.
|
|
func TestIntegrationEventProcessCrossNodeReconnectStorm(t *testing.T) {
|
|
a := startOrbitProcess(t)
|
|
_, _, token := integrationWorkerClient(t)
|
|
apiA := processClient(t, a, token)
|
|
b := startOrbitProcess(t)
|
|
apiB := processClient(t, b, token)
|
|
ctx, stop := context.WithTimeout(context.Background(), 45*time.Second)
|
|
defer stop()
|
|
sub := integrationEventSubscription(t, ctx, apiA, 3)
|
|
_ = integrationCancelledEvent(t, ctx, apiA, sub)
|
|
gateway := newProtocolFaultGateway(t, a.url, "/v1/events/connect", eventSubprotocol)
|
|
client, err := NewClient(gateway.server.URL, Options{Token: token, HTTPClient: gateway.server.Client()})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
var calls atomic.Int64
|
|
started := make(chan struct{})
|
|
release := make(chan struct{})
|
|
consumer, err := NewEventConsumer(client, EventConsumerOptions{
|
|
SubscriptionID: sub.ID, Capacity: 1, ShutdownTimeout: time.Second,
|
|
ReconnectMinDelay: 10 * time.Millisecond, ReconnectMaxDelay: 60 * time.Millisecond,
|
|
Handle: func(work context.Context, _ ResultEvent) error {
|
|
if calls.Add(1) == 1 {
|
|
close(started)
|
|
select {
|
|
case <-release:
|
|
case <-work.Done():
|
|
return work.Err()
|
|
}
|
|
}
|
|
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) }()
|
|
awaitConsumerEvent(t, consumer, "connected")
|
|
awaitConsumerEvent(t, consumer, "handler_started")
|
|
|
|
// Close whichever node currently owns the socket, then move the gateway to
|
|
// the other node. Repeat enough times to cover a reconnect burst, while the
|
|
// event itself remains one durable delivery.
|
|
for i := 0; i < 4; i++ {
|
|
gateway.disconnectActive()
|
|
if i%2 == 0 {
|
|
gateway.target.Store(b.url)
|
|
} else {
|
|
gateway.target.Store(a.url)
|
|
}
|
|
awaitConsumerEvent(t, consumer, "connected")
|
|
}
|
|
close(release)
|
|
for consumer.Stats().Acknowledged < 1 {
|
|
select {
|
|
case <-ctx.Done():
|
|
t.Fatal("cross-node consumer did not acknowledge delivery")
|
|
case err := <-done:
|
|
t.Fatalf("consumer stopped before acknowledgement: %v", err)
|
|
case <-time.After(20 * time.Millisecond):
|
|
}
|
|
}
|
|
cancel()
|
|
if err := awaitConsumerStop(t, done); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
stats := consumer.Stats()
|
|
if stats.Reconnects < 4 || stats.Acknowledged != 1 || calls.Load() != 1 {
|
|
t.Fatalf("cross-node reconnect storm duplicated work: stats=%+v calls=%d", stats, calls.Load())
|
|
}
|
|
page, err := apiB.ListEventDeliveries(ctx, sub.ID, DeliveryPageRequest{State: DeliveryAcked})
|
|
if err != nil || len(page.Items) != 1 || page.Items[0].Attempts != 1 {
|
|
t.Fatalf("durable delivery attempt changed during storm: err=%v deliveries=%+v", err, page.Items)
|
|
}
|
|
t.Logf("cross-node reconnect storm completed with %d reconnects, one handler call, and one durable delivery attempt", stats.Reconnects)
|
|
}
|