Skip to content

Adding locking to prevent race condition #2975

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

Closed
fnachabe wants to merge 2 commits into from
Closed

Adding locking to prevent race condition #2975

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
54f6be2
Adding locking to prevent race condition
fnachabe Sep 3, 2026
c29ea6c
Merge branch 'dev' into 2974-close-database-engines-before-cloud-sync
fnachabe Oct 26, 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
14 changes: 11 additions & 3 deletions src/prerna/cluster/util/clients/CentralCloudStorage.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -526,18 +526,26 @@ public void pushEngine(String engineId) throws IOException, InterruptedException
ReentrantLock lock = EngineSyncUtility.getEngineLock(engineId);
lock.lock();
classLogger.info("Engine {} is locked", aliasAndEngineId);
// Database engines can own local mutable state through their per-engine
// audit database, even when the primary database is remote and therefore
// reports that it does not hold file locks. Uploading that live audit file can
// race with H2 MVStore compaction and leave a storage client reading beyond the
// new end of the file. Evaluate this once because the engine is closed below.
boolean closeEngineForSync = engineType == IEngine.CATALOG_TYPE.DATABASE || engine.holdsFileLocks();
boolean engineRemovedForSync = false;
try {
if (engine.holdsFileLocks()) {
if (closeEngineForSync) {
UploadUtilities.removeEngineExcludingSMSSFromDIHelper(engineId);
engineRemovedForSync = true;
engine.close();
}

storageSyncLocalToStorage(localEngineFolder, cloudEngineFolder);
storageCopyToStorage(localSmssFilePath, cloudSmssFolder);
} finally {
try {
// Re-open the engine
if (engine.holdsFileLocks()) {
// Re-open an engine that was removed, including when close or transfer failed.
if (engineRemovedForSync) {
Utility.getEngine(engineId, false);
}
} catch (Exception e) {
Expand Down
84 changes: 81 additions & 3 deletions test/prerna/cluster/util/clients/CentralCloudStorageUnitTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,6 +41,7 @@
import static org.mockito.ArgumentMatchers.contains;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.ArgumentMatchers.isNull;
import static org.mockito.Mockito.inOrder;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.mockStatic;
import static org.mockito.Mockito.never;
Expand All @@ -49,6 +50,7 @@
import static org.mockito.Mockito.verifyNoInteractions;
import static org.mockito.Mockito.when;

import java.io.IOException;
import java.lang.reflect.Field;
import java.lang.reflect.Method;
import java.nio.file.Files;
Expand All @@ -66,6 +68,7 @@
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.TestInstance;
import org.junit.jupiter.api.io.TempDir;
import org.mockito.InOrder;
import org.mockito.MockedStatic;

import prerna.auth.utils.SecurityEngineUtils;
Expand Down Expand Up @@ -1549,7 +1552,7 @@ void testPullEngineAndProjectImageFolder(@TempDir Path testDir) throws Exception
// ------- pushEngine happy path -------

@Test
void testPushEngine_happyPath_noFileLocks(@TempDir Path testDir) throws Exception {
void testPushEngine_databaseWithoutPrimaryFileLocks_isClosedAndReopened(@TempDir Path testDir) throws Exception {
IEngine mockEngine = mock(IEngine.class);
when(mockEngine.getCatalogType()).thenReturn(IEngine.CATALOG_TYPE.DATABASE);
when(mockEngine.holdsFileLocks()).thenReturn(false);
Expand All @@ -1560,7 +1563,8 @@ void testPushEngine_happyPath_noFileLocks(@TempDir Path testDir) throws Exceptio
MockedStatic<SmssUtilities> smssMock = mockStatic(SmssUtilities.class);
MockedStatic<EngineUtility> euMock = mockStatic(EngineUtility.class);
MockedStatic<EngineSyncUtility> syncMock = mockStatic(EngineSyncUtility.class);
MockedStatic<ClusterUtil> clusterMock = mockStatic(ClusterUtil.class)) {
MockedStatic<ClusterUtil> clusterMock = mockStatic(ClusterUtil.class);
MockedStatic<DIHelper> diMock = mockStatic(DIHelper.class)) {

utilMock.when(() -> Utility.getEngine("eng1", false)).thenReturn(mockEngine);
utilMock.when(() -> Utility.normalizePath(anyString())).thenAnswer(inv -> inv.getArgument(0));
Expand All @@ -1570,13 +1574,87 @@ void testPushEngine_happyPath_noFileLocks(@TempDir Path testDir) throws Exceptio
euMock.when(() -> EngineUtility.getLocalEngineBaseDirectory(IEngine.CATALOG_TYPE.DATABASE))
.thenReturn(testDir.toString());
syncMock.when(() -> EngineSyncUtility.getEngineLock("eng1")).thenReturn(lock);
DIHelper mockDI = mock(DIHelper.class);
diMock.when(DIHelper::getInstance).thenReturn(mockDI);
when(mockDI.getEngineProperty(Constants.ENGINES)).thenReturn("eng1");
when(mockDI.getProjectProperty(Constants.PROJECTS)).thenReturn("proj1");

instance.pushEngine("eng1");

verify(mockStorageEngine).syncLocalToStorage(anyString(), anyString(), any());
InOrder closeBeforeSync = inOrder(mockEngine, mockStorageEngine);
closeBeforeSync.verify(mockEngine).close();
closeBeforeSync.verify(mockStorageEngine).syncLocalToStorage(anyString(), anyString(), any());
verify(mockStorageEngine).copyToStorage(argThat(s -> s.contains("MyEngine__eng1.smss")),
argThat(s -> s.contains(CentralCloudStorage.SMSS_POSTFIX)), any());
utilMock.verify(() -> Utility.getEngine("eng1", false), times(2));
assertFalse(lock.isLocked());
}
}

@Test
void testPushEngine_nonDatabaseWithoutFileLocks_staysOpen(@TempDir Path testDir) throws Exception {
IEngine mockEngine = mock(IEngine.class);
when(mockEngine.getCatalogType()).thenReturn(IEngine.CATALOG_TYPE.MODEL);
when(mockEngine.holdsFileLocks()).thenReturn(false);
ReentrantLock lock = new ReentrantLock();

try (MockedStatic<Utility> utilMock = mockStatic(Utility.class);
MockedStatic<SecurityEngineUtils> secMock = mockStatic(SecurityEngineUtils.class);
MockedStatic<SmssUtilities> smssMock = mockStatic(SmssUtilities.class);
MockedStatic<EngineUtility> euMock = mockStatic(EngineUtility.class);
MockedStatic<EngineSyncUtility> syncMock = mockStatic(EngineSyncUtility.class);
MockedStatic<ClusterUtil> clusterMock = mockStatic(ClusterUtil.class)) {

utilMock.when(() -> Utility.getEngine("eng1", false)).thenReturn(mockEngine);
utilMock.when(() -> Utility.normalizePath(anyString())).thenAnswer(inv -> inv.getArgument(0));
secMock.when(() -> SecurityEngineUtils.getEngineAliasForId("eng1")).thenReturn("MyEngine");
smssMock.when(() -> SmssUtilities.getUniqueName("MyEngine", "eng1")).thenReturn("MyEngine__eng1");
euMock.when(() -> EngineUtility.getLocalEngineBaseDirectory(IEngine.CATALOG_TYPE.MODEL))
.thenReturn(testDir.toString());
syncMock.when(() -> EngineSyncUtility.getEngineLock("eng1")).thenReturn(lock);

instance.pushEngine("eng1");

verify(mockEngine, never()).close();
utilMock.verify(() -> Utility.getEngine("eng1", false), times(1));
assertFalse(lock.isLocked());
}
}

@Test
void testPushEngine_databaseReopensAndUnlocksWhenTransferFails(@TempDir Path testDir) throws Exception {
IEngine mockEngine = mock(IEngine.class);
when(mockEngine.getCatalogType()).thenReturn(IEngine.CATALOG_TYPE.DATABASE);
when(mockEngine.holdsFileLocks()).thenReturn(false);
when(mockStorageEngine.syncLocalToStorage(anyString(), anyString(), isNull()))
.thenThrow(new IOException("simulated transfer failure"));
ReentrantLock lock = new ReentrantLock();

try (MockedStatic<Utility> utilMock = mockStatic(Utility.class);
MockedStatic<SecurityEngineUtils> secMock = mockStatic(SecurityEngineUtils.class);
MockedStatic<SmssUtilities> smssMock = mockStatic(SmssUtilities.class);
MockedStatic<EngineUtility> euMock = mockStatic(EngineUtility.class);
MockedStatic<EngineSyncUtility> syncMock = mockStatic(EngineSyncUtility.class);
MockedStatic<ClusterUtil> clusterMock = mockStatic(ClusterUtil.class);
MockedStatic<DIHelper> diMock = mockStatic(DIHelper.class)) {

utilMock.when(() -> Utility.getEngine("eng1", false)).thenReturn(mockEngine);
utilMock.when(() -> Utility.normalizePath(anyString())).thenAnswer(inv -> inv.getArgument(0));
secMock.when(() -> SecurityEngineUtils.getEngineAliasForId("eng1")).thenReturn("MyEngine");
smssMock.when(() -> SmssUtilities.getUniqueName("MyEngine", "eng1")).thenReturn("MyEngine__eng1");
euMock.when(() -> EngineUtility.getLocalEngineBaseDirectory(IEngine.CATALOG_TYPE.DATABASE))
.thenReturn(testDir.toString());
syncMock.when(() -> EngineSyncUtility.getEngineLock("eng1")).thenReturn(lock);
DIHelper mockDI = mock(DIHelper.class);
diMock.when(DIHelper::getInstance).thenReturn(mockDI);
when(mockDI.getEngineProperty(Constants.ENGINES)).thenReturn("eng1");
when(mockDI.getProjectProperty(Constants.PROJECTS)).thenReturn("proj1");

assertThrows(IOException.class, () -> instance.pushEngine("eng1"));

verify(mockEngine).close();
utilMock.verify(() -> Utility.getEngine("eng1", false), times(2));
assertFalse(lock.isLocked());
}
}

Expand Down
Loading