In FAIL_FAST, try opening change stream cursor.

I expected it to be a common mistake for people to forget to use a
replica set. In FAIL_FAST mode, it seems like that ought to reliably
throw an exception before the Bosk constructor returns.

Still won't help much as long as the default is DISCONNECT instead of
FAIL_FAST though.

Author: prdoyle

bosk-logback/src/main/java/works/bosk/logback/BoskLogFilter.java

@@ -6,6 +6,7 @@
 import ch.qos.logback.core.spi.FilterReply;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.function.Function;
 import java.util.stream.Stream;
 import org.slf4j.LoggerFactory;
 import org.slf4j.MDC;
@@ -61,7 +62,11 @@ public static final class LogController {
 		// We'd like to use SLF4J's "Level" but that doesn't support OFF
 		public void setLogging(Level level, Class<?>... loggers) {
 			// Put them all in one atomic operation
-			overrides.putAll(Stream.of(loggers).collect(toMap(Class::getName, c->level)));
+			overrides.putAll(Stream.of(loggers).collect(toMap(Class::getName, _->level)));
+		}
+
+		public void setLogging(Level level, String... loggers) {
+			overrides.putAll(Stream.of(loggers).collect(toMap(Function.identity(),_->level)));
 		}
 	}
 

bosk-mongo/README.md

@@ -5,4 +5,13 @@ containing `MongoDriver` and the associated machinery.
 This library is documented in the [User's Guide](../docs/USERS.md);
 see also the [unit tests](src/test) for usage examples.
 
+Add [MongoDriver](src/main/java/works/bosk/mongo/MongoDriver.java) to your
+driver stack for persistence and replication.
+During development, we recommend using `InitialDatabaseUnavailableMode.FAIL_FAST`
+to get helpful possible errors when the database is misconfigured or unavailable;
+in production, you should use the default `DISCONNECT` mode for better fault tolerance.
+
+Bosk uses change streams, so it requires MongoDB to be configured as a replica set.
+Our unit tests demonstrate how to do this with TestContainers.
+
 See the [javadocs](https://javadoc.io/doc/works.bosk/bosk-mongo/latest/works.bosk.mongo/module-summary.html) for more information.

bosk-mongo/src/main/java/works/bosk/drivers/mongo/MongoDriverSettings.java

@@ -3,6 +3,7 @@
 import lombok.Builder;
 import lombok.Builder.Default;
 import lombok.Value;
+import works.bosk.Bosk;
 import works.bosk.BoskDriver;
 
 import static works.bosk.drivers.mongo.MongoDriverSettings.DatabaseFormat.SEQUOIA;
@@ -39,6 +40,12 @@ public class MongoDriverSettings {
 	 * @see PandoFormat
 	 */
 	@Default DatabaseFormat preferredDatabaseFormat = SEQUOIA;
+
+	/**
+	 * Default is {@link InitialDatabaseUnavailableMode#DISCONNECT DISCONNECT}
+	 * because it simplifies production deployments and repairs,
+	 * but these very fault-tolerance features can be confusing during development.
+	 */
 	@Default InitialDatabaseUnavailableMode initialDatabaseUnavailableMode = InitialDatabaseUnavailableMode.DISCONNECT;
 
 	@Default Experimental experimental = Experimental.builder().build();
@@ -102,7 +109,7 @@ public enum InitialDatabaseUnavailableMode {
 
 		/**
 		 * If the database state can't be loaded during {@link BoskDriver#initialRoot},
-		 * throw an exception.
+		 * throw an exception from the {@link Bosk#Bosk Bosk constructor} call.
 		 * This is probably the desired behaviour during development,
 		 * but in production, it creates a boot sequencing dependency between the application and the database.
 		 */

bosk-mongo/src/main/java/works/bosk/drivers/mongo/exceptions/InitialCursorCommandException.java

@@ -0,0 +1,15 @@
+package works.bosk.drivers.mongo.exceptions;
+
+public class InitialCursorCommandException extends InitialRootFailureException {
+	public InitialCursorCommandException(String message) {
+		super(message);
+	}
+
+	public InitialCursorCommandException(String message, Throwable cause) {
+		super(message, cause);
+	}
+
+	public InitialCursorCommandException(Throwable cause) {
+		super(cause);
+	}
+}

bosk-mongo/src/main/java/works/bosk/drivers/mongo/exceptions/InitialCursorTimeoutException.java

@@ -0,0 +1,15 @@
+package works.bosk.drivers.mongo.exceptions;
+
+public class InitialCursorTimeoutException extends InitialRootFailureException {
+	public InitialCursorTimeoutException(String message) {
+		super(message);
+	}
+
+	public InitialCursorTimeoutException(String message, Throwable cause) {
+		super(message, cause);
+	}
+
+	public InitialCursorTimeoutException(Throwable cause) {
+		super(cause);
+	}
+}

bosk-mongo/src/main/java/works/bosk/drivers/mongo/internal/ChangeReceiver.java

@@ -1,6 +1,9 @@
 package works.bosk.drivers.mongo.internal;
 
+import com.mongodb.MongoCommandException;
+import com.mongodb.MongoException;
 import com.mongodb.MongoInterruptedException;
+import com.mongodb.MongoOperationTimeoutException;
 import com.mongodb.client.MongoChangeStreamCursor;
 import com.mongodb.client.MongoCollection;
 import com.mongodb.client.model.changestream.ChangeStreamDocument;
@@ -18,10 +21,14 @@
 import works.bosk.Identifier;
 import works.bosk.drivers.mongo.MongoDriverSettings;
 import works.bosk.drivers.mongo.exceptions.DisconnectedException;
+import works.bosk.drivers.mongo.exceptions.InitialCursorCommandException;
+import works.bosk.drivers.mongo.exceptions.InitialRootFailureException;
+import works.bosk.drivers.mongo.exceptions.InitialCursorTimeoutException;
 import works.bosk.logging.MdcKeys;
 
 import static java.lang.Thread.currentThread;
 import static java.util.concurrent.TimeUnit.MILLISECONDS;
+import static works.bosk.drivers.mongo.MongoDriverSettings.InitialDatabaseUnavailableMode.FAIL_FAST;
 import static works.bosk.logging.MappedDiagnosticContext.MDCScope;
 import static works.bosk.logging.MappedDiagnosticContext.setupMDC;
 
@@ -58,6 +65,11 @@ class ChangeReceiver implements Closeable {
 		this.settings = settings;
 		this.collection = collection;
 		this.creationPoint = new Exception("Additional context: ChangeReceiver creation stack trace:");
+		if (settings.initialDatabaseUnavailableMode() == FAIL_FAST) {
+			// User requested fail-fast behaviour; try to open the cursor right away
+			// to ensure the database is set up for change streams.
+			probeChangeStreamCursor();
+		}
 		ex.scheduleWithFixedDelay(
 			this::connectionLoop,
 			0,
@@ -66,6 +78,18 @@ class ChangeReceiver implements Closeable {
 		);
 	}
 
+	private void probeChangeStreamCursor() {
+		try (var _ = openCursor()) {
+			LOGGER.debug("Successfully opened MongoDB cursor");
+		} catch (MongoOperationTimeoutException e) {
+			throw new InitialCursorTimeoutException("Timed out attempting to open MongoDB cursor; check database connectivity", e);
+		} catch (MongoCommandException e) {
+			throw new InitialCursorCommandException("Failed to open change stream cursor; ensure MongoDB server configured as a replica set", e);
+		} catch (MongoException e) {
+			throw new InitialRootFailureException("Unexpected failure opening MongoDB cursor", e);
+		}
+	}
+
 	@Override
 	public void close() {
 		isClosed = true;
@@ -193,6 +217,7 @@ private void disconnect(String description, String remedy, Throwable e) {
 	private void addContextToException(Throwable x) {
 		x.addSuppressed(creationPoint);
 	}
+
 	private MongoChangeStreamCursor<ChangeStreamDocument<BsonDocument>> openCursor() {
 		MongoChangeStreamCursor<ChangeStreamDocument<BsonDocument>> result = collection
 			.watch()

bosk-mongo/src/test/java/works/bosk/drivers/mongo/internal/MongoService.java

@@ -45,6 +45,8 @@ public class MongoService implements Closeable {
 	// We do logging in some static initializers, so this needs to be initialized first
 	private static final Logger LOGGER = LoggerFactory.getLogger(MongoService.class);
 
+	public static final DockerImageName MONGODB_IMAGE_NAME = DockerImageName.parse("mongo:8.0");
+
 	private final MongoClient mongoClient = MongoClients.create(normalClientSettings);
 
 	// Expensive stuff shared among instances as much as possible, hence static
@@ -100,7 +102,7 @@ public void close() {
 	}
 
 	private static MongoDBContainer mongoContainer() {
-		MongoDBContainer container = new MongoDBContainer(DockerImageName.parse("mongo:8.0"))
+		MongoDBContainer container = new MongoDBContainer(MONGODB_IMAGE_NAME)
 			.withReplicaSet()
 			.withTmpFs(Map.of("/data/db", "rw"))
 			.withNetwork(NETWORK);

bosk-mongo/src/test/java/works/bosk/drivers/mongo/internal/ServerMisconfigurationTest.java

@@ -0,0 +1,84 @@
+package works.bosk.drivers.mongo.internal;
+
+import com.mongodb.MongoClientSettings;
+import com.mongodb.ServerAddress;
+import java.io.IOException;
+import java.net.ServerSocket;
+import org.junit.jupiter.api.Test;
+import org.testcontainers.mongodb.MongoDBContainer;
+import works.bosk.Bosk;
+import works.bosk.BoskConfig;
+import works.bosk.DriverStack;
+import works.bosk.drivers.mongo.BsonSerializer;
+import works.bosk.drivers.mongo.MongoDriver;
+import works.bosk.drivers.mongo.MongoDriverSettings;
+import works.bosk.drivers.mongo.exceptions.InitialCursorCommandException;
+import works.bosk.drivers.mongo.exceptions.InitialCursorTimeoutException;
+import works.bosk.logback.BoskLogFilter;
+import works.bosk.testing.drivers.state.TestEntity;
+
+import static ch.qos.logback.classic.Level.ERROR;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static works.bosk.drivers.mongo.MongoDriverSettings.InitialDatabaseUnavailableMode.FAIL_FAST;
+import static works.bosk.drivers.mongo.internal.MongoService.MONGODB_IMAGE_NAME;
+
+public class ServerMisconfigurationTest {
+
+	@Test
+	void unreachable() throws IOException {
+		int port;
+		try (var socket = new ServerSocket(0)) {
+			port = socket.getLocalPort();
+		}
+		MongoClientSettings clientSettings = MongoService.mongoClientSettings(
+			new ServerAddress("localhost", port)
+		);
+		assertThrows(InitialCursorTimeoutException.class, () -> createBosk(clientSettings));
+	}
+
+	/**
+	 * This is an important case: it doesn't fail "naturally" because
+	 * a standalone MongoDB server will accept connections and return the initial state,
+	 * but will then fail when attempting to open a change stream.
+	 * Hence, special effort is required to detect this case and pass this test.
+	 */
+	@Test
+	void notAReplicaSet() {
+		try (var mongo = new MongoDBContainer(MONGODB_IMAGE_NAME)) {
+			mongo.start();
+			MongoClientSettings clientSettings = MongoService.mongoClientSettings(
+				new ServerAddress(
+					mongo.getHost(),
+					mongo.getFirstMappedPort()
+				)
+			);
+			assertThrows(InitialCursorCommandException.class, () -> createBosk(clientSettings));
+		}
+	}
+
+	private Bosk<TestEntity> createBosk(MongoClientSettings clientSettings) {
+		// We're expecting some timeout warnings from the Mongo driver;
+		BoskLogFilter.LogController logController = new BoskLogFilter.LogController();
+		logController.setLogging(ERROR, "com.mongodb");
+
+		return new Bosk<>(
+			ServerMisconfigurationTest.class.getSimpleName(),
+			TestEntity.class,
+			AbstractMongoDriverTest::initialRootWithEmptyCatalog,
+			BoskConfig.<TestEntity>builder()
+				.driverFactory(DriverStack.of(
+					BoskLogFilter.withController(logController),
+					MongoDriver.factory(
+						clientSettings,
+						MongoDriverSettings.builder()
+							.database("bosk_" + getClass().getSimpleName())
+							.timescaleMS(100)
+							.initialDatabaseUnavailableMode(FAIL_FAST)
+							.build(),
+						new BsonSerializer()
+					)
+				))
+				.build()
+		);
+	}
+}