From 36d5bce9bc431ce3da7c6e27cdfa15f1e276f20e 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
---
 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 57ab0be862..fb6814c3da 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
@@ -220,6 +221,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
@@ -355,9 +359,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 e3c158834c..a19de60131 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
 
@@ -81,7 +82,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()
@@ -119,11 +120,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):
@@ -158,8 +157,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)
@@ -191,8 +191,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):