Repository: twisted/twisted

Issue:
ProcessMonitor.startProcess is not handling exception which could be raised by spawnProcess()
[IReactorProcess.spawnProcess()](https://docs.twisted.org/en/stable/api/twisted.internet.interfaces.IReactorProcess.html#spawnProcess) raises `OSError` upon error.

`ProcessMonitor.startProcess()` does not handle the case that `spawnProcess()` might raise an exception.

Discussion:
@adiroiban: Hi km

Thanks for the report.

Just to clarify this report.

Is this a defect or a feature request.

Can you provide an example for a case in which `spawnProcess` raises an OSError and how do you expect `startProcess` to handle it ?

As a general comment, I am not sure what you expect from `startProcess` to do on an OSError raised by `spawnProcess()`

Cheers@juanpeha: This is a bug. I made a patch which catches OSError and deals with it – when a process cannot be started, it should be attempted to be restarted, in a same way as when it exists.@juanpeha: https://github.com/twisted/twisted/pull/12422 is updated with tests – ready for review and merge.

Solution patch:
diff --git a/pyproject.toml b/pyproject.toml
index 05036256cd..55a9913f5f 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -621,7 +621,6 @@ module = [
     'twisted.python.win32',
     'twisted.python.zipstream',
     'twisted.runner.inetd',
-    'twisted.runner.procmon',
     'twisted.runner.test.test_procmon',
     'twisted.scripts._twistd_unix',
     'twisted.scripts.test.test_scripts',
diff --git a/src/twisted/runner/procmon.py b/src/twisted/runner/procmon.py
index 8b3749a6a4..ec3d55a0ca 100644
--- a/src/twisted/runner/procmon.py
+++ b/src/twisted/runner/procmon.py
@@ -5,16 +5,25 @@
 """
 Support for starting, monitoring, and restarting child process.
 """
-from typing import Dict, List, Optional
+from __future__ import annotations
+
+from typing import Any, Dict, List, Optional
 
 import attr
 import incremental
 
 from twisted.application import service
 from twisted.internet import error, protocol, reactor as _reactor
+from twisted.internet.interfaces import (
+    IDelayedCall,
+    IProcessTransport,
+    IReactorProcess,
+    IReactorTime,
+)
 from twisted.logger import Logger
 from twisted.protocols import basic
 from twisted.python import deprecate
+from twisted.python.failure import Failure
 
 
 @attr.s(frozen=True, auto_attribs=True)
@@ -46,7 +55,7 @@ class _Process:
     cwd: Optional[str] = None
 
     @deprecate.deprecated(incremental.Version("Twisted", 18, 7, 0))
-    def toTuple(self):
+    def toTuple(self) -> tuple[list[str], int | None, int | None, dict[str, str]]:
         """
         Convert process to tuple.
 
@@ -61,8 +70,6 @@ class _Process:
 
         This allows changing the internal structure of the process list,
         when warranted by bug fixes or additional features.
-
-        @return: tuple representation of process
         """
         return (self.args, self.uid, self.gid, self.env)
 
@@ -75,27 +82,39 @@ transport = DummyTransport()
 
 
 class LineLogger(basic.LineReceiver):
+    tag: str
+    stream: str
+    service: ProcessMonitor
+    delimiter: bytes
+
+    # These really ought to be set by a constructor, but the legacy API is a
+    # no-argument constructor.
     tag = None
     stream = None
     delimiter = b"\n"
     service = None
 
-    def lineReceived(self, line):
+    def lineReceived(self, line: bytes) -> None:
         try:
-            line = line.decode("utf-8")
+            sline = line.decode("utf-8")
         except UnicodeDecodeError:
-            line = repr(line)
+            sline = repr(line)
 
         self.service.log.info(
-            "[{tag}] {line}", tag=self.tag, line=line, stream=self.stream
+            "[{tag}] {line}", tag=self.tag, line=sline, stream=self.stream
         )
 
 
 class LoggingProtocol(protocol.ProcessProtocol):
+    service: ProcessMonitor
+    name: str
+
+    # These really ought to be set by a constructor, but the legacy API is a
+    # no-argument constructor.
     service = None
     name = None
 
-    def connectionMade(self):
+    def connectionMade(self) -> None:
         self._output = LineLogger()
         self._output.tag = self.name
         self._output.stream = "stdout"
@@ -111,56 +130,56 @@ class LoggingProtocol(protocol.ProcessProtocol):
         self._output.makeConnection(transport)
         self._error.makeConnection(transport)
 
-    def outReceived(self, data):
+    def outReceived(self, data: bytes) -> None:
         self._output.dataReceived(data)
         self._outputEmpty = data[-1] == b"\n"
 
-    def errReceived(self, data):
+    def errReceived(self, data: bytes) -> None:
         self._error.dataReceived(data)
         self._errorEmpty = data[-1] == b"\n"
 
-    def processEnded(self, reason):
+    def processEnded(self, reason: Failure) -> None:
         if not self._outputEmpty:
             self._output.dataReceived(b"\n")
         if not self._errorEmpty:
             self._error.dataReceived(b"\n")
-        self.service.connectionLost(self.name)
+        self.service._monitoredProcessExited(self.name, reason)
 
     @property
-    def output(self):
+    def output(self) -> LineLogger:
         return self._output
 
     @property
-    def empty(self):
+    def empty(self) -> bool:
         return self._outputEmpty
 
 
 class ProcessMonitor(service.Service):
     """
-    ProcessMonitor runs processes, monitors their progress, and restarts
-    them when they die.
+    ProcessMonitor runs processes, monitors their progress, and restarts them
+    when they die.
 
     The ProcessMonitor will not attempt to restart a process that appears to
     die instantly -- with each "instant" death (less than 1 second, by
-    default), it will delay approximately twice as long before restarting
-    it.  A successful run will reset the counter.
+    default), it will delay approximately twice as long before restarting it.
+    A successful run will reset the counter.
 
-    The primary interface is L{addProcess} and L{removeProcess}. When the
+    The primary interface is L{addProcess} and L{removeProcess}.  When the
     service is running (that is, when the application it is attached to is
     running), adding a process automatically starts it.
 
-    Each process has a name. This name string must uniquely identify the
-    process.  In particular, attempting to add two processes with the same
-    name will result in a C{KeyError}.
+    Each process has a name.  This name string must uniquely identify the
+    process.  In particular, attempting to add two processes with the same name
+    will result in a C{KeyError}.
 
     @type threshold: C{float}
     @ivar threshold: How long a process has to live before the death is
         considered instant, in seconds.  The default value is 1 second.
 
     @type killTime: C{float}
-    @ivar killTime: How long a process being killed has to get its affairs
-        in order before it gets killed with an unmaskable signal.  The
-        default value is 5 seconds.
+    @ivar killTime: How long a process being killed has to get its affairs in
+        order before it gets killed with an unmaskable signal.  The default
+        value is 5 seconds.
 
     @type minRestartDelay: C{float}
     @ivar minRestartDelay: The minimum time (in seconds) to wait before
@@ -171,13 +190,12 @@ class ProcessMonitor(service.Service):
         attempting to restart a process.  Default 3600s (1h).
 
     @type _reactor: L{IReactorProcess} provider
-    @ivar _reactor: A provider of L{IReactorProcess} and L{IReactorTime}
-        which will be used to spawn processes and register delayed calls.
+    @ivar _reactor: A provider of L{IReactorProcess} and L{IReactorTime} which
+        will be used to spawn processes and register delayed calls.
 
     @type log: L{Logger}
     @ivar log: The logger used to propagate log messages from spawned
         processes.
-
     """
 
     threshold = 1
@@ -186,18 +204,24 @@ class ProcessMonitor(service.Service):
     maxRestartDelay = 3600
     log = Logger()
 
-    def __init__(self, reactor=_reactor):
+    def __init__(
+        self,
+        reactor: IReactorProcess = _reactor,  # type:ignore
+    ) -> None:
         self._reactor = reactor
+        self._clock = IReactorTime(reactor)
 
-        self._processes = {}
-        self.protocols = {}
-        self.delay = {}
-        self.timeStarted = {}
-        self.murder = {}
-        self.restart = {}
+        self._processes: dict[str, _Process] = {}
+        self.protocols: dict[str, LoggingProtocol] = {}
+        self.delay: dict[str, float] = {}
+        self.timeStarted: dict[str, float] = {}
+        self.murder: dict[str, IDelayedCall] = {}
+        self.restart: dict[str, IDelayedCall] = {}
 
     @deprecate.deprecatedProperty(incremental.Version("Twisted", 18, 7, 0))
-    def processes(self):
+    def processes(
+        self,
+    ) -> dict[str, tuple[list[str], int | None, int | None, dict[str, str]]]:
         """
         Processes as dict of tuples
 
@@ -206,7 +230,7 @@ class ProcessMonitor(service.Service):
         return {name: process.toTuple() for name, process in self._processes.items()}
 
     @deprecate.deprecated(incremental.Version("Twisted", 18, 7, 0))
-    def __getstate__(self):
+    def __getstate__(self) -> Any:
         dct = service.Service.__getstate__(self)
         del dct["_reactor"]
         dct["protocols"] = {}
@@ -218,32 +242,41 @@ class ProcessMonitor(service.Service):
         dct["processes"] = self.processes
         return dct
 
-    def addProcess(self, name, args, uid=None, gid=None, env={}, cwd=None):
+    def addProcess(
+        self,
+        name: str,
+        args: list[str],
+        uid: int | None = None,
+        gid: int | None = None,
+        env: dict[str, str] = {},
+        cwd: str | None = None,
+    ) -> None:
         """
         Add a new monitored process and start it immediately if the
         L{ProcessMonitor} service is running.
 
-        Note that args are passed to the system call, not to the shell. If
+        Note that args are passed to the system call, not to the shell.  If
         running the shell is desired, the common idiom is to use
         C{ProcessMonitor.addProcess("name", ['/bin/sh', '-c', shell_script])}
 
-        @param name: A name for this process.  This value must be
-            unique across all processes added to this monitor.
-        @type name: C{str}
+        @param name: A name for this process.  This value must be unique across
+            all processes added to this monitor.
+
         @param args: The argv sequence for the process to launch.
-        @param uid: The user ID to use to run the process.  If L{None},
-            the current UID is used.
-        @type uid: C{int}
-        @param gid: The group ID to use to run the process.  If L{None},
-            the current GID is used.
-        @type uid: C{int}
-        @param env: The environment to give to the launched process. See
+
+        @param uid: The user ID to use to run the process.  If L{None}, the
+            current UID is used.
+
+        @param gid: The group ID to use to run the process.  If L{None}, the
+            current GID is used.
+
+        @param env: The environment to give to the launched process.  See
             L{IReactorProcess.spawnProcess}'s C{env} parameter.
-        @type env: C{dict}
-        @param cwd: The initial working directory of the launched process.
-            The default of C{None} means inheriting the laucnhing process's
-            working directory.
-        @type env: C{dict}
+
+        @param cwd: The initial working directory of the launched process.  The
+            default of C{None} means inheriting the laucnhing process's working
+            directory.
+
         @raise KeyError: If a process with the given name already exists.
         """
         if name in self._processes:
@@ -253,18 +286,17 @@ class ProcessMonitor(service.Service):
         if self.running:
             self.startProcess(name)
 
-    def removeProcess(self, name):
+    def removeProcess(self, name: str) -> None:
         """
         Stop the named process and remove it from the list of monitored
         processes.
 
-        @type name: C{str}
         @param name: A string that uniquely identifies the process.
         """
         self.stopProcess(name)
         del self._processes[name]
 
-    def startService(self):
+    def startService(self) -> None:
         """
         Start all monitored processes.
         """
@@ -272,7 +304,7 @@ class ProcessMonitor(service.Service):
         for name in list(self._processes):
             self.startProcess(name)
 
-    def stopService(self):
+    def stopService(self) -> None:
         """
         Stop all monitored processes and cancel all scheduled process restarts.
         """
@@ -286,22 +318,49 @@ class ProcessMonitor(service.Service):
         for name in list(self._processes):
             self.stopProcess(name)
 
-    def connectionLost(self, name):
+    @deprecate.deprecatedProperty(incremental.Version("Twisted", "NEXT", 0, 0))
+    def connectionLost(self, name: str) -> None:
+        """
+        Called when a monitored processes exits.  If
+        L{service.IService.running} is L{True} (ie the service is started), the
+        process will be restarted.  If the process had been running for more
+        than L{ProcessMonitor.threshold} seconds it will be restarted
+        immediately.  If the process had been running for less than
+        L{ProcessMonitor.threshold} seconds, the restart will be delayed and
+        each time the process dies before the configured threshold, the restart
+        delay will be doubled - up to a maximum delay of maxRestartDelay sec.
+
+        @param name: A string that uniquely identifies the process which
+            exited.
         """
-        Called when a monitored processes exits. If
+        return self._monitoredProcessExited(name)  # pragma: no cover
+
+    def _monitoredProcessExited(self, name: str, reason: Failure | None = None) -> None:
+        """
+        Called when a monitored processes exits.  If
         L{service.IService.running} is L{True} (ie the service is started), the
-        process will be restarted.
-        If the process had been running for more than
-        L{ProcessMonitor.threshold} seconds it will be restarted immediately.
-        If the process had been running for less than
+        process will be restarted.  If the process had been running for more
+        than L{ProcessMonitor.threshold} seconds it will be restarted
+        immediately.  If the process had been running for less than
         L{ProcessMonitor.threshold} seconds, the restart will be delayed and
         each time the process dies before the configured threshold, the restart
         delay will be doubled - up to a maximum delay of maxRestartDelay sec.
 
-        @type name: C{str}
-        @param name: A string that uniquely identifies the process
-            which exited.
+        @param name: A string that uniquely identifies the process which
+            exited.
+
+        @param reason: The reason why the connection was lost.
         """
+        # Log a warning if reason is something other than ProcessDone
+        if reason is not None and not reason.check(
+            error.ProcessDone,
+            error.ProcessTerminated,
+        ):
+            self.log.failure(
+                "Process '{name}' has exited: {failure!r}",
+                failure=reason,
+                name=name,
+            )
         # Cancel the scheduled _forceStopProcess function if the process
         # dies naturally
         if name in self.murder:
@@ -311,7 +370,7 @@ class ProcessMonitor(service.Service):
 
         del self.protocols[name]
 
-        if self._reactor.seconds() - self.timeStarted[name] < self.threshold:
+        if self._clock.seconds() - self.timeStarted[name] < self.threshold:
             # The process died too fast - backoff
             nextDelay = self.delay[name]
             self.delay[name] = min(self.delay[name] * 2, self.maxRestartDelay)
@@ -324,11 +383,11 @@ class ProcessMonitor(service.Service):
 
         # Schedule a process restart if the service is running
         if self.running and name in self._processes:
-            self.restart[name] = self._reactor.callLater(
+            self.restart[name] = self._clock.callLater(
                 nextDelay, self.startProcess, name
             )
 
-    def startProcess(self, name):
+    def startProcess(self, name: str) -> None:
         """
         @param name: The name of the process to be started
         """
@@ -343,27 +402,27 @@ class ProcessMonitor(service.Service):
         proto.service = self
         proto.name = name
         self.protocols[name] = proto
-        self.timeStarted[name] = self._reactor.seconds()
-        self._reactor.spawnProcess(
-            proto,
-            process.args[0],
-            process.args,
-            uid=process.uid,
-            gid=process.gid,
-            env=process.env,
-            path=process.cwd,
-        )
+        self.timeStarted[name] = self._clock.seconds()
+        try:
+            self._reactor.spawnProcess(
+                proto,
+                process.args[0],
+                process.args,
+                uid=process.uid,
+                gid=process.gid,
+                env=process.env,
+                path=process.cwd,
+            )
+        except OSError:
+            self._monitoredProcessExited(name, Failure())
 
-    def _forceStopProcess(self, proc):
-        """
-        @param proc: An L{IProcessTransport} provider
-        """
+    def _forceStopProcess(self, proc: IProcessTransport) -> None:
         try:
             proc.signalProcess("KILL")
         except error.ProcessExitedAlready:
             pass
 
-    def stopProcess(self, name):
+    def stopProcess(self, name: str) -> None:
         """
         @param name: The name of the process to be stopped
         """
@@ -373,16 +432,17 @@ class ProcessMonitor(service.Service):
         proto = self.protocols.get(name, None)
         if proto is not None:
             proc = proto.transport
+            assert proc is not None
             try:
                 proc.signalProcess("TERM")
             except error.ProcessExitedAlready:
                 pass
             else:
-                self.murder[name] = self._reactor.callLater(
+                self.murder[name] = self._clock.callLater(
                     self.killTime, self._forceStopProcess, proc
                 )
 
-    def restartAll(self):
+    def restartAll(self) -> None:
         """
         Restart all processes. This is useful for third party management
         services to allow a user to restart servers because of an outside change


Test patch:
diff --git a/src/twisted/runner/test/test_procmon.py b/src/twisted/runner/test/test_procmon.py
index 585cb0248f..bf65ef15a8 100644
--- a/src/twisted/runner/test/test_procmon.py
+++ b/src/twisted/runner/test/test_procmon.py
@@ -4,9 +4,17 @@
 """
 Tests for L{twisted.runner.procmon}.
 """
+from __future__ import annotations
+
+import errno
+import os
 import pickle
+from typing import AnyStr, Mapping, Optional, Sequence, Union
+
+from zope.interface import implementer
 
 from twisted.internet.error import ProcessDone, ProcessExitedAlready, ProcessTerminated
+from twisted.internet.interfaces import IProcessProtocol, IProcessTransport
 from twisted.internet.task import Clock
 from twisted.internet.testing import MemoryReactor
 from twisted.logger import globalLogPublisher
@@ -15,7 +23,8 @@ from twisted.runner.procmon import LoggingProtocol, ProcessMonitor
 from twisted.trial import unittest
 
 
-class DummyProcess:
+@implementer(IProcessTransport)
+class DummyProcess:  # type:ignore[misc]
     """
     An incomplete and fake L{IProcessTransport} implementation for testing how
     L{ProcessMonitor} behaves when its monitored processes exit.
@@ -93,31 +102,37 @@ class DummyProcessReactor(MemoryReactor, Clock):
     """
     @ivar spawnedProcesses: a list that keeps track of the fake process
         instances built by C{spawnProcess}.
-    @type spawnedProcesses: C{list}
+
+    @ivar spawnProcessException: An exception which spawnProcess() will raise.
     """
 
-    def __init__(self):
+    spawnedProcesses: list[DummyProcess]
+    spawnProcessException: Exception | None
+
+    def __init__(self) -> None:
         MemoryReactor.__init__(self)
         Clock.__init__(self)
-
         self.spawnedProcesses = []
+        self.spawnProcessException = None
 
     def spawnProcess(
         self,
-        processProtocol,
-        executable,
-        args=(),
-        env={},
-        path=None,
-        uid=None,
-        gid=None,
-        usePTY=0,
-        childFDs=None,
-    ):
+        processProtocol: IProcessProtocol,
+        executable: Union[bytes, str],
+        args: Sequence[Union[bytes, str]],
+        env: Optional[Mapping[AnyStr, AnyStr]] = None,
+        path: Union[None, bytes, str] = None,
+        uid: Optional[int] = None,
+        gid: Optional[int] = None,
+        usePTY: bool = False,
+        childFDs: Optional[Mapping[int, Union[int, str]]] = None,
+    ) -> IProcessTransport:
         """
         Fake L{reactor.spawnProcess}, that logs all the process
         arguments and returns a L{DummyProcess}.
         """
+        if self.spawnProcessException is not None:
+            raise self.spawnProcessException
 
         proc = DummyProcess(
             self,
@@ -279,6 +294,31 @@ class ProcmonTests(unittest.TestCase):
         """
         self.assertRaises(KeyError, self.pm.startProcess, "foo")
 
+    def test_startProcessSpawnCaughtException(self) -> None:
+        """
+        L{IReactorProcess.spawnProcess} might raise C{OSError}, ensure it
+        is caught and process is restarted.
+        """
+        self.reactor.spawnProcessException = OSError(
+            errno.EAGAIN, os.strerror(errno.EAGAIN)
+        )
+        self.pm.minRestartDelay = 123
+        self.pm.maxRestartDelay = 123
+        self.pm.addProcess("foo", ["foo"])
+        self.pm.startProcess("foo")
+        # process will be restarted in 123 seconds, per minRestartDelay and maxRestartDelay above
+        self.assertEquals(self.pm.delay["foo"], 123)
+        self.assertEquals(len(self.flushLoggedErrors(OSError)), 1)
+
+    def test_startProcessSpawnUncaughtException(self) -> None:
+        """
+        L{IReactorProcess.spawnProcess} might raise C{OSError}, ensure other
+        exceptions are not caught.
+        """
+        self.reactor.spawnProcessException = SystemError("Just another exception")
+        self.pm.addProcess("foo", ["foo"])
+        self.assertRaises(SystemError, self.pm.startProcess, "foo")
+
     def test_stopProcessNaturalTermination(self):
         """
         L{ProcessMonitor.stopProcess} immediately sends a TERM signal to the
@@ -467,7 +507,7 @@ class ProcmonTests(unittest.TestCase):
 
     def test_connectionLostLongLivedProcess(self):
         """
-        L{ProcessMonitor.connectionLost} should immediately restart a process
+        L{ProcessMonitor.processExit} should immediately restart a process
         if it has been running longer than L{ProcessMonitor.threshold} seconds.
         """
         self.pm.addProcess("foo", ["foo"])
@@ -487,7 +527,7 @@ class ProcmonTests(unittest.TestCase):
 
     def test_connectionLostMurderCancel(self):
         """
-        L{ProcessMonitor.connectionLost} cancels a scheduled process killer and
+        L{ProcessMonitor.processExit} cancels a scheduled process killer and
         deletes the DelayedCall from the L{ProcessMonitor.murder} list.
         """
         self.pm.addProcess("foo", ["foo"])
@@ -508,7 +548,7 @@ class ProcmonTests(unittest.TestCase):
 
     def test_connectionLostProtocolDeletion(self):
         """
-        L{ProcessMonitor.connectionLost} removes the corresponding
+        L{ProcessMonitor.processExit} removes the corresponding
         ProcessProtocol instance from the L{ProcessMonitor.protocols} list.
         """
         self.pm.startService()
@@ -520,7 +560,7 @@ class ProcmonTests(unittest.TestCase):
 
     def test_connectionLostMinMaxRestartDelay(self):
         """
-        L{ProcessMonitor.connectionLost} will wait at least minRestartDelay s
+        L{ProcessMonitor.processExit} will wait at least minRestartDelay s
         and at most maxRestartDelay s
         """
         self.pm.minRestartDelay = 2
@@ -536,7 +576,7 @@ class ProcmonTests(unittest.TestCase):
 
     def test_connectionLostBackoffDelayDoubles(self):
         """
-        L{ProcessMonitor.connectionLost} doubles the restart delay each time
+        L{ProcessMonitor.processExit} doubles the restart delay each time
         the process dies too quickly.
         """
         self.pm.startService()
@@ -642,7 +682,6 @@ class ProcmonTests(unittest.TestCase):
 
 
 class DeprecationTests(unittest.SynchronousTestCase):
-
     """
     Tests that check functionality that should be deprecated is deprecated.
     """