Messaging: Support subsetting on arbitrary labels

[derekperkins]

Currently Messaging supports subsetting based on name, keyspace, shard and/or keyrange. I propose adding another standard field that would support subsetting based on arbitrary labels. The most flexible would be to allow for a list of key/value labels, but I think even just supporting an array of string labels would be sufficient.

Use Cases

• State management - Generally we like to process messages 3-5 times to account for intermittent failures. At that point, we consider them "failed", meaning that a developer needs to investigate the root cause. These messages should no longer be queued, but also should not be purged.
• Prioritization - We have two queue types, standard and priority, which dictates the order that they are processed. In some cases the consumer orders by high priority, and in some cases we have separate priority and standard consumers.
• High cardinality - It could be useful for subscribing to messages based on a user or tenant id for websocket-like updates.

Alternate Solutions

Without digging into the internals, it's hard to say whether or not general label filtering would be feasible from a technical standpoint.

• Dedicated failure state - I am currently participating in the Google Cloud Tasks Alpha, and the number one requested feature is for dead/buried tasks, and a similar concept exists in many other queue libraries. I think this makes a lot of sense to natively support. I think it could be added as a single api, identical to MessageAck, called MessageFailed or something, that would set a timestamp in the table. Requeuing messages would be left to the user, similar to running an update on the next_time column to reschedule.
• Multiple queues / epoch: For our use case, just having a priority and a standard queue solves the issue. I think epoch helps somewhat, but based on the index in the docs, Messaging sorts by time and then epoch, where we are essentially looking for epoch and then time.
• Update stream - This high cardinality piece isn't something we need immediately, and for the use case mentioned, using the update stream or potentially adding streaming query support would likely be more useful.

@sougou What are your thoughts?

[sougou]

• Permanently failing messages sounds like a valid use case. However, the current design has no elegant way to accommodate it: An acked message essentially sets the time_acked field to non-null. We could introduce a new state column, but that would break backward compatibility. An alternate approach would be to set a time_next far away into the future and a huge epoch, which will amount to the same thing because the message will never be resent. As for scanning for these rows, the queries should be infrequent enough that a full table scan may be acceptable.
• For priority queues, we can look at starting the lower priority messages with epoch 1, and higher ones with epoch 0. Although the time seems to participate in the prioritization, it's not the case in reality. It's mainly used to decide if a message has come due. However, the problem with epoch is that the message is high priority only for first send. It then becomes epoch=1, which will then be on par with lower priority messages.
• Filtering of messages is something we should definitely look at doing as an advanced feature.