feat: add retention policy and stability verification

sdk/go/event_process_integration_test.go

@@ -193,3 +193,87 @@ func TestIntegrationEventProcessSlowConsumerPeerProgress(t *testing.T) {
 	}
 	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)
+}