refactor(shim): harden async message handling and clean stub/utils

- Replace global handler state with per-instance state to avoid cross-stream bleed
- Keep concurrent message handling but track task failures explicitly in logs
- Fix missing await paths on error message writes
- Remove wildcard import in stub and initialize safe defaults for creator/timestamp
- Add explicit error for empty function payload in stub
- Remove debug print side effect and unused import in utils

Validation:
- py_compile on handler/stub/utils
- pytest tests/test_handler.py
- smoke invoke/query for examples/ccaas/asset-transfer-basic
- smoke create/transfer/query for examples/ccaas/asset-transfer-sbe

Signed-off-by: kmilo <kmilo.denis.glez@yandex.com>

Author: kmilodenisglez

src/fabric_shim/handler.py

@@ -37,16 +37,31 @@ class STATES:
 COMPOSITEKEY_NS = '\x00'
 EMPTY_KEY_SUBSTITUTE = '\x01'
 
-STATE = STATES.CREATED
-
-
 class Handler:
     def __init__(self, cc_id: str, cc: Chaincode) -> None:
         self.chaincode_id = cc_pb2.ChaincodeID()
         self.chaincode_id.name = cc_id
         self.chaincode = cc
         self.msg_queue_handler = None
         self.context = None
+        self.state = STATES.CREATED
+        self._pending_tasks = set()
+
+    def _track_task(self, coro):
+        """Schedule message handling while surfacing task failures in logs."""
+        task = asyncio.create_task(coro)
+        self._pending_tasks.add(task)
+
+        def _done_callback(done_task):
+            self._pending_tasks.discard(done_task)
+            try:
+                exc = done_task.exception()
+            except asyncio.CancelledError:
+                return
+            if exc is not None:
+                LOGGER.exception('Unhandled exception while processing peer message', exc_info=exc)
+
+        task.add_done_callback(_done_callback)
 
     async def handle_stub_interaction(self, msg, action="Invoke"):
         """handle_message calls the Init | Invoke function of the associated chaincode."""
@@ -118,65 +133,51 @@ async def handle_message_ready(self, msg):
             await self.handle_stub_interaction(msg, "Invoke")
             return
         else:
-            self.context.write(new_error_msg(msg, STATE))
+            await self.context.write(new_error_msg(msg, self.state))
 
-    def handle_message_established(self, msg):
+    async def handle_message_established(self, msg):
         """
         handle_message_established handles messages received from the peer when the handler is in the "established" state.
         """
-        global STATE
         if msg.type != ccshim_pb2.ChaincodeMessage.READY:
             LOGGER.error(f'Chaincode is in "established" state, can only process messages of type "ready", '
                          f'but received "{msg.type}"')
-            # write is an async coroutine on the context
-            try:
-                return asyncio.create_task(self.context.write(new_error_msg(msg, STATE)))
-            except Exception:
-                return
+            await self.context.write(new_error_msg(msg, self.state))
         else:
             LOGGER.info('Successfully established communication with peer node. State transferred to "ready"')
-            STATE = STATES.READY
+            self.state = STATES.READY
 
-    def handle_message_created(self, msg):
+    async def handle_message_created(self, msg):
         """handle_message_created handles messages received from the peer when the handler is in the "created" state."""
-        global STATE
         if msg.type != ccshim_pb2.ChaincodeMessage.REGISTERED:
             LOGGER.error(f'Chaincode is in "created" state, can only process messages of type "registered", '
                          f'but received "{msg.type}"')
-            try:
-                return asyncio.create_task(self.context.write(new_error_msg(msg, STATE)))
-            except Exception:
-                return
+            await self.context.write(new_error_msg(msg, self.state))
         else:
             LOGGER.info('Successfully registered with peer node. State transferred to "established"')
-            STATE = STATES.ESTABLISHED
+            self.state = STATES.ESTABLISHED
 
     async def handle_message(self, msg: ccshim_pb2.ChaincodeMessage):
         """handle_message message handles loop for shim side of chaincode/peer stream."""
         LOGGER.warning('-->> Look out!')
-        global STATE
 
         # TODO: ?
         if msg.type == ccshim_pb2.ChaincodeMessage.KEEPALIVE:
             LOGGER.info('-| KEEPALIVE')
             return
 
-        if STATE == STATES.READY:
+        if self.state == STATES.READY:
             await self.handle_message_ready(msg)
-        elif STATE == STATES.ESTABLISHED:
-            self.handle_message_established(msg)
-        elif STATE == STATES.CREATED:
-            self.handle_message_created(msg)
+        elif self.state == STATES.ESTABLISHED:
+            await self.handle_message_established(msg)
+        elif self.state == STATES.CREATED:
+            await self.handle_message_created(msg)
         else:
-            try:
-                asyncio.create_task(self.context.write(new_error_msg(msg, STATE)))
-            except Exception:
-                LOGGER.exception('Failed to write error message to context')
+            await self.context.write(new_error_msg(msg, self.state))
 
     async def chat_with_peer(self, stream: AsyncIterable[ccshim_pb2.ChaincodeMessage], context: grpc.aio.ServicerContext):
         """chat stream for peer-chaincode interactions post connection"""
-        global STATE
-        STATE = STATES.CREATED
+        self.state = STATES.CREATED
 
         self.context = context
         self.msg_queue_handler = MsgQueueHandler(self)
@@ -197,13 +198,18 @@ async def chat_with_peer(self, stream: AsyncIterable[ccshim_pb2.ChaincodeMessage
                 return ccshim_pb2.ChaincodeMessage(
                     type=ccshim_pb2.ChaincodeMessage.ERROR, payload=err_str.encode(encoding='utf-8'))
             else:
-                asyncio.create_task(self.handle_message(receive_message))
+                # Keep handling asynchronous so the stream can continue receiving
+                # response frames needed by in-flight request futures.
+                self._track_task(self.handle_message(receive_message))
 
                 LOGGER.info(f'->>>>  proposal  {receive_message.proposal}')
                 LOGGER.info(f'->>>>  payload  {receive_message.payload}')
                 LOGGER.info(f'->>>>  channel ID  {receive_message.channel_id}')
                 LOGGER.info(f'->>>>  Tx ID  {receive_message.txid}')
 
+        if self._pending_tasks:
+            await asyncio.gather(*self._pending_tasks, return_exceptions=True)
+
     async def handle_get_state(self, collection, key, channel_id, tx_id):
         msg_pb = ccshim_pb2.GetState()
         msg_pb.key = key

src/fabric_shim/stub.py

@@ -2,7 +2,11 @@
 # contributors. All rights reserved.
 # SPDX-License-Identifier: Apache-2.0
 from src.fabric_shim.interfaces import ChaincodeStubInterface
-from src.fabric_shim.utils import *
+from src.fabric_shim.utils import (
+    COMPOSITEKEY_NS,
+    MIN_UNICODE_RUNE_VALUE,
+    validate_composite_key_attribute,
+)
 from fabric_protos_python.peer import chaincode_pb2 as pb
 from fabric_protos_python.common import common_pb2 as cm_pb
 from fabric_protos_python.peer import proposal_pb2 as pr_pb
@@ -23,6 +27,9 @@ def __init__(self, client, channel_id, tx_id, cc_input, signed_proposal_pb):
         self.tx_id = tx_id
         self.cc_input = cc_input
         self.signed_proposal_pb = signed_proposal_pb
+        self.creator = {}
+        self.tx_timestamp = None
+        self.proposal = None
         self.validationParameterMetakey = VALIDATION_PARAMETER
 
         if self.signed_proposal_pb:
@@ -104,6 +111,8 @@ def get_txid(self):
     def get_function_and_parameters(self):
         """Get function name and parameters of the chaincode calling transaction"""
         args = [arg.decode() for arg in self.cc_input.args]
+        if len(args) == 0:
+            raise Exception('no function name provided in transaction payload')
         function: str = args[0]
         params = args[1:]
 

src/fabric_shim/utils.py

@@ -1,5 +1,4 @@
 # Auxiliary tools
-from fabric_protos_python.peer import chaincode_shim_pb2 as ccshim_pb2
 
 
 def enum_type(*sequential, **named) -> type:
@@ -18,7 +17,6 @@ def enum_type(*sequential, **named) -> type:
         named: Collects all the keyword arguments in a dictionary.
 
     """
-    print(sequential, ": ", named)
     enums = dict(zip(sequential, range(len(sequential))), **named)
     return type('Enum', (), enums)