From 33dbe85355300fb3e825c0da275faf6472c9e1c8 Mon Sep 17 00:00:00 2001
From: Marius Guggenmos <marius@zircuit.com>
Date: Tue, 30 Jul 2024 21:58:52 +0200
Subject: [PATCH] fix flaky test TestHTTPNodeRPC

There is a race condition between subscribing and generating an event in
the test. The TestAPI now exposes the number of active subscriptions so
clients can ensure their subscription is active before calling the api.
---
 p2p/simulations/http_test.go | 40 ++++++++++++++++++++++++++++++++----
 1 file changed, 36 insertions(+), 4 deletions(-)

diff --git a/p2p/simulations/http_test.go b/p2p/simulations/http_test.go
index cd03e600f3..a832e3ba56 100644
--- a/p2p/simulations/http_test.go
+++ b/p2p/simulations/http_test.go
@@ -234,10 +234,11 @@ func (t *testService) Snapshot() ([]byte, error) {
 // * get and increment a counter
 // * subscribe to counter increment events
 type TestAPI struct {
-	state     *atomic.Value
-	peerCount *int64
-	counter   int64
-	feed      event.Feed
+	state               *atomic.Value
+	peerCount           *int64
+	counter             int64
+	feed                event.Feed
+	activeSubscriptions int64
 }
 
 func (t *TestAPI) PeerCount() int64 {
@@ -273,14 +274,17 @@ func (t *TestAPI) Events(ctx context.Context) (*rpc.Subscription, error) {
 		events := make(chan int64)
 		sub := t.feed.Subscribe(events)
 		defer sub.Unsubscribe()
+		atomic.AddInt64(&t.activeSubscriptions, 1)
 
 		for {
 			select {
 			case event := <-events:
 				notifier.Notify(rpcSub.ID, event)
 			case <-sub.Err():
+				atomic.AddInt64(&t.activeSubscriptions, -1)
 				return
 			case <-rpcSub.Err():
+				atomic.AddInt64(&t.activeSubscriptions, -1)
 				return
 			}
 		}
@@ -289,6 +293,10 @@ func (t *TestAPI) Events(ctx context.Context) (*rpc.Subscription, error) {
 	return rpcSub, nil
 }
 
+func (t *TestAPI) GetNumActiveSubscriptions() int64 {
+	return atomic.LoadInt64(&t.activeSubscriptions)
+}
+
 var testServices = adapters.LifecycleConstructors{
 	"test": newTestService,
 }
@@ -557,6 +565,14 @@ func TestHTTPNodeRPC(t *testing.T) {
 		t.Fatalf("error getting node RPC client: %s", err)
 	}
 
+	// get the number of subscriptions before subscribing to know what number to
+	// expect once it becomes active
+	var expectedActiveSubscriptions int64
+	if err := rpcClient1.CallContext(ctx, &expectedActiveSubscriptions, "test_getNumActiveSubscriptions"); err != nil {
+		t.Fatalf("error calling RPC method: %s", err)
+	}
+	expectedActiveSubscriptions += 1
+
 	// subscribe to events using client 1
 	events := make(chan int64, 1)
 	sub, err := rpcClient1.Subscribe(ctx, "test", events, "events")
@@ -565,6 +581,22 @@ func TestHTTPNodeRPC(t *testing.T) {
 	}
 	defer sub.Unsubscribe()
 
+	// make sure the subscription becomes active
+	var numActiveSubscriptions int64
+	for i := 0; i < 3; i++ {
+		err := rpcClient1.CallContext(ctx, &numActiveSubscriptions, "test_getNumActiveSubscriptions")
+		if err != nil {
+			t.Fatalf("error calling RPC method: %s", err)
+		}
+		if numActiveSubscriptions > 0 {
+			break
+		}
+		time.Sleep(100 * time.Millisecond)
+	}
+	if numActiveSubscriptions != expectedActiveSubscriptions {
+		t.Fatalf("subscription never became active")
+	}
+
 	// call some RPC methods using client 2
 	if err := rpcClient2.CallContext(ctx, nil, "test_add", 10); err != nil {
 		t.Fatalf("error calling RPC method: %s", err)