Trying from fix lock issue

sujata-m · sujata-m · commit 7c4a4ad16d5e · 2026-03-17T23:12:13.000+05:30
diff --git a/src/python_ms_core/core/topic/azure_topic.py b/src/python_ms_core/core/topic/azure_topic.py
@@ -1,6 +1,8 @@
 import json
 
+import gc
 import logging
+import os
 import time
 from ..config.config import TopicConfig
 from ..resource_errors import ExceptionHandler
@@ -12,6 +14,11 @@
 import concurrent.futures as cf
 import threading
 
+try:
+    import psutil
+except ImportError:  # pragma: no cover - dependency exists in the package, but keep logging resilient.
+    psutil = None
+
 
 logging.basicConfig(format='%(asctime)s %(levelname)-8s %(message)s', datefmt='%Y-%m-%d %H:%M:%S')
 logger = logging.getLogger('AzureTopic')
@@ -50,11 +57,37 @@ def __init__(self, config: TopicConfig=None, topic_name=None, max_concurrent_mes
         self.publisher = self.client.get_topic_sender(topic_name=topic_name)
         self.executor = ThreadPoolExecutor(max_workers=max_concurrent_messages)
         self.internal_count = 0
-        self.lock_renewal = AutoLockRenewer(max_workers=max_concurrent_messages)
-        self.max_renewal_duration = 86400 # Renew the message upto 1 day
+        self.max_renewal_duration = self._get_positive_int_from_env(
+            'TOPIC_MAX_LOCK_RENEWAL_DURATION_SECONDS',
+            86400,
+        )  # Renew the message upto 1 day
+        self.lock_renewal_margin = self._get_positive_int_from_env(
+            'TOPIC_LOCK_RENEWAL_MARGIN_SECONDS',
+            60,
+        )
+        renewer_max_workers = max(max_concurrent_messages, 2)
+        self.lock_renewal = AutoLockRenewer(
+            max_lock_renewal_duration=self.max_renewal_duration,
+            on_lock_renew_failure=self._handle_lock_renew_failure,
+            max_workers=renewer_max_workers,
+        )
+        # The SDK default renews only in the last 10 seconds of the lock window.
+        # Start earlier so long-running jobs have more headroom for scheduler jitter.
+        self.lock_renewal._renew_period = min(self.lock_renewal_margin, self.max_renewal_duration)
+        self.lock_renew_receiver = _LockRenewLoggingReceiver(self)
         self.wait_time_for_message = 5
         self.thread_lock = threading.Lock()
         self.pending_tasks = []
+        self._process = None
+        self._prime_runtime_samplers()
+        logger.info(
+            'Configured lock renewal for topic %s: max_lock_renewal_duration=%s seconds, '
+            'renew_margin=%s seconds, renewer_max_workers=%s',
+            self.topic_name,
+            self.max_renewal_duration,
+            self.lock_renewal_margin,
+            renewer_max_workers,
+        )
     
     
     def publish(self, data: QueueMessage):
@@ -97,7 +130,7 @@ def subscribe(self, subscription: str, callback, max_receivable_messages=-1):
                             self.internal_count += len(messages)
                         for message in messages: 
                             self.lock_renewal.register(
-                                self.receiver,
+                                self.lock_renew_receiver,
                                 message,
                                 max_lock_renewal_duration=self.max_renewal_duration,
                                 on_lock_renew_failure=self._handle_lock_renew_failure,
@@ -170,6 +203,13 @@ def _settle_task(self, x: cf.Future, incoming_message=None):
                 incoming_message = settled_message
             if incoming_message is None:
                 return
+            if getattr(incoming_message, '_lock_expired', False):
+                logger.error(
+                    f'Skipping settlement for message {self._get_message_id(incoming_message)} '
+                    f'because the lock expired at {getattr(incoming_message, "locked_until_utc", None)}. '
+                    f'auto_renew_error={getattr(incoming_message, "auto_renew_error", None)}'
+                )
+                return
             if is_success:
                 self.receiver.complete_message(incoming_message)
             else:
@@ -183,7 +223,101 @@ def _settle_task(self, x: cf.Future, incoming_message=None):
         return
 
     def _handle_lock_renew_failure(self, renewable, error):
-        message_id = getattr(renewable, 'message_id', None) or getattr(renewable, 'messageId', 'unknown')
-        logger.error(f'Error renewing lock for message {message_id}: {error}')
+        message_id = self._get_message_id(renewable)
+        failure_reason = error or getattr(renewable, 'auto_renew_error', None) or 'lock expired before renewal could complete'
+        logger.error(
+            f'Error renewing lock for message {message_id}: {failure_reason}; '
+            f'locked_until_utc={getattr(renewable, "locked_until_utc", None)}; '
+            f'runtime_snapshot={self._get_runtime_snapshot()}'
+        )
 
-    
+    @staticmethod
+    def _get_message_id(message):
+        return getattr(message, 'message_id', None) or getattr(message, 'messageId', 'unknown')
+
+    def _prime_runtime_samplers(self):
+        if psutil is None:
+            return
+        try:
+            self._process = psutil.Process(os.getpid())
+            self._process.cpu_percent(interval=None)
+            psutil.cpu_percent(interval=None)
+        except Exception:  # pragma: no cover - best effort diagnostics
+            self._process = None
+
+    def _get_runtime_snapshot(self):
+        return f'{self._get_memory_snapshot()}, {self._get_cpu_snapshot()}, {self._get_gc_snapshot()}'
+
+    def _get_memory_snapshot(self):
+        if self._process is None:
+            return 'memory=psutil-unavailable'
+        try:
+            memory_info = self._process.memory_info()
+            rss_mb = memory_info.rss / (1024 * 1024)
+            vms_mb = memory_info.vms / (1024 * 1024)
+            return f'memory=rss_mb={rss_mb:.2f}, vms_mb={vms_mb:.2f}, num_threads={self._process.num_threads()}'
+        except Exception as exc:  # pragma: no cover - diagnostic fallback
+            return f'memory=unavailable({exc})'
+
+    def _get_cpu_snapshot(self):
+        if self._process is None:
+            return 'cpu=psutil-unavailable'
+        try:
+            process_cpu_percent = self._process.cpu_percent(interval=None)
+            system_cpu_percent = psutil.cpu_percent(interval=None)
+            return (
+                f'cpu=process_percent={process_cpu_percent:.2f}, '
+                f'system_percent={system_cpu_percent:.2f}'
+            )
+        except Exception as exc:  # pragma: no cover - diagnostic fallback
+            return f'cpu=unavailable({exc})'
+
+    @staticmethod
+    def _get_gc_snapshot():
+        try:
+            gc_counts = gc.get_count()
+            gc_thresholds = gc.get_threshold()
+            gc_stats = gc.get_stats()
+            generation_summaries = []
+            for generation, stats in enumerate(gc_stats):
+                generation_summaries.append(
+                    'gen{generation}[collections={collections}, collected={collected}, uncollectable={uncollectable}]'.format(
+                        generation=generation,
+                        collections=stats.get('collections', 0),
+                        collected=stats.get('collected', 0),
+                        uncollectable=stats.get('uncollectable', 0),
+                    )
+                )
+            return (
+                f'gc=enabled={gc.isenabled()}, counts={gc_counts}, thresholds={gc_thresholds}, '
+                f'stats={"; ".join(generation_summaries)}'
+            )
+        except Exception as exc:  # pragma: no cover - diagnostic fallback
+            return f'gc=unavailable({exc})'
+
+    @staticmethod
+    def _get_positive_int_from_env(name, default):
+        value = os.environ.get(name)
+        if value is None:
+            return default
+        try:
+            parsed = int(value)
+            if parsed > 0:
+                return parsed
+        except (TypeError, ValueError):
+            pass
+        logger.warning(f'Invalid value for {name}: {value}. Using default {default}.')
+        return default
+
+class _LockRenewLoggingReceiver:
+    def __init__(self, topic):
+        self._topic = topic
+
+    def renew_message_lock(self, renewable):
+        logger.info(
+            'Attempting lock renewal for message %s; locked_until_utc=%s; runtime_snapshot=%s',
+            self._topic._get_message_id(renewable),
+            getattr(renewable, 'locked_until_utc', None),
+            self._topic._get_runtime_snapshot(),
+        )
+        return self._topic.receiver.renew_message_lock(renewable)
diff --git a/src/python_ms_core/version.py b/src/python_ms_core/version.py
@@ -1 +1 @@
-__version__ = '0.0.25'
+__version__ = '0.2.5.1'
diff --git a/tests/unit_tests/test_topic/test_azure_topic.py b/tests/unit_tests/test_topic/test_azure_topic.py
@@ -8,6 +8,25 @@
 
 class TestAzureTopic(unittest.TestCase):
 
+    @patch('src.python_ms_core.core.topic.azure_topic.AutoLockRenewer')
+    @patch('src.python_ms_core.core.topic.azure_topic.ServiceBusClient')
+    def test_init_sets_long_running_lock_renewal_defaults(self, mock_service_bus_client, mock_auto_lock_renewer):
+        mock_client = MagicMock()
+        mock_config = MagicMock(connection_string='Endpoint=sb://test/')
+        mock_renewer = MagicMock()
+
+        mock_service_bus_client.from_connection_string.return_value = mock_client
+        mock_client.get_topic_sender.return_value = MagicMock()
+        mock_auto_lock_renewer.return_value = mock_renewer
+
+        AzureTopic(config=mock_config, topic_name='mock-topic', max_concurrent_messages=1)
+
+        mock_auto_lock_renewer.assert_called_once()
+        _, kwargs = mock_auto_lock_renewer.call_args
+        self.assertEqual(kwargs['max_lock_renewal_duration'], 86400)
+        self.assertEqual(kwargs['max_workers'], 2)
+        self.assertEqual(mock_renewer._renew_period, 60)
+
     @patch('src.python_ms_core.core.topic.azure_topic.AutoLockRenewer')
     @patch('src.python_ms_core.core.topic.azure_topic.ServiceBusClient')
     def test_subscribe_settles_completed_tasks_on_receiver_loop(self, mock_service_bus_client, mock_auto_lock_renewer):
@@ -16,6 +35,7 @@ def test_subscribe_settles_completed_tasks_on_receiver_loop(self, mock_service_b
         mock_message = MagicMock()
         mock_future = MagicMock()
         mock_config = MagicMock(connection_string='Endpoint=sb://test/')
+        mock_message._lock_expired = False
 
         mock_service_bus_client.from_connection_string.return_value = mock_client
         mock_client.get_topic_sender.return_value = MagicMock()
@@ -48,6 +68,7 @@ def test_settle_task_logs_error_and_releases_slot(self, mock_service_bus_client,
         mock_message = MagicMock()
         mock_future = MagicMock()
         mock_config = MagicMock(connection_string='Endpoint=sb://test/')
+        mock_message._lock_expired = False
 
         mock_service_bus_client.from_connection_string.return_value = mock_client
         mock_client.get_topic_sender.return_value = MagicMock()
@@ -65,6 +86,122 @@ def test_settle_task_logs_error_and_releases_slot(self, mock_service_bus_client,
         mock_logger.error.assert_called_once_with('Error in settling message: Mocked settlement failure')
         self.assertEqual(topic.internal_count, 0)
 
+    @patch('src.python_ms_core.core.topic.azure_topic.logger')
+    @patch('src.python_ms_core.core.topic.azure_topic.AutoLockRenewer')
+    @patch('src.python_ms_core.core.topic.azure_topic.ServiceBusClient')
+    def test_settle_task_skips_expired_message(self, mock_service_bus_client, mock_auto_lock_renewer, mock_logger):
+        mock_client = MagicMock()
+        mock_receiver = MagicMock()
+        mock_message = MagicMock()
+        mock_future = MagicMock()
+        mock_config = MagicMock(connection_string='Endpoint=sb://test/')
+
+        mock_message._lock_expired = True
+        mock_message.message_id = 'message-1'
+        mock_message.locked_until_utc = '2026-03-17T09:39:28Z'
+        mock_message.auto_renew_error = None
+        mock_service_bus_client.from_connection_string.return_value = mock_client
+        mock_client.get_topic_sender.return_value = MagicMock()
+        mock_client.get_subscription_receiver.return_value = mock_receiver
+        mock_future.result.return_value = [True, mock_message]
+
+        topic = AzureTopic(config=mock_config, topic_name='mock-topic', max_concurrent_messages=1)
+        topic.receiver = mock_receiver
+        topic.internal_count = 1
+
+        topic._settle_task(mock_future, incoming_message=mock_message)
+
+        mock_receiver.complete_message.assert_not_called()
+        mock_receiver.abandon_message.assert_not_called()
+        mock_logger.error.assert_called_once_with(
+            'Skipping settlement for message message-1 because the lock expired at '
+            '2026-03-17T09:39:28Z. auto_renew_error=None'
+        )
+        self.assertEqual(topic.internal_count, 0)
+
+    @patch('src.python_ms_core.core.topic.azure_topic.logger')
+    @patch('src.python_ms_core.core.topic.azure_topic.AutoLockRenewer')
+    @patch('src.python_ms_core.core.topic.azure_topic.ServiceBusClient')
+    def test_handle_lock_renew_failure_logs_when_sdk_returns_no_error(
+        self,
+        mock_service_bus_client,
+        mock_auto_lock_renewer,
+        mock_logger,
+    ):
+        mock_client = MagicMock()
+        mock_config = MagicMock(connection_string='Endpoint=sb://test/')
+        mock_message = MagicMock()
+
+        mock_service_bus_client.from_connection_string.return_value = mock_client
+        mock_client.get_topic_sender.return_value = MagicMock()
+        mock_message.message_id = 'message-1'
+        mock_message.locked_until_utc = '2026-03-17T09:39:28Z'
+        mock_message.auto_renew_error = None
+
+        topic = AzureTopic(config=mock_config, topic_name='mock-topic', max_concurrent_messages=1)
+        topic._get_runtime_snapshot = MagicMock(
+            return_value=(
+                'memory=rss_mb=128.00, vms_mb=256.00, num_threads=4, '
+                'cpu=process_percent=80.00, system_percent=91.00, '
+                'gc=enabled=True, counts=(1, 2, 3), thresholds=(700, 10, 10), '
+                'stats=gen0[collections=1, collected=2, uncollectable=0]'
+            )
+        )
+
+        topic._handle_lock_renew_failure(mock_message, None)
+
+        mock_logger.error.assert_called_once_with(
+            'Error renewing lock for message message-1: lock expired before renewal could complete; '
+            'locked_until_utc=2026-03-17T09:39:28Z; '
+            'runtime_snapshot=memory=rss_mb=128.00, vms_mb=256.00, num_threads=4, '
+            'cpu=process_percent=80.00, system_percent=91.00, '
+            'gc=enabled=True, counts=(1, 2, 3), thresholds=(700, 10, 10), '
+            'stats=gen0[collections=1, collected=2, uncollectable=0]'
+        )
+
+    @patch('src.python_ms_core.core.topic.azure_topic.logger')
+    @patch('src.python_ms_core.core.topic.azure_topic.AutoLockRenewer')
+    @patch('src.python_ms_core.core.topic.azure_topic.ServiceBusClient')
+    def test_lock_renew_attempt_logs_memory_snapshot(
+        self,
+        mock_service_bus_client,
+        mock_auto_lock_renewer,
+        mock_logger,
+    ):
+        mock_client = MagicMock()
+        mock_config = MagicMock(connection_string='Endpoint=sb://test/')
+        mock_service_bus_client.from_connection_string.return_value = mock_client
+        mock_client.get_topic_sender.return_value = MagicMock()
+        mock_auto_lock_renewer.return_value = MagicMock()
+
+        topic = AzureTopic(config=mock_config, topic_name='mock-topic', max_concurrent_messages=1)
+        topic._get_runtime_snapshot = MagicMock(
+            return_value=(
+                'memory=rss_mb=64.00, vms_mb=128.00, num_threads=3, '
+                'cpu=process_percent=55.00, system_percent=70.00, '
+                'gc=enabled=True, counts=(4, 5, 6), thresholds=(700, 10, 10), '
+                'stats=gen0[collections=3, collected=10, uncollectable=0]'
+            )
+        )
+
+        mock_message = MagicMock()
+        mock_message.message_id = 'message-2'
+        mock_message.locked_until_utc = '2026-03-17T09:39:33Z'
+        topic.receiver = MagicMock()
+
+        topic.lock_renew_receiver.renew_message_lock(mock_message)
+
+        mock_logger.info.assert_any_call(
+            'Attempting lock renewal for message %s; locked_until_utc=%s; runtime_snapshot=%s',
+            'message-2',
+            '2026-03-17T09:39:33Z',
+            'memory=rss_mb=64.00, vms_mb=128.00, num_threads=3, '
+            'cpu=process_percent=55.00, system_percent=70.00, '
+            'gc=enabled=True, counts=(4, 5, 6), thresholds=(700, 10, 10), '
+            'stats=gen0[collections=3, collected=10, uncollectable=0]',
+        )
+        topic.receiver.renew_message_lock.assert_called_once_with(mock_message)
+
 
 if __name__ == '__main__':
     unittest.main()
diff --git a/version.py b/version.py
@@ -1 +1,4 @@
-version = '0.0.25'
+version = '0.2.5.1'
+lastCommit = '30e447b527d6744b5ce72e9460f21e22675a3126'
+lastCommitShort = '30e4'
+buildDate = '2026-03-17'

Original file line number	Diff line number	Diff line change
`@@ -1 +1 @@`
`1`		`-__version__ = '0.0.25'`
	`1`	`+__version__ = '0.2.5.1'`