fix: ObsoleteFileMaintainer is closed during catalog replacement

When catalog is replaced with another catalog contents, its ObsoleteFileMaintainer is closed which effectivelly stops old files purging. The replaced catalog continues to be used (and functional) while the purging logic is non functional.

Refs: #989
Signed-off-by: Jan Novotný <novotnaci@gmail.com>

Author: novoj

evita_common/src/main/java/io/evitadb/utils/FolderLock.java

@@ -27,6 +27,7 @@
 import io.evitadb.exception.FolderAlreadyUsedException;
 import io.evitadb.exception.GenericEvitaInternalError;
 import io.evitadb.exception.UnexpectedIOException;
+import io.evitadb.utils.IOUtils.ExceptionThrowingRunnable;
 import lombok.extern.slf4j.Slf4j;
 
 import javax.annotation.Nonnull;
@@ -115,8 +116,8 @@ public void close() throws IOException {
 				"Failed to release and close the lock file " + this.lockFilePath + ".",
 				"Failed to release and close the lock file."
 			),
-			(IOUtils.IOExceptionThrowingRunnable) this.lockFileLock::release,
-			(IOUtils.IOExceptionThrowingRunnable) this.lockFileChannel::close
+			(ExceptionThrowingRunnable) this.lockFileLock::release,
+			(ExceptionThrowingRunnable) this.lockFileChannel::close
 		);
 		FileUtils.deleteFileIfExists(this.lockFilePath);
 		ACQUIRED_LOCKS.remove(this.lockFilePath.toAbsolutePath().toString());

evita_common/src/main/java/io/evitadb/utils/IOUtils.java

@@ -111,26 +111,26 @@ public static long copy(
 	}
 
 	/**
-	 * Executes lambda logic encapsulated in {@link IOExceptionThrowingRunnable} instances, suppressing
+	 * Executes lambda logic encapsulated in {@link ExceptionThrowingRunnable} instances, suppressing
 	 * and aggregating any {@link IOException} that occurs during execution. If any exceptions are thrown, they
 	 * are encapsulated and re-thrown as a single exception provided by the {@code exceptionFactory}.
 	 *
 	 * @param <T>               the type of exception that will be thrown if any {@link IOException} occurs
 	 * @param exceptionFactory  a supplier that provides an exception of type {@code T}, used to wrap any
 	 *                          {@link IOException} thrown during the execution of the provided runnables
-	 * @param consumer          varargs of {@link IOExceptionThrowingRunnable} instances which encapsulate
+	 * @param consumer          varargs of {@link ExceptionThrowingRunnable} instances which encapsulate
 	 *                          the resources/actions to be closed or executed
 	 * @throws T                the consolidated exception containing any {@link IOException}s that were
 	 *                          thrown by the provided runnables
 	 */
 	public static <T extends RuntimeException> void executeSafely(
 		@Nonnull Supplier<T> exceptionFactory,
-		@Nonnull IOExceptionThrowingRunnable consumer
+		@Nonnull ExceptionThrowingRunnable consumer
 	) throws T {
 		T exception = null;
 		try {
 			consumer.run();
-		} catch (IOException e) {
+		} catch (Exception e) {
 			exception = exceptionFactory.get();
 			exception.addSuppressed(e);
 		}
@@ -179,7 +179,7 @@ public static <S, T extends RuntimeException> void executeSafely(
 	 *                          {@link IOException} thrown during the execution of the provided runnables
 	 * @param consumer          varargs of {@link IOExceptionThrowingConsumer} instances which encapsulate
 	 *                          the resources/actions to be closed or executed
-	 * @throws T                the consolidated exception containing any {@link IOException}s that were
+	 * @throws U                the consolidated exception containing any {@link Exception}s that were
 	 *                          thrown by the provided runnables
 	 */
 	public static <S, T, U extends RuntimeException> void executeSafely(
@@ -234,24 +234,24 @@ public static <S, T extends RuntimeException> S executeSafely(
 	}
 
 	/**
-	 * Closes multiple resources encapsulated in {@link IOExceptionThrowingRunnable} instances, suppressing
+	 * Closes multiple resources encapsulated in {@link ExceptionThrowingRunnable} instances, suppressing
 	 * and aggregating any {@link IOException} that occurs during execution. If any exceptions are thrown, they
 	 * are encapsulated and re-thrown as a single exception provided by the {@code exceptionFactory}.
 	 *
 	 * @param <T>               the type of exception that will be thrown if any {@link IOException} occurs
 	 * @param exceptionFactory  a supplier that provides an exception of type {@code T}, used to wrap any
 	 *                          {@link IOException} thrown during the execution of the provided runnables
-	 * @param runnable          varargs of {@link IOExceptionThrowingRunnable} instances which encapsulate
+	 * @param runnable          varargs of {@link ExceptionThrowingRunnable} instances which encapsulate
 	 *                          the resources/actions to be closed or executed
 	 * @throws T                the consolidated exception containing any {@link IOException}s that were
 	 *                          thrown by the provided runnables
 	 */
 	public static <T extends RuntimeException> void close(
 		@Nonnull Supplier<T> exceptionFactory,
-		@Nonnull IOExceptionThrowingRunnable... runnable
+		@Nonnull ExceptionThrowingRunnable... runnable
 	) throws T {
 		T exception = null;
-		for (IOExceptionThrowingRunnable lambda : runnable) {
+		for (ExceptionThrowingRunnable lambda : runnable) {
 			try {
 				lambda.run();
 			} catch (Exception e) {
@@ -269,7 +269,7 @@ public static <T extends RuntimeException> void close(
 	 * and aggregating any {@link Throwable} that occurs during execution. If any exceptions are thrown, they
 	 * are encapsulated and re-thrown as a single exception provided by the {@code exceptionFactory}.
 	 *
-	 * Unlike {@link #close(Supplier, IOExceptionThrowingRunnable...)}, this method catches any {@link Throwable}
+	 * Unlike {@link #close(Supplier, ExceptionThrowingRunnable...)}, this method catches any {@link Throwable}
 	 * rather than just {@link Exception}, providing a more robust safety net for resource cleanup.
 	 *
 	 * @param <T>               the type of exception that will be thrown if any exception occurs
@@ -299,7 +299,7 @@ public static <T extends RuntimeException> void closeSafely(
 	}
 
 	/**
-	 * Executes the provided {@link IOExceptionThrowingRunnable} instances, ensuring that exceptions thrown
+	 * Executes the provided {@link ExceptionThrowingRunnable} instances, ensuring that exceptions thrown
 	 * during their execution are logged but not propagated. This method is typically used for safely closing
 	 * resources without allowing individual close failures to disrupt the overall process.
 	 *
@@ -310,9 +310,9 @@ public static <T extends RuntimeException> void closeSafely(
 	 * @throws T if a runtime exception specific to the implementation needs propagation
 	 */
 	public static <T extends RuntimeException> void closeQuietly(
-		@Nonnull IOExceptionThrowingRunnable... runnable
+		@Nonnull ExceptionThrowingRunnable... runnable
 	) throws T {
-		for (IOExceptionThrowingRunnable lambda : runnable) {
+		for (ExceptionThrowingRunnable lambda : runnable) {
 			try {
 				lambda.run();
 			} catch (Exception e) {
@@ -327,7 +327,7 @@ public static <T extends RuntimeException> void closeQuietly(
 	 * during their execution are logged but not propagated. This method is typically used for safely closing
 	 * resources without allowing individual close failures to disrupt the overall process.
 	 *
-	 * Unlike {@link #closeQuietly(IOExceptionThrowingRunnable...)}, this method catches any {@link Throwable}
+	 * Unlike {@link #closeQuietly(ExceptionThrowingRunnable...)}, this method catches any {@link Throwable}
 	 * rather than just {@link Exception}, providing a more robust safety net for resource cleanup.
 	 *
 	 * @param runnable the runnable instances, which may throw any exception during execution
@@ -351,25 +351,25 @@ public static <T extends RuntimeException> void closeSafely(
 
 	/**
 	 * Represents a functional interface that can be used to encapsulate a block of code
-	 * that may throw an {@link IOException}. This interface is effectively a specialized
+	 * that may throw an {@link Exception}. This interface is effectively a specialized
 	 * form of {@link Runnable} for operations where checked I/O exceptions need to be handled.
 	 *
 	 * Implementations of this interface enable the execution of operations with the awareness
-	 * and explicit handling of {@link IOException}. This is especially useful in scenarios
+	 * and explicit handling of {@link Exception}. This is especially useful in scenarios
 	 * where multiple such operations need to be executed or wrapped with exception aggregation,
 	 * for example, in utility methods like resource handling or cleanup.
 	 *
 	 * Method {@code run} is similar to the {@link Runnable#run()} method but allows an
-	 * {@link IOException} to be thrown.
+	 * {@link Exception} to be thrown.
 	 *
 	 * Functional-style programming can make use of this interface to define inline behaviors
-	 * for operations expected to throw {@link IOException}. It can also be integrated with
+	 * for operations expected to throw {@link Exception}. It can also be integrated with
 	 * various utility methods that leverage this interface for exception handling and resource management.
 	 */
 	@FunctionalInterface
-	public interface IOExceptionThrowingRunnable {
+	public interface ExceptionThrowingRunnable {
 
-		void run() throws IOException;
+		void run() throws Exception;
 
 	}
 

evita_engine/src/main/java/io/evitadb/core/executor/DelayedAsyncTask.java

@@ -25,6 +25,7 @@
 
 import io.evitadb.core.metric.event.system.BackgroundTaskFinishedEvent;
 import io.evitadb.core.metric.event.system.BackgroundTaskStartedEvent;
+import io.evitadb.exception.GenericEvitaInternalError;
 import io.evitadb.utils.Assert;
 import lombok.Getter;
 import lombok.extern.slf4j.Slf4j;
@@ -147,6 +148,7 @@ public DelayedAsyncTask(
 	 * The task is scheduled using the scheduler's schedule method.
 	 */
 	public void scheduleImmediately() {
+		assertNotClosed();
 		final OffsetDateTime now = OffsetDateTime.now();
 		if (this.nextPlannedExecution.compareAndExchange(OffsetDateTime.MIN, now) == OffsetDateTime.MIN) {
 			scheduleTask(computeMinimalSchedulingGap(now.toInstant().toEpochMilli()));
@@ -162,6 +164,8 @@ public void scheduleImmediately() {
 	 * The task is scheduled using the scheduler's schedule method.
 	 */
 	public void schedule() {
+		assertNotClosed();
+
 		if (this.delay == Long.MAX_VALUE) {
 			// the task is manual task and will never be scheduled
 			return;
@@ -191,6 +195,20 @@ public void close() throws IOException {
 			}
 			// release the lambda to allow garbage collection
 			this.lambda.set(null);
+			this.nextPlannedExecution.set(OffsetDateTime.MIN);
+		}
+	}
+
+	/**
+	 * Ensures that the task is not closed before proceeding with its execution or scheduling.
+	 * If the task is already marked as closed, this method throws a GenericEvitaInternalError
+	 * to indicate that the operation cannot be performed on a closed task.
+	 *
+	 * @throws GenericEvitaInternalError if the task is marked as closed
+	 */
+	private void assertNotClosed() {
+		if (this.closed.get()) {
+			throw new GenericEvitaInternalError("Cannot schedule task `" + this.taskName + "` that has been closed.");
 		}
 	}
 
@@ -223,8 +241,9 @@ private long computeMinimalSchedulingGap(long nowMillis) {
 		if (lastFinishedExecutionTime.equals(OffsetDateTime.MIN)) {
 			return 0;
 		} else {
-			return Math.max(0, this.minimalSchedulingGap - (nowMillis - lastFinishedExecutionTime.toInstant()
-			                                                                                     .toEpochMilli())
+			return Math.max(
+				0, this.minimalSchedulingGap - (nowMillis - lastFinishedExecutionTime.toInstant()
+					.toEpochMilli())
 			);
 		}
 	}

evita_engine/src/main/java/io/evitadb/core/file/ExportFileService.java

@@ -494,24 +494,30 @@ void purgeFiles(@Nonnull OffsetDateTime thresholdDate) {
 			log.error("Failed to list files in the directory: {}", this.storageOptions.exportDirectory(), e);
 		}
 
-		// then check the size of the directory and delete oldest files until the directory size is below the limit
-		final long directorySize = FileUtils.getDirectorySize(this.storageOptions.exportDirectory());
-		// delete the oldest files until the directory size is below the limit
-		if (directorySize > this.storageOptions.exportDirectorySizeLimitBytes()) {
-			final List<FileForFetch> filesByCreationDate = this.files.stream()
-				.sorted(Comparator.comparing(FileForFetch::created))
-				.toList();
-			long savedSize = 0L;
-			for (FileForFetch it : filesByCreationDate) {
-				log.info("Purging the oldest file, because the export directory grew too big: {}", it);
-				final long metadataFileSize = it.metadataPath(this.storageOptions.exportDirectory()).toFile().length();
-				deleteFile(it.fileId());
-				savedSize += it.totalSizeInBytes() + metadataFileSize;
-				// finish removing files if the directory size is below the limit
-				if (directorySize - savedSize <= this.storageOptions.exportDirectorySizeLimitBytes()) {
-					break;
+		try {
+			// then check the size of the directory and delete oldest files until the directory size is below the limit
+			final long directorySize = FileUtils.getDirectorySize(this.storageOptions.exportDirectory());
+			// delete the oldest files until the directory size is below the limit
+			if (directorySize > this.storageOptions.exportDirectorySizeLimitBytes()) {
+				final List<FileForFetch> filesByCreationDate = this.files.stream()
+					.sorted(Comparator.comparing(FileForFetch::created))
+					.toList();
+				long savedSize = 0L;
+				for (FileForFetch it : filesByCreationDate) {
+					log.info("Purging the oldest file, because the export directory grew too big: {}", it);
+					final long metadataFileSize = it.metadataPath(this.storageOptions.exportDirectory())
+						.toFile()
+						.length();
+					deleteFile(it.fileId());
+					savedSize += it.totalSizeInBytes() + metadataFileSize;
+					// finish removing files if the directory size is below the limit
+					if (directorySize - savedSize <= this.storageOptions.exportDirectorySizeLimitBytes()) {
+						break;
+					}
 				}
 			}
+		} catch (UnexpectedIOException e) {
+			log.error("Failed to calculate size of the directory: {}", this.storageOptions.exportDirectory(), e);
 		}
 	}
 

evita_engine/src/main/java/io/evitadb/core/traffic/task/TrafficRecorderTask.java

@@ -39,6 +39,7 @@
 import io.evitadb.stream.RandomAccessFileInputStream;
 import io.evitadb.utils.Assert;
 import io.evitadb.utils.IOUtils;
+import io.evitadb.utils.IOUtils.ExceptionThrowingRunnable;
 import io.evitadb.utils.StringUtils;
 import lombok.extern.slf4j.Slf4j;
 
@@ -212,7 +213,7 @@ private FileForFetch start() {
 			if (exportSessionSink != null) {
 				IOUtils.close(
 					() -> new UnexpectedIOException("Failed to close export session sink."),
-					(IOUtils.IOExceptionThrowingRunnable) exportSessionSink::close
+					(ExceptionThrowingRunnable) exportSessionSink::close
 				);
 			}
 		}

evita_functional_tests/src/test/java/io/evitadb/store/offsetIndex/OffsetIndexTest.java

@@ -299,33 +299,42 @@ void tearDown() throws IOException {
 	@ParameterizedTest
 	@MethodSource("combineSettings")
 	void shouldSerializeAndReconstructEmptyOffsetIndex(Crc32Check crc32Check, Compression compression) {
-		final InsertionOutput insertionOutput = shouldSerializeAndReconstructOffsetIndex(
-			configure(StorageOptions.temporary(), crc32Check, compression), EntityBodyStoragePart::new, 0
-		);
-		IOUtils.closeQuietly(insertionOutput.fileOffsetIndex::close);
+		final StorageOptions storageOptions = configure(StorageOptions.temporary(), crc32Check, compression);
+		try (final ObservableOutputKeeper observableOutputKeeper = new ObservableOutputKeeper(TEST_CATALOG, storageOptions, Mockito.mock(Scheduler.class))) {
+			final InsertionOutput insertionOutput = shouldSerializeAndReconstructOffsetIndex(
+				storageOptions,
+				observableOutputKeeper,
+				EntityBodyStoragePart::new, 0
+			);
+			IOUtils.closeQuietly(insertionOutput.fileOffsetIndex::close);
+		}
 	}
 
 	@DisplayName("Offset index can be stored empty and then new records added.")
 	@ParameterizedTest
 	@MethodSource("combineSettings")
 	void shouldSerializeEmptyOffsetIndexWithLaterAddingRecordsAndReconstructCorrectly(Crc32Check crc32Check, Compression compression) {
 		final StorageOptions storageOptions = configure(StorageOptions.temporary(), crc32Check, compression);
-		final InsertionOutput insertionOutput = shouldSerializeAndReconstructOffsetIndex(storageOptions, EntityBodyStoragePart::new, 0);
-		final OffsetIndex offsetIndex = insertionOutput.fileOffsetIndex();
-		final InsertionOutput insertionOutput2 = createRecordsInFileOffsetIndex(offsetIndex, 100, 0, 1);
-		/* input count records +1 record for the OffsetIndex itself */
-		if (crc32Check == Crc32Check.YES) {
+		try (final ObservableOutputKeeper observableOutputKeeper = new ObservableOutputKeeper(TEST_CATALOG, storageOptions, Mockito.mock(Scheduler.class))) {
+			final InsertionOutput insertionOutput = shouldSerializeAndReconstructOffsetIndex(
+				storageOptions, observableOutputKeeper, EntityBodyStoragePart::new, 0
+			);
+			final OffsetIndex offsetIndex = insertionOutput.fileOffsetIndex();
+			final InsertionOutput insertionOutput2 = createRecordsInFileOffsetIndex(offsetIndex, 100, 0, 1);
+			/* input count records +1 record for the OffsetIndex itself */
+			if (crc32Check == Crc32Check.YES) {
+				assertEquals(
+					/* 100 records, 1 empty header, 1 header with single fragment */
+					100 + computeExpectedRecordCount(storageOptions, 100).fragments() + 1,
+					offsetIndex.verifyContents().getRecordCount()
+				);
+			}
 			assertEquals(
-				/* 100 records, 1 empty header, 1 header with single fragment */
-				100 + computeExpectedRecordCount(storageOptions, 100).fragments() + 1,
-				offsetIndex.verifyContents().getRecordCount()
+				100,
+				offsetIndex.count(insertionOutput2.catalogVersion())
 			);
+			IOUtils.closeQuietly(offsetIndex::close);
 		}
-		assertEquals(
-			100,
-			offsetIndex.count(insertionOutput2.catalogVersion())
-		);
-		IOUtils.closeQuietly(offsetIndex::close);
 	}
 
 	@DisplayName("Hundreds entities should be stored in OffsetIndex and retrieved intact.")
@@ -922,19 +931,21 @@ private InsertionOutput serializeAndReconstructBigFileOffsetIndex(
 		@Nonnull StorageOptions storageOptions,
 		@Nonnull IntFunction<EntityBodyStoragePart> bodyPartFactory
 	) {
-		return shouldSerializeAndReconstructOffsetIndex(storageOptions, bodyPartFactory, 600);
+		try (final ObservableOutputKeeper observableOutputKeeper = new ObservableOutputKeeper(TEST_CATALOG, storageOptions, Mockito.mock(Scheduler.class))) {
+			return shouldSerializeAndReconstructOffsetIndex(storageOptions, observableOutputKeeper, bodyPartFactory, 600);
+		}
 	}
 
 	@Nonnull
 	private InsertionOutput shouldSerializeAndReconstructOffsetIndex(
 		@Nonnull StorageOptions storageOptions,
+		@Nonnull ObservableOutputKeeper observableOutputKeeper,
 		@Nonnull IntFunction<EntityBodyStoragePart> bodyPartFactory,
 		int recordCount
 	) {
 		OffsetIndex loadedFileOffsetIndex = null;
 		try (
-			final ObservableOutputKeeper observableOutputKeeper = new ObservableOutputKeeper(TEST_CATALOG, storageOptions, Mockito.mock(Scheduler.class));
-			final WriteOnlyFileHandle writeHandle = new WriteOnlyFileHandle(this.targetFile, storageOptions, observableOutputKeeper);
+			final WriteOnlyFileHandle writeHandle = new WriteOnlyFileHandle(this.targetFile, storageOptions, observableOutputKeeper)
 		) {
 			final OffsetIndex fileOffsetIndex = new OffsetIndex(
 				0L,