test: verify slow consumer peer progress

sdk/go/event_process_integration_test.go

@@ -101,3 +101,95 @@ func TestIntegrationEventProcessAckCommitResponseLoss(t *testing.T) {
 	}
 	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")
+}