feat: unify streaming architecture for subscriptions and incremental delivery

- Add Absinthe.Streaming module with shared abstractions
- Add Absinthe.Streaming.Executor behaviour for pluggable task execution
- Add Absinthe.Streaming.TaskExecutor as default executor (Task.async_stream)
- Add Absinthe.Streaming.Delivery for pubsub incremental delivery
- Enable @defer/@stream in subscriptions (automatic multi-payload delivery)
- Refactor Transport to use shared TaskExecutor
- Update Subscription.Local to detect and handle incremental directives
- Add comprehensive backwards compatibility tests
- Update guides and documentation

Subscriptions with @defer/@stream now automatically deliver multiple payloads
using the standard GraphQL incremental format. Existing PubSub implementations
work unchanged - publish_subscription/2 is called multiple times.

Custom executors (Oban, RabbitMQ, etc.) can be configured via:
- Schema attribute: @streaming_executor MyApp.ObanExecutor
- Context: context: %{streaming_executor: MyApp.ObanExecutor}
- Application config: config :absinthe, :streaming_executor, MyApp.ObanExecutor

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

Author: jwaldrip

CHANGELOG.md

@@ -6,12 +6,26 @@
 
 * **draft-spec:** Add `@defer` and `@stream` directives for incremental delivery ([#1377](https://github.com/absinthe-graphql/absinthe/pull/1377))
   - **Note:** These directives are still in draft/RFC stage and not yet part of the finalized GraphQL specification
-  - **Opt-in required:** `import_types Absinthe.Type.BuiltIns.IncrementalDirectives` in your schema
+  - **Opt-in required:** `import_directives Absinthe.Type.BuiltIns.IncrementalDirectives` in your schema
   - Split GraphQL responses into initial + incremental payloads
   - Configure via `Absinthe.Pipeline.Incremental.enable/2`
   - Resource limits (max concurrent streams, memory, duration)
   - Dataloader integration for batched loading
   - SSE and WebSocket transport support
+* **subscriptions:** Support `@defer` and `@stream` in subscriptions
+  - Subscriptions with deferred content deliver multiple payloads automatically
+  - Existing PubSub implementations work unchanged (calls `publish_subscription/2` multiple times)
+  - Uses standard GraphQL incremental delivery format that clients already understand
+* **streaming:** Unified streaming architecture for queries and subscriptions
+  - New `Absinthe.Streaming` module consolidates shared abstractions
+  - `Absinthe.Streaming.Executor` behaviour for pluggable task execution backends
+  - `Absinthe.Streaming.TaskExecutor` default executor using `Task.async_stream`
+  - `Absinthe.Streaming.Delivery` handles pubsub delivery for subscriptions
+  - Both query and subscription incremental delivery share the same execution path
+* **executors:** Pluggable task execution backends
+  - Implement `Absinthe.Streaming.Executor` to use custom backends (Oban, RabbitMQ, etc.)
+  - Configure via `@streaming_executor` schema attribute, context, or application config
+  - Default executor uses `Task.async_stream` with configurable concurrency and timeouts
 * **telemetry:** Add telemetry events for incremental delivery
   - `[:absinthe, :incremental, :delivery, :initial]` - initial response
   - `[:absinthe, :incremental, :delivery, :payload]` - each deferred/streamed payload

guides/incremental-delivery.md

@@ -38,7 +38,7 @@ defmodule MyApp.Schema do
   use Absinthe.Schema
 
   # Import the draft-spec @defer and @stream directives
-  import_types Absinthe.Type.BuiltIns.IncrementalDirectives
+  import_directives Absinthe.Type.BuiltIns.IncrementalDirectives
 
   query do
     # ...
@@ -496,6 +496,176 @@ Existing queries work without changes. To add incremental delivery:
 4. **Configure transport** to handle streaming responses
 5. **Add monitoring** to track performance improvements
 
+## Subscriptions with @defer/@stream
+
+Subscriptions support the same `@defer` and `@stream` directives as queries. When a subscription contains deferred content, clients receive multiple payloads:
+
+1. **Initial payload**: Immediately available subscription data
+2. **Incremental payloads**: Deferred/streamed content as it resolves
+
+```graphql
+subscription OnOrderUpdated($orderId: ID!) {
+  orderUpdated(orderId: $orderId) {
+    id
+    status
+
+    # Defer expensive customer lookup
+    ... @defer(label: "customer") {
+      customer {
+        name
+        email
+        loyaltyTier
+      }
+    }
+  }
+}
+```
+
+This is handled automatically by the subscription system. Existing PubSub implementations work unchanged - the same `publish_subscription/2` callback is called multiple times with the standard GraphQL incremental format.
+
+### How It Works
+
+When a mutation triggers a subscription with `@defer`/`@stream`:
+
+1. `Subscription.Local` detects the directives in the subscription document
+2. The `StreamingResolution` phase executes, collecting deferred tasks
+3. `Streaming.Delivery` publishes the initial payload via `pubsub.publish_subscription/2`
+4. Deferred tasks are executed via the configured executor
+5. Each result is published as an incremental payload
+
+```elixir
+# What happens internally (you don't need to do this manually)
+pubsub.publish_subscription(topic, %{
+  data: %{orderUpdated: %{id: "123", status: "SHIPPED"}},
+  pending: [%{id: "0", label: "customer", path: ["orderUpdated"]}],
+  hasNext: true
+})
+
+# Later...
+pubsub.publish_subscription(topic, %{
+  incremental: [%{
+    id: "0",
+    data: %{customer: %{name: "John", email: "john@example.com", loyaltyTier: "GOLD"}}
+  }],
+  hasNext: false
+})
+```
+
+## Custom Executors
+
+By default, deferred and streamed tasks are executed using `Task.async_stream` for in-process concurrent execution. You can implement a custom executor for alternative backends:
+
+- **Oban** - Persistent, retryable job processing
+- **RabbitMQ** - Distributed task queuing
+- **GenStage** - Backpressure-aware pipelines
+- **Custom** - Any execution strategy you need
+
+### Implementing a Custom Executor
+
+Implement the `Absinthe.Streaming.Executor` behaviour:
+
+```elixir
+defmodule MyApp.ObanExecutor do
+  @behaviour Absinthe.Streaming.Executor
+
+  @impl true
+  def execute(tasks, opts) do
+    timeout = Keyword.get(opts, :timeout, 30_000)
+
+    # Queue tasks to Oban and stream results
+    tasks
+    |> Enum.map(&queue_to_oban/1)
+    |> stream_results(timeout)
+  end
+
+  defp queue_to_oban(task) do
+    %{task_id: task.id, execute_fn: task.execute}
+    |> MyApp.DeferredWorker.new()
+    |> Oban.insert!()
+  end
+
+  defp stream_results(jobs, timeout) do
+    # Return an enumerable of results matching this shape:
+    # %{
+    #   task: original_task,
+    #   result: {:ok, data} | {:error, reason},
+    #   has_next: boolean,
+    #   success: boolean,
+    #   duration_ms: integer
+    # }
+    Stream.resource(
+      fn -> {jobs, timeout} end,
+      &poll_next_result/1,
+      fn _ -> :ok end
+    )
+  end
+end
+```
+
+### Configuring a Custom Executor
+
+**Schema-level** (recommended):
+
+```elixir
+defmodule MyApp.Schema do
+  use Absinthe.Schema
+
+  # Use custom executor for all @defer/@stream operations
+  @streaming_executor MyApp.ObanExecutor
+
+  import_directives Absinthe.Type.BuiltIns.IncrementalDirectives
+
+  query do
+    # ...
+  end
+end
+```
+
+**Per-request** (via context):
+
+```elixir
+Absinthe.run(query, MyApp.Schema,
+  context: %{streaming_executor: MyApp.ObanExecutor}
+)
+```
+
+**Application config** (global default):
+
+```elixir
+# config/config.exs
+config :absinthe, :streaming_executor, MyApp.ObanExecutor
+```
+
+### When to Use Custom Executors
+
+| Use Case | Recommended Executor |
+|----------|---------------------|
+| Simple deployments | Default `TaskExecutor` |
+| Long-running deferred operations | Oban (with persistence) |
+| Distributed systems | RabbitMQ or similar |
+| High-throughput with backpressure | GenStage |
+| Retry on failure | Oban |
+
+## Architecture
+
+The streaming system is unified across queries, mutations, and subscriptions:
+
+```
+Absinthe.Streaming
+├── Executor        - Behaviour for pluggable execution backends
+├── TaskExecutor    - Default executor (Task.async_stream)
+└── Delivery        - Handles pubsub delivery for subscriptions
+
+Query/Mutation Path:
+  Request → Pipeline → StreamingResolution → Transport → Client
+
+Subscription Path:
+  Mutation → Subscription.Local → StreamingResolution → Streaming.Delivery
+           → pubsub.publish_subscription/2 (multiple times) → Client
+```
+
+Both paths share the same `Executor` for task execution, ensuring consistent behavior and allowing a single configuration point for custom backends.
+
 ## See Also
 
 - [Subscriptions](subscriptions.md) for real-time data

guides/subscriptions.md

@@ -266,3 +266,101 @@ Since we provided a `context_id`, Absinthe will only run two documents per publi
 
 1. Once for _user 1_ and _user 3_ because they have the same context ID (`"global"`) and sent the same document.
 2. Once for _user 2_. While _user 2_ has the same context ID (`"global"`), they provided a different document, so it cannot be de-duplicated with the other two.
+
+### Incremental Delivery with Subscriptions
+
+Subscriptions support `@defer` and `@stream` directives for incremental delivery. This allows you to receive subscription data progressively - immediately available data first, followed by deferred content.
+
+First, import the incremental directives in your schema:
+
+```elixir
+defmodule MyAppWeb.Schema do
+  use Absinthe.Schema
+
+  # Enable @defer and @stream directives
+  import_directives Absinthe.Type.BuiltIns.IncrementalDirectives
+
+  # ... rest of schema
+end
+```
+
+Then use `@defer` in your subscription queries:
+
+```graphql
+subscription {
+  commentAdded(repoName: "absinthe-graphql/absinthe") {
+    id
+    content
+    author {
+      name
+    }
+
+    # Defer expensive operations
+    ... @defer(label: "authorDetails") {
+      author {
+        email
+        avatarUrl
+        recentActivity {
+          type
+          timestamp
+        }
+      }
+    }
+  }
+}
+```
+
+When a mutation triggers this subscription, clients receive multiple payloads:
+
+**Initial payload** (sent immediately):
+```json
+{
+  "data": {
+    "commentAdded": {
+      "id": "123",
+      "content": "Great library!",
+      "author": { "name": "John" }
+    }
+  },
+  "pending": [{"id": "0", "label": "authorDetails", "path": ["commentAdded"]}],
+  "hasNext": true
+}
+```
+
+**Incremental payload** (sent when deferred data resolves):
+```json
+{
+  "incremental": [{
+    "id": "0",
+    "data": {
+      "author": {
+        "email": "john@example.com",
+        "avatarUrl": "https://...",
+        "recentActivity": [...]
+      }
+    }
+  }],
+  "hasNext": false
+}
+```
+
+This is handled automatically by the subscription system. Your existing PubSub implementation works unchanged - it receives multiple `publish_subscription/2` calls with the standard GraphQL incremental format.
+
+#### Custom Executors for Subscriptions
+
+For long-running deferred operations in subscriptions, you can configure a custom executor (e.g., Oban for persistence):
+
+```elixir
+defmodule MyAppWeb.Schema do
+  use Absinthe.Schema
+
+  # Use Oban for deferred task execution
+  @streaming_executor MyApp.ObanExecutor
+
+  import_directives Absinthe.Type.BuiltIns.IncrementalDirectives
+
+  # ...
+end
+```
+
+See the [Incremental Delivery guide](incremental-delivery.md) for details on implementing custom executors.