Skip to content

Commit 7f406c6

Browse files
committed
feat!: harden the runtime and finalize the 3.0 API surface
Full pre-release review sweep. Runtime fixes: - compression: self-encode wire content (ACM json mode corrupted every compressed publish into a JSON-serialized Buffer); retry republishes raw bytes - worker.close() drains in-flight handlers (drainTimeoutMs, default 30s) and the defensive nacks can no longer crash the process on a close race - per-consumer prefetch uses ACM's native consume option (no more QoS bleed across consumers on reconnect) - classic-queue immediate-requeue retries republish to the failing queue via the default exchange, not the original exchange - connection pool: idempotent leases, close/create race fixed, rejecting close no longer poisons the key; AmqpClient.close() memoized - telemetry helpers and the interceptor chain never throw into the data path; create() routes constructor throws to the defect channel - handlers with non-callable entries fail fast at create() Breaking API changes (3.0 window): - defineHandler(s)/defineMiddleware -> declareHandler(s)/declareMiddleware - defineRpc errors entries are { data, message? } - defineContract accepts standalone exchanges/queues/bindings - worker middleware accepts arrays; ConsumerOptions curated (noAck gone) - RPC handler responses typed as schema input; testing wait options are { count, timeoutMs }; asyncapi failOnMissingConverter defaults true; invalid connectTimeoutMs is a defect; deprecated _*ForTesting aliases removed; export-hygiene pass across all six packages New guarding tests for every fix; tests/ package now typechecks in CI.
1 parent 6d510d7 commit 7f406c6

61 files changed

Lines changed: 2273 additions & 776 deletions

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

.changeset/pre-3-0-api-review.md

Lines changed: 23 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,23 @@
1+
---
2+
"@amqp-contract/contract": major
3+
"@amqp-contract/core": major
4+
"@amqp-contract/client": major
5+
"@amqp-contract/worker": major
6+
"@amqp-contract/testing": major
7+
"@amqp-contract/asyncapi": major
8+
---
9+
10+
Pre-3.0 API review: breaking changes taken while the major window is open.
11+
12+
- **`defineHandler` / `defineHandlers` / `defineMiddleware``declareHandler` / `declareHandlers` / `declareMiddleware`.** The family convention (shared with temporal-contract) is `define*` for contract authoring and `declare*` for implementation-side APIs, making the contract boundary visible in the name.
13+
- **`defineRpc` error map entries are now `{ data: schema, message? }`** instead of `defineMessage(schema)`. `data` is the raw Standard Schema for the error payload; the optional `message` is a default human-readable message used when the handler constructs the error without one (and as the client-side fallback when a reply carries none).
14+
- **`defineContract` accepts standalone topology**: top-level `exchanges`, `queues`, and `bindings` entries declare resources with no publisher/consumer attached (a DLQ bound to the auto-extracted DLX, audit queues). Dead-letter exchanges and TTL-backoff infrastructure are auto-extracted for standalone queues exactly as for consumer queues.
15+
- **Worker `middleware` accepts an array** (first = outermost), composed like `composeMiddleware(...)`. Pre-compose with `composeMiddleware` when you want stepwise context types inferred.
16+
- **`worker.close()` now drains in-flight handlers** before closing the channel (their acks land; completed work is not redelivered), bounded by a new `drainTimeoutMs` option (default `DEFAULT_DRAIN_TIMEOUT_MS` = 30 s; `null` waits forever).
17+
- **Worker `ConsumerOptions` is now a curated subset** (`prefetch`, `priority`, `arguments`, `consumerTag`, `exclusive`). `noAck` is gone — it silently broke the ack-exactly-once and retry/DLQ invariants.
18+
- **RPC handler response types use the schema's _input_** (defaults optional, transforms not yet applied) — the worker validates before replying, matching the existing RPC error-data convention.
19+
- **Invalid `connectTimeoutMs`** (NaN, 0, negative, Infinity) now surfaces as a Defect from `create()` instead of silently disabling the timeout. Pass `null` to disable.
20+
- **Testing fixture wait options renamed**: `{ nbEvents, timeout }``{ count, timeoutMs }`, aligning with the library-wide `timeoutMs` convention. The fixture record is exported as `AmqpTestFixtures`, the wait options as `WaitForMessagesOptions`; the package now exposes `./package.json`.
21+
- **`@amqp-contract/asyncapi`: `failOnMissingConverter` defaults to `true`** — a spec that silently degrades schemas to `{ type: "object" }` placeholders lies to its consumers. It generates AsyncAPI 3.1 (docs previously claimed 3.0).
22+
- **Deprecated `_*ForTesting` aliases removed** from `@amqp-contract/core` (`_resetConnectionsForTesting`, `_getConnectionCountForTesting`, `_resetTelemetryCacheForTesting`) — use the `_internal_*` forms.
23+
- **Export hygiene**: `RetryOptions`, `NoneRetryOptions`, `DefineQueueOptionsWithDeadLetterExchange`, `QueueEntryWithDeadLetterExchange`, `RpcErrorDefinition`, `InferSchemaInput`, `InferSchemaOutput` (contract); `PublishError`, `CallError`, `ClientInferCallError` (client); `AnyWorkerMiddleware`, `DEFAULT_DRAIN_TIMEOUT_MS` (worker); `isTechnicalError`, `isMessageValidationError` guards (core). Client and worker re-export `Logger`, `LoggerContext`, `TelemetryProvider`, and `TechnicalError` so naming an option type never forces a direct dependency on core. `defineCommandPublisher` now preserves literal routing-key types. All packages declare `sideEffects: false`.
Lines changed: 16 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,16 @@
1+
---
2+
"@amqp-contract/contract": patch
3+
"@amqp-contract/core": patch
4+
"@amqp-contract/client": patch
5+
"@amqp-contract/worker": patch
6+
---
7+
8+
Robustness fixes from the pre-3.0 full review.
9+
10+
- **Compression was broken end-to-end**: the channel's JSON mode serialized the compressed Buffer into `'{"type":"Buffer","data":[...]}'` on the wire, so every compressed publish was DLQ'd on arrival. The client now encodes content itself (JSON for plain values, byte-for-byte for Buffers) and JSON mode is off. Retry republishing now passes the original bytes through untouched for all content types.
11+
- **Per-consumer `prefetch` no longer bleeds across consumers**: it now maps to amqp-connection-manager's native per-consumer prefetch, applied immediately before each `basic.consume` and re-applied per-consumer on reconnect (previously, any reconnect applied one consumer's QoS to all).
12+
- **Classic-queue immediate-requeue retries republish to the failing queue via the default exchange** instead of the original exchange — a fanout/topic topology no longer delivers retry duplicates (and foreign retry headers) to sibling queues.
13+
- **Closing a worker under load can no longer crash the process**: the defensive nack in the consume boundary is guarded (a nack racing channel teardown was an unhandled rejection), and `close()` drains in-flight handlers first.
14+
- **Connection pool races fixed**: releases are idempotent per-lease (a double `close()` can no longer close a shared connection under another live client), the last release removes the pool entry _before_ closing (an acquire racing it gets a fresh connection instead of a dead one), and a rejecting close no longer poisons the pool key. `AmqpClient.close()` is memoized (idempotent).
15+
- **Sync-throw escape hatches closed on the client**: a synchronously-throwing publish/call interceptor now resolves to a Defect instead of escaping `publish()`/`call()` as a raw throw, and all telemetry helpers swallow provider/span throws (a buggy TelemetryProvider degrades to "no telemetry" instead of poisoning the data path or converting a successful publish into a Defect). `create()` routes synchronous constructor throws to the defect channel.
16+
- **`handlers: { name: undefined }` now fails fast** at `create()` / `declareHandlers` with a clear error instead of defecting on every delivery.

examples/basic-order-processing-client/src/index.ts

Lines changed: 25 additions & 27 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
import { orderContract } from "@amqp-contract-examples/basic-order-processing-contract";
2-
import { type PublishOptions, TypedAmqpClient } from "@amqp-contract/client";
2+
import { type PublishInterceptor, TypedAmqpClient } from "@amqp-contract/client";
33
import pino from "pino";
44
import { z } from "zod";
55

@@ -20,11 +20,27 @@ const logger = pino({
2020
},
2121
});
2222

23+
// Publish interceptor: logs every publish (before and after) in one place
24+
// instead of wrapping each call site. The first interceptor in the array is
25+
// the outermost.
26+
const logPublishes: PublishInterceptor = (args, next) => {
27+
logger.debug(`Publishing to ${args.publisherName}`);
28+
return next()
29+
.tap(() => logger.debug(`Successfully published to ${args.publisherName}`))
30+
.tapFailure((failure) =>
31+
logger.error(
32+
{ error: failure.tag === "Err" ? failure.error : failure.cause },
33+
`Failed to publish: ${args.publisherName}`,
34+
),
35+
);
36+
};
37+
2338
async function main() {
2439
// Create type-safe client
2540
const client = await TypedAmqpClient.create({
2641
contract: orderContract,
2742
urls: [env.AMQP_URL],
43+
publishInterceptors: [logPublishes],
2844
})
2945
.tapDefect((cause) => logger.error({ error: cause }, "Failed to create client"))
3046
.get();
@@ -34,26 +50,6 @@ async function main() {
3450
logger.info("Publishing orders to demonstrate RabbitMQ topic pattern");
3551
logger.info("=".repeat(60));
3652

37-
// Helper function to publish and handle errors
38-
// In production code, you might want to return the Result to the caller
39-
// instead of throwing, but for this demo we throw to simplify the flow
40-
const publishWithLog = async <T extends Parameters<typeof client.publish>[0]>(
41-
publisherName: T,
42-
message: Parameters<typeof client.publish<T>>[1],
43-
options?: PublishOptions,
44-
): Promise<void> => {
45-
await client
46-
.publish(publisherName, message, options)
47-
.tapFailure((failure) =>
48-
logger.error(
49-
{ error: failure.tag === "Err" ? failure.error : failure.cause },
50-
`Failed to publish: ${publisherName}`,
51-
),
52-
)
53-
.tap(() => logger.debug(`Successfully published to ${publisherName}`))
54-
.getOrThrow();
55-
};
56-
5753
// 1. Publish a new order (routing key: order.created)
5854
logger.info("1️⃣ Publishing NEW ORDER (order.created)");
5955
const newOrder = {
@@ -65,7 +61,9 @@ async function main() {
6561
],
6662
totalAmount: 109.97,
6763
};
68-
await publishWithLog("orderCreated", newOrder);
64+
// In production code, you might want to match on the Result instead of
65+
// throwing, but for this demo we throw to simplify the flow
66+
await client.publish("orderCreated", newOrder).getOrThrow();
6967
logger.info(` ✓ Published order ${newOrder.orderId}`);
7068
logger.info(` → Will be received by: processing & notifications queues`);
7169
await new Promise((resolve) => setTimeout(resolve, 1500));
@@ -76,7 +74,7 @@ async function main() {
7674
orderId: "ORD-001",
7775
status: "processing" as const,
7876
};
79-
await publishWithLog("orderUpdated", orderUpdate);
77+
await client.publish("orderUpdated", orderUpdate).getOrThrow();
8078
logger.info(` ✓ Published update for ${orderUpdate.orderId}`);
8179
logger.info(` → Will be received by: notifications queue only`);
8280
await new Promise((resolve) => setTimeout(resolve, 1500));
@@ -87,7 +85,7 @@ async function main() {
8785
orderId: "ORD-001",
8886
status: "shipped" as const,
8987
};
90-
await publishWithLog("orderShipped", shippedOrder);
88+
await client.publish("orderShipped", shippedOrder).getOrThrow();
9189
logger.info(` ✓ Published shipment for ${shippedOrder.orderId}`);
9290
logger.info(` → Will be received by: notifications & shipping queues`);
9391
await new Promise((resolve) => setTimeout(resolve, 1500));
@@ -104,7 +102,7 @@ async function main() {
104102
eventSource: "new-order-service",
105103
eventVersion: 2,
106104
};
107-
await publishWithLog("orderCreated", newOrder2, { headers: newOrderHeaders2 });
105+
await client.publish("orderCreated", newOrder2, { headers: newOrderHeaders2 }).getOrThrow();
108106
logger.info(` ✓ Published order ${newOrder2.orderId}`);
109107
logger.info(` → Will be received by: processing & notifications queues`);
110108
await new Promise((resolve) => setTimeout(resolve, 1500));
@@ -115,7 +113,7 @@ async function main() {
115113
orderId: "ORD-002",
116114
status: "cancelled" as const,
117115
};
118-
await publishWithLog("orderUrgentUpdate", urgentUpdate);
116+
await client.publish("orderUrgentUpdate", urgentUpdate).getOrThrow();
119117
logger.info(` ✓ Published urgent update for ${urgentUpdate.orderId}`);
120118
logger.info(` → Will be received by: notifications & urgent queues`);
121119
await new Promise((resolve) => setTimeout(resolve, 1500));
@@ -129,7 +127,7 @@ async function main() {
129127
warehouseId: "WH-EU-1",
130128
priority: "express" as const,
131129
};
132-
await publishWithLog("requestFulfillment", fulfillment);
130+
await client.publish("requestFulfillment", fulfillment).getOrThrow();
133131
logger.info(` ✓ Sent fulfillment command for ${fulfillment.orderId}`);
134132
logger.info(` → Will be received by: the single fulfillment worker (task queue)`);
135133

examples/basic-order-processing-contract/README.md

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -21,7 +21,7 @@ import { TypedAmqpClient } from "@amqp-contract/client";
2121
const client = await TypedAmqpClient.create({
2222
contract,
2323
urls: ["amqp://localhost"],
24-
}).getOrThrow();
24+
}).get();
2525

2626
await client.publish("orderCreated", {/* fully typed */});
2727
```

examples/basic-order-processing-worker/README.md

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -40,12 +40,12 @@ const worker = await TypedAmqpWorker.create({
4040
},
4141
},
4242
urls: [env.AMQP_URL],
43-
}).getOrThrow();
43+
}).get();
4444
```
4545

4646
### External Handlers (src/handlers.ts)
4747

48-
Handlers can be organized in separate files using `defineHandler` or `defineHandlers`. The `src/handlers.ts` file demonstrates this pattern, which is recommended for:
48+
Handlers can be organized in separate files using `declareHandler` or `declareHandlers`. The `src/handlers.ts` file demonstrates this pattern, which is recommended for:
4949

5050
- Production applications
5151
- Better code organization and testability
@@ -54,7 +54,7 @@ Handlers can be organized in separate files using `defineHandler` or `defineHand
5454

5555
```typescript
5656
// handlers.ts
57-
export const processOrderHandler = defineHandler(orderContract, "processOrder", ({ payload }) => {
57+
export const processOrderHandler = declareHandler(orderContract, "processOrder", ({ payload }) => {
5858
// Handler logic here
5959
return OkAsync(undefined);
6060
});
@@ -69,7 +69,7 @@ const worker = await TypedAmqpWorker.create({
6969
// ... other handlers
7070
},
7171
urls: [env.AMQP_URL],
72-
}).getOrThrow();
72+
}).get();
7373
```
7474

7575
The main `src/index.ts` file uses inline handlers for simplicity, while `src/handlers.ts` provides an example of how to organize handlers externally for better maintainability.

examples/basic-order-processing-worker/src/index.spec.ts

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
import { orderContract } from "@amqp-contract-examples/basic-order-processing-contract";
22
import { it } from "@amqp-contract/testing/extension";
3-
import { TypedAmqpWorker, defineHandlers } from "@amqp-contract/worker";
3+
import { TypedAmqpWorker, declareHandlers } from "@amqp-contract/worker";
44
import { OkAsync } from "unthrown";
55
import { describe, expect, vi } from "vitest";
66

@@ -13,7 +13,7 @@ describe("Basic Order Processing Worker Integration", () => {
1313
const processedOrders: Array<unknown> = [];
1414
const worker = await TypedAmqpWorker.create({
1515
contract: orderContract,
16-
handlers: defineHandlers(orderContract, {
16+
handlers: declareHandlers(orderContract, {
1717
processOrder: ({ payload }) => {
1818
processedOrders.push(payload);
1919
return OkAsync(undefined);
@@ -64,7 +64,7 @@ describe("Basic Order Processing Worker Integration", () => {
6464
const notifications: Array<unknown> = [];
6565
const workerResult = await TypedAmqpWorker.create({
6666
contract: orderContract,
67-
handlers: defineHandlers(orderContract, {
67+
handlers: declareHandlers(orderContract, {
6868
processOrder: () => OkAsync(undefined),
6969
notifyOrder: ({ payload }) => {
7070
notifications.push(payload);
@@ -128,7 +128,7 @@ describe("Basic Order Processing Worker Integration", () => {
128128
const notifications: Array<unknown> = [];
129129
const workerResult = await TypedAmqpWorker.create({
130130
contract: orderContract,
131-
handlers: defineHandlers(orderContract, {
131+
handlers: declareHandlers(orderContract, {
132132
processOrder: ({ payload }) => {
133133
processedOrders.push(payload);
134134
return OkAsync(undefined);
@@ -184,7 +184,7 @@ describe("Basic Order Processing Worker Integration", () => {
184184
const fulfilled: Array<unknown> = [];
185185
const worker = await TypedAmqpWorker.create({
186186
contract: orderContract,
187-
handlers: defineHandlers(orderContract, {
187+
handlers: declareHandlers(orderContract, {
188188
processOrder: () => OkAsync(undefined),
189189
notifyOrder: () => OkAsync(undefined),
190190
shipOrder: () => OkAsync(undefined),

examples/basic-order-processing-worker/src/index.ts

Lines changed: 8 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
import { orderContract } from "@amqp-contract-examples/basic-order-processing-contract";
2-
import { RetryableError, TypedAmqpWorker, defineHandlers } from "@amqp-contract/worker";
2+
import { TypedAmqpWorker, declareHandlers, qualifyRetryable } from "@amqp-contract/worker";
33
import pino from "pino";
44
import { fromPromise } from "unthrown";
55
import { z } from "zod";
@@ -25,7 +25,7 @@ async function main() {
2525
// Create type-safe worker with handlers for each consumer
2626
const workerResult = TypedAmqpWorker.create({
2727
contract: orderContract,
28-
handlers: defineHandlers(orderContract, {
28+
handlers: declareHandlers(orderContract, {
2929
// Handler for processing NEW orders (order.created)
3030
processOrder: ({ payload, headers }) => {
3131
logger.info(
@@ -45,7 +45,7 @@ async function main() {
4545

4646
return fromPromise(
4747
new Promise<void>((resolve) => setTimeout(resolve, 500)),
48-
(e) => new RetryableError("Processing failed", e),
48+
qualifyRetryable("Processing failed"),
4949
).map(() => {
5050
logger.info({ orderId: payload.orderId }, "Order processed successfully");
5151
});
@@ -80,7 +80,7 @@ async function main() {
8080

8181
return fromPromise(
8282
new Promise<void>((resolve) => setTimeout(resolve, 300)),
83-
(e) => new RetryableError("Notification failed", e),
83+
qualifyRetryable("Notification failed"),
8484
).map(() => {
8585
logger.info("Notification sent");
8686
});
@@ -99,7 +99,7 @@ async function main() {
9999

100100
return fromPromise(
101101
new Promise<void>((resolve) => setTimeout(resolve, 400)),
102-
(e) => new RetryableError("Shipping failed", e),
102+
qualifyRetryable("Shipping failed"),
103103
).map(() => {
104104
logger.info({ orderId: payload.orderId }, "Shipping label prepared");
105105
});
@@ -118,7 +118,7 @@ async function main() {
118118

119119
return fromPromise(
120120
new Promise<void>((resolve) => setTimeout(resolve, 200)),
121-
(e) => new RetryableError("Urgent handling failed", e),
121+
qualifyRetryable("Urgent handling failed"),
122122
).map(() => {
123123
logger.warn({ orderId: payload.orderId }, "Urgent update handled");
124124
});
@@ -138,7 +138,7 @@ async function main() {
138138

139139
return fromPromise(
140140
new Promise<void>((resolve) => setTimeout(resolve, 400)),
141-
(e) => new RetryableError("Fulfillment failed", e),
141+
qualifyRetryable("Fulfillment failed"),
142142
).map(() => {
143143
logger.info({ orderId: payload.orderId }, "Order handed to the warehouse");
144144
});
@@ -158,7 +158,7 @@ async function main() {
158158

159159
return fromPromise(
160160
new Promise<void>((resolve) => setTimeout(resolve, 200)),
161-
(e) => new RetryableError("Failed order handling failed", e),
161+
qualifyRetryable("Failed order handling failed"),
162162
).map(() => {
163163
logger.error({ orderId: payload.orderId }, "Failed order logged for investigation");
164164
});

packages/asyncapi/README.md

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
# @amqp-contract/asyncapi
22

3-
**AsyncAPI 3.0.0 specification generator for amqp-contract.**
3+
**AsyncAPI 3.1.0 specification generator for amqp-contract.**
44

55
[![CI](https://github.com/btravstack/amqp-contract/actions/workflows/ci.yml/badge.svg)](https://github.com/btravstack/amqp-contract/actions/workflows/ci.yml)
66
[![npm version](https://img.shields.io/npm/v/@amqp-contract/asyncapi.svg?logo=npm)](https://www.npmjs.com/package/@amqp-contract/asyncapi)
@@ -60,7 +60,7 @@ writeFileSync("asyncapi.json", JSON.stringify(asyncAPISpec, null, 2));
6060

6161
## Features
6262

63-
-**AsyncAPI 3.0 compliant** with proper AMQP bindings (v0.3.0)
63+
-**AsyncAPI 3.1 compliant** with proper AMQP bindings (v0.3.0)
6464
-**Schema validation** - Converts Zod, Valibot, and ArkType schemas to JSON Schema
6565
-**Queue-exchange binding documentation** in channel descriptions
6666
-**Type-safe** with full TypeScript support

packages/asyncapi/package.json

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -32,6 +32,7 @@
3232
"docs"
3333
],
3434
"type": "module",
35+
"sideEffects": false,
3536
"main": "./dist/index.cjs",
3637
"module": "./dist/index.mjs",
3738
"types": "./dist/index.d.mts",

0 commit comments

Comments
 (0)