bug: non-numeric status in a 207 error entry raises inside handle_exception, kills the score consumer thread, flush()/shutdown() then hang #1967

Describe the bug

ScoreIngestionConsumer.upload() calls handle_exception(e) from inside its except block (score_ingestion_consumer.py#L141-L153). generate_error_message does int(error.status) on string statuses with no guard (parse_error.py#L75-L99), so a non-numeric status such as "n/a" raises ValueError from inside the error handler.

Nothing catches it in upload() or in run() (L129-L139), so the daemon consumer thread exits, and nothing restarts it (only the fork re-init does). The batch that triggered the error is still acknowledged by the finally block, but every later score event stays in the queue, unfinished_tasks never reaches 0, and Langfuse.flush() / shutdown() block forever on Queue.join() (resource_manager.py#L629-L640, no timeout).

Likelihood: the trigger needs a 207 response whose errors[].status is non-numeric. The Langfuse server does not send that today, so this is a defensive-hardening bug (a misbehaving proxy/gateway or a future server change would trigger it). The consequence is silent loss of later scores plus an unbounded hang at exit, and the fix is small.

Steps to reproduce

Uses httpx.MockTransport, no server needed:

import httpx, threading
from queue import Queue
from langfuse._utils.request import LangfuseClient
from langfuse._task_manager.score_ingestion_consumer import ScoreIngestionConsumer
import time; time.sleep = lambda s: None
def h(req): return httpx.Response(207, json={"errors": [{"status": "n/a", "message": "m"}]})
cl = LangfuseClient("pk", "sk", "http://x", "1", 5, httpx.Client(transport=httpx.MockTransport(h)))
q = Queue(); c = ScoreIngestionConsumer(ingestion_queue=q, identifier=0, client=cl, public_key="pk", flush_at=1, flush_interval=0.2)
c.start(); q.put({"id": "a", "type": "t", "body": {}}); c.join(3)
print("consumer alive:", c.is_alive())
q.put({"id": "b", "type": "t", "body": {}})
t = threading.Thread(target=q.join, daemon=True); t.start(); t.join(3); print("flush hung:", t.is_alive())

Observed output (traceback goes to stderr):

  File ".../langfuse/_utils/parse_error.py", line 86, in generate_error_message
    int(error.status) if isinstance(error.status, str) else error.status,
ValueError: invalid literal for int() with base 10: 'n/a'
consumer alive: False
flush hung: True

Direct check: handle_exception(APIErrors([APIError("abc", "m")])) raises ValueError: invalid literal for int() with base 10: 'abc' (a None status is fine).

Langfuse Cloud or self-hosted?

Not server dependent (reproduced with a mock transport).

If self-hosted, what version are you running?

n/a

SDK and integration versions

langfuse 4.17.0 (main @ bf11ec1), Python 3.14.0.

Additional information

Possible direction: convert statuses with a safe helper (try: int(s) except (TypeError, ValueError): None) in parse_error.py; make handle_exception never raise (fall back to logger.error); and wrap the loop body of ScoreIngestionConsumer.run() in try/except Exception that logs and continues so a single bad batch cannot end the thread. A bounded join() in flush() would be related hardening.

Related, different root cause: #1799 (shutdown() hangs on _score_ingestion_queue.join() after a cached resource manager is reused). Both end with consumers that are gone and an unbounded join().

Are you interested in contributing a fix for this bug?

Yes