MongoDriverSettings.timescaleMS.

Combine recoveryPollingMS and flushTimeoutMS into a single "timescale" setting.

Tuning a multitude of individual timeouts is not a reasonable thing to
ask users to do. Just ask them about how responsive they want the system
to be, and then we can take the responsibility of balancing the actual
individual timeout settings in a way that makes sense given how all the
parts of this rather complex driver work together.

This is something I already did in the SQL driver. I might end up
copying over the patienceFactor idea too.

Author: prdoyle

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

@@ -59,7 +59,7 @@ class ChangeReceiver implements Closeable {
 		ex.scheduleWithFixedDelay(
 			this::connectionLoop,
 			0,
-			settings.recoveryPollingMS(),
+			settings.timescaleMS(),
 			MILLISECONDS
 		);
 	}
@@ -73,7 +73,7 @@ public void close() {
 
 	/**
 	 * This method has a loop to do immediate reconnections and skip the
-	 * {@link MongoDriverSettings#recoveryPollingMS() recoveryPollingMS} delay,
+	 * {@link MongoDriverSettings#timescaleMS()} delay,
 	 * but besides that, exiting this method has the same effect as continuing
 	 * around the loop.
 	 */
@@ -200,7 +200,7 @@ private void addContextToException(Throwable x) {
 	private MongoChangeStreamCursor<ChangeStreamDocument<BsonDocument>> openCursor() {
 		MongoChangeStreamCursor<ChangeStreamDocument<BsonDocument>> result = collection
 			.watch()
-			.maxAwaitTime(settings.recoveryPollingMS(), MILLISECONDS)
+			.maxAwaitTime(settings.timescaleMS(), MILLISECONDS)
 			.cursor();
 		LOGGER.debug("Cursor is open");
 		return result;

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

@@ -49,7 +49,7 @@
  * that logic to include an epoch concept without touching other components.
  */
 class FlushLock implements Closeable {
-	private final MongoDriverSettings settings;
+	private final long flushTimeoutMS;
 	private final Lock queueLock = new ReentrantLock();
 	private final PriorityBlockingQueue<Waiter> queue = new PriorityBlockingQueue<>();
 	private volatile long alreadySeen;
@@ -60,9 +60,9 @@ class FlushLock implements Closeable {
 	 * too old, and we'll wait forever for intervening revisions that have already happened;
 	 * too new, and we'll proceed immediately without waiting for revisions that haven't happened yet.
 	 */
-	public FlushLock(MongoDriverSettings settings, long revisionAlreadySeen) {
+	public FlushLock(long revisionAlreadySeen, long flushTimeoutMS) {
 		LOGGER.debug("New flush lock at revision {} [{}]", revisionAlreadySeen, identityHashCode(this));
-		this.settings = settings;
+		this.flushTimeoutMS = flushTimeoutMS;
 		this.alreadySeen = revisionAlreadySeen;
 	}
 
@@ -92,7 +92,7 @@ void awaitRevision(BsonInt64 revision) throws InterruptedException, FlushFailure
 		}
 		if (revisionValue > past) {
 			LOGGER.debug("Awaiting revision {} > {} [{}]", revisionValue, past, identityHashCode(this));
-			if (!semaphore.tryAcquire(settings.flushTimeoutMS(), MILLISECONDS)) {
+			if (!semaphore.tryAcquire(flushTimeoutMS, MILLISECONDS)) {
 				throw new FlushFailureException("Timed out waiting for revision " + revisionValue + " > " + alreadySeen);
 			}
 			if (isClosed) {

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

@@ -30,22 +30,22 @@
 import works.bosk.Identifier;
 import works.bosk.Reference;
 import works.bosk.StateTreeNode;
-import works.bosk.logging.MappedDiagnosticContext.MDCScope;
 import works.bosk.drivers.mongo.MongoDriverSettings.DatabaseFormat;
 import works.bosk.drivers.mongo.MongoDriverSettings.InitialDatabaseUnavailableMode;
 import works.bosk.drivers.mongo.bson.BsonFormatter.DocumentFields;
 import works.bosk.drivers.mongo.bson.BsonSerializer;
 import works.bosk.drivers.mongo.status.MongoStatus;
 import works.bosk.exceptions.FlushFailureException;
 import works.bosk.exceptions.InvalidTypeException;
+import works.bosk.logging.MappedDiagnosticContext.MDCScope;
 
 import static com.mongodb.MongoException.TRANSIENT_TRANSACTION_ERROR_LABEL;
 import static java.util.concurrent.TimeUnit.MILLISECONDS;
 import static works.bosk.ReferenceUtils.rawClass;
 import static works.bosk.drivers.mongo.Formatter.REVISION_ONE;
 import static works.bosk.drivers.mongo.Formatter.REVISION_ZERO;
-import static works.bosk.logging.MappedDiagnosticContext.setupMDC;
 import static works.bosk.drivers.mongo.MongoDriverSettings.DatabaseFormat.SEQUOIA;
+import static works.bosk.logging.MappedDiagnosticContext.setupMDC;
 
 /**
  * This is the driver returned to the user by {@link MongoDriver#factory}.
@@ -67,6 +67,10 @@ final class MainDriver<R extends StateTreeNode> implements MongoDriver {
 	private final Listener listener;
 	final Formatter formatter;
 
+	final long flushTimeout;
+	final long initialRootTimeout;
+	final long driverReinitializeTimeout;
+
 	/**
 	 * {@link MongoClient#close()} throws if called more than once.
 	 * {@link MongoDriver} is more civilized: subsequent calls do nothing.
@@ -117,6 +121,17 @@ final class MainDriver<R extends StateTreeNode> implements MongoDriver {
 			this.formatter = new Formatter(boskInfo, bsonSerializer);
 			this.receiver = new ChangeReceiver(boskInfo.name(), boskInfo.instanceID(), listener, driverSettings, rawCollection);
 		}
+
+		// Flushes work by waiting for the latest version to arrive on the change stream.
+		// If we wait twice as long as that takes, and we don't see the update, something
+		// has gone wrong.
+		flushTimeout = 2L * driverSettings.timescaleMS();
+
+		// TODO: Justify this calculation.
+		initialRootTimeout = 5L * driverSettings.timescaleMS();
+
+		// TODO: Justify this calculation.
+		driverReinitializeTimeout = 5L * driverSettings.timescaleMS();
 	}
 
 	@Override
@@ -421,7 +436,7 @@ public void onConnectionSucceeded() throws
 		private void runInitialRootAction(FutureTask<R> initialRootAction) throws InterruptedException, TimeoutException, InitialRootActionException {
 			initialRootAction.run();
 			try {
-				initialRootAction.get(5 * driverSettings.recoveryPollingMS(), MILLISECONDS);
+				initialRootAction.get(initialRootTimeout, MILLISECONDS);
 				LOGGER.debug("initialRoot action completed successfully");
 			} catch (ExecutionException e) {
 				LOGGER.debug("initialRoot action failed", e);
@@ -541,7 +556,7 @@ private FormatDriver<R> newFormatDriver(long revisionAlreadySeen, DatabaseFormat
 				collection,
 				driverSettings,
 				bsonSerializer,
-				new FlushLock(driverSettings, revisionAlreadySeen),
+				new FlushLock(revisionAlreadySeen, flushTimeout),
 				downstream);
 		} else if (format instanceof PandoFormat pandoFormat) {
 			return new PandoFormatDriver<>(
@@ -550,7 +565,7 @@ private FormatDriver<R> newFormatDriver(long revisionAlreadySeen, DatabaseFormat
 				driverSettings,
 				pandoFormat,
 				bsonSerializer,
-				new FlushLock(driverSettings, revisionAlreadySeen),
+				new FlushLock(revisionAlreadySeen, flushTimeout),
 				downstream);
 		}
 		throw new IllegalArgumentException("Unexpected database format: " + format);
@@ -621,9 +636,8 @@ private <X extends Exception, Y extends Exception> void doRetryableDriverOperati
 	private <X extends Exception, Y extends Exception> void waitAndRetry(RetryableOperation<X, Y> operation, String description, Object... args) throws X, Y {
 		try {
 			formatDriverLock.lock();
-			long waitTimeMS = 5 * driverSettings.recoveryPollingMS();
-			LOGGER.debug("Waiting for new FormatDriver for {} ms", waitTimeMS);
-			boolean success = formatDriverChanged.await(waitTimeMS, MILLISECONDS);
+			LOGGER.debug("Waiting for new FormatDriver for {} ms", driverReinitializeTimeout);
+			boolean success = formatDriverChanged.await(driverReinitializeTimeout, MILLISECONDS);
 			if (!success) {
 				LOGGER.warn("Timed out waiting for MongoDB to recover; will retry anyway, but the operation may fail");
 			}

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

@@ -10,6 +10,8 @@
 import works.bosk.drivers.mongo.bson.BsonSerializer;
 import works.bosk.drivers.mongo.status.MongoStatus;
 
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
+
 /**
  * A {@link BoskDriver} that maintains the bosk state in a MongoDB database.
  * Multiple bosks, potentially in multiple separate processes,
@@ -71,7 +73,19 @@ static <RR extends StateTreeNode> MongoDriverFactory<RR> factory(
 		BsonSerializer bsonSerializer
 	) {
 		driverSettings.validate();
-		return (b, d) -> new MainDriver<>(b, clientSettings, driverSettings, bsonSerializer, d);
+		var actualClientSettings = overrideTimeouts(clientSettings, driverSettings);
+		return (b, d) -> new MainDriver<>(b, actualClientSettings, driverSettings, bsonSerializer, d);
+	}
+
+	/**
+	 * Timeouts from the driver settings take precedence over those from {@code clientSettings}.
+	 * Without this, the driver settings would be ineffective, and we'd end up waiting
+	 * longer than desired for the Mongo client operations.
+	 */
+	static MongoClientSettings overrideTimeouts(MongoClientSettings clientSettings, MongoDriverSettings driverSettings) {
+		return MongoClientSettings.builder(clientSettings)
+			.timeout(2L * driverSettings.timescaleMS(), MILLISECONDS)
+			.build();
 	}
 
 	interface MongoDriverFactory<RR extends StateTreeNode> extends DriverFactory<RR> {

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

@@ -13,8 +13,8 @@
 public class MongoDriverSettings {
 	String database;
 
-	@Default long flushTimeoutMS = 30_000;
-	@Default long recoveryPollingMS = 30_000;
+	@Default int timescaleMS = 30_000;
+
 	/**
 	 * @see DatabaseFormat#SEQUOIA
 	 * @see PandoFormat

bosk-mongo/src/test/java/works/bosk/drivers/mongo/MongoDriverRecoveryTest.java

@@ -54,16 +54,28 @@ static Stream<TestParameters.ParameterSet> parameters() {
 		return TestParameters.driverSettings(
 			Stream.of(
 				MongoDriverSettings.DatabaseFormat.SEQUOIA,
+				PandoFormat.oneBigDocument(),
 				PandoFormat.withGraftPoints("/catalog", "/sideTable")
 			),
 			Stream.of(TestParameters.EventTiming.NORMAL)
 		).map(b -> b.applyDriverSettings(s -> s
-			.recoveryPollingMS(1500) // Note that some tests can take as long as 10x this
-			.flushTimeoutMS(2000) // A little more than recoveryPollingMS
+			.timescaleMS(100) // Note that some tests can take as long as 25x this
 		));
 	}
 
-	enum FlushOrWait { FLUSH, WAIT }
+	enum FlushOrWait {
+		FLUSH,
+
+		/**
+		 * Technically, these tests should be using {@link BoskDriver#flush()},
+		 * but we also want to exhibit some "liveness" so that users who don't
+		 * call {@code flush} eventually see updates anyway.
+		 * <p>
+		 * This test mode inserts a delay instead of {@code flush} to ensure
+		 * updates eventually arrive.
+		 */
+		WAIT,
+	}
 
 	@SuppressWarnings("unused")
 	static Stream<FlushOrWait> flushOrWait() {
@@ -127,7 +139,11 @@ private void waitFor(BoskDriver driver) throws IOException, InterruptedException
 				driver.flush();
 				break;
 			case WAIT:
-				Thread.sleep(2 * driverSettings.recoveryPollingMS());
+				// The user really has no business expecting updates to occur promptly.
+				// Let's wait several times the timescale so that the test
+				// can set a short timescale to make FLUSH fast without risking
+				// failures in the WAIT tests.
+				Thread.sleep(5L * driverSettings.timescaleMS());
 				break;
 		}
 	}
@@ -274,7 +290,7 @@ private void testRecovery(Runnable disruptiveAction, Function<TestEntity, TestEn
 		LOGGER.debug("Setup database to beforeState");
 		TestEntity beforeState = initializeDatabase("before disruption");
 
-		Bosk<TestEntity> bosk = new Bosk<TestEntity>(boskName(getClass().getSimpleName()), TestEntity.class, this::initialRoot, driverFactory, Bosk.simpleRegistrar());
+		Bosk<TestEntity> bosk = new Bosk<>(boskName(getClass().getSimpleName()), TestEntity.class, this::initialRoot, driverFactory, Bosk.simpleRegistrar());
 
 		try (var _ = bosk.readContext()) {
 			assertEquals(beforeState, bosk.rootReference().value());

bosk-mongo/src/test/java/works/bosk/drivers/mongo/MongoDriverSpecialTest.java

@@ -69,11 +69,13 @@ static Stream<TestParameters.ParameterSet> parameters() {
 		return TestParameters.driverSettings(
 			Stream.of(
 				MongoDriverSettings.DatabaseFormat.SEQUOIA,
-//				PandoFormat.oneBigDocument(),
+				PandoFormat.oneBigDocument(),
 				PandoFormat.withGraftPoints("/catalog", "/sideTable")
 			),
 			Stream.of(TestParameters.EventTiming.NORMAL)
-		);
+		).map(b -> b.applyDriverSettings(s -> s
+			.timescaleMS(100) // Note that some tests can take as long as 25x this
+		));
 	}
 
 	@ParametersByName
@@ -132,8 +134,9 @@ public <T> void submitReplacement(Reference<T> target, T newValue) {
 		Reference<ListingEntry> ref = listingRef.then(entity123);
 		bosk.driver().submitReplacement(ref, LISTING_ENTRY);
 
-		// Give the driver a bit of time to make a mistake, if it's going to
-		long budgetMillis = 2000;
+		// Give the driver a bit of time to make a mistake, if it's going to,
+		// but not so long that we cause a timeout that wouldn't otherwise happen
+		long budgetMillis = (1+driverSettings.timescaleMS()) / 2;
 		while (budgetMillis > 0) {
 			long startTime = currentTimeMillis();
 			Reference<?> updatedRef = replacementsSeen.poll(budgetMillis, MILLISECONDS);

bosk-mongo/src/test/java/works/bosk/drivers/mongo/TestParameters.java

@@ -25,8 +25,7 @@ public static ParameterSet from(MongoDriverSettings.DatabaseFormat format, Event
 				timing + "," + format,
 				MongoDriverSettings.builder()
 					.preferredDatabaseFormat(format)
-					.recoveryPollingMS(3000) // Note that some tests can take as long as 10x this
-					.flushTimeoutMS(4000) // A little more than recoveryPollingMS
+					.timescaleMS(90_000) // Tests that actually exercise timeouts should use a much shorter value
 					.testing(MongoDriverSettings.Testing.builder()
 						.eventDelayMS(timing.eventDelayMS)
 						.build())