> ## Documentation Index
> Fetch the complete documentation index at: /docs/llms.txt
> Use this file to discover all available pages before exploring further.

# __init__

# `prefect.server.worker_communication.cleanup_queue`

## Functions

### `record_cleanup_queue_dead_letter` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/worker\_communication/cleanup\_queue/\_\_init\_\_.py#L284](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/worker_communication/cleanup_queue/__init__.py#L284)</sup>

```python theme={null}
record_cleanup_queue_dead_letter(dead_letter: CleanupQueueDeadLetter) -> None
```

Record observability signals for a cleanup message DLQ transition.

### `record_cleanup_queue_lease_expiry_result` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/worker\_communication/cleanup\_queue/\_\_init\_\_.py#L313](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/worker_communication/cleanup_queue/__init__.py#L313)</sup>

```python theme={null}
record_cleanup_queue_lease_expiry_result(result: CleanupQueueLeaseExpiryResult) -> None
```

Record aggregate lease-expiry transition metrics.

### `record_cleanup_queue_operation` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/worker\_communication/cleanup\_queue/\_\_init\_\_.py#L332](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/worker_communication/cleanup_queue/__init__.py#L332)</sup>

```python theme={null}
record_cleanup_queue_operation(operation: str) -> None
```

Record observability signals for a cleanup queue operation.

### `get_worker_cleanup_queue` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/worker\_communication/cleanup\_queue/\_\_init\_\_.py#L357](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/worker_communication/cleanup_queue/__init__.py#L357)</sup>

```python theme={null}
get_worker_cleanup_queue() -> WorkerCleanupQueue
```

Return a cleanup queue instance from the configured storage module.

## Classes

### `CleanupQueueMessage` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/worker\_communication/cleanup\_queue/\_\_init\_\_.py#L48](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/worker_communication/cleanup_queue/__init__.py#L48)</sup>

**Methods:**

#### `model_validate_list` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/\_internal/schemas/bases.py#L56](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/_internal/schemas/bases.py#L56)</sup>

```python theme={null}
model_validate_list(cls, obj: Any) -> list[Self]
```

#### `reset_fields` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/\_internal/schemas/bases.py#L85](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/_internal/schemas/bases.py#L85)</sup>

```python theme={null}
reset_fields(self: Self) -> Self
```

Reset the fields of the model that are in the `_reset_fields` set.

**Returns:**

* A new instance of the model with the reset fields.

### `CleanupQueueReservation` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/worker\_communication/cleanup\_queue/\_\_init\_\_.py#L63](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/worker_communication/cleanup_queue/__init__.py#L63)</sup>

### `CleanupQueueDeadLetter` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/worker\_communication/cleanup\_queue/\_\_init\_\_.py#L69](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/worker_communication/cleanup_queue/__init__.py#L69)</sup>

**Methods:**

#### `model_validate_list` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/\_internal/schemas/bases.py#L56](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/_internal/schemas/bases.py#L56)</sup>

```python theme={null}
model_validate_list(cls, obj: Any) -> list[Self]
```

#### `reset_fields` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/\_internal/schemas/bases.py#L85](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/_internal/schemas/bases.py#L85)</sup>

```python theme={null}
reset_fields(self: Self) -> Self
```

Reset the fields of the model that are in the `_reset_fields` set.

**Returns:**

* A new instance of the model with the reset fields.

### `CleanupQueueOperationResult` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/worker\_communication/cleanup\_queue/\_\_init\_\_.py#L81](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/worker_communication/cleanup_queue/__init__.py#L81)</sup>

**Methods:**

#### `model_validate_list` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/\_internal/schemas/bases.py#L56](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/_internal/schemas/bases.py#L56)</sup>

```python theme={null}
model_validate_list(cls, obj: Any) -> list[Self]
```

#### `reset_fields` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/\_internal/schemas/bases.py#L85](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/_internal/schemas/bases.py#L85)</sup>

```python theme={null}
reset_fields(self: Self) -> Self
```

Reset the fields of the model that are in the `_reset_fields` set.

**Returns:**

* A new instance of the model with the reset fields.

### `CleanupQueueLeaseExpiryResult` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/worker\_communication/cleanup\_queue/\_\_init\_\_.py#L92](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/worker_communication/cleanup_queue/__init__.py#L92)</sup>

**Methods:**

#### `model_validate_list` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/\_internal/schemas/bases.py#L56](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/_internal/schemas/bases.py#L56)</sup>

```python theme={null}
model_validate_list(cls, obj: Any) -> list[Self]
```

#### `reset_fields` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/\_internal/schemas/bases.py#L85](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/_internal/schemas/bases.py#L85)</sup>

```python theme={null}
reset_fields(self: Self) -> Self
```

Reset the fields of the model that are in the `_reset_fields` set.

**Returns:**

* A new instance of the model with the reset fields.

### `CleanupQueueWakeup` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/worker\_communication/cleanup\_queue/\_\_init\_\_.py#L99](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/worker_communication/cleanup_queue/__init__.py#L99)</sup>

**Methods:**

#### `model_validate_list` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/\_internal/schemas/bases.py#L56](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/_internal/schemas/bases.py#L56)</sup>

```python theme={null}
model_validate_list(cls, obj: Any) -> list[Self]
```

#### `reset_fields` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/\_internal/schemas/bases.py#L85](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/_internal/schemas/bases.py#L85)</sup>

```python theme={null}
reset_fields(self: Self) -> Self
```

Reset the fields of the model that are in the `_reset_fields` set.

**Returns:**

* A new instance of the model with the reset fields.

### `WorkerCleanupQueue` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/worker\_communication/cleanup\_queue/\_\_init\_\_.py#L106](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/worker_communication/cleanup_queue/__init__.py#L106)</sup>

Interface for cleanup delivery queue storage.

Implementations own cleanup message reservation correctness. WebSocket
dispatchers may keep process-local routing state, but ack, release, renew,
lease expiry, retry accounting, and DLQ transitions must go through this
queue. Implementations own the server retry and lease policy and completed
idempotency retention semantics.

**Methods:**

#### `ack` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/worker\_communication/cleanup\_queue/\_\_init\_\_.py#L157](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/worker_communication/cleanup_queue/__init__.py#L157)</sup>

```python theme={null}
ack(self) -> CleanupQueueOperationResult
```

Complete a reserved cleanup message.

The operation must validate the work-pool scope and current reservation
token atomically before removing the message from active delivery.
Implementations should retain completed idempotency state according to
their configured retention policy.

#### `enqueue` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/worker\_communication/cleanup\_queue/\_\_init\_\_.py#L117](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/worker_communication/cleanup_queue/__init__.py#L117)</sup>

```python theme={null}
enqueue(self) -> CleanupQueueMessage
```

Store a cleanup message if it has not already been produced.

Implementations should treat `message_id` and `idempotency_key` as stable
producer identifiers within a work pool, returning the existing message
for repeated enqueue attempts instead of creating duplicates. The
optional `work_queue_id` is advisory targeting metadata.

#### `expire_leases` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/worker\_communication/cleanup\_queue/\_\_init\_\_.py#L207](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/worker_communication/cleanup_queue/__init__.py#L207)</sup>

```python theme={null}
expire_leases(self) -> CleanupQueueLeaseExpiryResult
```

Expire overdue reservations in bounded batches.

Expired messages should become eligible for redelivery or move to the
dead-letter queue according to retry policy. Implementations may scope
the sweep to a work pool when `work_pool_id` is provided.

#### `read_dead_letter` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/worker\_communication/cleanup\_queue/\_\_init\_\_.py#L236](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/worker_communication/cleanup_queue/__init__.py#L236)</sup>

```python theme={null}
read_dead_letter(self) -> CleanupQueueDeadLetter | None
```

Read a dead-letter entry by work-pool scope and message ID.

This is an inspection helper for terminal cleanup failures. It must not
return a dead-letter entry from a different work pool.

#### `read_message` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/worker\_communication/cleanup\_queue/\_\_init\_\_.py#L222](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/worker_communication/cleanup_queue/__init__.py#L222)</sup>

```python theme={null}
read_message(self) -> CleanupQueueMessage | None
```

Read an active cleanup message by work-pool scope and message ID.

This is an inspection helper for messages that have not been acked or
dead-lettered. It must not return a message from a different work pool.

#### `read_wakeup_sequence` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/worker\_communication/cleanup\_queue/\_\_init\_\_.py#L259](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/worker_communication/cleanup_queue/__init__.py#L259)</sup>

```python theme={null}
read_wakeup_sequence(self, work_pool_id: UUID) -> int
```

Return the latest wakeup sequence observed for a work pool.

Callers use this value as the `after` cursor when waiting for future
wakeups.

#### `release` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/worker\_communication/cleanup\_queue/\_\_init\_\_.py#L174](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/worker_communication/cleanup_queue/__init__.py#L174)</sup>

```python theme={null}
release(self) -> CleanupQueueOperationResult
```

Give up the current reservation without completing the cleanup message.

The operation must validate the work-pool scope and current reservation
token atomically, then either make the message eligible for redelivery or
move it to the dead-letter queue when retry policy is exhausted.

#### `renew` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/worker\_communication/cleanup\_queue/\_\_init\_\_.py#L191](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/worker_communication/cleanup_queue/__init__.py#L191)</sup>

```python theme={null}
renew(self) -> CleanupQueueOperationResult
```

Extend the lease for the current reservation.

The operation must validate the work-pool scope and current reservation
token atomically. Renewing a reservation should not increment delivery
count because no new delivery has been committed.

#### `reserve` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/worker\_communication/cleanup\_queue/\_\_init\_\_.py#L138](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/worker_communication/cleanup_queue/__init__.py#L138)</sup>

```python theme={null}
reserve(self) -> CleanupQueueReservation | None
```

Atomically reserve one eligible cleanup message for delivery.

A successful reservation must increment the committed delivery count,
create exactly one active reservation, and return an unguessable token
required for follow-up operations. `preferred_work_queue_ids` should be
treated as an advisory preference, with pool-wide fallback controlled by
`allow_fallback_to_any_queue`.

#### `wait_for_wakeup` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/worker\_communication/cleanup\_queue/\_\_init\_\_.py#L268](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/worker_communication/cleanup_queue/__init__.py#L268)</sup>

```python theme={null}
wait_for_wakeup(self, work_pool_id: UUID) -> CleanupQueueWakeup | None
```

Wait for a work-pool wakeup sequence newer than `after`.

Returns the next wakeup when one is observed, or `None` when `timeout`
elapses before a newer wakeup is available.

#### `wake_dispatchers` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/worker\_communication/cleanup\_queue/\_\_init\_\_.py#L250](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/worker_communication/cleanup_queue/__init__.py#L250)</sup>

```python theme={null}
wake_dispatchers(self, work_pool_id: UUID) -> CleanupQueueWakeup
```

Notify dispatchers that cleanup work may be available for a work pool.

Implementations should advance and return a monotonic wakeup sequence so
local dispatchers can avoid missing notifications.