# Transactional partially-ordered message queuing

I'm having a really hard time finding a solution to a practical programming problem, and I think raw CS might have an answer.

The requirement is for a familiar message queue for which there are multiple message Generators and multiple message Consumers. Generators enqueue messages into a queue within the scope of an transaction. Consumers dequeue messages within the scope of a transaction, and process the message successfully, removing it from the queue, or fail and roll back the transaction in which case the message remains on the queue. The message may be left on the queue by the Consumer in an altered state - decrementing a retry-count, for example.

Adding to this the concept of a Sequence or Conversation - a partial ordering within the set of all messages in the queue. Messages which are part of a Sequence are to be consumed in-order. Sequences are unordered amongst Sequences.

Generators do not block other Generators unless the messages being enqueued are for the same Sequence. Generators never block Consumers. Consumers never block Generators. And Consumers never block other Consumers. That is, Enqueue() may block, Dequeue() never blocks. Generators Enqueue a message exactly once. Consumers successfully process each message exactly once.

It'd be nice if the implementation doesn't leave too much trash data around, but as long as old data can eventually be trimmed, that's fine.

It seems like this scenario should be possible, yet I cannot find or design correct rules on a common platform. SQL Service Broker might match all of these criteria, but if it does, how does it?

Is such a protocol provably impossible? If not, how might it be done/what properties must the underlying storage platform have to support it?

• Why do you say that Dequeue() never blocks? A consumer cannot pick and process a message N+1 of sequence S if another consumer is still processing (under transaction) the (previous) message N of sequence S – Marzio De Biasi Nov 24 '11 at 17:47
• No. If you attempt to Dequeue() and there are no available messages (that aren't blocked) than nothing is returned. – Jason Kleban Nov 24 '11 at 17:58

A standard RDBMS should be enough; this is a sketch of how it can be done in a RDBMS even without BEGIN TRANSACTION capabilities (a mix of SQL/pseudocode):

Create two tables with the following columns:

Table MESSAGES:
*MSGID            int not null auto_increment,  (primary key)
SEQID             int,
SEQORDER          int not null auto_increment,
RETRIES           int default 0,
MSGDATA           varchar(...)

Table SEQUENCES
*SEQID            int not null,  (primary key)
SEQLOCKED         int,
SEQHASH           varchar(256)


The Generators can simply insert into the MESSAGE table their requests (and optionally fill the SEQUENCES table):

For example suppose a Generator must insert a message with SEQID=120

1) INSERT INTO SEQUENCES VALUES (120, 0, MD5(RAND()) )
// if the SEQID is already present the previous query
// just fails, do not worry about it
2) INSERT INTO MESSAGES (SEQID, MSGDATA) VALUES (120, 'bla bla bla')
// the auto_increment option in columns will assign correct values for MSGID and
// SEQORDER


The Consumers can do their stuff without using BEGIN TRANSACTION in this way:

// retrieve the SEQID of an unlocked sequence that contains a message:
1) SELECT * FROM SEQUENCES s LEFT JOIN MESSAGES m ON  m.SEQID = s.SEQID
WHERE SEQLOCKED = 0
ORDER BY SEQORD LIMIT 1
// suppose we retrieve a sequence with SEQID=120 and SEQHASH='102f...'
// we first lock the sequence using a standard tecnique:
2) newhash = MD5(RAND())
3) UPDATE SEQUENCES SET (SEQHASH=newhash, SEQLOCKED=1)
WHERE (SEQID=120 AND SEQHASH='102f...')
4) IF (records_updated() = 1)   // lock acquired
// we pick the older message of the sequence
4.1) SELECT * FROM MESSAGES s WHERE SEQID=120 ORDER BY SEQORD LIMIT 1
4.2) .... if we get a message (suppose MSGID=1234) then process it ....
4.3) IF everything is ok THEN
DELETE MESSAGES WHERE MSGID=1234
ELSE
UPDATE MESSAGES SET (RETRIES = RETRIES+1) WHERE MSGID=1234
// finally we must unlock the sequence
4.4) UPDATE SEQUENCES SET (SEQLOCKED=0) WHERE SEQID=120

restart from step 1)

• Thanks for taking the time to read and respond in such detail. That's an interesting design, but without the transaction, the sequence might be locked forever if the Consumer fails between steps 4 and 4.4, right? Is there a good policy on how to recover from such a failure? Is the transactional way impossible? – Jason Kleban Nov 25 '11 at 21:27
• Yes you are right; if a Consumer can fail then there are several ways to avoid locks. The better way is to use a transactional RDBMS (put a BEGIN TRANSACTION at step 0: if Consumer fails, the SQL session is dropped and a ROLLBACK is automatically executed) and, even with dirty reads (lowest isolation level), step 3 should guarantees that everything is ok. Another approach is to add a timestamp when a sequence is locked; and use a "control" process (or an UPDATE query at the beginning of each Consumer) to unlock sequences that has been locked for a specified amount of time. – Marzio De Biasi Nov 26 '11 at 21:46
• Ok, assuming that UPDATE cannot block a row based on a dirty read of it's expired hash value, I think there remains an issue with when to DELETE from the Sequence table. A generator shouldn't delete Sequences because the items may not have been processed, but it can mark it "closed". The Consumer can mark a "closed" Sequence "complete" or I guess just delete it. But within a transaction, the UPDATE in 3) will be blocked by any Generator trying to "close" the Sequence. Not pruning Sequences is not practical. I'm looking into a special TERMINATE_SEQUENCE message instead. – Jason Kleban Nov 28 '11 at 15:48
• Yes, for closing sequences you can use an additional field FLAST in the message table that is set to TRUE in the last message of a sequence. Or as you say use a whole dedicated message. Then you add the "DELETE sequences ..." query to the consumers (IF flast THEN <delete sequence> ELSE <unlock sequence>). – Marzio De Biasi Nov 28 '11 at 23:08
• Thanks! I think I've got it now. To steer this, finally, back to TCS-relevancy, did you just make this up for me? or what's it called - where can I find references on this general problem? – Jason Kleban Nov 29 '11 at 0:55