feat: adaptive embedding rate limiting

[bigint]

No description provided.

[greptile-apps]

Greptile Summary

This PR introduces an adaptive AIMD-based embedding rate limiter (embedding_gate) that replaces the previous fixed-semaphore + manual cooldown mechanism across all three embedding providers (OpenAI, Cohere, Voyage). It also migrates the admin UI from WebSocket push (the entire services/realtime tree is removed) to HTTP polling via new /v1/status/* REST endpoints, and adds a dedicated CollectionsStatusResponse for document aggregate stats.

• embedding_gate.py: Redis-backed inflight ZSET with Lua scripts for atomic acquire/release and AIMD limit updates (multiplicative decrease on 429, additive increase on success). Falls back to a local asyncio.Condition-based limiter when Redis is unavailable, and fails open if Redis throws during acquire.
• Provider cleanup: All three embedding providers now delegate semaphore and cooldown logic entirely to embedding_gate; redundant wait_for_rate_limit_cooldown / record_rate_limit_cooldown calls in embed.py and embed_batches.py are removed.
• Realtime → polling migration: WebSocket router, all services/realtime/ topic handlers, and connector notify_connector_sources calls are removed; frontend hooks switch to React Query with a 5-second refetchInterval.

Confidence Score: 5/5

The change is safe to merge; the AIMD gate logic is sound and the provider migrations are mechanical. The only notable trade-off is the removal of the platform-stats cache, which could increase DB query frequency under active polling.

The core embedding_gate implementation correctly handles Redis failures (fail-open), lease expiry, and AIMD dynamics. Provider migrations are straight substitutions with no behavioral regressions. The realtime-to-polling switch is intentional and consistently applied across backend and frontend. The one finding is the removal of the 15-second cache from platform_stats_payload while the frontend polls every 5 seconds — a DB pressure concern at scale, but not a correctness issue.

api/bigrag/services/platform_stats.py — caching was removed while the frontend poll interval is 5 seconds

Important Files Changed

Filename	Overview
api/bigrag/services/embedding_gate.py	New AIMD-based concurrency gate with Redis-backed inflight ZSET and local fallback; core logic looks correct, but _on_success emits no log (noted in prior review) and reset_embedding_limiters leaves Redis limit keys intact (noted in prior review)
api/bigrag/services/embedding/voyage.py	Replaced semaphore + manual cooldown recording with embedding_gate; HTTP 429 is raised inside the context manager so _on_rate_limited fires correctly
api/bigrag/services/embedding/openai.py	Migrated to embedding_gate; rate limit handling delegated to gate correctly
api/bigrag/services/embedding/cohere.py	Migrated to embedding_gate; rate limit handling delegated to gate correctly
api/bigrag/services/queue_embedding/embed_batches.py	Removed cooldown_key parameter and duplicate record_rate_limit_cooldown call; batch-level sleep and gate-level cooldown coexist cleanly
api/bigrag/services/platform_stats.py	Removed 15-second cache from platform_stats_payload; combined with 5-second frontend polling this creates uncached DB load on every request
api/bigrag/services/status.py	New service extracting collection-level document aggregates into a dedicated CollectionsStatusResponse; query uses FILTER (WHERE ...) aggregate syntax correctly for PostgreSQL
api/bigrag/routers/status.py	New REST status router replaces the removed WebSocket realtime router; all endpoints properly auth-gated
api/bigrag/app_factory/routers.py	One-for-one swap of realtime_router with status_router; the old WebSocket endpoint and realtime-token endpoint are intentionally removed as part of the polling migration
api/bigrag/services/queue_embedding/embed.py	Removed redundant wait_for_rate_limit_cooldown and record_rate_limit_cooldown calls that are now handled inside embedding_gate
app/src/hooks/use-platform.ts	Migrated from WebSocket snapshot queries to React Query polling at 5-second interval; straightforward replacement

_{Reviews (4): Last reviewed commit: "fix: remove collections status poll hook" | Re-trigger Greptile}

[greptile-apps]

_on_success emits no log; spec promises embedding limit recovered

The design spec (under "Observability") lists two log events: embedding limit decreased and embedding limit recovered, each with endpoint, old_limit, new_limit, inflight. The plan's _on_success function returns silently with no logger.* call, so an agent implementing it verbatim will produce one-sided logs where operators can observe the limit dropping but never see it climbing back. The _on_rate_limited log is also missing the old_limit and inflight fields the spec lists. Updating _on_success to emit a logger.info("embedding limit recovered", ...) and adding the missing fields to _on_rate_limited before the plan is executed will keep the implementation consistent with the spec.

[greptile-apps]

reset_embedding_limiters only clears _local_limiters but leaves the Redis LIMIT_PREFIX keys untouched. When an admin changes embedding_concurrency in runtime settings, this function fires — but for multi-worker Redis-backed deployments, the authoritative AIMD limit stored in Redis is unaffected. Workers will continue operating at whatever throttled value was last written (potentially MIN_LIMIT = 1.0 after heavy rate limiting), ignoring the new ceiling until it gradually self-heals via _SUCCESS_LUA. In the old code, reset_embedding_semaphores destroyed and recreated semaphores immediately with the new value. The Redis counterpart keys should also be deleted here so the new ceiling takes effect right away.

Suggested change

	def reset_embedding_limiters() -> None:
	_local_limiters.clear()
	def reset_embedding_limiters() -> None:
	_local_limiters.clear()
	redis = redis_cache.get_redis()
	if redis is None:
	return
	try:
	import asyncio
	async def _del_limit_keys() -> None:
	keys = await redis.keys(LIMIT_PREFIX + "*")
	if keys:
	await redis.delete(*keys)
	asyncio.get_event_loop().run_until_complete(_del_limit_keys())
	except Exception:
	pass