Metadata-Version: 2.5
Name: ringo-task-queue
Version: 0.1.0.dev8
Summary: Python asyncio SDK for Ringo Task Queue with a bundled offline Go daemon
Author: Ringo Task Queue contributors
License-Expression: MIT
License-File: LICENSE
Keywords: asyncio,grpc,postgresql,sqlite,task-queue
Classifier: Development Status :: 3 - Alpha
Classifier: Framework :: AsyncIO
Classifier: Intended Audience :: Developers
Classifier: Operating System :: MacOS :: MacOS X
Classifier: Operating System :: Microsoft :: Windows
Classifier: Operating System :: POSIX :: Linux
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3 :: Only
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Requires-Python: >=3.11
Requires-Dist: grpcio-health-checking!=1.82.*,<2,>=1.71.0
Requires-Dist: grpcio-status<2,>=1.71.0
Requires-Dist: grpcio<2,>=1.71.0
Requires-Dist: protobuf<7,>=5.29.5
Provides-Extra: dev
Requires-Dist: build<2,>=1; extra == 'dev'
Requires-Dist: grpcio-tools==1.71.0; extra == 'dev'
Requires-Dist: hatchling<2,>=1.27; extra == 'dev'
Requires-Dist: mypy<2,>=1.15; extra == 'dev'
Requires-Dist: pytest-asyncio>=0.23; extra == 'dev'
Requires-Dist: pytest>=8; extra == 'dev'
Requires-Dist: ruff<1,>=0.11; extra == 'dev'
Provides-Extra: generate
Requires-Dist: grpcio-tools==1.71.0; extra == 'generate'
Provides-Extra: supervisor
Requires-Dist: psutil<8,>=7; (sys_platform != 'win32') and extra == 'supervisor'
Provides-Extra: test
Requires-Dist: pytest-asyncio>=0.23; extra == 'test'
Requires-Dist: pytest>=8; extra == 'test'
Description-Content-Type: text/markdown

# Ringo Task Queue — Python SDK

面向 Python 3.11+ 的 asyncio 任务队列 SDK。它支持 SDK 管理的本地 SQLite/PostgreSQL daemon 和远程 gRPC 服务，提供永久去重、至少一次执行、lease fencing、自动续租、重试和 DEAD/DLQ 语义。

**本文对应发行线：** Python `0.1.0.dev8` / Go `0.1.0-dev.8`，协议 **1.7 / Schema 3**。本文描述该版本代码与 wheel 的能力；是否已经公开上传，以 [PyPI 版本页](https://pypi.org/project/ringo-task-queue/0.1.0.dev8/) 和官方文件为准，候选构建本身不是发布证明。历史 dev7 使用协议 1.5，不包含 supervisor API。仅 Go 数据面与 Python SDK 持续维护；Node.js / Kotlin SDK 封存，保留历史代码/文档/测试，不维护、不发布，恢复须用户明确确认。

## 安装后即可离线运行

```shell
python -m pip install --pre "ringo-task-queue==0.1.0.dev8"
```

平台 wheel 各内置对应 daemon；以下是分发目标，**不等于全部平台已完成原生运行验收**：

| 系统 | wheel tag | 内置 daemon |
|---|---|---|
| Windows x64 | `win_amd64` | Windows x64 |
| Linux x64 | `manylinux_2_17_x86_64` | Linux x64 |
| Linux ARM64 | `manylinux_2_17_aarch64` | Linux ARM64 |
| macOS 13+ Intel | `macosx_13_0_x86_64` | macOS 13+ Intel |
| macOS 13+ Apple Silicon | `macosx_13_0_arm64` | macOS 13+ ARM64 |

`pip` 会自动选择当前平台的 wheel。每个 wheel 已包含匹配的 Go daemon；本地模式不会在安装或运行时下载二进制，不要求安装 Go，也不需要设置 `RINGO_DAEMON_PATH`。PostgreSQL 服务本身不包含在包内。

选择官方索引中实际存在的最新开发版本，需要允许预发布：

```shell
pip install --pre ringo-task-queue
```

## 快速开始

```python
import asyncio
from ringo_task_queue import Ringo


async def main() -> None:
    async with Ringo.local("./data/ringo") as client:
        queue = client.queue("email")
        handled = asyncio.Event()

        async def handle(task, ctx) -> None:
            print(task.id, task.payload)
            await ctx.report_progress(1, 1, "sent")
            handled.set()

        worker = queue.worker(
            handle,
            task_types=["send-email"],
            concurrency=4,
        )
        worker_task = asyncio.create_task(worker.run())

        result = await queue.enqueue(
            unique_key="welcome:user-42",
            task_type="send-email",
            payload={"user_id": 42},
        )
        print(result.task_id, result.created)

        await asyncio.wait_for(handled.wait(), timeout=10)
        await worker.close()
        await worker_task


asyncio.run(main())
```

`(queue, unique_key)` 在任务完整生命周期内永久唯一。重复 enqueue 返回原任务 ID 和 `created=False`，不会创建副本，也不会抛出重复异常。

## 本地模式

### SQLite

```python
async with Ringo.local("./data/ringo") as client:
    queue = client.queue("crawl")
```

默认本地模式使用进程私有 daemon：SDK 自动完成 daemon 校验、随机 loopback 端口启动、健康检查、协议协商和退出回收。SQLite 数据保存在 `data_dir`；daemon 日志追加到 `data_dir/ringo-daemon.stderr.log`。两个默认模式的 daemon 不能同时占用同一个 data dir。

若同一台机器上、**同一用户**的多个 SDK 进程需要共享一个 SQLite 队列，可显式启用共享模式：

```python
async with Ringo.local("./data/ringo", shared=True) as client:
    result = await client.queue("crawl").enqueue(
        unique_key="page:1", task_type="fetch", payload={"page": 1}
    )
```

第一个客户端自动启动唯一持锁的共享 daemon，后续客户端认证后附着；启动者退出不影响仍在工作的客户端。最后一个客户端离开后 daemon 延时退出；崩溃后存活客户端会重新发现并接入，不会重放结果不明的写请求。无需手动常驻服务。该模式仅支持本机 SQLite，不支持跨用户、网络共享目录或 PostgreSQL。默认私有模式保持不变。

### PostgreSQL

```python
import os
from ringo_task_queue import PostgresStorage, Ringo


async with Ringo.local(
    "./data/ringo-process",
    storage=PostgresStorage(os.environ["RINGO_POSTGRES_DSN"]),
) as client:
    queue = client.queue("crawl")
```

PostgreSQL 必须由应用提供。SDK 不下载、启动或停止 PostgreSQL。DSN 通过子进程环境传递，不放进进程参数，`PostgresStorage` 的 `repr` 也会隐藏它。

## 远程模式

```python
async with Ringo.connect(
    "https://queue.internal.example:7233",
    token="service-token",
) as client:
    queue = client.queue("crawl")
```

`grpc://`/`http://` 使用明文，`grpcs://`/`https://` 使用 TLS；省略端口时默认 `7233`。

受限链路可以启用整个 gRPC channel 的 gzip：

```python
async with Ringo.connect(
    "https://queue.internal.example:7233",
    token="service-token",
    compression="gzip",
) as client:
    ...
```

`compression` 只接受 `"none"` 或 `"gzip"`。它覆盖 unary RPC 和 Work stream 并在重连后保持；本地模式始终不压缩。

协议 1.6 的 ordered Work sources / SQLite ACK group commit 是当前源码的历史演进边界，1.7 的只读 `ProbeQueue` 用于可选疑似停滞观察；都不是已发布 dev7 / 协议 1.5 能力。远端可以按契约降级，bundled/shared-local 的本地 SDK/daemon 仍须精确匹配版本、协议和 schema，不能因 Schema 相同就混装。

## 提交任务

```python
from datetime import datetime, timedelta, timezone
from ringo_task_queue import RetryPolicy, RetryStrategy, TaskSpec

spec = TaskSpec(
    unique_key="page:10001",
    task_type="fetch-page",
    payload={"url": "https://example.com/10001"},
    priority=10,
    available_at=datetime.now(timezone.utc) + timedelta(seconds=5),
    retry_policy=RetryPolicy(
        max_attempts=5,
        strategy=RetryStrategy.EXPONENTIAL,
        initial_delay=timedelta(seconds=1),
        max_delay=timedelta(seconds=60),
        multiplier=2.0,
        jitter=0.2,
    ),
)
result = await queue.enqueue(spec)
```

也可以使用关键字参数：

```python
result = await queue.enqueue(
    unique_key="page:10001",
    task_type="fetch-page",
    payload={"url": "https://example.com/10001"},
)
```

payload 必须显式提供；显式 `None` 表示 JSON null。允许 JSON 对象、数组和标量，但：

- UTF-8 JSON 编码不超过 1 MiB；
- 不允许非有限浮点数；
- 整数必须位于 `[-2**53, 2**53]`；
- JSON 对象的键必须是字符串；
- 实际使用建议保持在 16–64 KiB 内，大对象只传外部引用。

### 批量提交

```python
batch = await queue.enqueue_many(
    [
        TaskSpec(
            unique_key=f"page:{i}",
            task_type="fetch-page",
            payload={"page": i},
        )
        for i in range(100)
    ]
)
print(batch.created_count, batch.duplicate_count)
```

`enqueue_many` 接受任意数量任务，SDK 自动按服务端单批上限 1,000 条切分并顺序发送，最终聚合为单个 `BatchResult`；`batch.results` 保留逐项结果。若任一批次失败，异常直接抛出，不返回部分结果。

## 查询和手工重试

```python
task = await queue.get_by_id(result.task_id)
task = await queue.get_by_unique_key("page:10001")

# 始终按业务 unique_key 解释
reopened = await queue.retry("page:10001")

# 显式按服务端 task ID
reopened = await queue.retry_by_id(result.task_id)
```

状态包括 `SCHEDULED`、`READY`、`LEASED`、`SUCCEEDED`、`DEAD` 和 `CANCELLED`。`Task` 还提供最后错误、进度、当前 lease 摘要和最近 attempt 预览（最多 32 条 / 512 KiB）；`task.attempts_truncated` 明确表示省略。手工 retry 只重置本轮 attempt，不删除累计历史或永久去重关系。

协议 1.5 的完整历史分页按稳定的 `history_id` 排序，首次读取的高水位保证后续新 attempt 不混入同一分页快照：

```python
page = await queue.get_attempts("page:10001", limit=32)
# 或按服务端 task ID：await queue.get_attempts_by_id(result.task_id)
while True:
    for attempt in page.attempts:
        print(attempt.history_id, attempt.number, attempt.outcome)
    if page.next_cursor is None:
        break
    page = await queue.get_attempts("page:10001", limit=32, cursor=page.next_cursor)
```

页上限为 100 条 / 1 MiB；Python 整数保留 uint64 ID 精度。`number` 在显式 retry 后可以再次从 1 开始，不是累计行 ID。只读 `await queue.stats()` 需要协议 1.4+，返回全生命周期的状态计数快照，不运行 reap/retry。

## Worker（推荐）

处理队列任务推荐直接使用 `Worker`，而不是下面的手工 Claim：Worker 自动管理注册、credit、lease 续租、并发和优雅关闭。

```python
from datetime import timedelta
from ringo_task_queue import PermanentTaskError, RetryTaskError


async def handle(task, ctx) -> None:
    await ctx.report_progress(1, 2, "started")
    if should_retry(task):
        raise RetryTaskError("rate limited", delay=30)  # 秒
    if invalid(task):
        raise PermanentTaskError("invalid input")
    await process(task)


worker = queue.worker(
    handle,
    task_types=["fetch-page"],
    concurrency=32,
    lease_duration=timedelta(seconds=60),
    grace_period=timedelta(seconds=30),
)
await worker.run()
```

`run()` 一直运行到其他协程调用 `await worker.close()`，并且只能调用一次。常驻消费时用 `asyncio.create_task(worker.run())` 启动；已知有限批次（见下）直接 `await worker.run()`。

### 三种并发模式、共享代理池和 pause/resume

整数是固定并发，创建时至少为 1；运行后可以把目标改成 `0..2**31-1`：

```python
worker = queue.worker(handle, concurrency=8)
run_task = asyncio.create_task(worker.run())
await worker.set_concurrency(0)  # pause：不再派发，已有 handler 不取消
await worker.set_concurrency(8)  # resume/resize
```

`ResourceConcurrency` 是同一进程/事件循环可由多个 queue/client 共享的总容量。容量是 admission permit，不是每个 Worker 的副本；缩容不会抢占已运行的 handler。

```python
from ringo_task_queue import ResourceConcurrency

proxies = ResourceConcurrency(capacity=12)
west = client.queue("west").worker(handle, concurrency=proxies, weight=1)
east = client.queue("east").worker(handle, concurrency=proxies, weight=3)
runs = [asyncio.create_task(w.run()) for w in (west, east)]

await proxies.set_capacity(20)  # 例如健康代理数变化
# 共享 controller 时不能调用 west.set_concurrency(...)

await west.close()
await east.close()
await asyncio.gather(*runs)
```

持续有任务时，权重按平滑加权轮转长期接近 1:3；空闲成员的临时 probe 会在 250 ms 后撤回，因此容量会给活跃成员 work-conserving 使用。可手动暂停整组：`await proxies.set_capacity(0)`，恢复时再设置非零容量。

`AdaptiveConcurrency` 使用 AIMD：当前容量个成功 ACK（且当时已饱和）后 `+1`，只有成功 NACK 且 `RetryTaskError(congestion=True)` 才按 `*0.5` 降低，始终限制在 `minimum..maximum`；普通异常、永久失败和 lease/传输丢失不改变它：

```python
from ringo_task_queue import AdaptiveConcurrency, RetryTaskError

limits = AdaptiveConcurrency(initial=4, minimum=1, maximum=32)

async def handle(task, ctx):
    try:
        await call_upstream(task.payload)
    except RateLimitError as exc:
        raise RetryTaskError(str(exc), delay=2, congestion=True) from exc

worker = queue.worker(handle, concurrency=limits)
```

动态 controller 和运行中的 fixed resize 需要协议 minor 2；不改变 fixed 并发的 Worker 仍兼容旧 minor。controller 可以跨 queue/client 共享，但不能跨进程共享。

Handler 映射：

| 结果 | 动作 |
|---|---|
| 正常返回 | ACK / `SUCCEEDED` |
| 普通异常 | NACK / 按策略重试 |
| `RetryTaskError` | NACK，可覆盖本次延迟；仅 `congestion=True` 且 NACK 被服务端接受时触发 Adaptive 的降速 |
| `PermanentTaskError` | Reject / `DEAD` |
| 取消 | 不发送终态，等待 lease 到期重投 |

Worker 持有 lease token，自动续租并严格控制并发。`TaskContext` 提供 `attempt`、`max_attempts`、`worker_id`、`cancelled`、`wait_cancelled()` 和 `report_progress()`。lease 丢失、stream 断开或强制关闭时，context 会被取消。

这是至少一次系统：handler 的外部副作用必须使用 `unique_key` 或任务 ID 实现幂等。

### 有限批次与 `on_idle`（PyPI 推荐范式）

已知有限批次（例如"把这批 seed 跑完就退出"）的推荐写法：**先在 `run()` 之前提交幂等 seed**，再给 Worker 配置一次性的 `on_idle` 回调，并**直接 `await worker.run()`**：

```python
import asyncio
import sys

from ringo_task_queue import Ringo


async def main() -> None:
    async with Ringo.local("./data/ringo") as client:
        queue = client.queue("email")

        # 1. 先提交幂等 seed（run() 之前 enqueue）。
        for user_id in (101, 102, 103):
            await queue.enqueue(
                unique_key=f"welcome:{user_id}",
                task_type="send-email",
                payload={"user_id": user_id},
            )

        async def handle(task, ctx) -> None:
            await send_welcome(task.payload["user_id"])

        # 2. 队列在该 Worker 的 task type 范围内连续空闲满 1 秒后触发一次。
        worker = queue.worker(
            handle,
            task_types=["send-email"],
            concurrency=4,
            on_idle=lambda: sys.exit(0),
        )

        # 3. 前台直接 await run()；on_idle 触发后整个批次随之结束。
        await worker.run()


asyncio.run(main())
```

语义与限制（务必阅读）：

- **固定参数**：SDK 以 250 ms 间隔轮询、连续空闲满 1 秒触发一次；这两个值是协议级常量，不提供配置参数。
- **单次触发**：回调最多触发一次。正常返回后 Worker 继续常驻运行且不再触发；要结束进程请在回调里退出（如 `sys.exit(0)`），异常会原样从 `await worker.run()` 传播。
- **type scope**：活动数只统计该 Worker `task_types` 范围内的任务（`SCHEDULED`/`READY`/`LEASED` 都算活动），不影响队列里其他类型的任务。
- **开放队列竞态**：空闲观察只是开放队列上的观测快照，不是持久完成判定。回调**正常返回**且 Worker 保持常驻时，观察之后提交的新任务仍会被正常消费；但如果回调直接退出进程（如 `sys.exit(0)`），观测之后、退出之前竞态提交的任务不会被本 Worker 处理，可能留给其他 Worker 或下一次运行。因此 `sys.exit(0)` 范式只适合生产者已经停止的有限批次。需要严格分布式完成判定时，必须另行设计 queue seal/generation。
- **不要在回调里 `close()`**：`on_idle` 由 `worker.run()` 协程内联执行，在回调里等待同一 Worker/client 的关闭会造成自等待。外层资源作用域（如上面的 `async with`）负责清理。
- **兼容性**：不带 `on_idle` 的 Worker 从不发送活动查询 RPC，与协议 minor 0 的旧 server 完全兼容；配置 `on_idle` 需要 minor ≥ 1 的 server，旧 server 会在 `run()` 快速抛出 `IncompatibleVersionError`。

## 手工 Claim

> 常规消费任务优先使用上面推荐的 `Worker`；手工 Claim 适用于需要自行控制 claim/ACK 时序的高级场景。

```python
from datetime import timedelta

leased = await queue.claim(
    limit=10,
    task_types=["fetch-page"],
    wait_timeout=timedelta(seconds=5),
    lease_duration=timedelta(seconds=60),
    worker_id="manual-worker-1",
)

for item in leased:
    try:
        await process(item.task)
        await item.ack()
    except TimeoutError as exc:
        await item.nack(str(exc), delay=timedelta(seconds=10))
    except ValueError as exc:
        await item.reject(str(exc))
```

续租：

```python
await item.extend_lease(timedelta(seconds=60))
```

lease token 不公开。旧 attempt 或迟到 ACK 会被 fencing 拒绝并映射成 `LeaseLostError`。

## 生命周期和关闭

推荐：

```python
async with Ringo.local("./data/ringo") as client:
    ...
```

或手工：

```python
client = Ringo.local("./data/ringo")
await client.start()
try:
    ...
finally:
    await client.close()
```

`close()` 可重复调用，且调用者取消不会中断底层清理。它会先 drain worker，在 grace period 内继续续租并允许 handler 完成，再关闭 channel 和 daemon。被强制取消的 handler 不发送 ACK/NACK。

本地 daemon 意外退出时最多自动重启三次。结果不确定的写操作不会被盲目重放；只读查询可以在重新协商后重试。

## 可选业务进程守护（dev8）

这是显式 opt-in API，不是 `Ringo.local()` 的自动重启选项，也不是通用服务平台。该 API 需要 dev8 或后续支持版本；历史 dev7 不提供。可从完整候选 wheel 集合验收，公开安装是否可用以官方 PyPI 文件为准：

```shell
python -m pip install --pre 'ringo-task-queue[supervisor]==0.1.0.dev8'
```

POSIX supervisor extra 安装 `psutil>=7,<8` 以观察进程身份/RSS；Windows 使用 native Job/进程 API，不强制 psutil。普通 Worker 不必启用 supervisor。源码 checkout 可自行安装 `.[supervisor]` extra 并准备同线 daemon，但源码/交叉构建不等于发行或全平台原生验证。

### 一个同步启动器、两个不可变配置对象

```python
from ringo_task_queue.supervisor import supervise, SupervisorPolicy, TaskTimeouts
# supervise(entrypoint: str, *, policy: SupervisorPolicy | None = None,
#           task_timeouts: TaskTimeouts | None = None) -> int
```

`supervise` 是 **blocking、main-thread** 启动器，必须在运行中的 asyncio loop 之外调用。入口是可导入、可零参调用的 `module:function`，支持 sync/async；同步入口可以自行拥有 `asyncio.run()`，异步入口的 awaitable 也可直接运行。不重建当前命令、不 pickle 闭包，不接受 `__main__` / 局部函数，也不递归监督自己。`SupervisorPolicy` / `TaskTimeouts` 是 immutable 配置，`by_type` 会复制成只读映射；所有公开时长使用 `datetime.timedelta`。

真实 handler 可能健康运行 20–30 分钟时，要**同时**留够业务总预算、Worker drain 和父级整代 shutdown 三个窗口。以下 45/45/46 分钟仅为清楚的配置示例，不是通用 SLA；快任务应另配更短预算。

```python
# myapp.py：可导入的零参业务入口，不在导入时启动事件循环
from datetime import timedelta as td
from ringo_task_queue import Ringo

async def main() -> None:
    async with Ringo.local("./data/ringo") as client:
        async def handle(task, ctx) -> None:
            # 替换成幂等的异步业务；实际阶段完成后才 mark_progress。
            print(task.id, task.payload)
            ctx.mark_progress()

        worker = client.queue("crawl").worker(
            handle, task_types=["fetch-page"], concurrency=4,
            grace_period=td(minutes=45),
            task_timeouts=None,  # 继承 launcher 的 TaskTimeouts
        )
        await worker.run()
```

```python
# launch.py：python launch.py，在最外层保护中同步调用，不放进 asyncio.run
from datetime import timedelta as td
from ringo_task_queue.supervisor import supervise, SupervisorPolicy, TaskTimeouts

if __name__ == "__main__":
    raise SystemExit(supervise(
        "myapp:main",
        policy=SupervisorPolicy(
            recycle_after=td(hours=12),
            shutdown_grace=td(minutes=46),
            no_start_timeout=td(minutes=5),
            restart_on_stall=False,
        ),
        task_timeouts=TaskTimeouts(
            default=td(minutes=45),
            by_type={"fast": td(minutes=2), "unbounded": None},
            progress_timeout=None,
            cancel_grace=td(seconds=5),
        ),
    ))
```

### 总时长、业务进度和外层守护分开

- `TaskTimeouts.default` 是**实际业务执行总预算**，不含本地排队、permit 等待或业务结束后的 ACK/NACK 回执；`by_type` 只覆盖总预算，某类型的 `None` 表示该类型总时长 unlimited，不禁用独立的进度预算。
- `progress_timeout` 仅在首次真实业务进度后计时。`ctx.mark_progress()` 同步、本地、无 RPC，在 Worker 所属 asyncio 循环调用；业务线程可通过 `loop.call_soon_threadsafe(ctx.mark_progress)` 回送标记。`await ctx.report_progress(...)` 也记本地进度。从未报进度不判死；lease 续租、ACK 和其他 Worker 的推进都不是本任务的业务进度。按实际阶段推进标记，不为喂 watchdog 增加逐任务网络调用。
- `queue.worker(..., task_timeouts=None)` 继承 supervisor；传 `TaskTimeouts(...)` **整体替换而不合并**，包括 `by_type`、`progress_timeout` 与 `cancel_grace`。`TaskTimeouts()` 明确使该 Worker 不限任务总时长、不开进度超时。
- `startup_timeout` 只管启动/导入/就绪，`heartbeat_timeout` 管可响应性，`soft_rss_bytes` / `max_rss_bytes` 管 owned 树采样 RSS；**它们不是 task timeout**。默认可选破坏性阈值为 `None`，RSS 不是内核硬限额。硬 RSS 观察在整代 drain 期间仍生效；排空中越限会缩短等待并回收，原计划轮换也按资源故障计入重启预算。纯同步阻塞也会妨碍 asyncio 续租；WebDriver 等阻塞调用应使用异步方式或 `asyncio.to_thread`，而非依靠守护假装续租安全。
- 超时先 fence 再取消；协作停止且 lease/source 仍有效时可用超时错误 NACK。吞取消、`uncancel()` 或超时后返回都不能 ACK 成功。不配合且超过 `cancel_grace` 才请求整代恢复；单独 Worker 只能协作取消，不能强杀 Python/native/browser 代码。父级观察也是采样，不是实时硬截止；含 unlimited 类型时不能臆造通用总时长界限，需要另行选择独立心跳 SLA。
- `no_start_timeout` 启用协议 1.7 的只读 `ProbeQueue`，只查匹配 READY / 到期 SCHEDULED 且剩余 attempt 的 EXISTS，不 claim/reap/count 历史。它按每 Worker/当前 source、实际获 permit 与服务端 accepted credit 判断**疑似停滞**，默认 warn。`restart_on_stall=True` 显式接受多消费者竞争误判风险；远端 minor <7 跳过此探测但保留其他守护，不绕过正常版本握手。

### 资源归属、退出与恢复

Windows 的 root 进程先以 **initially suspended** 状态创建，再 assign 到 native kill-on-close Job 后 resume，避免 venv redirector 在归属前派生 base interpreter；用户模块导入也等待 containment。shared-local daemon 使用显式 breakaway，失败不回退。POSIX 使用进程组/session 与身份跟踪，不是 cgroup，未被观察到就 setsid 逃逸的子进程不保证回收。共享 daemon、外部 PostgreSQL、远程 WebDriver 不归 business 所有；远端浏览器资源仍需业务 `try/finally` 或上下文清理。身份不明/清理不确定时 fail closed，不启动新代。

`shutdown_grace` 是**整代上限**，不是多个 Worker grace 相加；STOP 后所有 Worker 并行 drain。正常 return / exit 0 结束守护；计划轮换已触发时，即使收尾 exit 0 仍重启，且不消耗、不清空故障预算。Ctrl+C 停止而不重启。未确认任务按 lease expiry 恢复，已消耗 attempt 不退款、`maxAttempts` 耗尽进入 DEAD，不伪造成功或自动激活 DEAD，业务必须幂等。supervisor 自身/机器退出后的自动恢复属于 OS 外层服务管理，不由这个 API 承诺。

## 错误处理

```python
from ringo_task_queue import NotFoundError, RingoError, UnavailableError

try:
    task = await queue.get_by_unique_key("missing")
except NotFoundError:
    ...
except UnavailableError as exc:
    logger.warning("temporarily unavailable: %s", exc)
except RingoError as exc:
    logger.exception(
        "queue error code=%s request_id=%s metadata=%r",
        exc.code,
        exc.request_id,
        exc.metadata,
    )
```

公共异常：

- `InvalidArgumentError`；
- `DuplicateError`（普通重复 enqueue 返回 `created=False`）；
- `NotFoundError`；
- `ConflictError`；
- `LeaseLostError`；
- `UnavailableError`；
- `DeadlineExceededError`；
- `IncompatibleVersionError`；
- `InternalError`。

## 默认值与限制

| 项目 | 默认值或限制 |
|---|---|
| 本发行线协议 | 1.7（历史 dev7 为 1.5） |
| schema | 3 |
| lease | 60 秒 |
| worker concurrency | 1 |
| Worker graceful shutdown | 30 秒；长任务显式配置 `grace_period` |
| supervisor task total/progress timeout | 默认均为 `None`，不开任务预算 |
| supervisor shutdown/startup | 分别默认 30 秒；整代 shutdown 与任务总预算独立 |
| supervisor stall action | 默认 `restart_on_stall=False`；仅配置 no-start 窗口后观察并 warn |
| retry attempts | 每轮默认 3，最大 32；累计历史不设生命周期 32 条上限 |
| Task attempt 预览 | 最多 32 条 / 512 KiB，截断明确标记 |
| 完整历史分页 | 默认 32，最多 100 条 / 1 MiB，需要协议 1.5 |
| 默认 decoded gRPC envelope | 16 MiB；显式 channel 选项优先 |
| retry | 指数退避，1 秒到 60 秒，multiplier 2，jitter 0.2 |
| batch | SDK 自动分批，单批最大 1,000 |
| payload | 最大 1 MiB UTF-8 JSON |
| error text | 最大 256 KiB UTF-8 |
| daemon 自动重启 | 最多 3 次 |

## wheel 完整性和二进制解析

本地 daemon 解析优先级：

1. `Ringo.local(..., daemon_path=...)`；
2. `RINGO_DAEMON_PATH`；
3. wheel 内置平台 daemon；
4. 仅用于开发 checkout 的搜索。

普通 PyPI 安装直接使用第 3 项。SDK 在启动前验证 daemon 是普通文件、长度和 SHA-256 与 manifest 一致，然后运行 `version --json` 校验产品名、版本、协议和 schema。校验失败时拒绝启动，不会静默使用错误二进制。

## 常见问题

### `No matching distribution found`

确认 Python 为 3.11+，对应平台 wheel 已上传；安装开发版本时使用 `--pre`。

### 找不到 daemon

正式 wheel 不应出现。使用 `python -m pip show -f ringo-task-queue` 检查是否误装了 editable checkout 或非正式产物。

### `ConflictError`

另一个进程通常已占用同一个 SQLite data dir。复用已有客户端或换一个目录。

### 任务重复执行

这是至少一次语义的正常边界。使用任务 ID/`unique_key` 为外部副作用建立幂等约束。

### Worker 关闭较慢

默认允许 handler 在 30 秒内完成。可以调整 `grace_period`，并让 handler 响应 cancellation 或 `ctx.cancelled`。dev8 源码启用 supervisor 时还必须相应延长整代 `SupervisorPolicy.shutdown_grace`；例如健康 20–30 分钟任务可分别配置任务总预算 45 分钟、Worker grace 45 分钟、父级 shutdown 46 分钟。单独延长任务预算不会延长退出窗口。

## 许可证

MIT License。
