Recovering from ConsumerTooSlow?

[jetpacktuxedo]

I think the root cause of this is a queue of incoming posts reaching some max size because I'm not popping them out of the queue fast enough. I'm generally getting the ConsumerTooSlow error when I'm just filtering and not even actually processing anything.

Here is my on_message_handler. I'm just watching for likes on a specific post and triggering another action on them. Is there a more efficient way to filter this?

    async def on_message_handler(message) -> None:
        if message.header.t != '#commit':
            return
        commit = parse_subscribe_repos_message(message)
        if len(commit.ops) == 0:
            return
        op = commit.ops[0]
        if (op.action != "create"):
            return
        uri = AtUri.from_str(f'at://{commit.repo}/{op.path}')
        if (uri.collection != models.ids.AppBskyFeedLike):
            return
        post = CAR.from_bytes(commit.blocks).blocks[op.cid]
        if (post["subject"]["uri"] == watchpost):
            await on_hit(did=commit.repo)

I tried wrapping my client.start() in a while True like this:

while True:
    try:
        print("Listening to the firehose")
        await client.start(on_message_handler, on_error_handler)
    except (CancelledError, KeyboardInterrupt):
        print("Caught keyboard interrupt, exiting")
        if client:
            await client.stop()
        sys.exit(0)
    except FirehoseError as e:
        print("Caught firehose error:")
        print(e)
    except Exception as e:
        raise e
    finally:
        if client:
            await client.stop()

but this just results in an infinite loop of "Listening to the firehose"... Can the client not be started again once it is stopped?

Honestly if I can't keep up with the buffer I'd rather dump the contents and keep going, because it's relatively unlikely that any message in the buffer is one that would survive the filter anyway. Is there a way to do that? I lose a lot more events waiting to notice that the script has crashed than I would lose dumping the buffer.

[emilyhunt]

I think that after the error you need to re-initialize a client. See the implementation in the example feed generator.

I'm just watching for likes on a specific post and triggering another action on them. Is there a more efficient way to filter this?

If it's really just one post, then watching for reactions on all posts in the network is inefficient - using the ATProto client, you could just look up the likes on the post once every few minutes with AsyncClient.get_likes(). If you just need a total like count and don't mind who it's by, then AsyncClient.get_post() returns the total like count of a post (I think, if I'm remembering correctly).

There's an hourly ratelimit of 3000 on the Bluesky API - I wouldn't recommend getting anywhere near to that unless it's essential (we want to be nice!), but querying a post once every ~5 minutes with get_likes() won't break the bank there at all.

[jetpacktuxedo]

Ahhhh, I missed that they were reinitializing everything there. That should be an easy fix at least.

I think you're roght that it would be more efficient to just check the post directly every few minutes, though in this case it is part of a larger system... A labeler that already runs as close to the (seemingly undocumented) rate limit of label applications as possible. So I'm trying to minimize other calls where I can.