[vfs] IDEA-350058: protect PersistentFSRecords from concurrent access by another process

+ There are EA reports that could be explained by VFS storages being opened and used from >1 process. We have a protection from >1 IDE instance running -- but the protection is not 100% reliable, there are examples of >1 IDE instances running => VFS need it's own protection mechanism

GitOrigin-RevId: a784627cc5db470c09f69285b92c6bfc87ac36d0

platform/platform-impl/src/com/intellij/openapi/vfs/newvfs/persistent/FSRecordsImpl.java

@@ -241,8 +241,8 @@ public final class FSRecordsImpl implements Closeable {
   }
 
   public static int currentImplementationVersion() {
-    //bumped main version (60 -> 61) because of PFSRecords header enlargement (HEADER_ERRORS_ACCUMULATED)
-    final int mainVFSFormatVersion = 61;
+    //bumped main version (61 -> 62) because of records.connectionStatus values changed
+    final int mainVFSFormatVersion = 62;
     //@formatter:off (nextMask better be aligned)
     return nextMask(mainVFSFormatVersion + (PersistentFSRecordsStorageFactory.storageImplementation().getId()), /* acceptable range is [0..255] */ 8,
            nextMask(!USE_CONTENT_STORAGE_OVER_MMAPPED_FILE,  //former USE_CONTENT_HASHES=true, this is why negation

platform/platform-impl/src/com/intellij/openapi/vfs/newvfs/persistent/IPersistentFSRecordsStorage.java

@@ -1,4 +1,4 @@
-// Copyright 2000-2022 JetBrains s.r.o. and contributors. Use of this source code is governed by the Apache 2.0 license.
+// Copyright 2000-2024 JetBrains s.r.o. and contributors. Use of this source code is governed by the Apache 2.0 license.
 package com.intellij.openapi.vfs.newvfs.persistent;
 
 import com.intellij.openapi.Forceable;
@@ -97,7 +97,7 @@ public interface IPersistentFSRecordsStorage extends Forceable, AutoCloseable {
   interface HeaderForRead {
     long getTimestamp() throws IOException;
 
-    int getConnectionStatus() throws IOException;
+    //TODO boolean wasClosedProperly() throws IOException;
 
     int getVersion() throws IOException;
 
@@ -107,8 +107,6 @@ public interface IPersistentFSRecordsStorage extends Forceable, AutoCloseable {
   }
 
   interface HeaderForUpdate extends HeaderForRead {
-    void setConnectionStatus(final int code) throws IOException;
-
     void setVersion(final int version) throws IOException;
 
     //TODO void setErrorsAccumulated(final int errors) throws IOException;

platform/platform-impl/src/com/intellij/openapi/vfs/newvfs/persistent/PersistentFSConnection.java

@@ -215,7 +215,6 @@ public final class PersistentFSConnection {
   void markDirty() throws IOException {
     if (!dirty) {
       dirty = true;
-      records.setConnectionStatus(PersistentFSHeaders.CONNECTED_MAGIC);
     }
   }
 
@@ -253,7 +252,6 @@ public final class PersistentFSConnection {
       return;
     }
 
-    records.setConnectionStatus(PersistentFSHeaders.SAFELY_CLOSED_MAGIC);
     doForce();
 
     //ensure async loading is finished

platform/platform-impl/src/com/intellij/openapi/vfs/newvfs/persistent/PersistentFSHeaders.java

@@ -1,4 +1,4 @@
-// Copyright 2000-2023 JetBrains s.r.o. and contributors. Use of this source code is governed by the Apache 2.0 license.
+// Copyright 2000-2024 JetBrains s.r.o. and contributors. Use of this source code is governed by the Apache 2.0 license.
 package com.intellij.openapi.vfs.newvfs.persistent;
 
 import org.intellij.lang.annotations.MagicConstant;
@@ -23,9 +23,8 @@ final class PersistentFSHeaders {
 
   //CONNECTION_STATUS header field values:
   //@formatter:off
-  static final int CONNECTED_MAGIC      = 0x12ad34e4;
-  static final int SAFELY_CLOSED_MAGIC  = 0x1f2f3f4f;
-  static final int CORRUPTED_MAGIC      = 0xabcf7f7f;
+  static final int IN_USE_STAMP         = 0x12ad34e4;
+  static final int SAFELY_CLOSED_STAMP  = 0;
   //@formatter:on
 
   @MagicConstant(flagsFromClass = PersistentFSHeaders.class)

platform/platform-impl/src/com/intellij/openapi/vfs/newvfs/persistent/PersistentFSLoader.java

@@ -426,8 +426,8 @@ public final class PersistentFSLoader {
     //So the tradeoff: we use few heuristics to quickly check for the most likely signs of corruption,
     // and if we find any such sign -- switch to more vigilant checking:
 
-    if (recordsStorage.getConnectionStatus() != PersistentFSHeaders.SAFELY_CLOSED_MAGIC) {
-      addProblem(NOT_CLOSED_PROPERLY, "VFS wasn't safely shut down: records.connectionStatus != SAFELY_CLOSED");
+    if (!recordsStorage.wasClosedProperly()) {
+      addProblem(NOT_CLOSED_PROPERLY, "VFS wasn't safely shut down: records.wasClosedProperly is false");
     }
     int errorsAccumulated = recordsStorage.getErrorsAccumulated();
     if (errorsAccumulated > 0) {
@@ -779,7 +779,7 @@ public final class PersistentFSLoader {
   public @NotNull PersistentFSRecordsStorage createRecordsStorage(@NotNull Path recordsFile) throws IOException {
     StorageFactory<PersistentFSRecordsStorage> recordsStorageFactory = PersistentFSRecordsStorageFactory.storageImplementation();
 
-    LOG.trace("VFS uses " + recordsStorageFactory + " storage for main file records table");
+    LOG.info("VFS uses " + recordsStorageFactory + " storage for main file records table");
     return recordsStorageFactory.wrapStorageSafely(recordsFile, records -> {
       if (vfsLog != null) {
         var recordsInterceptors = vfsLog.getConnectionInterceptors().stream()
@@ -819,7 +819,6 @@ public final class PersistentFSLoader {
     records.setVersion(version);
     attributes.setVersion(version);
     contents.setVersion(version);
-    records.setConnectionStatus(PersistentFSHeaders.SAFELY_CLOSED_MAGIC);
   }
 
   private static void makeBestEffortToCleanStorage(@Nullable Object storage,

platform/platform-impl/src/com/intellij/openapi/vfs/newvfs/persistent/PersistentFSRecordsLockFreeOverMMappedFile.java

@@ -48,10 +48,18 @@ public final class PersistentFSRecordsLockFreeOverMMappedFile implements Persist
    * Instead, we store allocated records count in header, in a reserved field (HEADER_RESERVED_OFFSET_1)
    */
   private static final int HEADER_RECORDS_ALLOCATED = HEADER_RESERVED_OFFSET_1;
+  /**
+   * ConnectionStatus header field re-purposed to keep owner process pid instead of magic numbers in {@link PersistentFSHeaders},
+   * renamed to emphasise that new purpose.
+   */
+  private static final int OWNER_PROCESS_ID_OFFSET = HEADER_CONNECTION_STATUS_OFFSET;
+
+  public static final int NULL_OWNER_PID = 0;
 
   @VisibleForTesting
   static final int HEADER_SIZE = PersistentFSHeaders.HEADER_SIZE;
 
+
   @VisibleForTesting
   static final class RecordLayout {
     //@formatter:off
@@ -121,6 +129,10 @@ public final class PersistentFSRecordsLockFreeOverMMappedFile implements Persist
    */
   private int cachedMaxAllocatedId;
 
+  private final boolean wasClosedProperly;
+
+  private volatile int owningProcessId = 0;
+
   public PersistentFSRecordsLockFreeOverMMappedFile(@NotNull MMappedFileStorage storage) throws IOException {
     final int pageSize = storage.pageSize();
     if (pageSize < HEADER_SIZE) {
@@ -144,6 +156,9 @@ public final class PersistentFSRecordsLockFreeOverMMappedFile implements Persist
     }
 
     cachedMaxAllocatedId = maxAllocatedID();
+
+    int ownerProcessId = getIntHeaderField(OWNER_PROCESS_ID_OFFSET);
+    wasClosedProperly = (ownerProcessId == NULL_OWNER_PID);
   }
 
   @Override
@@ -336,11 +351,6 @@ public final class PersistentFSRecordsLockFreeOverMMappedFile implements Persist
       return records.getTimestamp();
     }
 
-    @Override
-    public int getConnectionStatus() throws IOException {
-      return records.getConnectionStatus();
-    }
-
     @Override
     public int getVersion() throws IOException {
       return records.getVersion();
@@ -351,11 +361,6 @@ public final class PersistentFSRecordsLockFreeOverMMappedFile implements Persist
       return records.getGlobalModCount();
     }
 
-    @Override
-    public void setConnectionStatus(final int code) throws IOException {
-      records.setConnectionStatus(code);
-    }
-
     @Override
     public void setVersion(final int version) throws IOException {
       records.setVersion(version);
@@ -578,14 +583,80 @@ public final class PersistentFSRecordsLockFreeOverMMappedFile implements Persist
   }
 
   @Override
-  public void setConnectionStatus(final int connectionStatus) {
-    setIntHeaderField(HEADER_CONNECTION_STATUS_OFFSET, connectionStatus);
-    dirty.compareAndSet(false, true);
+  public boolean wasClosedProperly() throws IOException {
+    return wasClosedProperly;
   }
 
-  @Override
-  public int getConnectionStatus() {
-    return getIntHeaderField(HEADER_CONNECTION_STATUS_OFFSET);
+
+  /**
+   * Tries to acquire an exclusive ownership over the storage, for the process identified by acquiringProcessId.
+   * Storages based on memory mapped files _could_ be used from >1 process concurrently -- but current implementation
+   * of those storages is not designed for such a use -- and there is no legit scenarios of such a co-use for today.
+   * Hence we need a way to protect the storage(s) from multi-process access. This method provides a way to acquire
+   * 'ownership' of a storage for given process (identified by integer pid) -- implementation ensures that in a concurrent
+   * settings only a single process could successfully acquire ownership (see below definition of 'successful').
+   * <p>
+   * BEWARE: storage provides a way to acquire ownership, but storage does NOT ensure that only the owner process
+   * accesses it. It is the responsibility of the application to check (i.e. try acquiring) the ownership early,
+   * before starting any other operations with the storage -- and step back if ownership is already acquired by
+   * another process.
+   *
+   * @param acquiringProcessId must be != 0
+   * @return pid of the exclusive owner after the call.
+   * If [returned pid == acquiringProcessId] => acquiring has succeeded, and acquiringProcessId is the new exclusive owner.
+   * Otherwise: returned pid provides an information about who is the storage exclusive owner now.
+   */
+  public int tryAcquireExclusiveAccess(int acquiringProcessId,
+                                       boolean forcibly) {
+    if (acquiringProcessId == NULL_OWNER_PID) {
+      throw new IllegalArgumentException("acquiringPid(=" + acquiringProcessId + ") must be !=0");
+    }
+    ByteBuffer headerPageBuffer = headerPage().rawPageBuffer();
+    while (true) {//CAS loop
+      int currentOwnerProcessId = (int)INT_HANDLE.getVolatile(headerPageBuffer, OWNER_PROCESS_ID_OFFSET);
+      if (currentOwnerProcessId == acquiringProcessId) {
+        owningProcessId = acquiringProcessId;
+        return acquiringProcessId; //already acquired => nothing to do
+      }
+      if (currentOwnerProcessId != NULL_OWNER_PID && !forcibly) {
+        //already acquired by other process => return owner's pid as an indication of failure
+        return currentOwnerProcessId;
+      }
+      if (INT_HANDLE.compareAndSet(headerPageBuffer, OWNER_PROCESS_ID_OFFSET, currentOwnerProcessId, acquiringProcessId)) {
+        owningProcessId = acquiringProcessId;
+        dirty.compareAndSet(false, true);
+        return acquiringProcessId;
+      }
+    }
+  }
+
+  /**
+   * Releases the ownership acquired by {@link #tryAcquireExclusiveAccess(int, boolean)}.
+   * If {@code tryAcquireExclusiveAccess(pid)} has succeeded, than following {@code tryReleaseExclusiveAccess(pid)} must
+   * also succeed (given nobody forcibly overwrites the ownership by calling {@link #tryAcquireExclusiveAccess(int, boolean)}
+   * with forcible=true).
+   *
+   * @param ownerProcessId pid of current storage owner. Could be 0 if there is no current owner
+   * @return pid of exclusive owner after the call.
+   * If returned pid=0 => there is no owner, i.e. ownership was released successfully.
+   * Otherwise: returned pid provides an information about who is currently owning the storage.
+   */
+  private int tryReleaseExclusiveAccess(int ownerProcessId) {
+    ByteBuffer headerPageBuffer = headerPage().rawPageBuffer();
+    while (true) {//CAS loop
+      int currentOwnerProcessId = (int)INT_HANDLE.getVolatile(headerPageBuffer, OWNER_PROCESS_ID_OFFSET);
+      if (currentOwnerProcessId == NULL_OWNER_PID) {
+        return NULL_OWNER_PID;//nothing to do (method expected to be idempotent)
+      }
+      if (currentOwnerProcessId != ownerProcessId) {
+        //acquired by another process => return its pid as an indication of failure
+        return currentOwnerProcessId;
+      }
+      if (INT_HANDLE.compareAndSet(headerPageBuffer, OWNER_PROCESS_ID_OFFSET, currentOwnerProcessId, NULL_OWNER_PID)) {
+        dirty.compareAndSet(false, true);
+        return 0;
+      }
+    }
   }
 
   @Override
@@ -674,9 +745,20 @@ public final class PersistentFSRecordsLockFreeOverMMappedFile implements Persist
 
   @Override
   public void close() throws IOException {
-    force();
-    storage.close();
-    headerPage = null;
+    if (storage.isOpen()) {
+      int ourPid = owningProcessId;
+      int currentOwnerPid = tryReleaseExclusiveAccess(ourPid);
+
+      force();
+      storage.close();
+
+      headerPage = null;
+
+      if (currentOwnerPid != NULL_OWNER_PID) {
+        throw new IOException(
+          "Storage is exclusively owned by another process[pid: " + currentOwnerPid + ", our pid: " + ourPid + "]");
+      }
+    }
   }
 
   @Override

platform/platform-impl/src/com/intellij/openapi/vfs/newvfs/persistent/PersistentFSRecordsOverLockFreePagedStorage.java

@@ -69,6 +69,7 @@ public final class PersistentFSRecordsOverLockFreePagedStorage implements Persis
 
   private final transient HeaderAccessor headerAccessor = new HeaderAccessor(this);
 
+  private final boolean wasClosedProperly;
 
   public PersistentFSRecordsOverLockFreePagedStorage(final @NotNull PagedFileStorageWithRWLockedPageContent storage) throws IOException {
     this.storage = storage;
@@ -80,6 +81,17 @@ public final class PersistentFSRecordsOverLockFreePagedStorage implements Persis
     allocatedRecordsCount.set(recordsCountInStorage);
 
     globalModCount.set(getIntHeaderField(HEADER_GLOBAL_MOD_COUNT_OFFSET));
+
+    int connectionStatus = getIntHeaderField(HEADER_CONNECTION_STATUS_OFFSET);
+    wasClosedProperly = (connectionStatus == SAFELY_CLOSED_STAMP);
+    setIntHeaderField(HEADER_CONNECTION_STATUS_OFFSET, IN_USE_STAMP);
+    force();//CONNECTION_STATUS change makes file dirty, but it is unexpected to have freshly opened storage dirty => flush it
+    //MAYBE RC: flushing on storage open is also not good for (startup) performance. Instead, we could update the position in
+    //          a page with raw update, without notifying storage of changes, which tricks storage think it is !dirty.
+    //          Normally this calls for a bug, but in this particular case flushing connection status change has no sense until
+    //          some _other_, actual change(s) happened -- i.e. if connection status change is the _only_ change happened, it is
+    //          perfectly fine to lose it, i.e. not store it on .force()/.close() -- we must store connection status only if
+    //          some other change(s) happened.
   }
 
   @VisibleForTesting
@@ -351,11 +363,6 @@ public final class PersistentFSRecordsOverLockFreePagedStorage implements Persis
       return records.getTimestamp();
     }
 
-    @Override
-    public int getConnectionStatus() throws IOException {
-      return records.getConnectionStatus();
-    }
-
     @Override
     public int getVersion() throws IOException {
       return records.getVersion();
@@ -366,11 +373,6 @@ public final class PersistentFSRecordsOverLockFreePagedStorage implements Persis
       return records.getGlobalModCount();
     }
 
-    @Override
-    public void setConnectionStatus(final int code) throws IOException {
-      records.setConnectionStatus(code);
-    }
-
     @Override
     public void setVersion(final int version) throws IOException {
       records.setVersion(version);
@@ -670,14 +672,8 @@ public final class PersistentFSRecordsOverLockFreePagedStorage implements Persis
   }
 
   @Override
-  public void setConnectionStatus(final int connectionStatus) throws IOException {
-    setIntHeaderField(HEADER_CONNECTION_STATUS_OFFSET, connectionStatus);
-    //intentionally don't increment globalModCount
-  }
-
-  @Override
-  public int getConnectionStatus() throws IOException {
-    return getIntHeaderField(HEADER_CONNECTION_STATUS_OFFSET);
+  public boolean wasClosedProperly() throws IOException {
+    return wasClosedProperly;
   }
 
   @Override
@@ -730,6 +726,8 @@ public final class PersistentFSRecordsOverLockFreePagedStorage implements Persis
   @Override
   public void close() throws IOException {
     if (!storage.isClosed()) {
+      setIntHeaderField(HEADER_CONNECTION_STATUS_OFFSET, SAFELY_CLOSED_STAMP);
+      
       force();
       storage.close();
     }

platform/platform-impl/src/com/intellij/openapi/vfs/newvfs/persistent/PersistentFSRecordsStorage.java

@@ -66,7 +66,8 @@ public interface PersistentFSRecordsStorage extends CleanableStorage, AutoClosea
 
   boolean setContentRecordId(int fileId, int recordId) throws IOException;
 
-  @PersistentFS.Attributes int getFlags(int fileId) throws IOException;
+  @PersistentFS.Attributes
+  int getFlags(int fileId) throws IOException;
 
   /**
    * Fills all record fields in one shot.
@@ -91,9 +92,13 @@ public interface PersistentFSRecordsStorage extends CleanableStorage, AutoClosea
 
   long getTimestamp() throws IOException;
 
-  void setConnectionStatus(int code) throws IOException;
-
-  int getConnectionStatus() throws IOException;
+  /**
+   * @return true if storage was closed properly (i.e. by {@link #close()} in a last session, or false if last session was
+   * finished without calling {@link #close()} -- storage content may be inconsistent or corrupted.
+   * The property describes events in a _previous_ session, so it is immutable during current session: changes to storage
+   * doesn't affect this property's value
+   */
+  boolean wasClosedProperly() throws IOException;
 
   int getErrorsAccumulated() throws IOException;
 

platform/platform-impl/src/com/intellij/openapi/vfs/newvfs/persistent/PersistentFSRecordsStorageFactory.kt

@@ -1,10 +1,7 @@
 // Copyright 2000-2024 JetBrains s.r.o. and contributors. Use of this source code is governed by the Apache 2.0 license.
 package com.intellij.openapi.vfs.newvfs.persistent
 
-import com.intellij.util.io.IOUtil
-import com.intellij.util.io.PageCacheUtils
-import com.intellij.util.io.PagedFileStorageWithRWLockedPageContent
-import com.intellij.util.io.StorageLockContext
+import com.intellij.util.io.*
 import com.intellij.util.io.dev.StorageFactory
 import com.intellij.util.io.dev.mmapped.MMappedFileStorageFactory
 import com.intellij.util.io.pagecache.impl.PageContentLockingStrategy
@@ -12,19 +9,50 @@ import org.jetbrains.annotations.ApiStatus.Internal
 import org.jetbrains.annotations.VisibleForTesting
 import java.io.IOException
 import java.nio.file.Path
+import kotlin.jvm.optionals.getOrNull
 
 @Internal
 abstract class PersistentFSRecordsStorageFactory(val id: Int) : StorageFactory<PersistentFSRecordsStorage> {
 
 
   /** Currently the default impl */
-  data class OverMMappedFile(val pageSize: Int = PersistentFSRecordsLockFreeOverMMappedFile.DEFAULT_MAPPED_CHUNK_SIZE)
+  data class OverMMappedFile(val pageSize: Int = PersistentFSRecordsLockFreeOverMMappedFile.DEFAULT_MAPPED_CHUNK_SIZE,
+                             val acquireStorageOwnership: Boolean = true)
     : PersistentFSRecordsStorageFactory(id = 0) {
 
     override fun open(storagePath: Path): PersistentFSRecordsLockFreeOverMMappedFile = MMappedFileStorageFactory.withDefaults()
       .pageSize(pageSize)
       .wrapStorageSafely<PersistentFSRecordsLockFreeOverMMappedFile, IOException>(storagePath) { mappedFileStorage ->
-        PersistentFSRecordsLockFreeOverMMappedFile(mappedFileStorage)
+        val recordsStorage = PersistentFSRecordsLockFreeOverMMappedFile(mappedFileStorage)
+
+        if (acquireStorageOwnership) {
+          val ownPid = ProcessHandle.current().pid().toInt()
+          val acquiredByPid = recordsStorage.tryAcquireExclusiveAccess(ownPid, /*forcibly: */ false)
+          if (acquiredByPid != ownPid) {
+            val ownerProcess = ProcessHandle.of(acquiredByPid.toLong()).getOrNull()
+            if (ownerProcess != null && ownerProcess.isAlive) {
+              //OSes re-use process ids, so false positives (pid collision) are possible.
+              //But this branch is reached only after improper shutdown, and we can't 100% guarantee VFS recovery after such
+              // a shutdowns anyway. So pid collision is just another case of bad luck -- it won't be a recovery, but a VFS
+              // rebuild.
+              //MAYBE RC: we could do a bit better by checking (how?) that an ownerProcess runs JVM -- this should greatly
+              //          reduce collision probability
+              throw StorageAlreadyInUseException("Records storage [$storagePath] is in use by another process [pid: $acquiredByPid, info: ${ownerProcess.info()}]")
+            }
+
+            //Previous owner process is terminated, i.e. storage wasn't closed properly on an app termination (crash?)
+            // => acquire storage forcibly
+            val acquiredByPidForcibly = recordsStorage.tryAcquireExclusiveAccess(ownPid, /*forcibly: */ true)
+            if (acquiredByPidForcibly != ownPid) {
+              //yet another process acquired records concurrently => whole scenario looks too damned, just fail:
+              val concurrentlyOwnedProcess = ProcessHandle.of(acquiredByPidForcibly.toLong()).getOrNull()
+              throw StorageAlreadyInUseException("Records storage [$storagePath] is in use by another process [pid: $acquiredByPidForcibly, info: ${concurrentlyOwnedProcess?.info()}]")
+            }
+            FSRecords.LOG.warn("Records storage [$storagePath] was in use by process [pid: $acquiredByPid] which is not exist now (wasn't closed properly/crashed?) -> re-acquired forcibly")
+          }
+        }
+
+        recordsStorage
       }
   }
 

platform/platform-impl/src/com/intellij/openapi/vfs/newvfs/persistent/PersistentInMemoryFSRecordsStorage.java

@@ -29,6 +29,7 @@ import static java.nio.file.StandardOpenOption.WRITE;
  * Intended for use as a reference implementation, to compare other impls against
  * (e.g. by performance)
  */
+//TODO RC: rename to PersistentFSRecordsOverInMemoryStorage
 @ApiStatus.Internal
 @TestOnly
 public final class PersistentInMemoryFSRecordsStorage implements PersistentFSRecordsStorage {
@@ -279,13 +280,8 @@ public final class PersistentInMemoryFSRecordsStorage implements PersistentFSRec
   }
 
   @Override
-  public void setConnectionStatus(final int connectionStatus) throws IOException {
-    setIntHeaderField(HEADER_CONNECTION_STATUS_OFFSET, connectionStatus);
-  }
-
-  @Override
-  public int getConnectionStatus() throws IOException {
-    return getIntHeaderField(HEADER_CONNECTION_STATUS_OFFSET);
+  public boolean wasClosedProperly() throws IOException {
+    return true;// 'previous session' make no sense for non-persistent in-memory storage
   }
 
   @Override

platform/platform-tests/testSrc/com/intellij/openapi/vfs/newvfs/persistent/PersistentFSRecordsStorageLockFreeOverMMappedFileTest.java

@@ -82,4 +82,59 @@ public class PersistentFSRecordsStorageLockFreeOverMMappedFileTest
     }
   }
 
+  @Test
+  public void tryAcquireExclusiveAccess_alwaysSucceedWithoutConcurrentRequests() {
+    int currentPid = 123;
+    int ownerPid = storage.tryAcquireExclusiveAccess(currentPid, false);
+    assertEquals(
+      "Acquire must be successful since there no other owners",
+      currentPid,
+      ownerPid
+    );
+  }
+
+  @Test
+  public void ifStorageIsAcquired_acquireWithDifferentPid_MustFailToChangeOwner() {
+    int currentPid = 123;
+    int competingPid = 124;
+    storage.tryAcquireExclusiveAccess(currentPid, false);
+    int ownerPid = storage.tryAcquireExclusiveAccess(competingPid, false);
+    assertEquals(
+      "If storage is already acquired, acquire must fail, and owner must not change",
+      currentPid,
+      ownerPid
+    );
+  }
+
+
+  @Test
+  public void processAlreadyAcquiredStorage_alwaysSucceedInAcquiringAgain() {
+    int currentPid = 123;
+    storage.tryAcquireExclusiveAccess(currentPid, false);
+    int ownerPid = storage.tryAcquireExclusiveAccess(currentPid, false);
+    assertEquals(
+      "Same process could always acquire storage again (i.e. acquire is idempotent)",
+      currentPid,
+      ownerPid
+    );
+  }
+
+  @Test
+  public void reopenedStorageHasNoOwner_henceCouldBeAcquired_ByAnyProcess() throws IOException {
+    int firstOwnerPid = 123;
+    storage.tryAcquireExclusiveAccess(firstOwnerPid, false);
+
+    storage.close();
+    storage = openStorage(storagePath);
+
+    int secondOwnerPid = 125;
+    int ownerPid = storage.tryAcquireExclusiveAccess(secondOwnerPid, false);
+    assertEquals(
+      ".close() clears the owner, so any process could acquire the reopened storage",
+      secondOwnerPid,
+      ownerPid
+    );
+
+  }
+
 }

platform/platform-tests/testSrc/com/intellij/openapi/vfs/newvfs/persistent/PersistentFSRecordsStorageTestBase.java

@@ -21,7 +21,6 @@ import java.util.concurrent.*;
 import java.util.stream.IntStream;
 
 import static com.intellij.openapi.vfs.newvfs.persistent.InvertedNameIndex.NULL_NAME_ID;
-import static com.intellij.openapi.vfs.newvfs.persistent.PersistentFSHeaders.CONNECTED_MAGIC;
 import static java.util.Comparator.comparing;
 import static java.util.concurrent.TimeUnit.SECONDS;
 import static java.util.stream.Collectors.joining;
@@ -65,7 +64,7 @@ public abstract class PersistentFSRecordsStorageTestBase<T extends PersistentFSR
 
 
   @Test
-  public void recordsCountIsZeroForEmptyStorage() {
+  public void freshStorage_hasZeroRecords() {
     assertEquals(
       "Should be 0 records in the empty storage",
       0,
@@ -73,6 +72,34 @@ public abstract class PersistentFSRecordsStorageTestBase<T extends PersistentFSR
     );
   }
 
+  @Test
+  public void freshStorage_isNotDirty() {
+    assertFalse(
+      "Fresh storage must not be dirty",
+      storage.isDirty()
+    );
+  }
+
+  @Test
+  public void wasClosedProperly_isTrue_forFreshStorage() throws IOException {
+    assertTrue(
+      "Fresh storage is always 'closed properly'",
+      storage.wasClosedProperly()
+    );
+  }
+
+  @Test
+  public void wasClosedProperly_isTrueForReopenedStorage() throws IOException {
+    storage.allocateRecord();//any modification
+    storage.close();
+
+    T storageReopened = openStorage(storagePath);
+    assertTrue(
+      "Reopened storage was closed properly",
+      storageReopened.wasClosedProperly()
+    );
+  }
+
   @Test
   public void firstRecordInserted_MustGetValidId() throws Exception {
     final int recordId = allocateRecordAndCheckConsistency(storage);
@@ -463,7 +490,6 @@ public abstract class PersistentFSRecordsStorageTestBase<T extends PersistentFSR
     final int version = 1;
     storage.setVersion(version);
     final long createdTimestamp = storage.getTimestamp();
-    storage.setConnectionStatus(CONNECTED_MAGIC);
 
     final FSRecord[] records = new FSRecord[maxRecordsToInsert];
     for (int i = 0; i < records.length; i++) {
@@ -482,11 +508,6 @@ public abstract class PersistentFSRecordsStorageTestBase<T extends PersistentFSR
       createdTimestamp,
       storage.getTimestamp()
     );
-    assertEquals(
-      "storage.connectedStatus must keep value assigned initially",
-      CONNECTED_MAGIC,
-      storage.getConnectionStatus()
-    );
   }
 
 
@@ -495,10 +516,8 @@ public abstract class PersistentFSRecordsStorageTestBase<T extends PersistentFSR
   @Test
   public void emptyStorageRemains_EmptyButHeaderFieldsStillRestored_AfterStorageClosedAndReopened() throws IOException {
     final int version = 10;
-    final int connectionStatus = CONNECTED_MAGIC;
 
     storage.setVersion(version);
-    storage.setConnectionStatus(connectionStatus);
     final int globalModCount = storage.getGlobalModCount();
     assertTrue("Storage must be 'dirty' after few header fields were written",
                storage.isDirty());
@@ -518,7 +537,7 @@ public abstract class PersistentFSRecordsStorageTestBase<T extends PersistentFSR
     assertFalse("Storage must be !dirty since no modifications since open", storageReopened.isDirty());
     assertEquals("globalModCount", globalModCount, storageReopened.getGlobalModCount());
     assertEquals("version", version, storageReopened.getVersion());
-    assertEquals("connectionStatus", connectionStatus, storageReopened.getConnectionStatus());
+    assertTrue("connectionStatus", storageReopened.wasClosedProperly());
     assertEquals("recordsCountBeforeClose", recordsCountBeforeClose, storageReopened.recordsCount());
   }
 
@@ -540,29 +559,6 @@ public abstract class PersistentFSRecordsStorageTestBase<T extends PersistentFSR
     assertEqualExceptModCount("Record written should be read back as-is", recordWritten, recordReadBack);
   }
 
-  private static int allocateRecordAndCheckConsistency(final PersistentFSRecordsStorage storage) throws IOException {
-    int newRecordId = storage.allocateRecord();
-    
-    int nameId = storage.getNameId(newRecordId);
-    int contentId = storage.getContentRecordId(newRecordId);
-    int attributeRecordId = storage.getAttributeRecordId(newRecordId);
-    int flags = storage.getFlags(newRecordId);
-    int parentId = storage.getParent(newRecordId);
-    long timestamp = storage.getTimestamp(newRecordId);
-    long length = storage.getLength(newRecordId);
-    if (nameId != NULL_NAME_ID || contentId != DataEnumerator.NULL_ID || attributeRecordId != DataEnumerator.NULL_ID
-        || parentId != PersistentFSRecordsStorage.NULL_ID
-        || flags != 0 || timestamp != 0 || length != 0) {
-      throw new AssertionError(
-        "new record (id: " + nameId + ") must be empty, but it has non-empty fields: " +
-        "nameId= " + nameId + ", contentId=" + contentId + ", attributeId=" + attributeRecordId + ", " +
-        "parentId=" + parentId + ", flags=" + flags + ", timestamp=" + timestamp + ", length=" + length
-      );
-    }
-    return newRecordId;
-  }
-
-
   @Test
   public void globalStorageModCount_ShouldNotChange_OnForceAndClose() throws IOException {
     int modCountBefore = storage.getGlobalModCount();
@@ -780,6 +776,28 @@ public abstract class PersistentFSRecordsStorageTestBase<T extends PersistentFSR
                recordOriginal.equalsExceptModCount(recordReadBack));
   }
 
+  private static int allocateRecordAndCheckConsistency(final PersistentFSRecordsStorage storage) throws IOException {
+    int newRecordId = storage.allocateRecord();
+
+    int nameId = storage.getNameId(newRecordId);
+    int contentId = storage.getContentRecordId(newRecordId);
+    int attributeRecordId = storage.getAttributeRecordId(newRecordId);
+    int flags = storage.getFlags(newRecordId);
+    int parentId = storage.getParent(newRecordId);
+    long timestamp = storage.getTimestamp(newRecordId);
+    long length = storage.getLength(newRecordId);
+    if (nameId != NULL_NAME_ID || contentId != DataEnumerator.NULL_ID || attributeRecordId != DataEnumerator.NULL_ID
+        || parentId != PersistentFSRecordsStorage.NULL_ID
+        || flags != 0 || timestamp != 0 || length != 0) {
+      throw new AssertionError(
+        "new record (id: " + nameId + ") must be empty, but it has non-empty fields: " +
+        "nameId= " + nameId + ", contentId=" + contentId + ", attributeId=" + attributeRecordId + ", " +
+        "parentId=" + parentId + ", flags=" + flags + ", timestamp=" + timestamp + ", length=" + length
+      );
+    }
+    return newRecordId;
+  }
+
   /**
    * Newly implemented storages provide experimental APIs for 'per-record' updates, but default API
    * should also be tested -- hence specific API variant to test is abstracted out, and could be

platform/platform-tests/testSrc/com/intellij/openapi/vfs/newvfs/persistent/VFSCorruptionRecoveryTest.java

@@ -1,4 +1,4 @@
-// Copyright 2000-2023 JetBrains s.r.o. and contributors. Use of this source code is governed by the Apache 2.0 license.
+// Copyright 2000-2024 JetBrains s.r.o. and contributors. Use of this source code is governed by the Apache 2.0 license.
 package com.intellij.openapi.vfs.newvfs.persistent;
 
 import com.intellij.openapi.util.io.FileUtil;
@@ -450,7 +450,7 @@ public class VFSCorruptionRecoveryTest {
     try (SeekableByteChannel channel = Files.newByteChannel(recordsPath, WRITE)) {
       ByteBuffer oneFieldBuffer = ByteBuffer.allocate(Integer.BYTES)
         .order(ByteOrder.nativeOrder());
-      oneFieldBuffer.putInt(PersistentFSHeaders.CONNECTED_MAGIC)
+      oneFieldBuffer.putInt(PersistentFSHeaders.IN_USE_STAMP)
         .clear();
 
       channel.position(HEADER_CONNECTION_STATUS_OFFSET);

platform/platform-tests/testSrc/com/intellij/openapi/vfs/newvfs/persistent/VFSInitializationTest.java

@@ -210,16 +210,16 @@ public class VFSInitializationTest {
 
         FSRecordsImpl fsRecords = FSRecordsImpl.connect(cachesDir);
         try {
-            //add something to VFS so it is not empty
-            int testFileId = fsRecords.createRecord();
-            fsRecords.setName(testFileId, "test");
-            try (var stream = fsRecords.writeContent(testFileId, false)) {
-              stream.writeUTF("test");
-            }
-            try (var stream = fsRecords.writeAttribute(testFileId, TEST_FILE_ATTRIBUTE)) {
-              stream.writeInt(42);
-            }
-            vfsVersion = fsRecords.getVersion();
+          //add something to VFS so it is not empty
+          int testFileId = fsRecords.createRecord();
+          fsRecords.setName(testFileId, "test");
+          try (var stream = fsRecords.writeContent(testFileId, false)) {
+            stream.writeUTF("test");
+          }
+          try (var stream = fsRecords.writeAttribute(testFileId, TEST_FILE_ATTRIBUTE)) {
+            stream.writeInt(42);
+          }
+          vfsVersion = fsRecords.getVersion();
         }
         finally {
           StorageTestingUtils.bestEffortToCloseAndUnmap(fsRecords);
@@ -324,20 +324,13 @@ public class VFSInitializationTest {
     final int version = 1;
 
     final PersistentFSConnection connection = tryInit(cachesDir, version, PersistentFSConnector.RECOVERERS);
-    try {
-      final PersistentFSRecordsStorage records = connection.getRecords();
-      records.setConnectionStatus(PersistentFSHeaders.CONNECTED_MAGIC);
-    }
-    finally {
-      //stamps connectionStatus=SAFELY_CLOSED_MAGIC
-      connection.close();
-    }
+    //stamps connectionStatus=SAFELY_CLOSED_MAGIC
+    connection.close();
 
     final PersistentFSConnection reopenedConnection = tryInit(cachesDir, version, PersistentFSConnector.RECOVERERS);
     try {
-      assertEquals("connectionStatus must be SAFELY_CLOSED since connection was disconnect()-ed",
-                   PersistentFSHeaders.SAFELY_CLOSED_MAGIC,
-                   reopenedConnection.getRecords().getConnectionStatus());
+      assertTrue("records must report 'closedProperly' since connection was properly disconnect()-ed",
+                   reopenedConnection.getRecords().wasClosedProperly());
     }
     finally {
       PersistentFSConnector.disconnect(reopenedConnection);
@@ -353,7 +346,6 @@ public class VFSInitializationTest {
     connection.doForce(); //persist VFS initial state
     try {
       final PersistentFSRecordsStorage records = connection.getRecords();
-      records.setConnectionStatus(PersistentFSHeaders.CONNECTED_MAGIC);
       records.force();
 
       //do NOT call connection.close() -- just reopen the connection:

platform/util/src/com/intellij/util/io/StorageAlreadyInUseException.java

@@ -0,0 +1,17 @@
+// Copyright 2000-2024 JetBrains s.r.o. and contributors. Use of this source code is governed by the Apache 2.0 license.
+package com.intellij.util.io;
+
+import java.io.IOException;
+
+/**
+ * Thrown if an opening storage detected to be already opened/used by somebody else.
+ */
+public class StorageAlreadyInUseException extends IOException {
+  public StorageAlreadyInUseException(String message) {
+    super(message);
+  }
+
+  public StorageAlreadyInUseException(String message, Throwable cause) {
+    super(message, cause);
+  }
+}