Fix commit order of CMFActivity SQL connection on nodes with several zserver threads

When a ZODB connection is closed, it usually returns to a ZODB pool and may be
reused by another thread. If the SQL connection was open and is still in ZODB
cache, the _v_database_connection attribute is still there:
ActivityConnection.connect() is not called and a new instance of ZMySQLDA.db.DB
is created for the new thread without initializing its sort key.

product/CMFActivity/ActivityConnection.py

@@ -26,13 +26,13 @@
 #
 ##############################################################################
 
-from Products.ZMySQLDA.DA import Connection
+from Products.ZMySQLDA.DA import Connection, ThreadedDB
 from Products.ERP5Type.Globals import InitializeClass
 from App.special_dtml import HTMLFile
 from Acquisition import aq_parent
 
-# If the sort order below doesn't work, we cannot guarantee the setSortKey()
-# call below will actually result in the activity connection being committed
+# If the sort order below doesn't work, we cannot guarantee the sort key
+# used below will actually result in the activity connection being committed
 # after the ZODB and Catalog data.
 assert None < 0 < '' < (), "Cannot guarantee commit of activities comes after the appropriate data"
 
@@ -58,15 +58,12 @@ class ActivityConnection(Connection):
     # reuse the permission from ZMySQLDA
     permission_type = 'Add Z MySQL Database Connections'
 
-    def connect(self, s):
-        result = Connection.connect(self, s)
-        if aq_parent(self) is None:
-            # Connection.connect() doesn't set _v_database_connection if there
-            # are no acquisition wrappers
-            return result
-        # the call above will set self._v_database_connection, and it won't
-        # have disappeared by now.
-        self._v_database_connection.setSortKey( (0,) )
-        return result
+    def factory(self):
+        return ActivityThreadedDB
 
 InitializeClass(ActivityConnection)
+
+
+class ActivityThreadedDB(ThreadedDB):
+
+    _sort_key = (0,)

product/CMFActivity/tests/testCMFActivity.py

@@ -3504,6 +3504,28 @@ class TestCMFActivity(ERP5TypeTestCase, LogInterceptor):
     newconn = portal.cmf_activity_sql_connection
     self.assertEquals(newconn.meta_type, 'CMFActivity Database Connection')
 
+  def test_connection_sortkey(self):
+    """
+    Check that SQL connection has properly initialized sort key,
+    even when its container (ZODB connection) is reused by another thread.
+    """
+    def sortKey():
+      app = ZopeTestCase.app()
+      try:
+        c = app[self.getPortalName()].cmf_activity_sql_connection()
+        return app._p_jar, c._access_db('sortKey', (), {})
+      finally:
+        ZopeTestCase.close(app)
+    jar, sort_key = sortKey()
+    self.assertNotEqual(1, sort_key)
+    result = []
+    t = threading.Thread(target=lambda: result.extend(sortKey()))
+    t.daemon = True
+    t.start()
+    t.join()
+    self.assertTrue(result[0] is jar)
+    self.assertEqual(result[1], sort_key)
+
   def test_onErrorCallback(self):
     activity_tool = self.portal.portal_activities
     obj = activity_tool.newActiveProcess()

product/ZMySQLDA/db.py

@@ -184,6 +184,8 @@ class ThreadedDB:
     conv[FIELD_TYPE.BIT] = ord_or_None
     del conv[FIELD_TYPE.TIME]
 
+    _sort_key = TM._sort_key
+
     def __init__(self,connection):
         """
           Parse the connection string.
@@ -290,6 +292,7 @@ class ThreadedDB:
             db = DB(kw_args=self._kw_args, use_TM=self._use_TM,
                     mysql_lock=self._mysql_lock,
                     transactions=self._transactions)
+            db.setSortKey(self._sort_key)
             self._pool_set(ident, db)
         return getattr(db, method_id)(*args, **kw)
 
@@ -305,9 +308,6 @@ class ThreadedDB:
     def string_literal(self, *args, **kw):
         return self._access_db(method_id='string_literal', args=args, kw=kw)
 
-    def setSortKey(self, *args, **kw):
-        return self._access_db(method_id='setSortKey', args=args, kw=kw)
-
 
 class DB(TM):