Trusted Effects
A handler sometimes needs the host application to commit a domain mutation
while its TaskQ attempt is still active. Reading taskq.jobs directly breaks
the table-privilege boundary; reading an observer projection does not lock or
fence the attempt.
The current SQL contract retains the domain-neutral function and Python adapter for this case:
from taskq.sql import lock_active_effect_attempt
It ships in migration 0018_trusted_effect_fence.sql. There is no matching
HTTP route; this is for a trusted, co-resident host transaction.
Handler side
A handler sees only a bounded request/response operation. Attempt identity stays inside the worker runtime:
from taskq import Complete, JobContext
async def apply_payment(ctx: JobContext, payload: PaymentInput) -> Complete:
result = await ctx.report_effect({
"operation": "payment.apply",
"payment_id": payload.payment_id,
})
return Complete(result={"applied": result["applied"]})
ctx.report_effect(...) is available only when the worker was constructed with
a trusted effect reporter. Requests and responses must be JSON objects and are
limited to 8 KiB by default (WorkerOptions.effect_request_max_bytes and
effect_response_max_bytes).
Host reporter
The reporter receives the hidden active-attempt identity. It must lock that attempt and perform the domain write in the same transaction:
from taskq import TaskqConflictError, WorkerEffectAttempt
from taskq.sql import lock_active_effect_attempt
class HostEffects:
def __init__(self, engine):
self.engine = engine
async def report_effect(
self,
attempt: WorkerEffectAttempt,
request: dict,
) -> dict:
async with self.engine.begin() as connection:
active = await lock_active_effect_attempt(
connection,
job_id=attempt.job_id,
attempt_id=attempt.attempt_id,
worker_id=attempt.worker_id,
queue=attempt.queue,
job_type=attempt.job_type,
)
if active is None:
raise TaskqConflictError(
details={"reason": "inactive_attempt"},
)
# Bind the requested subject to the admitted job payload before
# writing. Never trust request["payment_id"] by itself.
if request["payment_id"] != active.payload["payment_id"]:
raise TaskqConflictError(
details={"reason": "subject_mismatch"},
)
result = await apply_idempotently(
connection,
operation_key=f"{attempt.job_id}:payment.apply",
payment_id=request["payment_id"],
)
return {"applied": result.applied}
lock_active_effect_attempt(...) returns a row only when the job is running,
the attempt/worker/queue/job type all match, the database-clock lease is live,
and cancellation has not been requested. The returned payload, optional
workflow_id, and optional workflow status counts are safe projections; fence
material, headers, progress, result, and error are not exposed.
The database role needs EXECUTE on the function through taskq_producer and
normal rights on the host’s domain tables. It does not need direct access to
TaskQ tables or runner/operator capabilities.
Wire the worker
service = WorkerService(
runner_transport,
registry,
worker_id="worker-1",
options=worker_options,
effect_reporter=HostEffects(domain_engine),
)
WorkerSupervisor accepts the same effect_reporter= argument when you build
the lower-level runtime directly.
Failure and retry rules
- The worker retries
TaskqUnavailableError, timeouts, and connection errors with the settlement retry budget. - A retry replays the same request. The host effect must therefore have a durable idempotency key.
- Non-retryable TaskQ errors return to the handler as failures; the handler must not convert an inactive-attempt result into success.
- The held TaskQ job-row lock serializes the domain transaction with settlement, cancellation, lease expiry, and another trusted effect transaction.
- Commit or roll back the domain effect before returning a response.