Fix DCP test failures and log noise for magma buckets

self.vbuckets was hard-coded to range(1024) but magma buckets default
to 128 vbuckets; accessing vb_map[128+] raised IndexError in
select_dcp_client for every DCP collection test. Fix by deriving the
range from the actual vb_map length after fetching the cluster config.

Fix TypeError in handleSystemEvent where response['key'] arrives as
bytes in Python 3 and was stored raw into the manifest, causing
json.dumps(vb['manifest']) to fail. Decode with surrogateescape so
non-UTF-8 binary keys are round-tripped safely rather than raising
UnicodeDecodeError, consistent with the existing guard in handleMutation.

Fix recv_op to suppress hundreds of spurious "recv_op Exception: timed
out" lines that appeared during disk-checkpoint streams. Socket poll
timeouts are expected when the server is flushing to disk and are
already handled by the retry logic in process_dcp_traffic; printing
them on every poll obscures real errors. Also fix a dead-code branch
where 'Timeout' (capital T) never matched the actual socket exception
message 'timed out' (lowercase), and remove the _stream_timeout
assignment from this path since a transient poll timeout should not
permanently short-circuit the connection.

Co-authored-by: Claude <noreply@anthropic.com>
Change-Id: Iecb5d78aaeafd435d7da9882e9c2e2d3f7ee6e7f
Reviewed-on: https://review.couchbase.org/c/TAF/+/245571
Reviewed-by: Ashwin <ashwin.govindarajulu@couchbase.com>
Tested-by: Balakumaran G <balakumaran.gopal@couchbase.com>
Tested-by: Couchbase QE
Reviewed-by: Balakumaran G <balakumaran.gopal@couchbase.com>

Author: bkumaran

lib/dcp_bin_client.py

@@ -152,12 +152,10 @@ def stream_req(self, vbucket, takeover, start_seqno, end_seqno,
         response = self._handle_op(op)
 
         def __generator(response):
-
             yield response
             last_by_seqno = 0
 
             while True:
-
                 if not op.queue.empty():
                     response = op.queue.get()
                 else:
@@ -247,14 +245,16 @@ def recv_op(self, op):
                     self.ack_dcp_noop_req(opaque)
 
             except Exception as ex:
-                print("recv_op Exception:", ex)
-                if 'died' in str(ex):
+                ex_str = str(ex)
+                if 'died' in ex_str:
+                    print("recv_op Exception:", ex)
                     return {'opcode': op.opcode,
                             'status': 0xff}
-                elif 'Timeout' in str(ex):
-                    self._stream_timeout = True
+                elif 'timed out' in ex_str or 'Timeout' in ex_str:
+                    # Socket poll timeout — expected on slow streams, not an error
                     return None
                 else:
+                    print("recv_op Exception:", ex)
                     return None
 
     def ack_stream_req(self, opaque):

pytests/dcp_new/dcp_base.py

@@ -59,6 +59,7 @@ def initialise_cluster_connections(self):
 
         config_json = json.loads(DcpClient.get_config(init_dcp_client)[2])
         self.vb_map = config_json['vBucketServerMap']['vBucketMap']
+        self.vbuckets = range(len(self.vb_map))
 
         self.dcp_client_dict = dict()
 
@@ -183,6 +184,8 @@ def handle_stream_create_response(self, dcpStream):
 
     def handleSystemEvent(self, response, manifest):
         # Unpack a DCP system event
+        if isinstance(response.get('key'), bytes):
+            response['key'] = response['key'].decode('utf-8', errors='surrogateescape')
         if response['event'] == EVENT_CREATE_COLLECTION:
             if response['version'] == 0:
                 uid, sid, cid = struct.unpack(">QII", response['value'])