""" Embedding service (FastAPI). API (simple list-in, list-out; aligned by index): - POST /embed/text body: ["text1", "text2", ...] -> [[...], ...] - POST /embed/image body: ["url_or_path1", ...] -> [[...], ...] """ import logging import os import pathlib import threading import time import uuid from collections import deque from dataclasses import dataclass from logging.handlers import TimedRotatingFileHandler from typing import Any, Dict, List, Optional import numpy as np from fastapi import FastAPI, HTTPException, Request, Response from fastapi.concurrency import run_in_threadpool from config.services_config import get_embedding_backend_config from embeddings.config import CONFIG from embeddings.protocols import ImageEncoderProtocol app = FastAPI(title="saas-search Embedding Service", version="1.0.0") class _DefaultRequestIdFilter(logging.Filter): def filter(self, record: logging.LogRecord) -> bool: if not hasattr(record, "reqid"): record.reqid = "-1" return True def configure_embedding_logging() -> None: root_logger = logging.getLogger() if getattr(root_logger, "_embedding_logging_configured", False): return log_dir = pathlib.Path("logs") verbose_dir = log_dir / "verbose" log_dir.mkdir(exist_ok=True) verbose_dir.mkdir(parents=True, exist_ok=True) log_level = os.getenv("LOG_LEVEL", "INFO").upper() numeric_level = getattr(logging, log_level, logging.INFO) formatter = logging.Formatter( "%(asctime)s | reqid:%(reqid)s | %(name)s | %(levelname)s | %(message)s" ) request_filter = _DefaultRequestIdFilter() root_logger.setLevel(numeric_level) file_handler = TimedRotatingFileHandler( filename=log_dir / "embedding_api.log", when="midnight", interval=1, backupCount=30, encoding="utf-8", ) file_handler.setLevel(numeric_level) file_handler.setFormatter(formatter) file_handler.addFilter(request_filter) root_logger.addHandler(file_handler) error_handler = TimedRotatingFileHandler( filename=log_dir / "embedding_api_error.log", when="midnight", interval=1, backupCount=30, encoding="utf-8", ) error_handler.setLevel(logging.ERROR) error_handler.setFormatter(formatter) error_handler.addFilter(request_filter) root_logger.addHandler(error_handler) verbose_logger = logging.getLogger("embedding.verbose") verbose_logger.setLevel(numeric_level) verbose_logger.handlers.clear() verbose_logger.propagate = False verbose_handler = TimedRotatingFileHandler( filename=verbose_dir / "embedding_verbose.log", when="midnight", interval=1, backupCount=30, encoding="utf-8", ) verbose_handler.setLevel(numeric_level) verbose_handler.setFormatter(formatter) verbose_handler.addFilter(request_filter) verbose_logger.addHandler(verbose_handler) root_logger._embedding_logging_configured = True # type: ignore[attr-defined] configure_embedding_logging() logger = logging.getLogger(__name__) verbose_logger = logging.getLogger("embedding.verbose") # Models are loaded at startup, not lazily _text_model: Optional[Any] = None _image_model: Optional[ImageEncoderProtocol] = None _text_backend_name: str = "" open_text_model = os.getenv("EMBEDDING_ENABLE_TEXT_MODEL", "true").lower() in ("1", "true", "yes") open_image_model = os.getenv("EMBEDDING_ENABLE_IMAGE_MODEL", "true").lower() in ("1", "true", "yes") _text_encode_lock = threading.Lock() _image_encode_lock = threading.Lock() _TEXT_MICROBATCH_WINDOW_SEC = max( 0.0, float(os.getenv("TEXT_MICROBATCH_WINDOW_MS", "4")) / 1000.0 ) _TEXT_REQUEST_TIMEOUT_SEC = max( 1.0, float(os.getenv("TEXT_REQUEST_TIMEOUT_SEC", "30")) ) _TEXT_MAX_INFLIGHT = max(1, int(os.getenv("TEXT_MAX_INFLIGHT", "32"))) _IMAGE_MAX_INFLIGHT = max(1, int(os.getenv("IMAGE_MAX_INFLIGHT", "1"))) _OVERLOAD_STATUS_CODE = int(os.getenv("EMBEDDING_OVERLOAD_STATUS_CODE", "503")) _LOG_PREVIEW_COUNT = max(1, int(os.getenv("EMBEDDING_LOG_PREVIEW_COUNT", "3"))) _LOG_TEXT_PREVIEW_CHARS = max(32, int(os.getenv("EMBEDDING_LOG_TEXT_PREVIEW_CHARS", "120"))) _LOG_IMAGE_PREVIEW_CHARS = max(32, int(os.getenv("EMBEDDING_LOG_IMAGE_PREVIEW_CHARS", "180"))) _VECTOR_PREVIEW_DIMS = max(1, int(os.getenv("EMBEDDING_VECTOR_PREVIEW_DIMS", "6"))) class _InflightLimiter: def __init__(self, name: str, limit: int): self.name = name self.limit = max(1, int(limit)) self._sem = threading.BoundedSemaphore(self.limit) self._lock = threading.Lock() self._active = 0 self._rejected = 0 self._completed = 0 self._failed = 0 self._max_active = 0 def try_acquire(self) -> tuple[bool, int]: if not self._sem.acquire(blocking=False): with self._lock: self._rejected += 1 active = self._active return False, active with self._lock: self._active += 1 self._max_active = max(self._max_active, self._active) active = self._active return True, active def release(self, *, success: bool) -> int: with self._lock: self._active = max(0, self._active - 1) if success: self._completed += 1 else: self._failed += 1 active = self._active self._sem.release() return active def snapshot(self) -> Dict[str, int]: with self._lock: return { "limit": self.limit, "active": self._active, "rejected_total": self._rejected, "completed_total": self._completed, "failed_total": self._failed, "max_active": self._max_active, } _text_request_limiter = _InflightLimiter(name="text", limit=_TEXT_MAX_INFLIGHT) _image_request_limiter = _InflightLimiter(name="image", limit=_IMAGE_MAX_INFLIGHT) @dataclass class _SingleTextTask: text: str normalize: bool created_at: float request_id: str done: threading.Event result: Optional[List[float]] = None error: Optional[Exception] = None _text_single_queue: "deque[_SingleTextTask]" = deque() _text_single_queue_cv = threading.Condition() _text_batch_worker: Optional[threading.Thread] = None _text_batch_worker_stop = False def _compact_preview(text: str, max_chars: int) -> str: compact = " ".join((text or "").split()) if len(compact) <= max_chars: return compact return compact[:max_chars] + "..." def _preview_inputs(items: List[str], max_items: int, max_chars: int) -> List[Dict[str, Any]]: previews: List[Dict[str, Any]] = [] for idx, item in enumerate(items[:max_items]): previews.append( { "idx": idx, "len": len(item), "preview": _compact_preview(item, max_chars), } ) return previews def _preview_vector(vec: Optional[List[float]], max_dims: int = _VECTOR_PREVIEW_DIMS) -> List[float]: if not vec: return [] return [round(float(v), 6) for v in vec[:max_dims]] def _request_log_extra(request_id: str) -> Dict[str, str]: return {"reqid": request_id} def _resolve_request_id(http_request: Request) -> str: header_value = http_request.headers.get("X-Request-ID") if header_value and header_value.strip(): return header_value.strip()[:32] return str(uuid.uuid4())[:8] def _request_client(http_request: Request) -> str: client = getattr(http_request, "client", None) host = getattr(client, "host", None) return str(host or "-") def _encode_local_st(texts: List[str], normalize_embeddings: bool) -> Any: with _text_encode_lock: return _text_model.encode( texts, batch_size=int(CONFIG.TEXT_BATCH_SIZE), device=CONFIG.TEXT_DEVICE, normalize_embeddings=normalize_embeddings, ) def _start_text_batch_worker() -> None: global _text_batch_worker, _text_batch_worker_stop if _text_batch_worker is not None and _text_batch_worker.is_alive(): return _text_batch_worker_stop = False _text_batch_worker = threading.Thread( target=_text_batch_worker_loop, name="embed-text-microbatch-worker", daemon=True, ) _text_batch_worker.start() logger.info( "Started local_st text micro-batch worker | window_ms=%.1f max_batch=%d", _TEXT_MICROBATCH_WINDOW_SEC * 1000.0, int(CONFIG.TEXT_BATCH_SIZE), ) def _stop_text_batch_worker() -> None: global _text_batch_worker_stop with _text_single_queue_cv: _text_batch_worker_stop = True _text_single_queue_cv.notify_all() def _text_batch_worker_loop() -> None: max_batch = max(1, int(CONFIG.TEXT_BATCH_SIZE)) while True: with _text_single_queue_cv: while not _text_single_queue and not _text_batch_worker_stop: _text_single_queue_cv.wait() if _text_batch_worker_stop: return batch: List[_SingleTextTask] = [_text_single_queue.popleft()] deadline = time.perf_counter() + _TEXT_MICROBATCH_WINDOW_SEC while len(batch) < max_batch: remaining = deadline - time.perf_counter() if remaining <= 0: break if not _text_single_queue: _text_single_queue_cv.wait(timeout=remaining) continue while _text_single_queue and len(batch) < max_batch: batch.append(_text_single_queue.popleft()) try: queue_wait_ms = [(time.perf_counter() - task.created_at) * 1000.0 for task in batch] reqids = [task.request_id for task in batch] logger.info( "text microbatch dispatch | size=%d queue_wait_ms_min=%.2f queue_wait_ms_max=%.2f reqids=%s preview=%s", len(batch), min(queue_wait_ms) if queue_wait_ms else 0.0, max(queue_wait_ms) if queue_wait_ms else 0.0, reqids, _preview_inputs( [task.text for task in batch], _LOG_PREVIEW_COUNT, _LOG_TEXT_PREVIEW_CHARS, ), ) batch_t0 = time.perf_counter() embs = _encode_local_st([task.text for task in batch], normalize_embeddings=False) if embs is None or len(embs) != len(batch): raise RuntimeError( f"Text model response length mismatch in micro-batch: " f"expected {len(batch)}, got {0 if embs is None else len(embs)}" ) for task, emb in zip(batch, embs): vec = _as_list(emb, normalize=task.normalize) if vec is None: raise RuntimeError("Text model returned empty embedding in micro-batch") task.result = vec logger.info( "text microbatch done | size=%d reqids=%s dim=%d backend_elapsed_ms=%.2f", len(batch), reqids, len(batch[0].result) if batch and batch[0].result is not None else 0, (time.perf_counter() - batch_t0) * 1000.0, ) except Exception as exc: logger.error( "text microbatch failed | size=%d reqids=%s error=%s", len(batch), [task.request_id for task in batch], exc, exc_info=True, ) for task in batch: task.error = exc finally: for task in batch: task.done.set() def _encode_single_text_with_microbatch(text: str, normalize: bool, request_id: str) -> List[float]: task = _SingleTextTask( text=text, normalize=normalize, created_at=time.perf_counter(), request_id=request_id, done=threading.Event(), ) with _text_single_queue_cv: _text_single_queue.append(task) _text_single_queue_cv.notify() if not task.done.wait(timeout=_TEXT_REQUEST_TIMEOUT_SEC): with _text_single_queue_cv: try: _text_single_queue.remove(task) except ValueError: pass raise RuntimeError( f"Timed out waiting for text micro-batch worker ({_TEXT_REQUEST_TIMEOUT_SEC:.1f}s)" ) if task.error is not None: raise task.error if task.result is None: raise RuntimeError("Text micro-batch worker returned empty result") return task.result @app.on_event("startup") def load_models(): """Load models at service startup to avoid first-request latency.""" global _text_model, _image_model, _text_backend_name logger.info("Loading embedding models at startup...") if open_text_model: try: backend_name, backend_cfg = get_embedding_backend_config() _text_backend_name = backend_name if backend_name == "tei": from embeddings.text_embedding_tei import TEITextModel base_url = ( os.getenv("TEI_BASE_URL") or backend_cfg.get("base_url") or CONFIG.TEI_BASE_URL ) timeout_sec = int( os.getenv("TEI_TIMEOUT_SEC") or backend_cfg.get("timeout_sec") or CONFIG.TEI_TIMEOUT_SEC ) logger.info("Loading text backend: tei (base_url=%s)", base_url) _text_model = TEITextModel( base_url=str(base_url), timeout_sec=timeout_sec, ) elif backend_name == "local_st": from embeddings.text_embedding_sentence_transformers import Qwen3TextModel model_id = ( os.getenv("TEXT_MODEL_ID") or backend_cfg.get("model_id") or CONFIG.TEXT_MODEL_ID ) logger.info("Loading text backend: local_st (model=%s)", model_id) _text_model = Qwen3TextModel(model_id=str(model_id)) _start_text_batch_worker() else: raise ValueError( f"Unsupported embedding backend: {backend_name}. " "Supported: tei, local_st" ) logger.info("Text backend loaded successfully: %s", _text_backend_name) except Exception as e: logger.error("Failed to load text model: %s", e, exc_info=True) raise if open_image_model: try: if CONFIG.USE_CLIP_AS_SERVICE: from embeddings.clip_as_service_encoder import ClipAsServiceImageEncoder logger.info( "Loading image encoder via clip-as-service: %s (configured model: %s)", CONFIG.CLIP_AS_SERVICE_SERVER, CONFIG.CLIP_AS_SERVICE_MODEL_NAME, ) _image_model = ClipAsServiceImageEncoder( server=CONFIG.CLIP_AS_SERVICE_SERVER, batch_size=CONFIG.IMAGE_BATCH_SIZE, ) logger.info("Image model (clip-as-service) loaded successfully") else: from embeddings.clip_model import ClipImageModel logger.info( "Loading local image model: %s (device: %s)", CONFIG.IMAGE_MODEL_NAME, CONFIG.IMAGE_DEVICE, ) _image_model = ClipImageModel( model_name=CONFIG.IMAGE_MODEL_NAME, device=CONFIG.IMAGE_DEVICE, ) logger.info("Image model (local CN-CLIP) loaded successfully") except Exception as e: logger.error("Failed to load image model: %s", e, exc_info=True) raise logger.info("All embedding models loaded successfully, service ready") @app.on_event("shutdown") def stop_workers() -> None: _stop_text_batch_worker() def _normalize_vector(vec: np.ndarray) -> np.ndarray: norm = float(np.linalg.norm(vec)) if not np.isfinite(norm) or norm <= 0.0: raise RuntimeError("Embedding vector has invalid norm (must be > 0)") return vec / norm def _as_list(embedding: Optional[np.ndarray], normalize: bool = False) -> Optional[List[float]]: if embedding is None: return None if not isinstance(embedding, np.ndarray): embedding = np.array(embedding, dtype=np.float32) if embedding.ndim != 1: embedding = embedding.reshape(-1) embedding = embedding.astype(np.float32, copy=False) if normalize: embedding = _normalize_vector(embedding).astype(np.float32, copy=False) return embedding.tolist() @app.get("/health") def health() -> Dict[str, Any]: """Health check endpoint. Returns status and current throttling stats.""" return { "status": "ok", "text_model_loaded": _text_model is not None, "text_backend": _text_backend_name, "image_model_loaded": _image_model is not None, "limits": { "text": _text_request_limiter.snapshot(), "image": _image_request_limiter.snapshot(), }, "text_microbatch": { "window_ms": round(_TEXT_MICROBATCH_WINDOW_SEC * 1000.0, 3), "queue_depth": len(_text_single_queue), "worker_alive": bool(_text_batch_worker is not None and _text_batch_worker.is_alive()), "request_timeout_sec": _TEXT_REQUEST_TIMEOUT_SEC, }, } def _embed_text_impl( normalized: List[str], effective_normalize: bool, request_id: str, ) -> List[Optional[List[float]]]: if _text_model is None: raise RuntimeError("Text model not loaded") t0 = time.perf_counter() try: if _text_backend_name == "local_st": if len(normalized) == 1 and _text_batch_worker is not None: out = [ _encode_single_text_with_microbatch( normalized[0], normalize=effective_normalize, request_id=request_id, ) ] logger.info( "text backend done | backend=%s mode=microbatch-single inputs=%d normalize=%s dim=%d backend_elapsed_ms=%.2f", _text_backend_name, len(normalized), effective_normalize, len(out[0]) if out and out[0] is not None else 0, (time.perf_counter() - t0) * 1000.0, extra=_request_log_extra(request_id), ) return out embs = _encode_local_st(normalized, normalize_embeddings=False) mode = "direct-batch" else: embs = _text_model.encode( normalized, batch_size=int(CONFIG.TEXT_BATCH_SIZE), device=CONFIG.TEXT_DEVICE, normalize_embeddings=effective_normalize, ) mode = "backend-batch" except Exception as e: logger.error( "Text embedding backend failure: %s", e, exc_info=True, extra=_request_log_extra(request_id), ) raise RuntimeError(f"Text embedding backend failure: {e}") from e if embs is None or len(embs) != len(normalized): raise RuntimeError( f"Text model response length mismatch: expected {len(normalized)}, " f"got {0 if embs is None else len(embs)}" ) out: List[Optional[List[float]]] = [] for i, emb in enumerate(embs): vec = _as_list(emb, normalize=effective_normalize) if vec is None: raise RuntimeError(f"Text model returned empty embedding for index {i}") out.append(vec) logger.info( "text backend done | backend=%s mode=%s inputs=%d normalize=%s dim=%d backend_elapsed_ms=%.2f", _text_backend_name, mode, len(normalized), effective_normalize, len(out[0]) if out and out[0] is not None else 0, (time.perf_counter() - t0) * 1000.0, extra=_request_log_extra(request_id), ) return out @app.post("/embed/text") async def embed_text( texts: List[str], http_request: Request, response: Response, normalize: Optional[bool] = None, ) -> List[Optional[List[float]]]: request_id = _resolve_request_id(http_request) response.headers["X-Request-ID"] = request_id effective_normalize = bool(CONFIG.TEXT_NORMALIZE_EMBEDDINGS) if normalize is None else bool(normalize) normalized: List[str] = [] for i, t in enumerate(texts): if not isinstance(t, str): raise HTTPException(status_code=400, detail=f"Invalid text at index {i}: must be string") s = t.strip() if not s: raise HTTPException(status_code=400, detail=f"Invalid text at index {i}: empty string") normalized.append(s) accepted, active = _text_request_limiter.try_acquire() if not accepted: logger.warning( "embed_text rejected | client=%s backend=%s inputs=%d normalize=%s active=%d limit=%d preview=%s", _request_client(http_request), _text_backend_name, len(normalized), effective_normalize, active, _TEXT_MAX_INFLIGHT, _preview_inputs(normalized, _LOG_PREVIEW_COUNT, _LOG_TEXT_PREVIEW_CHARS), extra=_request_log_extra(request_id), ) raise HTTPException( status_code=_OVERLOAD_STATUS_CODE, detail=f"Text embedding service busy: active={active}, limit={_TEXT_MAX_INFLIGHT}", ) request_started = time.perf_counter() success = False try: logger.info( "embed_text request | client=%s backend=%s inputs=%d normalize=%s active=%d limit=%d preview=%s", _request_client(http_request), _text_backend_name, len(normalized), effective_normalize, active, _TEXT_MAX_INFLIGHT, _preview_inputs(normalized, _LOG_PREVIEW_COUNT, _LOG_TEXT_PREVIEW_CHARS), extra=_request_log_extra(request_id), ) verbose_logger.info( "embed_text detail | payload=%s normalize=%s backend=%s", normalized, effective_normalize, _text_backend_name, extra=_request_log_extra(request_id), ) out = await run_in_threadpool(_embed_text_impl, normalized, effective_normalize, request_id) success = True latency_ms = (time.perf_counter() - request_started) * 1000.0 logger.info( "embed_text response | backend=%s inputs=%d normalize=%s dim=%d first_vector=%s latency_ms=%.2f", _text_backend_name, len(normalized), effective_normalize, len(out[0]) if out and out[0] is not None else 0, _preview_vector(out[0] if out else None), latency_ms, extra=_request_log_extra(request_id), ) verbose_logger.info( "embed_text result detail | count=%d first_vector=%s latency_ms=%.2f", len(out), out[0][: _VECTOR_PREVIEW_DIMS] if out and out[0] is not None else [], latency_ms, extra=_request_log_extra(request_id), ) return out except HTTPException: raise except Exception as e: latency_ms = (time.perf_counter() - request_started) * 1000.0 logger.error( "embed_text failed | backend=%s inputs=%d normalize=%s latency_ms=%.2f error=%s", _text_backend_name, len(normalized), effective_normalize, latency_ms, e, exc_info=True, extra=_request_log_extra(request_id), ) raise HTTPException(status_code=502, detail=str(e)) from e finally: remaining = _text_request_limiter.release(success=success) logger.info( "embed_text finalize | success=%s active_after=%d", success, remaining, extra=_request_log_extra(request_id), ) def _embed_image_impl( urls: List[str], effective_normalize: bool, request_id: str, ) -> List[Optional[List[float]]]: if _image_model is None: raise RuntimeError("Image model not loaded") t0 = time.perf_counter() with _image_encode_lock: vectors = _image_model.encode_image_urls( urls, batch_size=CONFIG.IMAGE_BATCH_SIZE, normalize_embeddings=effective_normalize, ) if vectors is None or len(vectors) != len(urls): raise RuntimeError( f"Image model response length mismatch: expected {len(urls)}, " f"got {0 if vectors is None else len(vectors)}" ) out: List[Optional[List[float]]] = [] for i, vec in enumerate(vectors): out_vec = _as_list(vec, normalize=effective_normalize) if out_vec is None: raise RuntimeError(f"Image model returned empty embedding for index {i}") out.append(out_vec) logger.info( "image backend done | inputs=%d normalize=%s dim=%d backend_elapsed_ms=%.2f", len(urls), effective_normalize, len(out[0]) if out and out[0] is not None else 0, (time.perf_counter() - t0) * 1000.0, extra=_request_log_extra(request_id), ) return out @app.post("/embed/image") async def embed_image( images: List[str], http_request: Request, response: Response, normalize: Optional[bool] = None, ) -> List[Optional[List[float]]]: request_id = _resolve_request_id(http_request) response.headers["X-Request-ID"] = request_id effective_normalize = bool(CONFIG.IMAGE_NORMALIZE_EMBEDDINGS) if normalize is None else bool(normalize) urls: List[str] = [] for i, url_or_path in enumerate(images): if not isinstance(url_or_path, str): raise HTTPException(status_code=400, detail=f"Invalid image at index {i}: must be string URL/path") s = url_or_path.strip() if not s: raise HTTPException(status_code=400, detail=f"Invalid image at index {i}: empty URL/path") urls.append(s) accepted, active = _image_request_limiter.try_acquire() if not accepted: logger.warning( "embed_image rejected | client=%s inputs=%d normalize=%s active=%d limit=%d preview=%s", _request_client(http_request), len(urls), effective_normalize, active, _IMAGE_MAX_INFLIGHT, _preview_inputs(urls, _LOG_PREVIEW_COUNT, _LOG_IMAGE_PREVIEW_CHARS), extra=_request_log_extra(request_id), ) raise HTTPException( status_code=_OVERLOAD_STATUS_CODE, detail=f"Image embedding service busy: active={active}, limit={_IMAGE_MAX_INFLIGHT}", ) request_started = time.perf_counter() success = False try: logger.info( "embed_image request | client=%s inputs=%d normalize=%s active=%d limit=%d preview=%s", _request_client(http_request), len(urls), effective_normalize, active, _IMAGE_MAX_INFLIGHT, _preview_inputs(urls, _LOG_PREVIEW_COUNT, _LOG_IMAGE_PREVIEW_CHARS), extra=_request_log_extra(request_id), ) verbose_logger.info( "embed_image detail | payload=%s normalize=%s", urls, effective_normalize, extra=_request_log_extra(request_id), ) out = await run_in_threadpool(_embed_image_impl, urls, effective_normalize, request_id) success = True latency_ms = (time.perf_counter() - request_started) * 1000.0 logger.info( "embed_image response | inputs=%d normalize=%s dim=%d first_vector=%s latency_ms=%.2f", len(urls), effective_normalize, len(out[0]) if out and out[0] is not None else 0, _preview_vector(out[0] if out else None), latency_ms, extra=_request_log_extra(request_id), ) verbose_logger.info( "embed_image result detail | count=%d first_vector=%s latency_ms=%.2f", len(out), out[0][: _VECTOR_PREVIEW_DIMS] if out and out[0] is not None else [], latency_ms, extra=_request_log_extra(request_id), ) return out except HTTPException: raise except Exception as e: latency_ms = (time.perf_counter() - request_started) * 1000.0 logger.error( "embed_image failed | inputs=%d normalize=%s latency_ms=%.2f error=%s", len(urls), effective_normalize, latency_ms, e, exc_info=True, extra=_request_log_extra(request_id), ) raise HTTPException(status_code=502, detail=f"Image embedding backend failure: {e}") from e finally: remaining = _image_request_limiter.release(success=success) logger.info( "embed_image finalize | success=%s active_after=%d", success, remaining, extra=_request_log_extra(request_id), )