Release video temporary files before returning pipeline slots (1.0.3)

Changelog.md

@@ -4,6 +4,16 @@ All notable changes are recorded here, newest release first. Versions follow
 [Semantic Versioning](https://semver.org/). Dates use America/Los_Angeles.
 The history before this initial versioned snapshot remains in Git.
 
+## [1.0.3] - 2026-10-17
+
+### Fixed
+
+- Release video temporary files before returning pipeline slots, including
+  oversized rejections and interrupted uploads. A sender waiting on later API
+  work can no longer retain old media outside the two-pipeline bound.
+- Verify immediate cleanup with retained senders and inspect audio/video tracks
+  in the file delivered by Telegram after local upload files have been removed.
+
 ## [1.0.2] - 2026-10-16
 
 ### Fixed

README.md

@@ -97,7 +97,8 @@ Tests
 
 The consumer keeps at most `consumer.max_pending` submitted jobs (default: pool
 size). Two video jobs may download, mux, and upload concurrently. Uploads stream
-from disk, and each sender removes its private temporary files on close. Media
+from disk, and every attempt removes its private files before returning a
+pipeline slot, including oversized rejections and interrupted uploads. Media
 HTTP sessions close after each job; verified TLS contexts are shared across
 connections. On glibc systems,
 `pool.malloc_arenas` defaults to four to reduce memory retained by native

VERSION

@@ -1 +1 @@
-1.0.2
+1.0.3

reddit2telegram/utils/__init__.py

@@ -509,6 +509,7 @@ def _run_async(self, coro):
     def close(self):
         loop = getattr(self, '_loop', None)
         if loop is None or loop.is_closed():
+            self._cleanup_temp_files()
             return
         request = getattr(self, '_telegram_request', None)
         try:
@@ -523,9 +524,13 @@ def close(self):
             logging.exception('Failed to shutdown Telegram HTTP client cleanly.')
         finally:
             loop.close()
-            files = getattr(self, '_temp_files', None)
-            if files is not None:
-                files.cleanup()
+            self._cleanup_temp_files()
+
+    def _cleanup_temp_files(self):
+        files = getattr(self, '_temp_files', None)
+        if files is not None:
+            files.cleanup()
+            self._temp_files = None
 
     def __enter__(self):
         return self
@@ -722,7 +727,13 @@ def _get_dash_audio_url(self, dash_url):
 
     def send_video(self, url, text, parse_mode=None, dash_url=None):
         with _VIDEO_UPLOAD_SLOTS:
-            return self._send_video(url, text, parse_mode, dash_url)
+            try:
+                return self._send_video(url, text, parse_mode, dash_url)
+            finally:
+                # A rejected candidate can be followed by a long API retry on
+                # the same sender. Release its file cache before the next video
+                # takes this slot, on success, rejection, and upload failure.
+                self._cleanup_temp_files()
 
     def _send_video(self, url, text, parse_mode=None, dash_url=None):
         filename = self._get_file_name('.mp4')

tests/test_live_resources.py

@@ -90,6 +90,7 @@ async def send_video(self, **kwargs):
 class RecordingSender(utils.Reddit2TelegramSender):
     def __init__(self, config, root, state):
         self.messages = []
+        self.media_paths = set()
         self.root = root / uuid.uuid4().hex
         self.root.mkdir()
         super().__init__(TEST_CHANNEL, config)
@@ -99,8 +100,10 @@ def __init__(self, config, root, state):
         )
 
     def _get_file_name(self, ext='file'):
-        # Isolate files from production's cleaner and concurrent test senders.
-        return str(self.root / ('video' + ext))
+        # Exercise production's actual private directory and cleanup lifecycle.
+        filename = super()._get_file_name(ext)
+        self.media_paths.add(Path(filename))
+        return filename
 
     def _run_async(self, coro):
         result = super()._run_async(coro)
@@ -238,9 +241,16 @@ def test_real_video_download_audio_merge_and_delivery(self):
             message = sender.messages[-1]
             self.assert_delivery(message)
             self.assertIsNotNone(message.video)
+            self.assertTrue(sender.media_paths)
+            self.assertTrue(all(not path.parent.exists() for path in sender.media_paths),
+                            'A completed upload retained its production temporary directory')
+            # Inspect what Telegram actually received after local media cleanup.
+            delivered = sender.root / 'delivered.mp4'
+            remote = sender._run_async(sender.telegram_bot.get_file(message.video.file_id))
+            sender._run_async(remote.download_to_drive(custom_path=delivered))
             probe = subprocess.run([
                 'ffprobe', '-v', 'error', '-show_entries', 'stream=codec_type', '-of', 'json',
-                sender._get_file_name('.1.mp4'),
+                str(delivered),
             ], check=True, capture_output=True, text=True, timeout=10)
             self.assertEqual({stream['codec_type'] for stream in json.loads(probe.stdout)['streams']},
                              {'audio', 'video'})

tests/test_memory_budget.py

@@ -121,6 +121,34 @@ def stream(size):
     response.__exit__.assert_called_once()
 
 
+def test_rejected_videos_leave_no_files_while_senders_remain_alive(monkeypatch, tmp_path):
+    monkeypatch.setattr(utils, 'TEMP_FOLDER', str(tmp_path))
+
+    def download(url, filename):
+        size = (2 if filename.endswith('.aac') else 49) * 1024 * 1024
+        with open(filename, 'wb') as handle:
+            handle.truncate(size)  # Sparse files exercise limits without allocating GB of RAM/disk.
+        return True
+
+    def mux(args, **kwargs):
+        with open(args[-1], 'wb') as handle:
+            handle.truncate(51 * 1024 * 1024)
+        return subprocess.CompletedProcess(args, 0)
+
+    monkeypatch.setattr(utils, 'download_file', download)
+    monkeypatch.setattr(utils.subprocess, 'run', mux)
+    senders = [object.__new__(utils.Reddit2TelegramSender) for _ in range(12)]
+    for i, sender in enumerate(senders):
+        sender.t_channel = f'@test{i}'
+    with ThreadPoolExecutor(12) as workers:
+        results = list(workers.map(lambda sender: sender.send_video('https://local/video', 'test'), senders))
+    assert all(result == utils.SupplyResult.DO_NOT_WANT_THIS_SUBMISSION for result in results)
+    # A rejected candidate may be followed by a long Reddit retry on the same
+    # sender. Its files must be gone before another video gets the pipeline slot.
+    leftover = sum(path.stat().st_size for path in tmp_path.rglob('*') if path.is_file())
+    assert leftover == 0, f'Rejected videos retained {leftover / 2**20:.0f} MiB of media files'
+
+
 def test_large_video_reaches_http_backend_without_full_file_buffer(monkeypatch, tmp_path):
     size = 49 * 1024 * 1024
     fixture = tmp_path / 'fixture.mp4'
@@ -154,11 +182,13 @@ def log_message(self, *args):
     monkeypatch.setattr(utils, 'short_sleep', lambda: None)
     monkeypatch.setattr(utils.Reddit2TelegramSender, '_make_mongo_connections', lambda self: None)
     monkeypatch.setattr(utils, 'TEMP_FOLDER', str(tmp_path))
+    downloaded = []
 
     def download(url, filename):
         if filename.endswith('.aac'):
             return False
         shutil.copyfile(fixture, filename)
+        downloaded.append(Path(filename))
         return True
 
     monkeypatch.setattr(utils, 'download_file', download)
@@ -173,8 +203,8 @@ def download(url, filename):
                 peak = tracemalloc.get_traced_memory()[1]
             finally:
                 tracemalloc.stop()
-            directory = Path(sender._get_file_name('.mp4')).parent
-            assert directory.exists()
+            directory = downloaded[0].parent
+            assert not directory.exists(), 'Finished upload retained media until sender close'
         assert received and received[0] > size
         assert peak < 12 * 1024 * 1024, f'49 MiB upload allocated {peak / 2**20:.1f} MiB'
         assert not directory.exists(), 'Closed sender retained downloaded media'
@@ -184,6 +214,29 @@ def download(url, filename):
         thread.join(2)
 
 
+def test_failed_upload_releases_files_before_sender_close(monkeypatch, tmp_path):
+    monkeypatch.setattr(utils, 'TEMP_FOLDER', str(tmp_path))
+
+    def download(url, filename):
+        if filename.endswith('.aac'):
+            return False
+        Path(filename).write_bytes(b'video')
+        return True
+
+    async def upload(**kwargs):
+        assert not kwargs['video'].input_file_content.closed
+        raise RuntimeError('upload interrupted')
+
+    monkeypatch.setattr(utils, 'download_file', download)
+    sender = object.__new__(utils.Reddit2TelegramSender)
+    sender.t_channel = '@test'
+    sender.telegram_bot = mock.Mock(send_video=upload)
+    sender._run_async = asyncio.run
+    with pytest.raises(RuntimeError, match='upload interrupted'):
+        sender.send_video('https://local/video', 'test')
+    assert not list(tmp_path.iterdir()), 'Failed upload retained its temporary directory'
+
+
 def test_maintenance_reclaims_dead_job_cycles_without_waiting_for_automatic_gc(monkeypatch):
     class Job:
         pass