Consumer control

Pausing, throttling, or draining a consumer from Consumers only does something if your service checks in on that state. Here's how that works on each path.

If you're using the event runtime

Nothing to do. Consumers built on SQSConsumer (and the rest of the built-in event runtime) check in automatically as they poll for new messages, so dashboard controls just work.

If you're observing an existing consumer

ConsumerObserver exposes the current control state through observe.control. Check it before pulling your next message:

control = await observe.control.get()

if not control.state.should_pull:
    await asyncio.sleep(1)
    continue  # paused or draining, don't pull anything new

if control.state == "throttled":
    await asyncio.sleep(1.0 / (control.throttle_per_sec or 1.0))  # pace yourself

The four states

State	Meaning
Running	Normal operation, keep pulling.
Paused	Stop pulling entirely until resumed.
Throttled	Keep pulling, but at a reduced rate (throttle_per_sec).
Draining	Let in-flight work finish, then stop pulling. Used before taking a consumer offline.

If Observatory is briefly unreachable

Control checks fail open: if a check-in can't reach Observatory, your consumer keeps whatever state it last successfully saw instead of guessing. A short network blip won't accidentally pause your consumer, and a consumer that was already paused won't spring back to life during an outage either.