Merge branch 'testing' of git@gitorious.org:statusnet/mainline into testing

8 files changed, 405 insertions, 105 deletions

diff --git a/classes/Memcached_DataObject.php b/classes/Memcached_DataObject.php
index 2cc6377f8..f4dfe6314 100644
--- a/classes/Memcached_DataObject.php
+++ b/classes/Memcached_DataObject.php
@@ -147,6 +147,7 @@ class Memcached_DataObject extends DB_DataObject
     {
         $result = parent::insert();
         if ($result) {
+            $this->fixupTimestamps();
             $this->encache(); // in case of cached negative lookups
         }
         return $result;
@@ -159,6 +160,7 @@ class Memcached_DataObject extends DB_DataObject
         }
         $result = parent::update($orig);
         if ($result) {
+            $this->fixupTimestamps();
             $this->encache();
         }
         return $result;
@@ -366,7 +368,7 @@ class Memcached_DataObject extends DB_DataObject
     }
 
     /**
-     * sends query to database - this is the private one that must work 
+     * sends query to database - this is the private one that must work
      *   - internal functions use this rather than $this->query()
      *
      * Overridden to do logging.
@@ -428,7 +430,7 @@ class Memcached_DataObject extends DB_DataObject
         //
         // WARNING WARNING if we end up actually using multiple DBs at a time
         // we'll need some fancier logic here.
-        if (!$exists && !empty($_DB_DATAOBJECT['CONNECTIONS'])) {
+        if (!$exists && !empty($_DB_DATAOBJECT['CONNECTIONS']) && php_sapi_name() == 'cli') {
             foreach ($_DB_DATAOBJECT['CONNECTIONS'] as $index => $conn) {
                 if (!empty($conn)) {
                     $conn->disconnect();
@@ -529,4 +531,25 @@ class Memcached_DataObject extends DB_DataObject
 
         return $c->delete($cacheKey);
     }
+
+    function fixupTimestamps()
+    {
+        // Fake up timestamp columns
+        $columns = $this->table();
+        foreach ($columns as $name => $type) {
+            if ($type & DB_DATAOBJECT_MYSQLTIMESTAMP) {
+                $this->$name = common_sql_now();
+            }
+        }
+    }
+
+    function debugDump()
+    {
+        common_debug("debugDump: " . common_log_objstring($this));
+    }
+
+    function raiseError($message, $type = null, $behaviour = null)
+    {
+        throw new ServerException("DB_DataObject error [$type]: $message");
+    }
 }
diff --git a/classes/Notice.php b/classes/Notice.php
index 90e3e76ef..a60dd5bcd 100644
--- a/classes/Notice.php
+++ b/classes/Notice.php
@@ -326,13 +326,7 @@ class Notice extends Memcached_DataObject
         # XXX: someone clever could prepend instead of clearing the cache
         $notice->blowOnInsert();
 
-        if (common_config('queue', 'inboxes')) {
-            $qm = QueueManager::get();
-            $qm->enqueue($notice, 'distrib');
-        } else {
-            $handler = new DistribQueueHandler();
-            $handler->handle($notice);
-        }
+        $notice->distribute();
 
         return $notice;
     }
@@ -1447,4 +1441,31 @@ class Notice extends Memcached_DataObject
 
         $gi->free();
     }
+
+    function distribute()
+    {
+        if (common_config('queue', 'inboxes')) {
+            // If there's a failure, we want to _force_
+            // distribution at this point.
+            try {
+                $qm = QueueManager::get();
+                $qm->enqueue($this, 'distrib');
+            } catch (Exception $e) {
+                // If the exception isn't transient, this
+                // may throw more exceptions as DQH does
+                // its own enqueueing. So, we ignore them!
+                try {
+                    $handler = new DistribQueueHandler();
+                    $handler->handle($this);
+                } catch (Exception $e) {
+                    common_log(LOG_ERR, "emergency redistribution resulted in " . $e->getMessage());
+                }
+                // Re-throw so somebody smarter can handle it.
+                throw $e;
+            }
+        } else {
+            $handler = new DistribQueueHandler();
+            $handler->handle($this);
+        }
+    }
 }
diff --git a/classes/Session.php b/classes/Session.php
index 79a69a96e..2422f8b68 100644
--- a/classes/Session.php
+++ b/classes/Session.php
@@ -64,8 +64,12 @@ class Session extends Memcached_DataObject
         $session = Session::staticGet('id', $id);
 
         if (empty($session)) {
+            self::logdeb("Couldn't find '$id'");
             return '';
         } else {
+            self::logdeb("Found '$id', returning " .
+                         strlen($session->session_data) .
+                         " chars of data");
             return (string)$session->session_data;
         }
     }
@@ -77,14 +81,24 @@ class Session extends Memcached_DataObject
         $session = Session::staticGet('id', $id);
 
         if (empty($session)) {
+            self::logdeb("'$id' doesn't yet exist; inserting.");
             $session = new Session();
 
             $session->id           = $id;
             $session->session_data = $session_data;
             $session->created      = common_sql_now();
 
-            return $session->insert();
+            $result = $session->insert();
+
+            if (!$result) {
+                common_log_db_error($session, 'INSERT', __FILE__);
+                self::logdeb("Failed to insert '$id'.");
+            } else {
+                self::logdeb("Successfully inserted '$id' (result = $result).");
+            }
+            return $result;
         } else {
+            self::logdeb("'$id' already exists; updating.");
             if (strcmp($session->session_data, $session_data) == 0) {
                 self::logdeb("Not writing session '$id'; unchanged");
                 return true;
@@ -95,7 +109,16 @@ class Session extends Memcached_DataObject
 
                 $session->session_data = $session_data;
 
-                return $session->update($orig);
+                $result = $session->update($orig);
+
+                if (!$result) {
+                    common_log_db_error($session, 'UPDATE', __FILE__);
+                    self::logdeb("Failed to update '$id'.");
+                } else {
+                    self::logdeb("Successfully updated '$id' (result = $result).");
+                }
+
+                return $result;
             }
         }
     }
@@ -106,8 +129,17 @@ class Session extends Memcached_DataObject
 
         $session = Session::staticGet('id', $id);
 
-        if (!empty($session)) {
-            return $session->delete();
+        if (empty($session)) {
+            self::logdeb("Can't find '$id' to delete.");
+        } else {
+            $result = $session->delete();
+            if (!$result) {
+                common_log_db_error($session, 'DELETE', __FILE__);
+                self::logdeb("Failed to delete '$id'.");
+            } else {
+                self::logdeb("Successfully deleted '$id' (result = $result).");
+            }
+            return $result;
         }
     }
 
@@ -132,7 +164,10 @@ class Session extends Memcached_DataObject
 
         $session->free();
 
+        self::logdeb("Found " . count($ids) . " ids to delete.");
+
         foreach ($ids as $id) {
+            self::logdeb("Destroying session '$id'.");
             self::destroy($id);
         }
     }
diff --git a/index.php b/index.php
index b5edc0f94..5520d690b 100644
--- a/index.php
+++ b/index.php
@@ -152,6 +152,16 @@ function checkMirror($action_obj, $args)
 
     static $alwaysRW = array('session', 'remember_me');
 
+    // We ensure that these tables always are used
+    // on the master DB
+
+    $config['db']['database_rw'] = $config['db']['database'];
+    $config['db']['ini_rw'] = INSTALLDIR.'/classes/statusnet.ini';
+
+    foreach ($alwaysRW as $table) {
+        $config['db']['table_'.$table] = 'rw';
+    }
+
     if (common_config('db', 'mirror') && $action_obj->isReadOnly($args)) {
         if (is_array(common_config('db', 'mirror'))) {
             // "load balancing", ha ha
@@ -162,16 +172,6 @@ function checkMirror($action_obj, $args)
             $mirror = common_config('db', 'mirror');
         }
 
-        // We ensure that these tables always are used
-        // on the master DB
-
-        $config['db']['database_rw'] = $config['db']['database'];
-        $config['db']['ini_rw'] = INSTALLDIR.'/classes/statusnet.ini';
-
-        foreach ($alwaysRW as $table) {
-            $config['db']['table_'.$table] = 'rw';
-        }
-
         // everyone else uses the mirror
 
         $config['db']['database'] = $mirror;
diff --git a/lib/default.php b/lib/default.php
index c729193b5..c01508695 100644
--- a/lib/default.php
+++ b/lib/default.php
@@ -84,6 +84,8 @@ $default =
               'control_channel' => '/topic/statusnet-control', // broadcasts to all queue daemons
               'stomp_username' => null,
               'stomp_password' => null,
+              'stomp_persistent' => true, // keep items across queue server restart, if persistence is enabled
+              'stomp_manual_failover' => true, // if multiple servers are listed, treat them as separate (enqueue on one randomly, listen on all)
               'monitor' => null, // URL to monitor ping endpoint (work in progress)
               'softlimit' => '90%', // total size or % of memory_limit at which to restart queue threads gracefully
               'debug_memory' => false, // true to spit memory usage to log
diff --git a/lib/liberalstomp.php b/lib/liberalstomp.php
index c9233843a..3d38953fd 100644
--- a/lib/liberalstomp.php
+++ b/lib/liberalstomp.php
@@ -34,6 +34,22 @@ class LiberalStomp extends Stomp
     }
 
     /**
+     * Return the host we're currently connected to.
+     *
+     * @return string
+     */
+    function getServer()
+    {
+        $idx = $this->_currentHost;
+        if ($idx >= 0) {
+            $host = $this->_hosts[$idx];
+            return "$host[0]:$host[1]";
+        } else {
+            return '[unconnected]';
+        }
+    }
+
+    /**
      * Make socket connection to the server
      * We also set the stream to non-blocking mode, since we'll be
      * select'ing to wait for updates. In blocking mode it seems
@@ -71,10 +87,12 @@ class LiberalStomp extends Stomp
             // @fixme this sometimes hangs in blocking mode...
             // shouldn't we have been idle until we found there's more data?
             $read = fread($this->_socket, $rb);
-            if ($read === false) {
-                $this->_reconnect();
+            if ($read === false || ($read === '' && feof($this->_socket))) {
+                // @fixme possibly attempt an auto reconnect as old code?
+                throw new StompException("Error reading");
+                //$this->_reconnect();
                 // @fixme this will lose prior items
-                return $this->readFrames();
+                //return $this->readFrames();
             }
             $data .= $read;
             if (strpos($data, "\x00") !== false) {
diff --git a/lib/stompqueuemanager.php b/lib/stompqueuemanager.php
index 19e8c49b5..ec150bbb6 100644
--- a/lib/stompqueuemanager.php
+++ b/lib/stompqueuemanager.php
@@ -29,28 +29,37 @@
  */
 
 require_once 'Stomp.php';
+require_once 'Stomp/Exception.php';
 
 
 class StompQueueManager extends QueueManager
 {
-    var $server = null;
-    var $username = null;
-    var $password = null;
-    var $base = null;
-    var $con = null;
+    protected $servers;
+    protected $username;
+    protected $password;
+    protected $base;
     protected $control;
+
+    protected $useTransactions = true;
     
     protected $sites = array();
     protected $subscriptions = array();
 
-    protected $useTransactions = true;
-    protected $transaction = null;
-    protected $transactionCount = 0;
+    protected $cons = array(); // all open connections
+    protected $disconnect = array();
+    protected $transaction = array();
+    protected $transactionCount = array();
+    protected $defaultIdx = 0;
 
     function __construct()
     {
         parent::__construct();
-        $this->server   = common_config('queue', 'stomp_server');
+        $server = common_config('queue', 'stomp_server');
+        if (is_array($server)) {
+            $this->servers = $server;
+        } else {
+            $this->servers = array($server);
+        }
         $this->username = common_config('queue', 'stomp_username');
         $this->password = common_config('queue', 'stomp_password');
         $this->base     = common_config('queue', 'queue_basename');
@@ -99,9 +108,9 @@ class StompQueueManager extends QueueManager
             $message .= ':' . $param;
         }
         $this->_connect();
-        $result = $this->con->send($this->control,
-                                   $message,
-                                   array ('created' => common_sql_now()));
+        $result = $this->_send($this->control,
+                               $message,
+                               array ('created' => common_sql_now()));
         if ($result) {
             $this->_log(LOG_INFO, "Sent control ping to queue daemons: $message");
             return true;
@@ -166,28 +175,59 @@ class StompQueueManager extends QueueManager
     /**
      * Saves a notice object reference into the queue item table.
      * @return boolean true on success
+     * @throws StompException on connection or send error
      */
     public function enqueue($object, $queue)
     {
+        $this->_connect();
+        return $this->_doEnqueue($object, $queue, $this->defaultIdx);
+    }
+    
+    /**
+     * Saves a notice object reference into the queue item table
+     * on the given connection.
+     *
+     * @return boolean true on success
+     * @throws StompException on connection or send error
+     */
+    protected function _doEnqueue($object, $queue, $idx)
+    {
         $msg = $this->encode($object);
         $rep = $this->logrep($object);
 
-        $this->_connect();
-
-        // XXX: serialize and send entire notice
+        $props = array('created' => common_sql_now());
+        if ($this->isPersistent($queue)) {
+            $props['persistent'] = 'true';
+        }
 
-        $result = $this->con->send($this->queueName($queue),
-                                   $msg, 		// BODY of the message
-                                   array ('created' => common_sql_now(),
-                                          'persistent' => 'true'));
+        $con = $this->cons[$idx];
+        $host = $con->getServer();
+        $result = $con->send($this->queueName($queue), $msg, $props);
 
         if (!$result) {
-            common_log(LOG_ERR, "Error sending $rep to $queue queue");
+            common_log(LOG_ERR, "Error sending $rep to $queue queue on $host");
             return false;
         }
 
-        common_log(LOG_DEBUG, "complete remote queueing $rep for $queue");
+        common_log(LOG_DEBUG, "complete remote queueing $rep for $queue on $host");
         $this->stats('enqueued', $queue);
+        return true;
+    }
+
+    /**
+     * Determine whether messages to this queue should be marked as persistent.
+     * Actual persistent storage depends on the queue server's configuration.
+     * @param string $queue
+     * @return bool
+     */
+    protected function isPersistent($queue)
+    {
+        $mode = common_config('queue', 'stomp_persistent');
+        if (is_array($mode)) {
+            return in_array($queue, $mode);
+        } else {
+            return (bool)$mode;
+        }
     }
 
     /**
@@ -198,7 +238,29 @@ class StompQueueManager extends QueueManager
      */
     public function getSockets()
     {
-        return array($this->con->getSocket());
+        $sockets = array();
+        foreach ($this->cons as $con) {
+            if ($con) {
+                $sockets[] = $con->getSocket();
+            }
+        }
+        return $sockets;
+    }
+
+    /**
+     * Get the Stomp connection object associated with the given socket.
+     * @param resource $socket
+     * @return int index into connections list
+     * @throws Exception
+     */
+    protected function connectionFromSocket($socket)
+    {
+        foreach ($this->cons as $i => $con) {
+            if ($con && $con->getSocket() === $socket) {
+                return $i;
+            }
+        }
+        throw new Exception(__CLASS__ . " asked to read from unrecognized socket");
     }
 
     /**
@@ -210,27 +272,56 @@ class StompQueueManager extends QueueManager
      */
     public function handleInput($socket)
     {
-        assert($socket === $this->con->getSocket());
+        $idx = $this->connectionFromSocket($socket);
+        $con = $this->cons[$idx];
+        $host = $con->getServer();
+
         $ok = true;
-        $frames = $this->con->readFrames();
+        try {
+            $frames = $con->readFrames();
+        } catch (StompException $e) {
+            common_log(LOG_ERR, "Lost connection to $host: " . $e->getMessage());
+            $this->cons[$idx] = null;
+            $this->transaction[$idx] = null;
+            $this->disconnect[$idx] = time();
+            return false;
+        }
         foreach ($frames as $frame) {
             $dest = $frame->headers['destination'];
             if ($dest == $this->control) {
-                if (!$this->handleControlSignal($frame)) {
+                if (!$this->handleControlSignal($idx, $frame)) {
                     // We got a control event that requests a shutdown;
                     // close out and stop handling anything else!
                     break;
                 }
             } else {
-                $ok = $ok && $this->handleItem($frame);
+                $ok = $ok && $this->handleItem($idx, $frame);
             }
         }
         return $ok;
     }
 
     /**
+     * Attempt to reconnect in background if we lost a connection.
+     */
+    function idle()
+    {
+        $now = time();
+        foreach ($this->cons as $idx => $con) {
+            if (empty($con)) {
+                $age = $now - $this->disconnect[$idx];
+                if ($age >= 60) {
+                    $this->_reconnect($idx);
+                }
+            }
+        }
+        return true;
+    }
+
+    /**
      * Initialize our connection and subscribe to all the queues
-     * we're going to need to handle...
+     * we're going to need to handle... If multiple queue servers
+     * are configured for failover, we'll listen to all of them.
      *
      * Side effects: in multi-site mode, may reset site configuration.
      *
@@ -240,9 +331,14 @@ class StompQueueManager extends QueueManager
     public function start($master)
     {
         parent::start($master);
-        $this->_connect();
+        $this->_connectAll();
 
-        $this->con->subscribe($this->control);
+        common_log(LOG_INFO, "Subscribing to $this->control");
+        foreach ($this->cons as $con) {
+            if ($con) {
+                $con->subscribe($this->control);
+            }
+        }
         if ($this->sites) {
             foreach ($this->sites as $server) {
                 StatusNet::init($server);
@@ -251,7 +347,11 @@ class StompQueueManager extends QueueManager
         } else {
             $this->doSubscribe();
         }
-        $this->begin();
+        foreach ($this->cons as $i => $con) {
+            if ($con) {
+                $this->begin($i);
+            }
+        }
         return true;
     }
     
@@ -266,8 +366,12 @@ class StompQueueManager extends QueueManager
     {
         // If there are any outstanding delivered messages we haven't processed,
         // free them for another thread to take.
-        $this->rollback();
-        $this->con->unsubscribe($this->control);
+        foreach ($this->cons as $i => $con) {
+            if ($con) {
+                $this->rollback($i);
+                $con->unsubscribe($this->control);
+            }
+        }
         if ($this->sites) {
             foreach ($this->sites as $server) {
                 StatusNet::init($server);
@@ -289,24 +393,107 @@ class StompQueueManager extends QueueManager
     }
 
     /**
-     * Lazy open connection to Stomp queue server.
+     * Lazy open a single connection to Stomp queue server.
+     * If multiple servers are configured, we let the Stomp client library
+     * worry about finding a working connection among them.
      */
     protected function _connect()
     {
-        if (empty($this->con)) {
-            $this->_log(LOG_INFO, "Connecting to '$this->server' as '$this->username'...");
-            $this->con = new LiberalStomp($this->server);
-
-            if ($this->con->connect($this->username, $this->password)) {
-                $this->_log(LOG_INFO, "Connected.");
+        if (empty($this->cons)) {
+            $list = $this->servers;
+            if (count($list) > 1) {
+                shuffle($list); // Randomize to spread load
+                $url = 'failover://(' . implode(',', $list) . ')';
             } else {
-                $this->_log(LOG_ERR, 'Failed to connect to queue server');
-                throw new ServerException('Failed to connect to queue server');
+                $url = $list[0];
             }
+            $con = $this->_doConnect($url);
+            $this->cons = array($con);
+            $this->transactionCount = array(0);
+            $this->transaction = array(null);
+            $this->disconnect = array(null);
         }
     }
 
     /**
+     * Lazy open connections to all Stomp servers, if in manual failover
+     * mode. This means the queue servers don't speak to each other, so
+     * we have to listen to all of them to make sure we get all events.
+     */
+    protected function _connectAll()
+    {
+        if (!common_config('queue', 'stomp_manual_failover')) {
+            return $this->_connect();
+        }
+        if (empty($this->cons)) {
+            $this->cons = array();
+            $this->transactionCount = array();
+            $this->transaction = array();
+            foreach ($this->servers as $idx => $server) {
+                try {
+                    $this->cons[] = $this->_doConnect($server);
+                    $this->disconnect[] = null;
+                } catch (Exception $e) {
+                    // s'okay, we'll live
+                    $this->cons[] = null;
+                    $this->disconnect[] = time();
+                }
+                $this->transactionCount[] = 0;
+                $this->transaction[] = null;
+            }
+            if (empty($this->cons)) {
+                throw new ServerException("No queue servers reachable...");
+                return false;
+            }
+        }
+    }
+
+    protected function _reconnect($idx)
+    {
+        try {
+            $con = $this->_doConnect($this->servers[$idx]);
+        } catch (Exception $e) {
+            $this->_log(LOG_ERR, $e->getMessage());
+            $con = null;
+        }
+        if ($con) {
+            $this->cons[$idx] = $con;
+            $this->disconnect[$idx] = null;
+            
+            // now we have to listen to everything...
+            // @fixme refactor this nicer. :P
+            $host = $con->getServer();
+            $this->_log(LOG_INFO, "Resubscribing to $this->control on $host");
+            $con->subscribe($this->control);
+            foreach ($this->subscriptions as $site => $queues) {
+                foreach ($queues as $queue) {
+                    $this->_log(LOG_INFO, "Resubscribing to $queue on $host");
+                    $con->subscribe($queue);
+                }
+            }
+            $this->begin($idx);
+        } else {
+            // Try again later...
+            $this->disconnect[$idx] = time();
+        }
+    }
+
+    protected function _doConnect($server)
+    {
+        $this->_log(LOG_INFO, "Connecting to '$server' as '$this->username'...");
+        $con = new LiberalStomp($server);
+
+        if ($con->connect($this->username, $this->password)) {
+            $this->_log(LOG_INFO, "Connected.");
+        } else {
+            $this->_log(LOG_ERR, 'Failed to connect to queue server');
+            throw new ServerException('Failed to connect to queue server');
+        }
+
+        return $con;
+    }
+
+    /**
      * Subscribe to all enabled notice queues for the current site.
      */
     protected function doSubscribe()
@@ -317,7 +504,11 @@ class StompQueueManager extends QueueManager
             $rawqueue = $this->queueName($queue);
             $this->subscriptions[$site][$queue] = $rawqueue;
             $this->_log(LOG_INFO, "Subscribing to $rawqueue");
-            $this->con->subscribe($rawqueue);
+            foreach ($this->cons as $con) {
+                if ($con) {
+                    $con->subscribe($rawqueue);
+                }
+            }
         }
     }
 
@@ -331,7 +522,11 @@ class StompQueueManager extends QueueManager
         if (!empty($this->subscriptions[$site])) {
             foreach ($this->subscriptions[$site] as $queue => $rawqueue) {
                 $this->_log(LOG_INFO, "Unsubscribing from $rawqueue");
-                $this->con->unsubscribe($rawqueue);
+                foreach ($this->cons as $con) {
+                    if ($con) {
+                        $con->unsubscribe($rawqueue);
+                    }
+                }
                 unset($this->subscriptions[$site][$queue]);
             }
         }
@@ -346,27 +541,31 @@ class StompQueueManager extends QueueManager
      * Side effects: in multi-site mode, may reset site configuration to
      * match the site that queued the event.
      *
+     * @param int $idx connection index
      * @param StompFrame $frame
      * @return bool
      */
-    protected function handleItem($frame)
+    protected function handleItem($idx, $frame)
     {
+        $this->defaultIdx = $idx;
+
         list($site, $queue) = $this->parseDestination($frame->headers['destination']);
         if ($site != $this->currentSite()) {
             $this->stats('switch');
             StatusNet::init($site);
         }
 
+        $host = $this->cons[$idx]->getServer();
         if (is_numeric($frame->body)) {
             $id = intval($frame->body);
-            $info = "notice $id posted at {$frame->headers['created']} in queue $queue";
+            $info = "notice $id posted at {$frame->headers['created']} in queue $queue from $host";
 
             $notice = Notice::staticGet('id', $id);
             if (empty($notice)) {
                 $this->_log(LOG_WARNING, "Skipping missing $info");
-                $this->ack($frame);
-                $this->commit();
-                $this->begin();
+                $this->ack($idx, $frame);
+                $this->commit($idx);
+                $this->begin($idx);
                 $this->stats('badnotice', $queue);
                 return false;
             }
@@ -374,16 +573,16 @@ class StompQueueManager extends QueueManager
             $item = $notice;
         } else {
             // @fixme should we serialize, or json, or what here?
-            $info = "string posted at {$frame->headers['created']} in queue $queue";
+            $info = "string posted at {$frame->headers['created']} in queue $queue from $host";
             $item = $frame->body;
         }
 
         $handler = $this->getHandler($queue);
         if (!$handler) {
             $this->_log(LOG_ERR, "Missing handler class; skipping $info");
-            $this->ack($frame);
-            $this->commit();
-            $this->begin();
+            $this->ack($idx, $frame);
+            $this->commit($idx);
+            $this->begin($idx);
             $this->stats('badhandler', $queue);
             return false;
         }
@@ -395,18 +594,18 @@ class StompQueueManager extends QueueManager
             // FIXME we probably shouldn't have to do
             // this kind of queue management ourselves;
             // if we don't ack, it should resend...
-            $this->ack($frame);
+            $this->ack($idx, $frame);
             $this->enqueue($item, $queue);
-            $this->commit();
-            $this->begin();
+            $this->commit($idx);
+            $this->begin($idx);
             $this->stats('requeued', $queue);
             return false;
         }
 
         $this->_log(LOG_INFO, "Successfully handled $info");
-        $this->ack($frame);
-        $this->commit();
-        $this->begin();
+        $this->ack($idx, $frame);
+        $this->commit($idx);
+        $this->begin($idx);
         $this->stats('handled', $queue);
         return true;
     }
@@ -414,10 +613,11 @@ class StompQueueManager extends QueueManager
     /**
      * Process a control signal broadcast.
      *
+     * @param int $idx connection index
      * @param array $frame Stomp frame
      * @return bool true to continue; false to stop further processing.
      */
-    protected function handleControlSignal($frame)
+    protected function handleControlSignal($idx, $frame)
     {
         $message = trim($frame->body);
         if (strpos($message, ':') !== false) {
@@ -441,9 +641,9 @@ class StompQueueManager extends QueueManager
             $this->_log(LOG_ERR, "Ignoring unrecognized control message: $message");
         }
 
-        $this->ack($frame);
-        $this->commit();
-        $this->begin();
+        $this->ack($idx, $frame);
+        $this->commit($idx);
+        $this->begin($idx);
         return $shutdown;
     }
     
@@ -520,47 +720,49 @@ class StompQueueManager extends QueueManager
         common_log($level, 'StompQueueManager: '.$msg);
     }
 
-    protected function begin()
+    protected function begin($idx)
     {
         if ($this->useTransactions) {
-            if ($this->transaction) {
+            if (!empty($this->transaction[$idx])) {
                 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);
+            $this->transactionCount[$idx]++;
+            $this->transaction[$idx] = $this->master->id . '-' . $this->transactionCount[$idx] . '-' . time();
+            $this->cons[$idx]->begin($this->transaction[$idx]);
         }
     }
 
-    protected function ack($frame)
+    protected function ack($idx, $frame)
     {
         if ($this->useTransactions) {
-            if (!$this->transaction) {
+            if (empty($this->transaction[$idx])) {
                 throw new Exception("Tried to ack but not in a transaction");
             }
+            $this->cons[$idx]->ack($frame, $this->transaction[$idx]);
+        } else {
+            $this->cons[$idx]->ack($frame);
         }
-        $this->con->ack($frame, $this->transaction);
     }
 
-    protected function commit()
+    protected function commit($idx)
     {
         if ($this->useTransactions) {
-            if (!$this->transaction) {
+            if (empty($this->transaction[$idx])) {
                 throw new Exception("Tried to commit but not in a transaction");
             }
-            $this->con->commit($this->transaction);
-            $this->transaction = null;
+            $this->cons[$idx]->commit($this->transaction[$idx]);
+            $this->transaction[$idx] = null;
         }
     }
 
-    protected function rollback()
+    protected function rollback($idx)
     {
         if ($this->useTransactions) {
-            if (!$this->transaction) {
+            if (empty($this->transaction[$idx])) {
                 throw new Exception("Tried to rollback but not in a transaction");
             }
-            $this->con->commit($this->transaction);
-            $this->transaction = null;
+            $this->cons[$idx]->commit($this->transaction[$idx]);
+            $this->transaction[$idx] = null;
         }
     }
 }
diff --git a/lib/util.php b/lib/util.php
index 4312f9876..dd8189a58 100644
--- a/lib/util.php
+++ b/lib/util.php
@@ -178,7 +178,6 @@ function common_ensure_session()
 	}
 	if (isset($id)) {
 	    session_id($id);
-	    setcookie(session_name(), $id);
 	}
         @session_start();
         if (!isset($_SESSION['started'])) {