Skip to content

Refresh the engine once when a data update is applied in memory #92

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
jwrosewell merged 9 commits into from
Nov 10, 2026
Merged

Refresh the engine once when a data update is applied in memory #92

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
0b500e7
Test that an update applied in memory refreshes the engine once
jwrosewell Nov 9, 2026
66a5704
Refresh the engine once when an update is applied in memory
jwrosewell Nov 9, 2026
d99c9b7
Wait for the watcher's update in the watcher test, not the polling ti…
jwrosewell Nov 9, 2026
6ff3192
Test that a manual update sets the next update timer again
jwrosewell Nov 9, 2026
399afbb
Set the next update timer after a manual update, as the specification…
jwrosewell Nov 9, 2026
4215118
Check that a manual update leaves one update timer, not two
jwrosewell Nov 9, 2026
7d0b6e2
Test that a failed manual refresh keeps polling, and that unregisteri…
jwrosewell Nov 10, 2026
3e51394
Resume polling after a failed manual refresh, and clear the registrat…
jwrosewell Nov 10, 2026
763f6a4
Say that a refresh from the data file clears the update timer
jwrosewell Nov 10, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
23 changes: 21 additions & 2 deletions ...ne.engines/src/main/java/fiftyone/pipeline/engines/services/DataUpdateServiceDefault.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -226,6 +226,15 @@ private AutoUpdateStatus checkForUpdate(Object state, boolean manualUpdate) {
// yet been refreshed.
logger.debug("Calling updatedFileAvailable");
result = updatedFileAvailable(dataFile);
} else if (result == AUTO_UPDATE_SUCCESS) {
// The update was applied in memory. Clear the timer, as
// updatedFileAvailable does, so a new one is registered.
synchronized (futureLock) {
if (dataFile.getFuture() != null) {
dataFile.getFuture().cancel(true);
dataFile.setFuture(null);
}
}
} else {
// No update available.
// If this was a manual call to update then do nothing.
Expand All @@ -249,9 +258,15 @@ public void run() {
}
}
}
if (manualUpdate == false) {
if (manualUpdate == false ||
((result == AUTO_UPDATE_SUCCESS ||
result == AUTO_UPDATE_REFRESH_FAILED) &&
dataFile.getIsRegistered())) {
// Re-register the engine with the data update service
// so it knows when next set of data should be available.
// A manual update of a registered data file does this too,
// whether or not the refresh worked, as a refresh from the
// data file clears the timer, so polling resumes after it.
registerDataFile(dataFile);
}

Expand Down Expand Up @@ -565,6 +580,9 @@ public void unregisterDataFile(AspectEngineDataFile dataFile) {
}
}
}
if (dataFile != null) {
dataFile.setDataUpdateService(null);
}
}

/**
Expand Down Expand Up @@ -805,7 +823,8 @@ private AutoUpdateStatus checkForUpdateFromUrl(AspectEngineDataFile dataFile) {
dataFile.getConfiguration().setData(reader.toByteArray());
}
}
return memoryDownloadResult;
// The engine, or the configuration, now holds the new data.
return AUTO_UPDATE_SUCCESS;
} catch (Exception ex) {
logger.error("An error occurred when applying a data update to engine '{}'" ,
Objects.nonNull(dataFile.getEngine())?
Expand Down
303 changes: 300 additions & 3 deletions ...line.engines/src/test/java/fiftyone/pipeline/engines/services/DataUpdateServiceTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -45,6 +45,7 @@
import org.mockito.ArgumentCaptor;
import org.mockito.ArgumentMatcher;
import org.mockito.internal.matchers.LessOrEqual;
import org.mockito.invocation.Invocation;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
import org.slf4j.LoggerFactory;
Expand All @@ -61,6 +62,7 @@
import java.nio.file.Path;
import java.nio.file.Paths;
import java.nio.file.StandardCopyOption;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Calendar;
import java.util.Collection;
Expand All @@ -72,6 +74,7 @@

import static fiftyone.pipeline.engines.services.DataUpdateService.AutoUpdateStatus.AUTO_UPDATE_HTTPS_ERR;
import static fiftyone.pipeline.engines.services.DataUpdateService.AutoUpdateStatus.AUTO_UPDATE_NOT_NEEDED;
import static fiftyone.pipeline.engines.services.DataUpdateService.AutoUpdateStatus.AUTO_UPDATE_SUCCESS;
import static org.junit.Assert.*;
import static org.junit.Assume.assumeFalse;
import static org.mockito.Mockito.*;
Expand Down Expand Up @@ -415,14 +418,17 @@ public Object answer(InvocationOnMock invocationOnMock) throws Throwable {
}
});

// Configure a flag to be set when processing
// is complete.
// Configure a flag to be set when the watcher has applied the
// update. The polling timer's own check also completes, with no
// update found, and must not release the flag.
final Semaphore completeFlag = new Semaphore(1);
completeFlag.acquire();
dataUpdate.onUpdateComplete(new OnUpdateComplete() {
@Override
public void call(Object sender, DataUpdateService.DataUpdateCompleteArgs args) {
completeFlag.release();
if (args.getStatus() == AUTO_UPDATE_SUCCESS) {
completeFlag.release();
}
}
});

Expand Down Expand Up @@ -1229,6 +1235,297 @@ public boolean matches(byte[] argument) {
verify(futureFactory, times(1)).schedule(any(Runnable.class), anyLong());
}

/**
* A check that finds an update for a data file with no file path
* applies it in memory, so it refreshes the engine once, with the
* downloaded data, and does not refresh it again.
*/
@Test
public void DataUpdateService_CheckForUpdate_InMemory_RefreshesOnce()
throws IOException {
// Arrange
configureNoFileSystem();
when(httpClientConnection.getResponseCode())
.thenReturn(HttpURLConnection.HTTP_OK);
when(httpClientConnection.getResponseMessage()).thenReturn(null);
when(httpClientConnection.getHeaderField("Content-MD5"))
.thenReturn("08527dcbdd437e7fa6c084423d06dba6");
when(httpClientConnection.getInputStream()).thenReturn(
DataUpdateServiceTests.class.getClassLoader()
.getResourceAsStream("file.gz"));

OnPremiseAspectEngine<? extends AspectData, ? extends AspectPropertyMetaData> engine =
mock(OnPremiseAspectEngine.class);
DataFileConfigurationDefault config = new DataFileConfigurationDefault();
config.setDataUpdateUrl("http://www.test.com");
config.setVerifyMd5(true);
config.setDecompressContent(true);
config.setVerifyModifiedSince(false);

AspectEngineDataFileDefault file = new AspectEngineDataFileDefault();
file.setEngine(engine);
file.setConfiguration(config);
when(engine.getDataFileMetaData()).thenReturn(file);

// Act
DataUpdateService.AutoUpdateStatus status =
dataUpdate.checkForUpdate(engine);

// Assert
assertEquals(AUTO_UPDATE_SUCCESS, status);
List<String> refreshes = new ArrayList<>();
for (Invocation invocation : mockingDetails(engine).getInvocations()) {
if (invocation.getMethod().getName().equals("refreshData")) {
refreshes.add(invocation.getMethod().toGenericString());
}
}
assertEquals(
"The engine was not refreshed exactly once: " + refreshes,
1,
refreshes.size());
verify(engine, times(1)).refreshData(any(), any(byte[].class));
}

/**
* A manual check that applies an update from the data file to a data
* file registered for automatic updates sets the next update timer, as
* a check the timer started does.
*/
@Test
public void DataUpdateService_CheckForUpdate_FileUpdated_TimerSetAgain() {
// Arrange
OnPremiseAspectEngine<? extends AspectData, ? extends AspectPropertyMetaData> engine =
mock(OnPremiseAspectEngine.class);
DataFileConfigurationDefault config = new DataFileConfigurationDefault();
config.setAutomaticUpdatesEnabled(autoUpdateEnabled);
config.setFileSystemWatcherEnabled(false);

AspectEngineDataFileDefault file = new AspectEngineDataFileDefault();
file.setEngine(engine);
file.setConfiguration(config);
when(engine.getDataFileMetaData()).thenReturn(file);
configureFileUpdate(engine);

dataUpdate.registerDataFile(file);
int timers = autoUpdateEnabled ? 1 : 0;
verify(futureFactory, times(timers)).schedule(
any(Runnable.class),
anyLong());

ScheduledFuture<?> pending = file.getFuture();

// Act
DataUpdateService.AutoUpdateStatus status =
dataUpdate.checkForUpdate(engine);

// Assert
assertEquals(AUTO_UPDATE_SUCCESS, status);
verify(engine, times(1)).refreshData((String) any());
verify(futureFactory, times(2 * timers)).schedule(
any(Runnable.class),
anyLong());
assertEquals(
"The next update timer was not set after the manual update.",
autoUpdateEnabled,
file.getFuture() != null);

if (autoUpdateEnabled) {
// The pending timer is cancelled, so only the new one is left.
assertTrue(
"The timer pending before the manual update was not cancelled.",
pending.isCancelled());
assertNotSame(pending, file.getFuture());
}
}

/**
* A manual check whose refresh from the data file fails still leaves
* the next update timer set, as a check the timer started does, so
* polling resumes.
*/
@Test
public void DataUpdateService_CheckForUpdate_FileRefreshFails_TimerSetAgain() {
// updatedFileAvailable warns on each of its 10 tries.
ignoreWranings = 10;

// Arrange
OnPremiseAspectEngine<? extends AspectData, ? extends AspectPropertyMetaData> engine =
mock(OnPremiseAspectEngine.class);
doThrow(new IllegalStateException("The refresh failed."))
.when(engine).refreshData((String) any());
DataFileConfigurationDefault config = new DataFileConfigurationDefault();
config.setAutomaticUpdatesEnabled(autoUpdateEnabled);
config.setFileSystemWatcherEnabled(false);

AspectEngineDataFileDefault file = new AspectEngineDataFileDefault();
file.setEngine(engine);
file.setConfiguration(config);
when(engine.getDataFileMetaData()).thenReturn(file);
configureFileUpdate(engine);

dataUpdate.registerDataFile(file);
int timers = autoUpdateEnabled ? 1 : 0;
ScheduledFuture<?> pending = file.getFuture();

// Act
DataUpdateService.AutoUpdateStatus status =
dataUpdate.checkForUpdate(engine);

// Assert
assertEquals(
DataUpdateService.AutoUpdateStatus.AUTO_UPDATE_REFRESH_FAILED,
status);
verify(futureFactory, times(2 * timers)).schedule(
any(Runnable.class),
anyLong());
assertEquals(
"The next update timer was not set after the failed refresh.",
autoUpdateEnabled,
file.getFuture() != null);
if (autoUpdateEnabled) {
assertTrue(pending.isCancelled());
assertNotSame(pending, file.getFuture());
}
}

/**
* A data file that has been unregistered is no longer registered, so a
* manual update of it sets no update timer.
*/
@Test
public void DataUpdateService_Unregister_NoLongerRegistered() {
// Arrange
OnPremiseAspectEngine<? extends AspectData, ? extends AspectPropertyMetaData> engine =
mock(OnPremiseAspectEngine.class);
DataFileConfigurationDefault config = new DataFileConfigurationDefault();
config.setAutomaticUpdatesEnabled(autoUpdateEnabled);
config.setFileSystemWatcherEnabled(false);

AspectEngineDataFileDefault file = new AspectEngineDataFileDefault();
file.setEngine(engine);
file.setConfiguration(config);
when(engine.getDataFileMetaData()).thenReturn(file);
configureFileUpdate(engine);
dataUpdate.registerDataFile(file);
int timers = autoUpdateEnabled ? 1 : 0;

// Act
dataUpdate.unregisterDataFile(file);
DataUpdateService.AutoUpdateStatus status =
dataUpdate.checkForUpdate(engine);

// Assert
assertFalse(file.getIsRegistered());
assertEquals(AUTO_UPDATE_SUCCESS, status);
verify(futureFactory, times(timers)).schedule(
any(Runnable.class),
anyLong());
assertNull(file.getFuture());
}

/**
* A manual check that applies an update in memory to a data file
* registered for automatic updates sets the next update timer, as a
* check the timer started does.
*/
@Test
public void DataUpdateService_CheckForUpdate_InMemory_TimerSetAgain()
throws IOException {
// Arrange
configureNoFileSystem();
when(httpClientConnection.getResponseCode())
.thenReturn(HttpURLConnection.HTTP_OK);
when(httpClientConnection.getResponseMessage()).thenReturn(null);
when(httpClientConnection.getHeaderField("Content-MD5"))
.thenReturn("08527dcbdd437e7fa6c084423d06dba6");
when(httpClientConnection.getInputStream()).thenReturn(
DataUpdateServiceTests.class.getClassLoader()
.getResourceAsStream("file.gz"));

OnPremiseAspectEngine<? extends AspectData, ? extends AspectPropertyMetaData> engine =
mock(OnPremiseAspectEngine.class);
DataFileConfigurationDefault config = new DataFileConfigurationDefault();
config.setAutomaticUpdatesEnabled(autoUpdateEnabled);
config.setDataUpdateUrl("http://www.test.com");
config.setVerifyMd5(true);
config.setDecompressContent(true);
config.setVerifyModifiedSince(false);

AspectEngineDataFileDefault file = new AspectEngineDataFileDefault();
file.setEngine(engine);
file.setConfiguration(config);
when(engine.getDataFileMetaData()).thenReturn(file);

dataUpdate.registerDataFile(file);
int timers = autoUpdateEnabled ? 1 : 0;
verify(futureFactory, times(timers)).schedule(
any(Runnable.class),
anyLong());

ScheduledFuture<?> pending = file.getFuture();

// Act
DataUpdateService.AutoUpdateStatus status =
dataUpdate.checkForUpdate(engine);

// Assert
assertEquals(AUTO_UPDATE_SUCCESS, status);
verify(futureFactory, times(2 * timers)).schedule(
any(Runnable.class),
anyLong());
assertEquals(
"The next update timer was not set after the manual update.",
autoUpdateEnabled,
file.getFuture() != null);

if (autoUpdateEnabled) {
// The pending timer is cancelled, so only the new one is left.
assertTrue(
"The timer pending before the manual update was not cancelled.",
pending.isCancelled());
assertNotSame(pending, file.getFuture());
}
}

/**
* A manual check that finds no update leaves the timer of a data file
* registered for automatic updates as it was.
*/
@Test
public void DataUpdateService_CheckForUpdate_NoUpdate_TimerKept()
throws IOException {
// Arrange
configureNoFileSystem();
configureHttpNoUpdateAvailable();

OnPremiseAspectEngine<? extends AspectData, ? extends AspectPropertyMetaData> engine =
mock(OnPremiseAspectEngine.class);
DataFileConfigurationDefault config = new DataFileConfigurationDefault();
config.setAutomaticUpdatesEnabled(autoUpdateEnabled);
config.setDataUpdateUrl("http://www.test.com");
config.setVerifyModifiedSince(false);

AspectEngineDataFileDefault file = new AspectEngineDataFileDefault();
file.setEngine(engine);
file.setConfiguration(config);
when(engine.getDataFileMetaData()).thenReturn(file);

dataUpdate.registerDataFile(file);
Object timer = file.getFuture();

// Act
DataUpdateService.AutoUpdateStatus status =
dataUpdate.checkForUpdate(engine);

// Assert
assertEquals(AUTO_UPDATE_NOT_NEEDED, status);
verify(engine, never()).refreshData(anyString());
verify(futureFactory, times(autoUpdateEnabled ? 1 : 0)).schedule(
any(Runnable.class),
anyLong());
assertSame(timer, file.getFuture());
}



/**
Expand Down
Loading