104 lines
3.9 KiB
Go
104 lines
3.9 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")
|
|
}
|