Skip to content

Commit 05ad6b6

Browse files
llambeaullambeau
committed
Add DLQ requeue API endpoint
Add POST /dlq/:queueName/requeue endpoint to move messages from a dead letter queue to a specified destination queue using RabbitMQ Shovel plugin. API: POST /dlq/:queueName/requeue?count=N Body: { "destinationQueue": "target-queue" } Response: { "requeued": N, "failed": 0, "errors": [] } Features: - Uses RabbitMQ Shovel plugin for atomic message transfers - Optional count parameter to limit messages moved - Clear error messages when shovel plugin is not enabled
1 parent 8a5bf38 commit 05ad6b6

12 files changed

Lines changed: 807 additions & 8 deletions

File tree

‎.claude/tasks/ongoing/02-dlq-requeue-feature.md‎

Lines changed: 161 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,161 @@
1+
# Dead Letter Queue Requeue Feature
2+
3+
## Overview
4+
API endpoint to requeue messages from a dead letter queue to a specified destination queue using the RabbitMQ Shovel plugin.
5+
6+
## Requirements
7+
- Support requeue all messages OR a specific count (via query parameter)
8+
- Require explicit destination queue in request body
9+
- Use RabbitMQ Shovel plugin for atomic message transfers
10+
- Self-deleting shovel (no polling, no manual cleanup)
11+
- Fail gracefully if requeue already in progress
12+
- Return estimated message count in response
13+
- Clear error handling when shovel plugin is not enabled
14+
15+
## API Design
16+
17+
### Endpoint
18+
```
19+
POST /dlq/:queueName/requeue?count=N
20+
```
21+
22+
**Parameters:**
23+
- `queueName` (path): Name of the dead letter queue to consume from
24+
- `count` (query, optional): Number of messages to requeue. If omitted, requeues all.
25+
26+
**Request Body:**
27+
```json
28+
{
29+
"destinationQueue": "target-queue"
30+
}
31+
```
32+
33+
**Response:**
34+
```json
35+
{
36+
"status": "initiated",
37+
"requeued": 5,
38+
"failed": 0,
39+
"errors": []
40+
}
41+
```
42+
43+
- `status`: `"initiated"` (async, shovel created) or `"completed"` (sync, e.g. MemoryQueue)
44+
- `requeued`: Estimated number of messages to be moved (queried before shovel creation)
45+
46+
**Error Responses:**
47+
- `400`: Missing/invalid `destinationQueue` or invalid `count`
48+
- `409`-style error: Requeue already in progress for this queue
49+
- `500`: Shovel plugin not available, auth failure, or connection error
50+
51+
## Implementation
52+
53+
### 1. Types in Queue class
54+
**File:** `src/queue/index.ts`
55+
56+
```typescript
57+
export type RequeueOptions = {
58+
count?: number; // Number of messages to requeue (undefined = all)
59+
destinationQueue: string; // Target queue to move messages to
60+
}
61+
62+
export type RequeueResult = {
63+
status: 'initiated' | 'completed';
64+
requeued: number; // Estimated (initiated) or actual (completed) count
65+
failed: number;
66+
errors: Array<{ error: string }>;
67+
}
68+
```
69+
70+
### 2. AmqpQueue implementation via Shovel plugin
71+
**File:** `src/queue/drivers/amqp.ts`
72+
73+
Uses RabbitMQ Management API:
74+
75+
1. **Check for existing shovel** via `GET /api/shovels/{vhost}/{name}`
76+
- Uses deterministic name: `arnavon-requeue-{sourceQueue}`
77+
- Fails if shovel already exists (requeue in progress)
78+
79+
2. **Get queue message count** via `GET /api/queues/{vhost}/{queue}`
80+
- Used to return estimated requeue count
81+
82+
3. **Create self-deleting shovel** via `PUT /api/parameters/shovel/{vhost}/{name}`
83+
- `src-delete-after`: `count` or `'queue-length'` (auto-deletes when done)
84+
- `ack-mode`: `'on-confirm'` for safety
85+
86+
4. **Return immediately** with estimated count
87+
- No polling, no manual deletion
88+
- Shovel auto-deletes after moving messages
89+
90+
Key shovel configuration:
91+
```json
92+
{
93+
"value": {
94+
"src-uri": "amqp://...",
95+
"src-queue": "dead-letters",
96+
"dest-uri": "amqp://...",
97+
"dest-queue": "send-email",
98+
"src-delete-after": "queue-length",
99+
"ack-mode": "on-confirm"
100+
}
101+
}
102+
```
103+
104+
### 3. MemoryQueue implementation
105+
**File:** `src/queue/drivers/memory.ts`
106+
107+
Synchronous implementation moving items between named queues for testing.
108+
Returns `status: 'completed'` with actual count.
109+
110+
### 4. REST API endpoint
111+
**File:** `src/server/rest/index.ts`
112+
113+
```typescript
114+
api.post('/dlq/:queueName/requeue', async (req, res, next) => {
115+
const { queueName } = req.params;
116+
const count = req.query.count ? parseInt(req.query.count as string, 10) : undefined;
117+
const { destinationQueue } = req.body || {};
118+
// Validation and call to Arnavon.queue.requeue()
119+
});
120+
```
121+
122+
## Files Modified
123+
124+
1. `src/queue/index.ts` - Add `requeue` method and types
125+
2. `src/queue/drivers/amqp.ts` - Implement via Shovel API
126+
3. `src/queue/drivers/memory.ts` - Simple queue-to-queue move
127+
4. `src/server/rest/index.ts` - Add `/dlq/:queueName/requeue` endpoint
128+
129+
## Configuration
130+
131+
### RabbitMQ Plugins
132+
**File:** `example/rabbitmq/enabled_plugins`
133+
```erlang
134+
[rabbitmq_management,rabbitmq_shovel,rabbitmq_shovel_management].
135+
```
136+
137+
### Docker Compose
138+
**File:** `example/docker-compose.yml`
139+
- RabbitMQ 4.x with management image
140+
- Mount enabled_plugins file
141+
- Healthcheck for service readiness
142+
143+
## Testing
144+
145+
### Unit Tests
146+
- `tests/queue/index.spec.ts` - Base Queue class requeue method
147+
- `tests/queue/drivers/memory.spec.ts` - MemoryQueue requeue
148+
- `tests/server/rest/api.spec.ts` - REST API endpoint tests
149+
150+
### Integration Tests
151+
- `example/webspicy/arnavon/dlq/_queueName/requeue/post.yml` - Webspicy tests
152+
- `example/webspicy/schema.fio` - Finitio types for request/response
153+
154+
### Manual Testing
155+
1. Start RabbitMQ with shovel plugins enabled
156+
2. Push jobs to a queue with DLQ configured
157+
3. Force jobs to fail (so they go to DLQ)
158+
4. Call `POST /dlq/dead-letters/requeue -d '{"destinationQueue": "send-email"}'`
159+
5. Verify response contains estimated count
160+
6. Verify messages are moved to destination queue
161+
7. Verify shovel auto-deleted after completion

‎example/docker-compose.yml‎

Lines changed: 15 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -1,10 +1,9 @@
1-
version: '3.3'
2-
31
services:
42

53
api:
64
depends_on:
7-
- rabbitmq
5+
rabbitmq:
6+
condition: service_healthy
87
image: quadrabee/arnavon
98
ports:
109
- 3000:80
@@ -17,8 +16,10 @@ services:
1716

1817
workers:
1918
depends_on:
20-
- rabbitmq
21-
- fakesmtp
19+
rabbitmq:
20+
condition: service_healthy
21+
fakesmtp:
22+
condition: service_started
2223
image: example/consumers
2324
build:
2425
context: consumers
@@ -40,13 +41,21 @@ services:
4041
- 1080:1080
4142

4243
rabbitmq:
43-
image: rabbitmq:3.8.11-management
44+
image: rabbitmq:4-management
4445
ports:
4546
- 5672:5672
4647
- 15672:15672
4748
environment:
4849
RABBITMQ_DEFAULT_USER: rabbit
4950
RABBITMQ_DEFAULT_PASS: rabbit
51+
volumes:
52+
# Enable shovel plugins for DLQ requeue functionality
53+
- ./rabbitmq/enabled_plugins:/etc/rabbitmq/enabled_plugins
54+
healthcheck:
55+
test: ["CMD", "rabbitmq-diagnostics", "-q", "ping"]
56+
interval: 10s
57+
timeout: 5s
58+
retries: 5
5059

5160
webspicy:
5261
build: ./webspicy

‎example/rabbitmq/enabled_plugins‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1 @@
1+
[rabbitmq_management,rabbitmq_shovel,rabbitmq_shovel_management].

‎example/webspicy/arnavon/dlq/_queueName/requeue/post.yml‎

Lines changed: 83 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,83 @@
1+
---
2+
name: |-
3+
Requeue messages from DLQ
4+
5+
url: |-
6+
/dlq/{queueName}/requeue
7+
8+
services:
9+
- method: |-
10+
POST
11+
12+
description: |-
13+
Move messages from a dead letter queue to a destination queue
14+
using the RabbitMQ Management API.
15+
16+
preconditions:
17+
- The RabbitMQ Management plugin must be enabled
18+
- The source queue must exist
19+
- The destination queue must be specified
20+
21+
postconditions:
22+
- Messages are moved from source to destination queue
23+
- The response contains the count of requeued messages
24+
25+
input_schema: |-
26+
Dlq.RequeueRequest
27+
28+
output_schema: |-
29+
Dlq.RequeueResult
30+
31+
error_schema: |-
32+
Dlq.Error
33+
34+
examples:
35+
36+
- description: |-
37+
FOR: Messages are moved from source to destination queue
38+
params:
39+
queueName: dead-letters
40+
destinationQueue: send-email
41+
expected:
42+
content_type: application/json; charset=utf-8
43+
status: 200
44+
assert:
45+
- 'pathFD("", status: "initiated")'
46+
- 'pathFD("", requeued: 0)'
47+
- 'pathFD("", failed: 0)'
48+
- 'pathFD("", errors: [])'
49+
50+
- description: |-
51+
FOR: A specific count of messages can be moved
52+
params:
53+
queueName: dead-letters
54+
destinationQueue: send-email
55+
count: 10
56+
expected:
57+
content_type: application/json; charset=utf-8
58+
status: 200
59+
60+
counterexamples:
61+
62+
- description: |-
63+
FOR: The destination queue must be specified
64+
dress_params: false
65+
params:
66+
queueName: dead-letters
67+
expected:
68+
content_type: application/json; charset=utf-8
69+
status: 400
70+
assert:
71+
- match /destinationQueue is required/
72+
73+
- description: |-
74+
FOR: The destination queue must be a string
75+
dress_params: false
76+
params:
77+
queueName: dead-letters
78+
destinationQueue: 123
79+
expected:
80+
content_type: application/json; charset=utf-8
81+
status: 400
82+
assert:
83+
- match /destinationQueue is required/

‎example/webspicy/schema.fio‎

Lines changed: 25 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -35,6 +35,31 @@ Job.Error = {
3535
invalids :? [{...}]
3636
}
3737

38+
###
39+
### DLQ Requeue
40+
###
41+
42+
Dlq.QueueName = String(s | s.length > 0)
43+
44+
Dlq.RequeueRequest = {
45+
queueName : String
46+
destinationQueue : Dlq.QueueName
47+
count :? Numeric
48+
}
49+
50+
Dlq.RequeueStatus = String
51+
52+
Dlq.RequeueResult = {
53+
status : Dlq.RequeueStatus
54+
requeued : Numeric
55+
failed : Numeric
56+
errors : [{ error: String }]
57+
}
58+
59+
Dlq.Error = {
60+
error : String
61+
}
62+
3863
###
3964
### Example
4065
###

0 commit comments

Comments
 (0)