[Bug]: one unparseable record silently truncates an entire paginated stream read (search_traces, search_spans, experiments, migrate)

[feiiiiii5]

Bug description

A paginated stream read stops early and reports a short answer as if it were complete whenever the backend serves a record the SDK cannot parse. One unreadable record in the middle of a stream costs every page after it, silently.

read_and_parse_full_stream() decides whether the backend has run out of data by comparing the page size it asked for against the number of items it got back:

# sdks/python/src/opik/api_objects/rest_stream_parser.py:80 (main f81eed1b4)
if current_batch_size > len(parsed_items):
    break

len(parsed_items) is the count of records that survived parsing. _parse_stream_line() returns None for anything it cannot decode or validate (rest_stream_parser.py:121-137), and read_and_parse_stream() only appends non-None items (:101-104). So a page where the backend sent a full 50 records and 1 failed validation looks, to the loop, exactly like a page where the backend sent 49 records and there is nothing left. The remaining pages are never requested.

The two things the test at line 80 is trying to distinguish are different events:

• the backend sent fewer records than we asked for — end of data;
• the backend sent a full page, some of which we could not read — data continues past this page.

Affected user-facing paths

Every caller of read_and_parse_full_stream(), i.e. any read that can span more than one page:

• Opik.search_traces() — sdks/python/src/opik/api_objects/opik_client.py:2126 → api_objects/search_helpers.py:79
• Opik.search_spans() — opik_client.py:2236 → search_helpers.py:41
• Opik.search_threads() — search_helpers.py:113, api_objects/threads/threads_client.py:111
• experiment reads: evaluate_experiment() → evaluation/rest_operations.py:15 → api_objects/experiment/rest_operations.py:35
• opik migrate — cli/migrate/datasets/experiments.py:827, cli/migrate/datasets/version_replay.py:606

The experiment path is the worst case: if every record on the first page happens to be unreadable, parsed_items is empty, the loop breaks, and experiment/rest_operations.py:47-50 raises ExperimentNotFound for an experiment that does exist. A migration/rollback tool then reports missing data that is present.

Steps to reproduce

Backend/SDK field skew is what makes a record unreadable in practice (the generated model requires start_time, sdks/python/src/opik/rest_api/types/span_public.py:28), so the reproducer feeds the parser a record with that field absent. Three pages of two records each; the last page is genuinely short.

import json
from opik.api_objects import rest_stream_parser
from opik.rest_api.types import span_public

RECORD = {
    "id": "0195f6f1-9da2-7630-b285-60cf5580372f",
    "project_id": "0195f6f1-9c82-751f-a1ad-54fec4b5c7d8",
    "trace_id": "0195f6f1-9c3f-7b65-a334-56173d19bc00",
    "name": "query", "type": "general",
    "start_time": "2025-01-03T11:03:17.505783Z",
    "end_time": "2025-01-03T11:03:18.591897Z",
    "created_at": "2025-04-02T14:39:44.412550Z",
    "last_updated_at": "2025-04-02T14:39:44.412550Z",
    "created_by": "admin", "last_updated_by": "admin",
}

def line(span_id, unreadable=False):
    record = dict(RECORD, id=span_id)
    if unreadable:
        del record["start_time"]
    return (json.dumps(record) + "\r\n").encode("utf-8")

PAGES = [
    [line("bad-1", unreadable=True), line("span-a")],   # backend sent 2 (a full page)
    [line("span-b"), line("span-c")],                    # backend sent 2
    [line("span-d")],                                     # backend sent 1 -> real end
]
requests_made = []

def read_source(current_batch_size, last_retrieved_id):
    requests_made.append((current_batch_size, last_retrieved_id))
    return PAGES.pop(0)

spans = rest_stream_parser.read_and_parse_full_stream(
    read_source=read_source, parsed_item_class=span_public.SpanPublic,
    max_results=None, max_endpoint_batch_size=2,
)
print("requests:", requests_made)
print("records :", [s.id for s in spans])

Output on main (f81eed1b4), measured — python3 repro_stream_truncation.py, cwd sdks/python:

ERROR opik.api_objects.rest_stream_parser: Error parsing SpanPublic, reason: 1 validation error for SpanPublic  Field required [type=missing, input_value={'id': 'bad-1', ...}]
requests: [(2, None)]
records : ['span-a']

One request instead of three; span-b, span-c and span-d are never fetched. The only trace is a per-record ERROR line — nothing says the read stopped early, and the return value is indistinguishable from "there were only 1 record".

Expected behaviour

A dropped record counts as a record the backend sent, so the read continues to the next page and stops when a page is genuinely short. When a read did drop records, the fact should be visible in one log line rather than only in per-record errors.

One case cannot be recovered by skipping: a page whose records all fail to parse advances neither the result nor the last_retrieved_id cursor, so the next request would ask for exactly the same page forever. Whatever is decided there should be a deliberate one — today it stops only by accident, because an empty page also reads as "short".

What I checked before filing

Searched issues and PRs, open and closed, for this root cause (read_and_parse_full_stream, rest_stream_parser, incomplete chunked read, pagination stopped early, malformed record, skip unparseable, search_traces missing results). Closest prior art is in the same file but a different cause: #7330 / #7289 (page-size adaptive shrink, merged) and #7479 (guard against an unbounded full-project read). Neither addresses the parsed-count-vs-page-size comparison. No open PR touches rest_stream_parser.py.

[github-actions]

I have everything I need to write a thorough response. The bug is clear, the fix location is pinpointed, and I can see there's no existing test covering the parse-failure-mid-page scenario.

Hi, I'm Scout 🦉, the comet-ml/opik repository agent. Let me look into this.

Solution / Workaround

No existing workaround is available. The bug is confirmed and no prior PR or issue addresses this specific root cause (the reporter already verified this; my searches agree — the closest prior art is #7330/#7289 for adaptive page-size shrink and #7479 for unbounded reads, neither of which touches the parsed-count comparison).

Code Investigation

The bug is entirely in sdks/python/src/opik/api_objects/rest_stream_parser.py.

Root cause — line 80:

if current_batch_size > len(parsed_items):
    break

len(parsed_items) counts only successfully parsed records. _parse_stream_line (lines 121–137) returns None on any json.JSONDecodeError, TypeError, ValueError, or other exception, and read_and_parse_stream (lines 101–104) only appends non-None results. So a page where the backend sent 50 records and 1 failed validation is indistinguishable from a page where the backend only sent 49 records — both make current_batch_size (50) > len(parsed_items) (49) true, and the loop exits.

The two things the condition must distinguish:

Scenario	Backend sent	parsed_items	Expected action
End of data	< current_batch_size	< current_batch_size	break ✅
Parse failure	current_batch_size	< current_batch_size	continue ❌ (currently breaks)

Secondary edge case (also noted by reporter): if all records on a page fail to parse, parsed_items is empty and last_retrieved_id doesn't advance. A naive "continue on parse failure" would produce an infinite loop requesting the same page forever. This case needs deliberate handling.

Affected callers (all ultimately call read_and_parse_full_stream):

• sdks/python/src/opik/api_objects/search_helpers.py — search_spans(), search_traces(), search_threads()
• sdks/python/src/opik/api_objects/experiment/rest_operations.py:35 — experiment reads (worst case: all-bad first page raises ExperimentNotFound for an existing experiment)
• sdks/python/src/opik/cli/migrate/datasets/experiments.py:827
• sdks/python/src/opik/cli/migrate/datasets/version_replay.py:606

Existing test coverage: sdks/python/tests/unit/api_objects/test_rest_stream_parser.py has good coverage of the happy path and size-correlated error shrinking, but no test covers a parse failure mid-page. The reproducer in the issue report fills this gap exactly.

Next Steps

The fix is in read_and_parse_stream / read_and_parse_full_stream. The key insight is to track how many raw records the backend sent separately from how many survived parsing:

Option A (minimal, recommended): Have read_and_parse_stream return a (parsed_items, raw_count) tuple (or a small dataclass), where raw_count is incremented for every non-empty line the stream yields regardless of whether parsing succeeded. read_and_parse_full_stream then breaks on current_batch_size > raw_count instead of current_batch_size > len(parsed_items). For the all-bad-page edge case, add a guard: if raw_count == current_batch_size but parsed_items is empty, log a warning and advance via a skip (e.g., break after N consecutive all-bad pages, or expose the raw last ID for cursor advancement).

Option B (alternative): Keep the existing signature of read_and_parse_stream but add a separate counter inside read_and_parse_full_stream by making _parse_stream_line distinguish "empty line / EOF" from "decode/validation failure" via a sentinel, so callers can count attempted parses.

Concretely, the change in read_and_parse_full_stream collapses to:

# in read_and_parse_stream — return (items, lines_attempted) instead of items
# in read_and_parse_full_stream line 80:
if current_batch_size > lines_attempted:   # was: len(parsed_items)
    break

A summary warning log line after the loop ("Read completed; N records were dropped due to parse errors.") satisfies the reporter's visibility requirement without flooding logs with per-record errors.

A new test mirroring the reporter's reproducer should be added to test_rest_stream_parser.py — it's a self-contained unit test with no real network calls needed.

Feel free to open a pull request with this fix — the change is straightforward!

Was this helpful? React to this comment with 👍 or 👎 to rate my response.

---

View Scout trace in Opik

[chenhz01]

The read loop treats any mid-page parse failure as end-of-stream, so a short read is returned as if complete - and as the repo agent noted, nothing covers parse-failure-mid-page. I would make the stream reader distinguish the two endings: page-boundary reached (complete) versus a record failed to parse (partial, with errors collected and iteration continuing past the bad record). The result object then carries state plus the error list, so callers can decide to accept partial data knowingly instead of trusting a full-looking read. Pseudocode: for rec in page: try: emit(parse(rec)) except ParseError: errors.append(rec); state = partial if errors else complete. And a regression test constructing a bad record mid-page, asserting iteration continues and errors surface. Want me to take a crack at it? Happy to open a PR with working code.

[feiiiiii5]

@chenhz01 thanks for spelling that out — and yes, the two endings are the heart of it.

#8411 (still a draft; only baz-reviewer has looked at it, no maintainer yet) already takes the half you described first: the page-size decision now counts the records the backend delivered rather than the ones that parsed, so a single unreadable record no longer reads as end-of-stream, and a read that did drop records says so once at WARNING with the count. It keeps the result object's shape untouched.

Your state = partial plus an error list on the result is a different, bigger thing: it lets callers accept partial data knowingly, which changes what search_traces() and the experiment path return. I would not fold that in without a maintainer saying they want that on the result object — if they do, happy to take it in either PR.

So concretely: if you want to open a PR for the surfaced-errors/partial-state part, that is a clean complement to #8411 rather than duplicated work, and I would use it. If a maintainer would rather it all land in one place, I can move your design into #8411 and close the other. Either way the regression test you describe (bad record mid-page, assert iteration continues and errors surface) is the one this issue needs — mine covers the first half of it today.