Add manual task acknowledgement

[vvanglro]

Why

Some broker backends support explicit message acknowledgement, but taskiq currently only lets workers choose a fixed automatic acknowledgement point: when received, when executed, or when saved.

For some tasks, acknowledgement needs to happen after an application-level durable action is completed, but before the whole task finishes. Examples include committing an external transaction, persisting an idempotency marker, or coordinating with another system where the task code knows the correct acknowledgement boundary better than the worker.

This change adds an explicit manual acknowledgement mode so advanced users can acknowledge from inside the task through Context.ack(), while preserving the existing automatic acknowledgement behavior for all other modes.

Summary

• add manual acknowledgement type
• expose Context.ack(), Context.is_ackable, and Context.is_acked for task-level manual acknowledgement
• keep automatic acknowledgement idempotent through a shared ack controller
• document manual acknowledgement and add receiver/ack parser tests

Compatibility

Existing acknowledgement modes are unchanged. Manual acknowledgement only works for brokers that yield AckableMessage; otherwise Context.ack() raises a clear runtime error.

Tests

• env -u PYTHONPATH -u VIRTUAL_ENV .venv/bin/python -m pytest -q
• env -u PYTHONPATH -u VIRTUAL_ENV .venv/bin/python -m ruff check .
• env -u PYTHONPATH -u VIRTUAL_ENV .venv/bin/python -m black --check .

[codecov]

Codecov Report

❌ Patch coverage is 97.29730% with 1 line in your changes missing coverage. Please review.
✅ Project coverage is 80.20%. Comparing base (fe92683) to head (d6af7d6).

Files with missing lines	Patch %	Lines
taskiq/context.py	92.30%	1 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##           master     #637      +/-   ##
==========================================
+ Coverage   80.02%   80.20%   +0.18%     
==========================================
  Files          69       69              
  Lines        2548     2577      +29     
==========================================
+ Hits         2039     2067      +28     
- Misses        509      510       +1     

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

[s3rius]

I love to see this feature. Also it shines along with per-task ack support.

So, for example I can see a scenario for a long-running tasks, where you can do something like:

taskiq worker --ack-type=when_executed

@broker.task(ack_type="manual")
async def testing(context):
    # Long-running stuff
    await context.ack()

But there is a small problem, which I see. Not with this pull request tho, but there is one piece missing.

For long-running tasks we need to tell our broker to not send the message we're processing right now to any other subscriber.

For example in nats there is a special method for this:

AckProgress: https://nats-io.github.io/nats.net/api/NATS.Client.JetStream.NatsJSMsg-1.AckProgressAsync.html

What do you think?

[vvanglro]

Yes, for Redis streams this can happen because taskiq-redis uses XAUTOCLAIM for messages idle longer than idle_timeout.

I think this is a separate broker-level progress/heartbeat problem. Manual ack gives task code control over final acknowledgement, but long-running tasks may also need a way to extend/reset the pending idle timeout before the final ack.

We could discuss a follow-up API like context.ack_progress() / context.extend_ack_timeout(), implemented only by brokers that support it.

[s3rius]

Yeah, that sounds good to me.

[vvanglro]

I just asked Codex to investigate how this maps to the existing brokers, especially taskiq-nats. Here is what it found.

NATS JetStream already supports this.

In nats-py, the equivalent of AckProgress is:

await message.in_progress()

It sends the JetStream +WPI progress ack.

Current state in taskiq-nats:

Both PushBasedJetStreamBroker and PullBasedJetStreamBroker currently expose only the final ack callback:

AckableMessage(
    data=nats_message.data,
    ack=nats_message.ack,
)

So taskiq can call the final ack, but it cannot currently send progress ack / heartbeat while the task is still running.

Possible follow-up implementation:

Extend AckableMessage with an optional progress callback:

class AckableMessage(BaseModel):
    data: bytes
    ack: Callable[[], None | Awaitable[None]]
    ack_progress: Callable[[], None | Awaitable[None]] | None = None

Expose it from task context:

await context.ack_progress()

And wire it in taskiq-nats JetStream brokers:

AckableMessage(
    data=nats_message.data,
    ack=nats_message.ack,
    ack_progress=nats_message.in_progress,
)

For brokers that do not support this, context.ack_progress() could raise a clear RuntimeError, similar to manual ack on non-ackable messages.

Then long-running tasks could do something like:

@broker.task(ack_type="manual")
async def long_task(context: Context) -> None:
    while still_working:
        await do_some_work()
        await context.ack_progress()

    await context.ack()

So I think NATS would be a good first broker to support this API, because it maps directly to JetStream’s native progress ack.