Skip to content

release: prepare v0.2.0 #21

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
jamesainslie merged 1 commit into from
Mar 30, 2026
Merged

release: prepare v0.2.0 #21

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
157 changes: 92 additions & 65 deletions CHANGELOG.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.<clientID>`)
- 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.<clientID>`), 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

Expand All @@ -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
18 changes: 18 additions & 0 deletions stavefile.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down
Loading