test: prove broken locking patterns lose writes under contention

Add parameterized tests that run the old broken patterns (flock on
data file + os.replace, and no locking at all) under high contention
and verify they lose writes or corrupt data. These contrast with the
atomic_json_update and flock+seek/truncate tests which must always
preserve all writes.

The broken-pattern tests skip (rather than fail) if the race doesn't
trigger, since thread scheduling is non-deterministic.

Author: Nissan Pow

test/plugins/conda/test_conda_environment_unit.py

@@ -5,6 +5,7 @@
 hashing/dedup logic. Uses temp files — no conda installation needed.
 """
 
+import fcntl
 import json
 import os
 import tempfile
@@ -13,63 +14,166 @@
 import pytest
 
 
-class TestManifestConcurrentWrites:
-    """Test that manifest file operations are thread-safe."""
+def _flock_on_data_file(manifest_path, key, value):
+    """Old pattern: flock the data file itself + seek/truncate.
+
+    Serializes correctly but not crash-safe (truncate then write).
+    """
+    with open(manifest_path, "r+") as f:
+        fcntl.flock(f.fileno(), fcntl.LOCK_EX)
+        try:
+            data = json.load(f)
+            data[key] = value
+            f.seek(0)
+            f.truncate()
+            json.dump(data, f)
+            f.flush()
+            os.fsync(f.fileno())
+        finally:
+            fcntl.flock(f.fileno(), fcntl.LOCK_UN)
 
-    def test_concurrent_writes_no_corruption(self):
-        """Multiple threads writing to the manifest should not corrupt it."""
-        from metaflow.util import atomic_json_update
 
-        errors = []
-        num_threads = 10
-        writes_per_thread = 5
-
-        with tempfile.NamedTemporaryFile(
-            mode="w", suffix=".manifest", delete=False
-        ) as f:
-            manifest_path = f.name
-            json.dump({}, f)
-
-        def write_to_manifest(thread_id):
-            try:
-                for i in range(writes_per_thread):
-                    key = f"env_{thread_id}_{i}"
-                    atomic_json_update(
-                        manifest_path,
-                        lambda d, k=key, t=thread_id: {
-                            **d,
-                            k: {"platform": "linux-64", "thread": t},
-                        },
-                    )
-            except Exception as e:
-                errors.append(e)
-
-        threads = [
-            threading.Thread(target=write_to_manifest, args=(tid,))
-            for tid in range(num_threads)
-        ]
-        for t in threads:
-            t.start()
-        for t in threads:
-            t.join()
+def _flock_on_data_file_with_replace(manifest_path, key, value):
+    """Broken pattern: flock the data file + os.replace.
 
+    The os.replace creates a new inode, so the flock on the old file
+    descriptor doesn't prevent other threads from acquiring their own
+    lock on the new inode. Writes are lost under contention.
+    """
+    with open(manifest_path, "r+") as f:
+        fcntl.flock(f.fileno(), fcntl.LOCK_EX)
         try:
-            assert len(errors) == 0, f"Errors during concurrent writes: {errors}"
+            data = json.load(f)
+            data[key] = value
+            tmp = manifest_path + ".tmp"
+            with open(tmp, "w") as tmp_f:
+                json.dump(data, tmp_f)
+            os.replace(tmp, manifest_path)
+        finally:
+            fcntl.flock(f.fileno(), fcntl.LOCK_UN)
+
+
+def _no_locking(manifest_path, key, value):
+    """Old production_token.py pattern: no locking at all."""
+    with open(manifest_path, "r") as f:
+        data = json.load(f)
+    data[key] = value
+    with open(manifest_path, "w") as f:
+        json.dump(data, f)
+
 
+def _atomic_json_update_wrapper(manifest_path, key, value):
+    """New pattern: lock file + atomic replace."""
+    from metaflow.util import atomic_json_update
+
+    atomic_json_update(manifest_path, lambda d: {**d, key: value})
+
+
+def _run_concurrent_writes(write_fn, num_threads=10, writes_per_thread=5):
+    """Run concurrent writes using the given write function.
+
+    Returns (num_keys_written, expected_keys, errors).
+    """
+    errors = []
+
+    with tempfile.NamedTemporaryFile(mode="w", suffix=".manifest", delete=False) as f:
+        manifest_path = f.name
+        json.dump({}, f)
+
+    def writer(thread_id):
+        try:
+            for i in range(writes_per_thread):
+                key = f"env_{thread_id}_{i}"
+                write_fn(manifest_path, key, {"thread": thread_id})
+        except Exception as e:
+            errors.append(e)
+
+    threads = [
+        threading.Thread(target=writer, args=(tid,)) for tid in range(num_threads)
+    ]
+    for t in threads:
+        t.start()
+    for t in threads:
+        t.join()
+
+    try:
+        try:
             with open(manifest_path, "r") as f:
                 data = json.load(f)
+            return len(data), num_threads * writes_per_thread, errors
+        except (json.JSONDecodeError, FileNotFoundError) as e:
+            # File was corrupted or deleted by the broken pattern
+            errors.append(e)
+            return 0, num_threads * writes_per_thread, errors
+    finally:
+        if os.path.exists(manifest_path):
+            os.unlink(manifest_path)
+        for suffix in (".lock", ".tmp"):
+            p = manifest_path + suffix
+            if os.path.exists(p):
+                os.unlink(p)
 
-            expected_keys = num_threads * writes_per_thread
-            assert len(data) == expected_keys, (
-                f"Expected {expected_keys} entries, got {len(data)}. "
-                "Some writes may have been lost due to race conditions."
+
+class TestManifestConcurrentWrites:
+    """Demonstrate that the old locking patterns lose writes and the new one doesn't."""
+
+    @pytest.mark.parametrize(
+        "write_fn,description",
+        [
+            (
+                _flock_on_data_file_with_replace,
+                "flock+replace (inode changes break lock)",
+            ),
+            (_no_locking, "no locking (old production_token.py pattern)"),
+        ],
+    )
+    def test_broken_patterns_lose_writes(self, write_fn, description):
+        """Broken locking patterns lose writes or corrupt data under contention.
+
+        These patterns are known-broken. The race is non-deterministic, so we
+        run many trials with high contention. If the race never triggers, we
+        skip rather than fail — the important thing is that the *fixed* pattern
+        (test_atomic_json_update_no_lost_writes) always passes.
+        """
+        broken_at_least_once = False
+        for _ in range(30):
+            actual, expected, errors = _run_concurrent_writes(
+                write_fn, num_threads=20, writes_per_thread=10
+            )
+            if errors or actual < expected:
+                broken_at_least_once = True
+                break
+
+        if not broken_at_least_once:
+            pytest.skip(
+                f"Race condition in '{description}' did not trigger in 30 trials "
+                f"(non-deterministic). The pattern is still broken; the OS scheduler "
+                f"just didn't interleave threads enough to expose it this time."
+            )
+
+    def test_atomic_json_update_no_lost_writes(self):
+        """atomic_json_update with lock file never loses writes."""
+        for _ in range(5):
+            actual, expected, errors = _run_concurrent_writes(
+                _atomic_json_update_wrapper, num_threads=10, writes_per_thread=5
+            )
+            assert len(errors) == 0, f"Unexpected errors: {errors}"
+            assert actual == expected, (
+                f"Expected {expected} entries, got {actual}. "
+                "atomic_json_update lost writes."
+            )
+
+    def test_flock_seek_truncate_no_lost_writes(self):
+        """flock on data file + seek/truncate doesn't lose writes (same inode)."""
+        for _ in range(5):
+            actual, expected, errors = _run_concurrent_writes(
+                _flock_on_data_file, num_threads=10, writes_per_thread=5
+            )
+            assert len(errors) == 0, f"Unexpected errors: {errors}"
+            assert actual == expected, (
+                f"Expected {expected} entries, got {actual}. "
+                "flock+seek/truncate lost writes."
             )
-        finally:
-            os.unlink(manifest_path)
-            # Clean up lock file
-            lock_path = manifest_path + ".lock"
-            if os.path.exists(lock_path):
-                os.unlink(lock_path)
 
 
 class TestAtomicJsonUpdate:
@@ -114,28 +218,3 @@ def bad_update(d):
 
             with open(path, "r") as f:
                 assert json.load(f) == {"original": True}
-
-
-class TestCleanupCondaFile:
-    """Test temp file cleanup patterns used by the conda environment."""
-
-    def test_cleanup_temp_file(self):
-        """Verify temp files are cleaned up properly."""
-        with tempfile.NamedTemporaryFile(suffix=".conda_tmp", delete=False) as f:
-            tmp_path = f.name
-            f.write(b"test data")
-
-        assert os.path.exists(tmp_path)
-        os.unlink(tmp_path)
-        assert not os.path.exists(tmp_path)
-
-    def test_cleanup_nonexistent_file_no_error(self):
-        """Cleaning up a file that doesn't exist should not raise."""
-        tmp_path = "/tmp/nonexistent_conda_test_file_12345.tmp"
-        if os.path.exists(tmp_path):
-            os.unlink(tmp_path)
-        # Should not raise
-        try:
-            os.unlink(tmp_path)
-        except FileNotFoundError:
-            pass  # Expected behavior