Glide Mq

glide-mq

High-performance AI-native message queue for Node.js on Valkey/Redis Streams with a Rust NAPI core.

Quick Start

import { Queue, Worker } from 'glide-mq';

const connection = { addresses: [{ host: 'localhost', port: 6379 }] };

const queue = new Queue('tasks', { connection });
await queue.add('send-email', { to: 'user@example.com', subject: 'Hello' });

const worker = new Worker(
  'tasks',
  async (job) => {
    console.log(`Processing ${job.name}:`, job.data);
    return { sent: true };
  },
  { connection, concurrency: 10 },
);

worker.on('completed', (job) => console.log(`Done: ${job.id}`));
worker.on('failed', (job, err) => console.error(`Failed: ${job.id}`, err.message));

When to Apply

Use this skill when:

• Creating or configuring queues, workers, or producers
• Adding jobs (single, bulk, delayed, priority)
• Setting up retries, backoff, or dead-letter queues
• Building job workflows (parent-child, DAGs, chains)
• Implementing fan-out broadcast patterns
• Configuring cron/interval schedulers
• Setting up connection options (TLS, IAM, AZ-affinity)
• Working with batch processing or rate limiting
• Tracking AI/LLM usage (tokens, cost, model) per job or flow
• Streaming LLM output tokens in real-time
• Implementing human-in-the-loop approval with suspend/resume
• Setting budget caps (tokens, cost) on workflow flows
• Configuring fallback chains for model/provider failover
• Dual-axis rate limiting (RPM + TPM) for LLM API compliance
• Aggregating rolling usage/cost summaries across queues
• Searching jobs by vector similarity (KNN) with Valkey Search
• Exposing queues or broadcasts over the HTTP proxy, including SSE endpoints
• Integrating with frameworks (Hono, Fastify, NestJS, Hapi)
• Deploying in serverless environments (Lambda, Vercel Edge)

Core API by Priority

Key Patterns

Delayed & Priority Jobs

// Delayed: run after 5 minutes
await queue.add('reminder', data, { delay: 300_000 });

// Priority: lower number = higher priority (default: 0)
await queue.add('urgent', data, { priority: 0 });
await queue.add('low-priority', data, { priority: 10 });

// Retries with exponential backoff
await queue.add('webhook', data, {
  attempts: 5,
  backoff: { type: 'exponential', delay: 1000 },
});

Bulk Ingestion (10,000 jobs in ~350ms)

const jobs = items.map((item) => ({
  name: 'process',
  data: item,
  opts: { jobId: `item-${item.id}` },
}));
await queue.addBulk(jobs);

Batch Worker (Process Multiple Jobs at Once)

const worker = new Worker(
  'analytics',
  async (jobs) => {
    // jobs is Job[] when batch is enabled
    await db.insertMany(
      'events',
      jobs.map((j) => j.data),
    );
  },
  {
    connection,
    batch: { size: 50, timeout: 5000 },
  },
);

Batch mode is composable with priority and lifo: true jobs - list-popped jobs are dispatched into the same batch processor (chunked by batch.size).

Request-Reply (addAndWait)

const result = await queue.addAndWait(
  'compute',
  { input: 42 },
  {
    waitTimeout: 30_000,
  },
);
console.log(result); // processor return value

Serverless Producer (No EventEmitter Overhead)

import { Producer } from 'glide-mq';
const producer = new Producer('queue', { connection });
await producer.add('job-name', data);
await producer.close();

Graceful Shutdown

import { gracefulShutdown } from 'glide-mq';

// Registers SIGTERM/SIGINT handlers and returns a handle.
// await blocks until a signal fires - use as last line of your program.
const handle = gracefulShutdown([worker1, worker2, queue, events]);

// For programmatic shutdown (e.g., in tests):
await handle.shutdown();

// To remove signal handlers without closing:
handle.dispose();

Testing Without Valkey

import { TestQueue, TestWorker } from 'glide-mq/testing';
const queue = new TestQueue('tasks');
await queue.add('test-job', { key: 'value' });
const worker = new TestWorker(queue, processor);
await worker.run();

Problem-to-Reference Mapping

Critical Notes

• Node.js 20+ and Valkey 7.0+ (or Redis 7.0+) required
• At-least-once delivery - make processors idempotent
• Priority: lower number = higher priority (0 is default, highest)
• Cluster-native - hash-tagged keys (glide:{queueName}:*) work out of the box
• All queue logic runs as a single Valkey Server Function (FCALL) - 1 round-trip per job
• Connection format uses addresses: [{ host, port }] array, NOT { host, port } object
• Never use `customCommand` - use typed API methods with dummy keys for cluster routing

Done When

• npm test or the project-equivalent test command passes
• await queue.getJobCounts() matches the expected queue state
• no jobs are left unexpectedly stuck in active
• any QueueEvents or SSE behavior touched by the change has been smoke-tested
• temporary queues, workers, and listeners are closed cleanly

Full Documentation

https://www.glidemq.dev/

AI-Native Primitives Reference

glide-mq provides 7 AI-native primitives designed for LLM orchestration pipelines.

1. Usage Metadata (job.reportUsage)

Track model, tokens, cost, and latency per job.

const worker = new Worker('inference', async (job) => {
  const response = await openai.chat.completions.create({ ... });

  await job.reportUsage({
    model: 'gpt-5.4',
    provider: 'openai',
    tokens: {
      input: response.usage.prompt_tokens,
      output: response.usage.completion_tokens,
    },
    // totalTokens auto-computed as sum of all token categories if omitted
    costs: { total: 0.0032 },
    costUnit: 'usd',
    latencyMs: 1200,
    cached: false,
  });

  return response.choices[0].message.content;
}, { connection });

JobUsage Interface

interface JobUsage {
  model?: string;           // e.g. 'gpt-5.4', 'claude-sonnet-4-20250514'
  provider?: string;        // e.g. 'openai', 'anthropic'
  tokens?: Record<string, number>;   // e.g. { input: 500, output: 200, reasoning: 100 }
  totalTokens?: number;     // auto-computed as sum of tokens values if omitted
  costs?: Record<string, number>;    // e.g. { total: 0.003 } or { input: 0.001, output: 0.002 }
  totalCost?: number;       // auto-computed as sum of costs values if omitted
  costUnit?: string;        // e.g. 'usd', 'credits', 'ils' (informational)
  latencyMs?: number;       // inference latency (not queue wait)
  cached?: boolean;         // cache hit flag
}

• Calling reportUsage() multiple times overwrites previous values on that job.
• Token counts must not be negative (throws).
• Emits a 'usage' event on the events stream with the full usage object.
• Stored in the job hash as usage:model, usage:tokens (JSON), usage:costs (JSON), usage:totalTokens, usage:totalCost, usage:costUnit.
• Also updates rolling per-minute usage buckets used by queue.getUsageSummary().

Rolling Usage Summary (queue.getUsageSummary / Queue.getUsageSummary)

const summary = await queue.getUsageSummary({
  queues: ['inference', 'embeddings'],
  windowMs: 3_600_000,   // last hour
});

// {
//   totalTokens,
//   totalCost,
//   jobCount,
//   models: Record<string, number>,
//   perQueue: Record<string, { totalTokens, totalCost, jobCount, models }>
// }

Use Queue.getUsageSummary() when you want the same rollup without an existing queue instance. The HTTP proxy exposes the same aggregation at GET /usage/summary.

2. Token Streaming (job.stream / job.streamChunk / queue.readStream)

Emit and consume LLM output tokens in real-time via per-job Valkey Streams.

Producer Side (Worker)

const worker = new Worker('chat', async (job) => {
  const stream = await openai.chat.completions.create({ stream: true, ... });

  for await (const chunk of stream) {
    const token = chunk.choices[0]?.delta?.content;
    if (token) {
      await job.stream({ token, index: String(chunk.choices[0].index) });
    }
  }

  return { done: true };
}, { connection });

job.stream(chunk) appends a flat Record<string, string> to a per-job Valkey Stream via XADD. Returns the stream entry ID.

Convenience: job.streamChunk(type, content?)

Typed shorthand for streaming LLM chunks with a type field and optional content:

await job.streamChunk('reasoning', 'Let me think about this...');
await job.streamChunk('content', 'The answer is 42.');
await job.streamChunk('done');

Equivalent to job.stream({ type, content }) - useful for structured streaming with thinking models.

Consumer Side (Queue)

const entries = await queue.readStream(jobId);
// entries: { id: string; fields: Record<string, string> }[]

// Resume from last known position
const more = await queue.readStream(jobId, { lastId: entries.at(-1)?.id });

// Long-polling (blocks until new entries arrive)
const live = await queue.readStream(jobId, { lastId, block: 5000 });

ReadStreamOptions

interface ReadStreamOptions {
  lastId?: string;     // resume from this stream ID (exclusive)
  count?: number;      // max entries to return (default: 100)
  block?: number;      // XREAD BLOCK ms for long-polling (0 = non-blocking)
}

3. Suspend / Resume (Human-in-the-Loop)

Pause a job to wait for external approval, then resume with signals.

Suspending (Worker Side)

const worker = new Worker('content-review', async (job) => {
  // Check if this is a resume after suspension
  if (job.signals.length > 0) {
    const approval = job.signals.find(s => s.name === 'approve');
    if (approval) {
      return { published: true, approver: approval.data.approvedBy };
    }
    return { rejected: true };
  }

  // First run - generate content and suspend for review
  const content = await generateContent(job.data);
  await job.updateData({ ...job.data, generatedContent: content });

  await job.suspend({
    reason: 'Awaiting human review',
    timeout: 86_400_000,  // 24h timeout (0 = infinite, default)
  });
}, { connection });

job.suspend() throws SuspendError internally - no code after it executes. The job moves to 'suspended' state.

If timeout is set, glide-mq stores the deadline on the suspended sorted set and any live Queue or Worker runtime can fail expired suspended jobs with 'Suspend timeout exceeded'. This no longer depends on the original worker staying online, but it does require at least one glide-mq process to remain connected to the queue.

Resuming (Queue Side)

// Send a signal to resume the job
const resumed = await queue.signal(jobId, 'approve', { approvedBy: 'alice' });
// true if job was suspended and is now resumed, false otherwise

// Inspect suspension state
const info = await queue.getSuspendInfo(jobId);
// null if not suspended, otherwise:
// {
//   reason?: string,
//   suspendedAt: number (epoch ms),
//   timeout?: number (ms),
//   signals: SignalEntry[]
// }

SignalEntry

interface SignalEntry {
  name: string;        // signal name (e.g. 'approve', 'reject')
  data: any;           // arbitrary payload
  receivedAt: number;  // epoch ms
}

SuspendOptions

interface SuspendOptions {
  reason?: string;     // human-readable reason
  timeout?: number;    // ms, 0 = infinite (default)
}

4. Budget Middleware (Flow-Level Caps)

Cap total token usage and/or cost across all jobs in a flow. Supports per-category limits and weighted totals for thinking model budgets.

Setting Budget on a Flow

import { FlowProducer } from 'glide-mq';

const flow = new FlowProducer({ connection });
await flow.add(
  {
    name: 'research-report',
    queueName: 'ai',
    data: { topic: 'quantum computing' },
    children: [
      { name: 'search', queueName: 'ai', data: { query: 'latest papers' } },
      { name: 'summarize', queueName: 'ai', data: {} },
      { name: 'critique', queueName: 'ai', data: {} },
    ],
  },
  {
    budget: {
      maxTotalTokens: 50_000,
      maxTotalCost: 0.50,
      costUnit: 'usd',
      tokenWeights: { reasoning: 4, cachedInput: 0.25 },
      onExceeded: 'fail',    // 'fail' (default) or 'pause'
    },
  },
);

BudgetOptions

interface BudgetOptions {
  maxTotalTokens?: number;                // hard cap on weighted total tokens
  maxTokens?: Record<string, number>;     // per-category token caps (e.g. { input: 50000, reasoning: 5000 })
  tokenWeights?: Record<string, number>;  // weight multipliers for maxTotalTokens (unlisted = 1)
  maxTotalCost?: number;                  // hard cap on total cost
  maxCosts?: Record<string, number>;      // per-category cost caps
  costUnit?: string;                      // e.g. 'usd', 'credits', 'ils' (informational)
  onExceeded?: 'pause' | 'fail';         // default: 'fail'
}

Reading Budget State

const budget = await queue.getFlowBudget(parentJobId);
// null if no budget was set, otherwise:
// {
//   maxTotalTokens?: number,
//   maxTokens?: Record<string, number>,
//   tokenWeights?: Record<string, number>,
//   maxTotalCost?: number,
//   maxCosts?: Record<string, number>,
//   costUnit?: string,
//   usedTokens: number,
//   usedCost: number,
//   exceeded: boolean,
//   onExceeded: 'pause' | 'fail'
// }

Budget is enforced per flow by writing a budgetKey to every job hash in the tree.

5. Fallback Chains

Ordered list of model/provider alternatives tried on retryable failure.

Setting Fallbacks

await queue.add('inference', { prompt: 'Explain quantum entanglement' }, {
  attempts: 4,  // 1 original + 3 fallbacks
  fallbacks: [
    { model: 'gpt-5.4', provider: 'openai' },
    { model: 'claude-sonnet-4-20250514', provider: 'anthropic' },
    { model: 'llama-3-70b', provider: 'groq', metadata: { temperature: 0.7 } },
  ],
});

Reading Fallback State (Worker Side)

const worker = new Worker('inference', async (job) => {
  const fallback = job.currentFallback;
  // undefined on first attempt (original request)
  // { model: 'gpt-5.4', provider: 'openai' } on first fallback
  // { model: 'claude-sonnet-4-20250514', provider: 'anthropic' } on second, etc.

  const model = fallback?.model ?? job.data.defaultModel;
  const provider = fallback?.provider ?? job.data.defaultProvider;

  return await callLLM(provider, model, job.data.prompt);
}, { connection });

• job.fallbackIndex is 0 for the original request, 1+ for fallback entries.
• job.currentFallback returns fallbacks[fallbackIndex - 1] or undefined when index is 0.
• Each fallback entry has model (required), provider (optional), and metadata (optional).

6. Dual-Axis Rate Limiting (RPM + TPM)

Rate-limit workers by both requests-per-minute (RPM) and tokens-per-minute (TPM).

Configuration

const worker = new Worker('inference', processor, {
  connection,
  limiter: { max: 60, duration: 60_000 },          // RPM: 60 req/min
  tokenLimiter: {
    maxTokens: 100_000,
    duration: 60_000,
    scope: 'both',  // 'queue' | 'worker' | 'both' (default)
  },
});

TokenLimiter Options

interface TokenLimiter {
  maxTokens: number;        // max tokens per window
  duration: number;         // window duration in ms
  scope?: 'queue' | 'worker' | 'both';
  // 'queue': Valkey counter shared across all workers
  // 'worker': in-memory counter per worker instance
  // 'both': local check first, then Valkey (optimal, default)
}

Reporting Tokens

const worker = new Worker('inference', async (job) => {
  const result = await callLLM(job.data);

  // Option 1: report tokens directly for TPM tracking
  await job.reportTokens(result.totalTokens);

  // Option 2: reportUsage auto-extracts totalTokens for TPM
  await job.reportUsage({
    model: 'gpt-5.4',
    tokens: { input: result.promptTokens, output: result.completionTokens },
  });

  return result;
}, { connection, tokenLimiter: { maxTokens: 100_000, duration: 60_000 } });

Worker pauses fetching when either RPM or TPM limit is exceeded.

7. Flow Usage Aggregation (getFlowUsage)

Aggregate AI usage metadata across all jobs in a flow tree.

const usage = await queue.getFlowUsage(parentJobId);
// {
//   tokens: Record<string, number>,    // aggregated per-category tokens (e.g. { input: 2500, output: 1200 })
//   totalTokens: number,               // sum of all token categories
//   costs: Record<string, number>,     // aggregated per-category costs
//   totalCost: number,                 // sum of all cost categories
//   costUnit?: string,                 // unit from the first job that reported one
//   jobCount: number,
//   models: Record<string, number>     // model name -> call count
// }

Walks the parent and all children via the deps set. Useful for cost reporting, billing, and observability dashboards.

Gotchas

• job.suspend() and job.moveToWaitingChildren() both throw internally - no code after them executes.
• job.reportUsage() and job.reportTokens() reject negative values.
• reportUsage() overwrites previous usage data on the same job.
• getUsageSummary() reads rolling buckets, not job hashes, so it is cheap for queue-wide summaries but not a replacement for per-job detail.
• reportTokens() overwrites the previous value - it does not accumulate.
• Budget enforcement happens at the flow level, not per-job. Individual jobs report usage; the budget key tracks aggregates.
• Fallback chains require attempts >= fallbacks.length + 1 (original + N fallbacks).
• queue.signal() returns false if the job is not in suspended state.
• readStream() with block > 0 uses XREAD BLOCK (a blocking Valkey call) - do not use on a shared client that serves other queries.

Broadcast Reference

Overview

Broadcast is pub/sub fan-out. Unlike Queue (point-to-point), every message is delivered to all subscribers.

Broadcast Constructor

import { Broadcast, BroadcastWorker } from 'glide-mq';

const broadcast = new Broadcast('events', {
  connection: ConnectionOptions,
  maxMessages?: number,  // retain at most N messages in the stream
});

Publishing

// publish(subject, data, opts?) - subject is the first arg
await broadcast.publish('orders', { event: 'order.placed', orderId: 42 });

// With dotted subjects (for subject filtering)
await broadcast.publish('orders.created', { orderId: 42 });
await broadcast.publish('inventory.low', { sku: 'ABC', qty: 0 });

await broadcast.close();

BroadcastWorker Constructor

const worker = new BroadcastWorker(
  'events',                         // broadcast name
  async (job) => {                  // processor
    console.log(job.name, job.data);
  },
  {
    connection: ConnectionOptions,
    subscription: string,           // REQUIRED - unique subscriber name (consumer group)
    startFrom?: string,             // '$' (default, new only) | '0-0' (replay all history)
    subjects?: string[],            // NATS-style subject filter patterns
    concurrency?: number,           // same as Worker
    limiter?: { max, duration },    // same as Worker
    // All other Worker options supported (backoff, etc.)
  },
);

await worker.close();

Subject Filtering (NATS-style)

Patterns use . as token separator:

Pattern Examples

Usage

// Single pattern
const worker = new BroadcastWorker('events', processor, {
  connection,
  subscription: 'order-handler',
  subjects: ['orders.*'],
});

// Multiple patterns
const worker = new BroadcastWorker('events', processor, {
  connection,
  subscription: 'mixed-handler',
  subjects: ['orders.*', 'inventory.low', 'shipping.>'],
});

How Filtering Works

1. subjects compiled to matcher at construction via compileSubjectMatcher. 2. Non-matching messages are auto-acknowledged (XACK) and skipped. 3. Empty/unset subjects = all messages processed.

Utility Functions

import { matchSubject, compileSubjectMatcher } from 'glide-mq';

matchSubject('orders.*', 'orders.created');  // true
matchSubject('orders.*', 'orders.a.b');      // false

const matcher = compileSubjectMatcher(['orders.*', 'shipping.>']);
matcher('orders.created');    // true
matcher('shipping.us.west');  // true
matcher('inventory.low');     // false

Queue vs Broadcast

HTTP Proxy

Cross-language producers and consumers can use the proxy instead of Broadcast / BroadcastWorker directly:

SSE payloads arrive as event: message with JSON { id, subject, data, timestamp }.

Gotchas

• subscription is required on BroadcastWorker - it becomes the consumer group name.
• Proxy SSE subscription follows the same rule and becomes the consumer-group name.
• Subject filtering requires publishing with a name using dotted convention.
• > wildcard must be the last token in the pattern.
• startFrom: '0-0' replays all retained history (backfill).
• Per-subscriber retries - each subscriber independently retries failed messages.

Connection Reference

ConnectionOptions Interface

interface ConnectionOptions {
  addresses: { host: string; port: number }[];  // ARRAY of address objects
  useTLS?: boolean;
  credentials?: PasswordCredentials | IamCredentials;
  clusterMode?: boolean;
  readFrom?: ReadFrom;
  clientAz?: string;
  inflightRequestsLimit?: number;  // default: 1000
  requestTimeout?: number;         // command timeout in ms, default: 500
}

Basic Connection

const connection = { addresses: [{ host: 'localhost', port: 6379 }] };
const queue = new Queue('tasks', { connection });

TLS

const connection = {
  addresses: [{ host: 'my-server.com', port: 6379 }],
  useTLS: true,
};

Authentication

Password-based

interface PasswordCredentials {
  username?: string;
  password: string;
}

const connection = {
  addresses: [{ host: 'server.com', port: 6379 }],
  useTLS: true,
  credentials: { password: 'secret' },
};

IAM (AWS ElastiCache / MemoryDB)

interface IamCredentials {
  type: 'iam';
  serviceType: 'elasticache' | 'memorydb';
  region: string;         // e.g. 'us-east-1'
  userId: string;         // IAM user ID (maps to username in AUTH)
  clusterName: string;
  refreshIntervalSeconds?: number;  // default: 300 (5 min)
}

const connection = {
  addresses: [{ host: 'my-cluster.cache.amazonaws.com', port: 6379 }],
  clusterMode: true,
  credentials: {
    type: 'iam',
    serviceType: 'elasticache',
    region: 'us-east-1',
    userId: 'my-iam-user',
    clusterName: 'my-cluster',
  },
};

Cluster Mode

const connection = {
  addresses: [
    { host: 'node1', port: 7000 },
    { host: 'node2', port: 7001 },
  ],
  clusterMode: true,
};

Keys are hash-tagged automatically (glide:{queueName}:*) for cluster compatibility.

Read Strategies

const connection = {
  addresses: [{ host: 'cluster.cache.amazonaws.com', port: 6379 }],
  clusterMode: true,
  readFrom: 'AZAffinity',
  clientAz: 'us-east-1a',
};

AZ-based strategies require clientAz to be set.

Shared Client Pattern

By default each component creates its own GLIDE client. You can inject a shared client to reduce connections.

import { GlideClient } from '@glidemq/speedkey';

const client = await GlideClient.createClient({ addresses: [{ host: 'localhost' }] });

const queue  = new Queue('jobs', { client });           // borrows client
const flow   = new FlowProducer({ client });            // borrows client
const worker = new Worker('jobs', handler, {
  connection,         // REQUIRED - blocking client auto-created
  commandClient: client,  // shared client for non-blocking ops
});
const events = new QueueEvents('jobs', { connection }); // always own connection
// Total: 2 TCP connections (shared + worker's blocking client)

What can share

Queue, FlowProducer, Worker's command client - all non-blocking operations. GLIDE multiplexes up to 1000 in-flight requests over one TCP connection.

What cannot share

• Worker's blocking client (XREADGROUP BLOCK) - always auto-created
• QueueEvents (XREAD BLOCK) - always own connection. Throws if you pass client.

Close order

// Close components first, then shared client
await queue.close();    // detaches (does not close shared client)
await worker.close();   // closes only auto-created blocking client
await flow.close();
client.close();         // now safe

inflightRequestsLimit

Default 1000. At Worker concurrency=50, peak inflight is ~55 commands.

const connection = {
  addresses: [{ host: 'localhost' }],
  inflightRequestsLimit: 2000,
};

requestTimeout

Command timeout in milliseconds. Default: 500. Commands exceeding this throw a TimeoutError. Increase for operations that may take longer (e.g. FT.CREATE with many existing keys, FUNCTION LOAD with large libraries).

const connection = {
  addresses: [{ host: 'localhost', port: 6379 }],
  requestTimeout: 2000,  // 2 seconds
};

Valkey Modules (Search / JSON / Bloom)

Vector search (queue.createJobIndex(), queue.vectorSearch()) requires the valkey-search module loaded on the server. The easiest way to get all modules is to use valkey-bundle, which bundles search, JSON, bloom, and other modules:

# Docker (standalone with all modules)
docker run -p 6379:6379 valkey/valkey-bundle:latest

# Or load the search module explicitly
valkey-server --loadmodule /path/to/valkeysearch.so

Vector search is supported in standalone mode only (not cluster mode) due to Valkey Search module limitations.

Gotchas

• addresses is an array of { host, port } objects, not a single host/port.
• Worker always requires connection even when commandClient is provided.
• commandClient and client are aliases on Worker - use one, not both.
• Don't close shared client while components are alive.
• QueueEvents cannot accept an injected client - throws.
• Don't mutate shared client state externally (e.g., SELECT).

Observability Reference

QueueEvents

Stream-based lifecycle events via XREAD BLOCK. Real-time without polling.

import { QueueEvents } from 'glide-mq';

const events = new QueueEvents('tasks', { connection });

events.on('added', ({ jobId }) => { ... });
events.on('progress', ({ jobId, data }) => { ... });
events.on('completed', ({ jobId, returnvalue }) => { ... });
events.on('failed', ({ jobId, failedReason }) => { ... });
events.on('stalled', ({ jobId }) => { ... });
events.on('paused', () => { ... });
events.on('resumed', () => { ... });
events.on('usage', ({ jobId, data }) => { ... });  // AI usage reported

await events.close();

Disabling Server-Side Events

Save 1 redis.call() per job on high-throughput workloads:

const queue = new Queue('tasks', { connection, events: false });
const worker = new Worker('tasks', handler, { connection, events: false });

TS-side EventEmitter events (worker.on('completed', ...)) are unaffected.

QueueEvents Cannot Share Clients

QueueEvents uses XREAD BLOCK - always creates its own connection. Throws if you pass client.

Job Logs

// Inside processor
await job.log('Starting step 1');
await job.log('Step 1 done');

// Fetching externally
const { logs, count } = await queue.getJobLogs(jobId);
// logs: string[], count: number

// Paginated
const { logs } = await queue.getJobLogs(jobId, 0, 49);   // first 50
const { logs } = await queue.getJobLogs(jobId, 50, 99);  // next 50

Job Progress

// Inside processor
await job.updateProgress(50);              // number (0-100)
await job.updateProgress({ step: 3 });     // or object

// Listen via QueueEvents
events.on('progress', ({ jobId, data }) => { ... });

// Or via Worker events
worker.on('active', (job) => { ... });

Job Counts

const counts = await queue.getJobCounts();
// { waiting: 12, active: 3, delayed: 5, completed: 842, failed: 7 }

const waitingCount = await queue.count();  // stream length only

Time-Series Metrics

const metrics = await queue.getMetrics('completed');
// {
//   count: 15234,
//   data: [
//     { timestamp: 1709654400000, count: 142, avgDuration: 234 },
//     { timestamp: 1709654460000, count: 156, avgDuration: 218 },
//   ],
//   meta: { resolution: 'minute' }
// }

// Slice (e.g., last 10 data points)
const recent = await queue.getMetrics('completed', { start: -10 });

• Recorded server-side with zero extra RTTs.
• Minute-resolution buckets retained for 24 hours, trimmed automatically.
• Type: 'completed' or 'failed'.

Disabling Metrics

const worker = new Worker('tasks', handler, {
  connection,
  metrics: false,  // skip HINCRBY per job
});

Waiting for a Job

// Poll job hash until finished
const state = await job.waitUntilFinished(pollIntervalMs, timeoutMs);
// Returns 'completed' | 'failed'

// Request-reply (no polling)
const result = await queue.addAndWait('inference', data, { waitTimeout: 30_000 });

AI Usage Telemetry

Per-Job Usage

// Report usage inside a processor
await job.reportUsage({
  model: 'gpt-5.4',
  provider: 'openai',
  tokens: { input: 500, output: 200 },
  costs: { total: 0.003 },
  costUnit: 'usd',
  latencyMs: 800,
  cached: false,
});

// Emits a 'usage' event on the events stream
events.on('usage', ({ jobId, data }) => {
  const usage = JSON.parse(data);
  console.log(`Job ${jobId}: ${usage.model} - ${usage.totalTokens} tokens`);
});

// Read usage from a completed job
const job = await queue.getJob(jobId);
console.log(job.usage);
// { model, provider, tokens, totalTokens, costs, totalCost, costUnit, latencyMs, cached }

Flow-Level Aggregation

const usage = await queue.getFlowUsage(parentJobId);
// {
//   tokens: { input: 2500, output: 1200 },
//   totalTokens: 3700,
//   costs: { total: 0.015 },
//   totalCost: 0.015,
//   costUnit: 'usd',
//   jobCount: 4,
//   models: { 'gpt-5.4': 3, 'claude-sonnet-4-20250514': 1 }
// }

Walks the parent job and all children via the deps set. Includes usage from the parent itself.

Rolling Usage Summary

const summary = await queue.getUsageSummary({
  queues: ['tasks', 'embeddings'],
  windowMs: 3_600_000,
});

// { totalTokens, totalCost, jobCount, models, perQueue }

This reads rolling per-minute buckets instead of scanning job hashes, so it is the right primitive for dashboards and queue-wide cost telemetry.

Budget Monitoring

const budget = await queue.getFlowBudget(flowId);
if (budget && budget.exceeded) {
  console.warn(`Flow ${flowId} exceeded budget: ${budget.usedTokens} tokens, $${budget.usedCost}`);
}

Proxy SSE Surfaces

For cross-language observability, the HTTP proxy exposes:

These routes require the proxy to be created with connection, because they allocate blocking readers internally.

OpenTelemetry

Auto-emits spans when @opentelemetry/api is installed. No code changes needed.

npm install @opentelemetry/api

Initialize tracer provider before creating Queue/Worker (standard OTel setup).

Custom Tracer

import { setTracer, isTracingEnabled } from 'glide-mq';
import { trace } from '@opentelemetry/api';

setTracer(trace.getTracer('my-service', '1.0.0'));
console.log('Tracing:', isTracingEnabled());

Instrumented Operations

Gotchas

• QueueEvents always creates its own connection - cannot use shared client.
• Disabling events only affects the Valkey events stream, not TS-side EventEmitter.
• getMetrics() type is 'completed' or 'failed' only.
• OTel spans are automatic if @opentelemetry/api is installed - no explicit setup in glide-mq.
• job.waitUntilFinished() does NOT require QueueEvents (unlike BullMQ) - polls job hash directly.

Queue Reference

Constructor

import { Queue } from 'glide-mq';

const queue = new Queue('tasks', {
  connection: ConnectionOptions,   // required unless `client` provided
  client?: Client,                 // pre-existing GLIDE client (not owned)
  prefix?: string,                 // key prefix (default: 'glide')
  compression?: 'none' | 'gzip',  // default: 'none'
  serializer?: Serializer,        // default: JSON_SERIALIZER
  events?: boolean,               // emit 'added' events (default: true)
  deadLetterQueue?: { name: string; maxRetries?: number },
});

Adding Jobs

// Single job - returns Job | null (null if dedup/collision)
const job = await queue.add(name: string, data: any, opts?: JobOptions);

// Bulk add - 12.7x faster via GLIDE Batch API
const jobs = await queue.addBulk([
  { name: 'job1', data: { a: 1 }, opts?: JobOptions },
]);

// Request-reply - blocks until worker returns result
const result = await queue.addAndWait(name, data, {
  waitTimeout: 30_000,  // producer-side wait budget (separate from job timeout)
  // Does NOT support removeOnComplete or removeOnFail
  // Rejects if dedup returns null
});

JobOptions

Note: Compression is not a per-job option. Set compression: 'gzip' at Queue level in the Queue constructor.

Processing Order

priority > LIFO > FIFO. Priority jobs first, then LIFO list, then FIFO stream.

Queue Management

await queue.pause();             // workers stop picking up new jobs
await queue.resume();
const paused = await queue.isPaused();

// Drain - remove waiting jobs
await queue.drain();             // waiting only
await queue.drain(true);         // also delayed/scheduled

// Obliterate - remove ALL queue data
await queue.obliterate();              // fails if active jobs exist
await queue.obliterate({ force: true });

// Clean old jobs by age
const ids = await queue.clean(grace: number, limit: number, type: 'completed' | 'failed');

await queue.close();

Inspecting Jobs

const job = await queue.getJob('42');
const job = await queue.getJob('42', { excludeData: true });  // metadata only

const jobs = await queue.getJobs(state, start?, end?);
// state: 'waiting' | 'active' | 'delayed' | 'completed' | 'failed'
const lite = await queue.getJobs('waiting', 0, 99, { excludeData: true });

const counts = await queue.getJobCounts();
// { waiting, active, delayed, completed, failed }

const results = await queue.searchJobs({ state?, name?, data?, limit? });
// data: shallow key-value match. limit default: 100

const waitingCount = await queue.count();  // stream length

Rate Limiting

// Per-worker rate limit (in WorkerOptions)
limiter: { max: 100, duration: 60_000 }  // 100 jobs/min

// Global rate limit (across all workers)
await queue.setGlobalRateLimit({ max: 500, duration: 60_000 });
const limit = await queue.getGlobalRateLimit();
await queue.removeGlobalRateLimit();

// Global concurrency
await queue.setGlobalConcurrency(20);
await queue.setGlobalConcurrency(0);  // remove limit

Dead Letter Queue

// Configure on Worker
const worker = new Worker('tasks', processor, {
  connection,
  deadLetterQueue: { name: 'tasks-dlq' },
});

// Inspect DLQ
const dlqJobs = await queue.getDeadLetterJobs(0, 49);

Token Streaming

// Read entries from a job's streaming channel
const entries = await queue.readStream(jobId);
// entries: { id: string; fields: Record<string, string> }[]

// Resume from last position
const more = await queue.readStream(jobId, { lastId: entries.at(-1)?.id });

// Long-polling (blocks until new entries or timeout)
const live = await queue.readStream(jobId, {
  lastId: '0-0',
  count: 50,        // max entries (default: 100)
  block: 5000,      // XREAD BLOCK ms
});

Flow Usage Aggregation

const usage = await queue.getFlowUsage(parentJobId);
// {
//   tokens: Record<string, number>,    // aggregated per-category (e.g. { input, output })
//   totalTokens: number,
//   costs: Record<string, number>,     // aggregated per-category costs
//   totalCost: number,
//   costUnit?: string,
//   jobCount: number,
//   models: Record<string, number>     // model -> call count
// }

Rolling Usage Summary

const summary = await queue.getUsageSummary({
  queues: ['tasks', 'embeddings'],
  windowMs: 3_600_000,
});

// Static form:
const sameSummary = await Queue.getUsageSummary({ connection, queues: ['tasks'] });

Flow Budget

const budget = await queue.getFlowBudget(flowId);
// null if no budget set, otherwise:
// {
//   maxTotalTokens?: number,
//   maxTokens?: Record<string, number>,
//   tokenWeights?: Record<string, number>,
//   maxTotalCost?: number,
//   maxCosts?: Record<string, number>,
//   costUnit?: string,
//   usedTokens: number,
//   usedCost: number,
//   exceeded: boolean,
//   onExceeded: 'pause' | 'fail'
// }

Suspend / Resume

// Send a signal to resume a suspended job
const resumed = await queue.signal(jobId, 'approve', { approvedBy: 'alice' });
// true if job was resumed, false if not suspended

// Inspect suspension state
const info = await queue.getSuspendInfo(jobId);
// null if not suspended, otherwise:
// { reason?, suspendedAt, timeout?, signals: SignalEntry[] }

Vector Search

// Create a search index over job hashes
await queue.createJobIndex({
  vectorField: { name: 'embedding', dimensions: 1536 },
  fields: [{ type: 'TAG', name: 'category' }],
});

// Search by vector similarity
const results = await queue.vectorSearch(embedding, {
  k: 10,
  filter: '@state:{completed}',
});
// results: { job: Job, score: number }[]

// Drop the index (does not delete jobs)
await queue.dropJobIndex();

See references/ai-native.md and references/search.md for full details.

Gotchas

• Priority: 0 is highest priority. Lower number = higher priority. Max 2048.
• addAndWait() rejects if dedup returns null. Does not support removeOnComplete/removeOnFail.
• queue.add() returns null on custom jobId collision or deduplication skip.
• FlowProducer.add() throws on duplicate jobId (flows cannot be partial).
• getUsageSummary() is for queue-wide rollups. Use getJob() / job.usage for per-job detail.
• Payload size limit: job data must be <= 1 MB after serialization, before compression.
• Same serializer must be used on Queue, Worker, and FlowProducer. Mismatch causes silent corruption.
• lifo and ordering.key are mutually exclusive - throws at enqueue time.

Schedulers Reference

Overview

upsertJobScheduler defines repeatable jobs via cron or fixed interval. Schedulers survive restarts - next run time is stored in Valkey.

API

All scheduler operations are on the Queue instance:

const queue = new Queue('tasks', { connection });

Cron Schedule

await queue.upsertJobScheduler(
  'daily-report',                    // scheduler ID (unique per queue)
  { pattern: '0 8 * * *' },         // cron expression
  { name: 'generate-report', data: { type: 'daily' } },  // job template
);

Fixed Interval

await queue.upsertJobScheduler(
  'cleanup',
  { every: 5 * 60 * 1_000 },        // interval in ms
  { name: 'cleanup-old', data: {} },
);

Repeat After Complete

Schedules next job only after current completes (no overlap).

await queue.upsertJobScheduler(
  'sensor-poll',
  { repeatAfterComplete: 5000 },     // 5s after previous completes
  { name: 'poll', data: { sensor: 'temp-1' } },
);

Mutually exclusive with pattern and every.

Schedule Options

Only one of pattern, every, repeatAfterComplete per scheduler.

Bounded Schedulers

// Campaign window with max runs
await queue.upsertJobScheduler(
  'black-friday',
  {
    pattern: '0 */2 * * *',
    startDate: new Date('2026-11-28T00:00:00Z'),
    endDate: new Date('2026-12-01T00:00:00Z'),
    limit: 36,
  },
  { name: 'promote-deal', data: { campaign: 'bf' } },
);

// Interval with delayed start and hard stop
await queue.upsertJobScheduler(
  'warmup-cache',
  {
    every: 30_000,
    startDate: Date.now() + 60_000,
    endDate: new Date('2026-12-31'),
    limit: 100,
  },
  { name: 'warmup', data: { region: 'us-east' } },
);

Management

// List all schedulers
const schedulers = await queue.getRepeatableJobs();
// Returns stored bounds + iterationCount

// Get single scheduler details
const info = await queue.getJobScheduler('daily-report');

// Remove a scheduler (does not cancel in-flight jobs)
await queue.removeJobScheduler('cleanup');

// Upsert updates existing scheduler atomically
await queue.upsertJobScheduler('cleanup', { every: 10_000 }, { name: 'cleanup', data: {} });

Gotchas

• pattern, every, repeatAfterComplete are mutually exclusive.
• repeatAfterComplete prevents overlap - next job only after current finishes or terminally fails.
• Scheduler ID is unique per queue. upsert replaces if exists.
• removeJobScheduler does not cancel jobs already in flight.
• Bounded options (startDate, endDate, limit) work with all three modes.
• Internal Scheduler class fires a promotion loop that converts due entries into real jobs.
• getRepeatableJobs() / getJobScheduler() expose iterationCount for inspection.

Vector Search Reference

Create Valkey Search indexes over job hashes for vector similarity search (KNN).

Requires the valkey-search module loaded on the Valkey server (standalone mode only).

Creating an Index

import { Queue } from 'glide-mq';
import type { Field } from 'glide-mq';

const queue = new Queue('embeddings', { connection });

// Minimal index (base fields only, no vector search)
await queue.createJobIndex();

// Index with vector field for KNN search
await queue.createJobIndex({
  name: 'embeddings-idx',            // default: '{queueName}-idx'
  vectorField: {
    name: 'embedding',               // field name in the job hash
    dimensions: 1536,                // vector dimensions (e.g. OpenAI ada-002)
    algorithm: 'HNSW',              // 'HNSW' (default) or 'FLAT'
    distanceMetric: 'COSINE',       // 'COSINE' (default), 'L2', or 'IP'
  },
  fields: [                          // additional schema fields
    { type: 'TAG', name: 'category' } as Field,
    { type: 'TEXT', name: 'summary' } as Field,
    { type: 'NUMERIC', name: 'score' } as Field,
  ],
});

Auto-Included Base Fields

Every index automatically includes:

JobIndexOptions

interface JobIndexOptions {
  name?: string;                      // index name, default: '{queueName}-idx'
  fields?: Field[];                   // additional schema fields
  vectorField?: {
    name: string;                     // field name where vector is stored
    dimensions: number;               // vector dimensions
    algorithm?: 'HNSW' | 'FLAT';     // default: 'HNSW'
    distanceMetric?: 'COSINE' | 'L2' | 'IP';  // default: 'COSINE'
  };
  createOptions?: IndexCreateOptions; // pass-through to FT.CREATE
}

IndexCreateOptions

interface IndexCreateOptions {
  score?: number;              // default document score
  language?: string;           // default stemming language
  skipInitialScan?: boolean;   // skip indexing existing docs
  minStemSize?: number;
  withOffsets?: boolean;
  noOffsets?: boolean;
  noStopWords?: boolean;
  stopWords?: string[];
  punctuation?: string;
}

Vector Search (KNN)

// Generate an embedding for the query
const queryEmbedding = await openai.embeddings.create({
  model: 'text-embedding-ada-002',
  input: 'machine learning optimization',
});

const results = await queue.vectorSearch(
  queryEmbedding.data[0].embedding,   // number[] or Float32Array
  {
    k: 10,                             // nearest neighbours (default: 10)
    filter: '@state:{completed}',      // pre-filter expression
    indexName: 'embeddings-idx',       // default: '{queueName}-idx'
    scoreField: '__score',             // score field name (default: '__score')
  },
);

for (const { job, score } of results) {
  console.log(`Job ${job.id}: ${job.name} (score: ${score})`);
  console.log('  Data:', job.data);
}

VectorSearchOptions

interface VectorSearchOptions {
  indexName?: string;           // default: '{queueName}-idx'
  k?: number;                   // nearest neighbours (default: 10)
  filter?: string;              // pre-filter expression (default: '*')
  scoreField?: string;          // score field name (default: '__score')
  searchOptions?: SearchQueryOptions;
}

VectorSearchResult

interface VectorSearchResult<D = any, R = any> {
  job: Job<D, R>;     // fully hydrated Job object
  score: number;       // distance/similarity score
}

Score interpretation depends on distance metric:

• COSINE: 0 = identical, 2 = opposite (lower is more similar)
• L2: 0 = identical (lower is more similar)
• IP (inner product): higher is more similar

SearchQueryOptions

interface SearchQueryOptions {
  nocontent?: boolean;          // return only IDs
  dialect?: number;             // query dialect version
  verbatim?: boolean;           // disable stemming
  inorder?: boolean;            // proximity terms must be in order
  slop?: number;                // proximity matching slop
  sortby?: { field: string; order?: 'ASC' | 'DESC' };
  scorer?: string;              // scoring function name
}

Storing Vectors in Jobs

Create the job first, then store the embedding with job.storeVector(...). This writes the raw FLOAT32 buffer to the job hash in the format Valkey Search expects.

// When adding jobs with embeddings
const embedding = await getEmbedding(text);
const job = await queue.add('document', {
  text,
  summary: 'A document about...',
  category: 'research',
});
if (job) {
  await job.storeVector('embedding', embedding);
}

Testing mode provides parity via TestJob.storeVector(...), TestQueue.createJobIndex(...), and TestQueue.vectorSearch(...).

Dropping an Index

// Drop by default name
await queue.dropJobIndex();

// Drop by custom name
await queue.dropJobIndex('embeddings-idx');

Dropping an index does not delete the job hashes - only the search index is removed.

Pre-Filter Expressions

Use Valkey Search query syntax for pre-filtering before KNN:

// Filter by state
await queue.vectorSearch(embedding, { filter: '@state:{completed}' });

// Filter by job name
await queue.vectorSearch(embedding, { filter: '@name:{summarize}' });

// Filter by priority range
await queue.vectorSearch(embedding, { filter: '@priority:[0 5]' });

// Combine filters
await queue.vectorSearch(embedding, {
  filter: '@state:{completed} @name:{embed|summarize}',
});

// No filter (search all indexed jobs)
await queue.vectorSearch(embedding, { filter: '*' });

Gotchas

• Requires valkey-search module loaded on the Valkey server (standalone mode only, not cluster).
• When no vectorField is specified in createJobIndex(), a minimal 2-dimensional placeholder vector field (_vec) is added because valkey-search requires at least one vector field.
• The index prefix is automatically scoped to this queue's job hashes.
• dropJobIndex() only removes the index, not the underlying job data.
• Vector search returns fully hydrated Job objects - each result triggers an HMGET to fetch the full job hash.
• The Field type is re-exported from @glidemq/speedkey.

Serverless & Testing Reference

Producer (Lightweight Queue.add)

No EventEmitter, no Job instances, no state tracking. Same FCALL functions as Queue.

import { Producer } from 'glide-mq';

const producer = new Producer('emails', {
  connection: ConnectionOptions,  // required unless `client` provided
  client?: Client,               // pre-existing GLIDE client (not owned)
  prefix?: string,               // default: 'glide'
  compression?: 'none' | 'gzip', // default: 'none'
  serializer?: Serializer,       // default: JSON
  events?: boolean,              // emit 'added' events (default: true, set false to save 1 call)
});

// Returns string ID (not Job object) or null for dedup/collision
const id = await producer.add('send-welcome', { to: 'user@example.com' });
const id = await producer.add('urgent', data, { delay: 3600000, priority: 1 });

// Bulk - returns (string | null)[]
const ids = await producer.addBulk([
  { name: 'email', data: { to: 'a@test.com' } },
  { name: 'sms', data: { phone: '+123' } },
]);

await producer.close();  // if external client was provided, it is NOT closed

All JobOptions work: delay, priority, deduplication, jobId, ordering, ttl, lifo, cost.

ServerlessPool

Reuses connections across warm Lambda/Edge invocations.

import { serverlessPool, ServerlessPool } from 'glide-mq';

// Module-level singleton
const producer = serverlessPool.getProducer('notifications', {
  connection: { addresses: [{ host: process.env.VALKEY_HOST!, port: 6379 }] },
});
await producer.add('push', { userId: 42 });

// Or create your own pool
const pool = new ServerlessPool();
const p = pool.getProducer('queue', { connection });
await pool.closeAll();

AWS Lambda Example

import { serverlessPool } from 'glide-mq';

const CONNECTION = {
  addresses: [{ host: process.env.VALKEY_HOST!, port: 6379 }],
};

export async function handler(event: any) {
  const producer = serverlessPool.getProducer('notifications', {
    connection: CONNECTION,
  });
  const id = await producer.add('push-notification', {
    userId: event.userId,
    message: event.message,
  });
  return { statusCode: 200, body: JSON.stringify({ jobId: id }) };
}

process.on('SIGTERM', async () => { await serverlessPool.closeAll(); });

Connection Behavior

• Cold start: creates new GLIDE connection + loads function library
• Warm invocation: returns cached producer (zero overhead)
• Container freeze/thaw: GLIDE auto-reconnects on next command

HTTP Proxy

Express-based HTTP proxy for enqueueing, request-reply, queue telemetry, and SSE consumption from any language/environment.

import { createProxyServer } from 'glide-mq/proxy';

const proxy = createProxyServer({
  connection: ConnectionOptions,  // required unless client provided
  client?: Client,               // pre-existing GLIDE client
  prefix?: string,               // default: 'glide'
  queues?: string[],             // allowlist (403 for unlisted queues)
  compression?: 'none' | 'gzip',
  onError?: (err, queueName) => void,
});

proxy.app.listen(3000);
await proxy.close();  // shuts down all cached Queue instances

Proxy Endpoints

Proxy Notes

• Add your own auth/rate limiting middleware before exposing the proxy publicly.
• Queue-wide SSE and broadcast SSE require connection, not just client, because they allocate blocking readers.
• queues is an allowlist. Unlisted queue names return 403.
• POST /flows supports FlowProducer-style trees and DAG payloads. HTTP budgets are currently supported for tree flows only.

Testing (In-Memory)

No Valkey needed. Import from glide-mq/testing.

import { TestQueue, TestWorker } from 'glide-mq/testing';

const queue = new TestQueue('tasks');   // no connection config needed
const worker = new TestWorker(queue, async (job) => {
  return { processed: job.data };
});

worker.on('completed', (job, result) => { ... });
worker.on('failed', (job, err) => { ... });

await queue.add('send-email', { to: 'user@example.com' });
const counts = await queue.getJobCounts();
// { waiting: 0, active: 0, delayed: 0, completed: 1, failed: 0 }

await worker.close();
await queue.close();

TestQueue API

TestJob API

TestWorker Events

Same as Worker: active, completed, failed, drained.

Batch Testing

const worker = new TestWorker(queue, async (jobs) => {
  return jobs.map(j => ({ doubled: j.data.n * 2 }));
}, { batch: { size: 5, timeout: 100 } });

Key Testing Behaviors

• Processing is synchronous-ish - check state right after await queue.add().
• Delayed jobs become waiting immediately (delay not honored in test mode).
• moveToDelayed not supported in test mode.
• Custom jobId returns null on duplicate (mirrors production).
• All three dedup modes (simple, throttle, debounce) work.
• Retries work normally with attempts and backoff.
• Swap without changing processors - same interface as Queue/Worker.

Gotchas

• Producer returns string IDs, not Job objects.
• Producer close() does NOT close an externally provided client.
• serverlessPool is a module-level singleton - shared across handler invocations.
• HTTP proxy requires express as a peer dependency.
• Proxy queues option is an allowlist - unlisted names get 403, and the same allowlist applies to /usage/summary?queues=... and /broadcast/:name.
• Queue-wide/broadcast SSE proxy routes require connection, not only client.
• TestQueue isPaused() is synchronous (real Queue returns Promise).
• Test mode does not honor delay or moveToDelayed.

Worker Reference

Constructor

import { Worker } from 'glide-mq';

const worker = new Worker(
  'tasks',              // queue name
  async (job) => {      // processor function
    // job.data, job.name, job.id, job.opts
    await job.log('step done');
    await job.updateProgress(50);       // 0-100 or object
    await job.updateData({ ...job.data, enriched: true });
    return { ok: true };                // becomes job.returnvalue
  },
  {
    connection: ConnectionOptions,       // required (even if commandClient provided)
    commandClient?: Client,              // shared client for non-blocking ops (alias: client)
    concurrency?: number,                // parallel jobs (default: 1)
    blockTimeout?: number,               // XREADGROUP BLOCK ms (default: 5000)
    stalledInterval?: number,            // stall check interval ms (default: 30000)
    lockDuration?: number,               // stall detection window per job ms (default: 30000)
    maxStalledCount?: number,            // max stall recoveries before fail
    limiter?: { max, duration },         // rate limit per worker
    deadLetterQueue?: { name: string },  // inherited from QueueOptions - usually set on Queue
    events?: boolean,                    // emit completed/failed events (default: true)
    metrics?: boolean,                   // record metrics (default: true)
    prefix?: string,
    serializer?: Serializer,
    tokenLimiter?: {
      maxTokens: number,          // max tokens per window
      duration: number,           // window duration in ms
      scope?: 'queue' | 'worker' | 'both',  // default: 'both'
    },
    backoffStrategies?: Record<string, (attemptsMade: number, err: Error) => number>,
  },
);

Batch Processing

import { Worker, BatchError } from 'glide-mq';

const worker = new Worker(
  'bulk-insert',
  async (jobs) => {              // receives Job[] in batch mode
    const results = await db.insertMany(jobs.map(j => j.data));
    return results;              // must return R[] with length === jobs.length
  },
  {
    connection,
    batch: {
      size: 50,          // max jobs per batch (1-1000)
      timeout: 1000,     // ms to wait for full batch (optional)
    },
  },
);

// Partial failures - report per-job outcomes
async (jobs) => {
  const results = await Promise.allSettled(jobs.map(processOne));
  const mapped = results.map(r => r.status === 'fulfilled' ? r.value : r.reason);
  if (mapped.some(r => r instanceof Error)) {
    throw new BatchError(mapped);  // each job individually completed/failed
  }
  return mapped;
};

Worker Events

worker.on('completed', (job, result) => { ... });
worker.on('failed', (job, err) => { ... });
worker.on('error', (err) => { ... });
worker.on('stalled', (jobId) => { ... });

Stall Detection

• Worker extends job lock every lockRenewTime (default: lockDuration/2).
• If lock expires (job exceeds lockDuration without renewal), job is stalled.
• Stalled jobs are re-queued up to maxStalledCount times, then failed.
• Check interval controlled by stalledInterval.

LIFO Mode

Workers check sources in order: priority > LIFO > FIFO. Add jobs with { lifo: true } to process newest first. LIFO uses a dedicated Valkey LIST separate from the FIFO stream.

Job Revocation (AbortSignal)

// Queue-side: revoke a job
const result = await queue.revoke(job.id);
// 'revoked'    - was waiting/delayed, now failed
// 'flagged'    - active, worker will abort cooperatively
// 'not_found'  - job does not exist

// Worker-side: check for revocation
const worker = new Worker('tasks', async (job) => {
  for (const chunk of dataset) {
    if (job.abortSignal?.aborted) throw new Error('Revoked');
    await processChunk(chunk);
  }
}, { connection });

job.abortSignal is a standard AbortSignal - pass to fetch, axios, etc.

Pause / Resume / Close

await worker.pause();        // stop accepting new jobs (active finish)
await worker.pause(true);    // force-stop immediately
await worker.resume();

await worker.close();        // graceful: waits for active jobs
await worker.close(true);    // force-close immediately

AI Usage & Token Tracking

const worker = new Worker('inference', async (job) => {
  const result = await callLLM(job.data.prompt);

  // Report AI usage metadata (persisted to job hash, emits 'usage' event)
  await job.reportUsage({
    model: 'gpt-5.4',
    provider: 'openai',
    tokens: { input: result.promptTokens, output: result.completionTokens },
    costs: { total: 0.003 },
    costUnit: 'usd',
    latencyMs: 800,
  });

  // Or report just tokens for TPM rate limiting
  await job.reportTokens(result.totalTokens);

  return result.content;
}, {
  connection,
  limiter: { max: 60, duration: 60_000 },        // RPM limit
  tokenLimiter: { maxTokens: 100_000, duration: 60_000 },  // TPM limit
});

Worker pauses fetching when either RPM limiter or TPM tokenLimiter is exceeded.

Token Streaming

const worker = new Worker('chat', async (job) => {
  const stream = await openai.chat.completions.create({ stream: true, ... });
  for await (const chunk of stream) {
    const token = chunk.choices[0]?.delta?.content;
    if (token) {
      await job.stream({ token });  // XADD to per-job stream
    }
  }
  return { done: true };
}, { connection });

Consumers read via queue.readStream(jobId, opts).

Suspend / Resume (Human-in-the-Loop)

const worker = new Worker('review', async (job) => {
  // On resume, signals are populated
  if (job.signals.length > 0) {
    const approval = job.signals.find(s => s.name === 'approve');
    if (approval) return { approved: true };
    return { rejected: true };
  }

  // First run - suspend for human review
  await job.suspend({ reason: 'Needs approval', timeout: 86_400_000 });
  // throws SuspendError - no code after this executes
}, { connection });

Resume externally via queue.signal(jobId, 'approve', { ... }).

Fallback Chains

const worker = new Worker('inference', async (job) => {
  const fallback = job.currentFallback;
  // undefined on first attempt, then fallbacks[0], fallbacks[1], etc.
  const model = fallback?.model ?? 'gpt-5.4-nano';
  return await callLLM(model, job.data.prompt);
}, { connection });

Set via queue.add('inference', data, { fallbacks: [...], attempts: 4 }).

Skipping Retries

import { UnrecoverableError } from 'glide-mq';

// Option 1: UnrecoverableError - skips all remaining retries
throw new UnrecoverableError('bad input');

// Option 2: job.discard() + throw - same effect
job.discard();
throw new Error('discarded');

Step Jobs (moveToDelayed)

const worker = new Worker('drip', async (job) => {
  switch (job.data.step) {
    case 'send':
      await sendEmail(job.data);
      return job.moveToDelayed(Date.now() + 86400_000, 'check');
    case 'check':
      return 'done';
  }
}, { connection });

moveToDelayed(timestampMs, nextStep?) - pauses job until timestamp, optionally updates job.data.step.

Graceful Shutdown

import { gracefulShutdown } from 'glide-mq';
// Returns a handle that auto-registers SIGTERM/SIGINT handlers.
// await blocks until a signal fires. For manual shutdown: handle.shutdown()
const handle = gracefulShutdown([queue, worker, events]);
await handle.shutdown(); // programmatic trigger

Gotchas

• Worker always requires `connection` even with commandClient - blocking client is auto-created.
• commandClient and client are aliases - provide one, not both.
• Don't close shared client while worker is alive. Close worker first.
• Batch processor must return array with length === jobs.length.
• moveToDelayed() must be called from active processor. Throws DelayedError internally.
• job.suspend() throws SuspendError internally - no code after it executes.
• job.reportUsage() and job.reportTokens() reject negative values.
• reportTokens() overwrites previous value (does not accumulate).
• tokenLimiter scope 'both' checks local counter first, then Valkey (optimal for most setups).
• Fallback chains require attempts >= fallbacks.length + 1.

Workflows Reference

FlowProducer

Atomically enqueues a tree of parent-child jobs. Parent only runs after all children complete.

import { FlowProducer } from 'glide-mq';

const flow = new FlowProducer({ connection });
// Also accepts: { client } for shared client

const { job: parent } = await flow.add({
  name: 'aggregate',
  queueName: 'reports',
  data: { month: '2025-01' },
  children: [
    { name: 'fetch-sales', queueName: 'data', data: { region: 'eu' } },
    { name: 'fetch-returns', queueName: 'data', data: {} },
    {
      name: 'fetch-inventory', queueName: 'data', data: {},
      children: [  // nested children supported
        { name: 'load-a', queueName: 'data', data: {} },
      ],
    },
  ],
});

await flow.close();

FlowJob Structure

interface FlowJob {
  name: string;
  queueName: string;
  data: any;
  opts?: JobOptions;
  children?: FlowJob[];
}

Bulk Flows

const nodes = await flow.addBulk([
  { name: 'report-jan', queueName: 'reports', data: {}, children: [...] },
  { name: 'report-feb', queueName: 'reports', data: {}, children: [...] },
]);

Reading Child Results

const worker = new Worker('reports', async (job) => {
  const childValues = await job.getChildrenValues();
  // Keys are opaque internal IDs - use Object.values()
  const results = Object.values(childValues);
  return { total: results.reduce((s, v) => s + v.count, 0) };
}, { connection });

DAG Workflows (Multiple Parents)

addDAG() supports arbitrary DAG topologies where a job can depend on multiple parents.

import { FlowProducer, dag } from 'glide-mq';

// Helper function (simpler API)
const jobs = await dag([
  { name: 'A', queueName: 'tasks', data: { step: 1 } },
  { name: 'B', queueName: 'tasks', data: { step: 2 }, deps: ['A'] },
  { name: 'C', queueName: 'tasks', data: { step: 3 }, deps: ['A'] },
  { name: 'D', queueName: 'tasks', data: { step: 4 }, deps: ['B', 'C'] },  // fan-in
], connection);

// Or via FlowProducer directly
const flow = new FlowProducer({ connection });
const jobs = await flow.addDAG({
  nodes: [
    { name: 'A', queueName: 'tasks', data: {}, deps: [] },
    { name: 'B', queueName: 'tasks', data: {}, deps: ['A'] },
    { name: 'C', queueName: 'tasks', data: {}, deps: ['A'] },
    { name: 'D', queueName: 'tasks', data: {}, deps: ['B', 'C'] },
  ],
});
// Returns Map<string, Job> keyed by node name

DAGNode

• name - unique within the DAG (used in deps)
• queueName - target queue
• data - payload
• opts? - JobOptions
• deps? - array of node names that must complete first

Reading Multiple Parent Results

const worker = new Worker('tasks', async (job) => {
  if (job.name === 'D') {
    const parents = await job.getParents();
    // Returns { queue, id }[] - not Job instances
    // Fetch full jobs if needed:
    const parentJobs = await Promise.all(
      parents.map(p => new Queue(p.queue, { connection }).getJob(p.id))
    );
    const results = parentJobs.map(p => p.returnvalue);
    return { merged: results };
  }
}, { connection });

Convenience Helpers

chain() - Sequential Pipeline

Array is in reverse execution order (last element runs first).

import { chain } from 'glide-mq';

// Execution: download -> parse -> transform -> upload
await chain('pipeline', [
  { name: 'upload',    data: {} },   // runs LAST (root)
  { name: 'transform', data: {} },
  { name: 'parse',     data: {} },
  { name: 'download',  data: {} },   // runs FIRST (leaf)
], connection);

group() - Parallel Execution

import { group } from 'glide-mq';

await group('tasks', [
  { name: 'resize-sm', data: { size: 'sm' } },
  { name: 'resize-md', data: { size: 'md' } },
  { name: 'resize-lg', data: { size: 'lg' } },
], connection);
// Creates synthetic __group__ parent that waits for all children

chord() - Parallel + Callback

import { chord } from 'glide-mq';

await chord(
  'tasks',
  // Group (parallel)
  [
    { name: 'score-a', data: { model: 'a' } },
    { name: 'score-b', data: { model: 'b' } },
  ],
  // Callback (after group completes)
  { name: 'select-best', data: {} },
  connection,
);

Dynamic Children (moveToWaitingChildren)

Spawn children at runtime, then pause parent until they complete.

import { Queue, Worker, WaitingChildrenError } from 'glide-mq';

const worker = new Worker('orchestrator', async (job) => {
  // Detect re-entry
  const existing = await job.getChildrenValues();
  if (Object.keys(existing).length > 0) {
    return { merged: Object.values(existing) };  // aggregate results
  }

  // Spawn children dynamically
  const childQueue = new Queue('subtasks', { connection });
  for (const url of job.data.urls) {
    await childQueue.add('fetch', { url }, {
      parent: { id: job.id!, queue: job.queueQualifiedName },
    });
  }
  await childQueue.close();

  // Pause until all children complete - throws WaitingChildrenError
  await job.moveToWaitingChildren();
}, { connection });

Budget on Flows

Cap total token usage and/or cost across all jobs in a flow tree. Supports per-category limits and weighted totals.

const flow = new FlowProducer({ connection });

await flow.add(
  {
    name: 'research',
    queueName: 'ai',
    data: { topic: 'quantum computing' },
    children: [
      { name: 'search', queueName: 'ai', data: {} },
      { name: 'summarize', queueName: 'ai', data: {} },
    ],
  },
  {
    budget: {
      maxTotalTokens: 50_000,
      maxTotalCost: 0.50,
      costUnit: 'usd',
      tokenWeights: { reasoning: 4, cachedInput: 0.25 },
      onExceeded: 'fail',      // 'fail' (default) or 'pause'
    },
  },
);

// Check budget state
const budget = await queue.getFlowBudget(parentJobId);
// { maxTotalTokens, maxTokens, tokenWeights, maxTotalCost, maxCosts, costUnit,
//   usedTokens, usedCost, exceeded, onExceeded }

Budget is propagated to every job in the flow via a budgetKey field.

Suspend / Resume as Workflow Primitive

Suspend a job in a flow to await human approval, then resume and continue the pipeline.

const worker = new Worker('ai', async (job) => {
  if (job.name === 'review') {
    if (job.signals.length > 0) {
      return { approved: job.signals.some(s => s.name === 'approve') };
    }
    await job.suspend({ reason: 'Human review required', timeout: 86_400_000 });
  }
  // other job types...
}, { connection });

// Resume externally
await queue.signal(jobId, 'approve', { reviewer: 'alice' });

When a suspended job resumes, it re-enters the stream and the processor is invoked again with job.signals populated. The parent flow continues once all children (including the resumed one) complete.

Gotchas

• chain() array is reverse execution order - last element is leaf (runs first).
• moveToWaitingChildren() always throws WaitingChildrenError. No code after it executes.
• Processor re-runs from the top when children complete. Use getChildrenValues() to detect re-entry.
• Children must reference parent via opts.parent: { id, queue }.
• Cycles in DAGs are detected and rejected with CycleError.
• If a parent in a DAG fails, dependent jobs remain blocked indefinitely.
• FlowProducer.add() throws on duplicate jobId (cannot be partially created).
• Cross-queue dependencies are supported - each DAG node can have its own queueName.