feat: Implement @defer and @stream directives for incremental delivery

[jwaldrip]

Summary

This PR implements the GraphQL @defer and @stream directives for incremental delivery as specified in the GraphQL Incremental Delivery RFC.

Note: These directives are still in draft/RFC stage and not yet part of the finalized GraphQL specification. The implementation may change as the specification evolves.

Features

Core Directives

• @defer directive: Defer execution of fragments to reduce initial response time
• @stream directive: Stream list fields incrementally with configurable batch sizes
• Opt-in: Requires import_directives Absinthe.Type.BuiltIns.IncrementalDirectives in your schema

Unified Streaming Architecture

• Subscriptions with @defer/@stream: Subscriptions now support incremental delivery automatically
• Shared execution path: Both queries and subscriptions use the same TaskExecutor for deferred tasks
• Backwards compatible: Existing PubSub implementations work unchanged

Pluggable Executors

• Custom backends: Implement Absinthe.Streaming.Executor for Oban, RabbitMQ, GenStage, etc.
• Flexible configuration: Schema-level, per-request, or application-wide
• Default executor: Task.async_stream with configurable concurrency and timeouts

Transport & Integration

• Transport support: GraphQL-WS, Server-Sent Events, and extensible transport layer
• Dataloader integration: Maintains efficient batching with incremental delivery
• Relay compatibility: Full support for streaming Relay connections

Observability

• Telemetry events: [:absinthe, :incremental, :delivery, :*] for monitoring
• on_event callback: Custom monitoring integrations (Sentry, DataDog, etc.)
• Complexity analysis: Proper cost calculation for deferred/streamed operations

Architecture

Absinthe.Streaming
├── Executor        - Behaviour for pluggable execution backends
├── TaskExecutor    - Default executor (Task.async_stream)
└── Delivery        - Handles pubsub delivery for subscriptions

Query/Mutation Path:
  Request → StreamingResolution → Transport → TaskExecutor → Client

Subscription Path:
  Mutation → Subscription.Local → StreamingResolution → Streaming.Delivery
           → TaskExecutor → pubsub.publish_subscription/2 → Client

Usage

Enable in Schema

defmodule MyApp.Schema do
  use Absinthe.Schema

  # Required: opt-in to @defer/@stream directives
  import_directives Absinthe.Type.BuiltIns.IncrementalDirectives

  # Optional: custom executor for deferred tasks
  @streaming_executor MyApp.ObanExecutor

  query do
    # ...
  end

  subscription do
    # Subscriptions with @defer work automatically
  end
end

Query with @defer

query GetUser($id: ID!) {
  user(id: $id) {
    id
    name
    ... @defer(label: "profile") {
      email
      profile { bio avatar }
    }
  }
}

Subscription with @defer

subscription OnOrderUpdated($orderId: ID!) {
  orderUpdated(orderId: $orderId) {
    id
    status
    ... @defer(label: "customer") {
      customer { name email }
    }
  }
}

Custom Executor Example

defmodule MyApp.ObanExecutor do
  @behaviour Absinthe.Streaming.Executor

  @impl true
  def execute(tasks, opts) do
    tasks
    |> Enum.map(&queue_to_oban/1)
    |> stream_results(opts)
  end
end

Testing

Comprehensive test coverage including:

• Unit tests for directives and executor
• Backwards compatibility tests for subscriptions
• Integration tests with dataloader
• Transport protocol tests
• Error handling scenarios

1516 tests, 0 failures

Breaking Changes

None - incremental delivery is opt-in and backward compatible:

• Existing schemas work unchanged
• Existing PubSub implementations work unchanged
• @defer/@stream only available when explicitly imported

Documentation

• Incremental Delivery Guide - Full documentation
• Subscriptions Guide - Updated with @defer/@stream section
• CHANGELOG - Updated with new features

[cschiewek]

@jwaldrip This is fantastic. I'm going to try and get folks from my team to review this in the next couple weeks. If you could update your branch to kick off a new CI again, that'd be amazing. Thanks!

[bryanjos]

Great work! I had a few comments and suggestions on my initial look

[bryanjos]

Is this file needed for something or could it be removed?

[jwaldrip]

Removed in a recent commit. This was a leftover from early development.

[bryanjos]

I see this, INCREMENTAL_DELIVERY.md and README_INCREMENTAL.md and they look similar. Could one or more be removed and consolidated?

[jwaldrip]

Good catch! These have been consolidated. The duplicate files (INCREMENTAL_DELIVERY.md and README_INCREMENTAL.md) have been removed - only guides/incremental-delivery.md remains now.

[bryanjos]

Not sure if this should be kept in or not. If so it could be set to the lowest version or elixir and erlang supported

[jwaldrip]

Removed. The mainline repo doesn't have a version file, so this has been deleted to stay aligned.

[bryanjos]

This could be very useful!

[jwaldrip]

Thanks! It's designed to be an optional optimization layer - users can enable it to have @defer/@stream automatically suggested for expensive fields based on configurable thresholds.

[bryanjos]

Is there something that starts this supervisor? I couldn't find anything but I might have missed it.

[jwaldrip]

The supervisor needs to be added to your application's supervision tree manually. I've updated the moduledoc with clear instructions showing exactly how to do this in application.ex. The supervisor is only required for actual incremental delivery over transports (SSE, WebSocket) - standard execution works without it.

[bryanjos]

You can use start_supervised here that'll start and stop the supervisor for each test

Suggested change

	case Absinthe.Incremental.Supervisor.start_link(
	enabled: true,
	enable_defer: true,
	enable_stream: true
	) do
	{:ok, _pid} -> :ok
	{:error, {:already_started, _pid}} -> :ok
	end
	{:ok, _pid} = start_supervised({Absinthe.Incremental.Supervisor, enabled: true, enable_defer: true, enable_stream: true})

[jwaldrip]

Good suggestion! The test file has been refactored - it now focuses on directive definitions and parsing which don't require the supervisor. Integration tests that need the supervisor have been moved elsewhere.

[yordis]

Hey folks, any updates on this one? Is there anything I can do to help to push the feature forward?

[jwaldrip]

Hey @yordis! The PR has been updated and all review feedback has been addressed:

• Removed duplicate documentation files (consolidated to guides/incremental-delivery.md)
• Removed .tool-versions to align with mainline
• Removed leftover test files
• Added clear documentation for Supervisor startup and Dataloader integration
• Fixed several bugs in the incremental delivery implementation
• All 1483 tests pass

The branch is up to date with mainline and ready for review. The companion PRs for transport layers are also ready:

• absinthe_plug #306 - SSE transport
• absinthe_graphql_ws #36 - WebSocket transport

[jwaldrip]

Unified Streaming Architecture Update

This commit unifies the subscription and incremental delivery systems, enabling @defer/@stream support in subscriptions with pluggable task execution.

New Modules

Module	Purpose
Absinthe.Streaming	Umbrella module with shared helpers
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

Key Features

1. Subscriptions with @defer/@stream

Subscriptions now support incremental delivery automatically:

subscription {
  orderUpdated(orderId: "123") {
    id
    status
    ... @defer(label: "customer") {
      customer { name email }
    }
  }
}

Clients receive multiple payloads using the standard GraphQL incremental format. Existing PubSub implementations work unchanged.

2. Pluggable Executors

Custom execution backends (Oban, RabbitMQ, etc.) can be configured:

defmodule MyApp.Schema do
  use Absinthe.Schema
  
  @streaming_executor MyApp.ObanExecutor
  
  # ...
end

3. Unified Code Path

Both queries and subscriptions now share the same TaskExecutor for deferred task execution, ensuring consistent behavior and a single configuration point.

Architecture

Query/Mutation Path:
  Request → StreamingResolution → Transport → TaskExecutor → Client

Subscription Path:
  Mutation → Subscription.Local → StreamingResolution → Streaming.Delivery
           → TaskExecutor → pubsub.publish_subscription/2 → Client

Documentation

• Updated guides/incremental-delivery.md with custom executor guide
• Updated guides/subscriptions.md with @defer/@stream section
• Updated CHANGELOG.md with new features

Tests

All 1516 tests pass, including new tests for:

• TaskExecutor unit tests
• Backwards compatibility tests for subscriptions

[bryanjos]

Awesome work! Approving as I only had a few nits and questions

[bryanjos]

Could this be moved to its own file in the middleware folder so it is easier to find?

[bryanjos]

Could this also be put in its own file so it's easier to find?

[bryanjos]

Any worries getting stream tasks for a defer or the other way around?

[bryanjos]

The changelog should be updated automatically

[jwaldrip]

Good question! This is actually safe because both task types can coexist in the same query and share the same execution path. Each task has a :type field (:defer or :stream) that determines how it's processed downstream:

• In Streaming.Delivery.build_success_payload/3 (lines 200-218), we pattern match on task.type to call either Response.build_incremental/4 for defer or Response.build_stream_incremental/4 for stream
• In Transport.build_task_response/3 (lines 323-356), same pattern matching ensures correct response formatting
• The executor itself is type-agnostic - it just runs the task.execute/0 function and returns results

So mixing them is intentional and safe. A query like:

{
  user {
    ... @defer { profile }  # defer task
    posts @stream { title }  # stream task  
  }
}

Results in both task types being combined into a single stream and executed together, which is exactly what we want.

[jwaldrip]

Thanks for the heads up! I've manually updated the changelog to document the new features (unified streaming architecture, pluggable executors, subscriptions with @defer/@stream).

If you have an automated changelog generation workflow (e.g., via conventional commits or release-please), I'm happy to adjust. Just let me know the process and I'll follow it.

[jwaldrip]

PR Review Feedback Addressed

Thanks @bryanjos for the detailed review! I've addressed all the feedback:

1. ✅ Move IncrementalComplexity middleware to its own file

Commit: 96fa747

Moved Absinthe.Middleware.IncrementalComplexity from lib/absinthe/incremental/complexity.ex to lib/absinthe/middleware/incremental_complexity.ex for better discoverability.

2. ✅ Move TelemetryReporter to its own file

Commit: 96fa747

Moved Absinthe.Incremental.TelemetryReporter from lib/absinthe/incremental/supervisor.ex to lib/absinthe/incremental/telemetry_reporter.ex for better organization.

3. ✅ Defer/Stream task mixing concern

Responded in thread - this is intentional and safe. Both task types use the same executor and are distinguished by their :type field during response formatting.

4. ✅ CHANGELOG automatic update

Responded in thread - happy to adjust if you have an automated process.

Tests

All 1516 tests passing ✅

The refactoring improves code organization without changing any functionality.