PYCBC-1685: Remove timeout logic when waiting for C++ core HTTP response

Motivation
==========
Since the SDK uses a timeout (the same it passes down to the C++ core)
when waiting for streaming results from the C++ core, it is possible for
it to raise a `StopIteration` incorrectly.  Instead the SDK should
continue to wait for the C++ core to return a response.

Modification
============
* Remove logic to raise a `StopIteration` exception when waiting for a
  response for streaming operations.  Instead a message is logged (at
  the DEBUG level) to provide potentially useful information.
* Update tests to confirm changes.

Change-Id: Ia4ea33ebe9dbf3560874a5f9317b476a5ec5315a
Reviewed-on: https://review.couchbase.org/c/couchbase-python-client/+/227431
Tested-by: Build Bot <build@couchbase.com>
Reviewed-by: Dimitris Christodoulou <dimitris.christodoulou@couchbase.com>

Author: thejcfactor

acouchbase/tests/analytics_t.py

@@ -172,6 +172,7 @@ class AnalyticsTestSuite:
     TEST_MANIFEST = [
         'test_analytics_metadata',
         'test_analytics_with_metrics',
+        'test_query_large_result_set',
         'test_query_named_parameters',
         'test_query_named_parameters_no_options',
         'test_query_named_parameters_override',
@@ -229,6 +230,21 @@ async def test_analytics_with_metrics(self, cb_env):
         assert isinstance(metrics.processed_objects(), UnsignedInt64)
         assert metrics.error_count() == UnsignedInt64(0)
 
+    @pytest.mark.asyncio
+    async def test_query_large_result_set(self, cb_env):
+        # Prior to PYCBC-1685, this would raise a StopIteration b/c the timeout was
+        # reached on the Python side prior to the C++ core returning the result set.
+        # It is difficult to determine the timeout value in the Jenkins environment,
+        # so allow an AmbiguousTimeoutException.
+        count = 100000
+        statement = f'SELECT {{"x1": 1, "x2": 2, "x3": 3}} FROM range(1, {count}) r;'
+        try:
+            result = cb_env.cluster.analytics_query(statement, timeout=timedelta(seconds=2))
+            row_count = [1 async for _ in result.rows()]
+            assert len(row_count) == count
+        except AmbiguousTimeoutException:
+            pass
+
     @pytest.mark.asyncio
     async def test_query_named_parameters(self, cb_env):
         result = cb_env.cluster.analytics_query(f'SELECT * FROM `{cb_env.DATASET_NAME}` WHERE `type` = $atype LIMIT 1',
@@ -291,19 +307,17 @@ async def test_query_timeout(self, cb_env):
         username, pw = cb_env.config.get_username_and_pw()
         auth = PasswordAuthenticator(username, pw)
         # Prior to PYCBC-1521, this test would fail as each request would override the cluster level analytics_timeout.
-        # If a timeout was not provided in the request, the default 75s timeout would be used.  PYCBC-1521 corrects
-        # this behavior so this test will pass as we are essentially forcing an AmbiguousTimeoutException because
-        # we are setting the cluster level analytics_timeout such a small value.
-        timeout_opts = ClusterTimeoutOptions(analytics_timeout=timedelta(milliseconds=1))
+        # If a timeout was not provided in the request, the default 75s timeout would be used.
+        timeout_opts = ClusterTimeoutOptions(analytics_timeout=timedelta(seconds=1))
         cluster = await Cluster.connect(f'{conn_string}', ClusterOptions(auth, timeout_options=timeout_opts))
         # don't need to do this except for older server versions
         _ = cluster.bucket(f'{cb_env.bucket.name}')
-        q_str = f'SELECT * FROM `{cb_env.DATASET_NAME}` LIMIT 1;'
+        q_str = 'SELECT sleep("some value", 1500) AS some_field;'
         with pytest.raises(AmbiguousTimeoutException):
             res = cluster.analytics_query(q_str)
             [r async for r in res.rows()]
         # if we override the timeout w/in the request the query should succeed.
-        res = cluster.analytics_query(q_str, timeout=timedelta(seconds=10))
+        res = cluster.analytics_query(q_str, timeout=timedelta(seconds=2))
         rows = [r async for r in res.rows()]
         assert len(rows) > 0
 

acouchbase/tests/query_t.py

@@ -205,19 +205,6 @@ class QueryTestSuite:
         'test_simple_query_without_options_with_kwargs_positional_params',
     ]
 
-    @pytest_asyncio.fixture(name='setup_udf')
-    async def setup_teardown_udf(self, cb_env):
-        EnvironmentFeatures.check_if_feature_supported('query_user_defined_functions',
-                                                       cb_env.server_version_short,
-                                                       cb_env.mock_server_type)
-        await AsyncTestEnvironment.try_n_times(3,
-                                               1,
-                                               cb_env.load_udf)
-        yield
-        await AsyncTestEnvironment.try_n_times(3,
-                                               1,
-                                               cb_env.drop_udf)
-
     @pytest.fixture(scope='class')
     def check_preserve_expiry_supported(self, cb_env):
         EnvironmentFeatures.check_if_feature_supported('preserve_expiry',
@@ -243,11 +230,15 @@ async def test_mixed_positional_parameters(self, cb_env):
                                       QueryOptions(positional_parameters=['xgfflq']), f'{cb_env.get_batch_id()}')
         await cb_env.assert_rows(result, 1)
 
-    @pytest.mark.usefixtures('setup_udf')
     @pytest.mark.asyncio
     async def test_non_blocking(self, cb_env):
+
         async def run_query(cluster, idx):
-            result = cluster.query("EXECUTE FUNCTION loop(1000000000)")
+            slow_query = ['SELECT COUNT (1) AS c FROM',
+                          'ARRAY_RANGE(0,100) AS d1,'
+                          'ARRAY_RANGE(0,100) AS d2,'
+                          'ARRAY_RANGE(0,100) AS d3']
+            result = cluster.query(' '.join(slow_query))
             rows = []
             async for r in result:
                 rows.append(r)
@@ -363,25 +354,29 @@ async def test_query_ryow(self, cb_env):
     @pytest.mark.flaky(reruns=5, reruns_delay=1)
     @pytest.mark.asyncio
     async def test_query_timeout(self, cb_env):
+        if cb_env.server_version_short < 7.1:
+            pytest.skip("Query used in test only available on server versions >= 7.1")
         from acouchbase.cluster import Cluster
         from couchbase.auth import PasswordAuthenticator
         from couchbase.options import ClusterOptions, ClusterTimeoutOptions
         conn_string = cb_env.config.get_connection_string()
         username, pw = cb_env.config.get_username_and_pw()
         auth = PasswordAuthenticator(username, pw)
         # Prior to PYCBC-1521, this test would fail as each request would override the cluster level query_timeout.
-        # If a timeout was not provided in the request, the default 75s timeout would be used.  PYCBC-1521 corrects
-        # this behavior so this test will pass as we are essentially forcing an AmbiguousTimeoutException because
-        # we are setting the cluster level query_timeout such a small value.
-        timeout_opts = ClusterTimeoutOptions(query_timeout=timedelta(milliseconds=1))
+        # If a timeout was not provided in the request, the default 75s timeout would be used.
+        timeout_opts = ClusterTimeoutOptions(query_timeout=timedelta(seconds=1.5))
         cluster = await Cluster.connect(f'{conn_string}', ClusterOptions(auth, timeout_options=timeout_opts))
         # don't need to do this except for older server versions
         _ = cluster.bucket(f'{cb_env.bucket.name}')
-        q_str = f'SELECT * FROM `{cb_env.bucket.name}` LIMIT 10;'
+        slow_query = ' '.join(['SELECT COUNT (1) AS c FROM',
+                               'ARRAY_RANGE(0,110) AS d1,'
+                               'ARRAY_RANGE(0,110) AS d2,'
+                               'ARRAY_RANGE(0,110) AS d3'])
         with pytest.raises(AmbiguousTimeoutException):
-            await cluster.query(q_str).execute()
+            await cluster.query(slow_query).execute()
         # If we override the timeout w/in the request the query should succeed.
-        rows = await cluster.query(q_str, timeout=timedelta(seconds=10)).execute()
+        # NOTE: a timeout of < 10s is most likely acceptable, but for the Jenkins environment we give plenty of room.
+        rows = await cluster.query(slow_query, timeout=timedelta(seconds=30)).execute()
         assert len(rows) > 0
 
     @pytest.mark.asyncio

couchbase/tests/analytics_t.py

@@ -186,6 +186,7 @@ class AnalyticsTestSuite:
         'test_analytics_metadata',
         'test_analytics_query_in_thread',
         'test_analytics_with_metrics',
+        'test_query_large_result_set',
         'test_query_named_parameters',
         'test_query_named_parameters_no_options',
         'test_query_named_parameters_override',
@@ -264,6 +265,20 @@ def test_analytics_with_metrics(self, cb_env):
         assert isinstance(metrics.processed_objects(), UnsignedInt64)
         assert metrics.error_count() == UnsignedInt64(0)
 
+    def test_query_large_result_set(self, cb_env):
+        # Prior to PYCBC-1685, this would raise a StopIteration b/c the timeout was
+        # reached on the Python side prior to the C++ core returning the result set.
+        # It is difficult to determine the timeout value in the Jenkins environment,
+        # so allow an AmbiguousTimeoutException.
+        count = 100000
+        statement = f'SELECT {{"x1": 1, "x2": 2, "x3": 3}} FROM range(1, {count}) r;'
+        try:
+            result = cb_env.cluster.analytics_query(statement, timeout=timedelta(seconds=2))
+            row_count = [1 for _ in result.rows()]
+            assert len(row_count) == count
+        except AmbiguousTimeoutException:
+            pass
+
     def test_query_named_parameters(self, cb_env):
         result = cb_env.cluster.analytics_query(f'SELECT * FROM `{cb_env.DATASET_NAME}` WHERE `type` = $atype LIMIT 1',
                                                 AnalyticsOptions(named_parameters={'atype': 'vehicle'}))
@@ -318,20 +333,18 @@ def test_query_timeout(self, cb_env):
         username, pw = cb_env.config.get_username_and_pw()
         auth = PasswordAuthenticator(username, pw)
         # Prior to PYCBC-1521, this test would fail as each request would override the cluster level analytics_timeout.
-        # If a timeout was not provided in the request, the default 75s timeout would be used.  PYCBC-1521 corrects
-        # this behavior so this test will pass as we are essentially forcing an AmbiguousTimeoutException because
-        # we are setting the cluster level analytics_timeout such a small value.
-        timeout_opts = ClusterTimeoutOptions(analytics_timeout=timedelta(milliseconds=1))
+        # If a timeout was not provided in the request, the default 75s timeout would be used.
+        timeout_opts = ClusterTimeoutOptions(analytics_timeout=timedelta(seconds=1))
         cluster = Cluster.connect(f'{conn_string}', ClusterOptions(auth, timeout_options=timeout_opts))
         # don't need to do this except for older server versions
         _ = cluster.bucket(f'{cb_env.bucket.name}')
-        q_str = f'SELECT * FROM `{cb_env.DATASET_NAME}` LIMIT 1;'
+        q_str = 'SELECT sleep("some value", 1500) AS some_field;'
         with pytest.raises(AmbiguousTimeoutException):
             res = cluster.analytics_query(q_str)
             [r for r in res.rows()]
 
         # if we override the timeout w/in the request the query should succeed.
-        res = cluster.analytics_query(q_str, timeout=timedelta(seconds=10))
+        res = cluster.analytics_query(q_str, timeout=timedelta(seconds=2))
         rows = [r for r in res.rows()]
         assert len(rows) > 0
 

couchbase/tests/query_t.py

@@ -336,26 +336,30 @@ def test_query_raw_options(self, cb_env):
     # creating a new connection, allow retries
     @pytest.mark.flaky(reruns=5, reruns_delay=1)
     def test_query_timeout(self, cb_env):
+        if cb_env.server_version_short < 7.1:
+            pytest.skip("Query used in test only available on server versions >= 7.1")
         from couchbase.auth import PasswordAuthenticator
         from couchbase.cluster import Cluster
         from couchbase.options import ClusterOptions, ClusterTimeoutOptions
         conn_string = cb_env.config.get_connection_string()
         username, pw = cb_env.config.get_username_and_pw()
         auth = PasswordAuthenticator(username, pw)
         # Prior to PYCBC-1521, this test would fail as each request would override the cluster level query_timeout.
-        # If a timeout was not provided in the request, the default 75s timeout would be used.  PYCBC-1521 corrects
-        # this behavior so this test will pass as we are essentially forcing an AmbiguousTimeoutException because
-        # we are setting the cluster level query_timeout such a small value.
-        timeout_opts = ClusterTimeoutOptions(query_timeout=timedelta(milliseconds=1))
+        # If a timeout was not provided in the request, the default 75s timeout would be used.
+        timeout_opts = ClusterTimeoutOptions(query_timeout=timedelta(seconds=1.5))
         cluster = Cluster.connect(f'{conn_string}', ClusterOptions(auth, timeout_options=timeout_opts))
         # don't need to do this except for older server versions
         _ = cluster.bucket(f'{cb_env.bucket.name}')
-        q_str = f'SELECT * FROM `{cb_env.bucket.name}` LIMIT 10;'
+        slow_query = ' '.join(['SELECT COUNT (1) AS c FROM',
+                               'ARRAY_RANGE(0,110) AS d1,'
+                               'ARRAY_RANGE(0,110) AS d2,'
+                               'ARRAY_RANGE(0,110) AS d3'])
         with pytest.raises(AmbiguousTimeoutException):
-            cluster.query(q_str).execute()
+            cluster.query(slow_query).execute()
 
         # If we override the timeout w/in the request the query should succeed.
-        rows = cluster.query(q_str, timeout=timedelta(seconds=10)).execute()
+        # NOTE: a timeout of < 10s is most likely acceptable, but for the Jenkins environment we give plenty of room.
+        rows = cluster.query(slow_query, timeout=timedelta(seconds=30)).execute()
         assert len(rows) > 0
 
     def test_query_ryow(self, cb_env):

src/result.cxx

@@ -230,13 +230,7 @@ streamed_result_iternext(PyObject* self)
     Py_BEGIN_ALLOW_THREADS row = s_res->rows->get(s_res->timeout_ms);
     Py_END_ALLOW_THREADS
   }
-
-  if (row != nullptr) {
-    return row;
-  } else {
-    PyErr_SetString(PyExc_StopIteration, "Timeout occurred waiting for next item in queue.");
-    return nullptr;
-  }
+  return row;
 }
 
 static PyObject*

src/result.hxx

@@ -49,10 +49,21 @@ public:
     std::unique_lock<std::mutex> lock(mut_);
 
     while (rows_.empty()) {
-      auto now = std::chrono::system_clock::now();
-      if (cv_.wait_until(lock, now + timeout_ms) == std::cv_status::timeout) {
-        // this will cause iternext to return nullptr, which stops iteration
-        return nullptr;
+      if (cv_.wait_for(lock, timeout_ms) == std::cv_status::timeout) {
+        // This timeout (e.g. timeout_ms) is the same timeout we pass to the C++ core.
+        // If we timeout on the Python side this means:
+        //   - Edge case where the C++ core is about to timeout. We want to use the C++ core error
+        //   details,
+        //     so wait a little longer to get the C++ core timeout.
+        //   - The result set is large and since we don't have streaming support yet, we have to
+        //   wait for
+        //     the entire result set to be returned.  Again we should wait until we get the results.
+        // PYCBC-1685: Instead of trying to do some tricky error handling we instead wait for the
+        // C++ core results and log a message that can provide insight to users about the SDK
+        // behavior.
+        CB_LOG_DEBUG(
+          "PYCBC: No results received from C++ core after {}ms.  Continue to wait for results.",
+          timeout_ms.count());
       }
     }