ppy/osu-queue-processor

osu-queue-processor

A lightweight redis-backed message queue processor

Contributing

Contributions can be made via pull requests to this repository. We hope to credit and reward larger contributions via a bounty system. If you're unsure of what you can help with, check out the list of open issues.

Note that while we already have certain standards in place, nothing is set in stone. If you have an issue with the way code is structured; with any libraries we are using; with any processes involved with contributing, please bring it up. I welcome all feedback so we can make contributing to this project as pain-free as possible.

Licence

The osu! client code, framework, and server-side components are licensed under the MIT licence. Please see the licence file for more information. tl;dr you can do whatever you want as long as you include the original copyright and license notice in any copy of the software/source.

Please note that this does not cover the usage of the "osu!" or "ppy" branding in any software, resources, advertising or promotion, as this is protected by trademark law.

osu-queue-processor's Issues

Failed items may not get requeued if queue processing is cancelled

It seems like when a cancellation is requested, the continuation at https://github.com/ppy/osu-queue-processor/blob/master/osu.Server.QueueProcessor/QueueProcessor.cs#L132 doesn't get awaited to completion and the process can just exit without it finishing, causing items to not get requeued.

Using osu-elastic-indexer with a batchsize of 1000

---

1000 items are added to the queue:

127.0.0.1:6381> llen osu-queue:score-index-es8
(integer) 1000

All items get marked as Failed and Cancel() is called on the cancellation token:

Everything is dequeued but only some of the initial items get requeued:
https://gist.github.com/notbakaneko/fa7b9c345923326da94f11db03966a96

In redis there're only 203 items in the queue instead of the expected 1000:

127.0.0.1:6381> llen osu-queue:score-index-es8
(integer) 203

This seems to be the process exiting before all the task continuations are run?

---

If a delay is added to ProcessResults so it doesn't return immediately (which should delay the continuations from executing), nothing gets requeued:

env $(cat .env.8) schema=es8 dotnet run queue
No existing schema version set, is this intended?
Running queue with schema version es8

Using non-aliased `dev.8.solo_scores_20220719080438`.
stats: queue:1000 inflight:0 dequeued:0 processed:0 errors:0
Starting queue processing (Backlog of 1000)..
Invalid NEST response built from a successful (200) low level call on POST: /dev.8.solo_scores_20220719080438/_bulk

Shutting down..
stats: queue:0 inflight:1000 dequeued:1000 processed:0 errors:0
stats: queue:0 inflight:1000 dequeued:1000 processed:0 errors:0
stats: queue:0 inflight:1000 dequeued:1000 processed:0 errors:0

On redis:

127.0.0.1:6381> llen osu-queue:score-index-es8
(integer) 0

---

Everything gets added back to the queue if either the process' exit is delayed with something like Thread.Sleep(10000), or .Wait(CancellationToken.None) is added to the task continuation in the queue processor.

Support pushing multiple items at once

There's possibly sizable network overhead from latency when pushing a large number of items individually. Redis already supports pushing multiple items to a list, so this could be just adjusting QueueProcessor.PushToQueue to accept an array of items.

Failed queue items should not be dropped from the queue

Or at very least it should be optional. Some queues we have these days are supposed to be robust and guarantee that every incoming item is processed.

GetDatabaseConnection without creating QueueProcessor

Would be useful for osu-elastic-indexer getting the connection string by using the same envvars in places without creating an instance of QueueProcessor that isn't needed

Actual number of inflight items can be greater than configured MaxInFlightItems

This condition means the queue still gets popped even if total in flight items equals the configured MaxInFlightItems
https://github.com/ppy/osu-queue-processor/blob/master/osu.Server.QueueProcessor/QueueProcessor.cs#L100

This was confusing as I was wondering why items still kept getting popped off the queue even though it was still pending on the configured MaxInFlightItems