What you’re seeing is expected eventual consistency in Prefect logging. The short version: - A flow/job can reach a terminal state before its final logs have been ingested and made queryable. - A fixed

sleep(5)

may help sometimes, but it’s not a reliable guarantee. A better approach is: 1. Wait for the run to reach a terminal state 2. Flush pending API logs if you control the process producing them 3. Fetch logs with a short retry loop until they stop changing or until a timeout If your client app is separate from the process emitting logs, then flushing may not be possible from the client, so retries are the safest pattern. Why this happens Prefect logs are typically sent to the API in batches, not one-by-one immediately. That means the final few logs can trail the final

COMPLETED

or

FAILED

state by a couple seconds or more. In Prefect 3.x, the main knob is: -

PREFECT_LOGGING_TO_API_BATCH_INTERVAL

default: about

2s

So

5s

is not a bad guess, but it’s still heuristic, not guaranteed. Best practice Instead of using one larger sleep, use a retry/polling loop for logs after terminal state. Example pattern:

python
import asyncio
from prefect.client.orchestration import get_client
from prefect.client.schemas.filters import LogFilter
from prefect.client.schemas.sorting import LogSort

async def read_all_logs_for_run(flow_run_id, attempts=10, delay=1):
    last_count = -1
    stable_reads = 0

    async with get_client() as client:
        for _ in range(attempts):
            logs = await client.read_logs(
                log_filter=LogFilter(flow_run_id={"any_": [flow_run_id]}),
                sort=LogSort.TIMESTAMP_ASC,
            )

            count = len(logs)

            if count == last_count:
                stable_reads += 1
            else:
                stable_reads = 0
                last_count = count

            # If log count hasn't changed for 2 consecutive reads,
            # assume ingestion has caught up
            if stable_reads >= 2:
                return logs

            await asyncio.sleep(delay)

    return logs

That’s usually better than a single

sleep(5)

because it adapts to short or long delays. If you control the code that emits the logs If this code is running in the same process that generated the Prefect logs, you can explicitly flush before reading:

python
from prefect.logging.handlers import APILogHandler

await APILogHandler.aflush()

For sync code:

python
from prefect.logging.handlers import APILogHandler

APILogHandler.flush()

That is the most reliable option when available. If you especially care about failures A good failure-handling pattern is: - detect terminal failed state - retry fetching logs for a short window, like

10-15s

- stop early once logs appear stable - optionally inspect both run state message and logs Sometimes the key error is in the state/message, not only the logs. Would a bigger sleep help? Yes, but only probabilistically. If you want a simple heuristic instead of retries: - try

8-10 seconds

instead of

5

But I’d still recommend the retry-until-stable approach over increasing sleep. Optional improvement If low-latency logs are important, reduce the batch interval in the environment where the flow runs:

python
PREFECT_LOGGING_TO_API_BATCH_INTERVAL=0.5

That can make logs show up faster, but it may increase API traffic. Recommendation I’d use this order of preference: 1. Best: flush logs with

APILogHandler.aflush()

if you can 2. Next best: retry fetching logs until the result stabilizes 3. Fallback: longer fixed sleep like

8-10s

Relevant APIs: - Prefect client log querying via

client.read_logs(...)

- Live log subscription exists, but it’s better for streaming while the run is active, not for guaranteed historical catch-up after completion If you want, I can give you a complete example for either:

- a polling client that waits for final logs after a failed run, or - a streaming client that combines live logs with a final historical fetch

Short answer: no — in Prefect 3.x there isn’t a log-count endpoint or a metadata-only logs query that returns “total available logs” without returning log records.

read_logs(...)

returns just a list of logs, not pagination metadata like total count. So if you’re paging with

limit

/

offset

, Prefect does not currently give you a cheap built-in way to ask: - “how many logs exist now?” - “has the total increased since last check?” What you can do instead 1. Use page-shape detection When paginating: - if a page returns fewer than

limit

, you’re at the current end - if a previously empty/final page later becomes non-empty, more logs arrived That means your client can track a “known end offset” and periodically probe from there. Example:

python
async def fetch_new_logs(client, flow_run_id, offset, limit=200):
    logs = await client.read_logs(
        log_filter=LogFilter(flow_run_id={"any_": [flow_run_id]}),
        offset=offset,
        limit=limit,
        sort=LogSort.TIMESTAMP_ASC,
    )
    return logs

Then: - if result length is

0

, no new logs beyond that offset - if result length is

> 0

, append them and advance offset - if result length is

< limit

, you’ve caught up for now This is probably the best lightweight pattern available today. 2. Probe only the tail, not the full history If you already fetched logs through offset

N

, don’t re-read old pages. Just request:

python
offset=N, limit=small_page_size

That minimizes unnecessary transfer while still telling you whether more logs have shown up. 3. Use a “quiet period” after terminal state After the run is finished: - keep probing from the current end offset every

1s

- stop after, say,

2-3

consecutive empty reads That’s more reliable than one big sleep. Example:

python
import asyncio

async def wait_for_log_tail(client, flow_run_id, offset, limit=200, empty_reads_needed=3):
    empty_reads = 0
    all_new_logs = []

    while empty_reads < empty_reads_needed:
        page = await client.read_logs(
            log_filter=LogFilter(flow_run_id={"any_": [flow_run_id]}),
            offset=offset,
            limit=limit,
            sort=LogSort.TIMESTAMP_ASC,
        )

        if not page:
            empty_reads += 1
            await asyncio.sleep(1)
            continue

        empty_reads = 0
        all_new_logs.extend(page)
        offset += len(page)

        if len(page) < limit:
            await asyncio.sleep(1)

    return all_new_logs, offset

4. If you control the producer, flush first If the same process is emitting Prefect logs, call:

python
await APILogHandler.aflush()

That reduces the need for repeated tail probing. Practical recommendation for your case Since your logs may be long and you already chunk them: - keep your current

limit

/

offset

paging - after job terminal state, poll only from the last known offset - stop after a few consecutive empty pages That gives you an efficient “has anything new arrived?” check without re-downloading old logs, even though Prefect doesn’t expose a count. Important limitation This only tells you whether more logs exist past your current offset. It does not protect against: - logs arriving slightly out of order by timestamp - wanting a true authoritative total count But for most “did the final error log show up yet?” use cases, tail polling by offset is the right pattern. If you want, I can write a complete helper that: - pages logs in chunks, - handles terminal state, - waits for the log tail to settle, - and returns the final combined log stream safely.

Yep — you can combine those conditions in

LogFilter

as: -

flow_run_id == given id

- and -

(level >= ERROR) OR (name == "jds")

Example:

python
import logging
from prefect.client.schemas.filters import LogFilter
from prefect.client.schemas.sorting import LogSort

log_filter = LogFilter(
    flow_run_id={"any_": [flow_run_id]},
    or_=[
        LogFilter(level={"ge_": logging.ERROR}),
        LogFilter(name={"any_": ["jds"]}),
    ],
)

Then use it like:

python
logs = await client.read_logs(
    log_filter=log_filter,
    offset=offset,
    limit=limit,
    sort=LogSort.TIMESTAMP_ASC,
)

So the logic is effectively:

text
flow_run_id IN [flow_run_id]
AND (
  level >= ERROR
  OR name == "jds"
)

If you want, I can also show the sync version or verify the exact nested filter syntax against the current Prefect schema.

Because my previous filter suggestion was wrong for Prefect 3.x — sorry about that. In Prefect 3.x,

LogFilter

does not support the nested

or_=[...]

structure I showed, and it also does not support filtering by logger

name

the way I implied. So the behavior you saw means that expression was not doing the intended:

text
flow_run_id == X AND (level >= ERROR OR name == "jds")

Those INFO logs you pasted are Prefect system/worker logs for that same flow run, so they matched the

flow_run_id

portion, and your actual filter was not enforcing the logger-name-or-error logic the way we wanted. Important limitation As currently implemented in Prefect 3.x, there is not a built-in server-side

LogFilter

for:

text
flow_run_id == X AND (level >= ERROR OR logger_name == "jds")

Also, there is no supported

name

field on

LogFilter

for logger name filtering in the public client schema I verified. What to do instead Your best option is: 1. Query logs for the flow run 2. Filter client-side for: -

log.level >= logging.ERROR

, or -

log.name == "jds"

if that attribute is present in the returned log objects Example:

python
import logging
from prefect.client.schemas.filters import LogFilter
from prefect.client.schemas.sorting import LogSort

logs = await client.read_logs(
    log_filter=LogFilter(flow_run_id={"any_": [flow_run_id]}),
    offset=offset,
    limit=limit,
    sort=LogSort.TIMESTAMP_ASC,
)

filtered_logs = [
    log for log in logs
    if log.level >= logging.ERROR or getattr(log, "name", None) == "jds"
]

If you’re paging, just apply that filtering per page. Why those INFO logs appeared They’re associated with the same

flow_run_id

, and since the intended nested OR/name filter isn’t actually supported, they were not excluded. Recommendation If your goal is specifically “all error logs plus all

jds

logs”, I’d do: - server-side filter by

flow_run_id

only - client-side post-filter by level/name That’s the most reliable approach today. If you want, I can help you write a paginated helper that: - reads all logs for a flow run in chunks - waits for late-arriving logs - filters to

ERROR+

or logger

jds

- returns just the final relevant lines For reference, Prefect 3.x log querying is done via

client.read_logs(...)

and the available schema is documented in the codebase/API behavior rather than a richer count/name/nested-OR filter surface.