Transactional partially-ordered message queuing

Question

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?

Answer 1

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)