test(network): ack NATS messages before signalling test completion (backport)

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) {

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

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)
 		})