Pull and execute R tasks

Claims and runs tasks whose names appear in handlers, one attempt at a time, until max_tasks or idle_timeout is reached. Scale out with more worker processes connected through Quack. Leases are renewed by ca_step(), ca_sleep() and ca_heartbeat() only; no background thread does it, so supervise long external commands with ca_process(). Losing a lease never ends the worker process.

Usage

ca_work(
  db,
  handlers,
  queue = "default",
  max_tasks = Inf,
  poll_seconds = 0.1,
  idle_timeout = Inf,
  lease_seconds = 30,
  worker = paste0("R-", Sys.getpid()),
  reap_limit = 64L,
  failure_delay = 0,
  conflict_retries = 8L,
  on_result = NULL
)

Arguments

db

A handle returned by ca_open(), ca_serve(), or ca_connect().

handlers

Named list of functions taking (input, ctx).

queue

Queue name.

max_tasks

Maximum number of claimed attempts, including failures and suspensions. Inf runs without an attempt limit.

poll_seconds

Positive delay when no work is found.

idle_timeout

Nonnegative seconds without a claim before returning. Inf waits indefinitely. Zero drains currently eligible work.

lease_seconds

Positive lease duration in seconds. Long-running handlers must explicitly heartbeat before this interval expires.

worker

Worker label used for inspection; lease tokens establish ownership.

reap_limit

Maximum exhausted, expired leases to mark failed before each claim. Batching bounds contention with other workers; it does not limit recovery of leases that still have failure budget.

failure_delay

Nonnegative seconds before a failed task can be claimed again. This is separate from caller-managed SQL conflict retries.

conflict_retries

Nonnegative number of automatic retries for each statement aborted by a known write conflict. Zero installs no retry policy, leaving canard_retryable to the caller's own calling handler.

on_result

Optional function called with each ca_run() outcome after persistence. The default warns about handler failures with canard_task_failed. A supplied callback owns outcome reporting instead. Errors from the callback propagate to the caller.

Value

The number of executed attempts, invisibly. Use on_result to collect outcomes without accumulating an unbounded result list inside the worker.

Details

Competing workers pick the same eligible task, and DuckDB aborts the losing statement. The worker retries such a statement up to conflict_retries times, backing off from 10 ms to 0.5 s (about 1.6 seconds in total by default), repeating only the SQL and never an R callback, and signalling canard_conflict_retry beforehand; see ca_conditions(). Jitter comes from the clock and process ID, leaving R's random number stream untouched. An exhausted budget raises canard_conflict. Lease loss and transport failures are never retried.