From add08864a9a1c7883a14ee7b7437d37850126465 Mon Sep 17 00:00:00 2001 From: James Rosewell Date: Tue, 10 Nov 2026 20:28:29 +0000 Subject: [PATCH 1/7] Test that polling carries on after a refresh from a watched file A new data file found by the file watcher refreshes the engine, and then polling for updates must carry on, as the newFileAvailable steps of the data update specification say. The test runs the watcher's task itself and records each polling timer the service schedules. After the refresh there must be a second timer, held by the data file. --- .../DataUpdateServiceWatchedFileTests.java | 168 ++++++++++++++++++ 1 file changed, 168 insertions(+) create mode 100644 pipeline/pipeline.engines/src/test/java/fiftyone/pipeline/engines/services/DataUpdateServiceWatchedFileTests.java diff --git a/pipeline/pipeline.engines/src/test/java/fiftyone/pipeline/engines/services/DataUpdateServiceWatchedFileTests.java b/pipeline/pipeline.engines/src/test/java/fiftyone/pipeline/engines/services/DataUpdateServiceWatchedFileTests.java new file mode 100644 index 00000000..b353e9a7 --- /dev/null +++ b/pipeline/pipeline.engines/src/test/java/fiftyone/pipeline/engines/services/DataUpdateServiceWatchedFileTests.java @@ -0,0 +1,168 @@ +/* ********************************************************************* + * This Original Work is copyright of 51 Degrees Mobile Experts Limited. + * Copyright 2026 51 Degrees Mobile Experts Limited, Davidson House, + * Forbury Square, Reading, Berkshire, United Kingdom RG1 3EU. + * + * This Original Work is licensed under the European Union Public Licence + * (EUPL) v.1.2 and is subject to its terms as set out below. + * + * If a copy of the EUPL was not distributed with this file, You can obtain + * one at https://opensource.org/licenses/EUPL-1.2. + * + * The 'Compatible Licences' set out in the Appendix to the EUPL (as may be + * amended by the European Commission) shall be deemed incompatible for + * the purposes of the Work and the provisions of the compatibility + * clause in Article 5 of the EUPL shall not apply. + * + * If using the Work as, or as part of, a network application, by + * including the attribution notice(s) required under Article 5 of the EUPL + * in the end user terms of the application under an appropriate heading, + * such notice(s) shall fulfill the requirements of that article. + * ********************************************************************* */ + +package fiftyone.pipeline.engines.services; + +import fiftyone.common.wrappers.io.FileWrapperDefault; +import fiftyone.common.wrappers.io.FileWrapperFactory; +import fiftyone.pipeline.engines.configuration.DataFileConfigurationDefault; +import fiftyone.pipeline.engines.data.AspectData; +import fiftyone.pipeline.engines.data.AspectEngineDataFileDefault; +import fiftyone.pipeline.engines.data.AspectPropertyMetaData; +import fiftyone.pipeline.engines.flowelements.OnPremiseAspectEngine; +import fiftyone.pipeline.engines.services.update.FutureFactory; +import org.junit.Test; +import org.mockito.invocation.InvocationOnMock; +import org.mockito.stubbing.Answer; + +import java.io.File; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.atomic.AtomicReference; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assume.assumeFalse; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.mockingDetails; +import static org.mockito.Mockito.when; + +/** + * A new data file found by the file watcher is loaded, and then polling for + * updates carries on, as the newFileAvailable steps of the data update + * specification say ("resume polling, if any"). + */ +public class DataUpdateServiceWatchedFileTests { + + private static final long WAIT_FOR_EVENT_MILLIS = 10000; + + private static final long RETRY_MILLIS = 100; + + @Test + public void WatchedFileRefresh_PollingCarriesOn() throws Exception { + assumeFalse("File watchers are not well implemented in OS X, " + + "so don't run this unit test as it is unlikely to pass.", + System.getProperty("os.name").contains("Mac OS X")); + + // The polling timers scheduled, and the watcher's task, which the + // test runs itself. + final List timers = new ArrayList<>(); + final AtomicReference watcher = new AtomicReference<>(); + FutureFactory futureFactory = mock(FutureFactory.class); + when(futureFactory.schedule(any(Runnable.class), anyLong())) + .then(new Answer() { + @Override + public Object answer(InvocationOnMock invocation) { + timers.add((Long) invocation.getArgument(1)); + return mock(ScheduledFuture.class); + } + }); + when(futureFactory.scheduleRepeating(any(Runnable.class), anyLong())) + .then(new Answer() { + @Override + public Object answer(InvocationOnMock invocation) { + watcher.set((Runnable) invocation.getArgument(0)); + return mock(ScheduledFuture.class); + } + }); + FileWrapperFactory fileWrapperFactory = mock(FileWrapperFactory.class); + when(fileWrapperFactory.build(anyString())).then(new Answer() { + @Override + public Object answer(InvocationOnMock invocation) throws Throwable { + return new FileWrapperDefault( + (String) invocation.getArgument(0)); + } + }); + when(fileWrapperFactory.getLastModified(anyString())) + .then(new Answer() { + @Override + public Object answer(InvocationOnMock invocation) { + return new File((String) invocation.getArgument(0)) + .lastModified(); + } + }); + DataUpdateService dataUpdate = new DataUpdateServiceDefault( + null, + mock(HttpClient.class), + fileWrapperFactory, + futureFactory); + + @SuppressWarnings("unchecked") + OnPremiseAspectEngine engine = + mock(OnPremiseAspectEngine.class); + Path directory = Files.createTempDirectory("watched"); + Path dataFile = directory.resolve("data.dat"); + Files.write(dataFile, "Old".getBytes()); + + DataFileConfigurationDefault config = new DataFileConfigurationDefault(); + config.setAutomaticUpdatesEnabled(true); + config.setFileSystemWatcherEnabled(true); + config.setPollingIntervalSeconds(1); + config.setDataFilePath(dataFile.toString()); + AspectEngineDataFileDefault file = new AspectEngineDataFileDefault(); + file.setEngine(engine); + file.setConfiguration(config); + + try { + dataUpdate.registerDataFile(file); + assertEquals("Registering sets one polling timer", + 1, timers.size()); + assertNotNull("Registering sets up the file watcher", + watcher.get()); + + Files.write(dataFile, "New".getBytes()); + long end = System.currentTimeMillis() + WAIT_FOR_EVENT_MILLIS; + while (refreshCount(engine) == 0 && + System.currentTimeMillis() < end) { + watcher.get().run(); + Thread.sleep(RETRY_MILLIS); + } + + assertEquals("The watched file refreshes the engine once", + 1, refreshCount(engine)); + assertEquals("A new polling timer is set after the refresh", + 2, timers.size()); + assertNotNull("The data file holds the new polling timer", + file.getFuture()); + } + finally { + if (config.getWatchKey() != null) { + config.getWatchKey().cancel(); + } + Files.deleteIfExists(dataFile); + Files.deleteIfExists(directory); + } + } + + private static long refreshCount(Object engine) { + return mockingDetails(engine).getInvocations().stream() + .filter(i -> i.getMethod().getName().equals("refreshData")) + .count(); + } +} From 20167c3aad1f67385f6cb9d51a6aaed0c3e90d9e Mon Sep 17 00:00:00 2001 From: James Rosewell Date: Tue, 10 Nov 2026 20:32:33 +0000 Subject: [PATCH 2/7] Register the data file again after a refresh from a watched file updatedFileAvailable cancels the polling timer before it refreshes the engine. checkForUpdate registers the data file again afterwards, which sets a new timer, but dataFileUpdated, which the file watcher calls, did not, so HTTP polling stopped after the first refresh from a watched file. The newFileAvailable steps of the data update specification say to resume polling after loading the file and before the callbacks. dataFileUpdated now registers a registered data file again after the refresh and before the callbacks. Registering a file that is already registered only sets the timer, because the watcher and the start-up check are each set up once. The comment on updatedFileAvailable said it registered the file itself, which it does not, and now says the caller does. --- .../engines/services/DataUpdateServiceDefault.java | 12 +++++++++--- 1 file changed, 9 insertions(+), 3 deletions(-) diff --git a/pipeline/pipeline.engines/src/main/java/fiftyone/pipeline/engines/services/DataUpdateServiceDefault.java b/pipeline/pipeline.engines/src/main/java/fiftyone/pipeline/engines/services/DataUpdateServiceDefault.java index 6b8da4ad..d04da0f4 100644 --- a/pipeline/pipeline.engines/src/main/java/fiftyone/pipeline/engines/services/DataUpdateServiceDefault.java +++ b/pipeline/pipeline.engines/src/main/java/fiftyone/pipeline/engines/services/DataUpdateServiceDefault.java @@ -290,9 +290,10 @@ public void run() { /** * Called when a data update is available and the file at * engine.DataFilePath contains this new data. - * 1. Refresh the engine. - * 2. Dispose of the existing update timer if there is one. - * 3. Re-register the engine with the update service. + * 1. Dispose of the existing update timer if there is one. + * 2. Refresh the engine. + * Where the data file is registered, the caller then registers it again, + * which sets a new update timer. * * @param dataFile the data file to update * @return {@link AutoUpdateStatus#AUTO_UPDATE_SUCCESS} if the data file was @@ -383,6 +384,11 @@ private void dataFileUpdated(Object sender) { aspectDataFile.getDataFilePath(), e); } } + // The refresh cancelled the polling timer, so + // register the file again to set a new one. + if (dataFile.getIsRegistered()) { + registerDataFile(dataFile); + } } } } From 0beb384d9f1946473703329751e83ef01c0671a6 Mon Sep 17 00:00:00 2001 From: James Rosewell Date: Tue, 10 Nov 2026 21:21:23 +0000 Subject: [PATCH 3/7] Leave a data file Windows still holds to be deleted at exit On Windows the full build failed in the watched file test's clean up, with "The process cannot access the file because it is being used by another process". The data update service reads the file through BinaryReader, which maps it, and Windows keeps a mapped file until the buffer is collected. The test's assertions had passed. The clean up now deletes what it can and leaves the rest to be deleted when the test run ends, as the existing watcher test does by ignoring a failed delete. --- .../services/DataUpdateServiceWatchedFileTests.java | 11 +++++++++-- 1 file changed, 9 insertions(+), 2 deletions(-) diff --git a/pipeline/pipeline.engines/src/test/java/fiftyone/pipeline/engines/services/DataUpdateServiceWatchedFileTests.java b/pipeline/pipeline.engines/src/test/java/fiftyone/pipeline/engines/services/DataUpdateServiceWatchedFileTests.java index b353e9a7..7599f7ca 100644 --- a/pipeline/pipeline.engines/src/test/java/fiftyone/pipeline/engines/services/DataUpdateServiceWatchedFileTests.java +++ b/pipeline/pipeline.engines/src/test/java/fiftyone/pipeline/engines/services/DataUpdateServiceWatchedFileTests.java @@ -155,8 +155,15 @@ public Object answer(InvocationOnMock invocation) { if (config.getWatchKey() != null) { config.getWatchKey().cancel(); } - Files.deleteIfExists(dataFile); - Files.deleteIfExists(directory); + // Windows keeps the file until the reader's mapped buffer is + // collected, so whatever cannot be deleted now is deleted when + // the test run ends. Those are deleted in the reverse order they + // are registered, so the folder is registered first. + directory.toFile().deleteOnExit(); + if (dataFile.toFile().delete() == false) { + dataFile.toFile().deleteOnExit(); + } + directory.toFile().delete(); } } From 49057d1bde06dc26924ec6b5ed02384626bc7321 Mon Sep 17 00:00:00 2001 From: James Rosewell Date: Wed, 11 Nov 2026 07:34:38 +0000 Subject: [PATCH 4/7] Test that a watched file that cannot be opened is given up on The data update specification's Errors table says a watched file that cannot be opened for reading within 10 minutes ends the update with AUTO_UPDATE_TIMEOUT and logs nothing, as .NET does. Java tries for ever with no pause, so the new test, which shortens the wait to one second, never sees the update end. This commit adds the AUTO_UPDATE_TIMEOUT status, the ten minute wait and the half second pause as constants, and the field the test shortens, so the test compiles and fails on behaviour. The loop does not use them yet. The setup the two watched file tests share moves into helpers. --- .../engines/services/DataUpdateService.java | 4 +- .../services/DataUpdateServiceDefault.java | 18 ++ .../DataUpdateServiceWatchedFileTests.java | 289 ++++++++++++++---- 3 files changed, 247 insertions(+), 64 deletions(-) diff --git a/pipeline/pipeline.engines/src/main/java/fiftyone/pipeline/engines/services/DataUpdateService.java b/pipeline/pipeline.engines/src/main/java/fiftyone/pipeline/engines/services/DataUpdateService.java index bb262ae7..2ee0231c 100644 --- a/pipeline/pipeline.engines/src/main/java/fiftyone/pipeline/engines/services/DataUpdateService.java +++ b/pipeline/pipeline.engines/src/main/java/fiftyone/pipeline/engines/services/DataUpdateService.java @@ -127,7 +127,9 @@ enum AutoUpdateStatus { // The new data file can't be renamed to replace the previous one. AUTO_UPDATE_NEW_FILE_CANT_RENAME, // Refreshing the engine with the new data caused an error to occur. - AUTO_UPDATE_REFRESH_FAILED + AUTO_UPDATE_REFRESH_FAILED, + // An operation could not be completed within the expected time. + AUTO_UPDATE_TIMEOUT } class DataUpdateCompleteArgs { diff --git a/pipeline/pipeline.engines/src/main/java/fiftyone/pipeline/engines/services/DataUpdateServiceDefault.java b/pipeline/pipeline.engines/src/main/java/fiftyone/pipeline/engines/services/DataUpdateServiceDefault.java index d04da0f4..6ba5361e 100644 --- a/pipeline/pipeline.engines/src/main/java/fiftyone/pipeline/engines/services/DataUpdateServiceDefault.java +++ b/pipeline/pipeline.engines/src/main/java/fiftyone/pipeline/engines/services/DataUpdateServiceDefault.java @@ -72,6 +72,24 @@ public class DataUpdateServiceDefault implements DataUpdateService { */ private static final String LICENSE_KEYS_PARAMETER = "LicenseKeys"; + /** + * How long a watched data file is tried before the update gives up with + * {@link AutoUpdateStatus#AUTO_UPDATE_TIMEOUT}, as a copy over it may + * still be in progress. + */ + static final long WATCHED_FILE_OPEN_TIMEOUT_MILLIS = + TimeUnit.MINUTES.toMillis(10); + + /** + * The pause between tries to open a watched data file. + */ + static final long WATCHED_FILE_OPEN_PAUSE_MILLIS = 500; + + /** + * The wait for a watched data file, which tests shorten. + */ + long watchedFileOpenTimeoutMillis = WATCHED_FILE_OPEN_TIMEOUT_MILLIS; + final List configurations = new ArrayList<>(); private final Logger logger; private final boolean closeFutureFactory; diff --git a/pipeline/pipeline.engines/src/test/java/fiftyone/pipeline/engines/services/DataUpdateServiceWatchedFileTests.java b/pipeline/pipeline.engines/src/test/java/fiftyone/pipeline/engines/services/DataUpdateServiceWatchedFileTests.java index 7599f7ca..f41120da 100644 --- a/pipeline/pipeline.engines/src/test/java/fiftyone/pipeline/engines/services/DataUpdateServiceWatchedFileTests.java +++ b/pipeline/pipeline.engines/src/test/java/fiftyone/pipeline/engines/services/DataUpdateServiceWatchedFileTests.java @@ -22,6 +22,8 @@ package fiftyone.pipeline.engines.services; +import fiftyone.common.wrappers.data.BinaryReader; +import fiftyone.common.wrappers.io.FileWrapper; import fiftyone.common.wrappers.io.FileWrapperDefault; import fiftyone.common.wrappers.io.FileWrapperFactory; import fiftyone.pipeline.engines.configuration.DataFileConfigurationDefault; @@ -29,21 +31,28 @@ import fiftyone.pipeline.engines.data.AspectEngineDataFileDefault; import fiftyone.pipeline.engines.data.AspectPropertyMetaData; import fiftyone.pipeline.engines.flowelements.OnPremiseAspectEngine; +import fiftyone.pipeline.engines.services.DataUpdateService.AutoUpdateStatus; import fiftyone.pipeline.engines.services.update.FutureFactory; import org.junit.Test; -import org.mockito.invocation.InvocationOnMock; -import org.mockito.stubbing.Answer; +import org.mockito.Answers; +import org.slf4j.Logger; import java.io.File; +import java.io.IOException; import java.nio.file.Files; import java.nio.file.Path; import java.util.ArrayList; import java.util.List; +import java.util.concurrent.CountDownLatch; import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertTrue; import static org.junit.Assume.assumeFalse; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyLong; @@ -51,11 +60,13 @@ import static org.mockito.Mockito.mock; import static org.mockito.Mockito.mockingDetails; import static org.mockito.Mockito.when; +import static org.mockito.Mockito.withSettings; /** * A new data file found by the file watcher is loaded, and then polling for * updates carries on, as the newFileAvailable steps of the data update - * specification say ("resume polling, if any"). + * specification say ("resume polling, if any"). A new file that cannot be + * opened is given up on after a time, as its Errors table says. */ public class DataUpdateServiceWatchedFileTests { @@ -63,71 +74,45 @@ public class DataUpdateServiceWatchedFileTests { private static final long RETRY_MILLIS = 100; + /** + * The wait for a file that cannot be opened, shortened from ten minutes. + */ + private static final long OPEN_TIMEOUT_MILLIS = 1000; + + /** + * The most tries that wait allows, being one at the start, one after + * each pause and one more for a slow runner. + */ + private static final long MAX_TRIES = OPEN_TIMEOUT_MILLIS / + DataUpdateServiceDefault.WATCHED_FILE_OPEN_PAUSE_MILLIS + 2; + + private static final String MAC_MESSAGE = + "File watchers are not well implemented in OS X, " + + "so don't run this unit test as it is unlikely to pass."; + @Test public void WatchedFileRefresh_PollingCarriesOn() throws Exception { - assumeFalse("File watchers are not well implemented in OS X, " + - "so don't run this unit test as it is unlikely to pass.", - System.getProperty("os.name").contains("Mac OS X")); + assumeFalse(MAC_MESSAGE, isMac()); - // The polling timers scheduled, and the watcher's task, which the - // test runs itself. final List timers = new ArrayList<>(); final AtomicReference watcher = new AtomicReference<>(); - FutureFactory futureFactory = mock(FutureFactory.class); - when(futureFactory.schedule(any(Runnable.class), anyLong())) - .then(new Answer() { - @Override - public Object answer(InvocationOnMock invocation) { - timers.add((Long) invocation.getArgument(1)); - return mock(ScheduledFuture.class); - } - }); - when(futureFactory.scheduleRepeating(any(Runnable.class), anyLong())) - .then(new Answer() { - @Override - public Object answer(InvocationOnMock invocation) { - watcher.set((Runnable) invocation.getArgument(0)); - return mock(ScheduledFuture.class); - } - }); FileWrapperFactory fileWrapperFactory = mock(FileWrapperFactory.class); - when(fileWrapperFactory.build(anyString())).then(new Answer() { - @Override - public Object answer(InvocationOnMock invocation) throws Throwable { - return new FileWrapperDefault( - (String) invocation.getArgument(0)); - } - }); - when(fileWrapperFactory.getLastModified(anyString())) - .then(new Answer() { - @Override - public Object answer(InvocationOnMock invocation) { - return new File((String) invocation.getArgument(0)) - .lastModified(); - } - }); + when(fileWrapperFactory.build(anyString())).then(invocation -> + new FileWrapperDefault((String) invocation.getArgument(0))); + when(fileWrapperFactory.getLastModified(anyString())).then( + invocation -> lastModified(invocation.getArgument(0))); DataUpdateService dataUpdate = new DataUpdateServiceDefault( null, mock(HttpClient.class), fileWrapperFactory, - futureFactory); + futureFactory(timers, watcher)); - @SuppressWarnings("unchecked") OnPremiseAspectEngine engine = - mock(OnPremiseAspectEngine.class); + ? extends AspectPropertyMetaData> engine = engine(); Path directory = Files.createTempDirectory("watched"); Path dataFile = directory.resolve("data.dat"); Files.write(dataFile, "Old".getBytes()); - - DataFileConfigurationDefault config = new DataFileConfigurationDefault(); - config.setAutomaticUpdatesEnabled(true); - config.setFileSystemWatcherEnabled(true); - config.setPollingIntervalSeconds(1); - config.setDataFilePath(dataFile.toString()); - AspectEngineDataFileDefault file = new AspectEngineDataFileDefault(); - file.setEngine(engine); - file.setConfiguration(config); + AspectEngineDataFileDefault file = watchedFile(engine, dataFile); try { dataUpdate.registerDataFile(file); @@ -152,21 +137,199 @@ public Object answer(InvocationOnMock invocation) { file.getFuture()); } finally { - if (config.getWatchKey() != null) { - config.getWatchKey().cancel(); + cleanUp(file, directory, dataFile); + } + } + + /** + * A watched file that cannot be opened is tried, with a pause between + * tries, until the wait ends. The update then reports + * AUTO_UPDATE_TIMEOUT, logs nothing, and leaves the engine and the + * polling timer as they were. + */ + @Test + public void WatchedFileCannotBeOpened_GivesUpAfterTheWait() + throws Exception { + assumeFalse(MAC_MESSAGE, isMac()); + + final List timers = new ArrayList<>(); + final AtomicReference watcher = new AtomicReference<>(); + final AtomicBoolean locked = new AtomicBoolean(true); + final AtomicInteger tries = new AtomicInteger(); + final AtomicInteger logged = new AtomicInteger(); + + // The service may call these many times while it waits, so they + // are stub only mocks, which keep no record of their calls. + final BinaryReader reader = + mock(BinaryReader.class, withSettings().stubOnly()); + FileWrapper wrapper = + mock(FileWrapper.class, withSettings().stubOnly()); + when(wrapper.getReader()).then(invocation -> { + tries.incrementAndGet(); + if (locked.get()) { + // Stops a loop with no pause using a whole processor. + Thread.sleep(1); + throw new IOException("The file is locked"); } - // Windows keeps the file until the reader's mapped buffer is - // collected, so whatever cannot be deleted now is deleted when - // the test run ends. Those are deleted in the reverse order they - // are registered, so the folder is registered first. - directory.toFile().deleteOnExit(); - if (dataFile.toFile().delete() == false) { - dataFile.toFile().deleteOnExit(); + return reader; + }); + FileWrapperFactory fileWrapperFactory = + mock(FileWrapperFactory.class, withSettings().stubOnly()); + when(fileWrapperFactory.build(anyString())).thenReturn(wrapper); + when(fileWrapperFactory.getLastModified(anyString())).then( + invocation -> lastModified(invocation.getArgument(0))); + Logger logger = mock(Logger.class, withSettings().stubOnly() + .defaultAnswer(invocation -> { + String name = invocation.getMethod().getName(); + if (name.equals("warn") || name.equals("error")) { + logged.incrementAndGet(); + } + return Answers.RETURNS_DEFAULTS.answer(invocation); + })); + DataUpdateServiceDefault dataUpdate = new DataUpdateServiceDefault( + logger, + mock(HttpClient.class), + fileWrapperFactory, + futureFactory(timers, watcher)); + dataUpdate.watchedFileOpenTimeoutMillis = OPEN_TIMEOUT_MILLIS; + + final AtomicReference status = + new AtomicReference<>(); + final CountDownLatch done = new CountDownLatch(1); + dataUpdate.onUpdateComplete((sender, args) -> { + if (args.getStatus() != AutoUpdateStatus.AUTO_UPDATE_IN_PROGRESS && + status.compareAndSet(null, args.getStatus())) { + done.countDown(); } - directory.toFile().delete(); + }); + + OnPremiseAspectEngine engine = engine(); + Path directory = Files.createTempDirectory("watched"); + Path dataFile = directory.resolve("data.dat"); + Files.write(dataFile, "Old".getBytes()); + AspectEngineDataFileDefault file = watchedFile(engine, dataFile); + + // The watcher runs on its own thread, so the test can stop waiting + // for an update that never ends. + final AtomicBoolean stop = new AtomicBoolean(false); + Thread runner = new Thread(() -> { + while (stop.get() == false && done.getCount() > 0) { + watcher.get().run(); + try { + Thread.sleep(RETRY_MILLIS); + } catch (InterruptedException e) { + return; + } + } + }); + runner.setDaemon(true); + + try { + dataUpdate.registerDataFile(file); + assertNotNull("Registering sets up the file watcher", + watcher.get()); + int loggedBefore = logged.get(); + + Files.write(dataFile, "New".getBytes()); + runner.start(); + boolean gaveUp = done.await( + WAIT_FOR_EVENT_MILLIS, TimeUnit.MILLISECONDS); + + assertTrue("The watcher tries to open the new file", + tries.get() > 0); + assertTrue("The update gives up on a file it cannot open", + gaveUp); + assertEquals(AutoUpdateStatus.AUTO_UPDATE_TIMEOUT, status.get()); + assertTrue("The tries are paused, but " + tries.get() + + " were made", tries.get() <= MAX_TRIES); + assertEquals("The engine is not refreshed", + 0, refreshCount(engine)); + assertEquals("Nothing is logged", loggedBefore, logged.get()); + assertEquals("The polling timer is kept", 1, timers.size()); + assertNotNull("The data file holds the polling timer", + file.getFuture()); + } + finally { + // Lets an update that would never give up end, so the thread + // stops. + stop.set(true); + locked.set(false); + if (runner.isAlive()) { + runner.join(WAIT_FOR_EVENT_MILLIS); + } + cleanUp(file, directory, dataFile); } } + /** + * Records the polling timers scheduled, and keeps the watcher's task for + * the test to run itself. + */ + private static FutureFactory futureFactory( + final List timers, + final AtomicReference watcher) { + FutureFactory futureFactory = mock(FutureFactory.class); + when(futureFactory.schedule(any(Runnable.class), anyLong())) + .then(invocation -> { + timers.add((Long) invocation.getArgument(1)); + return mock(ScheduledFuture.class); + }); + when(futureFactory.scheduleRepeating(any(Runnable.class), anyLong())) + .then(invocation -> { + watcher.set((Runnable) invocation.getArgument(0)); + return mock(ScheduledFuture.class); + }); + return futureFactory; + } + + @SuppressWarnings("unchecked") + private static OnPremiseAspectEngine engine() { + return mock(OnPremiseAspectEngine.class); + } + + private static AspectEngineDataFileDefault watchedFile( + OnPremiseAspectEngine engine, + Path dataFile) { + DataFileConfigurationDefault config = new DataFileConfigurationDefault(); + config.setAutomaticUpdatesEnabled(true); + config.setFileSystemWatcherEnabled(true); + config.setPollingIntervalSeconds(1); + config.setDataFilePath(dataFile.toString()); + AspectEngineDataFileDefault file = new AspectEngineDataFileDefault(); + file.setEngine(engine); + file.setConfiguration(config); + return file; + } + + private static void cleanUp( + AspectEngineDataFileDefault file, + Path directory, + Path dataFile) { + if (file.getConfiguration().getWatchKey() != null) { + file.getConfiguration().getWatchKey().cancel(); + } + // Windows keeps the file until the reader's mapped buffer is + // collected, so whatever cannot be deleted now is deleted when the + // test run ends. Those are deleted in the reverse order they are + // registered, so the folder is registered first. + directory.toFile().deleteOnExit(); + if (dataFile.toFile().delete() == false) { + dataFile.toFile().deleteOnExit(); + } + directory.toFile().delete(); + } + + private static boolean isMac() { + return System.getProperty("os.name").contains("Mac OS X"); + } + + private static long lastModified(Object path) { + return new File((String) path).lastModified(); + } + private static long refreshCount(Object engine) { return mockingDetails(engine).getInvocations().stream() .filter(i -> i.getMethod().getName().equals("refreshData")) From 03999c5a60ebe8f28896144956856a96f80d1cfc Mon Sep 17 00:00:00 2001 From: James Rosewell Date: Wed, 11 Nov 2026 07:35:28 +0000 Subject: [PATCH 5/7] Give up on a watched file that cannot be opened after ten minutes The file watcher tried to open a new data file for ever, with no pause and a warning on every try, while holding the data file's update lock. It now tries every half second for up to ten minutes, logs nothing, and then ends the update with AUTO_UPDATE_TIMEOUT, leaving the engine and the polling timer as they were. This is what .NET does and what the Errors table of the data update specification says. The file is registered again only when the refresh ran, because only the refresh cancels the polling timer. --- .../services/DataUpdateServiceDefault.java | 32 +++++++++++++------ 1 file changed, 23 insertions(+), 9 deletions(-) diff --git a/pipeline/pipeline.engines/src/main/java/fiftyone/pipeline/engines/services/DataUpdateServiceDefault.java b/pipeline/pipeline.engines/src/main/java/fiftyone/pipeline/engines/services/DataUpdateServiceDefault.java index 6ba5361e..df482ae6 100644 --- a/pipeline/pipeline.engines/src/main/java/fiftyone/pipeline/engines/services/DataUpdateServiceDefault.java +++ b/pipeline/pipeline.engines/src/main/java/fiftyone/pipeline/engines/services/DataUpdateServiceDefault.java @@ -391,21 +391,35 @@ private void dataFileUpdated(Object sender) { // before notifying the engine, otherwise the copy // may still be in progress. boolean fileLockable = false; - while (fileLockable == false) { + long deadline = System.nanoTime() + + TimeUnit.MILLISECONDS.toNanos( + watchedFileOpenTimeoutMillis); + while (fileLockable == false && + System.nanoTime() - deadline < 0) { try (BinaryReader reader = fileWrapperFactory.build( aspectDataFile.getDataFilePath()).getReader()) { fileLockable = true; - // Complete the update - status = updatedFileAvailable(dataFile); } catch (IOException e) { - logger.warn("Exception while reading the data file at '{}'", - aspectDataFile.getDataFilePath(), e); + try { + Thread.sleep(WATCHED_FILE_OPEN_PAUSE_MILLIS); + } catch (InterruptedException ex) { + // Interrupted, so stop trying. + Thread.currentThread().interrupt(); + break; + } } } - // The refresh cancelled the polling timer, so - // register the file again to set a new one. - if (dataFile.getIsRegistered()) { - registerDataFile(dataFile); + if (fileLockable) { + // Complete the update + status = updatedFileAvailable(dataFile); + // The refresh cancelled the polling timer, so + // register the file again to set a new one. + if (dataFile.getIsRegistered()) { + registerDataFile(dataFile); + } + } else { + // The polling timer is left as it was. + status = AutoUpdateStatus.AUTO_UPDATE_TIMEOUT; } } } From bddd7468a5e29ab12e1f57211f2aae1a7f3a376a Mon Sep 17 00:00:00 2001 From: James Rosewell Date: Wed, 11 Nov 2026 08:39:00 +0000 Subject: [PATCH 6/7] Test that an interrupted wait for a watched file leaves the update unfinished An interrupt, as unregistering the data file gives the watcher's task, ends the wait for a file that cannot be opened, and the update reports AUTO_UPDATE_TIMEOUT although the file was not tried for the whole wait. The new test requires AUTO_UPDATE_IN_PROGRESS, the status for an update that did not finish, so it fails on the timeout. The two tests of a file that cannot be opened now share one fixture. The stand-in file pauses without sleeping, so an interrupt reaches the service rather than the stand-in. --- .../DataUpdateServiceWatchedFileTests.java | 226 ++++++++++++------ 1 file changed, 147 insertions(+), 79 deletions(-) diff --git a/pipeline/pipeline.engines/src/test/java/fiftyone/pipeline/engines/services/DataUpdateServiceWatchedFileTests.java b/pipeline/pipeline.engines/src/test/java/fiftyone/pipeline/engines/services/DataUpdateServiceWatchedFileTests.java index f41120da..b57a16ca 100644 --- a/pipeline/pipeline.engines/src/test/java/fiftyone/pipeline/engines/services/DataUpdateServiceWatchedFileTests.java +++ b/pipeline/pipeline.engines/src/test/java/fiftyone/pipeline/engines/services/DataUpdateServiceWatchedFileTests.java @@ -49,6 +49,7 @@ import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; +import java.util.concurrent.locks.LockSupport; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; @@ -66,7 +67,8 @@ * A new data file found by the file watcher is loaded, and then polling for * updates carries on, as the newFileAvailable steps of the data update * specification say ("resume polling, if any"). A new file that cannot be - * opened is given up on after a time, as its Errors table says. + * opened is given up on after a time, as its Errors table says, or at + * once when the wait is interrupted. */ public class DataUpdateServiceWatchedFileTests { @@ -86,6 +88,14 @@ public class DataUpdateServiceWatchedFileTests { private static final long MAX_TRIES = OPEN_TIMEOUT_MILLIS / DataUpdateServiceDefault.WATCHED_FILE_OPEN_PAUSE_MILLIS + 2; + /** + * A wait long enough that only an interrupt can end it within + * {@link #INTERRUPTED_WAIT_MILLIS}. + */ + private static final long LONG_OPEN_TIMEOUT_MILLIS = 60000; + + private static final long INTERRUPTED_WAIT_MILLIS = 5000; + private static final String MAC_MESSAGE = "File watchers are not well implemented in OS X, " + "so don't run this unit test as it is unlikely to pass."; @@ -151,108 +161,166 @@ public void WatchedFileRefresh_PollingCarriesOn() throws Exception { public void WatchedFileCannotBeOpened_GivesUpAfterTheWait() throws Exception { assumeFalse(MAC_MESSAGE, isMac()); + try (LockedFile locked = new LockedFile(OPEN_TIMEOUT_MILLIS)) { + locked.start(); + boolean gaveUp = locked.done.await( + WAIT_FOR_EVENT_MILLIS, TimeUnit.MILLISECONDS); + + assertTrue("The watcher tries to open the new file", + locked.tries.get() > 0); + assertTrue("The update gives up on a file it cannot open", + gaveUp); + assertEquals(AutoUpdateStatus.AUTO_UPDATE_TIMEOUT, + locked.status.get()); + assertTrue("The tries are paused, but " + locked.tries.get() + + " were made", locked.tries.get() <= MAX_TRIES); + locked.assertLeftAsItWas(); + } + } + + /** + * An interrupt, as unregistering the data file gives the watcher's task, + * ends the wait at once. The update is left unfinished, so it reports + * AUTO_UPDATE_IN_PROGRESS rather than a timeout, logs nothing, and leaves + * the engine and the polling timer as they were. + */ + @Test + public void WatchedFileWaitInterrupted_LeavesTheUpdateUnfinished() + throws Exception { + assumeFalse(MAC_MESSAGE, isMac()); + try (LockedFile locked = new LockedFile(LONG_OPEN_TIMEOUT_MILLIS)) { + locked.start(); + long end = System.currentTimeMillis() + WAIT_FOR_EVENT_MILLIS; + while (locked.tries.get() == 0 && + System.currentTimeMillis() < end) { + Thread.sleep(RETRY_MILLIS); + } + assertTrue("The watcher tries to open the new file", + locked.tries.get() > 0); + + locked.runner.interrupt(); + boolean ended = locked.done.await( + INTERRUPTED_WAIT_MILLIS, TimeUnit.MILLISECONDS); + + assertTrue("The wait ends when it is interrupted", ended); + assertEquals(AutoUpdateStatus.AUTO_UPDATE_IN_PROGRESS, + locked.status.get()); + locked.assertLeftAsItWas(); + } + } + + /** + * A watched data file that throws on every open, the service watching + * it, and the thread that runs the watcher's task. + */ + private static class LockedFile implements AutoCloseable { final List timers = new ArrayList<>(); final AtomicReference watcher = new AtomicReference<>(); final AtomicBoolean locked = new AtomicBoolean(true); final AtomicInteger tries = new AtomicInteger(); final AtomicInteger logged = new AtomicInteger(); - - // The service may call these many times while it waits, so they - // are stub only mocks, which keep no record of their calls. - final BinaryReader reader = - mock(BinaryReader.class, withSettings().stubOnly()); - FileWrapper wrapper = - mock(FileWrapper.class, withSettings().stubOnly()); - when(wrapper.getReader()).then(invocation -> { - tries.incrementAndGet(); - if (locked.get()) { - // Stops a loop with no pause using a whole processor. - Thread.sleep(1); - throw new IOException("The file is locked"); - } - return reader; - }); - FileWrapperFactory fileWrapperFactory = - mock(FileWrapperFactory.class, withSettings().stubOnly()); - when(fileWrapperFactory.build(anyString())).thenReturn(wrapper); - when(fileWrapperFactory.getLastModified(anyString())).then( - invocation -> lastModified(invocation.getArgument(0))); - Logger logger = mock(Logger.class, withSettings().stubOnly() - .defaultAnswer(invocation -> { - String name = invocation.getMethod().getName(); - if (name.equals("warn") || name.equals("error")) { - logged.incrementAndGet(); - } - return Answers.RETURNS_DEFAULTS.answer(invocation); - })); - DataUpdateServiceDefault dataUpdate = new DataUpdateServiceDefault( - logger, - mock(HttpClient.class), - fileWrapperFactory, - futureFactory(timers, watcher)); - dataUpdate.watchedFileOpenTimeoutMillis = OPEN_TIMEOUT_MILLIS; - final AtomicReference status = new AtomicReference<>(); final CountDownLatch done = new CountDownLatch(1); - dataUpdate.onUpdateComplete((sender, args) -> { - if (args.getStatus() != AutoUpdateStatus.AUTO_UPDATE_IN_PROGRESS && - status.compareAndSet(null, args.getStatus())) { - done.countDown(); - } - }); - - OnPremiseAspectEngine engine = engine(); - Path directory = Files.createTempDirectory("watched"); - Path dataFile = directory.resolve("data.dat"); - Files.write(dataFile, "Old".getBytes()); - AspectEngineDataFileDefault file = watchedFile(engine, dataFile); + final DataUpdateServiceDefault service; + final Path directory; + final Path dataFile; + final AspectEngineDataFileDefault file; + final Thread runner; + int loggedAtStart; + + LockedFile(long timeoutMillis) throws Exception { + // The service may call these many times while it waits, so they + // are stub only mocks, which keep no record of their calls. + final BinaryReader reader = + mock(BinaryReader.class, withSettings().stubOnly()); + FileWrapper wrapper = + mock(FileWrapper.class, withSettings().stubOnly()); + when(wrapper.getReader()).then(invocation -> { + tries.incrementAndGet(); + if (locked.get()) { + // Stops a loop with no pause using a whole processor, + // and leaves an interrupt for the service to see. + LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(1)); + throw new IOException("The file is locked"); + } + return reader; + }); + FileWrapperFactory fileWrapperFactory = + mock(FileWrapperFactory.class, withSettings().stubOnly()); + when(fileWrapperFactory.build(anyString())).thenReturn(wrapper); + when(fileWrapperFactory.getLastModified(anyString())).then( + invocation -> lastModified(invocation.getArgument(0))); + Logger logger = mock(Logger.class, withSettings().stubOnly() + .defaultAnswer(invocation -> { + String name = invocation.getMethod().getName(); + if (name.equals("warn") || name.equals("error")) { + logged.incrementAndGet(); + } + return Answers.RETURNS_DEFAULTS.answer(invocation); + })); + service = new DataUpdateServiceDefault( + logger, + mock(HttpClient.class), + fileWrapperFactory, + futureFactory(timers, watcher)); + service.watchedFileOpenTimeoutMillis = timeoutMillis; + // The first update after the change is the one that waits, + // because only the watcher sets the time it compares with. + service.onUpdateComplete((sender, args) -> { + if (status.compareAndSet(null, args.getStatus())) { + done.countDown(); + } + }); - // The watcher runs on its own thread, so the test can stop waiting - // for an update that never ends. - final AtomicBoolean stop = new AtomicBoolean(false); - Thread runner = new Thread(() -> { - while (stop.get() == false && done.getCount() > 0) { - watcher.get().run(); - try { - Thread.sleep(RETRY_MILLIS); - } catch (InterruptedException e) { - return; + directory = Files.createTempDirectory("watched"); + dataFile = directory.resolve("data.dat"); + Files.write(dataFile, "Old".getBytes()); + file = watchedFile(engine, dataFile); + + // The watcher runs on its own thread, so the test can stop + // waiting for an update that does not end. + runner = new Thread(() -> { + while (stop.get() == false && done.getCount() > 0) { + watcher.get().run(); + try { + Thread.sleep(RETRY_MILLIS); + } catch (InterruptedException e) { + return; + } } - } - }); - runner.setDaemon(true); + }); + runner.setDaemon(true); + } - try { - dataUpdate.registerDataFile(file); + /** + * Registers the file, changes it, and starts the watcher's thread. + */ + void start() throws Exception { + service.registerDataFile(file); assertNotNull("Registering sets up the file watcher", watcher.get()); - int loggedBefore = logged.get(); - + loggedAtStart = logged.get(); Files.write(dataFile, "New".getBytes()); runner.start(); - boolean gaveUp = done.await( - WAIT_FOR_EVENT_MILLIS, TimeUnit.MILLISECONDS); + } - assertTrue("The watcher tries to open the new file", - tries.get() > 0); - assertTrue("The update gives up on a file it cannot open", - gaveUp); - assertEquals(AutoUpdateStatus.AUTO_UPDATE_TIMEOUT, status.get()); - assertTrue("The tries are paused, but " + tries.get() + - " were made", tries.get() <= MAX_TRIES); + void assertLeftAsItWas() { assertEquals("The engine is not refreshed", 0, refreshCount(engine)); - assertEquals("Nothing is logged", loggedBefore, logged.get()); + assertEquals("Nothing is logged", loggedAtStart, logged.get()); assertEquals("The polling timer is kept", 1, timers.size()); assertNotNull("The data file holds the polling timer", file.getFuture()); } - finally { - // Lets an update that would never give up end, so the thread - // stops. + + @Override + public void close() throws Exception { + // Lets an update that would not give up end, so the thread stops. stop.set(true); locked.set(false); if (runner.isAlive()) { From 3439e5e24bf66f85eeaf90d607870eda15f58210 Mon Sep 17 00:00:00 2001 From: James Rosewell Date: Wed, 11 Nov 2026 08:40:07 +0000 Subject: [PATCH 7/7] Leave an interrupted wait for a watched file unfinished, not timed out An interrupt ended the wait and reported AUTO_UPDATE_TIMEOUT, although the file had not been tried for the whole wait. The update now keeps the status it started with, AUTO_UPDATE_IN_PROGRESS, which is what it reports for any watcher event that does not finish an update. The engine and the polling timer are left as they were, as for a timeout. The IOException the loop catches is no longer named e, as nothing reads it. --- .../engines/services/DataUpdateServiceDefault.java | 11 ++++++++--- 1 file changed, 8 insertions(+), 3 deletions(-) diff --git a/pipeline/pipeline.engines/src/main/java/fiftyone/pipeline/engines/services/DataUpdateServiceDefault.java b/pipeline/pipeline.engines/src/main/java/fiftyone/pipeline/engines/services/DataUpdateServiceDefault.java index df482ae6..a2c4af85 100644 --- a/pipeline/pipeline.engines/src/main/java/fiftyone/pipeline/engines/services/DataUpdateServiceDefault.java +++ b/pipeline/pipeline.engines/src/main/java/fiftyone/pipeline/engines/services/DataUpdateServiceDefault.java @@ -391,6 +391,7 @@ private void dataFileUpdated(Object sender) { // before notifying the engine, otherwise the copy // may still be in progress. boolean fileLockable = false; + boolean interrupted = false; long deadline = System.nanoTime() + TimeUnit.MILLISECONDS.toNanos( watchedFileOpenTimeoutMillis); @@ -399,12 +400,14 @@ private void dataFileUpdated(Object sender) { try (BinaryReader reader = fileWrapperFactory.build( aspectDataFile.getDataFilePath()).getReader()) { fileLockable = true; - } catch (IOException e) { + } catch (IOException ignored) { try { Thread.sleep(WATCHED_FILE_OPEN_PAUSE_MILLIS); } catch (InterruptedException ex) { - // Interrupted, so stop trying. + // Interrupted, as unregistering the data + // file does, so stop trying. Thread.currentThread().interrupt(); + interrupted = true; break; } } @@ -417,10 +420,12 @@ private void dataFileUpdated(Object sender) { if (dataFile.getIsRegistered()) { registerDataFile(dataFile); } - } else { + } else if (interrupted == false) { // The polling timer is left as it was. status = AutoUpdateStatus.AUTO_UPDATE_TIMEOUT; } + // An interrupted wait leaves the update unfinished, + // so the status stays AUTO_UPDATE_IN_PROGRESS. } } }