feat: distributed named locks, semaphores, and rate-limit pools
Nobody has claimed this yet.
- Dominant language
- Go
- Stars
- 4k
- Forks
- 337
- Avg merge
- 19h 26m
- Merged PRs (30d)
- 133
Description
## Is your feature request related to a problem?
Dagu provides DAG queues, `max_active_runs`, `max_active_steps`, parallel limits, and worker selectors. These control broad execution concurrency, but they cannot safely coordinate access to a named external resource across unrelated DAGs and steps.
Common examples are:
- only one writer may update a DuckDB/DuckLake catalog or table at a time;
- at most four jobs may use a memory-heavy parser regardless of which DAG starts them;
- a government API allows 100 requests per minute across all workflows;
- only one deployment may target a specific customer/environment;
- jobs for different keys may run concurrently, while jobs for the same key must serialize.
A queue is too coarse because it applies to whole DAG runs and usually has a static name. State actions can implement a best-effort flag, but not a safe distributed lease with fencing, expiry, fairness, and automatic cleanup. #2047 also shows that queue semantics for child DAGs are currently not sufficient as a general resource-coordination primitive.
## Describe the solution you'd like
Add distributed named coordination primitives that work consistently in local and distributed execution.
Illustrative syntax:
```yaml
steps:
- id: publish_companies
run: python publish.py
coordination:
locks:
- key: "ducklake:${CATALOG}:main.companies"
mode: exclusive
timeout: 30m
- id: parse_accounts
run: python parse.py
coordination:
semaphores:
- key: xbrl-parser
permits: 1
capacity: 4
- id: call_registry_api
run: python fetch.py
coordination:
rate_limits:
- key: "registry:${COUNTRY}"
limit: 100
interval: 1m
burst: 10
```
The exact syntax is illustrative. The requested behavior is:
1. **Named exclusive/shared locks**
- Dynamic keys derived from parameters, partition keys, item IDs, or environment values.
- DAG- or step-scoped acquisition.
- Optional shared/read and exclusive/write modes.
- Configurable wait timeout and behavior when the lock cannot be acquired.
2. **Named semaphores**
- A global capacity and a configurable number of permits requested by each run/step.
- Capacity must be enforced across DAGs, child DAGs, retries, schedulers, coordinators, and workers.
- Permit acquisition should happen before the guarded command starts.
3. **Rate-limit pools**
- Token-bucket or equivalent semantics with limit, interval, and optional burst.
- Shared across all consumers of the same key.
- Ability to honor a provider-supplied retry/reset time where an action exposes it.
- Waiting should not busy-loop or consume an execution slot unnecessarily where avoidable.
4. **Durable lease and fencing semantics**
- Acquisition must produce a lease/fencing token so a stale worker cannot continue writing after ownership moves elsewhere.
- Heartbeats and expiry for worker/coordinator failure.
- Automatic release on success, failure, abort, timeout, and process crash.
- Recovery must not create two simultaneous owners.
5. **Fairness and deadlock safety**
- FIFO or documented fair ordering for waiters.
- Deterministic ordering or validation when a step requests multiple locks.
- Detect/reject obvious self-deadlocks.
- Cancellation removes a waiter promptly.
6. **Visibility and operations**
- UI/API/CLI view of current holders, waiters, capacities, lease age, and guarded DAG run/step.
- Ability for an administrator to release an orphaned lease with an audit record.
- Metrics for wait time, contention, utilization, acquisition failures, and forced releases.
7. **Configuration and security**
- Optional centrally configured pools so a DAG cannot silently raise the capacity or rate limit.
- Per-workspace/profile scope where applicable.
- Backward-compatible behavior for DAGs that do not request coordination.
## Describe alternatives you've considered
- Put every potentially conflicting DAG in the same queue.
- Set `max_active_runs: 1` on each DAG.
- Implement lock files on a shared filesystem.
- Store a lock flag/version with Dagu state actions.
- Use PostgreSQL advisory locks, Redis, or a separate rate-limit service.
- Add locking logic to every script.
External locks are valid for complex deployments, but a lightweight built-in primitive would preserve Dagu's low-operations model and make local and distributed behavior consistent. Queue-based workarounds also serialize unrelated keys and waste capacity.
## Use case
Several ingestion and transformation workflows read from the same DuckLake catalog. Reads can overlap, but a table publish/compaction operation must have a single active writer. At the same time, enrichment DAGs call registry and WHOIS services with provider-specific global rate limits.
Dynamic named locks allow `main.companies` and `main.financials` to be updated independently, while rate-limit pools keep all callers compliant without a central custom proxy.
## Suggested acceptance criteria
- Two unrelated DAG steps requesting the same exclusive lock never run their guarded commands concurrently.
- Different dynamic lock keys may execute concurrently.
- Semaphore capacity is enforced across distributed workers and child DAGs.
- Leases recover safely after worker or coordinator failure, using fencing or equivalent stale-owner protection.
- Rate-limit consumption is shared across DAGs and survives service restarts.
- Lock holders and waiters are visible through API/UI, with contention metrics.
- Existing queue and concurrency behavior remains unchanged for workflows that do not use this feature.
Contributor guide
First steps
- Read the whole issue, then the project's contributing guide.
- Comment on the issue to say you are picking it up — it saves two people doing the same work.
- Fork the repository and make your change on a branch.
- Open a pull request that references the issue number.
Research direction
No files, tests, or entry points are named. Start by mapping Dagu's existing queue, concurrency, scheduler, coordinator, worker, and state-handling paths, then determine how local and distributed execution share coordination. Done requires a scoped design and implementation plan covering leases, fencing, fairness, recovery, rate limits, visibility, and the listed acceptance criteria.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- go
- Domain
- backend, distributed-systems
- Issue type
- Feature
- Difficulty
- 5/5
- Estimated time
- Over a week
- Activity status
- Active
- Clarity
- Needs clarification
- Newbie friendliness
- 25/100