gh-159690: Fix timed log rotation after a failed or duplicate rollover (GH-159692)

Author: lpyu001

Lib/logging/handlers.py

@@ -469,6 +469,7 @@ def doRollover(self):
                 timeTuple = time.localtime(t + addend)
         dfn = self.rotation_filename(self.baseFilename + "." +
                                      time.strftime(self.suffix, timeTuple))
+        self.rolloverAt = self.computeRollover(currentTime)
         if os.path.exists(dfn):
             # Already rolled over.
             return
@@ -482,7 +483,6 @@ def doRollover(self):
                 os.remove(s)
         if not self.delay:
             self.stream = self._open()
-        self.rolloverAt = self.computeRollover(currentTime)
 
 class WatchedFileHandler(logging.FileHandler):
     """

Lib/test/test_logging.py

@@ -6902,6 +6902,75 @@ def test_rollover_at_midnight(self, weekly=False):
     def test_rollover_at_weekday(self):
         self.test_rollover_at_midnight(weekly=True)
 
+    def _make_fake_time_handler(self, now, when='S', **kwargs):
+        dirname = tempfile.mkdtemp()
+        self.addCleanup(os_helper.rmtree, dirname)
+        fn = os.path.join(dirname, 'test.log')
+        with patch('time.time', lambda: now[0]):
+            fh = logging.handlers.TimedRotatingFileHandler(
+                fn, when=when, encoding='utf-8', delay=True, **kwargs)
+        self.addCleanup(fh.close)
+        return fh
+
+    def test_rollover_after_failed_rollover(self):
+        for target in 'os.remove', '_open':
+            with self.subTest(target=target):
+                now = [1_000_000_000]
+                fh = self._make_fake_time_handler(now, backupCount=1)
+                fn = fh.baseFilename
+                emit = lambda msg: fh.emit(logging.makeLogRecord({'msg': msg}))
+                with patch('time.time', lambda: now[0]):
+                    for msg in 'a', 'b':
+                        now[0] += 2
+                        emit(msg)
+                    now[0] += 2
+                    if target == '_open':
+                        failure = patch.object(fh, '_open', side_effect=OSError)
+                        fh.delay = False
+                    else:
+                        failure = patch('os.remove', side_effect=OSError)
+                    with failure as failed, support.captured_stderr():
+                        emit('c')
+                    failed.assert_called_once()
+                    self.assertGreater(fh.rolloverAt, now[0])
+                    fh.delay = True
+                    for msg in 'd', 'e':
+                        now[0] += 2
+                        emit(msg)
+                fh.close()
+                with open(fn, encoding='utf-8') as f:
+                    self.assertEqual(f.read(), 'e\n')
+                backups = [f for f in os.listdir(os.path.dirname(fn))
+                           if f != 'test.log']
+                self.assertEqual(len(backups), 1)
+
+    def test_rollover_existing_backup(self):
+        now = [1_000_000_000]
+        fh = self._make_fake_time_handler(now)
+        fn = fh.baseFilename
+        with patch('time.time', lambda: now[0]):
+            fh.emit(logging.makeLogRecord({'msg': 'a'}))
+            t = fh.rolloverAt - fh.interval
+            dfn = fh.rotation_filename(
+                fn + '.' + time.strftime(fh.suffix, time.localtime(t)))
+            with open(dfn, 'w', encoding='utf-8') as f:
+                f.write('existing\n')
+            now[0] += 2
+            fh.emit(logging.makeLogRecord({'msg': 'b'}))
+            self.assertGreater(fh.rolloverAt, now[0])
+            t = fh.rolloverAt - fh.interval
+            next_dfn = fh.rotation_filename(
+                fn + '.' + time.strftime(fh.suffix, time.localtime(t)))
+            now[0] += 2
+            fh.emit(logging.makeLogRecord({'msg': 'c'}))
+        fh.close()
+        with open(dfn, encoding='utf-8') as f:
+            self.assertEqual(f.read(), 'existing\n')
+        with open(next_dfn, encoding='utf-8') as f:
+            self.assertEqual(f.read(), 'a\nb\n')
+        with open(fn, encoding='utf-8') as f:
+            self.assertEqual(f.read(), 'c\n')
+
     def test_invalid(self):
         assertRaises = self.assertRaises
         assertRaises(ValueError, logging.handlers.TimedRotatingFileHandler,

Misc/NEWS.d/next/Library/2026-10-20-00-12-05.gh-issue-159690.vZgLP6.rst

@@ -0,0 +1,3 @@
+Fix :class:`logging.handlers.TimedRotatingFileHandler` permanently stopping
+rollover after a rollover failed or its backup file already existed, for
+example at a DST change.