Merge pull request #990 from FgForrest/989-obsoletefilemaintainer-is-closed-during-catalog-replacement

fix: ObsoleteFileMaintainer is closed during catalog replacement

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,