dagucloud / dagucloud/dagu

feat: distributed named locks, semaphores, and rate-limit pools

Open
#2,647 1 comment 0 reactions 0 assignees View on GitHub

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

Open the contributing guide

First steps

  1. Read the whole issue, then the project's contributing guide.
  2. Comment on the issue to say you are picking it up — it saves two people doing the same work.
  3. Fork the repository and make your change on a branch.
  4. 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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.