Merge branch 'feature/sanitizer' into release/2.1.0

Author: overheadhunter

src/main/java/org/cryptomator/cryptofs/health/api/AbstractHealthCheck.java

@@ -3,17 +3,25 @@
 import org.cryptomator.cryptofs.VaultConfig;
 import org.cryptomator.cryptolib.api.Cryptor;
 import org.cryptomator.cryptolib.api.Masterkey;
+import org.slf4j.LoggerFactory;
 
 import java.nio.file.Path;
 import java.util.Collection;
 import java.util.ServiceLoader;
 import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.ForkJoinPool;
+import java.util.function.Consumer;
 import java.util.stream.Stream;
+import java.util.stream.StreamSupport;
 
 public interface HealthCheck {
 
+	/**
+	 * @return All known health checks
+	 */
+	static Collection<HealthCheck> allChecks() {
+		return ServiceLoader.load(HealthCheck.class).stream().map(ServiceLoader.Provider::get).toList();
+	}
+
 	/**
 	 * @return A unique name for this check (that might be used as a translation key)
 	 */
@@ -24,24 +32,42 @@ default String identifier() {
 	/**
 	 * Checks the vault at the given path.
 	 *
-	 * @param pathToVault Path to the vault's root directory
-	 * @param config The parsed and verified vault config
-	 * @param masterkey The masterkey
-	 * @param cryptor A cryptor initialized for this vault
-	 * @param executor An executor service to run the health check
-	 * @return Diagnostic results
+	 * @param pathToVault     Path to the vault's root directory
+	 * @param config          The parsed and verified vault config
+	 * @param masterkey       The masterkey
+	 * @param cryptor         A cryptor initialized for this vault
+	 * @param resultCollector Callback called for each result.
 	 */
-	Stream<DiagnosticResult> check(Path pathToVault, VaultConfig config, Masterkey masterkey, Cryptor cryptor, ExecutorService executor);
+	void check(Path pathToVault, VaultConfig config, Masterkey masterkey, Cryptor cryptor, Consumer<DiagnosticResult> resultCollector);
 
 	/**
-	 * Attempts to cancel this health check (if it is running).
+	 * Invokes the health check on a background thread scheduled using the given executor service. The results will be
+	 * streamed. If the stream gets {@link Stream#close() closed} before it terminates, an attempt is made to cancel
+	 * the health check.
+	 * <p>
+	 * The check blocks if the stream is not consumed
 	 *
-	 * Calling this method does not guarantee that no further results are produced due to async behaviour.
+	 * @param pathToVault Path to the vault's root directory
+	 * @param config      The parsed and verified vault config
+	 * @param masterkey   The masterkey
+	 * @param cryptor     A cryptor initialized for this vault
+	 * @param executor    An executor service to run the health check
+	 * @return A lazily filled stream of diagnostic results.
 	 */
-	void cancel();
+	default Stream<DiagnosticResult> check(Path pathToVault, VaultConfig config, Masterkey masterkey, Cryptor cryptor, ExecutorService executor) {
+		var resultSpliterator = new TransferSpliterator<DiagnosticResult>(new PoisonResult());
 
-	static Collection<HealthCheck> allChecks() {
-		return ServiceLoader.load(HealthCheck.class).stream().map(ServiceLoader.Provider::get).toList();
+		var task = executor.submit(() -> {
+			try {
+				check(pathToVault, config, masterkey, cryptor, resultSpliterator);
+			} catch (TransferSpliterator.TransferClosedException e) {
+				LoggerFactory.getLogger(HealthCheck.class).debug("{} cancelled.", identifier());
+			} finally {
+				resultSpliterator.close();
+			}
+		});
+
+		return StreamSupport.stream(resultSpliterator, false).onClose(() -> task.cancel(true));
 	}
 
-}
+}

src/main/java/org/cryptomator/cryptofs/health/api/HealthCheck.java

@@ -0,0 +1,8 @@
+package org.cryptomator.cryptofs.health.api;
+
+record PoisonResult() implements DiagnosticResult {
+	@Override
+	public Severity getServerity() {
+		return null;
+	}
+}

src/main/java/org/cryptomator/cryptofs/health/api/PoisonResult.java

@@ -0,0 +1,82 @@
+package org.cryptomator.cryptofs.health.api;
+
+import com.google.common.base.Preconditions;
+
+import java.util.Objects;
+import java.util.Spliterators;
+import java.util.concurrent.LinkedTransferQueue;
+import java.util.concurrent.TransferQueue;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Consumer;
+
+/**
+ * A concurrent spliterator that only {@link java.util.Spliterator#tryAdvance(Consumer) advances} by transferring
+ * elements it {@link Consumer#accept(Object) consumes}. Consumption blocks if the spliterator is not advanced.
+ * <p>
+ * Once no futher elements are expected, this spliterator <b>must</b> be {@link #close() closed}, otherwise it'll
+ * wait indefinitely.
+ *
+ * @param <T> the type of elements consumed and returned
+ */
+class TransferSpliterator<T> extends Spliterators.AbstractSpliterator<T> implements Consumer<T>, AutoCloseable {
+
+	private final TransferQueue<T> queue = new LinkedTransferQueue<>();
+	private final AtomicBoolean poisoned = new AtomicBoolean();
+	private final T poison;
+
+	/**
+	 * @param poison A unique value that must be distinct to every single value expected to be transferred.
+	 */
+	public TransferSpliterator(T poison) {
+		super(Long.MAX_VALUE, DISTINCT | NONNULL | IMMUTABLE);
+		this.poison = Objects.requireNonNull(poison);
+	}
+
+	@Override
+	public boolean tryAdvance(Consumer<? super T> action) {
+		try {
+			var element = queue.take();
+			if (element == poison) {
+				return false;
+			} else {
+				action.accept(element);
+				return true;
+			}
+		} catch (InterruptedException e) {
+			Thread.currentThread().interrupt();
+			return false;
+		}
+	}
+
+	/**
+	 * Transfers the value to consuming thread. Blocks until transfer is complete or thread is interrupted.
+	 * @param value The value to transfer
+	 * @throws TransferClosedException If the transfer has been closed or this thread is interrupted while waiting for the consuming side.
+	 */
+	@Override
+	public void accept(T value) throws TransferClosedException {
+		Preconditions.checkArgument(value != poison, "must not feed poison");
+		if (poisoned.get()) {
+			throw new TransferClosedException();
+		}
+		try {
+			queue.transfer(value);
+		} catch (InterruptedException e) {
+			Thread.currentThread().interrupt();
+			throw new TransferClosedException();
+		}
+	}
+
+	@Override
+	public void close() {
+		poisoned.set(true);
+		queue.offer(poison);
+	}
+
+	/**
+	 * Thrown if an attempt is made to {@link #accept(Object) transfer} further elements after the TransferSpliterator
+	 * has been closed.
+	 */
+	public static class TransferClosedException extends IllegalStateException {}
+
+}

src/main/java/org/cryptomator/cryptofs/health/api/TransferSpliterator.java

@@ -2,7 +2,6 @@
 
 import org.cryptomator.cryptofs.VaultConfig;
 import org.cryptomator.cryptofs.common.Constants;
-import org.cryptomator.cryptofs.health.api.AbstractHealthCheck;
 import org.cryptomator.cryptofs.health.api.CheckFailed;
 import org.cryptomator.cryptofs.health.api.DiagnosticResult;
 import org.cryptomator.cryptofs.health.api.HealthCheck;
@@ -18,30 +17,22 @@
 import java.nio.file.Path;
 import java.nio.file.SimpleFileVisitor;
 import java.nio.file.attribute.BasicFileAttributes;
-import java.util.ArrayList;
-import java.util.Collection;
 import java.util.HashMap;
 import java.util.HashSet;
-import java.util.List;
 import java.util.Map;
-import java.util.Queue;
 import java.util.Set;
-import java.util.Spliterator;
-import java.util.Spliterators;
 import java.util.function.Consumer;
-import java.util.stream.Stream;
-import java.util.stream.StreamSupport;
 
 /**
  * Reads all dir.c9r files and checks if the corresponding dir exists.
  */
-public class DirIdCheck extends AbstractHealthCheck {
+public class DirIdCheck implements HealthCheck {
 
 	private static final Logger LOG = LoggerFactory.getLogger(DirIdCheck.class);
 	private static final int MAX_TRAVERSAL_DEPTH = 4; // d/2/30/Fo0==.c9r/dir.c9r
 
 	@Override
-	protected void check(Path pathToVault, VaultConfig config, Masterkey masterkey, Cryptor cryptor, Consumer<DiagnosticResult> resultCollector) {
+	public void check(Path pathToVault, VaultConfig config, Masterkey masterkey, Cryptor cryptor, Consumer<DiagnosticResult> resultCollector) {
 		// scan vault structure:
 		var dataDirPath = pathToVault.resolve(Constants.DATA_DIR_NAME);
 		var dirVisitor = new DirVisitor(dataDirPath, resultCollector);