From 17fcac4a563406e933ab1654b9acb3c44b5e85f5 Mon Sep 17 00:00:00 2001
From: Steven van der Vegt <steven.vandervegt@s-curve.nl>
Date: Wed, 9 Sep 2026 08:55:22 +0200
Subject: [PATCH] test(network): ack NATS messages before signalling test
 completion

TestNetwork_Reprocess called wg.Done() before msg.Ack(). Once Done
released the test, its cleanup shut down the embedded NATS server, and
the ack on the callback goroutine failed with "nats: connection closed"
on slow CI runners.

Also stop calling require inside NATS callback goroutines (FailNow must
run on the test goroutine) and read the ack error in the events
integration test under the mutex instead of through a shared variable.

Assisted-by: AI
---
 events/integration_test.go          | 7 +++++--
 network/network_integration_test.go | 4 ++--
 network/network_test.go             | 5 ++++-
 3 files changed, 11 insertions(+), 5 deletions(-)

diff --git a/events/integration_test.go b/events/integration_test.go
index 62422a6c1a..b7bc1b5b3a 100644
--- a/events/integration_test.go
+++ b/events/integration_test.go
@@ -45,12 +45,13 @@ func Test_pub_sub(t *testing.T) {
 		require.NoError(t, err)
 		defer conn.Close()
 		var found []byte
+		var ackErr error
 		foundMutex := sync.Mutex{}
 		err = stream.Subscribe(conn, "TEST", "TRANSACTIONS.tx", func(msg *nats.Msg) {
 			foundMutex.Lock()
 			defer foundMutex.Unlock()
 			found = msg.Data
-			err = msg.Ack()
+			ackErr = msg.Ack()
 		})
 
 		require.NoError(t, err)
@@ -64,7 +65,9 @@ func Test_pub_sub(t *testing.T) {
 			defer foundMutex.Unlock()
 			return bytes.Equal(found, []byte{1}), nil
 		}, 100*time.Millisecond, "timeout waiting for message")
-		require.NoError(t, err)
+		foundMutex.Lock()
+		defer foundMutex.Unlock()
+		require.NoError(t, ackErr)
 	})
 
 	t.Run("publish more than the subscriber can handle", func(t *testing.T) {
diff --git a/network/network_integration_test.go b/network/network_integration_test.go
index 0bc9af1c7d..0fcd602cb7 100644
--- a/network/network_integration_test.go
+++ b/network/network_integration_test.go
@@ -561,7 +561,7 @@ func TestNetworkIntegration_PrivateTransaction(t *testing.T) {
 			defer foundMutex.Unlock()
 			found = msg.Data
 			err := msg.Ack()
-			require.NoError(t, err)
+			assert.NoError(t, err)
 		})
 
 		node1DID := node1.network.nodeDID
@@ -829,7 +829,7 @@ func TestNetworkIntegration_AddedTransactionsAsEvents(t *testing.T) {
 		defer foundMutex.Unlock()
 		found = msg.Data
 		err := msg.Ack()
-		require.NoError(t, err)
+		assert.NoError(t, err)
 	})
 
 	// add a transaction
diff --git a/network/network_test.go b/network/network_test.go
index cd7e29b724..c813d713c6 100644
--- a/network/network_test.go
+++ b/network/network_test.go
@@ -898,8 +898,11 @@ func TestNetwork_Reprocess(t *testing.T) {
 		err = events.NewDisposableStream("REPROCESS_test", []string{"REPROCESS.*"}, 10).Subscribe(conn, t.Name(), "REPROCESS.*", func(m *nats.Msg) {
 			foundMutex.Lock()
 			defer foundMutex.Unlock()
+			// Signal completion only after the ack: once wg.Done() releases the test,
+			// its cleanup shuts down the NATS server and the ack would fail with
+			// "nats: connection closed".
+			defer wg.Done()
 			*counter++
-			wg.Done()
 			err := m.Ack()
 			assert.NoError(t, err)
 		})