Queues

Queues enable sophisticated multi-step data extraction workflows where one endpoint collects data (such as records or identifiers )that is used by subsequent endpoints. This is essential for APIs that require separate calls to fetch related data.

Queue Architecture

Queue Usage

Queues are auto-detected from your endpoint definitions — Sling discovers them by scanning processor output: and iterate.over: expressions. You do not need to declare them.

⚠️ Deprecated: The top-level queues: list is no longer required and is ignored at runtime. Old specs that still include it continue to load, but Sling emits a one-time deprecation warning. Remove the field when you next touch the spec.

At load time, Sling enforces that every consumed queue has at least one producer, so typos surface immediately as a queue(s) with no producer error rather than at runtime.

1. Sending Data to Queues

Use processors to send data from one endpoint to a queue:

endpoints:
  list_customers:
    description: "Get list of customers"
    
    request:
      url: "customers"
    
    response:
      records:
        jmespath: "data[]"
        primary_key: ["id"]
      
      processors:
        # Send customer IDs to queue for detailed processing
        - expression: "record.id"
          output: "queue.customer_ids"

2. Consuming Data from Queues

Use the iterate section to process each item from a queue:

endpoints:
  customer_details:
    description: "Get detailed customer information"
    
    # Process each customer ID from the queue
    iterate:
      over: "queue.customer_ids"
      into: "state.current_customer_id"
    
    request:
      url: "customers/{state.current_customer_id}"
    
    response:
      records:
        jmespath: "[*]"  # Single customer object
        primary_key: ["id"]
      
      processors:
        # Add the customer ID to each record for reference
        - expression: "state.current_customer_id"
          output: "record.customer_id"

Queue Consumption: deferred vs immediate

The iterate.consume option controls when a consumer reads its queue:

Value	Behavior
deferred (default)	The consumer waits for the producer to fully finish, then reads the queue from the start.
immediate	The consumer tails the queue live — it starts processing as soon as the producer appends records, running concurrently with the producer.

endpoints:
  search:
    queue_only: true
    request:
      url: "{state.base_url}/search"
    response:
      processors:
        - expression: "record.id"
          output: "queue.item_ids"

  details:
    iterate:
      over: "queue.item_ids"
      into: "state.item_id"
      consume: immediate   # tail the queue live instead of waiting
    request:
      url: "{state.base_url}/items/{state.item_id}"

When to use each

• deferred (default) — Use when the consumer needs the complete set of queue items before it can do its job, such as deduplication or aggregation across all records. Also the safest choice for ordinary fan-out where you don't need pipelining.

• immediate — Use when you want pipelined throughput (the consumer works while the producer is still fetching) and fail-fast behavior.

Fail-fast with immediate

When a consumer uses consume: immediate, Sling runs the producer and its live consumers concurrently (it forces threaded execution, equivalent to setting SLING_THREADS). The producer plus all the consumers that tail its queue form a fail-fast group:

• If a consumer fails, its producer (and the other consumers in the same group) are terminated — Sling stops pulling data that nothing will use.

• If a producer fails, the consumers tailing its queue are terminated — they would otherwise block forever waiting for a queue that will never complete.

• Unrelated streams and other queue groups are not affected — they continue and can finish successfully.

This avoids wasted work: with the default deferred consumption, a producer always runs to completion before the consumer even starts, so a later consumer failure means the entire producer fetch was wasted. immediate surfaces failures early while the producer is still running.

💡 Tip: immediate only changes behavior for queue iteration. It has no effect on non-queue iterate.over expressions (arrays, range(), etc.).

Queue-Only Endpoints

Some producer endpoints exist only to fan IDs into a queue for a downstream consumer — they don't produce records anyone wants to read. Mark them with queue_only: true and Sling will:

• run the endpoint and drain its records into the queue(s) it populates via processors,

• skip emitting a record stream (no target write, no row count),

• hide the endpoint from * wildcard discovery and from sling conns discover,

• still schedule the endpoint to run before any consumer that iterates over its queue (dependency order is auto-detected).

The endpoint is still selectable by explicit name when you need to debug it.

Example: Search → Detail

endpoints:
  # Producer: queue_only — no record output, just populates queue.imdb_ids
  search:
    queue_only: true
    request:
      url: "{state.base_url}/"
      method: GET
      parameters:
        apikey: "{env.OMDB_API_KEY}"
        s: "batman"
        type: "movie"
    response:
      records:
        jmespath: "Search"
      processors:
        - expression: "record.imdbID"
          output: "queue.imdb_ids"

  # Consumer: iterates the queue populated by `search`
  details:
    iterate:
      over: "queue.imdb_ids"
      into: "state.imdb_id"
      concurrency: 2
    request:
      url: "{state.base_url}/"
      method: GET
      parameters:
        apikey: "{env.OMDB_API_KEY}"
        i: "{state.imdb_id}"
    response:
      records:
        jmespath: "@"
        primary_key: ["imdbID"]

In a replication that targets *, Sling will run search first (filling queue.imdb_ids) and then details (writing one row per movie). search does not appear in conns discover output or in the wildcard endpoint list — list it explicitly by name to inspect it.

💡 Tip: queue_only replaces the older pattern of writing throwaway records from a producer endpoint just to satisfy the "every endpoint must produce a stream" assumption. Combined with auto-detected queues, your producer/consumer wiring stays declared in exactly one place: the endpoints themselves.

Queue Functions

Queues can be used with built-in functions for advanced processing:

Chunking Queue Data

Process queue items in batches for more efficient API calls:

endpoints:
  batch_customer_details:
    description: "Process customers in batches"
    
    iterate:
      # Process 50 customer IDs at a time
      over: "chunk(queue.customer_ids, 50)"
      into: "state.customer_batch"          # as an array
      concurrency: 3  # Process 3 batches concurrently
    
    request:
      url: "customers/batch"
      method: "POST"
      payload:
        ids: '{join(state.customer_batch, ","}'
    
    response:
      records:
        jmespath: "customers[]"
        primary_key: ["id"]

Real-World Example: Stripe API

This example from the Stripe API demonstrates a complete queue-based workflow:

endpoints:
  # Step 1: Collect customer IDs
  customer:
    description: "Retrieve list of customers"
    
    request:
      url: "customers"
    
    pagination:
      next_state:
        starting_after: '{response.records[-1].id}'
      stop_condition: "response.json.has_more == false"
    
    response:
      records:
        jmespath: "data[]"
        primary_key: ["id"]
      
      processors:
        # Send each customer ID to the queue
        - expression: "record.id"
          output: "queue.customer_ids"

  # Step 2: Get customer balance transactions
  customer_balance_transaction:
    description: "Retrieve customer balance transactions"
    
    iterate:
      over: "queue.customer_ids"
      into: "state.customer_id"
    
    request:
      url: "customers/{state.customer_id}/balance_transactions"
    
    response:
      records:
        jmespath: "data[]"
        primary_key: ["id"]
      
      processors:
        # Add customer_id to each transaction record
        - expression: "state.customer_id"
          output: "record.customer_id"

  # Step 3: Collect invoice IDs while getting invoice data
  invoice:
    description: "Retrieve invoices and queue IDs for line items"
    
    request:
      url: "invoices"
    
    pagination:
      next_state:
        starting_after: '{response.records[-1].id}'
      stop_condition: 'jmespath(response.json, "has_more") == false'
    
    response:
      records:
        jmespath: "data[]"
        primary_key: ["id"]
      
      processors:
        # Queue invoice IDs for line item extraction
        - expression: "record.id"
          output: "queue.invoice_ids"

  # Step 4: Get line items for each invoice
  invoice_line_item:
    description: "Retrieve invoice line items"
    
    iterate:
      over: "queue.invoice_ids"
      into: "state.invoice_id"
    
    request:
      url: "invoices/{state.invoice_id}/lines"
    
    response:
      records:
        jmespath: "data[]"
        primary_key: ["id"]
      
      processors:
        # Link line items back to their invoice
        - expression: "state.invoice_id"
          output: "record.invoice_id"

Queue Properties and Behavior

Queue Characteristics

Property	Description	Example
Temporary Storage	Queues are backed by temporary files	Automatically cleaned up after run
FIFO Order	Items are processed in first-in, first-out order	IDs processed in the order they were added
JSON Encoding	All data is JSON-encoded for safe storage	Handles strings, numbers, objects, arrays
Single Run Scope	Queues exist only within a single Sling execution	Cannot persist between separate runs

Queue Lifecycle

Direct Queue-to-Records Pattern

You can pipe queue data directly to records without making HTTP requests using the special syntax iterate.into: "response.records".

Basic Syntax

endpoints:
  deduplicate_customers:
    description: "Remove duplicate customer IDs"

    iterate:
      over: "queue.raw_customer_ids"
      into: "response.records"  # Special: Direct to records, no HTTP call

    # No request block needed!

    response:
      records:
        primary_key: ["customer_id"]  # Deduplicate

      processors:
        - expression: "record"
          output: "queue.clean_customer_ids"

When to Use

• Deduplicate queue items before further processing

• Enrich queue data with state variables

• Transform queue structure

• Export queue contents as a separate dataset

Example: Deduplication Workflow

endpoints:
  # Step 1: Collect IDs (may have duplicates)
  list_orders:
    request:
      url: "{state.base_url}/orders"
    response:
      processors:
        - expression: "record.customer_id"
          output: "queue.raw_customer_ids"

  # Step 2: Deduplicate using direct queue-to-records
  deduplicate_customers:
    iterate:
      over: "queue.raw_customer_ids"
      into: "response.records"
    response:
      records:
        primary_key: ["customer_id"]
      processors:
        - expression: "record"
          output: "queue.clean_customer_ids"

  # Step 3: Use clean IDs
  customer_details:
    iterate:
      over: "queue.clean_customer_ids"
      into: "state.customer_id"
    request:
      url: "{state.base_url}/customers/{state.customer_id}"

Data Type Handling

Scalar values (strings/numbers) are wrapped: "user123" → {"value": "user123"}

processors:
  - expression: "record.value"  # Access via .value
    output: "record.user_id"

Object values are used as-is: {"id": 1, "name": "Alice"} → same structure

processors:
  - expression: "record.id"  # Access fields directly
    output: "record.user_id"

Limitations

• No HTTP response data (response.status, response.headers unavailable)

• Cannot use pagination or response rules

• Don't define a request block (it will be ignored)

💡 Tip: This pattern is much faster than unnecessary HTTP requests for queue transformation steps.

Advanced Queue Patterns

Pattern 1: Multi-Level Hierarchies

Process nested data structures with multiple queue levels:

endpoints:
  accounts:
    response:
      # Level 1: Get accounts
      processors:
        - expression: "record.id"
          output: "queue.account_ids"

  customers:
    # Level 2: Get customers for each account
    iterate:
      over: "queue.account_ids"
      into: "state.account_id"
    response:
      processors:
        - expression: "record.id"
          output: "queue.customer_ids"

  subscriptions:
    # Level 3: Get subscriptions for each customer
    iterate:
      over: "queue.customer_ids"
      into: "state.customer_id"
    response:
      processors:
        - expression: "record.id"
          output: "queue.subscription_ids"

Pattern 2: Conditional Queue Population

Only queue certain items based on conditions:

processors:
  # Only queue active customers
  - expression: >
      if(record.status == "active", record.id, null)
    output: "queue.active_customer_ids"
  
  # Queue high-value customers for special processing
  - expression: >
      if(record.total_spent > 10000, record.id, null)
    output: "queue.vip_customer_ids"

Pattern 3: Queue Transformation

Transform data before queuing:

processors:
  # Create composite keys for the queue
  - expression: >
      {
        "customer_id": record.id,
        "type": record.customer_type,
        "priority": if(record.is_vip, "high", "normal")
      }
    output: "queue.enriched_customers"

Queue Best Practices

Performance Optimization

iterate:
  over: "chunk(queue.large_dataset, 100)"  # Process in batches
  into: "state.batch"
  concurrency: 5  # Parallel processing

Error Handling with Queues

rules:
  # Continue processing other queue items even if one fails
  - action: "continue"
    condition: "response.status == 404"
    message: "Item not found, skipping"
  
  # Retry transient errors
  - action: "retry"
    condition: "response.status >= 500"
    max_attempts: 3

💡 Tip: Use descriptive queue names that clearly indicate their purpose (e.g., customer_ids, pending_order_ids, failed_payment_ids).

⚠️ Warning: Queues consume disk space proportional to the number of items. For very large datasets (millions of items), monitor available disk space.

📝 Note: Queue items are automatically JSON-encoded, so complex objects, arrays, and special characters are handled safely.