fix(Foundation): Fix FileChannel purge race with compression (#4848)

Foundation/include/Poco/ArchiveStrategy.h

@@ -26,6 +26,7 @@
 #include "Poco/Mutex.h"
 #include "Poco/Condition.h"
 #include <atomic>
+#include <functional>
 
 
 namespace Poco {
@@ -42,6 +43,8 @@ class Foundation_API ArchiveStrategy
 	/// using the gzip file format.
 {
 public:
+	using PurgeCallback = std::function<void()>;
+
 	ArchiveStrategy();
 	virtual ~ArchiveStrategy();
 
@@ -58,6 +61,10 @@ public:
 	void compress(bool flag = true);
 		/// Enables or disables compression of archived files.
 
+	void setPurgeCallback(PurgeCallback callback);
+		/// Sets a callback to be invoked after compression completes.
+		/// Used by FileChannel to trigger purging at the right time.
+
 protected:
 	void moveFile(const std::string& oldName, const std::string& newName);
 	bool exists(const std::string& name);
@@ -79,6 +86,9 @@ private:
 
 	std::atomic<bool> _compress;
 	std::atomic<ArchiveCompressor*> _pCompressor;
+
+protected:
+	PurgeCallback _purgeCallback;
 };
 
 
@@ -116,6 +126,11 @@ public:
 	/// file name. If the new file name exists, additionally a monotonic
 	/// increasing number is appended to the log file name.
 	{
+		FastMutex::ScopedLock l(_rotateMutex);
+
+		while (_compressingCount > 0)
+			_compressingComplete.wait(_rotateMutex, 1000);
+
 		std::string path = pFile->path();
 		delete pFile;
 		std::string archPath = path;
@@ -125,6 +140,11 @@ public:
 		if (exists(archPath)) archiveByNumber(archPath);
 		else moveFile(path, archPath);
 
+		// If no compression was started, invoke purge callback now.
+		// Otherwise, it will be invoked when compression completes.
+		if (_compressingCount == 0 && _purgeCallback)
+			_purgeCallback();
+
 		return new LogFile(path);
 	}
 

Foundation/src/ArchiveStrategy.cpp

@@ -110,6 +110,12 @@ void ArchiveStrategy::compress(bool flag)
 }
 
 
+void ArchiveStrategy::setPurgeCallback(PurgeCallback callback)
+{
+	_purgeCallback = std::move(callback);
+}
+
+
 void ArchiveStrategy::moveFile(const std::string& oldPath, const std::string& newPath)
 {
 	bool compressed = false;
@@ -214,7 +220,11 @@ void ArchiveStrategy::compressFile(const std::string& path)
 
 	_compressingCount--;
 	if (_compressingCount < 1)
+	{
 		_compressingComplete.broadcast();
+		if (_purgeCallback)
+			_purgeCallback();
+	}
 }
 
 
@@ -272,6 +282,12 @@ LogFile* ArchiveByNumberStrategy::archive(LogFile* pFile)
 		moveFile(oldPath, newPath);
 		--n;
 	}
+
+	// If no compression was started, invoke purge callback now.
+	// Otherwise, it will be invoked when compression completes.
+	if (_compressingCount == 0 && _purgeCallback)
+		_purgeCallback();
+
 	return new LogFile(basePath);
 }
 

Foundation/src/FileChannel.cpp

@@ -49,6 +49,7 @@ FileChannel::FileChannel():
 	_pArchiveStrategy(new ArchiveByNumberStrategy),
 	_pPurgeStrategy(new NullPurgeStrategy())
 {
+	_pArchiveStrategy->setPurgeCallback([this]() { purge(); });
 }
 
 
@@ -63,6 +64,7 @@ FileChannel::FileChannel(const std::string& path):
 	_pArchiveStrategy(new ArchiveByNumberStrategy),
 	_pPurgeStrategy(new NullPurgeStrategy())
 {
+	_pArchiveStrategy->setPurgeCallback([this]() { purge(); });
 }
 
 
@@ -94,7 +96,6 @@ void FileChannel::open()
 			try
 			{
 				_pFile = _pArchiveStrategy->archive(_pFile);
-				purge();
 			}
 			catch (...)
 			{
@@ -130,7 +131,6 @@ void FileChannel::log(const Message& msg)
 		try
 		{
 			_pFile = _pArchiveStrategy->archive(_pFile);
-			purge();
 		}
 		catch (...)
 		{
@@ -326,6 +326,7 @@ void FileChannel::setArchiveStrategy(ArchiveStrategy* strategy)
 
 	delete _pArchiveStrategy;
 	_pArchiveStrategy = strategy;
+	_pArchiveStrategy->setPurgeCallback([this]() { purge(); });
 }
 
 
@@ -348,6 +349,7 @@ void FileChannel::setArchive(const std::string& archive)
 	else throw InvalidArgumentException("archive", archive);
 	delete _pArchiveStrategy;
 	pStrategy->compress(_compress);
+	pStrategy->setPurgeCallback([this]() { purge(); });
 	_pArchiveStrategy = pStrategy;
 	_archive = archive;
 }

Foundation/testsuite/src/FileChannelTest.cpp

@@ -608,14 +608,13 @@ void FileChannelTest::testCompressedRotation()
 	for (const auto& f: files)
 		std::cout << "log file: " << f << std::endl;
 
-	assertEqual(5+1+1, files.size()); // 5+1 rotated files, current file
+	assertEqual(5+1, files.size()); // 5 archived files + current file
 	assertEqual("test.log", files[0]);
 	assertEqual("test.log.0.gz", files[1]);
 	assertEqual("test.log.1.gz", files[2]);
 	assertEqual("test.log.2.gz", files[3]);
 	assertEqual("test.log.3.gz", files[4]);
 	assertEqual("test.log.4.gz", files[5]);
-	assertEqual("test.log.5.gz", files[6]);
 
 	logsDir.remove(true);
 }