Queues - Hyrex

Queues allow you to organize and route tasks to specific worker pools, enabling efficient resource allocation and task prioritization.

Queue Configuration

Basic Queue Usage

Assign tasks to queues using the queue configuration:

import { HyrexRegistry } from '@hyrex/hyrex';

const hy = new HyrexRegistry();

// Simple string queue
const emailTask = hy.task({
    name: 'sendEmail',
    config: {
        queue: 'emails'
    },
    func: async (data) => {
        // Task will be processed by workers listening to 'emails' queue
        return { sent: true };
    }
});

// Default queue (if not specified)
const defaultTask = hy.task({
    name: 'defaultTask',
    func: async (data) => {
        // Goes to 'default' queue
        return { processed: true };
    }
});

Queue with Concurrency Limits

Control concurrent task execution with queue objects:

import { HyrexQueue } from '@hyrex/hyrex';

// Define queue with concurrency limit
const cpuIntensiveTask = hy.task({
    name: 'cpuIntensive',
    config: {
        queue: {
            name: 'cpu-intensive',
            concurrencyLimit: 2  // Max 2 concurrent tasks
        }
    },
    func: async (data) => {
        // Heavy computation
        return { result: 'computed' };
    }
});

// Or create HyrexQueue instance
const gpuQueue = new HyrexQueue({
    name: 'gpu-processing',
    concurrencyLimit: 1  // Single GPU task at a time
});

const mlTask = hy.task({
    name: 'trainModel',
    config: {
        queue: gpuQueue
    },
    func: async (data) => {
        // GPU-intensive ML training
        return { model: 'trained' };
    }
});

Worker Queue Selection

Workers can subscribe to specific queues or patterns:

# Listen to single queue
hyrex run-worker hyrex-app.ts --queue emails

# Listen to all queues (default)
hyrex run-worker hyrex-app.ts --queue "*"

# Use glob patterns
hyrex run-worker hyrex-app.ts --queue "email-*"
hyrex run-worker hyrex-app.ts --queue "priority-?"
hyrex run-worker hyrex-app.ts --queue "{emails,sms,push}"

Glob Pattern Support

Queue patterns support standard glob syntax:

* - Matches any characters
? - Matches single character
[abc] - Matches any character in brackets
{a,b,c} - Matches any of the comma-separated patterns

Dynamic Queue Assignment

Override queue at send time:

const flexibleTask = hy.task({
    name: 'flexibleTask',
    config: {
        queue: 'default'
    },
    func: async (data) => {
        return { processed: true };
    }
});

// Send to default queue
const taskId1 = await flexibleTask.send({ data: 'normal' });

// Override to high-priority queue
const taskId2 = await flexibleTask.withConfig({
    queue: 'high-priority'
}).send({ data: 'urgent' });

Queue Organization Patterns

By Priority

// High priority queue
const urgentTask = hy.task({
    name: 'urgentTask',
    config: {
        queue: 'high-priority',
        priority: 9  // Higher task priority within queue
    },
    func: async (data) => {
        return { handled: 'immediately' };
    }
});

// Low priority queue
const batchTask = hy.task({
    name: 'batchTask',
    config: {
        queue: 'low-priority',
        priority: 2
    },
    func: async (data) => {
        return { processed: 'eventually' };
    }
});

By Resource Type

// CPU-intensive tasks
const analyzeData = hy.task({
    name: 'analyzeData',
    config: {
        queue: 'cpu-intensive',
        timeoutSeconds: 600
    },
    func: async (data) => {
        // Complex calculations
        return { analyzed: true };
    }
});

// I/O-bound tasks
const fetchExternalData = hy.task({
    name: 'fetchExternalData',
    config: {
        queue: 'io-bound',
        maxRetries: 5
    },
    func: async (data) => {
        // External API calls
        return { fetched: true };
    }
});

// Memory-intensive tasks
const processLargeFile = hy.task({
    name: 'processLargeFile',
    config: {
        queue: {
            name: 'memory-intensive',
            concurrencyLimit: 1  // Prevent OOM
        }
    },
    func: async (data) => {
        // Large file processing
        return { processed: true };
    }
});

By Business Domain

// Email service queue
const emailQueue = 'email-service';

const sendWelcomeEmail = hy.task({
    name: 'sendWelcomeEmail',
    config: { queue: emailQueue },
    func: async (data) => { /* ... */ }
});

const sendPasswordReset = hy.task({
    name: 'sendPasswordReset',
    config: { queue: emailQueue },
    func: async (data) => { /* ... */ }
});

// Payment processing queue
const paymentQueue = 'payment-processing';

const chargeCard = hy.task({
    name: 'chargeCard',
    config: { 
        queue: paymentQueue,
        idempotencyKey: 'charge-123'  // Prevent duplicate charges
    },
    func: async (data) => { /* ... */ }
});

const refundPayment = hy.task({
    name: 'refundPayment',
    config: { queue: paymentQueue },
    func: async (data) => { /* ... */ }
});

Worker Deployment Strategies

Dedicated Workers by Queue

# Email worker (high concurrency)
hyrex run-worker hyrex-app.ts 10 --queue emails

# Payment worker (limited concurrency for safety)
hyrex run-worker hyrex-app.ts 2 --queue payment-processing

# Background jobs worker
hyrex run-worker hyrex-app.ts 5 --queue "batch-*"

Resource-Based Workers

# CPU-optimized instance
hyrex run-worker hyrex-app.ts 8 --queue cpu-intensive

# Memory-optimized instance
hyrex run-worker hyrex-app.ts 2 --queue memory-intensive

# GPU instance
hyrex run-worker hyrex-app.ts 1 --queue gpu-processing

Queue Monitoring

Monitor queue depths and worker health through Hyrex Studio to:

View pending tasks per queue
Monitor worker connections per queue
Track queue processing rates
Identify bottlenecks

Best Practices

Use Descriptive Queue Names

// Good
config: { queue: 'email-notifications' }
config: { queue: 'payment-processing' }

// Avoid
config: { queue: 'q1' }
config: { queue: 'misc' }

Set Appropriate Concurrency Limits

// Limit concurrent database writes
config: {
    queue: {
        name: 'database-writes',
        concurrencyLimit: 5  // Prevent connection exhaustion
    }
}

Organize Workers Logically

# Separate workers for different concerns
hyrex run-worker app.ts 10 --queue "user-facing-*"
hyrex run-worker app.ts 5 --queue "background-*"
hyrex run-worker app.ts 1 --queue "scheduled-reports"

Use Queue Patterns for Flexibility

// Tasks can go to region-specific queues
const regionalTask = hy.task({
    name: 'processRegional',
    config: { queue: `process-${region}` },
    func: async (data) => { /* ... */ }
});

// Workers can handle multiple regions
// hyrex run-worker app.ts --queue "process-*"

Next Steps

Tasks

Configure tasks for queues

Monitoring

Monitor queue performance

Retries

Handle failures in queues

​Queue Configuration

​Basic Queue Usage

​Queue with Concurrency Limits

​Worker Queue Selection

​Glob Pattern Support

​Dynamic Queue Assignment

​Queue Organization Patterns

​By Priority

​By Resource Type

​By Business Domain

​Worker Deployment Strategies

​Dedicated Workers by Queue

​Resource-Based Workers

​Queue Monitoring

​Best Practices

​Next Steps