From 7269e4e72ed523788f5f677b708769a4a53c5db4 Mon Sep 17 00:00:00 2001 From: James Ainslie <42301770+jamesainslie@users.noreply.github.com> Date: Mon, 30 Mar 2026 06:40:08 -0400 Subject: [PATCH] release: prepare v0.2.0 changelog and add release:publish stave target MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Reorganize [Unreleased] into grouped v0.2.0 section with categories: AMQP hardening, advanced queues, management API, TLS/security, MQTT, clustering, shovels/federation, performance, testing/CI - Add Fixed section for bugs resolved during development - Add Release.Publish stave target: svu next → git tag → goreleaser --- CHANGELOG.md | 157 ++++++++++++++++++++++++++++++--------------------- stavefile.go | 18 ++++++ 2 files changed, 110 insertions(+), 65 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 54e38d8..a8a6aff 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,76 +7,98 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +## [0.2.0] - 2026-03-30 + ### Added -- Shovel engine for cross-broker message forwarding with AMQP source, AMQP destination, and HTTP destination -- Shovel `Source` and `Destination` interfaces for extensible endpoint types -- Three ack modes: `on-confirm` (default), `on-publish`, and `no-ack` -- Exponential backoff reconnect on shovel connection failure -- Federation links implemented as specialised shovels with `x-federation-hops` hop tracking -- Loop prevention: messages at or above `MaxHops` are silently dropped -- Shovel store with JSON persistence (`shovels.json`, `federation.json`) surviving broker restarts -- HTTP management API for shovels: `GET /api/shovels`, `PUT/DELETE /api/shovels/{vhost}/{name}` -- HTTP management API for federation: `GET /api/federation-links`, `PUT/DELETE /api/federation-links/{vhost}/{name}` -- URI scrubbing for federation link status display (credentials replaced with `***`) -- Leader-follower clustering with file-level replication over TCP -- `cluster` package: `FileIndex` (CRC32 checksums), `Action` encode/decode, `ReplicationServer`, `Follower`, `Election` -- Replication protocol with `GOMQR` header, password authentication, and action streaming -- Single-leader mode: leader broadcasts append/replace/delete actions to connected followers -- Full sync support via `FileIndex.Diff` for detecting missing or changed files -- `[clustering]` INI config section with `enabled`, `bind`, `port`, `password`, and `leader_uri` keys -- `--cluster`, `--cluster-bind`, `--cluster-port`, `--cluster-password`, `--cluster-leader` CLI flags -- `WithCluster`, `WithClusterBind`, `WithClusterPort`, `WithClusterPassword`, `WithClusterLeaderURI` embeddable API options -- MQTT 3.1.1 protocol support with full packet codec (all 14 control packet types) -- MQTT broker bridging to AMQP via `amq.topic` exchange with topic conversion (`/` to `.`, `+` to `*`) -- MQTT QoS 0, QoS 1 (PUBACK), and QoS 2 (PUBREC/PUBREL/PUBCOMP) message delivery -- MQTT retained messages with delivery to new subscribers -- MQTT Last Will and Testament (LWT) on ungraceful disconnect -- MQTT clean and persistent sessions with per-client AMQP queues (`mqtt.`) -- MQTT wildcard subscription matching (`+` single-level, `#` multi-level) -- MQTT keep-alive with PINGREQ/PINGRESP -- MQTT authentication via the AMQP user store (username/password) -- `WithMQTTPort` and `WithMQTTBind` options for the embeddable broker API -- `--mqtt-port` and `--mqtt-bind` CLI flags (default 127.0.0.1:1883, -1 to disable) -- `[mqtt]` INI config section with `bind` and `port` keys -- Cross-protocol integration: AMQP publish delivered to MQTT subscribers and vice versa -- TLS/AMQPS support with configurable cert/key and `--tls-cert`, `--tls-key`, `--amqps-port` flags -- `WithTLS` and `WithAMQPSPort` options for the embeddable broker API -- Policy system with regex pattern matching, priority-based resolution, and automatic reapplication -- HTTP API for policies: `GET/PUT/DELETE /api/policies/{vhost}/{name}` -- Per-vhost connection and queue limits (`max-connections`, `max-queues`) -- HTTP API for vhost limits: `GET/PUT/DELETE /api/vhost-limits/{vhost}` -- INI config file support (`--config` flag) with LavinMQ-compatible `[main]`, `[amqp]`, `[mgmt]` sections -- Priority queue support via `x-max-priority` queue argument (1-255 levels, per-priority stores, highest-first delivery) -- Stream queue type via `x-queue-type=stream` (append-only log, per-consumer offsets, ack is no-op, supports first/last/next/numeric offset specifiers) -- Consistent-hash exchange type (`x-consistent-hash`) with CRC32 hash ring and weighted virtual nodes -- Message priority field (`Priority uint8`) in storage binary format and AMQP property mapping -- HTTP management API (`/api/`) with RabbitMQ-compatible endpoints -- Endpoints for overview, connections, channels, exchanges, queues, bindings, users, vhosts, permissions, and definitions -- Prometheus metrics at `/api/metrics` and `/metrics` (queue depths, message rates, connection counts) -- Health checks at `/api/healthchecks/node` and `/api/aliveness-test/{vhost}` -- Definition export/import via `GET/POST /api/definitions` -- HTTP Basic Auth against the user store on all management endpoints -- `--http-port` and `--http-bind` CLI flags for the management API (default 127.0.0.1:15672) -- `WithHTTPPort` and `WithHTTPBind` options for the embeddable broker API -- Broker accessor methods for connections, vhosts, exchanges, queues, channels, and bindings +#### AMQP Protocol Hardening + +- Nack/reject with requeue support (messages redelivered with Redelivered flag) +- Dead letter exchange support (`x-dead-letter-exchange`, `x-dead-letter-routing-key`) +- `x-death` header on dead-lettered messages with queue, reason, and routing keys +- Queue-level message TTL enforcement (`x-message-ttl`) +- Per-message TTL via AMQP `expiration` property +- Queue expiry timer (`x-expires`) marks unused queues for deletion - Consumer priority via `x-priority` argument on `basic.consume` - Single active consumer mode via `x-single-active-consumer` queue argument - `basic.return` for mandatory messages with no route (reply code 312) - `connection.blocked`/`connection.unblocked` flow control with disk space monitoring - Delivery limits via `x-delivery-limit` queue argument with dead-letter on exceed +- Headers and expiration persistence in message store binary format + +#### Advanced Queue Types + +- Priority queue support via `x-max-priority` queue argument (1-255 levels, per-priority stores, highest-first delivery) +- Stream queue type via `x-queue-type=stream` (append-only log, per-consumer offsets, supports first/last/next/numeric offset specifiers) +- Consistent-hash exchange type (`x-consistent-hash`) with CRC32 hash ring and weighted virtual nodes + +#### HTTP Management API + +- RabbitMQ-compatible REST API on port 15672 with HTTP Basic Auth +- Endpoints for overview, connections, channels, exchanges, queues, bindings, users, vhosts, permissions, and definitions +- Prometheus metrics at `/metrics` (queue depths, message rates, connection counts) +- Health checks at `/api/healthchecks/node` and `/api/aliveness-test/{vhost}` +- Definition export/import via `GET/POST /api/definitions` +- Policy management via `/api/policies/{vhost}/{name}` +- Vhost limits via `/api/vhost-limits/{vhost}` +- Shovel management via `/api/shovels/{vhost}/{name}` +- Federation link management via `/api/federation-links/{vhost}/{name}` + +#### TLS, Security & Configuration + +- TLS/AMQPS support with `--tls-cert`, `--tls-key`, `--amqps-port` flags +- Policy system with regex pattern matching, priority-based resolution, and automatic reapplication +- Per-vhost connection and queue limits (`max-connections`, `max-queues`) +- INI config file support (`--config` flag) with `[main]`, `[amqp]`, `[mgmt]`, `[mqtt]`, `[clustering]` sections + +#### MQTT 3.1.1 + +- Full MQTT 3.1.1 packet codec (all 14 control packet types) +- MQTT broker bridging to AMQP via `amq.topic` exchange with topic conversion (`/` → `.`, `+` → `*`, `#` → `#`) +- QoS 0 (at-most-once), QoS 1 (at-least-once), QoS 2 (exactly-once) delivery +- Retained messages, Last Will and Testament, clean/persistent sessions +- Per-client AMQP queues (`mqtt.`), wildcard subscriptions, keep-alive +- Cross-protocol: AMQP publish delivered to MQTT subscribers and vice versa + +#### Clustering + +- Leader-follower replication with file-level sync over TCP +- CRC32 file checksums with directory scanning and diff computation +- Replication protocol (`GOMQR` header) with password authentication and action streaming +- Full sync via `FileIndex.Diff` for detecting missing or changed files + +#### Shovels & Federation + +- Shovel engine for cross-broker message forwarding (AMQP source, AMQP + HTTP destinations) +- Three ack modes: `on-confirm`, `on-publish`, `no-ack` with exponential backoff reconnect +- Federation links as specialized shovels with `x-federation-hops` hop tracking and loop prevention +- Shovel store with JSON persistence surviving broker restarts + +#### Performance + +- Zero-copy consumer delivery from mmap to socket via `ShiftFunc`/`ReadFunc` +- Batched network writes (method + header + body in single flush) +- Buffer pooling for message serialization, publish body accumulation, and frame encoding +- Event-driven consumer prefetch signaling (replaced 5ms polling) +- Method object caching in AMQP frame reader +- TCP_CORK for write batching on Linux +- 64KB buffered I/O for reader and writer + +#### Testing & CI + +- Correctness guarantee tests (ordering, confirms, durability, ack, requeue, exclusive, overflow) +- Performance regression gates (throughput and latency thresholds) - CI pipeline with GitHub Actions (lint, test, build) - Changelog enforcement on pull requests -- Release automation with goreleaser -- Docker image published to ghcr.io -- Cross-platform binaries (linux/darwin × amd64/arm64) -- Nack/reject with requeue support (messages redelivered with Redelivered flag) -- Dead letter exchange support (x-dead-letter-exchange, x-dead-letter-routing-key) -- x-death header on dead-lettered messages with queue, reason, and routing keys -- Queue-level message TTL enforcement (x-message-ttl) -- Per-message TTL via AMQP expiration property -- Queue expiry timer (x-expires) marks unused queues for deletion -- Headers and expiration persistence in message store binary format +- Release automation with goreleaser and Docker + +### Fixed + +- Queue lock deadlock: delivery callback no longer holds `q.mu` during socket writes +- Crash recovery: `loadStats` resizes segments to last valid position after scanning +- Frame size validation prevents DoS via oversized frame headers +- `Len()` includes requeued and inbox-buffered messages +- AMQP wire type mismatch: `x-max-length` and other integer arguments now handled correctly from all client libraries ## [0.1.0] - 2026-03-29 @@ -85,9 +107,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - AMQP 0-9-1 broker with TCP server, connection, channel, consumer lifecycle - Vhost with exchange/queue management and routing (direct, fanout, topic, headers) - Queue with message store, configurable limits, and consumer signaling -- Embeddable API, integration tests, and standalone binary -- gomqperf performance testing tool -- Hot-path optimizations with correctness guarantees preserved +- Memory-mapped segment storage engine with crash recovery +- AMQP 0-9-1 frame codec with all method types +- User authentication with SHA256 passwords and per-vhost permissions +- Publisher confirms and transaction support +- Embeddable Go library API with functional options +- Standalone `gomq` binary with signal handling +- `gomqperf` performance testing tool -[Unreleased]: https://github.com/jamesainslie/gomq/compare/v0.1.0...HEAD +[Unreleased]: https://github.com/jamesainslie/gomq/compare/v0.2.0...HEAD +[0.2.0]: https://github.com/jamesainslie/gomq/compare/v0.1.0...v0.2.0 [0.1.0]: https://github.com/jamesainslie/gomq/releases/tag/v0.1.0 diff --git a/stavefile.go b/stavefile.go index 0c2f307..16945ea 100644 --- a/stavefile.go +++ b/stavefile.go @@ -145,6 +145,24 @@ func (Test) Gate() error { ) } +// Release groups release targets. +type Release st.Namespace + +// Publish tags the release and runs goreleaser. +func (Release) Publish() error { + version, err := sh.Output("svu", "next") + if err != nil { + return fmt.Errorf("determine next version: %w", err) + } + if err := sh.RunV("git", "tag", "-a", version, "-m", version); err != nil { + return fmt.Errorf("tag %s: %w", version, err) + } + if err := sh.RunV("git", "push", "origin", version); err != nil { + return fmt.Errorf("push tag %s: %w", version, err) + } + return sh.RunV("goreleaser", "release", "--clean") +} + // Fmt groups formatting targets. type Fmt st.Namespace