diff --git a/src/prerna/cluster/util/clients/CentralCloudStorage.java b/src/prerna/cluster/util/clients/CentralCloudStorage.java index e6fc2ca0cdf..0425b7f17de 100644 --- a/src/prerna/cluster/util/clients/CentralCloudStorage.java +++ b/src/prerna/cluster/util/clients/CentralCloudStorage.java @@ -526,9 +526,17 @@ 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(); } @@ -536,8 +544,8 @@ public void pushEngine(String engineId) throws IOException, InterruptedException 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) { diff --git a/test/prerna/cluster/util/clients/CentralCloudStorageUnitTests.java b/test/prerna/cluster/util/clients/CentralCloudStorageUnitTests.java index 8d7902563f2..51f6b95e363 100644 --- a/test/prerna/cluster/util/clients/CentralCloudStorageUnitTests.java +++ b/test/prerna/cluster/util/clients/CentralCloudStorageUnitTests.java @@ -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; @@ -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; @@ -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; @@ -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); @@ -1560,7 +1563,8 @@ void testPushEngine_happyPath_noFileLocks(@TempDir Path testDir) throws Exceptio MockedStatic smssMock = mockStatic(SmssUtilities.class); MockedStatic euMock = mockStatic(EngineUtility.class); MockedStatic syncMock = mockStatic(EngineSyncUtility.class); - MockedStatic clusterMock = mockStatic(ClusterUtil.class)) { + MockedStatic clusterMock = mockStatic(ClusterUtil.class); + MockedStatic diMock = mockStatic(DIHelper.class)) { utilMock.when(() -> Utility.getEngine("eng1", false)).thenReturn(mockEngine); utilMock.when(() -> Utility.normalizePath(anyString())).thenAnswer(inv -> inv.getArgument(0)); @@ -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 utilMock = mockStatic(Utility.class); + MockedStatic secMock = mockStatic(SecurityEngineUtils.class); + MockedStatic smssMock = mockStatic(SmssUtilities.class); + MockedStatic euMock = mockStatic(EngineUtility.class); + MockedStatic syncMock = mockStatic(EngineSyncUtility.class); + MockedStatic 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 utilMock = mockStatic(Utility.class); + MockedStatic secMock = mockStatic(SecurityEngineUtils.class); + MockedStatic smssMock = mockStatic(SmssUtilities.class); + MockedStatic euMock = mockStatic(EngineUtility.class); + MockedStatic syncMock = mockStatic(EngineSyncUtility.class); + MockedStatic clusterMock = mockStatic(ClusterUtil.class); + MockedStatic 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()); } }