Added filter for late txpool notifications

src/mempool.cpp

@@ -65,4 +65,17 @@ void Mempool::swap(std::vector<TxMempoolData>& transactions)
 	}
 }
 
+void Mempool::remove(const std::vector<hash>& tx_hashes)
+{
+	if (tx_hashes.empty()) {
+		return;
+	}
+
+	WriteLock lock(m_lock);
+
+	for (const hash& h : tx_hashes) {
+		m_transactions.erase(h);
+	}
+}
+
 } // namespace p2pool

src/mempool.h

@@ -56,6 +56,8 @@ public:
 		m_transactions.clear();
 	}
 
+	void remove(const std::vector<hash>& tx_hashes);
+
 private:
 	mutable uv_rwlock_t m_lock;
 	unordered_map<hash, TxMempoolData> m_transactions;

src/p2pool.cpp

@@ -584,8 +584,11 @@ const char* BLOCK_FOUND = "\n\
 | ######   #######  #######   #####   #    #      #        #######   #####   #     #  ######  |\n\
 -----------------------------------------------------------------------------------------------";
 
-void p2pool::handle_chain_main(ChainMain& data, const char* extra)
+void p2pool::handle_chain_main(ChainMain& data, const char* extra, const std::vector<hash>& tx_hashes_in_block)
 {
+	// These transactions were already mined, so remove them from mempool if any of them slipped through
+	m_mempool->remove(tx_hashes_in_block);
+
 	{
 		WriteLock lock(m_mainchainLock);
 

src/p2pool.h

@@ -85,7 +85,7 @@ public:
 
 	virtual void handle_tx(TxMempoolData& tx) override;
 	virtual void handle_miner_data(MinerData& data) override;
-	virtual void handle_chain_main(ChainMain& data, const char* extra) override;
+	virtual void handle_chain_main(ChainMain& data, const char* extra, const std::vector<hash>& tx_hashes_in_block) override;
 	virtual void handle_monero_block_broadcast(std::vector<std::vector<uint8_t>>&& blobs) override;
 
 #ifdef WITH_MERGE_MINING_DONATION

src/util.h

@@ -91,7 +91,7 @@ struct MinerCallbackHandler
 
 	virtual void handle_tx(TxMempoolData& tx) = 0;
 	virtual void handle_miner_data(MinerData& data) = 0;
-	virtual void handle_chain_main(ChainMain& data, const char* extra) = 0;
+	virtual void handle_chain_main(ChainMain& data, const char* extra, const std::vector<hash>& tx_hashes_in_block) = 0;
 	virtual void handle_monero_block_broadcast(std::vector<std::vector<uint8_t>>&& blobs) = 0;
 };
 

src/zmq_reader.cpp

@@ -253,8 +253,10 @@ bool ZMQReader::connect(const std::string& address, bool keep_monitor)
 	return true;
 }
 
-static std::vector<uint8_t> construct_monero_block_blob(rapidjson::Value* value)
+static std::vector<uint8_t> construct_monero_block_blob(rapidjson::Value* value, std::vector<hash>& out_transaction_hashes)
 {
+	out_transaction_hashes.clear();
+
 	std::vector<uint8_t> empty_blob;
 
 #define X(type, name) type name; if (!parseValue(*value, #name, name)) return empty_blob;
@@ -405,6 +407,7 @@ static std::vector<uint8_t> construct_monero_block_blob(rapidjson::Value* value)
 	auto arr2 = tx_hashes->value.GetArray();
 
 	writeVarint(arr2.Size(), blob);
+	out_transaction_hashes.reserve(arr2.Size());
 
 	for (auto i = arr2.begin(); i != arr2.end(); ++i) {
 		if (!i->IsString()) {
@@ -416,6 +419,7 @@ static std::vector<uint8_t> construct_monero_block_blob(rapidjson::Value* value)
 			return empty_blob;
 		}
 		blob.insert(blob.end(), h.h, h.h + HASH_SIZE);
+		out_transaction_hashes.emplace_back(h);
 	}
 
 	const uint8_t* p = reinterpret_cast<const uint8_t*>(&data);
@@ -525,7 +529,8 @@ void ZMQReader::parse(char* data, size_t size)
 		blobs.reserve(arr.Size());
 
 		for (auto i = arr.begin(); i != arr.end(); ++i) {
-			blobs.emplace_back(construct_monero_block_blob(i));
+			std::vector<hash> tx_hashes;
+			blobs.emplace_back(construct_monero_block_blob(i, tx_hashes));
 
 			if (!PARSE(*i, m_chainmainData, timestamp)) {
 				LOGWARN(1, "json-full-chain_main timestamp failed to parse, skipping it");
@@ -586,7 +591,7 @@ void ZMQReader::parse(char* data, size_t size)
 				continue;
 			}
 
-			m_handler->handle_chain_main(m_chainmainData, extra_it->value.GetString());
+			m_handler->handle_chain_main(m_chainmainData, extra_it->value.GetString(), tx_hashes);
 		}
 
 		m_handler->handle_monero_block_broadcast(std::move(blobs));

tests/src/block_template_tests.cpp

@@ -81,13 +81,30 @@ TEST(block_template, update)
 	ASSERT_EQ(blobs_hash, H("da11e1ee86779a559df63a55e0b238ce5a67b977e0f68a0b347a39d37096a4bc"));
 
 	// Test 2: mempool with high fee and low fee transactions, it must choose high fee transactions
-	for (uint64_t i = 0; i < 512; ++i) {
+	for (uint64_t i = 0; i < 513; ++i) {
 		TxMempoolData tx;
 		*reinterpret_cast<uint64_t*>(tx.id.h) = i;
 		tx.fee = (i < 256) ? 30000000 : 60000000;
 		tx.weight = 1500;
 		mempool.add(tx);
 	}
+	ASSERT_EQ(mempool.size(), 513);
+
+	// Test transaction removing from mempool
+	{
+		std::vector<hash> tx_hashes;
+
+		// Empty list, should do nothing
+		mempool.remove(tx_hashes);
+		ASSERT_EQ(mempool.size(), 513);
+
+		hash h;
+		*reinterpret_cast<uint64_t*>(h.h) = 512;
+		tx_hashes.push_back(h);
+
+		// Should remove a single hash
+		mempool.remove(tx_hashes);
+	}
 	ASSERT_EQ(mempool.size(), 512);
 
 	tpl.update(data, mempool, &wallet);