wamo/gnu-social
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Fix for stuck queue messages: wrap processing in stomp transactions so our lack of an ACK if PHP dies actually triggers redelivery.

Browse Source 
Previously, messages once delivered would just get stuck in the queue seemingly forever if they never got ACKed.
Note this could lead to partial duplication, for instance if the OMB or Twitter queue handlers die after 1/2 of the outgoing sends.

Recommendations:
* catch exceptions more aggressively within queue handlers (so only PHP fatal errors are likely to kill in the middle)
* for processing that involves sending to multiple clients, consider a second queue similar to the XMPP output, eg for OMB:
 - first queue gets delivery list and builds message data, enqueueing it for each target address
 - second queue can handle each individual outgoing message (and attempt redelivery etc separately)

This would also protect better against a recurring error preventing delivery in the second part, and could spread out any slow sends over multiple threads.
This commit is contained in:
Brion Vibber 2010-01-22 12:35:05 -08:00
parent c7507e7e9d
commit 6e4cad71e5
1 changed files with 64 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

68
lib/stompqueuemanager.php
View File

@ -41,6 +41,10 @@ class StompQueueManager extends QueueManager
protected $sites = array(); protected $sites = array();
protected $useTransactions = true;
protected $transaction = null;
protected $transactionCount = 0;
function __construct() function __construct()
{ {
parent::__construct(); parent::__construct();
@ -201,6 +205,7 @@ class StompQueueManager extends QueueManager
} else { } else {
$this->doSubscribe(); $this->doSubscribe();
} }
$this->begin();
return true; return true;
} }
@ -213,6 +218,9 @@ class StompQueueManager extends QueueManager
*/ */
public function finish() public function finish()
{ {
// If there are any outstanding delivered messages we haven't processed,
// free them for another thread to take.
$this->rollback();
if ($this->sites) { if ($this->sites) {
foreach ($this->sites as $server) { foreach ($this->sites as $server) {
StatusNet::init($server); StatusNet::init($server);
@ -293,7 +301,9 @@ class StompQueueManager extends QueueManager
$notice = Notice::staticGet('id', $id); $notice = Notice::staticGet('id', $id);
if (empty($notice)) { if (empty($notice)) {
$this->_log(LOG_WARNING, "Skipping missing $info"); $this->_log(LOG_WARNING, "Skipping missing $info");
$this->con->ack($frame); $this->ack($frame);
$this->commit();
$this->begin();
$this->stats('badnotice', $queue); $this->stats('badnotice', $queue);
return false; return false;
} }
@ -308,7 +318,9 @@ class StompQueueManager extends QueueManager
$handler = $this->getHandler($queue); $handler = $this->getHandler($queue);
if (!$handler) { if (!$handler) {
$this->_log(LOG_ERROR, "Missing handler class; skipping $info"); $this->_log(LOG_ERROR, "Missing handler class; skipping $info");
$this->con->ack($frame); $this->ack($frame);
$this->commit();
$this->begin();
$this->stats('badhandler', $queue); $this->stats('badhandler', $queue);
return false; return false;
} }
@ -320,14 +332,18 @@ class StompQueueManager extends QueueManager
// FIXME we probably shouldn't have to do // FIXME we probably shouldn't have to do
// this kind of queue management ourselves; // this kind of queue management ourselves;
// if we don't ack, it should resend... // if we don't ack, it should resend...
$this->con->ack($frame); $this->ack($frame);
$this->enqueue($item, $queue); $this->enqueue($item, $queue);
$this->commit();
$this->begin();
$this->stats('requeued', $queue); $this->stats('requeued', $queue);
return false; return false;
} }
$this->_log(LOG_INFO, "Successfully handled $info"); $this->_log(LOG_INFO, "Successfully handled $info");
$this->con->ack($frame); $this->ack($frame);
$this->commit();
$this->begin();
$this->stats('handled', $queue); $this->stats('handled', $queue);
return true; return true;
} }
@ -369,5 +385,49 @@ class StompQueueManager extends QueueManager
{ {
common_log($level, 'StompQueueManager: '.$msg); common_log($level, 'StompQueueManager: '.$msg);
} }
protected function begin()
{
if ($this->useTransactions) {
if ($this->transaction) {
throw new Exception("Tried to start transaction in the middle of a transaction");
}
$this->transactionCount++;
$this->transaction = $this->master->id . '-' . $this->transactionCount . '-' . time();
$this->con->begin($this->transaction);
}
}
protected function ack($frame)
{
if ($this->useTransactions) {
if (!$this->transaction) {
throw new Exception("Tried to ack but not in a transaction");
}
}
$this->con->ack($frame, $this->transaction);
}
protected function commit()
{
if ($this->useTransactions) {
if (!$this->transaction) {
throw new Exception("Tried to commit but not in a transaction");
}
$this->con->commit($this->transaction);
$this->transaction = null;
}
}
protected function rollback()
{
if ($this->useTransactions) {
if (!$this->transaction) {
throw new Exception("Tried to rollback but not in a transaction");
}
$this->con->commit($this->transaction);
$this->transaction = null;
}
}
} }