Data structure to use for a consumable message queue where it's safe for messages to overwrite one another

Question

I have a queue of messages representing filesystem operations that need to be processed in order. A message may succeed or fail when it's sent. A change message, for example, is generated when a user creates or saves a file.

• Delete Foo.txt message sent
• Delete Bar.txt message sent
• Change Baz.txt message sent
• Change Foo.txt message sent

In the most common case, I pop the top item from the queue and send it off for processing, which succeeds.

But suppose this happens:

• Delete Foo.txt message attempted (FAIL)
• Delete Bar.txt message attempted (OK)
• Change Baz.txt message attempted (OK)
• Change Foo.txt message attempted (OK)

What should I do in this case?

I want to move on and process the next item in the queue, but it's not that simple. In this case, "Delete Foo.txt" must be processed before "Create Foo.txt", otherwise we're deleting Foo.txt from the search index (see "more details" below), even though it exists.

One approach I had considered: a key-value store, where the document path is the key, and the Message is the value. In the context of a filesystem, the only operation that matters is the most recent, so overwriting the Message associated with a file if it hasn't been sent might work.

What kind of data structure and algorithm is right for this situation? I don't want to reinvent the wheel.

A few more details:

• New items are constantly arrive in the queue, from multiple producer threads
• The messages are consumed from the collection by a single consumer
• A message is a serialized object, sent to a RabbitMQ broker
• This queue exists in case the RabbitMQ broker goes down or a Publisher Confirm fails
• Messages are consumed from the endpoint by a different service which performs various actions to keep a search index incrementally up to date
• C#, if it matters

Update 1:

Big picture

I am sending messages about filesystem operations that have occurred. These messages are used to keep a search index (elasticsearch) up to date. A few minutes behind is perfectly acceptable. If a user makes three saves in rapid succession, I don't need search index updates for the first two saves. (If it happens, that's fine, but it's not required.)

CRUD operations are done by users working with their files on a busy network share. Each write operation triggers a message that has to be sent to a persistent work queue (the RabbitMQ broker). An example message might contain data that boils down to "File F has been deleted" or "File Y has been changed". (No file contents are contained therein.)

(These messages are consumed by another component downstream that performs an action based on the contents of the message, which is outside the scope of this question.)

Because messages represent filesystem actions that were performed in a sequence by users they either:

• Have to be sent in order so they can be processed in order OR
• The message associated with the most recent write operation on File F should overwrite any queued message associated with File F that hasn't been sent yet

In my example above (second unordered list), it's safe for the message representing "Change Foo.txt" to overwrite a failed "Delete Foo.txt", because the Change message was triggered after the Delete message. (The time where Foo.txt didn't exist isn't important.) Hence my thought of a using key-value store of as a holding area in case of a Message send failure.

Answer 1

Sorry, you have to tell us more. We can't tell you what you should do in case of a failure; that will depend upon the application semantics and the type of a failure.

In general, in the event of a failure of an attempted operation, there are several options:

• Retry the operation later.

• Treat the failure as permanent.

If you decide you want to retry it later, then the implications of this for other operations depends on the dependencies between operations and the desired semantics. If operations A and B commute (performing A first then B has the same effect as B then A), and if you try A and it fails, then often you can perform B immediately and then retry A. If operations A and B don't commute, and you try A first and it fails, then performing B immediately and then retrying A will have different semantics, which might or might not be acceptable depending upon what guarantees you are trying to provide.

Related topics: ACID, database transactions.

---

Example: Suppose you have a one-level filesystem (with no directories), and the intended semantics are that operations on one file can be performed in an arbitrary order compared to operations on another file (but operations on a single file are performed in the order they were requested).

Then operations A and B commute if they are performed on different files, and they don't commute if they're performed on the same file. Thus, conceptually you can maintain one queue per file. When you receive a request to perform operation $O$ on file $F$, append $O$ to the queue for $F$. Whenever you are free to perform an operation, pick an arbitrary file whose queue is non-empty and try the operation at the head of the queue. If the operation succeeds, then you can pop it from the queue. If it fails and you want to retry it later, then you have to leave it at the head of the queue to be retried later.

Such a data structure can be implemented with a hash table of queues. The hash table is keyed on the filename; it yields a queue. It's easy to implement all the necessary behaviors using this data structure.

---

I am assuming that there are only two possible outcomes of an attempted operation: success (the operation is performed in full), or failure (and there is no effect: this is the same as a no-op). If it is possible that an operation can fail partway through without rolling back its effects, so it has a partial effect, then you've got much bigger problems.