Fix COPY OUT race conditions and state corruption (Python 3.9.9 specific)

This commit addresses critical concurrency issues in the `COPY OUT` protocol implementation. These issues were specifically observed in Python 3.9.9 environments (likely due to differences in asyncio event loop scheduling compared to newer versions like 3.13), leading to deadlocks and data corruption.

Changes:

1. Fix Data Loss (Backpressure Handling):
   - Issue: In Python 3.9, `_dispatch_result` could be triggered multiple times while the sink was paused, overwriting `_pending_result`.
   - Fix: Implemented data merging logic. New incoming data is now appended to the existing pending buffer instead of replacing it.

2. Fix State Pollution on Cancel:
   - Issue: Cancelling a `COPY` task left residual data in `_pending_result`. Subsequent queries (e.g., `SELECT 1`) would incorrectly consume this stale data, causing `AttributeError: 'tuple' object has no attribute '_init_types'`.
   - Fix: Added a `try...finally` block in `copy_out` to unconditionally clear `_pending_result` and reset protocol state upon exit.

3. Fix Deadlocks:
   - Issue: The connection could get stuck in a paused state if execution was interrupted or if `resume_reading` was missed.
   - Fix: Ensured `self.resume_reading()` is called immediately after consuming buffered data in `_new_waiter` and in the cleanup phase.

4. Fix Logic Errors:
   - corrected `_new_waiter` to prevent assigning completed Futures to `self.waiter`, avoiding `InternalClientError`.

Fixes #8

Author: zhanghaitao3

async_gaussdb/protocol/protocol.pxd

@@ -51,6 +51,8 @@ cdef class BaseProtocol(CoreProtocol):
 
         bint _is_ssl
 
+        object _pending_result
+
         PreparedStatementState statement
 
     cdef get_connection(self)

async_gaussdb/protocol/protocol.pyx

@@ -102,6 +102,8 @@ cdef class BaseProtocol(CoreProtocol):
 
         self._is_ssl = False
 
+        self._pending_result = None
+
         try:
             self.create_future = loop.create_future
         except AttributeError:
@@ -414,7 +416,10 @@ cdef class BaseProtocol(CoreProtocol):
                         self._request_cancel()
                         # Make asyncio shut up about unretrieved
                         # QueryCanceledError
-                        waiter.add_done_callback(lambda f: f.exception())
+                        if waiter and not waiter.done():
+                            waiter.cancel()
+                        elif waiter and waiter.done() and not waiter.cancelled():
+                            waiter.exception()
                         raise
 
                 # done will be True upon receipt of CopyDone.
@@ -424,6 +429,7 @@ cdef class BaseProtocol(CoreProtocol):
                 waiter = self._new_waiter(timer.get_remaining_budget())
 
         finally:
+            self._pending_result = None
             self.resume_reading()
 
         return status_msg
@@ -776,6 +782,14 @@ cdef class BaseProtocol(CoreProtocol):
             self.abort()
 
     cdef _new_waiter(self, timeout):
+        if self._pending_result is not None:
+            res = self._pending_result
+            self._pending_result = None
+            self.resume_reading()
+            waiter = self.loop.create_future()
+            waiter.set_result(res)
+            return waiter
+
         if self.waiter is not None:
             raise apg_exc.InterfaceError(
                 'cannot perform operation: another operation is in progress')
@@ -848,10 +862,31 @@ cdef class BaseProtocol(CoreProtocol):
         waiter = self.waiter
         self.waiter = None
 
-        if PG_DEBUG:
-            if waiter is None:
+        if waiter is None:
+            if PG_DEBUG:
                 raise apg_exc.InternalClientError('_on_result: waiter is None')
+            
+            if self.state == PROTOCOL_COPY_OUT_DATA or \
+               self.state == PROTOCOL_COPY_OUT_DONE:
+                
+                copy_done = self.state == PROTOCOL_COPY_OUT_DONE
+                if copy_done:
+                    status_msg = self.result_status_msg.decode(self.encoding)
+                else:
+                    status_msg = None
+
+                self.pause_reading()
+                if self._pending_result is not None:
+                    old_data, old_done, old_status = self._pending_result
+                    current_data = self.result if self.result is not None else b''
+                    merged_data = (old_data if old_data is not None else b'') + current_data
+                    self._pending_result = (merged_data, copy_done, status_msg)
+                else:
+                    self._pending_result = (self.result, copy_done, status_msg)
 
+                return
+            else:
+                return
         if waiter.cancelled():
             return