diff --git a/ingestor/adapters/grpc_server.py b/ingestor/adapters/grpc_server.py index 05cf6e8..a9fee06 100644 --- a/ingestor/adapters/grpc_server.py +++ b/ingestor/adapters/grpc_server.py @@ -17,7 +17,9 @@ class ReadTagServicer(opc_gateway_pb2_grpc.OpcGatewayServicer): self.active_read = active_read self.logger = logger # ponytail: slot_lookup wiring (address -> names from the current slot) - # lands in Task 18 (app wiring); optional callable keeps this servicer usable now. + # lands at the SIENTIAPDE-1974 integration -- `names` only exist in slot v2, + # which the Orchestrator produces after 1974. Until then ReadTag never + # persists (persisted=False); optional callable keeps this servicer usable now. self.slot_lookup = slot_lookup async def ReadTag(self, request, context): diff --git a/ingestor/app.py b/ingestor/app.py index 720d300..bf96140 100644 --- a/ingestor/app.py +++ b/ingestor/app.py @@ -52,18 +52,27 @@ async def main(): # NOSONAR # Gateway write/active-read/gRPC wiring (Task 18). Only built if prepare_ingestor # got far enough to have a data_manager (needed by CertProvider/DataSink) -- - # skipped entirely if prepare failed before that point. + # skipped entirely if prepare failed before that point. Construction can hit + # the network (MinIO bucket check, Postgres engine): a failure there must not + # kill the read path, so the app continues without the gateway runtime. runtime: GatewayRuntime | None = None if ingestor.ingestor_manager is not None: - runtime = GatewayRuntime( - config=GatewayConfig.from_env(), - data_manager=ingestor.ingestor_manager.data_manager, - logger=ingestor.logger, - notification_handler=ingestor.notification_handler, - metrics_controller=ingestor.metrics_controller, - metadata=ingestor.metadata, - ) - await runtime.start() + try: + runtime = GatewayRuntime( + config=GatewayConfig.from_env(), + data_manager=ingestor.ingestor_manager.data_manager, + logger=ingestor.logger, + notification_handler=ingestor.notification_handler, + metrics_controller=ingestor.metrics_controller, + metadata=ingestor.metadata, + ) + await runtime.start() + except Exception as e: + runtime = None + metrics.APP_ERRORS_TOTAL.labels(pod_id=POD_ID).inc() + ingestor.logger.error( + f'GatewayRuntime failed to initialize, continuing without gateway: {e}' + ) while not exit_signal.is_set(): # NOSONAR start_time = time() # Start loop timer diff --git a/ingestor/gateway_runtime.py b/ingestor/gateway_runtime.py index 22e14ec..16fe85f 100644 --- a/ingestor/gateway_runtime.py +++ b/ingestor/gateway_runtime.py @@ -53,6 +53,7 @@ class GatewayRuntime(SientiaMonitoring): self.config = config self.metadata = metadata or {} self._nats_started = False + self._nats_notified = False # notify once per outage, not per 5s retry minio = MinioRepository( endpoint=getenv('MINIO_ENDPOINT', 'localhost:9000'), @@ -138,8 +139,12 @@ class GatewayRuntime(SientiaMonitoring): try: await self.nats_consumer.start() self._nats_started = True + self._nats_notified = False except Exception as e: self.logger.error(f'GatewayRuntime: failed to start NATS write consumer: {e}') + if self._nats_notified: + return + self._nats_notified = True await self.send_notification_async( metadata=self.metadata, notification_id='GATEWAY_NATS_CONSUMER_START_FAILED', diff --git a/tests/unit/test_app.py b/tests/unit/test_app.py index 21454a9..ab038a3 100644 --- a/tests/unit/test_app.py +++ b/tests/unit/test_app.py @@ -346,3 +346,34 @@ async def test_main_prepare_ingestor_failure(mock_app_env, capsys): mock_ingestor_instance.logger.error.assert_called_once_with( 'Failed to prepare ingestor: Preparation failed' ) + + +@pytest.mark.asyncio +async def test_main_gateway_runtime_init_failure_keeps_read_path_alive(mock_app_env): + """GatewayRuntime construction hitting the network (MinIO/Postgres) and failing + must not kill the app: the main loop (read path) still runs and shuts down cleanly.""" + mock_ingestor_instance = mock_app_env['Ingestor'].return_value + mock_exit_signal = mock_app_env['mock_exit_signal'] + + mock_app_env['GatewayRuntime'].side_effect = Exception('MinIO unreachable') + mock_exit_signal.is_set.side_effect = [False, True] + mock_app_env['time_time'].side_effect = [10.0, 11.0] + + with pytest.raises(OsExitCalledError) as excinfo: + await app.main() + assert excinfo.value.code == 0 + + # Loop still ran and shutdown was clean, without any runtime calls + mock_ingestor_instance.loop.assert_called_once() + mock_ingestor_instance.shutdown.assert_called_once() + mock_runtime_instance = mock_app_env['GatewayRuntime'].return_value + mock_runtime_instance.idle_sweep.assert_not_called() + mock_runtime_instance.stop.assert_not_called() + + # Failure was logged and counted, but did not set exit_signal + assert any( + 'GatewayRuntime failed to initialize' in str(c) + for c in mock_ingestor_instance.logger.error.call_args_list + ) + mock_app_env['metrics_APP_ERRORS_TOTAL_labels_inc'].assert_called_once() + mock_exit_signal.set.assert_not_called() diff --git a/tests/unit/test_gateway_runtime.py b/tests/unit/test_gateway_runtime.py index f4ad64b..1a29050 100644 --- a/tests/unit/test_gateway_runtime.py +++ b/tests/unit/test_gateway_runtime.py @@ -106,7 +106,7 @@ async def test_nats_start_failure_does_not_raise_and_idle_sweep_retries(mocks): runtime = make_runtime() runtime.send_notification_async = AsyncMock() nats_mock = mocks['NatsWriteConsumer'].return_value - nats_mock.start = AsyncMock(side_effect=[Exception('boom'), None]) + nats_mock.start = AsyncMock(side_effect=[Exception('boom'), Exception('boom'), None]) await runtime.start() # must not raise @@ -114,10 +114,17 @@ async def test_nats_start_failure_does_not_raise_and_idle_sweep_retries(mocks): runtime.logger.error.assert_called_once() runtime.send_notification_async.assert_awaited_once() + await runtime.idle_sweep() # second failure: retry connects, but no re-notify + + assert runtime._nats_started is False + runtime.send_notification_async.assert_awaited_once() # still just one notification + await runtime.idle_sweep() assert runtime._nats_started is True - assert nats_mock.start.await_count == 2 + assert nats_mock.start.await_count == 3 + # flag reset on success: a future outage notifies again + assert runtime._nats_notified is False @pytest.mark.asyncio