Merge pull request boskworks#137 from prdoyle/mongo-timeouts

MongoDriver timescale setting

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,27 @@
 public class MongoDriverSettings {
 	String database;
 
-	@Default long flushTimeoutMS = 30_000;
-	@Default long recoveryPollingMS = 30_000;
+	/**
+	 * The general responsiveness of the system. Changes to the database will be
+	 * "noticed" in about this many milliseconds, and other time-related behaviours
+	 * will be scaled accordingly. Lower values are more responsive, while higher
+	 * values are more efficient because they'll do less polling.
+	 * <p>
+	 * For test cases,
+	 *
+	 * <ul>
+	 *     <li>
+	 *         If you are exercising timeout conditions, set this to a low value
+	 *         (say, 1/10 of its default) to make your tests run quickly;
+	 *     </li>
+	 *     <li>
+	 *         otherwise, set it to a high value (say, 10x its default)
+	 *         to avoid spurious test failures.
+	 *     </li>
+	 * </ul>
+	 */
+	@Default int timescaleMS = 1000;
+
 	/**
 	 * @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())