[Router] vectorstore: bound ingestion memory with streaming batches

config/config.yaml

@@ -1839,6 +1839,7 @@ global:
       embedding_model: multimodal
       embedding_dimension: 384
       ingestion_workers: 4
+      ingestion_batch_size: 64
       ingestion_drain_timeout_seconds: 30
       supported_formats: [.txt, .md, .json, .csv, .html, .pdf]
       milvus:

dashboard/frontend/src/pages/configPageRouterDefaultsCatalog.ts

@@ -111,6 +111,7 @@ export const DEFAULT_SECTIONS: Record<RouterSystemKey, unknown> = {
     embedding_model: 'mmbert',
     embedding_dimension: 384,
     ingestion_workers: 2,
+    ingestion_batch_size: 64,
     supported_formats: ['.txt', '.md', '.json', '.csv', '.html'],
     memory: {
       max_entries_per_store: 100000,

dashboard/frontend/src/pages/configPageRouterDefaultsSupport.ts

@@ -591,6 +591,7 @@ function fieldsForKey(key: RouterSystemKey): FieldConfig[] {
         { name: 'embedding_model', label: 'Embedding Model', type: 'select', options: ['bert', 'qwen3', 'gemma', 'mmbert', 'multimodal'] },
         { name: 'embedding_dimension', label: 'Embedding Dimension', type: 'number', placeholder: '384' },
         { name: 'ingestion_workers', label: 'Ingestion Workers', type: 'number', placeholder: '2' },
+        { name: 'ingestion_batch_size', label: 'Ingestion Batch Size', type: 'number', placeholder: '64' },
         routerStructuredField(key, 'supported_formats'),
         routerStructuredField(key, 'memory'),
         routerStructuredField(key, 'milvus'),

dashboard/frontend/src/pages/configPageSupport.ts

@@ -506,6 +506,7 @@ export interface VectorStoreConfig {
   embedding_model?: string
   embedding_dimension?: number
   ingestion_workers?: number
+  ingestion_batch_size?: number
   supported_formats?: string[]
   milvus?: VectorStoreMilvusConfig
   memory?: VectorStoreMemoryConfig

deploy/helm/semantic-router/values.yaml

@@ -5200,6 +5200,7 @@ config:
         embedding_model: mmbert
         embedding_dimension: 384
         ingestion_workers: 2
+        ingestion_batch_size: 64
         ingestion_drain_timeout_seconds: 30
         supported_formats:
           - .txt
@@ -10619,6 +10620,7 @@ config:
         embedding_model: mmbert
         embedding_dimension: 384
         ingestion_workers: 2
+        ingestion_batch_size: 64
         ingestion_drain_timeout_seconds: 30
         supported_formats:
           - .txt

e2e/profiles/rag-hybrid-search/values.yaml

@@ -54,6 +54,7 @@ config:
         embedding_model: mmbert
         embedding_dimension: 384
         ingestion_workers: 2
+        ingestion_batch_size: 64
         ingestion_drain_timeout_seconds: 30
         supported_formats:
           - .txt

src/semantic-router/pkg/config/vectorstore.go

@@ -46,6 +46,15 @@ type VectorStoreConfig struct {
 	// Default: 2
 	IngestionWorkers int `json:"ingestion_workers,omitempty" yaml:"ingestion_workers,omitempty"`
 
+	// IngestionBatchSize bounds how many chunks are embedded and inserted per
+	// batch during ingestion. Instead of embedding every chunk of a file and
+	// then inserting them all at once — which holds the whole file's text,
+	// chunks, and embedding vectors in memory simultaneously — the pipeline
+	// processes chunks in fixed-size windows. This bounds peak per-job memory to
+	// roughly O(batch_size × embedding_dimension) regardless of file size.
+	// Default: 64.
+	IngestionBatchSize int `json:"ingestion_batch_size,omitempty" yaml:"ingestion_batch_size,omitempty"`
+
 	// IngestionDrainTimeoutSeconds bounds how long shutdown waits for in-flight
 	// ingestion jobs to drain before cancelling them. This is a shutdown grace
 	// window, not a per-file ingest budget: on shutdown the pipeline stops
@@ -316,6 +325,9 @@ func (c *VectorStoreConfig) ApplyDefaults() {
 	if c.IngestionWorkers <= 0 {
 		c.IngestionWorkers = 2
 	}
+	if c.IngestionBatchSize <= 0 {
+		c.IngestionBatchSize = 64
+	}
 	if c.IngestionDrainTimeoutSeconds <= 0 {
 		c.IngestionDrainTimeoutSeconds = 30
 	}

src/semantic-router/pkg/routerruntime/vectorstore_runtime.go

@@ -71,6 +71,7 @@ func NewVectorStoreRuntime(cfg *config.RouterConfig) (*VectorStoreRuntime, error
 	pipeline := vectorstore.NewIngestionPipeline(backend, fileStore, manager, embedder, vectorstore.PipelineConfig{
 		Workers:   cfg.VectorStore.IngestionWorkers,
 		QueueSize: 100,
+		BatchSize: cfg.VectorStore.IngestionBatchSize,
 	})
 	pipeline.Start()
 

src/semantic-router/pkg/vectorstore/pipeline.go

@@ -46,6 +46,18 @@ type IngestionJob struct {
 // backend or embedder is wedged.
 const defaultStopTimeout = 30 * time.Second
 
+// defaultBatchSize is the number of chunks embedded and inserted per batch when
+// PipelineConfig does not specify one. It bounds peak per-job memory to roughly
+// O(batchSize × embeddingDimension) instead of holding every chunk's text and
+// embedding for a file at once.
+const defaultBatchSize = 64
+
+// batchCleanupTimeout bounds the best-effort removal of already-inserted chunks
+// when a multi-batch ingestion fails partway through. It is detached from the
+// (possibly cancelled) job context so cleanup still runs during shutdown, but
+// is itself bounded so a wedged backend cannot block the worker indefinitely.
+const batchCleanupTimeout = 10 * time.Second
+
 // IngestionPipeline processes file attachment jobs asynchronously.
 // It reads files, extracts text, chunks, embeds, and stores the
 // resulting vectors in the backend.
@@ -56,6 +68,7 @@ type IngestionPipeline struct {
 	embedder     Embedder
 	jobQueue     chan IngestionJob
 	workers      int
+	batchSize    int
 	lifecycleMu  sync.Mutex
 	mu           sync.RWMutex
 	fileStatuses map[string]*VectorStoreFile // vsf_id -> status
@@ -73,6 +86,7 @@ type IngestionPipeline struct {
 type PipelineConfig struct {
 	Workers   int // number of concurrent workers (default 2)
 	QueueSize int // job queue buffer size (default 100)
+	BatchSize int // chunks embedded+inserted per batch (default 64)
 }
 
 // NewIngestionPipeline creates a new ingestion pipeline.
@@ -85,6 +99,10 @@ func NewIngestionPipeline(backend VectorStoreBackend, fileStore *FileStore, mana
 	if queueSize <= 0 {
 		queueSize = 100
 	}
+	batchSize := cfg.BatchSize
+	if batchSize <= 0 {
+		batchSize = defaultBatchSize
+	}
 
 	return &IngestionPipeline{
 		backend:      backend,
@@ -93,6 +111,7 @@ func NewIngestionPipeline(backend VectorStoreBackend, fileStore *FileStore, mana
 		embedder:     embedder,
 		jobQueue:     make(chan IngestionJob, queueSize),
 		workers:      workers,
+		batchSize:    batchSize,
 		fileStatuses: make(map[string]*VectorStoreFile),
 		stopCh:       make(chan struct{}),
 	}
@@ -385,20 +404,15 @@ func (p *IngestionPipeline) processJob(ctx context.Context, job IngestionJob) {
 		return
 	}
 
-	// Step 5: Embed each chunk.
-	embeddedChunks, ok := p.embedChunks(ctx, job, record.Filename, chunks)
-	if !ok {
-		return
-	}
-
-	if err := ctx.Err(); err != nil {
-		p.failJob(ctx, job, "cancelled", "ingestion cancelled before storage")
-		return
-	}
-
-	// Step 6: Insert into backend.
-	if err := p.backend.InsertChunks(ctx, job.VectorStoreID, embeddedChunks); err != nil {
-		p.failJob(ctx, job, "storage_error", fmt.Sprintf("failed to store chunks: %v", err))
+	// Steps 5 & 6: Embed and store chunks in bounded batches.
+	//
+	// Rather than embedding every chunk and then inserting them all at once —
+	// which holds the whole file's chunk text and embedding vectors in memory
+	// simultaneously — process chunks in fixed-size windows. This bounds peak
+	// per-job memory to roughly O(batchSize × embeddingDimension) regardless of
+	// file size. ctx is checked before each batch so a cancelled lifecycle
+	// aborts promptly between batches.
+	if !p.embedAndStoreBatches(ctx, job, record.Filename, chunks) {
 		return
 	}
 
@@ -410,6 +424,75 @@ func (p *IngestionPipeline) processJob(ctx context.Context, job IngestionJob) {
 	})
 }
 
+// embedAndStoreBatches embeds and inserts chunks in fixed-size windows of
+// p.batchSize. It returns true when every chunk has been stored, and false when
+// a batch failed or the job was cancelled — in which case it has already marked
+// the job failed and, if any earlier batch had already been inserted, made a
+// best-effort attempt to remove the partial chunks so a failed file leaves no
+// searchable state (full transactional reconciliation is tracked separately in
+// #2474).
+func (p *IngestionPipeline) embedAndStoreBatches(ctx context.Context, job IngestionJob, filename string, chunks []TextChunk) bool {
+	inserted := false
+	for start := 0; start < len(chunks); start += p.batchSize {
+		end := start + p.batchSize
+		if end > len(chunks) {
+			end = len(chunks)
+		}
+
+		if err := ctx.Err(); err != nil {
+			p.failBatchJob(ctx, job, inserted, "cancelled",
+				fmt.Sprintf("ingestion cancelled before embedding batch at chunk %d", start))
+			return false
+		}
+
+		embeddedChunks, ok := p.embedChunks(ctx, job, filename, chunks[start:end])
+		if !ok {
+			// embedChunks already recorded the failure status; still clean up any
+			// chunks inserted by earlier batches so the failed file is not
+			// partially searchable.
+			p.cleanupPartialChunks(ctx, job, inserted)
+			return false
+		}
+
+		if err := ctx.Err(); err != nil {
+			p.failBatchJob(ctx, job, inserted, "cancelled",
+				fmt.Sprintf("ingestion cancelled before storing batch at chunk %d", start))
+			return false
+		}
+
+		if err := p.backend.InsertChunks(ctx, job.VectorStoreID, embeddedChunks); err != nil {
+			p.failBatchJob(ctx, job, inserted, "storage_error",
+				fmt.Sprintf("failed to store chunks: %v", err))
+			return false
+		}
+		inserted = true
+	}
+	return true
+}
+
+// failBatchJob marks a job failed and, when earlier batches were already
+// inserted, removes the partial chunks first so a failed file leaves no
+// searchable state.
+func (p *IngestionPipeline) failBatchJob(ctx context.Context, job IngestionJob, inserted bool, code, message string) {
+	p.cleanupPartialChunks(ctx, job, inserted)
+	p.failJob(ctx, job, code, message)
+}
+
+// cleanupPartialChunks best-effort removes chunks already inserted for a job
+// whose ingestion failed partway through. The delete is detached from the
+// job context (which may be cancelled during shutdown) but bounded by
+// batchCleanupTimeout so a wedged backend cannot block the worker. Failure to
+// clean up is logged implicitly by leaving the file marked failed; it is not
+// surfaced as a separate error because the job is already failing.
+func (p *IngestionPipeline) cleanupPartialChunks(ctx context.Context, job IngestionJob, inserted bool) {
+	if !inserted {
+		return
+	}
+	cleanupCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), batchCleanupTimeout)
+	defer cancel()
+	_ = p.backend.DeleteByFileID(cleanupCtx, job.VectorStoreID, job.FileID)
+}
+
 // embedChunks embeds each chunk, checking ctx before each embedding call so a
 // cancelled lifecycle context aborts promptly. On any error it fails the job
 // and returns ok=false; the caller should stop processing.