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