Repository: psf/black

Issue:
Manager not properly shut down in concurrency module - violates multiprocessing best practices
In src/black/concurrency.py, schedule_formatting() creates a multiprocessing.Manager() when write_back is DIFF or COLOR_DIFF, but the Manager is never explicitly shut down.
 [code location:](https://github.com/psf/black/blob/main/src/black/concurrency.py#L159-L166)
```python 
async def schedule_formatting(...):
lock = None
if write_back in (WriteBack.DIFF, WriteBack.COLOR_DIFF):
manager = Manager()
lock = manager.Lock()
```
Since Manager starts a separate server process, not calling shutdown() (or using a context manager) can lead to resource leaks when Black is used as a library or repeatedly in a long-running process. This is mostly hidden in CLI usage because the process exits immediately.

It would be safer to explicitly shut down the Manager or manage it via a context manager to match Python’s multiprocessing best practices.

Discussion hints:

Solution patch:
diff --git a/CHANGES.md b/CHANGES.md
index 24e16e431f3..4f809f80b8f 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -82,6 +82,7 @@ directories.
 
 ### Stable style
 
+- Fix Shutdown multiprocessing Manager in schedule_formatting (#4952)
 - Fix bug where comments preceding `# fmt: off`/`# fmt: on` blocks were incorrectly
   removed, particularly affecting Jupytext's `# %% [markdown]` comments (#4845)
 - Fix crash when multiple `# fmt: skip` comments are used in a multi-part if-clause, on
diff --git a/src/black/concurrency.py b/src/black/concurrency.py
index e939dc34681..2359dba5fe8 100644
--- a/src/black/concurrency.py
+++ b/src/black/concurrency.py
@@ -159,46 +159,52 @@ async def schedule_formatting(
     cancelled = []
     sources_to_cache = []
     lock = None
+    manager = None
     if write_back in (WriteBack.DIFF, WriteBack.COLOR_DIFF):
         # For diff output, we need locks to ensure we don't interleave output
         # from different processes.
         manager = Manager()
         lock = manager.Lock()
-    tasks = {
-        asyncio.ensure_future(
-            loop.run_in_executor(
-                executor, format_file_in_place, src, fast, mode, write_back, lock
-            )
-        ): src
-        for src in sorted(sources)
-    }
-    pending = tasks.keys()
+
     try:
-        loop.add_signal_handler(signal.SIGINT, cancel, pending)
-        loop.add_signal_handler(signal.SIGTERM, cancel, pending)
-    except NotImplementedError:
-        # There are no good alternatives for these on Windows.
-        pass
-    while pending:
-        done, _ = await asyncio.wait(pending, return_when=asyncio.FIRST_COMPLETED)
-        for task in done:
-            src = tasks.pop(task)
-            if task.cancelled():
-                cancelled.append(task)
-            elif exc := task.exception():
-                if report.verbose:
-                    traceback.print_exception(type(exc), exc, exc.__traceback__)
-                report.failed(src, str(exc))
-            else:
-                changed = Changed.YES if task.result() else Changed.NO
-                # If the file was written back or was successfully checked as
-                # well-formatted, store this information in the cache.
-                if write_back is WriteBack.YES or (
-                    write_back is WriteBack.CHECK and changed is Changed.NO
-                ):
-                    sources_to_cache.append(src)
-                report.done(src, changed)
-    if cancelled:
-        await asyncio.gather(*cancelled, return_exceptions=True)
-    if sources_to_cache and not no_cache and cache is not None:
-        cache.write(sources_to_cache)
+        tasks = {
+            asyncio.ensure_future(
+                loop.run_in_executor(
+                    executor, format_file_in_place, src, fast, mode, write_back, lock
+                )
+            ): src
+            for src in sorted(sources)
+        }
+        pending = tasks.keys()
+        try:
+            loop.add_signal_handler(signal.SIGINT, cancel, pending)
+            loop.add_signal_handler(signal.SIGTERM, cancel, pending)
+        except NotImplementedError:
+            # There are no good alternatives for these on Windows.
+            pass
+        while pending:
+            done, _ = await asyncio.wait(pending, return_when=asyncio.FIRST_COMPLETED)
+            for task in done:
+                src = tasks.pop(task)
+                if task.cancelled():
+                    cancelled.append(task)
+                elif exc := task.exception():
+                    if report.verbose:
+                        traceback.print_exception(type(exc), exc, exc.__traceback__)
+                    report.failed(src, str(exc))
+                else:
+                    changed = Changed.YES if task.result() else Changed.NO
+                    # If the file was written back or was successfully checked as
+                    # well-formatted, store this information in the cache.
+                    if write_back is WriteBack.YES or (
+                        write_back is WriteBack.CHECK and changed is Changed.NO
+                    ):
+                        sources_to_cache.append(src)
+                    report.done(src, changed)
+        if cancelled:
+            await asyncio.gather(*cancelled, return_exceptions=True)
+        if sources_to_cache and not no_cache and cache is not None:
+            cache.write(sources_to_cache)
+    finally:
+        if manager is not None:
+            manager.shutdown()


Test patch:
diff --git a/tests/test_concurrency_manager_shutdown.py b/tests/test_concurrency_manager_shutdown.py
new file mode 100644
index 00000000000..18758da476b
--- /dev/null
+++ b/tests/test_concurrency_manager_shutdown.py
@@ -0,0 +1,66 @@
+from __future__ import annotations
+
+import asyncio
+from concurrent.futures import ThreadPoolExecutor
+from pathlib import Path
+from typing import Any, Optional
+
+import black.concurrency as concurrency
+from black import Mode, WriteBack
+from black.report import Report
+
+
+class FakeManager:
+    shutdown_called: bool
+
+    def __init__(self) -> None:
+        self.shutdown_called = False
+
+    def Lock(self) -> object:
+        return object()
+
+    def shutdown(self) -> None:
+        self.shutdown_called = True
+
+
+def test_manager_shutdown_called_for_diff(monkeypatch: Any, tmp_path: Path) -> None:
+    """
+    schedule_formatting() creates multiprocessing.Manager() for DIFF/COLOR_DIFF
+    and must shut it down deterministically.
+    """
+    fake_manager = FakeManager()
+
+    monkeypatch.setattr(concurrency, "Manager", lambda: fake_manager)
+
+    def fake_format_file_in_place(
+        src: Path,
+        fast: bool,
+        mode: Mode,
+        write_back: WriteBack,
+        lock: Optional[object],
+    ) -> bool:
+        assert lock is not None
+        return False
+
+    monkeypatch.setattr(concurrency, "format_file_in_place", fake_format_file_in_place)
+
+    src = tmp_path / "a.py"
+    src.write_text("x=1\n", encoding="utf8")
+
+    async def run() -> None:
+        loop = asyncio.get_running_loop()
+        with ThreadPoolExecutor(max_workers=1) as executor:
+            await concurrency.schedule_formatting(
+                sources={src},
+                fast=False,
+                write_back=WriteBack.DIFF,
+                mode=Mode(),
+                report=Report(),
+                loop=loop,
+                executor=executor,
+                no_cache=True,
+            )
+
+    asyncio.run(run())
+
+    assert fake_manager.shutdown_called is True