Don't use a process pool to delete push temp files (#5156)

Cleanup ran after the data upload but before the manifest push, and Pool.map
re-raises the first worker exception, so a temp file that was already gone --
including when two logical keys share one serialized object, deleting it twice
-- failed an otherwise-complete push. Cleanup is now a sequential best-effort
loop: an already-gone file is ignored, any other removal error is logged.

Dropping the pool also lets push run from a daemonic process, where
multiprocessing.Pool raises outright, and stops the library imposing an
`if __name__ == "__main__":` guard on callers via spawn's re-import of __main__.
Concurrency bought nothing: the pool spent ~93 ms of fixed setup on ~0.2 ms of
unlink() work.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

Author: sir-sigurd

api/python/quilt3/packages.py

@@ -16,7 +16,6 @@
 import uuid
 import warnings
 from collections import deque
-from multiprocessing import Pool
 
 import botocore.exceptions
 import jsonlines
@@ -104,11 +103,6 @@ def f(wrapped):
 )
 
 
-def _delete_local_physical_key(pk):
-    assert pk.is_local(), "This function only works on files that live on a local disk"
-    pathlib.Path(pk.path).unlink()
-
-
 def _filesystem_safe_encode(key):
     """Returns the sha256 of the key. This ensures there are no slashes, uppercase/lowercase conflicts,
     avoids `OSError: [Errno 36] File name too long:`, etc."""
@@ -1645,17 +1639,19 @@ def physical_key_is_temp_file(pk):
                 return False
             return pathlib.Path(pk.path).parent.resolve() == APP_DIR_TEMPFILE_DIR.resolve()
 
+        # Materialized first: _set() below mutates what walk() iterates.
         temp_file_logical_keys = [lk for lk, entry in self.walk() if physical_key_is_temp_file(entry.physical_key)]
-        if temp_file_logical_keys:
-            temp_file_physical_keys = [self[lk].physical_key for lk in temp_file_logical_keys]
-
-            # Now that data has been pushed, delete tmp files created by pkg.set('KEY', obj)
-            with Pool(10) as p:
-                p.map(_delete_local_physical_key, temp_file_physical_keys)
+        for lk in temp_file_logical_keys:
+            # Delete tmp files created by pkg.set('KEY', obj). Cleanup is best-effort: a file we
+            # cannot remove is a leaked scratch file, not a reason to fail a completed push.
+            temp_pk = self[lk].physical_key
+            try:
+                pathlib.Path(temp_pk.path).unlink(missing_ok=True)
+            except OSError as e:
+                logger.warning("Failed to remove temporary file %s: %s", temp_pk.path, e)
 
-            # Update old package to point to the materialized location of the file since the tempfile no longest exists
-            for lk in temp_file_logical_keys:
-                self._set(lk, pkg[lk])
+            # Point the entry at the materialized location.
+            self._set(lk, pkg[lk])
 
         # Check top hash again just before pushing, to minimize the race condition.
         if not force:

api/python/tests/integration/test_packages.py

@@ -809,6 +809,59 @@ def test_set_package_entry_as_object(self):
             file_path = pkg[lk].physical_key.path
             assert not pathlib.Path(file_path).exists(), "These temp files should have been deleted during push()"
 
+    @patch('quilt3.workflows.validate', mock.MagicMock(return_value=None))
+    def test_push_survives_missing_temp_file(self):
+        self.patch_s3_registry('shorten_top_hash', return_value='7a67ff4')
+        pkg = Package()
+        df = pd.DataFrame({'col_num': [11, 22, 33]})
+        pkg.set("mydataframe1.parquet", df)
+        pkg.set("mydataframe2.parquet", df)
+        pkg._calculate_missing_hashes()
+
+        # Something else removed one temp file already: a tempdir sweep, a retried push, another process.
+        gone = pathlib.Path(pkg["mydataframe1.parquet"].physical_key.path)
+        remaining = pathlib.Path(pkg["mydataframe2.parquet"].physical_key.path)
+        gone.unlink()
+        assert remaining.exists(), "the two entries must not share a temp file, or this test proves nothing"
+
+        with (
+            patch('quilt3.Package._push_manifest'),
+            patch('quilt3.packages.copy_file_list', _mock_copy_file_list),
+            self.assertNoLogs('quilt3.packages', level='WARNING'),
+        ):
+            pkg.push('Quilt/test_pkg_name', 's3://test-bucket', force=True)
+
+        assert not remaining.exists(), "Cleanup should continue past a temp file that is already gone"
+
+    @patch('quilt3.workflows.validate', mock.MagicMock(return_value=None))
+    def test_push_logs_when_temp_file_cannot_be_removed(self):
+        self.patch_s3_registry('shorten_top_hash', return_value='7a67ff4')
+        pkg = Package()
+        pkg.set("mydataframe1.parquet", pd.DataFrame({'col_num': [11, 22, 33]}))
+        pkg._calculate_missing_hashes()
+        # temp_key, not str(temp_path): the log formats the physical key, whose separators differ
+        # from pathlib's on Windows.
+        temp_key = pkg["mydataframe1.parquet"].physical_key.path
+        temp_path = pathlib.Path(temp_key)
+        real_unlink = pathlib.Path.unlink
+
+        def unlink(self, *args, **kwargs):
+            if self == temp_path:
+                raise PermissionError("file is in use")
+            return real_unlink(self, *args, **kwargs)
+
+        with (
+            patch('quilt3.Package._push_manifest'),
+            patch('quilt3.packages.copy_file_list', _mock_copy_file_list),
+            patch('pathlib.Path.unlink', unlink),
+            self.assertLogs('quilt3.packages', level='WARNING') as logs,
+        ):
+            pkg.push('Quilt/test_pkg_name', 's3://test-bucket', force=True)
+
+        assert any(f"Failed to remove temporary file {temp_key}" in line for line in logs.output)
+        assert temp_path.exists(), "a cleanup failure leaves the temp file behind"
+        temp_path.unlink()
+
     @patch("quilt3.packages.get_size_and_version", mock.Mock(return_value=(123, "v1")))
     def test_set_package_entry_unversioned_flag(self):
         for flag_value, version_id in {

docs/CHANGELOG.md

@@ -23,6 +23,8 @@ Entries inside each section should be ordered by type:
 * [Removed] Drop support for Python 3.9 (end-of-life); `quilt3` now requires Python >= 3.10 ([#4941](https://github.com/quiltdata/quilt/pull/4941))
 * [Fixed] `quilt3.admin.buckets.list` no longer raises `TypeError` when its type hints are introspected on Python 3.14 ([#4940](https://github.com/quiltdata/quilt/pull/4940))
 * [Fixed] `quilt3.delete_package()` on a local registry no longer deletes other packages sharing the same namespace ([#5140](https://github.com/quiltdata/quilt/pull/5140))
+* [Fixed] `Package.push()` no longer fails after the data is uploaded when cleaning up a temporary file created by `Package.set()`: a file that is already gone is ignored, which is also what happens when two logical keys share one serialized object and its temporary file is deleted twice, and any other removal error is logged instead of raised ([#5156](https://github.com/quiltdata/quilt/pull/5156))
+* [Fixed] `Package.push()` no longer uses a process pool to delete temporary files, so it works from a daemonic process (e.g. a prefork worker) and no longer requires callers to guard their entry point with `if __name__ == "__main__":` ([#5156](https://github.com/quiltdata/quilt/pull/5156))
 
 ### CLI