From dfe37be952c260a00b400676afff80c46b47bf8d Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?C=C3=A9dric=20Krier?= <ced@b2ck.com>
Date: Mon, 14 Feb 2022 23:12:00 +0100
Subject: [PATCH] Avoid concurrency failure in cache with channel test

We need to ensure that the listener is running on the return of the transaction
start otherwise notification can be missed.
We must also just from time to time check that the listener is still the
current listener.
And finally we can not wait an arbitrary time for cache sync, instead we must
rely on the _clean_last being reset with the sync happens.

issue9413
review364831002
(grafted from a32b84d9664917df3274663c06ab57fd5a7789ef)

--HG--
branch : 6.0
---
 trytond/cache.py            |  7 ++++++-
 trytond/tests/test_cache.py | 19 +++++++++++--------
 2 files changed, 17 insertions(+), 9 deletions(-)

diff --git a/trytond/cache.py b/trytond/cache.py
index 7934f4ef8a..2d2ffb43e1 100644
--- a/trytond/cache.py
+++ b/trytond/cache.py
@@ -6,6 +6,7 @@ import logging
 import os
 import select
 import threading
+import time
 from collections import OrderedDict, defaultdict
 from datetime import datetime
 from weakref import WeakKeyDictionary
@@ -216,6 +217,9 @@ class MemoryCache(BaseCache):
                     cls._listener[pid, dbname] = listener = threading.Thread(
                         target=cls._listen, args=(dbname,), daemon=True)
                     listener.start()
+                    while (not getattr(listener, 'listening', False)
+                            and listener.is_alive()):
+                        time.sleep(.01)
             return
         if (datetime.now() - cls._clean_last).total_seconds() < _clear_timeout:
             return
@@ -354,9 +358,10 @@ class MemoryCache(BaseCache):
         try:
             cursor = conn.cursor()
             cursor.execute('LISTEN "%s"' % cls._channel)
+            current_thread.listening = True
 
             while cls._listener.get((pid, dbname)) == current_thread:
-                readable, _, _ = select.select([conn], [], [])
+                readable, _, _ = select.select([conn], [], [], 60)
                 if not readable:
                     continue
 
diff --git a/trytond/tests/test_cache.py b/trytond/tests/test_cache.py
index 87b0351f66..8bd91e0a38 100644
--- a/trytond/tests/test_cache.py
+++ b/trytond/tests/test_cache.py
@@ -1,6 +1,7 @@
 # This file is part of Tryton.  The COPYRIGHT file at the top level of
 # this repository contains the full copyright notices and license terms.
 
+import datetime as dt
 import time
 import unittest
 
@@ -58,7 +59,7 @@ class MemoryCacheTestCase(unittest.TestCase):
     def tearDown(self):
         MemoryCache.drop(DB_NAME)
 
-    def wait_cache_sync(self):
+    def wait_cache_sync(self, after=None):
         pass
 
     @with_transaction()
@@ -96,11 +97,9 @@ class MemoryCacheTestCase(unittest.TestCase):
         with Transaction().set_current_transaction(transaction1):
             self.assertEqual(cache.get('foo'), 'bar')
 
+        commit_time = dt.datetime.now()
         transaction2.commit()
-        for n in range(10):
-            if cache.get('foo') == 'baz':
-                break
-            self.wait_cache_sync()
+        self.wait_cache_sync(after=commit_time)
         self.assertEqual(cache.get('foo'), 'baz')
 
     def test_memory_cache_nested_transactions(self):
@@ -135,8 +134,9 @@ class MemoryCacheTestCase(unittest.TestCase):
         transaction2 = transaction1.new_transaction()
         self.addCleanup(transaction2.stop)
         cache.clear()
+        commit_time = dt.datetime.now()
         transaction2.commit()
-        self.wait_cache_sync()
+        self.wait_cache_sync(after=commit_time)
 
         # Set value from old transaction
         Transaction().set_current_transaction(transaction1)
@@ -168,8 +168,11 @@ class MemoryCacheChannelTestCase(MemoryCacheTestCase):
         self.addCleanup(
             setattr, cache_mod, '_clear_timeout', clear_timeout)
 
-    def wait_cache_sync(self):
-        time.sleep(1)
+    def wait_cache_sync(self, after=None):
+        if after is None:
+            after = dt.datetime.now()
+        while MemoryCache._clean_last < after:
+            time.sleep(.01)
 
     @unittest.skip("No cache sync on transaction start with channel")
     def test_memory_cache_sync(self):