From 54f6be27177ef90ebd0542d821f08ab6cb626309 Mon Sep 17 00:00:00 2001 From: fnachabe <111369875+fnachabe@users.noreply.github.com> Date: Thu, 3 Sep 2026 10:07:25 -0400 Subject: [PATCH] Adding locking to prevent race condition --- .../central_cloud_storage.md | 9 +- .../util/clients/CentralCloudStorage.java | 14 +++- .../clients/CentralCloudStorageUnitTests.java | 84 ++++++++++++++++++- 3 files changed, 100 insertions(+), 7 deletions(-) diff --git a/docs/cloud_and_cluster/central_cloud_storage.md b/docs/cloud_and_cluster/central_cloud_storage.md index f22a22f9f67..2a7b742b0c4 100644 --- a/docs/cloud_and_cluster/central_cloud_storage.md +++ b/docs/cloud_and_cluster/central_cloud_storage.md @@ -36,9 +36,10 @@ This class is implemented as a singleton and provides the core logic for interac * `pushEngine(String engineId)`: * Takes an `engineId`. * Retrieves the local engine's folder (e.g., `SEMOSS_HOME/db/__/`) and its `.smss` file. - * If the engine `holdsFileLocks()`, it's temporarily closed and removed from `DIHelper`'s cache. + * Database engines are temporarily closed and removed from `DIHelper`'s cache before their folders are uploaded. Other engine types are closed when `holdsFileLocks()` is `true`. * Syncs the entire local engine folder to a corresponding path in the central storage (e.g., `//`). * Copies the local `.smss` file to a separate location in central storage (e.g., `/-smss/`). + * Reopens an engine that was closed for synchronization, including when the transfer fails. * Uses `EngineSyncUtility.getEngineLock(engineId)` to ensure exclusive access during the operation. * `pullEngine(String engineId, IEngine.CATALOG_TYPE engineType, boolean engineAlreadyLoaded)`: * Downloads the engine's folder and `.smss` file from the central store to the corresponding local SEMOSS directories. @@ -67,6 +68,12 @@ This class is implemented as a singleton and provides the core logic for interac * To prevent race conditions and ensure data integrity in a multi-instance environment, `CentralCloudStorage` uses `java.util.concurrent.locks.ReentrantLock`. * `EngineSyncUtility.getEngineLock(engineId)` and `ProjectSyncUtility.getProjectLock(projectId)` provide static methods to obtain a lock specific to an engine ID or project ID. This ensures that operations on the same asset are serialized across different threads or even different SEMOSS instances (if the lock mechanism is cluster-aware, though its direct cluster-awareness isn't detailed in this class itself; ZooKeeper likely handles distributed locking aspects). +#### Database audit files + +Every database engine can own a local modification-audit database in its engine folder, such as `audit_log_database.mv.db`. This remains true for an external database whose primary data and connection are remote. An external database can therefore return `false` from `holdsFileLocks()` while its local H2 audit database still has an active background writer. + +`pushEngine` closes all database engines before synchronizing their folders. Closing the engine flushes and closes the local audit database so the storage provider reads a stable file. Uploading a live H2 MVStore file is unsafe: compaction can shorten the file after an uploader records its original length, which can cause a zero-progress chunked upload and block other storage requests. The engine is reloaded in the `finally` path so both successful and failed transfers restore normal availability. + ### Interaction with Cluster Synchronizer (ZooKeeper) * The constructor of `CentralCloudStorage` conditionally initializes `ClusterSynchronizer.getInstance()` if `ClusterUtil.IS_CLUSTER_ZK` is true. 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()); } }