You signed in with another tab or window. Reload to refresh your session.You signed out in another tab or window. Reload to refresh your session.You switched accounts on another tab or window. Reload to refresh your session.Dismiss alert
This commit was created on GitHub.com and signed with GitHub’s verified signature.
Consumer
[Feature] Add PGMQ::Consumer, a minimal single-threaded consumer loop modelled on the rdkafka-ruby simple
consumer. PGMQ::Consumer.new(client, queue, vt:, batch_size:, max_poll_seconds:, poll_interval_ms:, on_success:, max_attempts:) wraps an existing client; #each long-polls with read_with_poll, yields one PGMQ::Message at a
time and deletes (or archives, on_success: :archive) it when the block returns. A block exception stops the loop
and is re-raised without acknowledging the message, so it reappears after vt (at-least-once delivery). Messages
whose read_ct exceeds max_attempts (requires batch_size: 1) are archived without being yielded. #stop only
sets a flag, so it is safe in a signal trap; the loop finishes the messages it already read, then returns.
[Feature]PGMQ::Consumer consumes several queues when given an Array of names (PGMQ::Consumer.new(client, %w[orders emails])), like an rdkafka-ruby consumer subscribed to several topics. It reads them with read_multi_with_poll, rotating the read order each poll, sets queue_name on every yielded message and
acknowledges each message on its own queue. With max_attempts, it reads one message per poll across all queues.
Metrics
[Feature]PGMQ::Metrics exposes queue_visible_length (messages whose visibility timeout has passed, PGMQ
v1.5.0+) and default_partition_length (estimated rows in the default partitions of a partitioned queue and its
archive, PGMQ v1.13.0+; nil for non-partitioned queues, non-zero means pg_partman maintenance is failing). Both
are raw values like the other fields, and nil on PGMQ versions that do not return them.
Queue Management
[Feature]create_partitioned accepts premake:, the number of partitions pg_partman creates ahead of the
current one for the queue and archive tables (PGMQ's default is 4). Raise it when writes outrun the pre-made
partitions and land in the default partition. Requires PGMQ v1.13.0+ when given; without it the SQL call is
unchanged.
Documentation
[Fix] YARD examples used a non-existent msg.payload; they now use msg.message.
[Fix] The README long-polling example now treats the read_with_poll result as an Array of messages.
[Fix] The enable_notify_insert and wait_for_notify docs and the README now say that on PGMQ older than
v1.13.1, insert notifications never fire for a queue name with uppercase letters (pgmq/pgmq#593). Create queues with
lower-case names, or use PGMQ v1.13.1+.
Producer
[Fix] A Float delay: in produce, produce_batch, produce_topic and produce_batch_topic no longer fails
with PGMQ::Errors::ConnectionError (invalid input syntax for type integer). A fractional delay such as 1.5 now
makes the message visible 1.5 seconds later: it is sent to PGMQ's timestamptz overload as clock_timestamp()
plus the delay (PGMQ v1.10.0+), the same way PGMQ handles an integer delay. Integer and Time delays are unchanged.
[Breaking]produce and produce_batch raise PGMQ::Errors::MessageInvalidError when a message is not a
String (including nil) or headers are neither a String nor nil. Previously a Hash or nil was bound as SQL
NULL, so the call returned a msg_id but the message or headers were stored as NULL without an error. Any other
non-String value now raises too, including Integer, Float, true and Array values that were stored as JSON before, false headers that were dropped, and Hash values on a connection with a query type map
(PG::BasicTypeMapForQueries), which were stored as JSON objects. The error message names the invalid key
(message or headers; in a batch messages.<index>.message or messages.<index>.headers) and the reason, such
as must be a JSON String. Inside transaction the error is wrapped in PGMQ::Errors::ConnectionError like any
other error. Serialize payloads to JSON before producing. The check is the new PGMQ::Contracts::Message, built on PGMQ::Contractable, the contract layer extracted from karafka-core (no new dependency).
Validation
[Breaking] Every argument check now runs through a contract in PGMQ::Contracts (Message, Batch, MultiQueue, Consumer, Autovacuum, Connection), with the error messages in config/locales/errors.yml.
Client calls raise new validation errors instead of ArgumentError, like WaterDrop: PGMQ::Errors::MessageInvalidError (produce, produce_batch, produce_batch_topic), PGMQ::Errors::MultiQueueInvalidError (read_multi, read_multi_with_poll, pop_multi, delete_multi, archive_multi, set_vt_multi) and PGMQ::Errors::AutovacuumSettingsInvalidError (tune_autovacuum). They
inherit from PGMQ::Errors::BaseError, not from ArgumentError, so rescue ArgumentError no longer catches them. PGMQ::Consumer.new, the connection parameters and the PGMQ::Connection reconnectable error settings still raise PGMQ::Errors::ConfigurationError. The message is now a Hash of each invalid key and its reason, such as queue_names and cannot be empty. The headers length error of produce_batch and produce_batch_topic no longer
includes the two lengths. New checks: produce_batch and produce_batch_topic raise PGMQ::Errors::MessageInvalidError when messages is not an Array or headers is neither an Array nor nil (String
headers failed in SQL with PGMQ::Errors::ConnectionError before), and tune_autovacuum checks the queue and
archive settings before it changes any table (an unknown archive setting raised only after the queue table was
changed). A setting value that is not numeric raises PGMQ::Errors::AutovacuumSettingsInvalidError too, also before
any table changes (it raised ArgumentError from Float() or Integer() for a String, TypeError for other values
such as nil or true, so rescue TypeError no longer catches it). Two calls that returned without an error now
raise a validation error: produce_batch and produce_batch_topic with an empty messages Array and non-empty or
String headers (they returned []), and tune_autovacuum (and create with tune_autovacuum:) with archive: false and an unknown or non-numeric archive setting (it ignored the archive settings).
Transactions
[Fix] The transactional client yielded by PGMQ::Client#transaction now raises the new PGMQ::Errors::TransactionError when used from another thread or fiber, or after the block has returned.
Previously such calls checked out a different pool connection and committed outside the transaction, so a
rollback did not undo them (and with pool_size: 1 they waited for the pool timeout). An unrescued TransactionError propagates from #transaction as is, not wrapped in PGMQ::Errors::ConnectionError.
Testing
[Fix] Transaction tests produce JSON strings instead of Hashes, which were stored as NULL.
[Change] Add spec/integration/concurrency_overlap_spec.rb, which proves concurrency by asserting overlapping
execution intervals (not order or absolute timings): threads and fibers overlap on long polls and LISTEN waits,
the reactor keeps running, a pool of one serializes work, and the transactional client rejects other threads and
fibers. ConcurrencyTracker now uses the monotonic clock, and the fiber long-polling example asserts its result.
[Fix]FiberSchedulerHelper.with_scheduler uses Sync instead of a top-level Async, so a failed assertion
inside a fiber scheduler integration spec now fails the spec instead of being only logged.