Pipelines

The pipeline API keeps the code you need to index data simple and organized. A typical pipeline has two stages: extraction and chunking on CPU, followed by embedding on GPU. This guide walks through a best-practice layout for that pipeline; the concepts expand to N stages.

A Pipeline resource declares the scaling characteristics you want for ingesting data. Ingestion typically runs in stages: a CPU stage for chunking and extraction, followed by a GPU stage for embedding. You can declare the spec in YAML, from code through the pipeline API, or a combination of both — it is recommended you declare your pipeline scaling characteristics in YAML while setting your namespace via the client. spec.sourceRef lets you declare your pipeline’s upstream details as well — the operator hands it to the worker as an environment variable, so the worker reads its source from config instead of hardcoding it.

apiVersion: hevlayer.com/v1alpha1
kind: Pipeline
metadata:
  name: product-images
  namespace: layer
spec:
  target:
    namespace: products
  sourceRef:
    kind: sqs
    queueUrl: https://sqs.us-east-1.amazonaws.com/123456789/product-images
  schedule:
    cron: "0 2 * * *"
    leaseSeconds: 600
  worker:
    image: <acct>.dkr.ecr.us-east-1.amazonaws.com/hev-product-image-worker:latest
    computeClass: cpu
    batchSize: 64
    timeoutSeconds: 60
  scaling:
    pool: cpu
    mode: autoscale
    replicas:
      min: 0
      max: 8

Document lifecycle

              put chunks             put vectors
  (new doc) ──────────► pending ──────────────► indexed
                           ▲
                           │ re-stage (idempotent)

• pending — chunks stored, waiting for embedding.
• indexed — vectors written to the namespace’s configured VectorStore.

embedding is a claim stage: documents sit in it only while leased to a worker, and recover to pending when a lease expires. Re-staging a document resets it to pending with new chunks, which is how you reprocess after source data changes.

File tree

indexer/
├── pipelines/
│   ├── extract-chunk.yaml   # CPU stage — Pipeline resource
│   └── embed.yaml           # GPU stage — Pipeline resource
├── extract_chunk.py         # read the source, stage chunks
├── embed.py                 # claim pending docs, write vectors
└── app.py                   # REST API: trigger a run, wait for completion

The two YAML files declare the worker images, pools, and scaling — see Target through Scaling below for the fields. Both set pipelineId: products so the two workers share one queue. The rest of this page is the worker code — shown in Python and Go; every call is also a plain REST endpoint (see Write & Stage).

Extract and chunk

The CPU worker reads the source, splits text into chunks, and stages them. Staging chunks stores them durably (S3, cached in the document cache) and marks the document pending.

The worker hardcodes nothing: the operator injects the pipeline id, the gateway URL, and spec.sourceRef as environment variables — see the worker variables below. The queue URL below comes from the sourceRef declared in pipelines/extract-chunk.yaml.

# extract_chunk.py
import asyncio
import json
import os

import boto3
from hevlayer import AsyncHevlayer

PIPELINE = os.environ["HEVLAYER_PIPELINE_ID"]
SOURCE = json.loads(os.environ["HEVLAYER_SOURCE_REF"])
sqs = boto3.client("sqs")

def chunks(text: str, size: int = 800) -> list[str]:
    return [text[i : i + size] for i in range(0, len(text), size)]

async def main() -> None:
    async with AsyncHevlayer(
        base_url=os.environ["HEVLAYER_BASE_URL"],
        api_key=os.environ.get("LAYER_GATEWAY_API_KEY"),
    ) as layer:
        while True:
            batch = sqs.receive_message(
                QueueUrl=SOURCE["queueUrl"], MaxNumberOfMessages=10,
            ).get("Messages", [])
            for m in batch:
                doc = json.loads(m["Body"])
                await layer.put_pipeline_document_chunks(PIPELINE, doc["id"], {
                    "chunks": [
                        {"id": f"{doc['id']}-{i}", "text": t}
                        for i, t in enumerate(chunks(doc["text"]))
                    ],
                })
                sqs.delete_message(QueueUrl=SOURCE["queueUrl"], ReceiptHandle=m["ReceiptHandle"])

asyncio.run(main())

// extract_chunk.go
package main

import (
	"context"
	"encoding/json"
	"fmt"
	"os"

	"github.com/aws/aws-sdk-go-v2/config"
	"github.com/aws/aws-sdk-go-v2/service/sqs"
	hevlayer "github.com/hev/layer-go"
)

func chunks(text string, size int) []string {
	var out []string
	for i := 0; i < len(text); i += size {
		out = append(out, text[i:min(i+size, len(text))])
	}
	return out
}

func main() {
	ctx := context.Background()
	pipeline := os.Getenv("HEVLAYER_PIPELINE_ID")
	var source struct {
		QueueURL string `json:"queueUrl"`
	}
	json.Unmarshal([]byte(os.Getenv("HEVLAYER_SOURCE_REF")), &source)

	cfg, _ := config.LoadDefaultConfig(ctx)
	queue := sqs.NewFromConfig(cfg)
	layer := hevlayer.NewClient(
		hevlayer.WithBaseURL(os.Getenv("HEVLAYER_BASE_URL")),
		hevlayer.WithAPIKey(os.Getenv("LAYER_GATEWAY_API_KEY")),
	)

	for {
		batch, err := queue.ReceiveMessage(ctx, &sqs.ReceiveMessageInput{
			QueueUrl:            &source.QueueURL,
			MaxNumberOfMessages: 10,
		})
		if err != nil {
			continue
		}
		for _, m := range batch.Messages {
			var doc struct {
				ID   string `json:"id"`
				Text string `json:"text"`
			}
			json.Unmarshal([]byte(*m.Body), &doc)
			var staged []hevlayer.Chunk
			for i, t := range chunks(doc.Text, 800) {
				staged = append(staged, hevlayer.Chunk{ID: fmt.Sprintf("%s-%d", doc.ID, i), Text: t})
			}
			layer.PutPipelineDocumentChunks(ctx, pipeline, doc.ID,
				&hevlayer.PutChunksRequest{Chunks: staged})
			queue.DeleteMessage(ctx, &sqs.DeleteMessageInput{
				QueueUrl:      &source.QueueURL,
				ReceiptHandle: m.ReceiptHandle,
			})
		}
	}
}

// extract_chunk.ts
import {
  DeleteMessageCommand,
  ReceiveMessageCommand,
  SQSClient,
} from "@aws-sdk/client-sqs";
import { Hevlayer } from "hevlayer";

const PIPELINE = process.env.HEVLAYER_PIPELINE_ID!;
const SOURCE = JSON.parse(process.env.HEVLAYER_SOURCE_REF!);
const sqs = new SQSClient({});
const layer = new Hevlayer({
  baseUrl: process.env.HEVLAYER_BASE_URL,
  apiKey: process.env.LAYER_GATEWAY_API_KEY,
});

function chunks(text: string, size = 800): string[] {
  const out: string[] = [];
  for (let i = 0; i < text.length; i += size) out.push(text.slice(i, i + size));
  return out;
}

while (true) {
  const batch = await sqs.send(new ReceiveMessageCommand({
    QueueUrl: SOURCE.queueUrl,
    MaxNumberOfMessages: 10,
  }));
  for (const message of batch.Messages ?? []) {
    const doc = JSON.parse(message.Body ?? "{}");
    await layer.putPipelineDocumentChunks(PIPELINE, doc.id, {
      chunks: chunks(doc.text).map((text, i) => ({ id: `${doc.id}-${i}`, text })),
    });
    await sqs.send(new DeleteMessageCommand({
      QueueUrl: SOURCE.queueUrl,
      ReceiptHandle: message.ReceiptHandle,
    }));
  }
}

Embed

The GPU worker claims pending documents, reads their chunks back, and writes vectors. Writing vectors upserts to the namespace’s configured VectorStore and marks the document indexed. Claims are leased, so a worker that crashes loses nothing. For multivector namespaces on a kind: search store, send vectors: [[...], [...]] on an entry instead of vector: [...]; the gateway forwards the bag and caches the first inner vector for nearest_to_id lookup.

# embed.py
import asyncio
import os

from hevlayer import AsyncHevlayer
from sentence_transformers import SentenceTransformer

PIPELINE = os.environ["HEVLAYER_PIPELINE_ID"]
model = SentenceTransformer("all-MiniLM-L6-v2")

async def main() -> None:
    async with AsyncHevlayer(
        base_url=os.environ["HEVLAYER_BASE_URL"],
        api_key=os.environ.get("LAYER_GATEWAY_API_KEY"),
    ) as layer:
        while True:
            claimed = await layer.claim_documents(PIPELINE, {
                "stage": "pending",
                "claim_stage": "embedding",
                "limit": 16,
                "worker_id": "embed-0",
            })
            for doc_id in claimed.documents:
                doc_chunks = await layer.get_pipeline_document_chunks(PIPELINE, doc_id)
                vectors = model.encode([c.text for c in doc_chunks])
                await layer.put_pipeline_document_vectors(PIPELINE, doc_id, {
                    "vectors": [
                        {"id": c.id, "vector": v.tolist(), "attributes": {"text": c.text}}
                        for c, v in zip(doc_chunks, vectors)
                    ],
                })

asyncio.run(main())

// embed.go
package main

import (
	"context"
	"os"

	hevlayer "github.com/hev/layer-go"
)

func main() {
	ctx := context.Background()
	pipeline := os.Getenv("HEVLAYER_PIPELINE_ID")
	layer := hevlayer.NewClient(
		hevlayer.WithBaseURL(os.Getenv("HEVLAYER_BASE_URL")),
		hevlayer.WithAPIKey(os.Getenv("LAYER_GATEWAY_API_KEY")),
	)

	for {
		claimed, err := layer.ClaimDocuments(ctx, pipeline, &hevlayer.ClaimDocumentsRequest{
			Stage:      "pending",
			ClaimStage: "embedding",
			Limit:      16,
			WorkerID:   "embed-0",
		})
		if err != nil {
			continue
		}
		for _, docID := range claimed.Documents {
			docChunks, err := layer.GetPipelineDocumentChunks(ctx, pipeline, docID)
			if err != nil {
				continue
			}
			texts := make([]string, len(*docChunks))
			for i, c := range *docChunks {
				texts[i] = c.Text
			}
			vectors := embed(texts) // your embedding model or service
			entries := make([]hevlayer.VectorEntry, len(*docChunks))
			for i, c := range *docChunks {
				entries[i] = hevlayer.VectorEntry{
					ID:         c.ID,
					Vector:     vectors[i],
					Attributes: map[string]interface{}{"text": c.Text},
				}
			}
			layer.PutPipelineDocumentVectors(ctx, pipeline, docID,
				&hevlayer.PutVectorsRequest{Vectors: entries})
		}
	}
}

// embed.ts
import { Hevlayer } from "hevlayer";

const PIPELINE = process.env.HEVLAYER_PIPELINE_ID!;
const layer = new Hevlayer({
  baseUrl: process.env.HEVLAYER_BASE_URL,
  apiKey: process.env.LAYER_GATEWAY_API_KEY,
});

while (true) {
  const claimed = await layer.claimDocuments(PIPELINE, {
    stage: "pending",
    claim_stage: "embedding",
    limit: 16,
    worker_id: "embed-0",
  });
  for (const docId of claimed.documents) {
    const docChunks = await layer.getPipelineDocumentChunks(PIPELINE, docId);
    const vectors = await embed(docChunks.map((chunk) => chunk.text));
    await layer.putPipelineDocumentVectors(PIPELINE, docId, {
      vectors: docChunks.map((chunk, i) => ({
        id: chunk.id,
        vector: vectors[i],
        attributes: { text: chunk.text },
      })),
    });
  }
}

Write staged rows

A stage that doesn’t compute vectors can instead write a document’s staged rows as ordinary namespace rows, and let the namespace’s schema embed them natively. The stock PDF extraction write stage works this way.

curl -X PUT "$LAYER_GATEWAY_URL/v2/pipelines/$PIPELINE/documents/$DOC_ID/rows" \
  -H "Authorization: Bearer $LAYER_GATEWAY_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "upsert_rows": [
      {"id": "'"$DOC_ID"'#p1", "_hevlayer_parent_id": "'"$DOC_ID"'", "text": "Page one."}
    ],
    "distance_metric": "cosine_distance",
    "schema": {"text": {"type": "string", "full_text_search": true,
                        "embed": {"model": "voyage/voyage-4"}}}
  }'

upsert_rows may contain only the document’s staged row ids, and each row’s _hevlayer_parent_id must be the document id. schema and distance_metric are optional and follow the write API. The usual reserved-attribute rules apply.

The route does not mark the document indexed. The worker completes its claim afterwards through the stage API, so a crash between the two leaves the document to be claimed again.

The generated Go, Python and TypeScript clients include this route.

Write several documents’ rows in one request

Every namespace write takes the namespace’s write path one at a time, and each write carries a fixed cost (the Function write guard, the store round trip and the embedding call) that does not shrink with the number of rows. A stage that finishes many small documents can share that cost:

curl -X PUT "$LAYER_GATEWAY_URL/v2/pipelines/$PIPELINE/rows" \
  -H "Authorization: Bearer $LAYER_GATEWAY_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "documents": [
      {"doc_id": "a", "upsert_rows": [{"id": "a#p1", "_hevlayer_parent_id": "a", "text": "Page one."}]},
      {"doc_id": "b", "upsert_rows": [{"id": "b#p1", "_hevlayer_parent_id": "b", "text": "Page one."}]}
    ],
    "distance_metric": "cosine_distance",
    "schema": {"text": {"type": "string", "embed": {"model": "voyage/voyage-4"}}}
  }'

Up to 64 documents and 2,048 rows. Each document is validated as in the per-document route, and one invalid document rejects the whole request before anything is written. The route does not mark any document indexed: complete each claim only after the request succeeds. After an error or an unknown outcome, release the documents and write them again; page ids are stable, so the rewrite is idempotent. Page patches stay per document.

Build the two workers into the images your YAML references and push them to a registry your cluster can pull — Layer does not build images. Then apply the resources:

kubectl apply -f pipelines/

The operator creates one Deployment per resource and the embed pool’s KEDA object. Order doesn’t matter here: the app creates the gateway pipeline before it enqueues a batch (staging into a pipeline id that doesn’t exist returns 404), so workers never see a missing pipeline. Nothing else to wire: the CRD types themselves install with the Helm chart.

Trigger a run

The app exposes the pipeline to the rest of your system as one endpoint: POST /index-runs sends a batch to the source queue, then waits for the run to complete and returns the snapshot it produced. The pipeline is created on first use — this is where the target namespace is set in code.

Layer deduplicates create_missing staging by pipeline and document id. Large staging calls publish queue segments incrementally, so downstream workers can claim committed rows and KEDA can observe their depth while the same call is still adding later segments.

# app.py
import asyncio
import json
import os
import time

import boto3
from fastapi import FastAPI
from hevlayer import AsyncHevlayer, HevlayerError

QUEUE = "https://sqs.us-east-1.amazonaws.com/123456789/product-updates"
sqs = boto3.client("sqs")
app = FastAPI()
layer = AsyncHevlayer(
    base_url=os.environ["HEVLAYER_BASE_URL"],
    api_key=os.environ.get("LAYER_GATEWAY_API_KEY"),
)

@app.post("/index-runs")
async def index_run(documents: list[dict]) -> dict:
    started_ms = int(time.time() * 1000)
    try:
        await layer.create_pipeline({"id": "products", "target_namespace": "products"})
    except HevlayerError as e:
        if e.status_code != 409:  # 409: already exists
            raise
    for doc in documents:
        sqs.send_message(QueueUrl=QUEUE, MessageBody=json.dumps(doc))
    await drain()
    sha = await next_snapshot(after_ms=started_ms)
    return {"documents": len(documents), "snapshot": sha}

// app.go
var (
	queueURL = "https://sqs.us-east-1.amazonaws.com/123456789/product-updates"
	queue    *sqs.Client // sqs.NewFromConfig in main
	layer    = hevlayer.NewClient(
		hevlayer.WithBaseURL(os.Getenv("HEVLAYER_BASE_URL")),
		hevlayer.WithAPIKey(os.Getenv("LAYER_GATEWAY_API_KEY")),
	)
)

func indexRun(w http.ResponseWriter, r *http.Request) {
	ctx := r.Context()
	startedMs := time.Now().UnixMilli()
	var documents []map[string]interface{}
	json.NewDecoder(r.Body).Decode(&documents)

	_, err := layer.CreatePipeline(ctx, &hevlayer.CreatePipelineRequest{
		ID:              "products",
		TargetNamespace: "products",
	})
	var herr *hevlayer.HevlayerError
	if err != nil && !(errors.As(err, &herr) && herr.StatusCode == 409) { // 409: already exists
		http.Error(w, err.Error(), http.StatusBadGateway)
		return
	}
	for _, doc := range documents {
		body, _ := json.Marshal(doc)
		mb := string(body)
		queue.SendMessage(ctx, &sqs.SendMessageInput{QueueUrl: &queueURL, MessageBody: &mb})
	}
	drain(ctx)
	sha := nextSnapshot(ctx, startedMs)
	json.NewEncoder(w).Encode(map[string]interface{}{
		"documents": len(documents),
		"snapshot":  sha,
	})
}

// app.ts
import { SendMessageCommand, SQSClient } from "@aws-sdk/client-sqs";
import { Hevlayer } from "hevlayer";

const queueUrl = "https://sqs.us-east-1.amazonaws.com/123456789/product-updates";
const queue = new SQSClient({});
const layer = new Hevlayer({
  baseUrl: process.env.HEVLAYER_BASE_URL,
  apiKey: process.env.LAYER_GATEWAY_API_KEY,
});

async function indexRun(documents: Record<string, unknown>[]) {
  const startedMs = Date.now();
  await layer.ensurePipeline({ id: "products", target_namespace: "products" });
  for (const doc of documents) {
    await queue.send(new SendMessageCommand({
      QueueUrl: queueUrl,
      MessageBody: JSON.stringify(doc),
    }));
  }
  await drain();
  return { documents: documents.length, snapshot: await nextSnapshot(startedMs) };
}

Wait for completion

A run is complete in two steps: the queue drains, then the consistency watcher observes the namespace stable and writes a snapshot past the run’s watermark. pending_count is the same signal KEDA scales on — when it reaches zero, the embed pool scales back to zero. status is waiting_on_upstream when a downstream worker has no pending rows to claim while upstream stages are still active. If a worker cannot load a document’s durable chunk payload after the configured retry budget, Layer moves that document to failed and adds the dead-letter reason to failed_reasons, such as {"chunks_unavailable": 2}. The snapshot SHA addresses facet listings and counts exact at that watermark; flip your application to it.

# app.py
async def drain() -> None:
    while True:
        status = await layer.get_pipeline_status("products")
        if status.pending_count == 0:  # status.counts: {"pending": 0, "indexed": 8530}
            return
        await asyncio.sleep(10)

async def next_snapshot(after_ms: int) -> str:
    while True:
        history = await layer.list_namespace_history("products", limit=1)
        if history and history[0].watermark_ms >= after_ms:
            return history[0].sha
        await asyncio.sleep(30)

// app.go
func drain(ctx context.Context) {
	for {
		status, err := layer.GetPipelineStatus(ctx, "products")
		if err == nil && status.PendingCount == 0 { // status.Counts: {"pending": 0, "indexed": 8530}
			return
		}
		time.Sleep(10 * time.Second)
	}
}

func nextSnapshot(ctx context.Context, afterMs int64) string {
	for {
		history, err := layer.ListNamespaceHistory(ctx, "products",
			&hevlayer.ListNamespaceHistoryParams{Limit: 1})
		if err == nil && len(history) > 0 && history[0].WatermarkMs >= afterMs {
			return history[0].Sha
		}
		time.Sleep(30 * time.Second)
	}
}

// app.ts
const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms));

async function drain() {
  while (true) {
    const status = await layer.getPipelineStatus("products");
    if (status.pending_count === 0) return;
    await sleep(10_000);
  }
}

async function nextSnapshot(afterMs: number): Promise<string> {
  while (true) {
    const history = await layer.listNamespaceHistory("products", { limit: 1 });
    if (history.length > 0 && history[0].watermark_ms >= afterMs) {
      return history[0].sha;
    }
    await sleep(30_000);
  }
}

Once vectors are indexed, query and fetch them through the namespace API — see Query & Fetch.

Target

spec.target.namespace is the turbopuffer namespace the pipeline writes. The gateway pipeline API owns document state, chunks, and vector writes for that target namespace.

Pipeline id

spec.pipelineId names the gateway pipeline (the queue) the worker stages into and scales on. It defaults to the resource name. Set it when multiple worker resources share one queue: the extract and embed stages of a two-stage pipeline both set pipelineId: products.

Source

spec.sourceRef declares the external source that feeds the worker. Its kind selects how the operator treats it.

For open kinds — SQS, Kafka, S3 events, a partner API, a one-off migration — sourceRef is arbitrary JSON injected into the worker pod verbatim as HEVLAYER_SOURCE_REF; the worker image owns source-specific behavior. See Extract and chunk for a worker reading it. For PDFs in object storage, the stock layer-extract worker reads kind: objects; see PDF extraction.

Typed sources

For warehouse-backed kinds — snowflake, huggingface, and rest — kind selects a typed shape the operator validates. The source names a Warehouse with warehouseRef; the operator resolves it (it must be Verified), mounts its credential Secret, and injects connection details as HEVLAYER_WAREHOUSE with no credential material. Set spec.worker.image to the stock worker for that kind from the mesh-account ECR registry (for example, <acct>.dkr.ecr.us-east-1.amazonaws.com/hev-huggingface-source or <acct>.dkr.ecr.us-east-1.amazonaws.com/hev-rest-source), so a typed source needs no custom code, or to your own image. worker.image is required on every Pipeline; the operator does not fill it in. The per-kind source fields are on the Warehouses page.

A REST source can enumerate an incremental JSON projection without a request.forEach block. Use pagination.kind: searchAfter, name the query parameter that receives the prior cursor, and point cursorPath at the next cursor echoed by each response. The stock worker stores the last cursor in S3 after its row writes succeed and resumes there on the next scheduled refresh. See the REST search-after example. REST attribute mappings can cast numeric strings with { path: $.total, cast: float }. Declare the target attribute as a filterable float to use inclusive numeric ranges through MCP search and aggregate. Nested API fan-out also supports list/detail endpoints; see the REST mapping and fan-out contract.

Chunking

A source’s text column is often a whole document that must be split before it is embedded. An optional chunk block declares how, with no code for the common strategies. It applies to any source whose worker honors it — the stock workers do.

spec:
  sourceRef:
    kind: huggingface
    warehouseRef: hf-public
    dataset: wikimedia/wikipedia
    config: 20231101.en
    split: train
    mapping:
      text: text
      attributes: [title, url]
    chunk:
      strategy: recursive     # none | fixed | recursive | sentence | markdown
      unit: tokens            # tokens | characters
      size: 512
      overlap: 64
      tokenizer: cl100k_base  # when unit: tokens

Each row maps to one document; text splits into chunks. The chunk is the unit indexed and embedded — a row with id {documentId}#{i} carrying the document’s attributes plus reserved _hevlayer_parent_id and _hevlayer_chunk_index. For splits the stock strategies can’t express, set spec.worker.image to your own chunker.

Schedule

spec.schedule is optional. When it is set, the operator wakes the Pipeline worker on a KEDA cron window instead of scaling it on pending pipeline queue depth:

schedule:
  cron: "0 2 * * *"      # 5-field UTC cron; minute must be a single integer
  leaseSeconds: 600      # sizes the cron window

The worker still owns source semantics: what to pull on wake, how to advance cursors, and how to stage rows. The schedule only controls when the worker runs. Scheduled Pipelines must use scaling.replicas.min: 0; the cron window is the wake trigger.

A rest source can instead keep one worker running and re-enumerate on an interval with sourceRef.sync.interval, measured from the end of each sweep. Do not set both; see Staying in sync. Each sweep writes a report to the snapshot bucket with its counts and errors.

Worker

The operator creates one Deployment per Pipeline and injects:

Scaling

scaling:
  pool: cpu
  mode: autoscale
  replicas:
    min: 0
    max: 8

spec.scaling.pool, when set, must name a pool in the built-in set or InfraRules/default. When omitted, the operator uses worker.computeClass to choose the stock cpu or gpu pool. Layer provides the well-known cpu, cpu-large, and gpu pools without requiring an InfraRules object. mode: autoscale creates a KEDA ScaledObject backed by pipeline queue depth, or by the cron window when spec.schedule is set. mode: fixed pins the Deployment to replicas.min; mode: disabled scales it to zero.

spec.scaling.warmWindowSeconds sets a cooldown (and node retention) that holds the worker warm after its queue drains — see Workload scaling.

spec.paused: true also scales the worker to zero.

Status

Use the pipeline status API for status: queue counts, stage progress, and worker state. The resource itself reports only managed object references and readiness conditions.

Failure model

• VectorStore write failures are hard: the vectors route returns 502 and the document stays in embedding for re-claim.
• Aerospike cache failures do not block chunk reads when S3 backing is present; PostgreSQL connectivity failures return 500 and should be retried with backoff. The stop-writes recovery path and the metrics to watch live in the failure-mode runbook.
• Lease expiry is handled server-side. A worker that crashes mid-embedding has its documents recovered on the next claim sweep.