[R21BC-161] Bug fix: do not re-broadcast transactions with past deadline

src/catapult/cache_tx/MemoryUtCache.cpp

@@ -23,6 +23,7 @@
 #include "CacheSizeLogger.h"
 #include "catapult/model/EntityInfo.h"
 #include "catapult/model/FeeUtils.h"
+#include "catapult/utils/NetworkTime.h"
 
 namespace catapult { namespace cache {
 
@@ -92,6 +93,8 @@ namespace catapult { namespace cache {
 		uint64_t totalSize = 0;
 		UnknownTransactions transactions;
 		for (const auto& data : m_transactionDataContainer) {
+			if (data.pEntity->Deadline <= utils::NetworkTime())
+				continue;
 			if (data.pEntity->MaxFee < model::CalculateTransactionFee(minFeeMultiplier, *data.pEntity, feeInterest, feeInterestDenominator))
 				continue;
 
@@ -144,8 +147,8 @@ namespace catapult { namespace cache {
 				if (m_maxCacheSize <= m_transactionDataContainer.size())
 					return false;
 
-				if (m_idLookup.cend() != m_idLookup.find(transactionInfo.EntityHash))
-					return false;
+				// if (m_idLookup.cend() != m_idLookup.find(transactionInfo.EntityHash))
+				// 	return false;
 
 				m_idLookup.emplace(transactionInfo.EntityHash, ++m_idSequence);
 				m_transactionDataContainer.emplace(transactionInfo, m_idSequence);

src/catapult/local/server/LocalNode.cpp

@@ -213,6 +213,18 @@ namespace catapult { namespace local {
 				return m_cacheHolder.cache();
 			}
 
+			cache::MemoryUtCacheProxy& utCache() const override {
+				return *m_pUtCache;
+			}
+
+			extensions::ServiceState& serviceState() const override {
+				return *m_pServiceState;
+			}
+
+			const extensions::ServiceLocator& serviceLocator() const override {
+				return m_serviceLocator;
+			}
+
 			model::ChainScore score() const override {
 				return m_score.get();
 			}

src/catapult/local/server/LocalNode.h

@@ -21,6 +21,9 @@
 #pragma once
 #include "catapult/local/ProcessHost.h"
 #include "catapult/utils/DiagnosticCounter.h"
+#include "catapult/cache_tx/MemoryUtCache.h"
+#include "catapult/extensions/ServiceState.h"
+#include "catapult/extensions/ServiceLocator.h"
 #include <memory>
 #include <vector>
 
@@ -75,6 +78,15 @@ namespace catapult { namespace local {
 		/// Gets the current cache.
 		virtual const cache::CatapultCache& cache() const = 0;
 
+		/// Gets the current MemoryUtCacheProxy
+		virtual cache::MemoryUtCacheProxy& utCache() const = 0;
+
+		/// Gets the current service state 
+		virtual extensions::ServiceState& serviceState() const = 0;
+
+		/// Gets the current service locator 
+		virtual const extensions::ServiceLocator& serviceLocator() const = 0;
+
 		/// Gets the current chain score.
 		virtual model::ChainScore score() const = 0;
 

tests/int/node/basic/LocalNodeApiRequestTests.cpp

@@ -98,6 +98,25 @@ namespace catapult { namespace local {
 		AssertReaderConnection(stats);
 	}
 
+	TEST(TEST_CLASS, ExpiredTransactionNotPushed) {
+		// Arrange:
+		TestContext context;
+		test::WaitForBoot(context);
+
+		// Act:
+		test::ExternalSourceConnection connection;
+		auto pIo = test::PushExpiredTransaction(connection);
+		WAIT_FOR_ONE_EXPR(connection.writeAttempts());
+
+		// Assert: no transaction element was added
+		auto stats = context.stats();
+		EXPECT_EQ(0u, stats.NumAddedBlockElements);
+		EXPECT_EQ(0u, stats.NumAddedTransactionElements);
+
+		// - the connection is still active
+		AssertReaderConnection(stats);
+	}
+
 	// endregion
 
 	// region pull (unsupported)

tests/int/node/basic/LocalNodeRequestTests.cpp

@@ -23,6 +23,8 @@
 #include "tests/int/node/test/LocalNodeRequestTestUtils.h"
 #include "tests/int/node/test/PeerLocalNodeTestContext.h"
 #include "tests/TestHarness.h"
+#include "catapult/model/EntityInfo.h"
+#include "catapult/utils/MemoryUtils.h"
 
 namespace catapult { namespace local {
 
@@ -158,10 +160,65 @@ namespace catapult { namespace local {
 		context.assertSingleReaderConnection();
 	}
 
+	TEST(TEST_CLASS, ExpiredTransactionNotPushedToLocalNode) {
+		// Arrange:
+		TestContext context;
+		test::WaitForBoot(context);
+
+		// Act:
+		test::ExternalSourceConnection connection;
+		auto pIo = test::PushExpiredTransaction(connection);
+		WAIT_FOR_ONE_EXPR(connection.writeAttempts());
+
+		// Assert: no transaction element was added
+		auto stats = context.stats();
+		EXPECT_EQ(0u, stats.NumAddedBlockElements);
+		EXPECT_EQ(0u, stats.NumAddedTransactionElements);
+
+		// - the connection is still active
+		context.assertSingleReaderConnection();
+	}
 	// endregion
 
 	// region pull
 
+
+	TEST(TEST_CLASS, ExpiredTransactionNotPulledToLocalNode) {
+		// Arrange:
+		TestContext context;
+		test::Sleep(5000);
+
+		// Act: 
+
+		// // Sanity: 1st local node has zero unconfirmed transaction
+
+		EXPECT_EQ(0u, context.localNode().serviceState().utCache().modifier().size());
+
+		//add valid transaction to first local node
+		auto pTransaction1 = utils::UniqueToShared(test::createValidTransaction());
+		model::TransactionInfo transactionInfo1(pTransaction1, Height(123));
+		EXPECT_EQ(true, context.localNode().serviceState().utCache().modifier().add(transactionInfo1));
+
+		// // // Sanity: 1st local node has one unconfirmed transaction
+		EXPECT_EQ(1u, context.localNode().serviceState().utCache().modifier().size());
+
+		// // //add expired transaction to first local node
+		auto pTransaction2 = utils::UniqueToShared(test::createExpiredTransaction());
+		model::TransactionInfo transactionInfo2(pTransaction2, Height(123));
+		EXPECT_EQ(true, context.localNode().serviceState().utCache().modifier().add(transactionInfo2));
+
+		// // Sanity: 1st local node has two unconfirmed transactions
+		EXPECT_EQ(2u, context.localNode().serviceState().utCache().modifier().size());
+
+		// Act: reboot local partner node
+		// context.bootPartnerNode();
+		test::pullUnconfirmedTransactions(context.partnerNode().serviceLocator(), context.partnerNode().serviceState());
+		test::Sleep(5000);
+
+		// Assert: only the expired transaction is not pulled
+		EXPECT_EQ(1u, context.partnerNode().serviceState().utCache().modifier().size());
+	}
+
 	CHAIN_API_INT_VALID_TRAITS_BASED_TEST(CanGetResponse) {
 		// Assert: the connection is still active
 		test::AssertCanGetResponse<TApiTraits>(TestContext(), TestContext::AssertSingleReaderConnection);

tests/int/node/test/LocalNodeRequestTestUtils.cpp

@@ -30,6 +30,17 @@
 #include "tests/test/core/BlockTestUtils.h"
 #include "tests/test/core/mocks/MockBlockchainConfigurationHolder.h"
 #include "tests/test/local/RealTransactionFactory.h"
+#include "catapult/utils/NetworkTime.h"
+#include "catapult/net/PacketWriters.h"
+#include "catapult/api/LocalChainApi.h"
+#include "catapult/api/RemoteChainApi.h"
+#include "catapult/api/RemoteTransactionApi.h"
+#include "catapult/chain/UtSynchronizer.h"
+#include "catapult/config/BlockchainConfiguration.h"
+#include "catapult/extensions/LocalNodeChainScore.h"
+#include "catapult/extensions/PeersConnectionTasks.h"
+#include "catapult/extensions/SynchronizerTaskCallbacks.h"
+#include "catapult/thread/Task.h"
 
 namespace catapult { namespace test {
 
@@ -64,6 +75,7 @@ namespace catapult { namespace test {
 
 		WAIT_FOR(isWriteFinished);
 		CATAPULT_LOG(debug) << " <<< push finished";
+		connection.incrementWriteAttempts();
 		return connection.io();
 	}
 
@@ -97,11 +109,57 @@ namespace catapult { namespace test {
 	}
 
 	std::shared_ptr<ionet::PacketIo> PushValidTransaction(ExternalSourceConnection& connection) {
+		auto pTransaction = createValidTransaction();
+		return PushEntity(connection, ionet::PacketType::Push_Transactions, std::shared_ptr<model::Transaction>(std::move(pTransaction)));
+	}
+
+	std::shared_ptr<ionet::PacketIo> PushExpiredTransaction(ExternalSourceConnection& connection) {
+		auto pTransaction = createExpiredTransaction();
+		return PushEntity(connection, ionet::PacketType::Push_Transactions, std::shared_ptr<model::Transaction>(std::move(pTransaction)));
+	}
+
+	model::UniqueEntityPtr<model::Transaction> createValidTransaction() {
 		auto recipient = test::GenerateRandomUnresolvedAddress();
 		auto pTransaction = CreateTransferTransaction(GetNemesisAccountKeyPair(), recipient, Amount(10000));
-		return PushEntity(connection, ionet::PacketType::Push_Transactions, std::shared_ptr<model::Transaction>(std::move(pTransaction)));
+		return pTransaction;
+	}
+
+	model::UniqueEntityPtr<model::Transaction> createExpiredTransaction() {
+		auto pTransaction = createValidTransaction();
+		pTransaction->Deadline = Timestamp(utils::NetworkTime().unwrap() - 1000);
+		return pTransaction;
+	}
+
+	std::shared_ptr<net::PacketWriters> GetPacketWriters(const extensions::ServiceLocator& locator) {
+		return locator.service<net::PacketWriters>("writers");
+	}
+
+	thread::Task CreatePullUtTask(const extensions::ServiceState& state, net::PacketWriters& packetWriters) {
+		auto utSynchronizer = chain::CreateUtSynchronizer(
+				[pConfigHolder = state.pluginManager().configHolder()]() {
+					return config::GetMinFeeMultiplier(pConfigHolder->Config());
+				},
+				[&cache = state.utCache()]() { return cache.view().shortHashes(); },
+				state.hooks().transactionRangeConsumerFactory()(disruptor::InputSource::Remote_Pull));
+
+		thread::Task task;
+		task.Name = "pull unconfirmed transactions task";
+		task.Callback = CreateChainSyncAwareSynchronizerTaskCallback(
+				std::move(utSynchronizer),
+				api::CreateRemoteTransactionApi,
+				packetWriters,
+				state,
+				task.Name);
+		return task;
+	}
+
+	void pullUnconfirmedTransactions(const extensions::ServiceLocator& locator, extensions::ServiceState& state) {
+		auto& packetWriters = *GetPacketWriters(locator);
+		auto task = CreatePullUtTask(state, packetWriters);
+		task.Callback();
 	}
 
+
 	// endregion
 
 	// region height

tests/int/node/test/LocalNodeRequestTestUtils.h

@@ -34,6 +34,9 @@
 #include "tests/test/local/LocalTestUtils.h"
 #include "tests/test/nodeps/MijinConstants.h"
 #include "tests/test/other/RemoteApiFactory.h"
+#include "catapult/extensions/ServiceLocator.h"
+#include "catapult/net/PacketWriters.h"
+#include "catapult/thread/Task.h"
 
 namespace catapult { namespace test {
 
@@ -52,6 +55,7 @@ namespace catapult { namespace test {
 				, m_clientKeyPair(crypto::KeyPair::FromPrivate(GenerateRandomPrivateKey()))
 				, m_pConnector(net::CreateServerConnector(m_pPool, m_clientKeyPair, net::ConnectionSettings()))
 				, m_localNode(node)
+				, m_writeAttempts(0)
 		{}
 
 	public:
@@ -78,6 +82,16 @@ namespace catapult { namespace test {
 			});
 		}
 
+		/// increments number of write attempts
+		void incrementWriteAttempts() {
+			m_writeAttempts++;
+		}
+
+		/// return number of write attempts
+		int writeAttempts() {
+			return m_writeAttempts;
+		}
+
 	private:
 		static model::TransactionRegistry CreateTransactionRegistry();
 
@@ -88,6 +102,7 @@ namespace catapult { namespace test {
 		ionet::Node m_localNode;
 
 		std::shared_ptr<ionet::PacketIo> m_pIo;
+		int m_writeAttempts;
 	};
 
 	// endregion
@@ -131,6 +146,25 @@ namespace catapult { namespace test {
 	/// Pushes a valid transaction to \a connection.
 	std::shared_ptr<ionet::PacketIo> PushValidTransaction(ExternalSourceConnection& connection);
 
+	/// Pushes an expired transaction to \a connection.
+	std::shared_ptr<ionet::PacketIo> PushExpiredTransaction(ExternalSourceConnection& connection);
+
+	// creat a valid transaction
+	model::UniqueEntityPtr<model::Transaction> createValidTransaction();
+
+	// create an expired transaction
+	model::UniqueEntityPtr<model::Transaction> createExpiredTransaction();
+
+	std::shared_ptr<net::PacketWriters> GetPacketWriters(const extensions::ServiceLocator& locator);
+
+	thread::Task CreatePullUtTask(const extensions::ServiceState& state, net::PacketWriters& packetWriters);
+
+
+
+	// pull unconfirmed transaction given \a locator and \a state
+	void pullUnconfirmedTransactions(const extensions::ServiceLocator& locator, extensions::ServiceState& state);
+
+
 	// endregion
 
 	// region pull

tests/int/node/test/LocalNodeTestContext.h

@@ -74,14 +74,15 @@ namespace catapult { namespace test {
 				, m_partnerTempDir("lntc_partner" + tempDirPostfix) {
 			initializeDataDirectory(m_tempDir.name());
 
-			if (HasFlag(NodeFlag::With_Partner, nodeFlag)) {
-				initializeDataDirectory(m_partnerTempDir.name());
+			if (!HasFlag(NodeFlag::Require_Explicit_Partner_Boot, nodeFlag))
+				if (HasFlag(NodeFlag::With_Partner, nodeFlag)) {
+					initializeDataDirectory(m_partnerTempDir.name());
 
-				// need to call configTransform first so that partner node loads all required transaction plugins
-				auto config = CreatePrototypicalBlockchainConfiguration(m_partnerTempDir.name());
-				m_configTransform(config);
-				m_pLocalPartnerNode = BootLocalPartnerNode(std::move(config), m_partnerServerKeyPair, nodeFlag);
-			}
+					// need to call configTransform first so that partner node loads all required transaction plugins
+					auto config = CreatePrototypicalBlockchainConfiguration(m_partnerTempDir.name());
+					m_configTransform(config);
+					m_pLocalPartnerNode = BootLocalPartnerNode(std::move(config), m_partnerServerKeyPair, nodeFlag);
+				}
 
 			if (!HasFlag(NodeFlag::Require_Explicit_Boot, nodeFlag))
 				boot();
@@ -119,6 +120,11 @@ namespace catapult { namespace test {
 			return *m_pLocalNode;
 		}
 
+		/// Gets the local partner node.
+		local::LocalNode& partnerNode() const {
+			return *m_pLocalPartnerNode;
+		}
+
 		/// Gets the node stats.
 		auto stats() const {
 			return TTraits::CountersToLocalNodeStats(m_pLocalNode->counters());
@@ -147,6 +153,15 @@ namespace catapult { namespace test {
 			return config;
 		}
 
+		void bootPartnerNode() {
+			initializeDataDirectory(m_partnerTempDir.name());
+
+			// need to call configTransform first so that partner node loads all required transaction plugins
+			auto config = CreatePrototypicalBlockchainConfiguration(m_partnerTempDir.name());
+			m_configTransform(config);
+			m_pLocalPartnerNode = BootLocalPartnerNode(std::move(config), m_partnerServerKeyPair, NodeFlag::Require_Explicit_Partner_Boot);
+		}
+
 	public:
 		/// Boots a new local node.
 		/// \note This overload is intended to be called only for nodes that require explicit booting.

tests/int/node/test/LocalNodeTestUtils.h

@@ -70,7 +70,9 @@ namespace catapult { namespace test {
 		Require_Explicit_Boot = 256,
 
 		/// Node supporting auto sync cleanup.
-		Auto_Sync_Cleanup = 512
+		Auto_Sync_Cleanup = 512,
+
+		Require_Explicit_Partner_Boot = 1024
 	};
 
 	MAKE_BITWISE_ENUM(NodeFlag)