Timeout: introduce publisher_timeout configuration.

Vincent Pelletier:
- Ignore TimerService-originated publication requests.
- Add copyright headers.
- Disable timeout by default.
- Give a meaning to requested negative deadlines.
- Docstrings.
- Assorted simplifications.

product/ERP5Form/tests/testListBox.py

@@ -738,6 +738,77 @@ class TestListBox(ERP5TypeTestCase):
     self.assertEqual(form.getId(), request.get('listbox_form_id'))
     self.assertEqual(form.listbox.getId(), request.get('listbox_field_id'))
 
+  def test_query_timeout(self):
+    portal = self.getPortal()
+    portal.ListBoxZuite_reset()
+
+    # Set short enough publisher timeout configuration
+    import Products.ERP5Type.Timeout
+    Products.ERP5Type.Timeout.publisher_timeout = 2.0
+
+    # We create a Z SQL Method that takes too long
+    list_method_id = 'ListBox_zSlowQuery'
+    portal.portal_skins.custom.manage_addProduct['ZSQLMethods'].manage_addZSQLMethod(
+      id=list_method_id,
+      title='',
+      connection_id='erp5_sql_connection',
+      arguments='',
+      template="SELECT uid, path FROM catalog WHERE SLEEP(3) = 0 AND path='%s'" % portal.foo_module.getPath(),
+    )
+    portal.changeSkin(None)
+
+    # set the listbox to use this as list method
+    listbox = portal.FooModule_viewFooList.listbox
+    listbox.ListBox_setPropertyList(
+      field_list_method=list_method_id,
+      field_count_method='',
+    )
+
+    # access the form
+    result = self.publish(
+      '%s/FooModule_viewFooList' % portal.foo_module.absolute_url(relative=True),
+      'ERP5TypeTestCase:',
+    )
+    self.assertEqual(result.getStatus(), 500)
+    self.assertTrue('Error Type: TimeoutReachedError' in result.getBody())
+    self.assertTrue('Error Value: 1969: Query execution was interrupted (max_statement_time exceeded): SET STATEMENT' in result.getBody())
+
+  def test_zodb_timeout(self):
+    portal = self.getPortal()
+    portal.ListBoxZuite_reset()
+
+    # Set short enough publisher timeout configuration
+    import Products.ERP5Type.Timeout
+    Products.ERP5Type.Timeout.publisher_timeout = 2.0
+
+    # We create a Z SQL Method that takes too long
+    list_method_id = 'ListBox_getSlowObjectValues'
+    createZODBPythonScript(
+        portal.portal_skins.custom,
+        list_method_id,
+        'selection=None, **kw',
+        """
+from time import sleep
+sleep(3)
+return context.objectValues()
+        """)
+
+    # set the listbox to use this as list method
+    listbox = portal.FooModule_viewFooList.listbox
+    listbox.ListBox_setPropertyList(
+      field_list_method=list_method_id,
+      field_count_method='',
+    )
+
+    portal.foo_module.newContent()
+
+    # access the form
+    result = self.publish(
+      '%s/FooModule_viewFooList' % portal.foo_module.absolute_url(relative=True),
+      'ERP5TypeTestCase:',
+    )
+    self.assertEqual(result.getStatus(), 500)
+    self.assertTrue('Error Type: TimeoutReachedError' in result.getBody())
 
 def test_suite():
   suite = unittest.TestSuite()

product/ERP5Type/Timeout.py

+##############################################################################
+# Copyright (c) 2019 Nexedi SA and Contributors. All Rights Reserved.
+#                     Kazuhiko <kazuhiko@nexedi.com>
+#
+# WARNING: This program as such is intended to be used by professional
+# programmers who take the whole responsability of assessing all potential
+# consequences resulting from its eventual inadequacies and bugs
+# End users who are looking for a ready-to-use solution with commercial
+# garantees and support are strongly adviced to contract a Free Software
+# Service Company
+#
+# This program is Free Software; you can redistribute it and/or
+# modify it under the terms of the GNU General Public License
+# as published by the Free Software Foundation; either version 2
+# of the License, or (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program; if not, write to the Free Software
+# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
+#
+##############################################################################
+from contextlib import contextmanager
+import threading
+import time
+from AccessControl.SecurityInfo import ModuleSecurityInfo
+from ZPublisher.HTTPResponse import status_codes
+from Products.TimerService.timerserver.TimerServer import TimerRequest
+
+__all__ = (
+  'TimeoutReachedError', 'Deadline', 'getDeadline', 'getTimeLeft',
+  'wrap_call_object',
+)
+
+class TimeoutReachedError(Exception):
+  """
+  A deadline was reached.
+  """
+  pass
+
+# There is no appropriate status code for timeout.
+status_codes['timeoutreachederror'] = 500 # Internal Server Error
+del status_codes
+
+# Placeholder timeouts until product's "initialize" is called.
+publisher_timeout = None
+
+_site_local = threading.local()
+
+@contextmanager
+def Deadline(offset):
+  """
+  Context manager for defining the code-wise scope of a deadline.
+  offset (float)
+    Number of seconds the context-managed piece of code should be allowed to
+    run for. Positive values are based on current time, while negative values
+    are based on pre-existing deadline, and no deadline will be set if none
+    was pre-existing.
+    There is no automated enforcement of this delay, it is up to the code to
+    check whether it exceeded the allotted preiod, and to raise
+    TimeoutReachedError.
+    If None, it has no effect on a possible current deadline, for caller code
+    simplicity.
+  """
+  if offset is None:
+    yield
+  else:
+    old_deadline = getattr(_site_local, 'deadline', None)
+    if old_deadline is None:
+      if offset >= 0:
+        _site_local.deadline = time.time() + offset
+    elif offset < 0:
+      _site_local.deadline = old_deadline + offset
+    else:
+      # Ignore attempts to extend an existing deadline.
+      _site_local.deadline = min(old_deadline, time.time() + offset)
+    try:
+      yield
+    finally:
+      _site_local.deadline = old_deadline
+
+ModuleSecurityInfo('Products.ERP5Type.Timeout').declarePublic('Deadline')
+
+def getDeadline():
+  """
+  Return currently-applicable deadline as a timestamp, or None if there is
+  no currently applicable deadline.
+  """
+  return getattr(_site_local, 'deadline', None)
+
+def getTimeLeft():
+  """
+  Return the number of seconds left until current deadline, or None if there is
+  no currently applicable deadline.
+  """
+  deadline = getattr(_site_local, 'deadline', None)
+  return None if deadline is None else max(deadline - time.time(), 0.000001)
+
+def wrap_call_object(call_object):
+  """
+  Return argument wrapped so it is executed in a timeout context using
+  publisher_timeout.
+
+  call_object (callable (object, args, request) -> any)
+    call_object-like function, which should be executed under a deadline based
+    on publisher_timeout value at call-time.
+    Dedline will not be applied if request is a TimerRequest instance, as these
+    requests are not strictly published, and hence do not fall under control of
+    the same setting.
+  """
+  def deadlined_call_object(object, args, request):
+    with Deadline(
+      None if isinstance(request, TimerRequest) else publisher_timeout,
+    ):
+      return call_object(object, args, request)
+  return deadlined_call_object

product/ERP5Type/ZopePatch.py

@@ -90,6 +90,7 @@ from Products.ERP5Type.patches import ZopePageTemplate
 from Products.ERP5Type.patches import ZSQLMethod
 from Products.ERP5Type.patches import MimetypesRegistry
 from Products.ERP5Type.patches import users
+from Products.ERP5Type.patches import Publish
 
 # These symbols are required for backward compatibility
 from Products.ERP5Type.patches.PropertyManager import ERP5PropertyManager

product/ERP5Type/__init__.py

@@ -30,6 +30,7 @@
     ERP5Type is provides a RAD environment for Zope / CMF
     All ERP5 classes derive from ERP5Type
 """
+from App.config import getConfiguration
 from patches import python, pylint, globalrequest
 from zLOG import LOG, INFO
 DISPLAY_BOOT_PROCESS = False
@@ -156,6 +157,11 @@ def initialize( context ):
     LOG('ERP5Type.__init__', INFO, 'installInteractorClassRegistry')
   installInteractorClassRegistry()
 
+  from Products.ERP5Type import Timeout
+  erp5_conf = getattr(getConfiguration(), 'product_config', {}).get('erp5')
+  # Note: erp5_conf attributes are missing in unit tests, fallback to no timeout
+  # in that case.
+  Timeout.publisher_timeout = getattr(erp5_conf, 'publisher_timeout', None)
 
 from AccessControl.SecurityInfo import allow_module
 from AccessControl.SecurityInfo import ModuleSecurityInfo

product/ERP5Type/component.xml

@@ -11,5 +11,10 @@
         Description
       </description>
     </key>
+    <key name="publisher-timeout" required="no" datatype="integer">
+      <description>
+        If a request takes more than this value (in seconds), data connection can be terminated.
+      </description>
+    </key>
   </sectiontype>
 </component>

product/ERP5Type/patches/Publish.py

+##############################################################################
+# Copyright (c) 2019 Nexedi SA and Contributors. All Rights Reserved.
+#                     Kazuhiko <kazuhiko@nexedi.com>
+#
+# WARNING: This program as such is intended to be used by professional
+# programmers who take the whole responsability of assessing all potential
+# consequences resulting from its eventual inadequacies and bugs
+# End users who are looking for a ready-to-use solution with commercial
+# garantees and support are strongly adviced to contract a Free Software
+# Service Company
+#
+# This program is Free Software; you can redistribute it and/or
+# modify it under the terms of the GNU General Public License
+# as published by the Free Software Foundation; either version 2
+# of the License, or (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program; if not, write to the Free Software
+# Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA  02111-1307, USA.
+#
+##############################################################################
+from Products.ERP5Type.Timeout import wrap_call_object
+from ZPublisher import Publish
+
+call_object_orig = Publish.call_object
+call_object = wrap_call_object(call_object_orig)
+Publish.call_object = call_object
+
+publish = Publish.__dict__['publish']
+assert publish.__module__ == 'ZPublisher.Publish', repr(publish.__module__)
+if publish.__name__ == 'new_publish': # already patched by Localizer/patches.py
+  publish = publish.__defaults__[1]
+publish.__defaults__ = tuple(call_object if x is call_object_orig else x for x in publish.__defaults__)

product/ERP5Type/patches/WSGIPublisher.py

@@ -33,6 +33,7 @@ from AccessControl.SecurityManagement import noSecurityManager
 from Acquisition import aq_acquire
 from Acquisition import aq_inner
 from Acquisition import aq_parent
+from Products.ERP5Type.Timeout import wrap_call_object
 from transaction.interfaces import TransientError
 from zExceptions import Redirect
 from zExceptions import Unauthorized
@@ -49,7 +50,8 @@ from ZPublisher import pubevents, Retry
 from ZPublisher.HTTPRequest import HTTPRequest
 from ZPublisher.Iterators import IUnboundStreamIterator
 from ZPublisher.mapply import mapply
-from ZPublisher.WSGIPublisher import call_object, missing_name, WSGIResponse
+from ZPublisher.WSGIPublisher import call_object as call_object_orig
+from ZPublisher.WSGIPublisher import missing_name, WSGIResponse
 
 
 if sys.version_info >= (3, ):
@@ -61,6 +63,7 @@ _DEFAULT_DEBUG_MODE = False
 _DEFAULT_REALM = None
 _MODULE_LOCK = allocate_lock()
 _MODULES = {}
+call_object = wrap_call_object(call_object_orig)
 
 
 AC_LOGGER = logging.getLogger('event.AccessControl')

product/ERP5Type/patches/ZODBConnection.py

@@ -17,6 +17,8 @@
 # transactions so that, for instance, activity processing can see a recent
 # enough state of the ZODB.
 
+import time
+from Products.ERP5Type.Timeout import TimeoutReachedError, getDeadline
 from ZODB.Connection import Connection
 
 FORCE_STORAGE_SYNC_ON_CONNECTION_OPENING = False
@@ -70,3 +72,12 @@ if 1: # keep indentation. Also good for quick disabling.
 
         Connection_open = Connection.open
         Connection.open = open
+
+    setstate_orig = Connection.setstate
+    def setstate(self, obj):
+        deadline = getDeadline()
+        if deadline is not None and deadline < time.time():
+          raise TimeoutReachedError
+        setstate_orig(self, obj)
+
+    Connection.setstate = setstate

product/ERP5Type/tests/testTimeout.py

+##############################################################################
+#
+# Copyright (c) 2019 Nexedi SA and Contributors. All Rights Reserved.
+#
+# WARNING: This program as such is intended to be used by professional
+# programmers who take the whole responsability of assessing all potential
+# consequences resulting from its eventual inadequacies and bugs
+# End users who are looking for a ready-to-use solution with commercial
+# garantees and support are strongly adviced to contract a Free Software
+# Service Company
+#
+# This program is Free Software; you can redistribute it and/or
+# modify it under the terms of the GNU General Public License
+# as published by the Free Software Foundation; either version 2
+# of the License, or (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program; if not, write to the Free Software
+# Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA  02111-1307, USA.
+#
+##############################################################################
+from Products.ERP5Type.tests.ERP5TypeTestCase import ERP5TypeTestCase
+from Products.ERP5Type.Timeout import TimeoutReachedError, Deadline
+import time
+
+class TestTimeout(ERP5TypeTestCase):
+  """
+    Test MariaDB / ZODB Deadline
+  """
+  def getBusinessTemplateList(self):
+    return ()
+
+  def getTitle(self):
+    return "Query Deadline"
+
+  def afterSetUp(self):
+    self.login()
+
+  def test_query_deadline(self):
+    # We create a Z SQL Method that takes too long
+    method_id = 'Base_zSlowQuery'
+    self.portal.portal_skins.custom.manage_addProduct['ZSQLMethods'].manage_addZSQLMethod(
+      id=method_id,
+      title='',
+      connection_id='erp5_sql_connection',
+      arguments='t',
+      template="SELECT uid, path FROM catalog WHERE SLEEP(<dtml-var t>) = 0 AND path='%s'" % self.portal.portal_templates.getPath(),
+    )
+    self.portal.changeSkin(None)
+    # finish within deadline
+    with Deadline(3.0):
+      getattr(self.portal, method_id)(t=2)
+    # query timeout by deadline
+    with Deadline(1.0):
+      with self.assertRaises(TimeoutReachedError):
+        getattr(self.portal, method_id)(t=2)
+    # can be nested, but cannot extend the deadline
+    with Deadline(1.0):
+      with Deadline(3.0):
+        with self.assertRaises(TimeoutReachedError):
+          getattr(self.portal, method_id)(t=2)
+    with Deadline(5.0):
+      with Deadline(1.0):
+        with self.assertRaises(TimeoutReachedError):
+          getattr(self.portal, method_id)(t=2)
+      # not yet reached for outer deadline
+      getattr(self.portal, method_id)(t=1)
+
+  def test_zodb_deadline(self):
+    with Deadline(1.0):
+      time.sleep(2)
+      with self.assertRaises(TimeoutReachedError):
+        [x.getObject() for x in self.portal.portal_templates.searchFolder()]

product/ZMySQLDA/db.py

@@ -93,6 +93,7 @@ import MySQLdb
 import warnings
 from contextlib import contextmanager, nested
 from _mysql_exceptions import OperationalError, NotSupportedError, ProgrammingError
+from Products.ERP5Type.Timeout import TimeoutReachedError, getTimeLeft
 MySQLdb_version_required = (0,9,2)
 
 _v = getattr(_mysql, 'version_info', (0,0,0))
@@ -122,6 +123,10 @@ lock_error = (
     ER.LOCK_DEADLOCK,
     )
 
+query_timeout_error = (
+    1969, # ER_STATEMENT_TIMEOUT in MariaDB
+)
+
 key_types = {
     "PRI": "PRIMARY KEY",
     "MUL": "INDEX",
@@ -194,6 +199,11 @@ def ord_or_None(s):
     if s is not None:
         return ord(s)
 
+match_select = re.compile(
+    r'(?:SET\s+STATEMENT\s+(.+?)\s+FOR\s+)?SELECT\s+(.+)',
+    re.IGNORECASE | re.DOTALL,
+).match
+
 class DB(TM):
     """This is the ZMySQLDA Database Connection Object."""
 
@@ -379,6 +389,8 @@ class DB(TM):
               raise OperationalError(m[0], '%s: %s' % (m[1], query))
             if m[0] in lock_error:
               raise ConflictError('%s: %s: %s' % (m[0], m[1], query))
+            if m[0] in query_timeout_error:
+              raise TimeoutReachedError('%s: %s: %s' % (m[0], m[1], query))
             if (allow_reconnect or not self._use_TM) and \
               m[0] in hosed_connection:
               self._forceReconnection()
@@ -389,7 +401,13 @@ class DB(TM):
         except ProgrammingError:
           LOG('ZMySQLDA', ERROR, 'query failed: %s' % (query,))
           raise
-        return self.db.store_result()
+        try:
+          return self.db.store_result()
+        except OperationalError, m:
+          if m[0] in query_timeout_error:
+            raise TimeoutReachedError('%s: %s: %s' % (m[0], m[1], query))
+          else:
+            raise
 
     def query(self, query_string, max_rows=1000):
         """Execute 'query_string' and return at most 'max_rows'."""
@@ -403,8 +421,18 @@ class DB(TM):
         for qs in query_string.split('\0'):
             qs = qs.strip()
             if qs:
-                if qs[:6].upper() == "SELECT" and max_rows:
-                    qs = "%s LIMIT %d" % (qs, max_rows)
+                select_match = match_select(qs)
+                if select_match:
+                    query_timeout = getTimeLeft()
+                    if query_timeout is not None:
+                        statement, select = select_match.groups()
+                        if statement:
+                            statement += ", max_statement_time=%f" % query_timeout
+                        else:
+                            statement = "max_statement_time=%f" % query_timeout
+                        qs = "SET STATEMENT %s FOR SELECT %s" % (statement, select)
+                    if max_rows:
+                        qs = "%s LIMIT %d" % (qs, max_rows)
                 c = self._query(qs)
                 if c:
                     if desc is not None is not c.describe():
@@ -606,12 +634,12 @@ class DeferredDB(DB):
         assert self._use_TM
         self._sql_string_list = []
 
-    def query(self,query_string, max_rows=1000):
+    def query(self, query_string, max_rows=1000):
         self._register()
         for qs in query_string.split('\0'):
             qs = qs.strip()
             if qs:
-                if qs[:6].upper() == "SELECT":
+                if match_select(qs):
                     raise NotSupportedError(
                         "can not SELECT in deferred connections")
                 self._sql_string_list.append(qs)

product/ZSQLCatalog/SQLCatalog.py

@@ -2280,6 +2280,7 @@ class Catalog(Folder,
         # XXX should get zsql_brain from ZSQLMethod class itself
         zsql_brain=None,
         implicit_join=False,
+        query_timeout=None,
         **kw
       ):
     if build_sql_query_method is None:
@@ -2301,6 +2302,7 @@ class Catalog(Folder,
       from_expression=query['from_expression'],
       sort_on=query['order_by_expression'],
       limit_expression=query['limit_expression'],
+      query_timeout=query_timeout,
     )
 
   def getSqlSearchResults(self):