fix(schema): Add file locking to prevent concurrent cache corruption (#4598)

* fix(schema): Add file locking to prevent concurrent cache corruption

When multiple cfn-lint processes run concurrently (e.g., pre-commit hook
partitions), they can all attempt to update the schema cache simultaneously.
The previous implementation deleted and extracted files in-place without
synchronization, causing FileNotFoundError when one process deletes a file
another process is about to unlink.

This change:
- Adds cross-platform file locking (fcntl on Unix, msvcrt on Windows)
- Extracts schemas to a temp directory then atomically swaps via rename
- Re-checks version after acquiring lock to skip redundant downloads
- Adds comprehensive tests for locking, atomic replacement, and concurrency

Fixes #4597

* test(schema): Add pragma no cover for platform-specific code

Windows-specific lock paths and defensive OSError handling cannot be
tested on Linux CI. Mark them with pragma: no cover to satisfy
coverage requirements.

* style(schema): Fix ruff E501 line-length and unused imports

Apply ruff formatting to wrap long lines (E501) and remove unused
os/sys imports flagged by the lint step.

* fix(schema): Add type: ignore for Windows-only msvcrt attributes

mypy on non-Windows platforms flags msvcrt.locking/LK_NBLCK/LK_UNLCK as
missing attributes since they only exist in the Windows typeshed stub.
The calls are guarded by sys.platform == 'win32' at runtime, so suppress
the attr-defined errors.

* fix(schema): Add type: ignore for Unix-only fcntl attributes

mypy runs on both Linux and Windows in CI. On Windows, fcntl.flock/
LOCK_EX/LOCK_NB/LOCK_UN are flagged as missing (Unix-only), mirroring
the msvcrt case on Linux. Suppress attr-defined on both platform paths;
warn_unused_ignores is disabled so the inactive-platform ignores are
harmless.

* fix(schema): Log accurate messages per update failure mode

Address CR feedback: the broad 'except Exception' around the file_lock
block mislabeled any error from _update_locked (BadZipFile, extraction
or atomic-replace OSError) as 'Failed to acquire schema cache lock'.

- Outer handler now distinguishes TimeoutError (lock wait), OSError
  (lock acquisition), and an unexpected-error fallback.
- _update_locked wraps extraction + atomic replace and logs an accurate
  'Failed to extract and install schema cache' message, so filesystem/
  zip errors are no longer reported as lock failures.
- Add tests asserting each failure mode logs its correct message and
  that update/extract errors are never labeled as lock errors.

* fix(schema): Handle network errors in version checks distinctly

Address CR follow-up: url_has_newer_version() makes a network request and
URLError subclasses OSError, so a network failure during the under-lock
re-check landed in the outer 'except OSError' and was mislabeled as
'Failed to acquire schema cache lock'.

- Wrap both version checks (the pre-lock check in update() and the
  re-check in _update_locked) and log 'Failed to check schema version'.
- The pre-lock check was previously unwrapped and would crash with an
  uncaught URLError; it now returns 2 consistently.
- _update_locked now never raises — it always returns an exit code — so
  the outer handler only ever sees genuine lock-acquisition errors.
- Add tests for network errors at both version-check call sites.

* fix(schema): Let force bypass the pre-lock version check

Address CR follow-up: 'url_has_newer_version() or force' evaluated the
network check before force, so a --force update on an unreachable network
raised URLError and returned 2 instead of proceeding. Swap operands to
'force or url_has_newer_version()' so force short-circuits, matching the
order already used in _update_locked.

- Fix test_update_initial_version_check_network_error to use force=False
  (the previous force=True assertion locked in the bug).
- Add test_update_force_bypasses_version_check asserting force=True
  downloads without ever invoking the version check.

Author: kddejong

src/cfnlint/schema/_lock.py

@@ -0,0 +1,129 @@
+"""
+Cross-platform file locking for schema cache updates.
+
+Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+SPDX-License-Identifier: MIT-0
+"""
+
+from __future__ import annotations
+
+import logging
+import sys
+import time
+from contextlib import contextmanager
+from pathlib import Path
+from typing import IO, Iterator
+
+LOGGER = logging.getLogger(__name__)
+
+
+@contextmanager
+def file_lock(lock_path: Path, timeout: float = 300.0) -> Iterator[IO[str]]:
+    """Acquire an exclusive file lock, blocking until available or timeout.
+
+    Creates the lock file and parent directories if they don't exist.
+    The lock is released when the context manager exits.
+
+    Args:
+        lock_path: Path to the lock file
+        timeout: Maximum seconds to wait for the lock (default 5 minutes)
+
+    Yields:
+        The open lock file handle
+
+    Raises:
+        TimeoutError: If the lock cannot be acquired within timeout
+        OSError: If lock file cannot be created or locked
+    """
+    lock_path.parent.mkdir(parents=True, exist_ok=True)
+
+    # Open in write mode to create if missing
+    lock_file = open(lock_path, "w", encoding="utf-8")
+    try:
+        _acquire_lock(lock_file, timeout)
+        yield lock_file
+    finally:
+        _release_lock(lock_file)
+        lock_file.close()
+
+
+def _acquire_lock(lock_file: IO[str], timeout: float) -> None:
+    """Platform-specific lock acquisition with retry loop."""
+    if sys.platform == "win32":  # pragma: no cover
+        _acquire_lock_windows(lock_file, timeout)
+    else:
+        _acquire_lock_unix(lock_file, timeout)
+
+
+def _acquire_lock_unix(lock_file: IO[str], timeout: float) -> None:
+    """Acquire lock on Unix using fcntl.flock()."""
+    import fcntl
+
+    start = time.monotonic()
+    while True:
+        try:
+            fcntl.flock(  # type: ignore[attr-defined]
+                lock_file.fileno(),
+                fcntl.LOCK_EX | fcntl.LOCK_NB,  # type: ignore[attr-defined]
+            )
+            LOGGER.debug("Acquired schema cache lock")
+            return
+        except (BlockingIOError, OSError):
+            if time.monotonic() - start >= timeout:
+                raise TimeoutError(
+                    f"Could not acquire schema cache lock within {timeout}s. "
+                    "Another cfn-lint process may be updating the cache."
+                )
+            time.sleep(0.1)
+
+
+def _acquire_lock_windows(
+    lock_file: IO[str], timeout: float
+) -> None:  # pragma: no cover
+    """Acquire lock on Windows using msvcrt.locking()."""
+    import msvcrt
+
+    start = time.monotonic()
+    while True:
+        try:
+            # Lock byte 0 with exclusive lock (non-blocking)
+            msvcrt.locking(  # type: ignore[attr-defined]
+                lock_file.fileno(),
+                msvcrt.LK_NBLCK,  # type: ignore[attr-defined]
+                1,
+            )
+            LOGGER.debug("Acquired schema cache lock")
+            return
+        except OSError:
+            if time.monotonic() - start >= timeout:
+                raise TimeoutError(
+                    f"Could not acquire schema cache lock within {timeout}s. "
+                    "Another cfn-lint process may be updating the cache."
+                )
+            time.sleep(0.1)
+
+
+def _release_lock(lock_file: IO[str]) -> None:
+    """Platform-specific lock release."""
+    if sys.platform == "win32":  # pragma: no cover
+        import msvcrt
+
+        try:
+            msvcrt.locking(  # type: ignore[attr-defined]
+                lock_file.fileno(),
+                msvcrt.LK_UNLCK,  # type: ignore[attr-defined]
+                1,
+            )
+        except OSError:
+            pass  # Already unlocked or file closed
+    else:
+        import fcntl
+
+        try:
+            fcntl.flock(  # type: ignore[attr-defined]
+                lock_file.fileno(),
+                fcntl.LOCK_UN,  # type: ignore[attr-defined]
+            )
+        except OSError:  # pragma: no cover
+            pass  # Already unlocked or file closed
+    LOGGER.debug("Released schema cache lock")

src/cfnlint/schema/manager.py

@@ -9,7 +9,9 @@
 import logging
 import os
 import re
+import shutil
 import sys
+import tempfile
 import zipfile
 from functools import lru_cache
 from pathlib import Path
@@ -24,6 +26,7 @@
 )
 from cfnlint.schema._exceptions import ResourceNotFoundError
 from cfnlint.schema._getatts import AttributeDict
+from cfnlint.schema._lock import file_lock
 from cfnlint.schema._schema import Schema
 
 if TYPE_CHECKING:
@@ -298,52 +301,113 @@ def get_resource_types(self, region: str) -> list[str]:
     def update(self, force: bool) -> int:
         """Update schemas from the enhanced schemas repository.
 
-        Writes to the user cache directory. After update, switches
-        to reading from the cache so the fresh schemas are used.
+        Uses file locking to prevent concurrent processes from corrupting the
+        cache. Extracts to a temporary directory and atomically replaces the
+        active cache directories.
 
         Args:
             force (bool): force the schemas to be downloaded
         Returns:
             int: exit code (0=success, 2=failure)
         """
-        if not (url_has_newer_version(_ENHANCED_SCHEMAS_URL) or force):
-            LOGGER.info("Schemas are up to date")
-            return 0
+        # url_has_newer_version() performs a network HEAD request. URLError is
+        # an OSError subclass, so wrap it here — otherwise a network failure
+        # would surface later as a misleading lock-acquisition error.
+        # `force` is evaluated first so a --force update bypasses the network
+        # check entirely (matches the short-circuit order in _update_locked).
+        try:
+            if not (force or url_has_newer_version(_ENHANCED_SCHEMAS_URL)):
+                LOGGER.info("Schemas are up to date")
+                return 0
+        except OSError as e:
+            LOGGER.error("Failed to check schema version: %s", e)
+            return 2
+
+        _cache = Path(get_cache_dir())
+        lock_path = _cache / ".update.lock"
+
+        try:
+            with file_lock(lock_path):
+                return self._update_locked(_cache, force)
+        except TimeoutError as e:
+            LOGGER.error("Timed out waiting for schema cache lock: %s", e)
+            return 2
+        except OSError as e:
+            # Raised by file_lock while creating/locking the lock file
+            LOGGER.error("Failed to acquire schema cache lock: %s", e)
+            return 2
+        except Exception as e:  # pragma: no cover
+            LOGGER.error("Schema update failed: %s", e)
+            return 2
+
+    def _update_locked(self, cache_dir: Path, force: bool) -> int:
+        """Perform the actual update while holding the lock.
+
+        Extracts schemas to a temporary directory, then atomically replaces
+        the live providers/ and resources/ directories.
+
+        Args:
+            cache_dir: The cache directory root
+            force: Whether the update was forced
+        Returns:
+            int: exit code (0=success, 2=failure)
+        """
+        # Re-check version under lock in case another process just updated.
+        # url_has_newer_version() makes a network request; URLError subclasses
+        # OSError, so handle it here rather than letting it propagate to the
+        # caller's lock-acquisition handler. This keeps the invariant that
+        # _update_locked never raises — it always returns an exit code.
+        try:
+            if not force and not url_has_newer_version(_ENHANCED_SCHEMAS_URL):
+                LOGGER.info("Schemas were updated by another process")
+                return 0
+        except OSError as e:
+            LOGGER.error("Failed to check schema version: %s", e)
+            return 2
 
         try:
             filehandle = get_url_retrieve(_ENHANCED_SCHEMAS_URL, caching=True)
         except Exception as e:
             LOGGER.error("Failed to download enhanced schemas: %s", e)
             return 2
 
-        _cache = Path(get_cache_dir())
-        providers_dir = _cache / "providers"
-        resources_dir = _cache / "resources"
-
-        with zipfile.ZipFile(filehandle, "r") as zip_ref:
-            providers_dir.mkdir(parents=True, exist_ok=True)
-            resources_dir.mkdir(parents=True, exist_ok=True)
-
-            for f in providers_dir.glob("*.json"):
-                f.unlink()
-            for f in resources_dir.glob("*.json"):
-                f.unlink()
+        providers_dir = cache_dir / "providers"
+        resources_dir = cache_dir / "resources"
 
-            for name in zip_ref.namelist():
-                if not name.endswith(".json"):
-                    continue
-                if name.startswith("providers/"):
-                    dest = providers_dir / Path(name).name
-                    with zip_ref.open(name) as src, open(dest, "wb") as dst:
-                        dst.write(src.read())
-                elif name.startswith("resources/"):
-                    dest = resources_dir / Path(name).name
-                    with zip_ref.open(name) as src, open(dest, "wb") as dst:
-                        dst.write(src.read())
+        # Extract to a temporary directory, then atomically swap
+        try:
+            with tempfile.TemporaryDirectory(dir=cache_dir) as tmpdir:
+                tmp_path = Path(tmpdir)
+                tmp_providers = tmp_path / "providers"
+                tmp_resources = tmp_path / "resources"
+                tmp_providers.mkdir()
+                tmp_resources.mkdir()
+
+                with zipfile.ZipFile(filehandle, "r") as zip_ref:
+                    for name in zip_ref.namelist():
+                        if not name.endswith(".json"):
+                            continue
+                        if name.startswith("providers/"):
+                            dest = tmp_providers / Path(name).name
+                            with zip_ref.open(name) as src, open(dest, "wb") as dst:
+                                dst.write(src.read())
+                        elif name.startswith("resources/"):
+                            dest = tmp_resources / Path(name).name
+                            with zip_ref.open(name) as src, open(dest, "wb") as dst:
+                                dst.write(src.read())
+
+                # Atomic replacement: remove old, rename new. On POSIX, rename()
+                # is atomic when src and dst share a filesystem, which is
+                # guaranteed here by extracting under the same cache_dir.
+                self._atomic_replace_dir(tmp_providers, providers_dir)
+                self._atomic_replace_dir(tmp_resources, resources_dir)
+        except (OSError, zipfile.BadZipFile) as e:
+            LOGGER.error("Failed to extract and install schema cache: %s", e)
+            return 2
 
         try:
             version_content = get_url_content(_VERSION_URL)
-            with open(_cache / "version.json", "w", encoding="utf-8") as vf:
+            with open(cache_dir / "version.json", "w", encoding="utf-8") as vf:
                 vf.write(version_content)
         except Exception:
             LOGGER.debug("Could not download version.json")
@@ -354,6 +418,42 @@ def update(self, force: bool) -> int:
         self.reset()
         return 0
 
+    @staticmethod
+    def _atomic_replace_dir(src: Path, dst: Path) -> None:
+        """Atomically replace dst directory with src.
+
+        Renames any existing dst to a backup, renames src to dst,
+        then removes the backup. If rename fails (cross-device),
+        falls back to shutil.move.
+
+        Args:
+            src: Source directory (will be moved)
+            dst: Destination directory (will be replaced)
+        """
+        backup = dst.with_suffix(".bak")
+
+        # Remove any stale backup from a previous failed update
+        if backup.exists():
+            shutil.rmtree(backup, ignore_errors=True)
+
+        # Move existing dst out of the way
+        if dst.exists():
+            try:
+                dst.rename(backup)
+            except OSError:
+                # Cross-device or other issue; use shutil
+                shutil.move(str(dst), str(backup))
+
+        # Move new dir into place
+        try:
+            src.rename(dst)
+        except OSError:
+            shutil.move(str(src), str(dst))
+
+        # Clean up backup
+        if backup.exists():
+            shutil.rmtree(backup, ignore_errors=True)
+
     def patch(self, patch: SchemaPatch, region: str) -> None:
         """Patch the schemas as needed