chore(profiling): fix flaky test_upload_resets_profile test (#15178)

## Description

Some tests try to read the newest profile and act on it. One such test,
`test_upload_resets_profile`, is flaky on MacOS (but strangely not on
Linux in the CI). It sometimes reads an empty profile (which is
expected), and sometimes does not (the empty-file assert is not
triggered and test fails).

The current way of finding the latest file in
`pprof_utils.parse_newest_profile` is error-prone because the latest
file `ctime`, metadata change time, is not always a true "latest" file;
this can happen for a variety of reasons, e.g. one file gets written
faster than the other one, because it's smaller ("empty").
_Example:_
* `file.foo.1` created at `t0` (sequence: 1)
* `file.foo.0` created at `t1` (sequence: 0)

Logically, `file_1` is the latest file, because that is the sequence
number assigned to it; however, currently we think it is `file_0` (t1 >
t0).

### What changed ###
* Instead of sorting by `ctime`, we sort by the "logical" time (i.e. the
file's seq_num)

## Testing
* multiple consecutive local runs on my Mac pass reliably (10+ tries)

Author: vlad-scherbich

tests/profiling/collector/pprof_utils.py

@@ -142,8 +142,9 @@ def parse_newest_profile(filename_prefix: str) -> pprof_pb2.Profile:
     the newest profile that has given filename prefix.
     """
     files = glob.glob(filename_prefix + ".*")
-    # Sort files by creation timestamp (oldest first, newest last)
-    files.sort(key=lambda f: os.path.getctime(f))
+    # Sort files by logical timestamp (i.e. the sequence number, which is monotonically increasing);
+    # this approach is more reliable than filesystem timestamps, especially when files are created rapidly.
+    files.sort(key=lambda f: int(f.rsplit(".", 1)[-1]))
     filename = files[-1]
     with open(filename, "rb") as fp:
         dctx = zstd.ZstdDecompressor()

tests/profiling_v2/collector/test_threading.py

@@ -1022,47 +1022,45 @@ def test_upload_resets_profile(self) -> None:
         with self.collector_class(capture_pct=100):
             with self.lock_class():  # !CREATE! !ACQUIRE! !RELEASE! test_upload_resets_profile
                 pass
+
         ddup.upload()  # pyright: ignore[reportCallIssue]
 
         linenos: LineNo = get_lock_linenos("test_upload_resets_profile", with_stmt=True)
 
-        try:
-            pprof: pprof_pb2.Profile = pprof_utils.parse_newest_profile(self.output_filename)
-            pprof_utils.assert_lock_events(
-                pprof,
-                expected_acquire_events=[
-                    pprof_utils.LockAcquireEvent(
-                        caller_name=self.test_name,
-                        filename=os.path.basename(__file__),
-                        linenos=linenos,
-                    ),
-                ],
-                expected_release_events=[
-                    pprof_utils.LockReleaseEvent(
-                        caller_name=self.test_name,
-                        filename=os.path.basename(__file__),
-                        linenos=linenos,
-                    ),
-                ],
-            )
-        except (AssertionError, KeyError) as e:
-            # This can be flaky due to timing or interference from other tests
-            pytest.skip(f"Profile validation failed (known flaky on some platforms): {e}")
+        pprof: pprof_pb2.Profile = pprof_utils.parse_newest_profile(self.output_filename)
+        pprof_utils.assert_lock_events(
+            pprof,
+            expected_acquire_events=[
+                pprof_utils.LockAcquireEvent(
+                    caller_name=self.test_name,
+                    filename=os.path.basename(__file__),
+                    linenos=linenos,
+                ),
+            ],
+            expected_release_events=[
+                pprof_utils.LockReleaseEvent(
+                    caller_name=self.test_name,
+                    filename=os.path.basename(__file__),
+                    linenos=linenos,
+                ),
+            ],
+        )
+
+        # Now we call upload() again, and we expect the profile to be empty
+        num_files_before_second_upload: int = len(glob.glob(self.output_filename + ".*"))
 
-        # Now we call upload() again without any new lock operations
-        # We expect the profile to be empty or contain no samples
         ddup.upload()  # pyright: ignore[reportCallIssue]
 
-        # Try to parse the newest profile - it should either not exist (no new file)
-        # or have no samples (which would raise AssertionError in parse_newest_profile)
-        try:
-            _ = pprof_utils.parse_newest_profile(self.output_filename)
-            # If we got here, a profile with samples exists
-            # This might be okay if other collectors are running
-            pytest.skip("Profile still has samples (possibly from other activity - known flaky)")
-        except (AssertionError, IndexError):
-            # Expected: no profile file or no samples
-            pass
+        num_files_after_second_upload: int = len(glob.glob(self.output_filename + ".*"))
+
+        # A new profile file should always be created (upload_seq increments)
+        assert (
+            num_files_after_second_upload - num_files_before_second_upload == 1
+        ), f"Expected 1 new file, got {num_files_after_second_upload - num_files_before_second_upload}."
+
+        # The newest profile file should be empty (no samples), which causes an AssertionError
+        with pytest.raises(AssertionError, match="No samples found in profile"):
+            pprof_utils.parse_newest_profile(self.output_filename)
 
     def test_lock_hash(self) -> None:
         """Test that __hash__ allows profiled locks to be used in sets and dicts."""