From 8c144b2186228cbcef44edf256a3a81bd101d9f1 Mon Sep 17 00:00:00 2001 From: Lance-Drane Date: Wed, 9 Sep 2026 15:49:49 -0400 Subject: [PATCH 01/12] Userspace message: replace boolean has_error with enum message_state Signed-off-by: Lance-Drane --- .../control_plane/messages/userspace.py | 32 +++++++++++-------- tests/unit/test_userspace_message_headers.py | 11 ++++--- 2 files changed, 25 insertions(+), 18 deletions(-) diff --git a/src/intersect_sdk_common/control_plane/messages/userspace.py b/src/intersect_sdk_common/control_plane/messages/userspace.py index 6fd4a79..4abc33f 100644 --- a/src/intersect_sdk_common/control_plane/messages/userspace.py +++ b/src/intersect_sdk_common/control_plane/messages/userspace.py @@ -18,7 +18,7 @@ import datetime import uuid -from typing import Annotated +from typing import Annotated, Literal from pydantic import AwareDatetime, BaseModel, Field, field_serializer @@ -28,6 +28,14 @@ ) from ...version import intersect_sdk_version_string +MessageState = Literal[ + 'ERROR', # something threw an Exception and this Request/Response chain is OVER + 'DATA_REQUEST_PROCESSING_STARTED', # calling Data Plane to send the request body + 'USER_FUNCTION_PROCESSING_STARTED', # calling user function + 'DATA_RESPONSE_PROCESSING_STARTED', # calling Data Plane to send the response body + 'COMPLETE', # Response is done. ALWAYS the message state from a Request. +] + class UserspaceMessageHeaders(BaseModel): """ALL request/response/command messages must contain this header. @@ -118,17 +126,15 @@ class UserspaceMessageHeaders(BaseModel): usage, the payload would indicate the URI to where the data is stored on MinIO. """ - has_error: Annotated[ - bool, + message_state: Annotated[ + MessageState, Field( - False, - description='If this value is True, the payload will contain the error message (a string)', + 'COMPLETE', + description='If request message - always COMPLETE. If response message: Current state of the request as handled by the Service. COMPLETE = finished processing with success. ERROR = finished processing with error. Other responses indicate that processing is ongoing.', ), ] """ - If this flag is set to True, the payload will contain the error message (always a string). - - This should only be set to "True" on return messages sent by services - NEVER clients. + Primary control flow logic for Clients/Orchestrators in determining how to handle the message. """ # make sure all non-string fields are serialized into strings, even in Python code @@ -141,9 +147,9 @@ def _ser_uuid(self, uuid: uuid.UUID) -> str: def _ser_datetime(self, dt: datetime.datetime) -> str: return dt.isoformat() - @field_serializer('has_error', mode='plain') - def _ser_boolean(self, boolean: bool) -> str: - return str(boolean).lower() + # @field_serializer('has_error', mode='plain') + # def _ser_boolean(self, boolean: bool) -> str: + # return str(boolean).lower() @field_serializer('data_handler', mode='plain') def _ser_enum(self, enum: IntersectDataHandler) -> str: @@ -157,7 +163,7 @@ def create_userspace_message_headers( data_handler: IntersectDataHandler, campaign_id: uuid.UUID, request_id: uuid.UUID, - has_error: bool = False, + message_state: MessageState = 'COMPLETE', ) -> dict[str, str]: """Generate raw headers and write them into a generic data structure which can be handled by any broker protocol.""" return UserspaceMessageHeaders( @@ -170,7 +176,7 @@ def create_userspace_message_headers( created_at=datetime.datetime.now(tz=datetime.timezone.utc), operation_id=operation_id, data_handler=data_handler, - has_error=has_error, + message_state=message_state, ).model_dump(by_alias=True) diff --git a/tests/unit/test_userspace_message_headers.py b/tests/unit/test_userspace_message_headers.py index b941b3c..3149352 100644 --- a/tests/unit/test_userspace_message_headers.py +++ b/tests/unit/test_userspace_message_headers.py @@ -29,7 +29,7 @@ def test_valid_userspace_message_deserializes() -> None: headers = validate_userspace_message_headers(raw_headers) # check defaults assert headers.data_handler == IntersectDataHandler.MESSAGE - assert headers.has_error is False + assert headers.message_state == 'COMPLETE' # check type serializations assert isinstance(headers.message_id, uuid.UUID) assert isinstance(headers.created_at, datetime.datetime) @@ -47,11 +47,11 @@ def test_unusual_userspace_message_deserializes() -> None: 'sdk_version': '0.5.0', 'created_at': '2024', 'data_handler': 'MINIO', - 'has_error': 'true', + 'message_state': 'ERROR', } headers = validate_userspace_message_headers(raw_headers) assert headers.data_handler == IntersectDataHandler.MINIO - assert headers.has_error is True + assert headers.message_state == 'ERROR' # even on strict mode, Pydantic can validate an integer as a string type, i.e. '"2024"' - it parses this as number of seconds since the Unix epoch assert headers.created_at.year == 1970 @@ -85,7 +85,7 @@ def test_invalid_does_not_deserialize() -> None: 'sdk_version': '1.0.0+20130313144700', 'created_at': '2024-01-19T20:21:14.045591', 'data_handler': 'COBOL', - 'has_error': 'I_AM_NOT_A_BOOLEAN', + 'message_state': 'I_AM_NOT_A_BOOLEAN', } with pytest.raises(ValidationError) as err: validate_userspace_message_headers(raw_headers) @@ -105,6 +105,7 @@ def test_invalid_does_not_deserialize() -> None: assert {'type': 'string_pattern_mismatch', 'loc': ('sdk_version',)} in errors # can't transpose these values into the enumerations assert {'type': 'enum', 'loc': ('data_handler',)} in errors + assert {'type': 'literal_error', 'loc': ('message_state',)} in errors def test_create_userspace_message() -> None: @@ -131,7 +132,7 @@ def test_create_userspace_message() -> None: # enforce UTC assert msg['created_at'][-6:] == '+00:00' # this should be lowercase for maximum language capability - assert msg['has_error'] == 'false' + assert msg['message_state'] == 'COMPLETE' assert msg['operation_id'] == 'operation' assert msg['data_handler'] == 'MESSAGE' assert msg['sdk_version'] == intersect_sdk_version_string From d4167449cca8fa05838379bc875c55876552d229 Mon Sep 17 00:00:00 2001 From: Lance-Drane Date: Wed, 9 Sep 2026 19:15:15 -0400 Subject: [PATCH 02/12] remove HierarchyConfig and add IntersectConfig Signed-off-by: Lance-Drane --- src/intersect_sdk_common/__init__.py | 6 +- src/intersect_sdk_common/config.py | 86 ++++++------------- .../data_plane/data_plane_manager.py | 12 +-- .../data_plane/minio_utils.py | 16 ++-- tests/unit/test_config.py | 62 ++++++------- 5 files changed, 71 insertions(+), 111 deletions(-) diff --git a/src/intersect_sdk_common/__init__.py b/src/intersect_sdk_common/__init__.py index d9e36a9..37d3218 100644 --- a/src/intersect_sdk_common/__init__.py +++ b/src/intersect_sdk_common/__init__.py @@ -15,7 +15,7 @@ ControlProvider, DataStoreConfig, DataStoreConfigMap, - HierarchyConfig, + IntersectConfig, ) from .control_plane.control_plane_manager import ControlPlaneManager from .control_plane.definitions import MessageCallback @@ -40,8 +40,8 @@ 'DataPlaneManager', 'DataStoreConfig', 'DataStoreConfigMap', - 'HierarchyConfig', 'IntersectApplicationError', + 'IntersectConfig', 'IntersectDataHandler', 'IntersectError', 'IntersectMimeType', @@ -60,7 +60,7 @@ 'ControlProvider': '.config', 'DataStoreConfig': '.config', 'DataStoreConfigMap': '.config', - 'HierarchyConfig': '.config', + 'IntersectConfig': '.config', 'ControlPlaneManager': '.control_plane.control_plane_manager', 'MessageCallback': '.control_plane.definitions', 'IntersectDataHandler': '.core_definitions', diff --git a/src/intersect_sdk_common/config.py b/src/intersect_sdk_common/config.py index 50483ad..d98ca95 100644 --- a/src/intersect_sdk_common/config.py +++ b/src/intersect_sdk_common/config.py @@ -3,82 +3,30 @@ from dataclasses import dataclass, field from typing import Annotated, Literal -from pydantic import BaseModel, ConfigDict, Field, PositiveInt +from pydantic import BaseModel, Field, PositiveInt from .core_definitions import IntersectDataHandler -HIERARCHY_REGEX = r'^[a-z]((?!--)[a-z0-9-]){2,62}$' +HIERARCHY_REGEX = r'[a-z0-9][-a-z0-9]{2,62}' """ -The hierarchy regex needs to be fairly restricted due to the number of different -systems we want to be compatible with. The rules: +The hierarchy regex should accomplish the following: - Only allow unreserved characters (alphanumeric and .-~_): https://datatracker.ietf.org/doc/html/rfc3986#section-2.3 - Require lowercase letters to avoid incompatibilities with case-insensitive systems. -- MinIO has been found to forbid _ and ~ characters -- MinIO requires an alphanumeric character at the start of the string -- No adjacent non-alphanumeric characters allowed - Range should be from 3-63 characters +- Be CLI friendly (don't start with hyphen) and URI friendly (don't use underscores) -The following commit tracks several issues with MINIO: https://code.ornl.gov/intersect/additive-manufacturing/ros-intersect-adapter/-/commit/fa71b791be0ccf1a5884910b5be3b5239cf9896f +This enables maximum integration with other systems (i.e. MINIO) without having to transform the hierarchy. """ +# HIERARCHY_REGEX = r'^[a-z0-9]((?!--)[a-z0-9-]){2,62}$' +# If using this regex, you must configure the Pydantic ConfigDict regex_engine to be 'python-re' to support lookaheads +# Used to disallow multiple hyphens +# should work in HTML5 forms out of the gate ControlProvider = Literal['mqtt5.0', 'amqp0.9.1'] """The type of broker we connect to.""" -class HierarchyConfig(BaseModel): - """Configuration for registering this service in a system-of-system architecture.""" - - service: Annotated[str, Field(pattern=HIERARCHY_REGEX)] - """ - The name of this application - should be unique within an INTERSECT system - """ - - subsystem: str | None = Field(default=None, pattern=HIERARCHY_REGEX) - """ - An associated subsystem / service-grouping of the system (should be unique within an INTERSECT system) - """ - - system: Annotated[str, Field(pattern=HIERARCHY_REGEX)] - """ - Name of the "system", could also be thought of as a "device" (should be unique within a facility) - """ - - facility: Annotated[str, Field(pattern=HIERARCHY_REGEX)] - """ - Name of the facility (an ORNL institutional designation, i.e. 'neutrons') (NOT abbreviated, should be unique within an organization) - """ - - organization: Annotated[str, Field(pattern=HIERARCHY_REGEX)] - """ - Name of the organization (i.e. 'ornl') (NOT abbreviated) (should be unique in an INTERSECT cluster) - """ - - def hierarchy_string(self, join_str: str = '') -> str: - """Get the full hierarchy string. This is mostly used internally, but if you're developing a client, it could potentially be helpful. - - Params - join_str: String used to separate different hierarchy parts in the full string (default: empty string). - - Returns: - Single string, which will contain all system-of-system parts. For optional parts not configured (i.e. - no subsystem), they will be represented by a "-" character. - """ - if not self.subsystem: - return join_str.join([self.organization, self.facility, self.system, '-', self.service]) - return join_str.join( - [ - self.organization, - self.facility, - self.system, - self.subsystem, - self.service, - ] - ) - - # we need to use the Python regex engine instead of the Rust regex engine here, because Rust's does not support lookaheads - model_config = ConfigDict(regex_engine='python-re') - - @dataclass class ControlPlaneConfig: """Configuration for interacting with a broker.""" @@ -181,3 +129,19 @@ def get_missing_data_store_types(self) -> set[IntersectDataHandler]: if not self.minio: missing.add(IntersectDataHandler.MINIO) return missing + + +class IntersectConfig(BaseModel): + """The response type used by INTERSECT-SDK Services and Clients to understand how to connect to the INTERSECT ecosystem. + + This object should only be used by the SDK, and not users directly. + """ + + system_name: Annotated[str, Field(pattern=HIERARCHY_REGEX)] + """The highest level of namespacing on the broker address, provided by value configured by registry service administrator. Important for connecting different INTERSECT systems.""" + service_name: str + """The service namespacing of these credentials, reserved by user on registry service.""" + brokers: Annotated[list[ControlPlaneConfig], Field(min_length=1)] + """List of control plane configurations the SDK should use.""" + data_stores: Annotated[DataStoreConfigMap, Field(default_factory=lambda: DataStoreConfigMap())] + """List of data plane configurations the SDK should use.""" diff --git a/src/intersect_sdk_common/data_plane/data_plane_manager.py b/src/intersect_sdk_common/data_plane/data_plane_manager.py index 5c86c6d..4497d3b 100644 --- a/src/intersect_sdk_common/data_plane/data_plane_manager.py +++ b/src/intersect_sdk_common/data_plane/data_plane_manager.py @@ -4,7 +4,7 @@ from pydantic import TypeAdapter, ValidationError -from ..config import DataStoreConfigMap, HierarchyConfig +from ..config import DataStoreConfigMap from ..core_definitions import IntersectDataHandler, IntersectMimeType from ..exceptions import IntersectError from ..logger import logger @@ -19,14 +19,16 @@ class DataPlaneManager: The API supports extensive plug-and-play for different data providers. """ - def __init__(self, hierarchy: HierarchyConfig, data_configs: DataStoreConfigMap) -> None: + def __init__(self, system: str, service: str, data_configs: DataStoreConfigMap) -> None: """Inside the constructor, we verify that all data configuration credentials are correct. Params: - hierarchy: Hierarchy configuration + system: name of system + service: name of service data_configs: data configuration """ - self._hierarchy = hierarchy + self._system = system + self._service = service self._minio_providers = list(map(create_minio_store, data_configs.minio)) # warn users about missing data plane @@ -99,7 +101,7 @@ def outgoing_message_data_handler( raise IntersectError provider = random.choice(self._minio_providers) # noqa: S311 (TODO choose a MINIO provider better than at random - this may be determined from external message params) minio_payload = send_minio_object( - function_response, provider, content_type, self._hierarchy + function_response, provider, content_type, self._system, self._service ) return MINIO_ADAPTER.dump_json(minio_payload) diff --git a/src/intersect_sdk_common/data_plane/minio_utils.py b/src/intersect_sdk_common/data_plane/minio_utils.py index 501593f..579ec13 100644 --- a/src/intersect_sdk_common/data_plane/minio_utils.py +++ b/src/intersect_sdk_common/data_plane/minio_utils.py @@ -11,7 +11,7 @@ from urllib3.exceptions import MaxRetryError from urllib3.util import parse_url -from ..config import DataStoreConfig, HierarchyConfig +from ..config import DataStoreConfig from ..core_definitions import IntersectMimeType from ..exceptions import IntersectError from ..logger import logger @@ -34,19 +34,20 @@ class MinioPayload(TypedDict): """ -def _condense_minio_bucket_name(hierarchy: HierarchyConfig) -> str: +def _condense_minio_bucket_name(system: str, service: str) -> str: """Condense a hierarchy string into a string less than 64 characters. This function is needed to handle MINIO bucket names. Collisions should be extremely rare, and it should be fairly straightforward to identify a specific bucket for MINIO admins. - Bucket name = first characters (up to 6) of service name + hyphen + sha224 of full hierarchy string + Bucket name = first characters (up to 6) of service name + hyphen + sha224 of system-service combo string TODO in the future, MINIO calls should be system-level only, so only the system + facility + organization should need to be hashed. Also, potentially come up with a better hashing algorithm (though sha224 is compressed enough, and gives us 56 characters). """ - return f'{hierarchy.service[:6]}-{sha224(hierarchy.hierarchy_string().encode()).hexdigest()}' + encoded = f'{system}-{service}'.encode() + return f'{service[:6]}-{sha224(encoded).hexdigest()}' def create_minio_store(config: DataStoreConfig) -> Minio: @@ -89,7 +90,7 @@ def create_minio_store(config: DataStoreConfig) -> Minio: def send_minio_object( - data: bytes, provider: Minio, content_type: IntersectMimeType, hierarchy: HierarchyConfig + data: bytes, provider: Minio, content_type: IntersectMimeType, system: str, service: str ) -> MinioPayload: """Core function to save data in MINIO. @@ -97,14 +98,15 @@ def send_minio_object( data: user response data, as bytes provider: the Minio client content_type: the content type of the body - hierarchy: the hierarchy configuration + system: the system name + service: the service name Returns: The MINIO payload which gets sent in the actual message. Raises: IntersectException - if any non-fatal MinIO error is caught """ - bucket_name = _condense_minio_bucket_name(hierarchy) + bucket_name = _condense_minio_bucket_name(system, service) # mimetypes.guess_extension() is a nice-to-have for MINIO preview, but isn't essential. object_id = str(uuid4()) + (mimetypes.guess_extension(content_type) or '') try: diff --git a/tests/unit/test_config.py b/tests/unit/test_config.py index 8bbb66d..cbff2bc 100644 --- a/tests/unit/test_config.py +++ b/tests/unit/test_config.py @@ -4,45 +4,11 @@ from intersect_sdk_common.config import ( ControlPlaneConfig, DataStoreConfig, - HierarchyConfig, + IntersectConfig, ) # TESTS ##################### - -def test_empty_hierarchy(): - with pytest.raises(ValidationError) as ex: - HierarchyConfig() # type: ignore[call-arg] - errors = ex.value.errors() - assert len(errors) == 4 - assert all(e['type'] == 'missing' for e in errors) - locations = [e['loc'] for e in errors] - assert ('organization',) in locations - assert ('facility',) in locations - assert ('system',) in locations - assert ('service',) in locations - - -def test_invalid_hierarchy(): - with pytest.raises(ValidationError) as ex: - HierarchyConfig( - organization='no.periods', - facility='no_underscores', - system='', - subsystem='no/slashes', - service='a', - ) - errors = ex.value.errors() - assert len(errors) == 5 - assert all(e['type'] == 'string_pattern_mismatch' for e in errors) - locations = [e['loc'] for e in errors] - assert ('organization',) in locations - assert ('facility',) in locations - assert ('system',) in locations - assert ('subsystem',) in locations - assert ('service',) in locations - - # NOTE: with dataclasses, need to validate dictionaries instead of the dataclass directly @@ -101,3 +67,29 @@ def test_invalid_data_plane_config(): assert {'type': 'string_too_short', 'loc': ('password',)} in errors assert {'type': 'string_too_short', 'loc': ('host',)} in errors assert {'type': 'greater_than', 'loc': ('port',)} in errors + + +def test_missing_intersect_config(): + with pytest.raises(ValidationError) as ex: + TypeAdapter(IntersectConfig).validate_python({}) + errors = [{'type': e['type'], 'loc': e['loc']} for e in ex.value.errors()] + assert len(errors) == 3 + assert {'type': 'missing', 'loc': ('brokers',)} in errors + assert {'type': 'missing', 'loc': ('system_name',)} in errors + assert {'type': 'missing', 'loc': ('service_name',)} in errors + + +def test_invalid_intersect_config(): + with pytest.raises(ValidationError) as ex: + TypeAdapter(IntersectConfig).validate_python( + IntersectConfig( + system_name='I AM INVALID', + service_name=7, + brokers=[], + ).__dict__ + ) + errors = [{'type': e['type'], 'loc': e['loc']} for e in ex.value.errors()] + assert len(errors) == 3 + assert {'type': 'string_pattern_mismatch', 'loc': ('system_name',)} in errors + assert {'type': 'string_type', 'loc': ('service_name',)} in errors + assert {'type': 'too_short', 'loc': ('brokers',)} in errors From 6e8053ee9555793b84bb2f4e0558c5a7df0ca0e1 Mon Sep 17 00:00:00 2001 From: Lance-Drane Date: Wed, 9 Sep 2026 22:05:47 -0400 Subject: [PATCH 03/12] IntersectConfig mandates single broker Signed-off-by: Lance-Drane --- src/intersect_sdk_common/config.py | 7 +++---- .../control_plane/brokers/amqp_client.py | 2 +- tests/unit/test_config.py | 5 ++--- 3 files changed, 6 insertions(+), 8 deletions(-) diff --git a/src/intersect_sdk_common/config.py b/src/intersect_sdk_common/config.py index d98ca95..f94b701 100644 --- a/src/intersect_sdk_common/config.py +++ b/src/intersect_sdk_common/config.py @@ -70,8 +70,7 @@ class ControlPlaneConfig: NOTE: INTERSECT currently only supports AMQP and MQTT. """ - # TODO default this to False once the registry service is in place - is_root: bool = True + is_root: bool = False """ Whether or not the broker credentials are for connecting as a root user. @@ -141,7 +140,7 @@ class IntersectConfig(BaseModel): """The highest level of namespacing on the broker address, provided by value configured by registry service administrator. Important for connecting different INTERSECT systems.""" service_name: str """The service namespacing of these credentials, reserved by user on registry service.""" - brokers: Annotated[list[ControlPlaneConfig], Field(min_length=1)] - """List of control plane configurations the SDK should use.""" + broker: ControlPlaneConfig + """Control plane configuration the SDK should use with this Service.""" data_stores: Annotated[DataStoreConfigMap, Field(default_factory=lambda: DataStoreConfigMap())] """List of data plane configurations the SDK should use.""" diff --git a/src/intersect_sdk_common/control_plane/brokers/amqp_client.py b/src/intersect_sdk_common/control_plane/brokers/amqp_client.py index 2d0c424..70b88a5 100644 --- a/src/intersect_sdk_common/control_plane/brokers/amqp_client.py +++ b/src/intersect_sdk_common/control_plane/brokers/amqp_client.py @@ -1,7 +1,7 @@ """This module handles ALL AMQP protocol logic in INTERSECT. We seek to entirely abstract protocols away from users. This is a very specific pub-sub model which assumes a single topic exchange. -AMQP topics in INTERSECT generally look like ${ORGANIZATION}.${FACILITY}.${SYSTEM}.${SUBSYSTEM}.${SERVICE}.${MESSAGE_TYPE} . (TODO change to ${SYSTEM}.${SERVICE}.${MESSAGE_TYPE}) +AMQP topics in INTERSECT generally look like ${SYSTEM}.${SERVICE}.${MESSAGE_TYPE} , with additional extensions after ${MESSAGE_TYPE} optional. MESSAGE_TYPE refers to INTERSECT domain messages - we do not allow users to determine their own message types directly, and every message has a message type. SERVICE refers to a specific application, generally the microservice which is handling the message. SYSTEM is generally the level where Auth should occur, and where you should configure access control on the broker itself. diff --git a/tests/unit/test_config.py b/tests/unit/test_config.py index cbff2bc..edfedeb 100644 --- a/tests/unit/test_config.py +++ b/tests/unit/test_config.py @@ -74,7 +74,7 @@ def test_missing_intersect_config(): TypeAdapter(IntersectConfig).validate_python({}) errors = [{'type': e['type'], 'loc': e['loc']} for e in ex.value.errors()] assert len(errors) == 3 - assert {'type': 'missing', 'loc': ('brokers',)} in errors + assert {'type': 'missing', 'loc': ('broker',)} in errors assert {'type': 'missing', 'loc': ('system_name',)} in errors assert {'type': 'missing', 'loc': ('service_name',)} in errors @@ -85,11 +85,10 @@ def test_invalid_intersect_config(): IntersectConfig( system_name='I AM INVALID', service_name=7, - brokers=[], ).__dict__ ) errors = [{'type': e['type'], 'loc': e['loc']} for e in ex.value.errors()] assert len(errors) == 3 assert {'type': 'string_pattern_mismatch', 'loc': ('system_name',)} in errors assert {'type': 'string_type', 'loc': ('service_name',)} in errors - assert {'type': 'too_short', 'loc': ('brokers',)} in errors + assert {'type': 'missing', 'loc': ('broker',)} in errors From 286e1a18aa4c703940fdc3bb9fa579f641273400 Mon Sep 17 00:00:00 2001 From: Lance-Drane Date: Thu, 10 Sep 2026 02:07:21 -0400 Subject: [PATCH 04/12] add publish_message_single and refresh_config API, move system_name to ControlPlaneConfig Signed-off-by: Lance-Drane --- src/intersect_sdk_common/config.py | 7 +- .../control_plane/brokers/amqp_client.py | 53 ++++++++------- .../control_plane/brokers/broker_client.py | 10 +++ .../control_plane/brokers/mqtt_client.py | 37 ++++++----- .../control_plane/control_plane_manager.py | 66 ++++++++++++++----- tests/fixtures/mock_client.py | 5 +- tests/fixtures/mock_service.py | 5 +- tests/integration/test_control_plane.py | 52 ++++++++------- tests/unit/test_config.py | 14 ++-- 9 files changed, 159 insertions(+), 90 deletions(-) diff --git a/src/intersect_sdk_common/config.py b/src/intersect_sdk_common/config.py index f94b701..1375bdd 100644 --- a/src/intersect_sdk_common/config.py +++ b/src/intersect_sdk_common/config.py @@ -37,6 +37,9 @@ class ControlPlaneConfig: """ # TODO - support more protocols and protocol versions as needed - see https://www.asyncapi.com/docs/reference/specification/v2.6.0#serverObject + system_name: Annotated[str, Field(pattern=HIERARCHY_REGEX)] + """The highest level of namespacing on the broker address, provided by value configured by registry service administrator. Important for connecting different INTERSECT systems.""" + username: Annotated[str, Field(min_length=1)] """ Username credentials for broker connection. @@ -136,10 +139,8 @@ class IntersectConfig(BaseModel): This object should only be used by the SDK, and not users directly. """ - system_name: Annotated[str, Field(pattern=HIERARCHY_REGEX)] - """The highest level of namespacing on the broker address, provided by value configured by registry service administrator. Important for connecting different INTERSECT systems.""" service_name: str - """The service namespacing of these credentials, reserved by user on registry service.""" + """The service namespacing of this microservice, reserved by user (or, if client, obtained from credential) on registry service.""" broker: ControlPlaneConfig """Control plane configuration the SDK should use with this Service.""" data_stores: Annotated[DataStoreConfigMap, Field(default_factory=lambda: DataStoreConfigMap())] diff --git a/src/intersect_sdk_common/control_plane/brokers/amqp_client.py b/src/intersect_sdk_common/control_plane/brokers/amqp_client.py index 70b88a5..02a3b42 100644 --- a/src/intersect_sdk_common/control_plane/brokers/amqp_client.py +++ b/src/intersect_sdk_common/control_plane/brokers/amqp_client.py @@ -27,6 +27,7 @@ from pika.frame import Frame from pika.spec import Basic, BasicProperties + from ...config import ControlPlaneConfig from ..control_plane_manager import ControlPlaneManager from ..definitions import MessageCallback @@ -89,36 +90,16 @@ class AMQPClient(BrokerClient): def __init__( self, - host: str, - port: int, - username: str, - password: str, + control_plane_config: ControlPlaneConfig, control_plane_manager: ControlPlaneManager, - is_root: bool, ) -> None: """The default constructor. Args: - host: String for hostname of AMQP broker - port: port number of AMQP broker - username: username credentials for AMQP broker - password: password credentials for AMQP broker + control_plane_config: configuration for the AMQP broker connection control_plane_manager: reference to the ControlPlaneManager instance, remember to ONLY use functions which do not mutate state - is_root: Whether or not the client can configure exchanges and queues themselves (core services), or if this must be delegated to a Core Service (SDK Clients/Services) """ - self._connection_params = pika.ConnectionParameters( - host=host, - port=port, - virtual_host='/', - credentials=pika.PlainCredentials(username, password), - connection_attempts=3, - # if not specified, this value is obtained by the broker. RabbitMQ sets it to 60s by default - heartbeat=10, - blocked_connection_timeout=5.0, - retry_delay=0.5, - ) - - self._is_root = is_root + self._handle_config(control_plane_config) # The pika connection to the broker self._connection: pika.adapters.SelectConnection = None @@ -141,6 +122,24 @@ def __init__( # tracking both channels is the best way to handle continuations self._channel_flags = MultiFlagThreadEvent(2) + def _handle_config(self, control_plane_config: ControlPlaneConfig) -> None: + self._connection_params = pika.ConnectionParameters( + host=control_plane_config.host, + port=control_plane_config.port or 5672, + virtual_host='/', + credentials=pika.PlainCredentials( + control_plane_config.username, control_plane_config.password + ), + connection_attempts=3, + # if not specified, this value is obtained by the broker. RabbitMQ sets it to 60s by default + heartbeat=10, + blocked_connection_timeout=5.0, + retry_delay=0.5, + ) + + self._is_root = control_plane_config.is_root + self._system_name = control_plane_config.system_name + def connect(self) -> None: """Connect to the defined broker. @@ -253,6 +252,14 @@ def unsubscribe(self, topic: str) -> None: if consumer_tag_info: self._cancel_consumer_tag(amqp_topic, consumer_tag_info.consumer_tag) + def system_name(self) -> str: + """Return the ecosystem system name.""" + return self._system_name + + def refresh_config(self, config: ControlPlaneConfig) -> None: + """Refresh the config with the new one from the registry service.""" + self._handle_config(config) + def _cancel_consumer_tag(self, topic: str, consumer_tag: str) -> None: if self._channel_in and self._channel_in.is_open: cb = functools.partial( diff --git a/src/intersect_sdk_common/control_plane/brokers/broker_client.py b/src/intersect_sdk_common/control_plane/brokers/broker_client.py index 3839358..2b7f511 100644 --- a/src/intersect_sdk_common/control_plane/brokers/broker_client.py +++ b/src/intersect_sdk_common/control_plane/brokers/broker_client.py @@ -5,6 +5,8 @@ from typing import Protocol +from ...config import ControlPlaneConfig + class BrokerClient(Protocol): """Abstract definition of a Broker Client. @@ -80,3 +82,11 @@ def unsubscribe(self, topic: str) -> None: topic: Topic to unsubscribe from. """ ... + + def system_name(self) -> str: + """Returns the system name.""" + ... + + def refresh_config(self, config: ControlPlaneConfig) -> None: + """If the configuration needs to be updated, update it.""" + ... diff --git a/src/intersect_sdk_common/control_plane/brokers/mqtt_client.py b/src/intersect_sdk_common/control_plane/brokers/mqtt_client.py index 7d49ea9..a046515 100644 --- a/src/intersect_sdk_common/control_plane/brokers/mqtt_client.py +++ b/src/intersect_sdk_common/control_plane/brokers/mqtt_client.py @@ -25,6 +25,7 @@ from paho.mqtt.client import DisconnectFlags from paho.mqtt.reasoncodes import ReasonCode + from ...config import ControlPlaneConfig from ..control_plane_manager import ControlPlaneManager @@ -55,27 +56,16 @@ class MQTTClient(BrokerClient): def __init__( self, - host: str, - port: int, - username: str, - password: str, + control_plane_config: ControlPlaneConfig, control_plane_manager: ControlPlaneManager, - uid: str | None = None, ) -> None: """The default constructor. Args: - host: String for hostname of MQTT broker - port: port number of MQTT broker - username: username credentials for MQTT broker - password: password credentials for MQTT broker + control_plane_config: configuration for the MQTT broker connection control_plane_manager: reference to the ControlPlaneManager instance, remember to ONLY use functions which do not mutate state - uid: A string representing the unique id to identify the client. """ - # Unique id for the MQTT broker to associate this client with - self.uid = uid if uid else str(uuid.uuid4()) - self.host = host - self.port = port + self.uid = str(uuid.uuid4()) # Create a client to connect to RabbitMQ self._connection = paho_client.Client( @@ -83,7 +73,8 @@ def __init__( protocol=paho_client.MQTTv5, client_id=self.uid, ) - self._connection.username_pw_set(username=username, password=password) + + self._handle_config(control_plane_config) # Whether the connection is currently active self._connected = False @@ -104,6 +95,14 @@ def __init__( self._connection.on_disconnect = self._handle_disconnect self._connection.on_message = self._on_message + def _handle_config(self, control_plane_config: ControlPlaneConfig) -> None: + self.host = control_plane_config.host + self.port = control_plane_config.port or 1883 + self._connection.username_pw_set( + username=control_plane_config.username, password=control_plane_config.password + ) + self._system_name = control_plane_config.system_name + def connect(self) -> None: """Connect to the defined broker.""" # Create a client to connect to RabbitMQ @@ -186,6 +185,14 @@ def unsubscribe(self, topic: str) -> None: resolved_topic = _hierarchy_2_mqtt(topic) self._connection.unsubscribe(resolved_topic) + def system_name(self) -> str: + """Get the ecosystem system name.""" + return self._system_name + + def refresh_config(self, config: ControlPlaneConfig) -> None: + """Refresh the config with the new one from the registry service.""" + self._handle_config(config) + def _on_message( self, client: paho_client.Client, # noqa: ARG002 diff --git a/src/intersect_sdk_common/control_plane/control_plane_manager.py b/src/intersect_sdk_common/control_plane/control_plane_manager.py index fa65112..8bcb3a7 100644 --- a/src/intersect_sdk_common/control_plane/control_plane_manager.py +++ b/src/intersect_sdk_common/control_plane/control_plane_manager.py @@ -34,22 +34,15 @@ def _create_control_provider( ) return AMQPClient( - host=config.host, - port=config.port or 5672, - username=config.username, - password=config.password, + control_plane_config=config, control_plane_manager=control_plane_manager, - is_root=config.is_root, ) # MQTT from .brokers.mqtt_client import MQTTClient # noqa: PLC0415 (lazy load MQTT modules) return MQTTClient( - host=config.host, - port=config.port or 1883, - username=config.username, - password=config.password, + control_plane_config=config, control_plane_manager=control_plane_manager, ) @@ -77,6 +70,13 @@ def __init__( self._topics_to_handlers: dict[str, TopicHandler] = {} self._wildcards: dict[str, TopicHandler] = {} + def refresh_broker_config(self, control_config: ControlPlaneConfig) -> None: + """If a broker configuration is out of date, update it.""" + for provider in filter( + lambda p: p.system_name() == control_config.system_name, self._control_providers + ): + provider.refresh_config(control_config) + def add_subscription_channel( self, channel: str, @@ -202,7 +202,7 @@ def disconnect(self) -> None: for provider in self._control_providers: provider.disconnect() - def publish_message( + def publish_message_bulk( self, channel: str, payload: bytes, @@ -210,13 +210,47 @@ def publish_message( headers: dict[str, str], persist: bool, ) -> None: - """Publish message on channel for all brokers.""" - if self.is_connected(): - for provider in self._control_providers: + """Publish message on channel for all brokers. + + Used for Event and Lifecycle messages. + """ + for i, provider in enumerate(self._control_providers): + if provider.is_connected(): provider.publish(channel, payload, content_type, headers, persist) - else: - # TODO may want more robust error handling here - logger.error('Cannot send message, providers are not connected') + else: + logger.error( + 'Could not publish message to provider at index %d , debug %s', i, headers + ) + + def publish_message_single( + self, + channel: str, + payload: bytes, + content_type: str, + headers: dict[str, str], + persist: bool, + system_name: str, + ) -> bool: + """Publish message on channel for single broker, as determined by system_name param. + + Used for Userspace messages. + + Returns true if publish successful, false if publish failed + """ + provider = next( + (p for p in self._control_providers if p.system_name() == system_name), None + ) + if provider and provider.is_connected(): + provider.publish(channel, payload, content_type, headers, persist) + return True + + # TODO may want more robust error handling here + logger.error( + 'Could not publish message to provider with system_name %s , debug %s', + system_name, + headers, + ) + return False def is_connected(self) -> bool: """Check that we are connected to ALL configured brokers. diff --git a/tests/fixtures/mock_client.py b/tests/fixtures/mock_client.py index a9f8916..74d530d 100644 --- a/tests/fixtures/mock_client.py +++ b/tests/fixtures/mock_client.py @@ -11,6 +11,7 @@ class MockClient: """A mock client for testing purposes.""" def __init__(self, config: ControlPlaneConfig, client_sub_channel: str) -> None: + self.system_name = config.system_name self.control_plane_manager = ControlPlaneManager(control_configs=[config]) self.campaign_id = uuid.uuid4() self.event = threading.Event() @@ -37,7 +38,9 @@ def publish_message(self, channel: str, message: bytes) -> None: campaign_id=self.campaign_id, request_id=uuid.uuid4(), ) - self.control_plane_manager.publish_message(channel, message, 'text/plain', headers, True) + self.control_plane_manager.publish_message_single( + channel, message, 'text/plain', headers, True, self.system_name + ) def wait(self, timeout: float) -> None: self.event.wait(timeout) diff --git a/tests/fixtures/mock_service.py b/tests/fixtures/mock_service.py index e7ea5f9..e21e616 100644 --- a/tests/fixtures/mock_service.py +++ b/tests/fixtures/mock_service.py @@ -10,6 +10,7 @@ class MockService: """A mock service for testing purposes.""" def __init__(self, config: ControlPlaneConfig, channels: list[str]) -> None: + self.system_name = config.system_name self.control_plane_manager = ControlPlaneManager(control_configs=[config]) queue_name = str(uuid.uuid4()) # make multiple queues to ensure we make sure messages are directed to the appropriate queues @@ -30,8 +31,8 @@ def on_message(self, message: bytes, content_type: str, headers: dict[str, str]) uuid.UUID(headers['request_id']), ) content_type = 'text/plain' - self.control_plane_manager.publish_message( - headers['source'], reply_message, content_type, reply_headers, True + self.control_plane_manager.publish_message_single( + headers['source'], reply_message, content_type, reply_headers, True, self.system_name ) def connect(self) -> None: diff --git a/tests/integration/test_control_plane.py b/tests/integration/test_control_plane.py index 160c91e..3a8e26a 100644 --- a/tests/integration/test_control_plane.py +++ b/tests/integration/test_control_plane.py @@ -7,18 +7,24 @@ from ..fixtures.mock_client import MockClient from ..fixtures.mock_service import MockService +TEST_SVC_SYSTEM_NAME = 'intersect-system' +TEST_SVC_CLIENT_NAME = 'intersect-system-client' -def get_config(protocol: ControlProvider, port: int) -> ControlPlaneConfig: + +def get_config(protocol: ControlProvider, port: int, system_name: str) -> ControlPlaneConfig: return ControlPlaneConfig( protocol=protocol, username='intersect_username', password='intersect_password', host='127.0.0.1', port=port, + system_name=system_name, + is_root=True, # cannot easily test is_root=False without the registry service or a mock which creates the queues ) # IMPORTANT: make sure that at least one channel topic is unique between test cases per client and service, otherwise messages may not be handled properly on the brokers +# IMPORTANT: if not testing wildcards, first item needs to be the system name @pytest.mark.parametrize( ('protocol', 'port', 'client_pub_channel', 'svc_sub_channel', 'client_sub_channel'), [ @@ -26,60 +32,60 @@ def get_config(protocol: ControlProvider, port: int) -> ControlPlaneConfig: ( 'amqp0.9.1', 5672, - 'org/fac/sys/sub/svc/test-subscription-strings-svc-1', - 'org/fac/sys/sub/svc/test-subscription-strings-svc-1', - 'org/fac/sys/sub/svc/test-subscription-strings-client-1', + f'{TEST_SVC_SYSTEM_NAME}/fac/sys/sub/svc/test-subscription-strings-svc-1', + f'{TEST_SVC_SYSTEM_NAME}/fac/sys/sub/svc/test-subscription-strings-svc-1', + f'{TEST_SVC_CLIENT_NAME}/fac/sys/sub/svc/test-subscription-strings-client-1', ), ( 'mqtt5.0', 1883, - 'org/fac/sys/sub/svc/test-subscription-strings-svc-2', - 'org/fac/sys/sub/svc/test-subscription-strings-svc-2', - 'org/fac/sys/sub/svc/test-subscription-strings-client-2', + f'{TEST_SVC_SYSTEM_NAME}/fac/sys/sub/svc/test-subscription-strings-svc-2', + f'{TEST_SVC_SYSTEM_NAME}/fac/sys/sub/svc/test-subscription-strings-svc-2', + f'{TEST_SVC_CLIENT_NAME}/fac/sys/sub/svc/test-subscription-strings-client-2', ), # single-level wildcards, needed for iHub ( 'amqp0.9.1', 5672, - 'org/fac/sys/sub/svc/test-subscription-strings-svc-3', + f'{TEST_SVC_SYSTEM_NAME}/fac/sys/sub/svc/test-subscription-strings-svc-3', '*/*/*/*/*/test-subscription-strings-svc-3', - 'tmp/tmp/tmp/tmp/tmp/test-subscription-strings-client-3', + f'{TEST_SVC_CLIENT_NAME}/sublayer/tmp/tmp/tmp/test-subscription-strings-client-3', ), ( 'mqtt5.0', 1883, - 'org/fac/sys/sub/svc/test-subscription-strings-svc-4', + f'{TEST_SVC_SYSTEM_NAME}/fac/sys/sub/svc/test-subscription-strings-svc-4', '*/*/*/*/*/test-subscription-strings-svc-4', - 'tmp/tmp/tmp/tmp/tmp/test-subscription-strings-client-4', + f'{TEST_SVC_CLIENT_NAME}/sublayer/tmp/tmp/tmp/test-subscription-strings-client-4', ), ( 'amqp0.9.1', 5672, - 'org/fac/test-subscription-strings-svc-5/sys/sub/svc', + f'{TEST_SVC_SYSTEM_NAME}/sublayer1/test-subscription-strings-svc-5/sublayer2/sublayer3/sublayer4', '*/*/test-subscription-strings-svc-5/*/*/*', - 'tmp/tmp/tmp/tmp/tmp/test-subscription-strings-client-5', + f'{TEST_SVC_CLIENT_NAME}/sublayer/tmp/tmp/tmp/test-subscription-strings-client-5', ), ( 'mqtt5.0', 1883, - 'org/fac/test-subscription-strings-svc-6/sys/sub/svc', + f'{TEST_SVC_SYSTEM_NAME}/sublayer1/test-subscription-strings-svc-6/sublayer2/sublayer3/sublayer4', '*/*/test-subscription-strings-svc-6/*/*/*', - 'tmp/tmp/tmp/tmp/tmp/test-subscription-strings-client-6', + f'{TEST_SVC_CLIENT_NAME}/sublayer/tmp/tmp/tmp/test-subscription-strings-client-6', ), # multi-level wildcards, not needed in ecosystem for the moment (NOTE: multi-level wildcards can only be used at the end of the topic string according to MQTT spec, so we put it at the end here to ensure that the test is valid for both AMQP and MQTT) ( 'amqp0.9.1', 5672, - 'test-subscription-strings-svc-7/whatever/required/hierarchy/path', - 'test-subscription-strings-svc-7/*/required/#', - 'tmp/tmp/tmp/tmp/tmp/test-subscription-strings-client-7', + f'{TEST_SVC_SYSTEM_NAME}/l1/l2/l3/l4/test-subscription-strings-svc-7/whatever/required/hierarchy/path', + f'{TEST_SVC_SYSTEM_NAME}/l1/l2/l3/l4/test-subscription-strings-svc-7/*/required/#', + f'{TEST_SVC_CLIENT_NAME}/sublayer/tmp/tmp/tmp/test-subscription-strings-client-7', ), ( 'mqtt5.0', 1883, - 'test-subscription-strings-svc-8/whatever/required/hierarchy/path', - 'test-subscription-strings-svc-8/*/required/#', - 'tmp/tmp/tmp/tmp/tmp/test-subscription-strings-client-8', + f'{TEST_SVC_SYSTEM_NAME}/test-subscription-strings-svc-8/whatever/required/hierarchy/path', + f'{TEST_SVC_SYSTEM_NAME}/test-subscription-strings-svc-8/*/required/#', + f'{TEST_SVC_CLIENT_NAME}/sublayer/tmp/tmp/tmp/test-subscription-strings-client-8', ), ], ) @@ -91,8 +97,8 @@ def test_subscription_strings( client_sub_channel: str, ) -> None: """Test that wildcard subscriptions are properly matched and receive messages.""" - service_config = get_config(protocol, port) - client_config = get_config(protocol, port) + service_config = get_config(protocol, port, TEST_SVC_SYSTEM_NAME) + client_config = get_config(protocol, port, TEST_SVC_CLIENT_NAME) mock_service = MockService(service_config, [svc_sub_channel]) mock_client = MockClient(client_config, client_sub_channel) diff --git a/tests/unit/test_config.py b/tests/unit/test_config.py index edfedeb..8a3830b 100644 --- a/tests/unit/test_config.py +++ b/tests/unit/test_config.py @@ -16,10 +16,11 @@ def test_missing_control_plane_config(): with pytest.raises(ValidationError) as ex: TypeAdapter(ControlPlaneConfig).validate_python({}) errors = [{'type': e['type'], 'loc': e['loc']} for e in ex.value.errors()] - assert len(errors) == 3 + assert len(errors) == 4 assert {'type': 'missing', 'loc': ('username',)} in errors assert {'type': 'missing', 'loc': ('password',)} in errors assert {'type': 'missing', 'loc': ('protocol',)} in errors + assert {'type': 'missing', 'loc': ('system_name',)} in errors def test_invalid_control_plane_config(): @@ -31,10 +32,12 @@ def test_invalid_control_plane_config(): password='', port=0, protocol='mqtt', # type: ignore[arg-type] + system_name='', ).__dict__ ) errors = [{'type': e['type'], 'loc': e['loc']} for e in ex.value.errors()] - assert len(errors) == 5 + assert len(errors) == 6 + assert {'type': 'string_pattern_mismatch', 'loc': ('system_name',)} in errors assert {'type': 'string_too_short', 'loc': ('username',)} in errors assert {'type': 'string_too_short', 'loc': ('password',)} in errors assert {'type': 'string_too_short', 'loc': ('host',)} in errors @@ -73,9 +76,8 @@ def test_missing_intersect_config(): with pytest.raises(ValidationError) as ex: TypeAdapter(IntersectConfig).validate_python({}) errors = [{'type': e['type'], 'loc': e['loc']} for e in ex.value.errors()] - assert len(errors) == 3 + assert len(errors) == 2 assert {'type': 'missing', 'loc': ('broker',)} in errors - assert {'type': 'missing', 'loc': ('system_name',)} in errors assert {'type': 'missing', 'loc': ('service_name',)} in errors @@ -83,12 +85,10 @@ def test_invalid_intersect_config(): with pytest.raises(ValidationError) as ex: TypeAdapter(IntersectConfig).validate_python( IntersectConfig( - system_name='I AM INVALID', service_name=7, ).__dict__ ) errors = [{'type': e['type'], 'loc': e['loc']} for e in ex.value.errors()] - assert len(errors) == 3 - assert {'type': 'string_pattern_mismatch', 'loc': ('system_name',)} in errors + assert len(errors) == 2 assert {'type': 'string_type', 'loc': ('service_name',)} in errors assert {'type': 'missing', 'loc': ('broker',)} in errors From 6686d744af9bc38686a0fb5baf53270cd0308080 Mon Sep 17 00:00:00 2001 From: Lance-Drane Date: Thu, 10 Sep 2026 14:03:39 -0400 Subject: [PATCH 05/12] handle graceful shutdowns in a better fashion Signed-off-by: Lance-Drane --- .../control_plane/brokers/amqp_client.py | 22 ++++++++++++++----- .../control_plane/brokers/broker_client.py | 9 ++++++++ .../control_plane/brokers/mqtt_client.py | 10 +++++++-- 3 files changed, 34 insertions(+), 7 deletions(-) diff --git a/src/intersect_sdk_common/control_plane/brokers/amqp_client.py b/src/intersect_sdk_common/control_plane/brokers/amqp_client.py index 02a3b42..a4ce65e 100644 --- a/src/intersect_sdk_common/control_plane/brokers/amqp_client.py +++ b/src/intersect_sdk_common/control_plane/brokers/amqp_client.py @@ -20,7 +20,7 @@ from ...logger import logger from ...utils.multi_flag_thread_event import MultiFlagThreadEvent -from .broker_client import BrokerClient +from .broker_client import GRACEFUL_SHUTDOWN_TIMEOUT_SECONDS, BrokerClient if TYPE_CHECKING: from pika.channel import Channel @@ -166,8 +166,12 @@ def disconnect(self) -> None: self._connection.close() if self._thread: - # If gracefully shutting down, we should finish up the current job. - self._thread.join(5 if self.considered_unrecoverable() else None) + # If gracefully shutting down, give the in-flight message being handled (if any) a + # bounded amount of time to finish/publish before we give up on it; if the broker is + # unrecoverable there's no point waiting at all. + self._thread.join( + 0 if self.considered_unrecoverable() else GRACEFUL_SHUTDOWN_TIMEOUT_SECONDS + ) self._thread = None def is_connected(self) -> bool: @@ -284,8 +288,9 @@ def _cancel_consumer_tag_cb( logger.info('Unsubscribed from %s', topic) try: thread = self._consumer_tags_to_threads[consumer_tag] - # kill thread immediately if not recoverable, wait to send last message if we are shutting down gracefully - thread.join(0 if self.considered_unrecoverable() else None) + # kill thread immediately if not recoverable, otherwise give it a bounded amount of + # time to finish handling/publishing its current message before we give up on it + thread.join(0 if self.considered_unrecoverable() else GRACEFUL_SHUTDOWN_TIMEOUT_SECONDS) del self._consumer_tags_to_threads[consumer_tag] logger.debug('Consumer cancelled') except KeyError: @@ -334,6 +339,10 @@ def _on_connection_open_error( This function usually implies a misconfiguration in the application config. """ + if self._should_disconnect: + logger.info('Disconnect requested, giving up AMQP reconnection attempt') + connection.ioloop.stop() + return self._connection_retries += 1 logger.error( f'On connect error received (probable broker config error), have tried {self._connection_retries} times' @@ -640,6 +649,9 @@ def _consume_message( basic_deliver.delivery_tag, persist, ), + # daemon so that a message which is still being handled past GRACEFUL_SHUTDOWN_TIMEOUT_SECONDS + # cannot block the application from exiting once we've given up waiting on it + daemon=True, ) self._consumer_tags_to_threads[consumer_tag_info.consumer_tag] = thrd thrd.start() diff --git a/src/intersect_sdk_common/control_plane/brokers/broker_client.py b/src/intersect_sdk_common/control_plane/brokers/broker_client.py index 2b7f511..369b1a1 100644 --- a/src/intersect_sdk_common/control_plane/brokers/broker_client.py +++ b/src/intersect_sdk_common/control_plane/brokers/broker_client.py @@ -7,6 +7,15 @@ from ...config import ControlPlaneConfig +GRACEFUL_SHUTDOWN_TIMEOUT_SECONDS = 30.0 +"""Maximum amount of time (in seconds) a broker client will wait, upon receiving a disconnect +request (i.e. from an OS signal such as SIGTERM), for the in-flight message it is currently +handling to finish (so that a reply/publish can still go out). After this window elapses, +the client gives up waiting and proceeds with shutdown regardless of whether the in-flight +message finished. Broker clients should NOT attempt to consume/pull any additional messages +once a disconnect has been requested, only finish the one already in progress. +""" + class BrokerClient(Protocol): """Abstract definition of a Broker Client. diff --git a/src/intersect_sdk_common/control_plane/brokers/mqtt_client.py b/src/intersect_sdk_common/control_plane/brokers/mqtt_client.py index a046515..5555a18 100644 --- a/src/intersect_sdk_common/control_plane/brokers/mqtt_client.py +++ b/src/intersect_sdk_common/control_plane/brokers/mqtt_client.py @@ -19,7 +19,7 @@ from paho.mqtt.properties import Properties from ...logger import logger -from .broker_client import BrokerClient +from .broker_client import GRACEFUL_SHUTDOWN_TIMEOUT_SECONDS, BrokerClient if TYPE_CHECKING: from paho.mqtt.client import DisconnectFlags @@ -123,7 +123,13 @@ def disconnect(self) -> None: self._should_disconnect = True if self._connection: self._connection.disconnect() - self._connection.loop_stop() + # loop_stop() blocks until the network thread exits, with no timeout of its own; if a + # message handler is slow to finish publishing its reply, run the wait on a separate + # (daemon) thread and only wait up to GRACEFUL_SHUTDOWN_TIMEOUT_SECONDS for it here, so + # a stuck/slow handler cannot block shutdown forever. + stopper = threading.Thread(target=self._connection.loop_stop, daemon=True) + stopper.start() + stopper.join(GRACEFUL_SHUTDOWN_TIMEOUT_SECONDS) def is_connected(self) -> bool: """Check if there is an active connection to the broker. From b15be678fe73c2b771e3eb7dc091331b09c20dfc Mon Sep 17 00:00:00 2001 From: Lance-Drane Date: Wed, 16 Sep 2026 22:36:55 -0400 Subject: [PATCH 06/12] update CHANGELOG.md Signed-off-by: Lance-Drane --- CHANGELOG.md | 18 ++++++++++++++++++ 1 file changed, 18 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index d8e1e7b..8dc4d84 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,24 @@ We follow [Common Changelog](https://common-changelog.org/) formatting for this document. +## Unreleased + +### Changed + +- Replaced `has_error` field in `UserspaceMessage` with `message_state`. Instead of a boolean, this is an enumerated value. This enumerated value can capture ongoing processing in addition to error notifications. ([commit](https://github.com/INTERSECT-SDK/intersect-python-common/commit/8c144b2186228cbcef44edf256a3a81bd101d9f1) (Lance Drane)) + +### Added + +- Added `IntersectConfig` as representative of the configuration retrieved from the Registry Service at runtime ([commit 1](https://github.com/INTERSECT-SDK/intersect-python-common/commit/d4167449cca8fa05838379bc875c55876552d229) [commit 2](https://github.com/INTERSECT-SDK/intersect-python-common/commit/6e8053ee9555793b84bb2f4e0558c5a7df0ca0e1) (Lance Drane)) + +### Removed + +- HierarchyConfig has been removed entirely. We no longer use organization, facility, or subsystem anywhere across the ecosystem. `System` is preserved as a top-level namespace to be configured by the ecosystem, and `Service` is the namespace the user/client specifies. ([commit](https://github.com/INTERSECT-SDK/intersect-python-common/commit/d4167449cca8fa05838379bc875c55876552d229) (Lance Drane)) + +### Fixed + +- Force ControlPlaneManager `disconnect()` command to stop hanging ([commit](https://github.com/INTERSECT-SDK/intersect-python-common/commit/6686d744af9bc38686a0fb5baf53270cd0308080) (Lance Drane)) + ## [0.9.6] - 2026-06-02 ### Added From c1af2f52e0b18a6c5fe7757f96c6fe3e9395d375 Mon Sep 17 00:00:00 2001 From: Lance-Drane Date: Wed, 16 Sep 2026 22:40:02 -0400 Subject: [PATCH 07/12] 0.10.0a1 release Signed-off-by: Lance-Drane --- pyproject.toml | 130 +++++++++++++++++++++++-------------------------- uv.lock | 4 +- 2 files changed, 63 insertions(+), 71 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index a859200..297ac3a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -18,7 +18,7 @@ readme = "README.md" license = { text = "BSD-3-Clause" } requires-python = ">=3.10,<4.0" keywords = ["intersect"] -version = "0.9.6" +version = "0.10.0a1" classifiers = [ "Programming Language :: Python :: 3", "Programming Language :: Python :: 3.10", @@ -26,12 +26,7 @@ classifiers = [ "Programming Language :: Python :: 3.12", "Programming Language :: Python :: 3.13", ] -dependencies = [ - "pydantic>=2.7.0", - "paho-mqtt>=2.1.0,<3.0.0", - "pika>=1.3.2,<2.0.0", - "minio>=7.2.3", -] +dependencies = ["pydantic>=2.7.0", "paho-mqtt>=2.1.0,<3.0.0", "pika>=1.3.2,<2.0.0", "minio>=7.2.3"] [project.urls] Homepage = "https://github.com/INTERSECT-SDK/intersect-python-common" @@ -66,88 +61,85 @@ mccabe = { max-complexity = 20 } pylint = { max-args = 10, max-branches = 20, max-returns = 15, max-statements = 75 } # pyflakes and the relevant pycodestyle rules are already configured extend-select = [ - 'C90', # mccabe complexity - 'I', # isort - 'N', # pep8-naming - 'D', # pydocstyle - 'UP', # pyupgrade - 'YTT', # flake8-2020 - 'ANN', # flake8-annotations + 'C90', # mccabe complexity + 'I', # isort + 'N', # pep8-naming + 'D', # pydocstyle + 'UP', # pyupgrade + 'YTT', # flake8-2020 + 'ANN', # flake8-annotations 'ASYNC', # flake8-async - 'S', # flake8-bandit - 'BLE', # flake8-blind-except - 'B', # flake8-bugbear - 'A', # flake8-builtins - 'COM', # flake8-commas - 'C4', # flake8-comprehensions - 'DTZ', # flake8-datetimez - 'T10', # flake8-debugger - 'EM', # flake8-error-message - 'FA', # flake8-future-annotations - 'ISC', # flake8-implicit-string-concat - 'ICN', # flake8-import-conventions - 'G', # flake8-logging-format - 'INP', # flake8-no-pep420 - 'PIE', # flake8-PIE - 'T20', # flake8-T20 - 'PYI', # flake8-pyi - 'PT', # flake8-pytest-style - 'Q', # flake8-quotes - 'RSE', # flake8-raise - 'RET', # flake8-return - 'SLF', # flake8-self - 'SLOT', # flake8-slots - 'SIM', # flake8-simplify - 'TC', # flake8-type-checking - 'ARG', # flake8-unused-arguments - 'PTH', # flake8-use-pathlib - 'PGH', # pygrep-hooks - 'PL', # pylint - 'TRY', # tryceratops - 'FLY', # flynt - 'RUF', # RUFF additional rules - 'INT', # flake8-gettext + 'S', # flake8-bandit + 'BLE', # flake8-blind-except + 'B', # flake8-bugbear + 'A', # flake8-builtins + 'COM', # flake8-commas + 'C4', # flake8-comprehensions + 'DTZ', # flake8-datetimez + 'T10', # flake8-debugger + 'EM', # flake8-error-message + 'FA', # flake8-future-annotations + 'ISC', # flake8-implicit-string-concat + 'ICN', # flake8-import-conventions + 'G', # flake8-logging-format + 'INP', # flake8-no-pep420 + 'PIE', # flake8-PIE + 'T20', # flake8-T20 + 'PYI', # flake8-pyi + 'PT', # flake8-pytest-style + 'Q', # flake8-quotes + 'RSE', # flake8-raise + 'RET', # flake8-return + 'SLF', # flake8-self + 'SLOT', # flake8-slots + 'SIM', # flake8-simplify + 'TC', # flake8-type-checking + 'ARG', # flake8-unused-arguments + 'PTH', # flake8-use-pathlib + 'PGH', # pygrep-hooks + 'PL', # pylint + 'TRY', # tryceratops + 'FLY', # flynt + 'RUF', # RUFF additional rules + 'INT', # flake8-gettext ] # If you're seeking to disable a rule, first consider whether the rule is overbearing, or if it should only be turned off for your usecase. ignore = [ - 'COM812', # formatter, handled by Ruff format - 'ISC001', # formatter, handled by Ruff format - 'SIM105', # "with contextlib.suppress():" is slower than try-except-pass - 'ANN401', # allow explicit "Any" typing, use with care + 'COM812', # formatter, handled by Ruff format + 'ISC001', # formatter, handled by Ruff format + 'SIM105', # "with contextlib.suppress():" is slower than try-except-pass + 'ANN401', # allow explicit "Any" typing, use with care 'PLR2004', # allow "magic numbers" ] [tool.ruff.lint.flake8-type-checking] runtime-evaluated-base-classes = ["pydantic.BaseModel"] -runtime-evaluated-decorators = [ - "pydantic.dataclasses.dataclass", - "pydantic.validate_call", -] +runtime-evaluated-decorators = ["pydantic.dataclasses.dataclass", "pydantic.validate_call"] [tool.ruff.lint.extend-per-file-ignores] '__init__.py' = [ - 'F401', # __init__.py commonly has unused imports + 'F401', # __init__.py commonly has unused imports 'TC004', # do lazy imports when importing from the base module ] 'docs/*' = [ - 'D', # the documentation folder does not need documentation + 'D', # the documentation folder does not need documentation 'INP001', # docs are not a namespace package ] 'tests/*' = [ - 'S101', # allow assert statements in tests - 'S106', # don't care about credentials in tests - 'S311', # don't care about cryptographic security in tests + 'S101', # allow assert statements in tests + 'S106', # don't care about credentials in tests + 'S311', # don't care about cryptographic security in tests 'SLF001', # allow private member access in tests - 'ANN', # tests in general don't need types, unless they are runtime types. - 'ARG', # allow unused parameters in tests - 'D', # ignore documentation in tests + 'ANN', # tests in general don't need types, unless they are runtime types. + 'ARG', # allow unused parameters in tests + 'D', # ignore documentation in tests 'RUF012', # permit "mutable" class attributes to not be annotated with typing.ClassVar (these shouldn't be mutated anyways...) ] # see https://mypy.readthedocs.io/en/stable/config_file.html for a complete reference [tool.mypy] strict = true -ignore_missing_imports = true # don't require typing for library stubs if they don't exist +ignore_missing_imports = true # don't require typing for library stubs if they don't exist disallow_untyped_decorators = false # this is needed for library decorator compatibility, i.e. "retrying" plugins = ["pydantic.mypy"] @@ -166,12 +158,12 @@ omit = [ '*__init__*', # __init__ files should just re-export other classes and functions ] exclude_also = [ - 'pragma: no-cover', # standard + 'pragma: no-cover', # standard 'if (typing\\.)?TYPE_CHECKING:', # type checking blocks are not executed in coverage, but we don't care - '@(abc\\.)?abstractmethod', # don't try to cover abstract methods - "class .*\\bProtocol\\):", # don't cover protocol classes (similar to abstract classes) - 'raise NotImplementedError', # it's not implemented so shouldn't be covered - 'except.* ImportError', # these are usually used to throw a "friendlier" error and are not really worth testing + '@(abc\\.)?abstractmethod', # don't try to cover abstract methods + "class .*\\bProtocol\\):", # don't cover protocol classes (similar to abstract classes) + 'raise NotImplementedError', # it's not implemented so shouldn't be covered + 'except.* ImportError', # these are usually used to throw a "friendlier" error and are not really worth testing ] diff --git a/uv.lock b/uv.lock index 70beeb8..fb92694 100644 --- a/uv.lock +++ b/uv.lock @@ -313,7 +313,7 @@ name = "exceptiongroup" version = "1.3.1" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "typing-extensions", marker = "python_full_version < '3.13'" }, + { name = "typing-extensions" }, ] sdist = { url = "https://files.pythonhosted.org/packages/50/79/66800aadf48771f6b62f7eb014e352e5d06856655206165d775e675a02c9/exceptiongroup-1.3.1.tar.gz", hash = "sha256:8b412432c6055b0b7d14c310000ae93352ed6754f70fa8f7c34141f91c4e3219", size = 30371, upload-time = "2025-11-21T23:01:54.787Z" } wheels = [ @@ -355,7 +355,7 @@ wheels = [ [[package]] name = "intersect-sdk-common" -version = "0.9.6" +version = "0.10.0a1" source = { editable = "." } dependencies = [ { name = "minio" }, From bb0c1b1124d916b383e3967f8a9d4dff231165f1 Mon Sep 17 00:00:00 2001 From: Lance-Drane Date: Wed, 16 Sep 2026 22:50:24 -0400 Subject: [PATCH 08/12] bump git-release version to v7 Signed-off-by: Lance-Drane --- .github/workflows/publish.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/publish.yml b/.github/workflows/publish.yml index c13f755..78eee29 100644 --- a/.github/workflows/publish.yml +++ b/.github/workflows/publish.yml @@ -27,7 +27,7 @@ jobs: - name: Publish package to PyPI run: uv publish - name: upload to github release - uses: docker://antonyurchenko/git-release:v5 + uses: docker://antonyurchenko/git-release:v7 env: GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} # assume any version with "a" (gets "ALPHA") or "b" (gets BETA) or "rc" (release candidates) will be a prerelease From 038725cf18a6b22ba6f24ccebec389f78e25b49f Mon Sep 17 00:00:00 2001 From: Lance-Drane Date: Thu, 24 Sep 2026 22:27:11 -0400 Subject: [PATCH 09/12] broker topics: consistently use '/' character everywhere to indicate a separator limit any separator conversions to protocol layer Signed-off-by: Lance-Drane --- src/intersect_sdk_common/constants.py | 4 ++-- .../control_plane/control_plane_manager.py | 2 +- src/intersect_sdk_common/control_plane/topic_handler.py | 2 +- tests/fixtures/mock_client.py | 4 ++-- tests/integration/test_control_plane.py | 2 +- tests/unit/test_event_message_headers.py | 2 +- tests/unit/test_lifecycle_message_headers.py | 2 +- tests/unit/test_userspace_message_headers.py | 4 ++-- 8 files changed, 11 insertions(+), 11 deletions(-) diff --git a/src/intersect_sdk_common/constants.py b/src/intersect_sdk_common/constants.py index 2a1c499..2f68717 100644 --- a/src/intersect_sdk_common/constants.py +++ b/src/intersect_sdk_common/constants.py @@ -1,11 +1,11 @@ """These are miscellaneous constants used in INTERSECT which SDK users may obtain value from knowing about.""" -SYSTEM_OF_SYSTEM_REGEX = r'^[a-z0-9][-a-z0-9.]*[-a-z0-9]$' +SYSTEM_OF_SYSTEM_REGEX = r'^[a-z0-9][-a-z0-9/]*[-a-z0-9]$' """ This is the regex used as a representation of a source/destination. This is only needed externally if you are building a client, services can ignore this. -NOTE: for future compatibility reasons, we are NOT specifying the number of "parts" (separated by a '.') in this regex. All that matters is that you don't start or end with a period, or start with a hyphen. +NOTE: for future compatibility reasons, we are NOT specifying the number of "parts" (separated by a '/') in this regex. All that matters is that you don't start or end with a separator, or start with a hyphen. """ # see the internal schema file for full validation details diff --git a/src/intersect_sdk_common/control_plane/control_plane_manager.py b/src/intersect_sdk_common/control_plane/control_plane_manager.py index 8bcb3a7..a9df421 100644 --- a/src/intersect_sdk_common/control_plane/control_plane_manager.py +++ b/src/intersect_sdk_common/control_plane/control_plane_manager.py @@ -16,7 +16,7 @@ - alphanumeric characters - hyphens - underscores -- `/` (topic separator; note that this is '.' on the broker) +- `/` (topic separator) - `*` (wildcard for exactly one word) - `#` (wildcard for any number of words, but can ONLY appear at the end of the topic string, see section 4.7.1.2 at https://docs.oasis-open.org/mqtt/mqtt/v3.1.1/os/mqtt-v3.1.1-os.html#_Toc398718107) diff --git a/src/intersect_sdk_common/control_plane/topic_handler.py b/src/intersect_sdk_common/control_plane/topic_handler.py index a727e22..7b1e11e 100644 --- a/src/intersect_sdk_common/control_plane/topic_handler.py +++ b/src/intersect_sdk_common/control_plane/topic_handler.py @@ -50,7 +50,7 @@ def __init__( # match any sequence of 0 or more words (make sure to allow for the separator) regex_builder.append('[a-zA-Z0-9/_-]*') elif char == '/': - # in-memory topics use '/' as the separator, it is the protocol handler's responsibility to convert topics to use '/' as the separator + # separator character regex_builder.append('/') else: # ordinary character diff --git a/tests/fixtures/mock_client.py b/tests/fixtures/mock_client.py index 74d530d..905ba63 100644 --- a/tests/fixtures/mock_client.py +++ b/tests/fixtures/mock_client.py @@ -31,8 +31,8 @@ def publish_message(self, channel: str, message: bytes) -> None: # in case you want to reuse this Client again, reset the event self.event.clear() headers = create_userspace_message_headers( - source=self.client_sub_channel.replace('/', '.'), - destination=channel.replace('/', '.'), + source=self.client_sub_channel, + destination=channel, operation_id='operation_not_used', data_handler=IntersectDataHandler.MESSAGE, campaign_id=self.campaign_id, diff --git a/tests/integration/test_control_plane.py b/tests/integration/test_control_plane.py index 3a8e26a..5aa01a9 100644 --- a/tests/integration/test_control_plane.py +++ b/tests/integration/test_control_plane.py @@ -119,7 +119,7 @@ def test_subscription_strings( received_message, content_type, headers = mock_client.get_received_value() assert received_message == b'god yzal eht revo spmuj xof nworb kciuq eht' assert content_type == 'text/plain' - assert headers['source'] == client_pub_channel.replace('/', '.') + assert headers['source'] == client_pub_channel finally: mock_service.disconnect() mock_client.disconnect() diff --git a/tests/unit/test_event_message_headers.py b/tests/unit/test_event_message_headers.py index 14d5eec..b287617 100644 --- a/tests/unit/test_event_message_headers.py +++ b/tests/unit/test_event_message_headers.py @@ -36,7 +36,7 @@ def test_valid_event_message_deserializes() -> None: def test_unusual_event_message_deserializes() -> None: raw_headers = { 'message_id': 'cc88a2c9-7e47-409f-82c5-ef49914ae140', - 'source': 'source.one', + 'source': 'source/one', 'sdk_version': '0.5.0', 'created_at': '2024', 'data_handler': 'MINIO', diff --git a/tests/unit/test_lifecycle_message_headers.py b/tests/unit/test_lifecycle_message_headers.py index a18266d..acecbc0 100644 --- a/tests/unit/test_lifecycle_message_headers.py +++ b/tests/unit/test_lifecycle_message_headers.py @@ -32,7 +32,7 @@ def test_valid_lifecycle_message_deserializes() -> None: def test_unusual_lifecycle_message_deserializes() -> None: raw_headers = { 'message_id': 'cc88a2c9-7e47-409f-82c5-ef49914ae140', - 'source': 'source.one', + 'source': 'source/one', 'sdk_version': '0.5.0', 'created_at': '2024', 'lifecycle_type': 'LCT_STARTUP', diff --git a/tests/unit/test_userspace_message_headers.py b/tests/unit/test_userspace_message_headers.py index 3149352..c2e8b28 100644 --- a/tests/unit/test_userspace_message_headers.py +++ b/tests/unit/test_userspace_message_headers.py @@ -42,8 +42,8 @@ def test_unusual_userspace_message_deserializes() -> None: 'campaign_id': 'dd88a2c9-7e47-409f-82c5-ef49914ae141', 'request_id': 'ee88a2c9-7e47-409f-82c5-ef49914ae142', 'operation_id': 'operation', - 'source': 'source.one', - 'destination': 'destination.two', + 'source': 'source/one', + 'destination': 'destination/two', 'sdk_version': '0.5.0', 'created_at': '2024', 'data_handler': 'MINIO', From bb215f447292fc8a0f8176c43685b0cdd88dc09b Mon Sep 17 00:00:00 2001 From: Lance-Drane Date: Thu, 24 Sep 2026 22:29:22 -0400 Subject: [PATCH 10/12] 0.10.0a2 release Signed-off-by: Lance-Drane --- CHANGELOG.md | 1 + pyproject.toml | 2 +- uv.lock | 2 +- 3 files changed, 3 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 8dc4d84..7e20593 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,7 @@ We follow [Common Changelog](https://common-changelog.org/) formatting for this ### Changed - Replaced `has_error` field in `UserspaceMessage` with `message_state`. Instead of a boolean, this is an enumerated value. This enumerated value can capture ongoing processing in addition to error notifications. ([commit](https://github.com/INTERSECT-SDK/intersect-python-common/commit/8c144b2186228cbcef44edf256a3a81bd101d9f1) (Lance Drane)) +- Consistently use '/' character everywhere to indicate a separator character, only change this in the protocol layer as needed ([commit](https://github.com/INTERSECT-SDK/intersect-python-common/commit/038725cf18a6b22ba6f24ccebec389f78e25b49f) (Lance Drane)) ### Added diff --git a/pyproject.toml b/pyproject.toml index 297ac3a..427bccd 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -18,7 +18,7 @@ readme = "README.md" license = { text = "BSD-3-Clause" } requires-python = ">=3.10,<4.0" keywords = ["intersect"] -version = "0.10.0a1" +version = "0.10.0a2" classifiers = [ "Programming Language :: Python :: 3", "Programming Language :: Python :: 3.10", diff --git a/uv.lock b/uv.lock index fb92694..c57c935 100644 --- a/uv.lock +++ b/uv.lock @@ -355,7 +355,7 @@ wheels = [ [[package]] name = "intersect-sdk-common" -version = "0.10.0a1" +version = "0.10.0a2" source = { editable = "." } dependencies = [ { name = "minio" }, From 0554cdb6c31bb80cd0e6fc762944e15b7df6fbc6 Mon Sep 17 00:00:00 2001 From: Lance-Drane Date: Tue, 29 Sep 2026 17:05:43 -0400 Subject: [PATCH 11/12] have broker clients use a user-defined callback to obtain credentials Signed-off-by: Lance-Drane --- .../control_plane/brokers/amqp_client.py | 31 ++++++++++ .../control_plane/brokers/broker_client.py | 9 +++ .../control_plane/brokers/mqtt_client.py | 28 +++++++++ .../control_plane/control_plane_manager.py | 35 ++++++++++++ tests/unit/test_broker_credentials.py | 57 +++++++++++++++++++ 5 files changed, 160 insertions(+) create mode 100644 tests/unit/test_broker_credentials.py diff --git a/src/intersect_sdk_common/control_plane/brokers/amqp_client.py b/src/intersect_sdk_common/control_plane/brokers/amqp_client.py index a4ce65e..8a3ccde 100644 --- a/src/intersect_sdk_common/control_plane/brokers/amqp_client.py +++ b/src/intersect_sdk_common/control_plane/brokers/amqp_client.py @@ -74,6 +74,19 @@ def _amqp_2_hierarchy(amqp_routing_key: str) -> str: return amqp_routing_key.replace('.', '/') +def _is_auth_error(err: BaseException) -> bool: + """Check if a connection error indicates that our credentials were rejected.""" + return isinstance( + err, + pika.exceptions.AuthenticationError + | pika.exceptions.ProbableAuthenticationError + | pika.exceptions.ProbableAccessDeniedError, + ) or ( + isinstance(err, pika.exceptions.ConnectionClosedByBroker) + and err.reply_code == 403 # ACCESS_REFUSED + ) + + class AMQPClient(BrokerClient): """Client for performing broker actions backed by a AMQP broker. @@ -119,6 +132,7 @@ def __init__( self._should_disconnect = False self._connection_retries = 0 self._unrecoverable = False + self._credentials_invalid = False # tracking both channels is the best way to handle continuations self._channel_flags = MultiFlagThreadEvent(2) @@ -193,6 +207,14 @@ def considered_unrecoverable(self) -> bool: """ return self._unrecoverable + def credentials_invalid(self) -> bool: + """Checks if the most recent connection attempt was rejected due to credentials/authorization. + + Returns: + A boolean. True if credentials were rejected, False otherwise. + """ + return self._credentials_invalid + def publish( self, topic: str, payload: bytes, content_type: str, headers: dict[str, str], persist: bool ) -> None: @@ -348,6 +370,7 @@ def _on_connection_open_error( f'On connect error received (probable broker config error), have tried {self._connection_retries} times' ) logger.error(err) + self._credentials_invalid = _is_auth_error(err) if self._connection_retries >= _AMQP_MAX_RETRIES: # This will allow us to break out of the while loop # where we establish the connection, as ioloop.stop @@ -358,12 +381,20 @@ def _on_connection_open_error( self._channel_flags.set_all() connection.ioloop.stop() else: + # only try to refresh credentials if we're still going to reconnect + if self._credentials_invalid and ( + new_config := self._control_plane_manager.fetch_refreshed_config(self._system_name) + ): + # the next SelectConnection in _init_connection will use the new connection params + logger.info('Obtained refreshed AMQP broker config') + self._handle_config(new_config) logger.error('Reopening in 5 seconds') connection.ioloop.call_later(5, connection.ioloop.stop) def _on_connection_open(self, connection: pika.SelectConnection) -> None: logger.info('AMQP connection open') self._connection_retries = 0 + self._credentials_invalid = False self._topics_to_consumer_tags.clear() connection.channel(on_open_callback=self._on_input_channel_open) connection.channel(on_open_callback=self._on_output_channel_open) diff --git a/src/intersect_sdk_common/control_plane/brokers/broker_client.py b/src/intersect_sdk_common/control_plane/brokers/broker_client.py index 369b1a1..06ac3a5 100644 --- a/src/intersect_sdk_common/control_plane/brokers/broker_client.py +++ b/src/intersect_sdk_common/control_plane/brokers/broker_client.py @@ -48,6 +48,15 @@ def considered_unrecoverable(self) -> bool: """ ... + def credentials_invalid(self) -> bool: + """Checks if the most recent connection attempt was rejected due to credentials/authorization. + + Returns: + A boolean. True if credentials were rejected, False otherwise. + Must always be False if a disconnect was requested or the broker is considered unrecoverable, as shutdown takes priority. + """ + ... + def publish( self, topic: str, payload: bytes, content_type: str, headers: dict[str, str], persist: bool ) -> None: diff --git a/src/intersect_sdk_common/control_plane/brokers/mqtt_client.py b/src/intersect_sdk_common/control_plane/brokers/mqtt_client.py index 5555a18..cfd88cf 100644 --- a/src/intersect_sdk_common/control_plane/brokers/mqtt_client.py +++ b/src/intersect_sdk_common/control_plane/brokers/mqtt_client.py @@ -31,6 +31,9 @@ _MQTT_MAX_RETRIES = 10 +_MQTT_AUTH_REASON_CODES = (134, 135) +"""'Bad user name or password' and 'Not authorized'. paho also maps MQTTv3 CONNACK codes 4 and 5 to these.""" + # TODO we should be handling hierarchy parts as a list of strings until they get to the client # this will be a breaking change, so only add it when ready to break @@ -80,6 +83,7 @@ def __init__( self._connected = False self._should_disconnect = False self._unrecoverable = False + self._credentials_invalid = False self._connection_retries = 0 self._connected_flag = threading.Event() @@ -147,6 +151,14 @@ def considered_unrecoverable(self) -> bool: """ return self._unrecoverable + def credentials_invalid(self) -> bool: + """Checks if the most recent connection attempt was rejected due to credentials/authorization. + + Returns: + A boolean. True if credentials were rejected, False otherwise. + """ + return self._credentials_invalid + def publish( self, topic: str, payload: bytes, content_type: str, headers: dict[str, str], persist: bool ) -> None: @@ -306,6 +318,7 @@ def _handle_connect( ) self._connected = True self._connection_retries = 0 + self._credentials_invalid = False self._should_disconnect = False # mimic "automatic QoS downgrade" of MQTTv3 for MQTTv5 @@ -327,7 +340,22 @@ def _handle_connect( ) logger.error('Connection error userdata: %s', userdata) logger.error('Connection error flags: %s', flags) + self._credentials_invalid = reason_code.value in _MQTT_AUTH_REASON_CODES if self._connection_retries >= _MQTT_MAX_RETRIES: logger.error('Giving up MQTT reconnection attempt') self._connected_flag.set() self._unrecoverable = True + # only try to refresh credentials if we're still going to reconnect + elif ( + self._credentials_invalid + and not self._should_disconnect + and ( + new_config := self._control_plane_manager.fetch_refreshed_config( + self._system_name + ) + ) + ): + # paho's automatic reconnect will use the new username/password + # host/port changes only apply on the next explicit connect(), since paho caches them from connect() + logger.info('Obtained refreshed MQTT broker config') + self._handle_config(new_config) diff --git a/src/intersect_sdk_common/control_plane/control_plane_manager.py b/src/intersect_sdk_common/control_plane/control_plane_manager.py index a9df421..b046287 100644 --- a/src/intersect_sdk_common/control_plane/control_plane_manager.py +++ b/src/intersect_sdk_common/control_plane/control_plane_manager.py @@ -2,6 +2,7 @@ import itertools import re +from collections.abc import Callable from ..config import ControlPlaneConfig from ..exceptions import IntersectSetupError @@ -53,13 +54,21 @@ class ControlPlaneManager: def __init__( self, control_configs: list[ControlPlaneConfig], + get_refreshed_config: Callable[[str], ControlPlaneConfig | None] | None = None, ) -> None: """Basic constructor. Some interaction with message brokers can change based on whether or not a Service or a Client is calling it. queue_name_generator should be a hardcoded value for Core Services, the SDK should provide its own function to generate queue names. + + Params: + control_configs: configurations for each broker + get_refreshed_config: optional callback, called with a broker's system_name when that broker rejects our credentials. + It should return a new ControlPlaneConfig (i.e. freshly obtained from the registry service), or None to decline. + NOTE: this runs on the broker's IO thread while that broker is disconnected; it may block, but should not take forever. """ + self._get_refreshed_config = get_refreshed_config self._control_providers = [ _create_control_provider(config, self) for config in control_configs ] @@ -77,6 +86,22 @@ def refresh_broker_config(self, control_config: ControlPlaneConfig) -> None: ): provider.refresh_config(control_config) + def fetch_refreshed_config(self, system_name: str) -> ControlPlaneConfig | None: + """Obtain new credentials for the broker with system_name, if a callback was provided. + + This function is safe to call from the broker clients, as it does not mutate state. + + Returns: + the new ControlPlaneConfig, or None if no callback was provided, the callback declined, or the callback raised + """ + if self._get_refreshed_config is None: + return None + try: + return self._get_refreshed_config(system_name) + except Exception: # noqa: BLE001 (user callback, must not kill the broker IO thread) + logger.exception('Unable to obtain refreshed broker config for system %s', system_name) + return None + def add_subscription_channel( self, channel: str, @@ -272,3 +297,13 @@ def considered_unrecoverable(self) -> bool: control_provider.considered_unrecoverable() for control_provider in self._control_providers ) + + def credentials_invalid(self) -> bool: + """Check if any broker most recently rejected our credentials. + + Returns: + - True if any broker rejected our credentials on its latest connection attempt, False otherwise + """ + return any( + control_provider.credentials_invalid() for control_provider in self._control_providers + ) diff --git a/tests/unit/test_broker_credentials.py b/tests/unit/test_broker_credentials.py new file mode 100644 index 0000000..5797999 --- /dev/null +++ b/tests/unit/test_broker_credentials.py @@ -0,0 +1,57 @@ +from unittest.mock import MagicMock + +import pika.exceptions +from paho.mqtt.packettypes import PacketTypes +from paho.mqtt.reasoncodes import ReasonCode + +from intersect_sdk_common.config import ControlPlaneConfig +from intersect_sdk_common.control_plane.control_plane_manager import ControlPlaneManager + + +def _config(protocol: str, password: str) -> ControlPlaneConfig: + return ControlPlaneConfig( + protocol=protocol, # type: ignore[arg-type] + system_name='system', + username='user', + password=password, + ) + + +def test_amqp_refreshes_credentials_on_auth_error(): + refresher = MagicMock(return_value=_config('amqp0.9.1', 'new')) + manager = ControlPlaneManager([_config('amqp0.9.1', 'old')], refresher) + client = manager._control_providers[0] + + client._on_connection_open_error(MagicMock(), pika.exceptions.ProbableAuthenticationError()) + assert client.credentials_invalid() + assert manager.credentials_invalid() + refresher.assert_called_once_with('system') + assert client._connection_params.credentials.password == 'new' # noqa: S105 + + refresher.reset_mock() + client._on_connection_open_error(MagicMock(), pika.exceptions.AMQPConnectionError()) + assert not client.credentials_invalid() + refresher.assert_not_called() + + +def test_mqtt_refreshes_credentials_on_auth_error(): + refresher = MagicMock(return_value=_config('mqtt5.0', 'new')) + manager = ControlPlaneManager([_config('mqtt5.0', 'old')], refresher) + client = manager._control_providers[0] + + client._handle_connect( + client._connection, None, {}, ReasonCode(PacketTypes.CONNACK, identifier=134), None + ) + assert client.credentials_invalid() + refresher.assert_called_once_with('system') + assert client._connection._password == b'new' + + +def test_refresher_exception_keeps_old_config(): + manager = ControlPlaneManager( + [_config('amqp0.9.1', 'old')], MagicMock(side_effect=RuntimeError('registry down')) + ) + client = manager._control_providers[0] + client._on_connection_open_error(MagicMock(), pika.exceptions.ProbableAuthenticationError()) + assert client.credentials_invalid() + assert client._connection_params.credentials.password == 'old' # noqa: S105 From cb8cbe5c8c1796f9f4f67d51cbb3bbcecf19e240 Mon Sep 17 00:00:00 2001 From: Lance-Drane Date: Wed, 30 Sep 2026 11:32:45 -0400 Subject: [PATCH 12/12] 0.10.0a3 release Signed-off-by: Lance-Drane --- CHANGELOG.md | 1 + pyproject.toml | 2 +- uv.lock | 2 +- 3 files changed, 3 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 7e20593..1eecc9d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,6 +12,7 @@ We follow [Common Changelog](https://common-changelog.org/) formatting for this ### Added - Added `IntersectConfig` as representative of the configuration retrieved from the Registry Service at runtime ([commit 1](https://github.com/INTERSECT-SDK/intersect-python-common/commit/d4167449cca8fa05838379bc875c55876552d229) [commit 2](https://github.com/INTERSECT-SDK/intersect-python-common/commit/6e8053ee9555793b84bb2f4e0558c5a7df0ca0e1) (Lance Drane)) +- Added mechanism for registry service or other utilities to provide credentials ([commit](https://github.com/INTERSECT-SDK/intersect-python-common/commit/0554cdb6c31bb80cd0e6fc762944e15b7df6fbc6) (Lance Drane)) ### Removed diff --git a/pyproject.toml b/pyproject.toml index 427bccd..0b4bee5 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -18,7 +18,7 @@ readme = "README.md" license = { text = "BSD-3-Clause" } requires-python = ">=3.10,<4.0" keywords = ["intersect"] -version = "0.10.0a2" +version = "0.10.0a3" classifiers = [ "Programming Language :: Python :: 3", "Programming Language :: Python :: 3.10", diff --git a/uv.lock b/uv.lock index c57c935..411a0c4 100644 --- a/uv.lock +++ b/uv.lock @@ -355,7 +355,7 @@ wheels = [ [[package]] name = "intersect-sdk-common" -version = "0.10.0a2" +version = "0.10.0a3" source = { editable = "." } dependencies = [ { name = "minio" },