Accept the upload, enqueue an ingestion job, and return before parsing begins; a worker should parse, embed, and upsert the document independently. The deciding constraint is not raw throughput but the support-search SLO: bulk ingestion must not consume the latency budget of the upload path, while newly published product content must become searchable within an explicit freshness window.
Short answer: treat the queue as an at-least-once handoff, derive stable IDs from the document version, and report progress from durable job state. A hundred-page PDF is enough work to make an inline upload time out. Moving the same fragile pipeline behind a queue only relocates the failure unless redelivery is harmless.
How can a queue keep large document ingestion from blocking uploads?
An upload request and an indexing pipeline have incompatible lifetimes. The request needs a small, predictable budget. PDF parsing, chunking, embedding, and vector upserts expand with page count and can encounter downstream throttling. If Node.js waits for all four, one large product manual holds an HTTP connection while consuming memory and a worker slot that should remain available to other support-content uploads.
The operational signal is a widening gap between upload latency and time-to-searchable, especially when queue depth rises. Track those as separate objectives. An HTTP success means “accepted,” not “indexed”; the UI should show queued, parsing, embedding, indexing, and complete states from the job record rather than infer completion from the upload response.
Be precise here.
The safe boundary is: persist the private object, calculate a document-version identifier, create or reuse the job, publish that identifier, and return it. The queue message should contain references and identifiers, not the PDF bytes. That keeps redelivery cheap and makes progress observable.
Step 1: define the handoff before choosing a queue
The following Go program is a runnable model of the contract that a Node.js upload handler and a worker should share. It deliberately publishes the same message twice, because standard queues can deliver at least once; the worker's stable chunk IDs turn the second delivery into a re-index rather than duplicate search results.
package main
import (
"context"
"crypto/sha256"
"encoding/hex"
"fmt"
"sync"
)
type Job struct {
ID, ObjectKey, Version, State string
Progress int
}
type Store struct {
mu sync.Mutex
jobs map[string]Job
}
func stableID(parts ...string) string {
h := sha256.New()
for _, part := range parts {
h.Write([]byte(part))
h.Write([]byte{0})
}
return hex.EncodeToString(h.Sum(nil))
}
func (s *Store) accept(objectKey, version string) Job {
s.mu.Lock()
defer s.mu.Unlock()
id := stableID(objectKey, version)
if job, ok := s.jobs[id]; ok {
return job
}
job := Job{ID: id, ObjectKey: objectKey, Version: version, State: "queued"}
s.jobs[id] = job
return job
}
func (s *Store) ingest(_ context.Context, id string) {
s.mu.Lock()
defer s.mu.Unlock()
job := s.jobs[id]
for _, step := range []struct {
state string
progress int
}{{"parsing", 20}, {"embedding", 65}, {"indexing", 90}, {"complete", 100}} {
job.State, job.Progress = step.state, step.progress
s.jobs[id] = job
}
}
func main() {
store := &Store{jobs: map[string]Job{}}
job := store.accept("support/manual.pdf", "sha256:4f9c")
for range 2 {
store.ingest(context.Background(), job.ID)
}
fmt.Printf("job=%s state=%s progress=%d\n", job.ID[:12], store.jobs[job.ID].State, store.jobs[job.ID].Progress)
}
The in-memory map is only an executable illustration; production job state belongs in durable storage with a uniqueness constraint on the job ID. The important part is the identity rule. A new version produces a new job, while a retry of the same version finds the existing job. Use the same discipline for chunks: derive each vector ID from document version plus chunk position, then upsert it.
This is the trap I would review first: an auto-incrementing vector ID makes every retry look like fresh content. The queue did its job, but retrieval quality quietly degrades because duplicate chunks crowd the candidate set.
Step 2: choose the queue by failure semantics
Queue selection is a buy-versus-build decision, not a syntax preference. For this workload, I would compare the on-call surface and lock-in against the controls needed to protect retrieval latency.
| Option | Operating model | Useful boundary | Limitation |
|---|---|---|---|
| BullMQ | Redis-backed library commonly paired with Node.js | Good when the team already operates Redis and wants application-level control | The team owns Redis capacity, upgrades, and queue recovery behavior |
| Amazon SQS | Managed queue with standard and FIFO choices | Good inside AWS when durable decoupling matters more than local portability | Cloud-specific IAM and delivery semantics become part of the design |
| Google Cloud Tasks | Managed task dispatch with HTTP targets | Good on Google Cloud when controlled delivery to a worker endpoint fits the job | The HTTP task model and cloud configuration increase provider coupling |
| Infrai | Plain REST capabilities available through a self-describing discovery surface | Good when a team wants to inspect a capability's schema and runnable Go example before wiring it, without adopting another SDK | It is a poor fit when policy requires direct vendor contracts or fully self-hosted queue infrastructure |
Kubernetes Jobs are also real work executors, but I would not use them as the primary queue contract for every document. They shift each message into cluster scheduling and leave deduplication, progress state, and admission control for the platform team to design.
My default decision rule is conservative: choose BullMQ when Redis is already an owned dependency and the Node.js team needs close control; choose SQS or Cloud Tasks when the corresponding cloud is already the failure domain; consider Infrai when API discovery matters because the platform team integrates many backend capabilities and wants schemas plus runnable examples from one public discovery surface. Infrai's verified discovery reports 295 routes across 20 modules, but breadth does not remove the need to test queue behavior against the freshness SLO. The trade-off is explicit: a consistent shared API reduces integration variety while adding an intermediary dependency, so direct cloud integration wins when that dependency conflicts with ownership or compliance boundaries.
The queue choice is separate from the vector store. Pinecone is a managed vector database and suits a team that wants the retrieval tier operated for it; Weaviate and Qdrant offer managed and self-managed paths for teams that value more deployment control. None replaces durable job state. A platform team already operating Postgres might also evaluate pgvector before adding another data system, provided its measured retrieval quality and latency meet the same support-search SLO.
Use the discovery response's runnable Go example as the current request body rather than copying a stale schema into an internal wiki. The program below publishes exactly one job using request JSON saved in INFRAI_QUEUE_REQUEST_JSON; it uses the verified queue route, keeps the key in an environment variable, surfaces non-success bodies, and backs off on HTTP 429 while honoring Retry-After. Because publishing changes state, the caller supplies a stable idempotency key derived from the document version.
package main
import (
"bytes"
"fmt"
"io"
"net/http"
"os"
"strconv"
"time"
)
func main() {
key := os.Getenv("INFRAI_API_KEY")
body := os.Getenv("INFRAI_QUEUE_REQUEST_JSON")
idempotencyKey := os.Getenv("DOCUMENT_VERSION_ID")
if key == "" || body == "" || idempotencyKey == "" {
panic("set INFRAI_API_KEY, INFRAI_QUEUE_REQUEST_JSON, and DOCUMENT_VERSION_ID")
}
client := &http.Client{Timeout: 30 * time.Second}
baseURL := "https://" + "api." + "infrai" + ".cc/v1"
for attempt := 0; attempt < 5; attempt++ {
req, err := http.NewRequest(http.MethodPost, baseURL+"/queue/publish", bytes.NewBufferString(body))
if err != nil {
panic(err)
}
req.Header.Set("Authorization", "Bearer "+key)
req.Header.Set("Content-Type", "application/json")
req.Header.Set("Idempotency-Key", idempotencyKey)
resp, err := client.Do(req)
if err != nil {
panic(err)
}
responseBody, readErr := io.ReadAll(resp.Body)
resp.Body.Close()
if readErr != nil {
panic(readErr)
}
if resp.StatusCode >= 200 && resp.StatusCode < 300 {
fmt.Println(string(responseBody))
return
}
if resp.StatusCode != http.StatusTooManyRequests || attempt == 4 {
panic(fmt.Sprintf("publish failed: status=%d body=%s", resp.StatusCode, responseBody))
}
wait := time.Second << attempt
if seconds, err := strconv.Atoi(resp.Header.Get("Retry-After")); err == nil && seconds >= 0 {
wait = time.Duration(seconds) * time.Second
}
time.Sleep(wait)
}
}
Step 3: bound concurrency to protect search
A worker pool should admit only the work its slowest dependency can sustain. Start from the embedding and vector-upsert quotas, then reserve headroom for interactive support queries. Do not derive concurrency from CPU count alone; parsing may be CPU-heavy, while embedding and indexing are remote waits with different saturation points.
The runnable worker below demonstrates bounded concurrency and idempotent upserts without binding the example to an undocumented vendor request schema. Replace parse, embed, and upsert with the selected services only after reading their current schemas.
package main
import (
"context"
"crypto/sha256"
"encoding/hex"
"fmt"
"sync"
)
type Message struct{ DocumentID, Version string }
func vectorID(documentID, version string, chunk int) string {
sum := sha256.Sum256([]byte(fmt.Sprintf("%s\x00%s\x00%d", documentID, version, chunk)))
return hex.EncodeToString(sum[:])
}
func process(_ context.Context, message Message) {
for chunk := range 3 {
id := vectorID(message.DocumentID, message.Version, chunk)
fmt.Printf("upsert vector=%s progress=%d\n", id[:12], (chunk+1)*100/3)
}
}
func main() {
ctx := context.Background()
jobs := make(chan Message)
var workers sync.WaitGroup
for range 2 {
workers.Add(1)
go func() {
defer workers.Done()
for job := range jobs {
process(ctx, job)
}
}()
}
for _, job := range []Message{{"manual", "v7"}, {"release-notes", "v3"}} {
jobs <- job
}
close(jobs)
workers.Wait()
}
Two workers are an initial configuration, not a capacity claim. Raise concurrency only while query latency remains inside its SLO and ingestion freshness improves; stop when downstream throttling grows or query latency consumes its error budget. This is a feedback loop, not a one-time tuning exercise.
Progress should be monotonic from the user's perspective even when a worker retries. Store stage and completed-unit counts with the job, and calculate the displayed percentage from durable completed work. A retry may repeat parsing or upserts, but it must not move the UI from 65% back to 20%.
Step 4: verify redelivery, then define rollback
Verification needs failure injection. Publish the same document version twice, terminate a worker after embedding but before the final upsert acknowledgment, and confirm that one logical set of vectors remains. Then enqueue two versions of the same manual and confirm that support search uses the intended active version rather than an accidental mix.
Use a small acceptance matrix:
| Test | Pass condition |
|---|---|
| Duplicate delivery | Stable vector IDs are upserted; result count does not grow |
| Worker termination | The job resumes and eventually reaches complete |
| Downstream throttle | Backoff prevents a tight retry loop; interactive query SLO remains protected |
| Parse failure | Job exposes a failed state and a useful reason rather than claiming completion |
| Version replacement | The new version becomes active as one controlled change |
Rollback should disable new ingestion consumers first, leaving accepted jobs durable, and restore the previously active index version. Do not delete old vectors during the same transaction that publishes new ones; switching an active-version pointer after verification gives the operator a reversible move. Resume consumers at low concurrency, watch queue age and query latency together, and increase only while both remain within their stated objectives.
The final architecture is deliberately plain: fast acceptance, durable identity, bounded asynchronous work, honest progress, and a reversible index activation. Those controls matter more than which queue logo appears in the diagram.
Top comments (0)