feat: Message data classes

.github/workflows/lint-test.yaml

@@ -118,6 +118,14 @@ jobs:
     strategy:
       matrix:
         python_version: [3.12, 3.13, 3.14]
+    # Each Python version has its own dedicated broker lane in plugboard-infra, so the
+    # matrix legs within a run execute concurrently. A per-version concurrency group with
+    # `queue: max` serialises same-version legs across concurrent pipeline runs (queueing
+    # up to 100 pending jobs in FIFO order rather than cancelling earlier pending ones).
+    concurrency:
+      group: test-integration-${{ matrix.python_version }}
+      cancel-in-progress: false
+      queue: max
     steps:
       - name: Checkout
         uses: actions/checkout@v4
@@ -142,12 +150,52 @@ jobs:
           sleep 10 # Wait for services to start
 
       - name: Run integration tests
-        run: COVERAGE_FILE=.coverage.py${{ matrix.python_version }}.integration uv run coverage run -m pytest ./tests/integration/ -m "not tuner"
+        # The message broker round trips are part of this suite and run on every Python
+        # leg. Each leg resolves its own dedicated broker lane (one SNS+SQS pair and
+        # Pub/Sub topic+subscription per Python version, provisioned in plugboard-infra
+        # and stored in the BROKER_LANES_JSON variable), so the legs run concurrently
+        # without contending on a shared queue or subscription. When broker tests are off
+        # the lane resolution is skipped and the broker tests self-skip.
+        run: |
+          # Resolve this leg's broker lane. The broker tests self-skip unless these env vars
+          # are set, so when broker tests are off, or lanes have not been provisioned for this
+          # run (e.g. a fork pull request where vars/secrets are unavailable), they simply skip.
+          if [ "${PLUGBOARD_MESSAGE_BROKER_TESTS}" = "true" ] && [ -n "${BROKER_LANES_JSON:-}" ]; then
+            # A lanes file is present but has no entry for this Python version: that is a
+            # configuration mismatch (version in the matrix but not in terraform), so fail loud.
+            if ! jq -e --arg v "$PYTHON_VERSION" 'has($v)' <<< "$BROKER_LANES_JSON" > /dev/null; then
+              echo "::error::No broker lane for Python $PYTHON_VERSION in BROKER_LANES_JSON"
+              exit 1
+            fi
+            printf '%s' "$GCP_SA_KEY_JSON" > /tmp/gcp-sa-key.json
+            # The lane values are queue URLs, topic ARNs and resource ids (no shell metacharacters),
+            # so the generated `export` lines are safe to eval.
+            eval "$(jq -r --arg v "$PYTHON_VERSION" \
+              '.[$v] | "export AWS_SQS_QUEUE_URL=\(.aws_sqs_queue_url)\nexport AWS_SNS_TOPIC_ARN=\(.aws_sns_topic_arn)\nexport GCP_TOPIC_ID=\(.gcp_topic_id)\nexport GCP_SUBSCRIPTION_ID=\(.gcp_subscription_id)"' \
+              <<< "$BROKER_LANES_JSON")"
+            export GOOGLE_APPLICATION_CREDENTIALS=/tmp/gcp-sa-key.json
+            echo "Broker lane for Python $PYTHON_VERSION:"
+            echo "  AWS_SQS_QUEUE_URL=$AWS_SQS_QUEUE_URL"
+            echo "  GCP_TOPIC_ID=$GCP_TOPIC_ID"
+          fi
+          COVERAGE_FILE=.coverage.py${{ matrix.python_version }}.integration \
+            uv run coverage run -m pytest ./tests/integration/ -m "not tuner"
         env:
           RABBITMQ_URL: amqp://user:password@localhost:5672/
           REDIS_URL: redis://default:password@localhost:6379/
           RAY_ENABLE_UV_RUN_RUNTIME_ENV: 0
           PLUGBOARD_IO_READ_TIMEOUT: 5.0
+          # Message broker test inputs. The per-lane identifiers are resolved from
+          # BROKER_LANES_JSON in the step above (only when the tests are enabled).
+          PYTHON_VERSION: ${{ matrix.python_version }}
+          PLUGBOARD_MESSAGE_BROKER_TESTS: ${{ vars.PLUGBOARD_MESSAGE_BROKER_TESTS }}
+          BROKER_LANES_JSON: ${{ vars.BROKER_LANES_JSON }}
+          GCP_SA_KEY_JSON: ${{ secrets.GCP_SA_KEY_JSON }}
+          AWS_REGION: ${{ vars.AWS_REGION }}
+          GCP_PROJECT_ID: ${{ vars.GCP_PROJECT_ID }}
+          GCP_PUBSUB_PROJECT_ID: ${{ vars.GCP_PUBSUB_PROJECT_ID }}
+          AWS_ACCESS_KEY_ID: ${{ secrets.AWS_ACCESS_KEY_ID }}
+          AWS_SECRET_ACCESS_KEY: ${{ secrets.AWS_SECRET_ACCESS_KEY }}
 
       - name: Upload integration test coverage
         uses: actions/upload-artifact@v4

README.md

@@ -69,7 +69,7 @@ Optional integrations for different cloud providers can be installed using `plug
 
 Support for parallelisation and hyperparameter optimisation can be installed using `plugboard[ray]`.
 
-Additional optional extras: `plugboard[llm]` for LLM components, `plugboard[redis]` for Redis-based connectors, `plugboard[omq]` for the pyomq backend for ZMQ connectors, and `plugboard[websockets]` for WebSocket I/O.
+Additional optional extras: `plugboard[llm]` for LLM components, `plugboard[redis]` for Redis-based connectors, `plugboard[omq]` for the pyomq backend for ZMQ connectors, `plugboard[websockets]` for WebSocket I/O, and `plugboard[gcp-pubsub]` or `plugboard[aws-messaging]` for [message data components](https://docs.plugboard.dev/latest/usage/message-data/).
 
 ## ⚡ Quickstart with AI
 

docs/usage/configuration.md

@@ -24,6 +24,25 @@ Plugboard can make use of a message broker for data exchange between components
 | `RABBITMQ_URL`      | URL for RabbitMQ AMQP message broker (must include credentials if required)  | |
 | `REDIS_URL`         | URL for Redis message broker (must include credentials if required)  | |
 
+These brokers carry data *between components* within a run. To read or write records
+through an external broker as part of the model itself, see
+[Message Data](message-data.md).
+
+### Message data brokers
+
+The message data components ([`MessageDataReader`][plugboard.library.MessageDataReader] /
+[`MessageDataWriter`][plugboard.library.MessageDataWriter] and their broker
+implementations) take connection details as constructor arguments, falling back to the
+environment below when they are not supplied:
+
+| Option Name               | Description                            | Default Value |
+|---------------------------|----------------------------------------|---------------|
+| `GCP_PUBSUB_PROJECT_ID`   | GCP project for PubSub topics and subscriptions | |
+| `AWS_REGION`              | AWS region for SQS and SNS clients     | |
+
+Note that `AWS_REGION` is also read by the AWS SDK itself, so setting it affects both
+Plugboard's defaults and the credential/endpoint resolution of the underlying client.
+
 ## Job ID
 
 Each plugboard run has a unique job ID associated with it. This is used to: track state for each run; and separate data messages between runs when using a message broker. Typically, a run would be started without explicitly setting the job ID, in which case a unique job ID will be created automatically. However, there are instances when it may be desirable to specify the job ID, such as stopping a run and resuming the same run later with the existing persisted state. In these scenarios the job ID can be set with the below environment variable which will then be used by any `StateBackend`, `Process` and `Component` while the value is set.

docs/usage/message-data.md

@@ -0,0 +1,172 @@
+# Message Data
+
+Message data components read and write **records** through a pub/sub message broker,
+rather than through a channel wired between components. They are the broker equivalent
+of [`FileReader`][plugboard.library.FileReader] / [`FileWriter`][plugboard.library.FileWriter]:
+each message carries one record, and the fields named by `field_names` become the
+component's outputs (for a reader) or inputs (for a writer).
+
+| Class | Broker | Reads from | Writes to |
+|---|---|---|---|
+| [`GCPPubSubDataReader`][plugboard.library.GCPPubSubDataReader] | Google Cloud PubSub | a subscription | |
+| [`GCPPubSubDataWriter`][plugboard.library.GCPPubSubDataWriter] | Google Cloud PubSub | | a topic |
+| [`AWSSQSDataReader`][plugboard.library.AWSSQSDataReader] | AWS SQS | a queue | |
+| [`AWSSNSDataWriter`][plugboard.library.AWSSNSDataWriter] | AWS SNS | | a topic |
+
+All four build on the [`MessageDataReader`][plugboard.library.MessageDataReader] and
+[`MessageDataWriter`][plugboard.library.MessageDataWriter] base classes, which handle
+connecting, reconnection with backoff, batching and acknowledgment.
+
+## Installing
+
+Each broker uses its own client library, supplied as an extra:
+
+```shell
+pip install "plugboard[gcp-pubsub]"     # google-cloud-pubsub
+pip install "plugboard[aws-messaging]"  # aiobotocore
+```
+
+Constructing a component without its extra raises an `ImportError` naming the extra to
+install.
+
+## Reading and writing records
+
+Messages are expected to carry a JSON object per record, whose keys match
+`field_names`:
+
+```python
+import asyncio
+
+from plugboard.component import Component, IOController as IO
+from plugboard.connector import AsyncioConnector
+from plugboard.library import AWSSNSDataWriter, GCPPubSubDataReader
+from plugboard.process import LocalProcess
+from plugboard.schemas import ConnectorSpec
+
+
+class Bucket(Component):
+    """Turns a raw measurement into a bucketed label before it leaves the model."""
+
+    io = IO(inputs=["x", "y"], outputs=["bucket"])
+
+    async def step(self) -> None:
+        self.bucket = "high" if self.x + self.y > 50 else "low"
+
+
+async def main() -> None:
+    reader = GCPPubSubDataReader(
+        name="reader",
+        subscription_id="measurements-pull",
+        # project_id is optional; it defaults to the GCP_PUBSUB_PROJECT_ID variable
+        field_names=["x", "y"],
+        chunk_size=20,
+    )
+    bucket = Bucket(name="bucket")
+    writer = AWSSNSDataWriter(
+        name="writer",
+        topic_arn="arn:aws:sns:eu-west-1:123456789012:measurements",
+        region="eu-west-1",  # Optional; defaults to AWS_REGION
+        field_names=["bucket"],
+        chunk_size=20,
+    )
+
+    process = LocalProcess(
+        name="measurements",
+        components=[reader, bucket, writer],
+        connectors=[
+            AsyncioConnector(spec=ConnectorSpec(source="reader.x", target="bucket.x")),
+            AsyncioConnector(spec=ConnectorSpec(source="reader.y", target="bucket.y")),
+            AsyncioConnector(spec=ConnectorSpec(source="bucket.bucket", target="writer.bucket")),
+        ],
+    )
+    async with process:
+        await process.run()
+
+
+asyncio.run(main())
+```
+
+The components are ordinary `Component`s, so they also work in a `RayProcess` and with
+any connector.
+
+## Behaviour worth knowing
+
+- **A reader waits; it does not finish.** An empty poll means "nothing yet", not "no
+  more data" (unlike a file). A reader started before its producer keeps running and
+  picks up messages as they arrive; to end a run that would otherwise wait forever, stop
+  the process. A reader only stops on its own when the broker reports the source as gone
+  (for example a deleted subscription or queue), at which point it closes its stream and
+  the run ends.
+- **Messages are acknowledged after they are processed.** A batch is acknowledged only
+  once every record in it has been published downstream, so a crash mid-batch leaves the
+  rest to be redelivered (at-least-once). For SQS, "acknowledge" means deleting the
+  messages.
+- **Writers batch by `chunk_size`.** A writer flushes to the broker when its input
+  buffer reaches `chunk_size` records, and flushes whatever is left over when the run
+  ends. `chunk_size` defaults to `None`, so a writer with no `chunk_size` buffers every
+  record in memory and publishes them all at the end rather than streaming during the
+  run — set `chunk_size` (for example `chunk_size=1` for live per-record publishing) to
+  stream. For SQS the effective size is capped at 10 messages per send, the broker
+  limit.
+- **Failures are retried with backoff.** Transient broker errors trigger a reconnect and
+  retry; permanent errors (such as `AccessDenied`) fail immediately. Override the policy
+  with the `retry_policy` argument, e.g.
+  `retry_policy=RetryPolicy(max_retries=5, base_delay=0.5, max_delay=30.0)`; an empty
+  poll pauses for `idle_poll_delay` seconds.
+- **Encoding.** By default each record is JSON-encoded (`parse_json=True`). With
+  `parse_json=False` a writer sends only the first field's value and a reader exposes the
+  raw payload under a `data` field.
+- **SNS fan-out is transparent.** Subscribing an SQS queue to an SNS topic delivers the
+  notification envelope, which `AWSSQSDataReader` unwraps automatically; a non-object
+  body is exposed under `data`.
+- **Messages are scoped by `job_id`.** Writers attach the active Plugboard `job_id` as a
+  `job_id` message attribute, and readers filter incoming messages by `job_id`
+  (acknowledging and discarding non-matching messages). Pass `job_id=...` to override the
+  active process job ID when reading messages produced by another job.
+
+## Configuration
+
+Connection details are passed explicitly or read from the environment (explicit arguments
+always win), so the same model can target a different account or region without code
+changes. The message-data components read the `GCP_PUBSUB_PROJECT_ID` and `AWS_REGION`
+environment variables; see [Configuration](configuration.md#message-data-brokers) for
+how these are resolved. Credentials are never configured through Plugboard — see
+[Credentials](#credentials) for the mechanisms each provider picks up.
+
+## Credentials
+
+Plugboard never configures, stores or refreshes credentials itself: the broker client
+libraries resolve them from your environment, so authenticate **before** starting a run.
+The mechanisms below are picked up in the provider's own precedence order — see the
+provider documentation for full details.
+
+**AWS (SQS/SNS)** uses the
+[standard credential chain](https://docs.aws.amazon.com/sdkref/latest/guide/standardized-credentials.html):
+
+- Static keys in the environment: `AWS_ACCESS_KEY_ID` + `AWS_SECRET_ACCESS_KEY`
+  (plus `AWS_SESSION_TOKEN` for temporary credentials).
+- Shared credentials/config files, optionally selected with `AWS_PROFILE` — including
+  IAM Identity Center (SSO) and assume-role profiles.
+- The instance or container role when running on EC2/ECS/EKS.
+
+**GCP (PubSub)** uses
+[Application Default Credentials](https://cloud.google.com/docs/authentication/application-default-credentials):
+
+- A service account key file pointed at by `GOOGLE_APPLICATION_CREDENTIALS`.
+- User credentials from `gcloud auth application-default login`.
+- The metadata server / Workload Identity when running on GCP infrastructure.
+
+### Long-running jobs and expiring sessions
+
+Credentials that renew themselves — instance/container roles, service account keys, and
+ADC refresh tokens — keep working for the life of a long-running job: the client
+libraries refresh them automatically as access tokens expire.
+
+IAM Identity Center (SSO) sessions are the exception: the login token is valid only for
+your organisation's configured session length (1-12 hours, commonly 8) and can only be
+renewed by an interactive `aws sso login`. If a job outlives the session, the next broker
+call fails; the component retries with backoff — each reconnect re-runs credential
+resolution, which heals any provider that can refresh — and then fails with the
+provider's own error rather than a synthetic one. Prefer a non-interactive credential
+source (static keys, an instance role, or a service account) for jobs that may run
+longer than a login session.

mkdocs.yaml

@@ -129,6 +129,7 @@ nav:
     - Event-driven models: examples/tutorials/event-driven-models.md
     - Tuning a process: examples/tutorials/tuning-a-process.md
   - Configuration: usage/configuration.md
+  - Message Data: usage/message-data.md
   - AI-Assisted Development: usage/ai.md
   - Topics: usage/topics.md
 - Demos:

plugboard/exceptions/__init__.py

@@ -116,3 +116,27 @@ class ProcessStatusError(Exception):
     """Raised when a `Process` is in an invalid state for the requested operation."""
 
     pass
+
+
+class MessageBrokerError(Exception):
+    """Base exception for message broker errors."""
+
+    pass
+
+
+class MessageBrokerConnectionError(MessageBrokerError):
+    """Raised when connection to a message broker fails."""
+
+    pass
+
+
+class MessageBrokerTransientError(MessageBrokerError):
+    """Raised on transient message broker errors (eligible for retry)."""
+
+    pass
+
+
+class MessageBrokerPermanentError(MessageBrokerError):
+    """Raised on permanent message broker errors (not eligible for retry)."""
+
+    pass

plugboard/library/__init__.py

@@ -1,20 +1,32 @@
 """Provides implementations of Plugboard objects for use in user models."""
 
+from plugboard.utils.retry import RetryPolicy
+from .aws_messaging_io import AWSSNSDataWriter, AWSSQSDataReader
 from .data_reader import DataReader
 from .data_writer import DataWriter
 from .file_io import FileReader, FileWriter
+from .gcp_pubsub_io import GCPPubSubDataReader, GCPPubSubDataWriter
 from .llm import LLMChat, LLMImageProcessor
+from .message_reader import MessageDataReader
+from .message_writer import MessageDataWriter
 from .sql_io import SQLReader, SQLWriter
 from .websocket_io import WebsocketBase, WebsocketReader, WebsocketWriter
 
 
 __all__ = [
+    "AWSSQSDataReader",
+    "AWSSNSDataWriter",
     "DataReader",
     "DataWriter",
-    "LLMChat",
-    "LLMImageProcessor",
     "FileReader",
     "FileWriter",
+    "GCPPubSubDataReader",
+    "GCPPubSubDataWriter",
+    "LLMChat",
+    "LLMImageProcessor",
+    "MessageDataReader",
+    "MessageDataWriter",
+    "RetryPolicy",
     "SQLReader",
     "SQLWriter",
     "WebsocketBase",