Add MysqlMessageQueue to @fedify/mysql#599
Conversation
Adds a MySQL/MariaDB-backed MessageQueue implementation for users who
want a production-grade message queue without adding Redis or
PostgreSQL to their stack.
Since MySQL and MariaDB lack LISTEN/NOTIFY, delivery relies on
periodic polling (SELECT … FOR UPDATE SKIP LOCKED). Ordering-key
serialization uses MySQL advisory locks (GET_LOCK/RELEASE_LOCK) on
a dedicated connection. Requires MySQL 8.0+ or MariaDB 10.6+.
Key details:
- Default poll interval: 1 second (configurable via pollInterval)
- Default handler timeout: 60 seconds (configurable via
handlerTimeout)
- enqueueMany() batch-inserts in a single transaction
- Advisory lock names longer than 64 chars are hashed to fit
MySQL's advisory lock name limit
Documentation updated in mq.md, federation.md, deploy.md, relay.md,
send.md, and inbox.md.
Closes fedify-dev#586
Co-Authored-By: Claude <noreply@anthropic.com>
Summary of ChangesHello, I'm Gemini Code Assist1! I'm currently reviewing this pull request and will post my feedback shortly. In the meantime, here's a summary to help you and other reviewers quickly get up to speed! This pull request significantly enhances Fedify's message queuing capabilities by integrating a MySQL/MariaDB-backed message queue. This new Highlights
Changelog
Activity
Using Gemini Code AssistThe full guide for Gemini Code Assist can be found on our documentation page, here are some quick tips. Invoking Gemini You can request assistance from Gemini at any point by creating a comment using either
Customization To customize Gemini Code Assist for GitHub experience, repository maintainers can create a configuration file and/or provide a custom code review style guide (such as PEP-8 for Python) by creating and adding files to a Limitations & Feedback Gemini Code Assist may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counter productive. You can react with 👍 and 👎 on @gemini-code-assist comments. If you're interested in giving your feedback about your experience with Gemini Code Assist for Github and other Google products, sign up here. You can also get AI-powered code generation, chat, as well as code reviews directly in the IDE at no cost with the Gemini Code Assist IDE Extension. Footnotes
|
dahlia
moved this to In progress
in STF 2025–2026 Milestone 4: Storage and Infrastructure
[ci skip]
dahlia
added
component/integration
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request introduces MysqlMessageQueue, a new MessageQueue implementation for MySQL and MariaDB. The implementation is robust, handling concurrency with SELECT ... FOR UPDATE SKIP LOCKED and advisory locks for ordering keys. It also includes features like delayed delivery and handler timeouts. The accompanying test suite is comprehensive, covering various scenarios including concurrent workers and error conditions. The documentation has been updated across multiple files to reflect this new addition.
My review found one opportunity for performance improvement in the enqueueMany method by batching inserts into a single query. This comment aligns with general best practices for database interaction and does not contradict any provided rules. Overall, this is a high-quality addition to the library.
There was a problem hiding this comment.
Pull request overview
Adds a MySQL/MariaDB-backed MessageQueue implementation to @fedify/mysql, enabling Fedify deployments that already depend on MySQL/MariaDB to use it for background message processing (with polling + advisory locks).
Changes:
- Implement
MysqlMessageQueuewith polling-basedlisten(), delayed delivery, ordering keys (viaGET_LOCK/RELEASE_LOCK), and transactional batch enqueue. - Add a comprehensive MySQL-backed test suite for the new queue implementation.
- Export/package/document
MysqlMessageQueueacross module entrypoints, package exports, docs, andCHANGES.md.
Reviewed changes
Copilot reviewed 14 out of 14 changed files in this pull request and generated 5 comments.
Show a summary per file
| File | Description |
|---|---|
| packages/mysql/tsdown.config.ts | Includes src/mq.ts in build entries and ensures Temporal polyfill intro applies to it. |
| packages/mysql/src/mq.ts | New MySQL/MariaDB message queue implementation (polling, delays, ordering keys via advisory locks). |
| packages/mysql/src/mq.test.ts | New test suite for MysqlMessageQueue behavior and concurrency scenarios. |
| packages/mysql/src/mod.ts | Re-exports MysqlMessageQueue from package root. |
| packages/mysql/package.json | Adds ./mq export mapping for npm consumers. |
| packages/mysql/deno.json | Adds ./mq export mapping for Deno/JSR consumers. |
| packages/mysql/README.md | Documents MysqlMessageQueue alongside MysqlKvStore and updates usage snippet/links. |
| docs/manual/send.md | Mentions MysqlMessageQueue as a production-ready queue option. |
| docs/manual/relay.md | Adds MysqlMessageQueue to production queue recommendations. |
| docs/manual/mq.md | Adds a new MysqlMessageQueue section (install + behavior + pros/cons + example) and updates feature tables. |
| docs/manual/inbox.md | Mentions MysqlMessageQueue as a production-ready queue option. |
| docs/manual/federation.md | Adds @fedify/mysql / MysqlMessageQueue to the list of MQ backends. |
| docs/manual/deploy.md | Updates production deployment recommendations to include MySQL/MariaDB KV+MQ pairing. |
| CHANGES.md | Adds changelog entry for the new MysqlMessageQueue feature. |
Deno type-checks @example blocks in JSDoc comments, and the createFederation() call was missing the required kv property, causing TS2345 in CI. Add MysqlKvStore alongside MysqlMessageQueue to make the example self-contained and type-correct. Co-Authored-By: Claude <noreply@anthropic.com>
Replace the per-message INSERT loop with a single INSERT … VALUES (…), (…), … statement, reducing the number of database round-trips from N to 1 regardless of batch size. fedify-dev#599 (comment) Co-Authored-By: Claude <noreply@anthropic.com>
MySQL's GET_LOCK() returns NULL when the server encounters an internal error (e.g. exhausted lock resources), as distinct from 0 (lock held by another session) and 1 (lock acquired). Previously both NULL and 0 fell through the same silent "try next ordering key" path, making it impossible to distinguish a transient server error from a normal contention event. Now when acquired is NULL a WARN log is emitted so operators can detect the condition, and the worker skips that ordering key and continues to the next candidate instead of silently dropping work. fedify-dev#599 (comment) Co-Authored-By: Claude <noreply@anthropic.com>
Without clearTimeout, the event loop would remain alive until the poll interval timer fired naturally, even after the AbortSignal was triggered. This caused listen() to not return promptly on abort, keeping resources alive longer than necessary. Fix: capture the timeoutId returned by setTimeout and call clearTimeout(timeoutId) in the abort handler so the inter-poll sleep resolves immediately. Also skip the trailing safeSerializedPoll() call when the signal is already aborted after the sleep resolves. Regression test: a queue configured with a 60-second poll interval must have its listen() promise resolve within 3 seconds of an abort signal, well within the 60-second timer it would have waited for previously. fedify-dev#599 (comment) Co-Authored-By: Claude <noreply@anthropic.com>
The #findOrderingKeyCandidate() query scans for the oldest ready message per ordering key. Without an index covering both columns, MySQL must do a full table scan for every poll cycle on tables with many ordering keys. Add idx_<tableName>_ok_da on (ordering_key, deliver_after) so that the query can use the index to find candidates efficiently. The index is created during initialize() alongside the existing deliver_after index, with the same ER_DUP_KEYNAME guard to handle concurrent initialization. Update the initialize() test to assert that both the single-column deliver_after index and the new composite index are present after initialization. Also tighten the existing deliver_after index check to match by index_name rather than column_name to avoid a spurious double-count now that deliver_after also appears in the composite index. fedify-dev#599 (comment) Co-Authored-By: Claude <noreply@anthropic.com>
…n test The previous "lock name is deterministic" test only validated that the MysqlMessageQueue constructor accepted certain table-name lengths; it never actually called GET_LOCK and so could not detect a regression in the name-hashing logic at runtime. Replace it with two focused tests: - A pure unit test (no DB) that checks the 46-char table-name limit by asserting doesNotThrow / throws on constructor calls. - A DB integration test that enqueues a message with a 200-character ordering key and waits for it to be processed. The combined raw lock name would be 221 chars — well over MySQL's 64-char cap — so if the hashing were removed or broken, GET_LOCK would return an error or NULL and the message would never be delivered. fedify-dev#599 (comment) Co-Authored-By: Claude <noreply@anthropic.com>
Bun's test runner does not honour the { skip } option, so any test that
uses { skip: dbUrl == null } but does not also guard its body with an
early return will attempt to execute even when MYSQL_URL is unset and
crash with a connection error.
Add 'if (dbUrl == null) return; // Bun does not support skip option'
as the first statement of every DB-dependent test body to ensure the
test exits cleanly on Bun when the database is unavailable, matching
the pattern already used in @fedify/postgres.
Also restore the enqueueMany 100-message bulk test and the listener-
survives-handler-errors test that were accidentally merged during an
earlier edit.
fedify-dev#599 (comment)
Co-Authored-By: Claude <noreply@anthropic.com>
|
/gemini review |
Codecov Report❌ Patch coverage is
... and 1 file with indirect coverage changes 🚀 New features to boost your workflow:
|
There was a problem hiding this comment.
Code Review
This pull request introduces a MysqlMessageQueue implementation for @fedify/mysql, providing a robust, polling-based message queue backed by MySQL or MariaDB. While the implementation is generally robust, a security concern was identified regarding the logging of sensitive information (PII) in debug logs during the message enqueuing process. Additionally, there is a suggestion to improve the robustness of the drop() method against potential race conditions.
The advisory lock name returned by getMysqlLockName() is "fdy:" (4 chars) + two 8-char hex values = 20 characters in total, not 21 as the comment previously stated. The "rejects table names longer than 46 chars" test added in an earlier commit duplicated assertions already present in the existing "rejects invalid table names" test (which already checks both the 47-char throw and the 46-char no-throw). Remove the redundant test to reduce maintenance overhead. fedify-dev#599 (comment) fedify-dev#599 (comment) Co-Authored-By: Claude <noreply@anthropic.com>
If drop() was called while initialize() was still in progress, drop() could complete first and clear #initialized / #initPromise, and then #doInitialize() would eventually set #initialized = true — leaving the instance believing the table exists even though it had already been dropped. Any subsequent enqueue() or listen() call would then hit a "Table doesn't exist" error. Fix: at the top of drop(), await the existing #initPromise (if any) and ignore any initialization error, so the CREATE TABLE and index creation are always fully committed to the database before the DROP TABLE runs. Also adds a regression test that fires initialize() and immediately calls drop(), then verifies the instance can be re-initialized and used normally afterward. fedify-dev#599 (comment) Co-Authored-By: Claude <noreply@anthropic.com>
MySQL evaluates NOW(6) once per statement, so all rows in a bulk INSERT share the same deliver_after timestamp. When messages share an orderingKey, dequeueing (which ORDER BY deliver_after) then selects among them non-deterministically. Fix by using delayMs * 1000 + index microseconds for each row's interval, so row i gets a deliver_after exactly i µs later than row 0. This ensures the insertion order is preserved under any orderingKey without requiring a new column or schema change. Add a regression test that enqueues five messages under the same ordering key via enqueueMany() and asserts they are delivered in order. fedify-dev#599 (comment) Co-Authored-By: claude-sonnet-4-6 <claude-sonnet-4-6@anthropic.com>
ordering_key is an arbitrary string (often a URL in Fedify), so the previous VARCHAR(512) limit could silently truncate or reject values longer than 512 characters at insert time with an opaque DB error. Switch the column to TEXT so there is no fixed-length cap. Because InnoDB requires a prefix length when indexing TEXT columns, the composite index now uses ordering_key(766): 766 chars * 4 bytes (utf8mb4) + 8 bytes (DATETIME(6)) = 3072 bytes which exactly hits InnoDB's 3072-byte key limit. Existing tables created with the old VARCHAR(512) definition will continue to work; operators who want to lift the 512-char cap must ALTER the column manually (or drop and recreate the table via drop()). fedify-dev#599 (comment) Co-Authored-By: claude-sonnet-4-6 <claude-sonnet-4-6@anthropic.com>
|
/gemini review |
There was a problem hiding this comment.
Code Review
This pull request introduces MysqlMessageQueue, a new MessageQueue implementation for the @fedify/mysql package, backed by MySQL or MariaDB. The implementation is robust, demonstrating a deep understanding of MySQL-specific features like SELECT ... FOR UPDATE SKIP LOCKED for concurrent dequeuing and advisory locks for ordering key serialization. The code is well-structured, handles concurrency and potential race conditions correctly, and includes comprehensive error handling. The accompanying test suite is thorough, covering a wide range of scenarios including edge cases, concurrency, and failure modes. The documentation updates are clear and provide excellent guidance for users. Overall, this is a high-quality contribution with no issues found.
Note: Security Review did not run due to the size of the PR.
MySQL evaluates NOW(6) once per statement, so all rows in a bulk INSERT share the same deliver_after timestamp. When messages share an orderingKey, dequeueing (which ORDER BY deliver_after) then selects among them non-deterministically. Fix by using delayMs * 1000 + index microseconds for each row's interval, so row i gets a deliver_after exactly i µs later than row 0. This ensures the insertion order is preserved under any orderingKey without requiring a new column or schema change. Add a regression test that enqueues five messages under the same ordering key via enqueueMany() and asserts they are delivered in order. fedify-dev#599 (comment) Co-Authored-By: Claude <noreply@anthropic.com>
ordering_key is an arbitrary string (often a URL in Fedify), so the previous VARCHAR(512) limit could silently truncate or reject values longer than 512 characters at insert time with an opaque DB error. Switch the column to TEXT so there is no fixed-length cap. Because InnoDB requires a prefix length when indexing TEXT columns, the composite index now uses ordering_key(766): 766 chars * 4 bytes (utf8mb4) + 8 bytes (DATETIME(6)) = 3072 bytes which exactly hits InnoDB's 3072-byte key limit. Existing tables created with the old VARCHAR(512) definition will continue to work; operators who want to lift the 512-char cap must ALTER the column manually (or drop and recreate the table via drop()). fedify-dev#599 (comment) Co-Authored-By: Claude <noreply@anthropic.com>
CAST(? AS JSON) is a MySQL-specific expression that is either unsupported or behaves differently on some MariaDB versions. Both MySQL and MariaDB accept a valid JSON string inserted directly into a JSON column via a parameterized placeholder; the server validates the value and stores it as JSON. Since JSON.stringify() already produces a valid JSON string, the CAST is redundant on MySQL and problematic on MariaDB. Replace CAST(? AS JSON) with ? in both enqueue() and enqueueMany(). fedify-dev#599 (comment) Co-Authored-By: Claude <noreply@anthropic.com>
The poll() calls inside listen() are always awaited sequentially — initial poll before the loop, then one poll per timer iteration. Because each safePoll call awaits its predecessor before returning, there is no scenario in which two poll() invocations can run concurrently, so the pollLock / serializedPoll machinery added no actual serialization. Remove the dead-code lock variables and rename safeSerializedPoll to safePoll to reflect its actual role: running poll() with error logging. fedify-dev#599 (comment) Co-Authored-By: Claude <noreply@anthropic.com>
withTimeout() races the handler against a timer and stops waiting when the timer wins, but it has no way to cancel the underlying handler promise. The handler therefore continues executing in the background after the timeout fires, which can cause interleaved side effects with the next handler for the same ordering key. Update the handlerTimeout JSDoc to explain this soft-timeout behaviour so operators are not surprised by concurrent execution after a timeout event. fedify-dev#599 (comment) Co-Authored-By: Claude <noreply@anthropic.com>
The ordered-key poll loop previously called #findOrderingKeyCandidate() once per lock attempt, each time issuing a query with a growing NOT IN (...) exclusion list as failed lock targets accumulated. Under high contention this produced O(n²) bytes of query text and caused repeated full-scans of the index range. Replace the loop with a single #findOrderingKeyCandidates() call that uses GROUP BY + ORDER BY MIN(deliver_after) LIMIT N to retrieve up to ORDERING_KEY_CANDIDATE_LIMIT (10) distinct ordering keys in one round- trip. The caller then iterates over the pre-fetched slice in memory, issuing GET_LOCK() calls without any further SELECT queries. If none of the N candidates is lockable the poll cycle ends and retries on the next timer tick, which is acceptable in practice. fedify-dev#599 (comment) Co-Authored-By: Claude <noreply@anthropic.com>
Two comments still referenced the old singular method name after it was renamed to #findOrderingKeyCandidates() in the GROUP-BY refactor. Updated both occurrences to match the current production method name. fedify-dev#599 (comment) fedify-dev#599 (comment) Co-Authored-By: Claude <noreply@anthropic.com>
github-project-automation
bot
moved this from In progress
to Done
in STF 2025–2026 Milestone 4: Storage and Infrastructure
Closes #586
Summary
Adds
MysqlMessageQueueto the@fedify/mysqlpackage, aMessageQueueimplementation backed by MySQL or MariaDB. This allows developers who already use MySQL or MariaDB as their primary database to use it as a Fedify message queue backend without introducing additional infrastructure such as Redis or RabbitMQ.Implementation notes
Since MySQL and MariaDB have no
LISTEN/NOTIFYequivalent, messages are delivered via periodic polling. Key design decisions:SELECT … FOR UPDATE SKIP LOCKEDensures safe concurrent dequeuing across multiple workers without row-level contention. Requires MySQL 8.0+ or MariaDB 10.6+.GET_LOCK/RELEASE_LOCK) on a dedicated connection serialize processing per ordering key, the same guarantee provided byPostgresMessageQueue.deliver_aftercolumn stores a pre-computedDATETIME(6)timestamp (DATE_ADD(NOW(6), INTERVAL ? MICROSECOND)), and the poll query filters out messages not yet due.enqueueMany()wraps all inserts in a single transaction.pollInterval).handlerTimeout). Prevents a hung handler from permanently blocking the queue.nativeRetrial:false— Fedify handles retries itself.Testing
23 tests covering constructor validation,
initialize()/drop()idempotency, concurrent initialization, delayed delivery, ordering-key ordering, advisory lock leak regression, multi-worker exclusivity, and handler timeout behavior. All tests pass against MySQL 8.0.Documentation
Updated docs/manual/mq.md (new
MysqlMessageQueuesection with install instructions, description, pros/cons, and a code example), and added references toMysqlMessageQueuein federation.md, deploy.md, relay.md, send.md, and inbox.md.Checklist
MysqlMessageQueueclassenqueue(),enqueueMany(), andlisten()with pollingdeliver_aftertimestampCHANGES.mdentry