refactor: remove correlation dead code

docs/explanations/events.md

@@ -34,18 +34,9 @@ every time the engine:
 In the latter case, information about the error is also included.
 
 
-## Correlation ID
+## Trace Context
 
-
-When controlling plans programmatically, it can be useful to verify that event model documents really are related to
-the plan you just asked the worker to run. The worker will therefore bundle a correlation ID into the headers of
-messages containing documents.
-
-see also [`Microsoft Playbook on Correlation IDs`](https://microsoft.github.io/code-with-engineering-playbook/observability/correlation-id/)
-
-ActiveMQ will give this header a different name depending on the protocol you use.
-
-| Protocol | Header name      |
-|----------|------------------|
-| JMS      | jms_correlationID|
-| STOMP    | correlation-id   |
+When controlling plans programmatically, it can be useful to verify that events really are related to
+the plan you just asked the worker to run. Every event carries the task ID in its body, and the worker also
+propagates the [OpenTelemetry](https://opentelemetry.io/) trace context in the message headers
+(the W3C `traceparent` header), so events can be tied to the request that started the task.

docs/reference/asyncapi.yaml

@@ -31,7 +31,7 @@ channels:
           - $ref: "#/components/messages/taggedResourceStream"
           - $ref: "#/components/messages/taggedDatumStream"
 components:
-  messages: # Defined as Components to allow referencing from other APIs. # TODO: Define protocol specific correlationId bindings
+  messages: # Defined as Components to allow referencing from other APIs.
     taggedStartDocument:
       messageId: runStart
       summary: Indicates the start of a Bluesky run based on a previously specified Plan; describes the initial position/metadata of a scan and its involved devices
@@ -183,9 +183,6 @@ components:
         replyDestination:
           description: Name of the temporary queue, specified by the caller where they will listen for response messages to requests submitted on the worker.X channels
           type: string
-        correlationId:
-          description: Unique identifier of an exchange supplied by its originator
-          type: string
     workerState:
       description: The state of the Worker.
       enum:

docs/reference/openapi.yaml

@@ -388,9 +388,6 @@ components:
           - $ref: '#/components/schemas/TaskError'
           - type: 'null'
           title: Outcome
-        request_id:
-          title: Request Id
-          type: string
         task:
           $ref: '#/components/schemas/Task'
         task_id:
@@ -459,7 +456,7 @@ info:
     name: Apache 2.0
     url: https://www.apache.org/licenses/LICENSE-2.0.html
   title: BlueAPI Control
-  version: 1.6.0
+  version: 1.6.1
 openapi: 3.1.0
 paths:
   /api/v1/devices:

src/blueapi/client/client.py

@@ -407,11 +407,7 @@ def inner_on_event(event: AnyEvent, ctx: MessageContext) -> None:
                         log.error(
                             f"Callback ({cb}) failed for event: {event}", exc_info=e
                         )
-                if (
-                    isinstance(event, WorkerEvent)
-                    and (event.is_complete())
-                    and (ctx.correlation_id == task_id)
-                ):
+                if isinstance(event, WorkerEvent) and event.is_complete():
                     if event.task_status is None:
                         complete.set_exception(
                             BlueskyRemoteControlError(

src/blueapi/config.py

@@ -335,7 +335,7 @@ class ApplicationConfig(BlueapiBaseModel):
     """
 
     #: API version to publish in OpenAPI schema
-    REST_API_VERSION: ClassVar[str] = "1.6.0"
+    REST_API_VERSION: ClassVar[str] = "1.6.1"
 
     LICENSE_INFO: ClassVar[dict[str, str]] = {
         "name": "Apache 2.0",

src/blueapi/core/event.py

@@ -19,12 +19,12 @@ class EventStream(ABC, Generic[E, S]):
     """
 
     @abstractmethod
-    def subscribe(self, __callback: Callable[[E, str | None], None]) -> S:
+    def subscribe(self, __callback: Callable[[E], None]) -> S:
         """
         Subscribe to new events with a callback
 
         Args:
-            __callback: What to do with each event, optionally takes a correlation id
+            __callback: What to do with each event
 
         Returns:
             S: A unique token representing the subscription
@@ -51,14 +51,14 @@ class EventPublisher(EventStream[E, int]):
     Simple Observable that can be fed values to publish
     """
 
-    _subscriptions: dict[int, Callable[[E, str | None], None]]
+    _subscriptions: dict[int, Callable[[E], None]]
     _count: itertools.count
 
     def __init__(self) -> None:
         self._subscriptions = {}
         self._count = itertools.count()
 
-    def subscribe(self, callback: Callable[[E, str | None], None]) -> int:
+    def subscribe(self, callback: Callable[[E], None]) -> int:
         sub_id = next(self._count)
         self._subscriptions[sub_id] = callback
         return sub_id
@@ -69,19 +69,17 @@ def unsubscribe(self, subscription: int) -> None:
     def unsubscribe_all(self) -> None:
         self._subscriptions = {}
 
-    def publish(self, event: E, correlation_id: str | None = None) -> None:
+    def publish(self, event: E) -> None:
         """
         Publish a new event to all subscribers
 
         Args:
             event: The event to publish
-            correlation_id: An optional ID that may be used to correlate this
-                event with other events
         """
         errs = []
         for callback in list(self._subscriptions.values()):
             try:
-                callback(event, correlation_id)
+                callback(event)
             except Exception as e:
                 errs.append(e)
         if errs:

src/blueapi/service/interface.py

@@ -189,11 +189,9 @@ def _publish_event_streams(
 
 
 def _publish_event_stream(stream: EventStream, destination: DestinationBase) -> None:
-    def forward_message(event: Any, correlation_id: str | None) -> None:
+    def forward_message(event: Any) -> None:
         if (stomp_client_ref := stomp_client()) is not None:
-            stomp_client_ref.send(
-                destination, event, None, correlation_id=correlation_id
-            )
+            stomp_client_ref.send(destination, event, None)
 
     stream.subscribe(forward_message)
 
@@ -288,9 +286,7 @@ def begin_task(
         )
         subscribers.append((active_context.run_engine, tiled_writer_token))
 
-        def remove_callback_when_task_finished(
-            event: WorkerEvent, correlation_id: str | None
-        ) -> None:
+        def remove_callback_when_task_finished(event: WorkerEvent) -> None:
             if (
                 event.task_status
                 and event.task_status.task_id == task.task_id
@@ -373,11 +369,7 @@ class SubHandles:
 def pipe_events(sender: Connection) -> SubHandles:
     tw = worker()
 
-    def handler(
-        worker_event: WorkerEvent | DataEvent | ProgressEvent,
-        _cor_id: str | None,
-    ) -> None:
-
+    def handler(worker_event: WorkerEvent | DataEvent | ProgressEvent) -> None:
         try:
             sender.send(worker_event)
         except BrokenPipeError:

src/blueapi/worker/task_worker.py

@@ -16,11 +16,9 @@
     setup_tracing,
     start_as_current_span,
 )
-from opentelemetry.baggage import get_baggage
 from opentelemetry.context import Context, get_current
 from opentelemetry.trace import SpanKind
 from pydantic import Field
-from pydantic.json_schema import SkipJsonSchema
 
 from blueapi.core import (
     OTLP_EXPORT_ENABLED,
@@ -67,7 +65,6 @@ class TrackableTask(BlueapiBaseModel):
 
     task_id: str
     task: Task
-    request_id: str | SkipJsonSchema[None] = None
     is_complete: bool = False
     is_pending: bool = True
     errors: list[str] = Field(default_factory=list)
@@ -275,16 +272,7 @@ def submit_task(self, task: Task) -> str:
         task_id: str = str(uuid.uuid4())
         task.metadata["blueapi_task_id"] = task_id
         add_span_attributes({"TaskId": task_id})
-        request_id = get_baggage("correlation_id")
-        # If request id is not a string, we do not pass it into a TrackableTask
-        if not isinstance(request_id, str):
-            LOGGER.warning(f"Invalid correlation id detected: {request_id}")
-            request_id = None
-        trackable_task = TrackableTask(
-            task_id=task_id,
-            request_id=request_id,
-            task=task,
-        )
+        trackable_task = TrackableTask(task_id=task_id, task=task)
         self._pending_tasks[task_id] = trackable_task
         return task_id
 
@@ -303,7 +291,7 @@ def _submit_trackable_task(self, trackable_task: TrackableTask) -> None:
 
         task_started = Event()
 
-        def mark_task_as_started(event: WorkerEvent, _: str | None) -> None:
+        def mark_task_as_started(event: WorkerEvent) -> None:
             if (
                 event.task_status is not None
                 and event.task_status.task_id == trackable_task.task_id
@@ -544,7 +532,6 @@ def _report_status(
                 task_failed=bool(self._current.errors),
                 result=self._current.outcome,
             )
-            correlation_id = self._current.task_id
             add_span_attributes(
                 {
                     "task_id": self._current.task_id,
@@ -554,15 +541,14 @@ def _report_status(
             )
         else:
             task_status = None
-            correlation_id = None
 
         event = WorkerEvent(
             state=self._state,
             task_status=task_status,
             errors=errors,
             warnings=warnings,
         )
-        self._worker_events.publish(event, correlation_id)
+        self._worker_events.publish(event)
 
     def _on_document(self, name: str, document: Mapping[str, Any]) -> None:
         if self._current is not None:
@@ -585,12 +571,10 @@ def _on_document(self, name: str, document: Mapping[str, Any]) -> None:
                         }
                     )
 
-                    correlation_id = self._current.request_id
                     self._data_events.publish(
                         DataEvent(
                             name=name, task_id=self._current.task_id, doc=document
-                        ),
-                        correlation_id,
+                        )
                     )
             else:
                 raise ValueError(
@@ -670,8 +654,7 @@ def _publish_status_snapshot(self) -> None:
                 ProgressEvent(
                     task_id=self._current.task_id,
                     statuses=self._status_snapshot,
-                ),
-                self._current.task_id,
+                )
             )
 
 

tests/unit_tests/cli/test_cli.py

@@ -291,7 +291,6 @@ def mock_events(topic: MessageTopic, callback: Callable[[Any, Any], Any]):
         if topic.name != "public.worker.event":
             return
         ctx = Mock()
-        ctx.correlation_id = task_id
         callback(
             WorkerEvent(
                 state=WorkerState.RUNNING,

tests/unit_tests/client/test_client.py

@@ -48,7 +48,6 @@
 def mock_events() -> EventBusClient:
     mock_events = MagicMock(spec=EventBusClient)
     ctx = Mock()
-    ctx.correlation_id = "foo"
     mock_events.subscribe_to_all_events = lambda on_event: on_event(ctx, COMPLETE_EVENT)
     return mock_events
 
@@ -374,7 +373,6 @@ def test_run_stomp_sets_up_control(
     mock_rest.create_task.return_value = TaskResponse(task_id="foo")
     mock_rest.update_worker_task.return_value = TaskResponse(task_id="foo")
     ctx = Mock()
-    ctx.correlation_id = "foo"
     mock_events.subscribe_to_all_events = lambda on_event: on_event(COMPLETE_EVENT, ctx)
 
     client_with_events.run_stomp(
@@ -395,7 +393,6 @@ def test_run_stomp_fails_on_failing_event(
     mock_rest.update_worker_task.return_value = TaskResponse(task_id="foo")
 
     ctx = Mock()
-    ctx.correlation_id = "foo"
     mock_events.subscribe_to_all_events = lambda on_event: on_event(FAILED_EVENT, ctx)
 
     on_event = Mock()
@@ -438,7 +435,6 @@ def test_run_stomp_calls_event_callback(
     mock_rest.update_worker_task.return_value = TaskResponse(task_id="foo")
 
     ctx = Mock()
-    ctx.correlation_id = "foo"
 
     def callback(on_event: Callable[[AnyEvent, MessageContext], None]):
         on_event(test_event, ctx)
@@ -480,7 +476,6 @@ def test_run_stomp_ignores_non_matching_events(
     mock_rest.update_worker_task.return_value = TaskResponse(task_id="foo")  # type: ignore
 
     ctx = Mock()
-    ctx.correlation_id = "foo"
 
     def callback(on_event: Callable[[AnyEvent, MessageContext], None]):
         on_event(test_event, ctx)  # type: ignore
@@ -715,7 +710,6 @@ def test_client_callbacks(
     mock_rest.update_worker_task.return_value = TaskResponse(task_id="foo")
 
     ctx = Mock()
-    ctx.correlation_id = "foo"
 
     def subscribe(on_event: Callable[[AnyEvent, MessageContext], None]):
         on_event(test_event, ctx)
@@ -743,7 +737,6 @@ def test_client_callback_failures(
     mock_rest.update_worker_task.return_value = TaskResponse(task_id="foo")
 
     ctx = Mock()
-    ctx.correlation_id = "foo"
 
     evt = DataEvent(name="start", doc={}, task_id="foo")