Skip to content

CONTRIBUTING.md

Latest commit

 

History

History
222 lines (162 loc) · 10.4 KB

CONTRIBUTING.md

File metadata and controls

222 lines (162 loc) · 10.4 KB
Raw

Contributing

You are interested in contributing to LavinMQ, but you are not sure how to? Right now, you can contribute to LavinMQ in one of two ways:

  1. Code contributions
  2. Non-code contributions

Code contributions

The first step to making a code contribution, is starting a conversation around what you'd like to work on. You can start a conversation around an existing issue or a new one that you've created - refer to reporting an issue. Next:

  1. Fork, create feature branch
  2. Submit pull request

Pull Request Workflow

We use GitHub's draft and open status to manage pull requests:

  • Draft: For code under construction. Use this when you're still working on your changes and they're not ready for review yet.
  • Open: For code that is perceived to be ready for review. Change your PR to this status when you've finished your implementation and want feedback from reviewers.

If you receive a review that requests changes, switch your PR back to draft mode while you're working on addressing the feedback. Once you've made the requested changes, set it back to open. This workflow helps us clearly see what is ready to be merged and what still needs work.

Develop

  1. Run specs with make test
  2. Run a specific spec file with make test SPEC=spec/storage_spec.cr
  3. Build locally with make bin/lavinmq CRYSTAL_FLAGS=
  4. Compile and run locally with crystal run src/lavinmq.cr
  5. Pull js dependencies with make js
  6. Build API docs with make docs (requires npx)
  7. Build with shards build

AI-assisted review

PRs are reviewed by AI in CI. To run the same review locally, install local-review and run local-review install in the repo. This adds an optional pre-push hook and a CLI (local-review, local-review 42).

Frontend development

Normally ECR files and static resources are baked into the binary, which is great for portability but requires recompiling the entire app for any view or JavaScript change. To streamline development, you can compile LavinMQ without release mode, compile the ECR files separately, and serve static files from the file system:

  1. Build the debug binary: make bin/lavinmq CRYSTAL_FLAGS=
  2. Watch and recompile views on change: make watch-views
  3. Run lavinmq in another terminal: ./bin/lavinmq -D /tmp/amqp
  4. Refresh the browser with Ctrl-R after making changes

Release

  1. Update CHANGELOG.md
  2. Bump version in shards.yml
  3. Create and push an annotated tag (git tag -a v$(shards version)), put the changelog of the version in the tagging message

Branching strategy

Our project's core development happens on the main branch, where new feature branches are merged as soon as they are complete. To prepare a new minor version, we create a long-lived release branch from main (e.g., 2.4.x). Bug fixes are applied to the relevant release branch and must also be merged back into main to be included in future releases.

%%{ init: { 'theme': 'base', 'gitGraph': {'mainBranchOrder':2 }} }%%
gitGraph
    commit
    branch feature-1 order: 1
    checkout feature-1
    commit
    checkout main
    commit id:"v2.4.0-rc.1" tag:"v2.4.0-rc.1"
    commit id:"rc-fix"
    commit id:"v2.4.0" tag:"v2.4.0"
    branch 2.4.x order: 3
    checkout main
    commit id:"bugfix-1"
    checkout 2.4.x
    cherry-pick id:"bugfix-1" tag:"v2.4.1"
    checkout feature-1
    commit
    checkout main
    branch feature-2 order: 0
    checkout feature-2
    commit
    checkout main
    merge feature-1
    commit id:"bugfix-2"
    checkout 2.4.x
    cherry-pick id:"bugfix-2" tag:"v2.4.2"
    checkout main
    commit id:"v2.5.0-rc.1" tag:"v2.5.0-rc.1"
    commit id:"v2.5.0" tag:"v2.5.0"
    branch 2.5.x order: 4
    checkout feature-2
    commit
    checkout main
    merge feature-2
    commit id:"bugfix-3"
    checkout 2.5.x
    cherry-pick id:"bugfix-3" tag:"v2.5.1"
    checkout main
    commit
Loading

Non-Code contributions

Your schedule won't allow you make code contributions? Still fine, you can:

Report an issue

  • This could be an easily reproducible bug or even a feature request.
  • If you spot an unexpected behaviour but you are not yet sure what the underlying bug is, the best place to post is LavinMQ's community Slack. This would allow us to interactively figure out what is going on.

Give us some feedback

We are also curious and happy to hear about your experience with LavinMQ. You can email us via contact@cloudamqp.com or reach us on Slack. Not sure what to write to us?

  • You can write to us about your first impression of LavinMQ
  • You can talk to us about features that are most important to you or your organization
  • You can also just tell us what we can do better

LavinMQ Implementation Details

LavinMQ is written in Crystal, a modern language built on the LLVM, with a Ruby-like syntax. It uses an event loop library for IO, is garbage collected, adopts a CSP-like concurrency model and compiles down to a single binary. You can liken it to Go, but with a nicer syntax.

Storage Architecture

Instead of trying to cache messages in RAM, we write all messages as fast as we can to disk and let the OS cache do the caching.

Each queue is backed by a message store on disk, which is just a series of files (segments), by default 8MB each. Message segments are memory-mapped files allocated using the mmap syscall. Each incoming message is appended to the last segment, prefixed with a timestamp, its exchange name, routing key and message headers.

Message Processing

When a message is being consumed it reads sequentially from the segments. Each acknowledged (or rejected) message position in the segment is written to an "ack" file (per segment). If a message is requeued its position is added to a in memory queue.

On boot all acked message positions are read from the "ack" files and then when delivering messages skip those when reading sequentially from the message segments. Segments are deleted when all message in them are acknowledged.

Definitions

Declarations of queues, exchanges and bindings are written to a definitions file (if the target is durable), encoded as the AMQP frame they came in as. Periodically this file is compacted/garbage-collected by writing only the current in-memory state to the file (getting rid of all delete events). This file is read on boot to restore all definitions.

All non-AMQP objects like users, vhosts, policies, etc. are stored in JSON files. Most often these types of objects does not have a high turnover rate, so we believe that JSON in this case makes it easy for operators to modify things when the server is not running, if ever needed.

In the data directory we store users.json and vhosts.json as mentioned earlier, and each vhost has a directory in which we store definitions.amqp (encoded as AMQP frames), policies.json and the messages named such as msgs.0000000124. Each vhost directory is named after the sha1 hash of its real name. The same goes for the queue directories in the vhost directory. The queue directories only has two files, ack and enq, also described earlier.

Transient queues are stored in a separate transient folder in the vhost directory, and it is removed any time LavinMQ shuts down or starts up.

Flows

Here is an architectural description of the different flows in the server.

Publish

Client#read_loop reads from the socket, it calls Channel#start_publish for the Basic.Publish frame and Channel#add_content for Body frames. When all content has been received (and appended to an IO::Memory object) it calls VHost#publish with a Message struct. VHost#publish finds all matching queues, writes the message to the message store and then calls Queue#publish with the segment position. Queue#publish writes to the message store.

Consume

When Client#read_loop receives a Basic.Consume frame it will create a Consumer class and add it to the queue's list of consumers. Each consumer has a deliver_loop fiber that will be notified by an internal Channel when new messages are available in the queue.

Stream Queues

Stream queues provide an append-only log structure that allows multiple consumers to read the same messages independently. Unlike standard queues, messages in stream queues aren't deleted when consumed, making them ideal for event sourcing patterns and multi-consumer scenarios.

Each consumer can start reading from anywhere in the queue using the x-stream-offset consumer argument and can process messages at their own pace.

Stream Queue Filtering

Stream queues support message filtering, allowing consumers to receive only messages that match specific criteria. This is useful for consuming a subset of messages without creating multiple queues.

How Filtering Works

  • Consumers can filter messages based on message headers
  • Two matching strategies are supported:
    • "all" (default): A message must match ALL filter conditions to be delivered
    • "any": A message matching ANY filter condition will be delivered
  • Messages without filter values won't be delivered to consumers with filters unless x-stream-match-unfiltered is set to true
  • Consumers without any filter conditions will receive all messages

Publishing Filtered Messages

When publishing messages that should be filtered, include header values that can be matched against:

  • The x-stream-filter-value header (a comma-separated string of values)
  • Any custom headers for more advanced filtering scenarios

Consuming Filtered Messages

To enable filtering on the consumer, set the following arguments when creating a consumer:

  • x-stream-filter: Supports multiple formats:
    • String: A comma-separated list of values to match against x-stream-filter-value headers
    • Table: Key-value pairs where both the header name and value must match
    • Array: A collection of filter conditions in either string or table format
  • x-filter-match-type: Determines the matching strategy:
    • "all" (default): All filter conditions must match
    • "any": At least one filter condition must match
  • x-stream-match-unfiltered: Boolean (default: false) that determines whether to receive messages without filter values