preserve reserved queue metadata#1305
Draft
bgentry wants to merge 5 commits into
Draft
Conversation
bgentry
changed the title
Dynamic queue addition registers a producer before starting it. A startup failure can therefore leave the queue name occupied and, for plugin pilots, leave persisted producer state behind even though `QueueBundle.Add` returned an error. Clean up initialized producer state under a bounded timeout when startup fails. Remove successfully cleaned producers from the client registry so the queue name can be retried. If cleanup also fails, retain the producer behind a typed error so `Remove` can retry without losing track of it. Also release partially registered notifier subscriptions and cover both the successful-cleanup and cleanup-retry paths through a pilot-backed client.
Rate and concurrency coordination need readers that can coexist while still excluding queue-wide maintenance. Expose PostgreSQL shared advisory transaction locks through the internal driver interface and both PostgreSQL drivers. SQLite continues to return `ErrNotImplemented`. Extend the cross-driver transaction suite to verify that shared holders coexist and block an exclusive holder until both transactions finish.
Pilot implementations that interpret producer state need the same liveness horizon that River uses to reap unresponsive producers. Keeping a second duration in plugins can make stale producers count toward one limiter while River has already stopped treating them as live. Name River's existing five-minute horizon once and pass it through `PilotInitParams`. Continue using the same value in producer maintenance and verify plugin pilots receive it during client initialization.
Queue metadata needs a namespace for River-managed runtime state without exposing implementation details or allowing ordinary user updates to erase them. Treat top-level `river:*` keys as reserved while preserving the existing public metadata contract. Filter reserved keys from creates, updates, reads, lists, notifications, and producer polling. Preserve them atomically in driver `QueueUpdate` queries. Wrap non-object user metadata under an internal key so arbitrary JSON values continue to round trip while runtime state remains an object alongside them. Exercise the behavior through clients and the shared PostgreSQL and SQLite driver conformance suite.
Queue cleanup currently deletes rows solely from `updated_at`. A queue can therefore age past the retention horizon while reserved runtime state still contains unexpired debt needed by a plugin. Pass one clock value through each cleanup pass and retain rows whose internal rate-limit rollup has a future expiry. Treat malformed runtime state conservatively so cleanup cannot destroy it. Lock selected rows with `FOR UPDATE SKIP LOCKED` to avoid racing concurrent runtime updates and to let parallel cleaners make progress safely. Cover expired, live, absent, and malformed runtime state across PostgreSQL and SQLite drivers.
bgentry
force-pushed
the
bg/rate-limit-runtime-metadata
branch
from
accaa41 to
29ea9a3
Compare
brandur
reviewed
Jul 12, 2026
| THEN (river_queue.metadata -> 'river:rate_limit_rollup' ->> 'expires_at_unix')::numeric | ||
| <= extract(epoch FROM coalesce(sqlc.narg('now')::timestamptz, now())) | ||
| ELSE false | ||
| END |
Contributor
There was a problem hiding this comment.
Just wanted to make sure you saw this part — is it intentional that we're calling out specific pro metadata keys in these queries? Seems a bit leaky.
|
|
||
| // QueueMetadataForUserWrite removes reserved keys from user-supplied queue | ||
| // metadata while preserving every nonreserved value. | ||
| func QueueMetadataForUserWrite(metadata []byte) []byte { |
Contributor
There was a problem hiding this comment.
Can you clarify the "user metadata" thing for me a bit?
For metadata on jobs, aren't we kind of saying that the entire space is user metadata minus the River specific stuff that we prefix with river:*? I wonder if we should try to do the same thing here for queues.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Adds support for reserved internal queue metadata. Queue updates preserve reserved keys, while public queue reads and notifications omit them.
Queue cleanup now locks candidates before deletion and retains rows with unexpired internal state. The driver interface also exposes shared transaction advisory locks, and pilot initialization receives River's existing producer stale-retention period.