Queued ingestion

Move graph construction behind BullMQ and Redis while preserving correct visibility, retries, and shutdown behavior.

When queue mode helps

Use queue mode when segmentation, extraction, and embedding make foreground response latency unacceptable. The LLM response can complete while workers build graph memory afterward.

Configure the queue

queue-config.ts

const agent = new MemoGrafterAgent({
  db: { connectionString: process.env.DATABASE_URL! },
  llm,
  embedder,
  queue: {
    redisUrl: process.env.REDIS_URL!,
    removeOnComplete: true,
    removeOnFail: false,
  },
});

Acceptance versus visibility

In queue mode, an awaited ingestion call confirms that the job was accepted. It does not mean topic and memory nodes are already searchable. Reads must wait for the relevant worker job to complete before assuming new memory is visible.

Retry safety

The stored ingest cursor prevents reprocessing message ranges already committed successfully.

The cursor advances only after graph writes succeed.

Keep failed jobs long enough to inspect their errors.

Design application retries around job identity and observed completion rather than arbitrary delays.

Worker operations

Run workers with the same database, provider, embedding, and Redis configuration as producers.

Monitor queue depth, job age, failures, provider latency, and database errors.

Use deliberate concurrency limits to protect providers and PostgreSQL.

Close queue and worker resources during graceful shutdown.

Failure behavior

Redis connection problems are warnings in normal chatbot invocation paths, but production systems still need alerting because accepted-looking conversations may not become graph memory. Required database or provider failures during worker execution should surface through failed-job monitoring.

Common mistakes

Expecting immediate recall after an enqueue.

Removing failed jobs before recording their errors.

Running workers with a different embedding model or dimension.

Using sleep timers instead of job completion state.

Stopping a process without closing queue resources.