Job Queues

When processing invoices at scale, synchronous API calls aren't always practical. This guide covers best practices for implementing job queues and async processing patterns with GoRoute.

Why Use Job Queues?

Challenge	Solution
Rate limits on API calls	Queue and throttle requests
Network timeouts	Retry failed jobs automatically
Batch processing thousands of invoices	Process in parallel with workers
System outages	Persist jobs and resume later
Audit trail	Track every job's status

Architecture Pattern

┌─────────────┐     ┌─────────────┐     ┌─────────────┐
│  Your App   │────▶│ Job Queue   │────▶│  Workers    │
│  (Producer) │     │ (Redis/SQS) │     │ (Consumers) │
└─────────────┘     └─────────────┘     └──────┬──────┘
                                               │
                                               ▼
                                        ┌─────────────┐
                                        │ GoRoute API │
                                        └─────────────┘

Implementation Examples

Python with Celery

# tasks.py
import os

import requests
from celery import Celery

app = Celery('invoice_tasks', broker=os.environ['REDIS_URL'])

DOCUMENTS_URL = 'https://app.goroute.ai/peppol-api/api/v1/documents'
API_KEY = os.environ['GOROUTE_API_KEY']


@app.task(
    bind=True,
    max_retries=5,
    default_retry_delay=60,
    autoretry_for=(requests.RequestException,),
    retry_backoff=True
)
def send_invoice(self, invoice_xml: str, receiver_scheme: str,
                 receiver_id: str, metadata: dict):
    """Send an invoice with POST /api/v1/documents, with automatic retries."""
    response = requests.post(
        DOCUMENTS_URL,
        headers={
            'X-API-Key': API_KEY,
            'Content-Type': 'application/json',
            # Stable per invoice, so a retry of this task cannot double-send.
            # The body must not change between retries either — see the
            # caution below.
            'Idempotency-Key': f"invoice-{metadata['invoice_id']}",
        },
        json={
            'receiver_scheme': receiver_scheme,
            'receiver_id': receiver_id,
            'document': invoice_xml,
            'metadata': metadata,
        },
        timeout=30,
    )

    # 202 Accepted - queued for delivery
    if response.status_code == 202:
        result = response.json()
        store_result(metadata['invoice_id'], result)
        return {'status': 'success', 'transaction_id': result['transaction_id']}

    if response.status_code == 429:
        # Rate limited - wait exactly as long as the response asks
        retry_after = int(response.headers.get('Retry-After', 60))
        raise self.retry(countdown=retry_after)

    if response.status_code == 400:
        # Validation refusal - permanent. Retrying will not change the answer.
        # An over-long key (error_code INVALID_IDEMPOTENCY_KEY) lands here too.
        return {
            'status': 'failed',
            'error': response.text,
            'retryable': False,
        }

    if response.status_code == 422:
        # Also permanent - do not raise. The usual cause is error_code
        # IDEMPOTENCY_KEY_REUSED: this key already carried a *different*
        # request within the last 24 hours. Nothing was sent, and nothing
        # will be under this key, so retrying cannot help.
        body = response.json()
        return {
            'status': 'failed',
            'error': response.text,
            'retryable': False,
            # Present on IDEMPOTENCY_KEY_REUSED: the send that did happen
            # under this key. Reconcile against it before sending again.
            'existing_transaction_id': body.get('existing_transaction_id'),
        }

    # Anything else (5xx, network) - raise so Celery retries with backoff
    response.raise_for_status()


# Producer code
def queue_invoices(invoices: list):
    """Queue multiple invoices for async processing."""
    for invoice in invoices:
        send_invoice.delay(
            invoice_xml=invoice.to_xml(),
            receiver_scheme=invoice.receiver_scheme,
            receiver_id=invoice.receiver_id,
            metadata={
                'invoice_id': invoice.id,
                'customer_id': invoice.customer_id
            }
        )

The send is a JSON call: the UBL document travels as a string in the document field, and a successful send answers 202 Accepted. See Sending an Invoice for the full request and response.

A stable key only works if the body stays the same

A reused Idempotency-Key protects you only as long as you send the same request under it. Send a different request under a key you have already used, within 24 hours, and GoRoute refuses it with 422 and error_code IDEMPOTENCY_KEY_REUSED. Nothing is sent, and a retry loop that regenerates the document every time will fail on its second attempt, every time.

All three examples on this page get this right on purpose: the worker receives invoice_xml as job data and sends exactly those bytes on every attempt. The producer builds the XML once, before the job is queued. Keep it that way. Move document generation inside the worker and each retry sends a new issue date, a recalculated total or a re-serialised XML — a different request under the same key, and a hard failure.

All three workers therefore handle 422 in a branch of its own, next to the 400 branch: they record the failure and return, and they never throw or re-raise, because a queue that retries a 422 will burn every attempt it has for a refusal that can never clear. The 422 body carries existing_transaction_id when the cause is IDEMPOTENCY_KEY_REUSED — that is the send that did happen under this key, so reconcile against it rather than sending blind.

If a document genuinely changes and you need to send the new version, give it a key of its own rather than reusing the old one. See Idempotency for both refusals and for the 24-hour window.

Node.js with BullMQ

// queue.ts
import { Queue, Worker } from 'bullmq';
import Redis from 'ioredis';

const connection = new Redis(process.env.REDIS_URL);

const GOROUTE_DOCUMENTS_URL =
  'https://app.goroute.ai/peppol-api/api/v1/documents';

// Create the queue
export const invoiceQueue = new Queue('invoices', { connection });

// Create the worker
const worker = new Worker('invoices', async (job) => {
  const { invoiceXml, receiverScheme, receiverId, metadata } = job.data;

  const headers = new Headers();
  headers.set('X-API-Key', process.env.GOROUTE_API_KEY);
  headers.set('Content-Type', 'application/json');
  // Stable per invoice, so a retry of this job cannot double-send. The body
  // must not change between retries either — see the caution above.
  headers.set('Idempotency-Key', 'invoice-' + metadata.invoiceId);

  const body = JSON.stringify({
    receiver_scheme: receiverScheme,
    receiver_id: receiverId,
    document: invoiceXml,
    metadata,
  });

  const init: RequestInit = { headers, body };
  init.method = 'POST';

  const response = await fetch(GOROUTE_DOCUMENTS_URL, init);

  // 202 Accepted - queued for delivery
  if (response.status === 202) {
    const result = await response.json();
    return { status: 'success', transactionId: result.transaction_id };
  }

  if (response.status === 429) {
    // Rate limited - throw so BullMQ retries with backoff
    throw new Error('Rate limited, retry-after: ' + response.headers.get('Retry-After'));
  }

  if (response.status === 400) {
    // Validation error - permanent, do not retry. An over-long key
    // (error_code INVALID_IDEMPOTENCY_KEY) lands here too.
    const detail = await response.text();
    return { status: 'failed', error: detail, retryable: false };
  }

  if (response.status === 422) {
    // Also permanent - do not throw. The usual cause is error_code
    // IDEMPOTENCY_KEY_REUSED: this key already carried a different request
    // within the last 24 hours. Nothing was sent, and nothing will be under
    // this key, so retrying cannot help.
    const body = await response.json();
    return {
      status: 'failed',
      error: JSON.stringify(body),
      retryable: false,
      // Present on IDEMPOTENCY_KEY_REUSED - the send that did happen under
      // this key. Reconcile against it before sending again.
      existingTransactionId: body.existing_transaction_id,
    };
  }

  // Anything else (5xx, network) - throw so BullMQ retries
  throw new Error('Send failed: ' + response.status);
}, {
  connection,
  concurrency: 10,  // Process 10 jobs in parallel
  limiter: {
    max: 100,        // Max 100 jobs
    duration: 60000  // Per minute
  }
});

// Handle events
worker.on('completed', (job, result) => {
  console.log('Invoice sent:', job.data.metadata.invoiceId, result);
});

worker.on('failed', (job, error) => {
  console.error('Invoice failed:', job.data.metadata.invoiceId, error);
});

C# with Hangfire

// InvoiceJobs.cs
using System.Net;
using System.Net.Http;
using System.Net.Http.Json;
using System.Text.Json;
using Hangfire;

public class InvoiceJobs
{
    private readonly HttpClient _http;
    private readonly ILogger<InvoiceJobs> _logger;

    // Registered via IHttpClientFactory with BaseAddress
    // https://app.goroute.ai/peppol-api/ and the X-API-Key default header.
    public InvoiceJobs(HttpClient http, ILogger<InvoiceJobs> logger)
    {
        _http = http;
        _logger = logger;
    }

    [AutomaticRetry(Attempts = 5, DelaysInSeconds = new[] { 60, 300, 900, 3600, 7200 })]
    public async Task<SendResult> SendInvoiceAsync(string invoiceXml, InvoiceMetadata metadata)
    {
        var payload = new Dictionary<string, object>();
        payload["receiver_scheme"] = metadata.ReceiverScheme;
        payload["receiver_id"] = metadata.ReceiverId;
        payload["document"] = invoiceXml;

        using var request = new HttpRequestMessage(
            HttpMethod.Post, "api/v1/documents");
        request.Content = JsonContent.Create(payload);
        // Stable per invoice, so a Hangfire retry cannot double-send. The body
        // must not change between retries either — see the caution above.
        request.Headers.Add("Idempotency-Key", "invoice-" + metadata.InvoiceId);

        var response = await _http.SendAsync(request);

        // 202 Accepted - queued for delivery
        if (response.StatusCode == HttpStatusCode.Accepted)
        {
            using var doc = JsonDocument.Parse(
                await response.Content.ReadAsStringAsync());

            var transactionId = doc.RootElement
                .GetProperty("transaction_id").GetString();

            _logger.LogInformation(
                "Invoice {InvoiceId} sent successfully. Transaction: {TransactionId}",
                metadata.InvoiceId, transactionId);

            return new SendResult { Success = true, TransactionId = transactionId };
        }

        if (response.StatusCode == HttpStatusCode.TooManyRequests)
        {
            _logger.LogWarning("Rate limited. Retry after {Seconds}s", response.Headers.RetryAfter?.Delta?.TotalSeconds);
            throw new HttpRequestException("Rate limited"); // Retried by Hangfire
        }

        var error = await response.Content.ReadAsStringAsync();

        if (response.StatusCode == HttpStatusCode.BadRequest)
        {
            // Permanent. An over-long key (error_code INVALID_IDEMPOTENCY_KEY)
            // lands here too.
            _logger.LogError("Validation failed for {InvoiceId}: {Error}", 
                metadata.InvoiceId, error);
            
            // Return failure without throwing - don't retry
            return new SendResult { Success = false, Error = error };
        }

        if (response.StatusCode == HttpStatusCode.UnprocessableEntity)
        {
            // Also permanent - do not throw. The usual cause is error_code
            // IDEMPOTENCY_KEY_REUSED: this key already carried a different
            // request within the last 24 hours. Nothing was sent, and nothing
            // will be under this key, so retrying cannot help.
            using var errorDoc = JsonDocument.Parse(error);

            if (errorDoc.RootElement.TryGetProperty(
                    "existing_transaction_id", out var existingTransactionId))
            {
                // The send that did happen under this key. Reconcile against
                // it before sending again.
                _logger.LogError(
                    "Idempotency key reused for {InvoiceId}. Nothing was sent; the earlier send is transaction {TransactionId}",
                    metadata.InvoiceId, existingTransactionId.GetString());
            }
            else
            {
                _logger.LogError("Request refused for {InvoiceId}: {Error}",
                    metadata.InvoiceId, error);
            }

            // Return failure without throwing - don't retry
            return new SendResult { Success = false, Error = error };
        }

        // Anything else (5xx, network) - throw so Hangfire retries
        throw new HttpRequestException("Send failed: " + (int)response.StatusCode);
    }
}

// Usage
public class InvoiceService
{
    public void QueueInvoice(Invoice invoice)
    {
        BackgroundJob.Enqueue<InvoiceJobs>(
            x => x.SendInvoiceAsync(invoice.ToXml(), invoice.Metadata));
    }

    public void QueueBatch(IEnumerable<Invoice> invoices)
    {
        foreach (var invoice in invoices)
        {
            BackgroundJob.Enqueue<InvoiceJobs>(
                x => x.SendInvoiceAsync(invoice.ToXml(), invoice.Metadata));
        }
    }
}

Queue Design Patterns

Priority Queues

Process urgent invoices first:

# High priority for due-date invoices
@app.task(queue='high_priority')
def send_urgent_invoice(invoice_xml, metadata):
    return send_invoice(invoice_xml, metadata)

@app.task(queue='default')
def send_normal_invoice(invoice_xml, metadata):
    return send_invoice(invoice_xml, metadata)

# Queue based on urgency
def queue_invoice(invoice):
    if invoice.days_until_due < 3:
        send_urgent_invoice.delay(invoice.xml, invoice.metadata)
    else:
        send_normal_invoice.delay(invoice.xml, invoice.metadata)

Dead Letter Queues

Handle permanently failed jobs:

// Configure DLQ for failed jobs
const invoiceQueue = new Queue('invoices', {
  connection,
  defaultJobOptions: {
    attempts: 5,
    backoff: {
      type: 'exponential',
      delay: 60000
    },
    removeOnComplete: 100,
    removeOnFail: false  // Keep failed jobs for analysis
  }
});

// Process DLQ periodically
async function processDLQ() {
  const failedJobs = await invoiceQueue.getFailed(0, 100);
  
  for (const job of failedJobs) {
    // Notify customer service
    await notifyFailure(job.data.metadata);
    
    // Move to manual review queue
    await manualReviewQueue.add('review', {
      originalJob: job.data,
      error: job.failedReason,
      attempts: job.attemptsMade
    });
    
    // Remove from failed queue
    await job.remove();
  }
}

Batch Processing

Group invoices for efficient processing:

from celery import group

def send_invoice_batch(invoices: list, batch_size: int = 50):
    """Send invoices in batches with controlled parallelism."""
    
    # Create groups of tasks
    for i in range(0, len(invoices), batch_size):
        batch = invoices[i:i + batch_size]
        
        # Create a group of parallel tasks
        job_group = group(
            send_invoice.s(inv.xml, inv.metadata)
            for inv in batch
        )
        
        # Execute batch and wait for results
        result = job_group.apply_async()
        
        # Wait for batch to complete before next batch
        # This prevents overwhelming the API
        results = result.get(timeout=300)
        
        # Log batch results
        success = sum(1 for r in results if r['status'] == 'success')
        logger.info(f"Batch {i//batch_size + 1}: {success}/{len(batch)} successful")

Rate Limiting

Respecting API Limits

GoRoute has rate limits to ensure fair usage:

Tier	Requests/min	Concurrent
Standard	100	10
Professional	500	50
Enterprise	Custom	Custom

Implementing Rate Limiting

from ratelimit import limits, sleep_and_retry

# Limit to 100 calls per minute
@sleep_and_retry
@limits(calls=100, period=60)
def rate_limited_send(invoice_xml, metadata):
    return client.documents.send(invoice_xml, metadata)

@app.task
def send_invoice(invoice_xml, metadata):
    return rate_limited_send(invoice_xml, metadata)

Token Bucket Pattern

import Bottleneck from 'bottleneck';

const limiter = new Bottleneck({
  reservoir: 100,      // Initial tokens
  reservoirRefreshAmount: 100,
  reservoirRefreshInterval: 60 * 1000,  // Refill every minute
  maxConcurrent: 10,
  minTime: 100  // Min 100ms between requests
});

async function sendWithRateLimit(invoice: Invoice) {
  return limiter.schedule(() => 
    client.documents.send(invoice)
  );
}

Monitoring

Job Metrics

Track queue health:

from prometheus_client import Counter, Histogram, Gauge

jobs_total = Counter('invoice_jobs_total', 'Total jobs processed', ['status'])
job_duration = Histogram('invoice_job_duration_seconds', 'Job processing time')
queue_depth = Gauge('invoice_queue_depth', 'Current queue depth')

@app.task
def send_invoice(invoice_xml, metadata):
    with job_duration.time():
        try:
            result = client.documents.send(invoice_xml, metadata)
            jobs_total.labels(status='success').inc()
            return result
        except Exception as e:
            jobs_total.labels(status='error').inc()
            raise

Alerting

Set up alerts for queue issues:

# prometheus-alerts.yaml
groups:
  - name: invoice_queue
    rules:
      - alert: HighQueueDepth
        expr: invoice_queue_depth > 1000
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "Invoice queue depth is high"
          
      - alert: HighFailureRate
        expr: rate(invoice_jobs_total{status="error"}[5m]) > 0.1
        for: 5m
        labels:
          severity: critical
        annotations:
          summary: "High invoice job failure rate"

Best Practices

1. Idempotency

Ensure jobs can be safely retried:

@app.task
def send_invoice(invoice_xml, metadata):
    # Check if already sent (idempotency)
    existing = get_transaction(metadata['invoice_id'])
    if existing:
        return {'status': 'already_sent', 'transaction_id': existing.id}
    
    result = client.documents.send(invoice_xml, metadata)
    
    # Store result atomically
    store_transaction(metadata['invoice_id'], result)
    
    return result

2. Graceful Shutdown

Handle shutdown without losing jobs:

import signal

def shutdown_handler(signum, frame):
    logger.info("Shutting down gracefully...")
    worker.shutdown()  # Stop accepting new jobs
    worker.wait()      # Wait for current jobs to finish
    sys.exit(0)

signal.signal(signal.SIGTERM, shutdown_handler)

3. Job Serialization

Keep job payloads small:

# Good - store reference
@app.task
def send_invoice(invoice_id: str):
    invoice = Invoice.get(invoice_id)  # Fetch from DB
    return client.documents.send(invoice.xml)

# Avoid - large payloads in queue
@app.task
def send_invoice(invoice_xml: str):  # Could be 100KB+
    return client.documents.send(invoice_xml)

4. Correlation IDs

Track jobs across systems:

import uuid

def queue_invoice(invoice):
    correlation_id = str(uuid.uuid4())
    
    send_invoice.apply_async(
        args=[invoice.xml],
        kwargs={'correlation_id': correlation_id},
        task_id=correlation_id  # Use same ID for tracing
    )
    
    return correlation_id

Next Steps

• Production Checklist →
• Error Recovery →
• Webhooks →