feat: Configurable embedding concurrency and rate-limit handling (INDEXING_MAX_WORKERS_NUMBER, EMBEDDING_BATCH_DELAY)

## Summary

When indexing documents into a knowledge base with an OpenAI-compatible embedding provider that has API rate limits, Dify fails with **429 Too Many Requests** errors. After deep debugging, we found the problem exists at **three independent concurrency layers**, none of which are currently configurable:

```
Layer 3: CELERY_WORKER_AMOUNT (default 4)
└─ multiple documents indexed IN PARALLEL as separate tasks
Layer 2: ThreadPoolExecutor(max_workers=10) — hardcoded
└─ per-document chunk groups embedded concurrently
Layer 1: max_chunks (openai_api_compatible defaults to 1)
└─ every single text = one HTTP request
→ no delay, no backoff, no retry on 429
```

With default settings, aggregate request rate can easily reach **10-40 requests/second**, which exceeds most self-hosted or free-tier embedding endpoints.

## Environment

- Dify version: 1.16.1 (Docker)
- Embedding provider: OpenAI-compatible (`bge-m3` behind an API router)
- Index technique: high_quality, parent-child chunking
- Vector DB: Weaviate

## Root Cause Analysis

### 1. Hardcoded per-document concurrency — `api/core/indexing_runner.py:643`

```python
max_workers = 10 # hardcoded, no config
with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
```

### 2. Cross-document parallelism via Celery

Even after fixing layer 1, **each document is a separate Celery task**. With `CELERY_WORKER_AMOUNT=4`, four documents embed simultaneously — each spawning its own ThreadPoolExecutor. Notably, `retry_document_indexing_task` appears to bypass the tenant isolation queue used by new-document indexing (`priority_document_indexing_task`), so "retry all failed" triggers full parallelism again.

### 3. No throttling between requests — `api/core/rag/embedding/cached_embedding.py:60-70`

```python
max_chunks = model_schema.model_properties.get(ModelPropertyKey.MAX_CHUNKS, 1)
for i in range(0, len(embedding_queue_texts), max_chunks):
batch_texts = embedding_queue_texts[i : i + max_chunks]
embedding_result = self._model_instance.invoke_text_embedding(...) # fires immediately, no delay
```

With `openai_api_compatible`, `max_chunks` defaults to 1, so a 4000-chunk document produces 4000 back-to-back HTTP requests. There is **no exponential backoff and no retry** — a single 429 fails the whole document.

### Related issues

- #28720 (Gemini 429 RESOURCE_EXHAUSTED when chunks > 200)
- #33053 (rate limit causes everything to restart)
- #34837 (document queuing)
- langgenius/dify-official-plugins#2866 (max_chunks not configurable)

## Proposed Solution

Add three environment variables:

**1. `api/configs/feature/__init__.py`**

```python
class IndexingConfig(BaseSettings):
# Per-document embedding thread pool size
INDEXING_MAX_WORKERS_NUMBER: PositiveInt = Field(
description="Concurrent workers for embedding indexing within one document. Lower values help avoid API rate limits.",
default=10,
)
# Delay between embedding HTTP requests (seconds)
EMBEDDING_BATCH_DELAY: float = Field(
description="Delay in seconds between embedding requests to avoid provider rate limits. 0 disables.",
default=0,
)
```

**2. `api/core/indexing_runner.py`**

```python
max_workers = dify_config.INDEXING_MAX_WORKERS_NUMBER
```

**3. `api/core/rag/embedding/cached_embedding.py`** — add delay + basic backoff:

```python
for i in range(0, len(embedding_queue_texts), max_chunks):
batch_texts = embedding_queue_texts[i : i + max_chunks]
embedding_result = self._model_instance.invoke_text_embedding(
texts=batch_texts, input_type=EmbeddingInputType.DOCUMENT
)
if dify_config.EMBEDDING_BATCH_DELAY > 0:
time.sleep(dify_config.EMBEDDING_BATCH_DELAY)
```

Ideally also: catch 429 errors and retry with exponential backoff instead of failing the whole document, and make `max_chunks` configurable per credential in `openai_api_compatible` (see langgenius/dify-official-plugins#2866).

**4. `docker/.env.example`** documentation:

```bash
# Embedding rate-limit protection
# Concurrency of chunk embedding threads per document (default 10; lower to 1-2 for strict rate limits)
INDEXING_MAX_WORKERS_NUMBER=
# Seconds to wait between embedding requests (default 0)
EMBEDDING_BATCH_DELAY=
```

### Recommended values

| Scenario | INDEXING_MAX_WORKERS_NUMBER | EMBEDDING_BATCH_DELAY | CELERY_WORKER_AMOUNT |
|---|---|---|---|
| Strict limit (self-hosted ollama/xinference, free tier) | 1 | 0.5–1 | 1–2 |
| Normal cloud providers | 4 | 0 | 4 |
| High quota (OpenAI etc.) | 10 | 0 | 4+ |

## Verified Workaround (production-tested)

Without code changes, patch inside running containers:

```bash
# 1. Serialize background tasks (also protects cross-document parallelism)
CELERY_WORKER_AMOUNT=1 # in docker/.env, then: docker compose up -d worker

# 2. Single-threaded embedding + 1s delay between requests
docker exec -u root docker-worker-1 python3 -c "
p='/app/api/core/indexing_runner.py'
s=open(p).read().replace('max_workers = 10','max_workers = 1'); open(p,'w').write(s)
p='/app/api/core/rag/embedding/cached_embedding.py'
s=open(p).read().replace(
'embedding_result = self._model_instance.invoke_text_embedding(\n texts=batch_texts, input_type=EmbeddingInputType.DOCUMENT\n )',
'embedding_result = self._model_instance.invoke_text_embedding(\n texts=batch_texts, input_type=EmbeddingInputType.DOCUMENT\n )\n import time; time.sleep(1)')
open(p,'w').write(s)
"
docker restart docker-worker-1 docker-api-1
```

Result: sustained ~0.8 req/s embedding throughput, zero 429 errors over a 4000+ chunk document that previously failed repeatedly. Embedding cache ensures already-vectorized chunks are reused on retry, so failed attempts are not fully wasted.

Start with api/configs/feature/__init__.py, api/core/indexing_runner.py, and api/core/rag/embedding/cached_embedding.py to trace the existing indexing concurrency and embedding loop. Check how docker/.env.example documents configuration and verify that worker concurrency and inter-request delay are configurable without breaking indexing; rate-limit retry behavior is also proposed but not fully specified.

Written by the indexing model from the issue text.