Queue is not processed

[akrus]

Hello!

We have a strange issue with this plugin - the queue itself seems to work fine, it has data in Redis (used all 2G of memory), but it's not inserted to database at all.

queuedtracking:print-queued-requests returns me a list of requests (but not full, just some of them)

And other output:

su www-data -c './console queuedtracking:monitor -vvv'

DEBUG [2015-10-29 08:41:08] UserSynchronizer::makeConfigured(): LDAP access synchronization not enabled.
DEBUG [2015-10-29 08:41:08] UserSynchronizer::makeConfigured: configuring with defaultSitesWithViewAccess =
Queue is enabled
Request sets in the queue will be processed automatically after a tracking request
Up to 1 workers will be used
Processor will start once there are at least 25 request sets in the queue
279662 (279662) request sets left in queue. 1.91G used memory (2.54G peak). 0 workers active. ^C
Session terminated, terminating shell... ...terminated.

su www-data -c './console queuedtracking:process -vvv'

DEBUG [2015-10-29 08:41:24] UserSynchronizer::makeConfigured(): LDAP access synchronization not enabled.
DEBUG [2015-10-29 08:41:24] UserSynchronizer::makeConfigured: configuring with defaultSitesWithViewAccess =
Starting to process request sets, this can take a while

---

This worker finished queue processing with 0req/s (0 requests in 0.00 seconds)

---

What's wrong with it? I've followed through the documentation several times and everything is correct...
Redis version is 3.0.4, php version is 5.5.9 (Ubuntu 14.04 LTS build), php-redis is 2.2.4.

[mattab]

Hi @akrus can you try to disable Ldap plugin and see if this changes anything?

[akrus]

Disabled, debug messages disappeared, but still no data is processed.

[mattab]

@akrus sorry but we don't know what could cause this issue. Does anyone else experience this?

[tsteur]

Are you using any other non-standard plugins apart from Ldap? If so, can you maybe disable them and try again? Also can you check your server logs if there are any errors?

Can you maybe have a look in the config (config/config.ini.php) whether [Tracker]record_statistics is set to 0? If so, it would explain why it does not actually insert them.

[akrus]

We have the following extra plugins:
AdvancedCampaignReporting
CustomAlerts
ReferrersManager

Error logs are empty, I checked them already :)

record_statistics is set to 1.

I'll try disabling plugins later and let you know.

[tsteur]

OK please let me know. I was hoping it would be the record_statistics. Can you maybe check the config file again and see if there's an entry with installation_in_progress? Also I presume a [database]username is set right?

[akrus]

Tried disabling plugins, nothing changed. installation_in_progress is not present and database credentials are configured.

[tsteur]

I wonder if it's maybe related with Redis 3.X. Can't think of anything else right now, it should work. By any chance can you try to use the latest Redis 2.X easily?

[tsteur]

I just updated from Redis 2.8 to latest 3.0.5 and it worked fine. Are you familiar with PHP programming language maybe?

[tsteur]

Or maybe you can provide us access to your server so we can debug? If so, feel free to send us credentials via email to hello at piwik.org

[akrus]

I can switch to 2.x, should I? I'm familiar with php, but I'd need some assistance from your side to debug this :) unfortunately I cannot provide access there :(

[tsteur]

Here is basically where the magic happens: https://github.com/piwik/plugin-QueuedTracking/blob/master/Queue/Processor.php#L64-L110

What I thought in the first place, is that for some reason it seams to stop here: https://github.com/piwik/plugin-QueuedTracking/blob/master/Queue/Processor.php#L68-L69

Maybe you can just add some var_dump messages to see whether it actually does something there like the following. Can you maybe replace the content of that process method with the one here and post the output?

    public function process(Tracker $tracker = null)
    {
        $tracker = $tracker ?: new Tracker();
var_dump('Process');
        if (!$tracker->shouldRecordStatistics()) {
var_dump('ignore');
            return $tracker;
        }

        $request = new RequestSet();
        $request->rememberEnvironment();

        $loops = 0;

        try {

            while ($queue = $this->queueManager->lockNext()) {
var_dump($loops);
                if ($loops > $this->numMaxBatchesToProcess) {
                    Common::printDebug('This worker processed ' . $loops . ' times, stopping now.');
                    break;
                } else {
                    $loops++;
                }

                $queuedRequestSets = $queue->getRequestSetsToProcess();
var_dump('Num Request Sets: ' . count($queuedRequestSets));
                if (!empty($queuedRequestSets)) {
                    $requestSetsToRetry = $this->processRequestSets($tracker, $queuedRequestSets);
var_dump('Num Request Sets to retry because they failed: ' . count($requestSetsToRetry));
                    $this->processRequestSets($tracker, $requestSetsToRetry);
                    $queue->markRequestSetsAsProcessed();
                }

                $this->queueManager->unlock();
            }

        } catch (Exception $e) {
var_dump('Exception: ' . $e->getMessage());
            Common::printDebug('Failed to process a request set: ' . $e->getMessage());

            $this->queueManager->unlock();
            $request->restoreEnvironment();
            throw $e;
        }

        $request->restoreEnvironment();
var_dump('Process end');
exit;
        return $tracker;
    }

[akrus]

Sorry for delay, was on holidays :) here is output:

./console queuedtracking:process

Starting to process request sets, this can take a while
string(7) "Process"
string(11) "Process end"

[tsteur]

Thanks for that. I have a suspect but let's see. Do you mind doing the same with another file? This time plugins/QueuedTracking/Queue/Manager.php ?

It would be the method lockNext() which should be around line 229:

    public function lockNext()
    {
var_dump('lockNext');
        $this->unlock();
var_dump('CurrentQueueId: ' . $this->currentQueueId);
var_dump('Num RequestsToProcess: ' . $this->numRequestsToProcessInBulk);
var_dump('Num QueuesAvailable: ' . $this->numQueuesAvailable);
        if ($this->currentQueueId < 0) {
            // we just want to avoid to always start looking for the queue at position 0
            $this->currentQueueId = rand(0, $this->numQueuesAvailable - 1);
        }
var_dump('Actual CurrentQueueId: ' . $this->currentQueueId);
        // here we look for all available queues whether at least one should and can be processed

        $start = $this->currentQueueId + 1; // this way we make sure to rotate through all queues
        $end   = $start + $this->numQueuesAvailable;

var_dump(sprintf('Start:End %s: %s', $start, $end));

        for (; $start < $end; $start++) {
            $this->currentQueueId = $start % $this->numQueuesAvailable;
var_dump('--Testing queue ' . $this->currentQueueId);
            $queue = $this->createQueue($this->currentQueueId);
var_dump('Num Requests To Process: ' . $queue->getNumberOfRequestSetsInQueue());
var_dump('getNumberOfRequestsToProcessAtSameTime: ' . $queue->getNumberOfRequestsToProcessAtSameTime());
var_dump('Should Process: ' . (int) $queue->shouldProcess());
var_dump('Get Num aquired locks: ' . $this->lock->getNumberOfAcquiredLocks());
            if ($queue->shouldProcess() && $this->lock->acquireLock($this->currentQueueId)) {
var_dump('using queue');
                return $queue;
            }
        }
    }

[akrus]

Here we are:

./console queuedtracking:process

Starting to process request sets, this can take a while
string(8) "lockNext"
string(18) "CurrentQueueId: -1"
string(25) "Num RequestsToProcess: 25"
string(22) "Num QueuesAvailable: 1"
string(24) "Actual CurrentQueueId: 0"
string(14) "Start:End 1: 2"
string(17) "--Testing queue 0"
string(31) "Num Requests To Process: 279662"
string(42) "getNumberOfRequestsToProcessAtSameTime: 25"
string(17) "Should Process: 1"
string(24) "Get Num aquired locks: 0"

---

This worker finished queue processing with 0req/s (0 requests in 0.14 seconds)

---

[tsteur]

Thx a lot. I will get back to you on Monday. As I expected there seems to be a problem with this command: https://github.com/piwik/plugin-QueuedTracking/blob/master/Queue/Backend/Redis.php#L127 (http://redis.io/commands/set set with EX & NX option which is also recommended to implement a locking pattern http://redis.io/commands/set#patterns ).

Do you know the version of the phpredis extension? You can possibly find out via echo phpinfo() in a PHP script or php -i in command line. Maybe php -i | grep phpredis works. I cannot test it right now.

[akrus]

Extension is 2.2.4, Redis is 3.0.4.

[tsteur]

Can you try to execute the following command on the server and post the output: redis-cli SET Queued1 "Value" NX EX 5? You might have to specify host and port on top like this: redis-cli -h 127.0.0.1 -p 6379.

Also can we debug the file plugins/QueuedTracking/Queue/Backend/Redis.php the method setIfNotExists this time?

    public function setIfNotExists($key, $value, $ttlInSeconds)
    {
var_dump('Key ' . $key);
var_dump('value  ' . $value);
var_dump('ttl  ' . $ttlInSeconds);
        $this->connectIfNeeded();

        $wasSet = $this->redis->set($key, $value, array('nx', 'ex' => $ttlInSeconds));
var_dump($wasSet);
        return $wasSet;
    }

[akrus]

Sure. Btw, upgraded to Redis 3.0.5, but nothing changed.

redis-cli SET Queued1 "Value" NX EX 5
OK

Starting to process request sets, this can take a while
string(23) "Key QueuedTrackingLock0"
string(19) "value 9662e3b03d0f"
string(7) "ttl 60"
bool(false)

---

This worker finished queue processing with 0req/s (0 requests in 0.00 seconds)

---

[tsteur]

OK your Redis server actually sends the correct response when executing redis-cli but phpredis extension seems to return a wrong value since this should have worked. Unfortunately, they don't have a changelog from the looks. Can you try to install the latest phpredis extension version which should be 2.2.7 see https://pecl.php.net/package/redis ? 2.2.4 seems to be > 2 years old. I couldn't find directly something related to this in the changelog but it might be still fixed.

[akrus]

Upgraded to 2.2.7, but still the same :(

[tsteur]

Thx. That's strange but I have one more idea. Can you execute the following commands and paste the output here?

redis-cli get "QueuedTrackingLock0"
redis-cli keys "QueuedTrackingLock*"

If any of these return a value you might try to execute (actually you can execute it either way) and test the process command afterwards if it works then.

redis-cli del "QueuedTrackingLock0"

[akrus]

Here we are.

$ redis-cli get "QueuedTrackingLock0"
(nil)
$ redis-cli keys "QueuedTrackingLock*"
(empty list or set)

[tsteur]

I'm running a bit out of ideas here.

the problem seems to be in the phpredis extension but same version works fine for me. The SET NX EX commands works fine (redis-cli SET Queued1 "Value" NX EX 5 ==> OK) but when phpredis executes it, it seems to fail for some reason and returns false. I tried to find a bug report for that but couldn't and I'm kinda surprised such a standard/basic command fails since other seem to work fine.

Let's try a last test to see if we can find some more information. Can you update the execute method in plugins/QueuedTracking/Commands/Process.php like this just temporarily to see which kinda commands work and which don't.

    protected function execute(InputInterface $input, OutputInterface $output)
    {
        $trackerEnvironment = new Environment('tracker');
        $trackerEnvironment->init();

        $settings = StaticContainer::get('Piwik\Plugins\QueuedTracking\Settings');
        $host     = $settings->redisHost->getValue();
        $port     = $settings->redisPort->getValue();
        $timeout  = $settings->redisTimeout->getValue();
        $password = $settings->redisPassword->getValue();
        $database = $settings->redisDatabase->getValue();

        $backend = Queue\Factory::makeBackend();
        $test = $backend->setIfNotExists('foo', 'bar', 1);
        var_dump($test);
        var_dump($backend->getServerVersion());

        $redis = new \Redis();
        $success = $redis->connect($host, $port, $timeout, null, 100);

        if ($success && !empty($password)) {
            var_dump('withpassword');
            $success = $redis->auth($password);
            var_dump($success);
        }

        if (!empty($database) || 0 === $database) {
            var_dump('with database');
            $redis->select($database);
        }

        var_dump($redis->set('testKey', 'value'));
        var_dump($redis->set('testKeyWithttl', 'value', 5));
        var_dump($redis->setnx('testnxkey', 'value'));
        var_dump($redis->setex('testexkey', 5, 'value'));
        var_dump($redis->set('testKeyWithNx', 'value', array('nx')));
        var_dump($redis->set('testKeyWithEx', 'value', array('ex' => 5)));

        var_dump($redis->del('testKey'));
        var_dump($redis->del('testKeyWithttl'));
        var_dump($redis->del('testnxkey'));
        var_dump($redis->del('testexkey'));
        var_dump($redis->del('testKeyWithNx'));
        var_dump($redis->del('testKeyWithEx'));


        $x = new \ReflectionExtension('redis');
        var_dump('Redis version: ' . $x->getVersion());
        var_dump('PHP version: ' . phpversion());
        var_dump('Connected: ' . (int) $redis->isConnected());
    }

In my case the output looks like this

bool(true)
string(5) "3.0.5"
string(13) "with database"
bool(true)
bool(true)
bool(true)
bool(true)
bool(true)
bool(true)
int(1)
int(1)
int(1)
int(1)
int(1)
int(1)
string(20) "Redis version: 2.2.7"
string(19) "PHP version: 5.5.28"
string(12) "Connected: 1"

[akrus]

Here :)

bool(false)
string(5) "3.0.5"
string(13) "with database"
bool(false)
bool(false)
bool(false)
bool(false)
bool(false)
bool(false)
int(0)
int(0)
int(0)
int(0)
int(0)
int(0)
string(20) "Redis version: 2.2.7"
string(30) "PHP version: 5.5.9-1ubuntu4.14"
string(12) "Connected: 1"

[tsteur]

Thanks for that. So it doesn't seem to set any value at all no matter what, at the same time it seems to add values to a list perfectly fine (rpush, lLen, ... commands). I'm running a bit out of ideas. Do you have some kind of special setup or did you make any changes in the configuration? Did you build everything yourself or did you use packages from repositories?

[tsteur]

Interesting is this one

1450765343.887508 [0 lua] "GET" "QueuedTrackingLock0"
1450765343.887521 [0 lua] "EXPIRE" "QueuedTrackingLock0" "20"
1450765363.930201 [0 127.0.0.1:56503] "EVAL" "if redis.call(\"GET\",KEYS[1]) == ARGV[1] then\n    return redis.call(\"EXPIRE\",KEYS[1], ARGV[2])\nelse\n    return 0\nend" "1" "QueuedTrackingLock0" "7de4cb630395" 
"20"
1450765363.930257 [0 lua] "GET" "QueuedTrackingLock0"
1450765363.930371 [0 127.0.0.1:56503] "GET" "QueuedTrackingLock0"
1450765363.968407 [0 127.0.0.1:56503] "EVAL" "if redis.call(\"GET\",KEYS[1]) == ARGV[1] then\n    return redis.call(\"DEL\",KEYS[1])\nelse\n    return 0\nend" "1" "QueuedTrackingLock0" "7de4cb630395"
1450765363.968460 [0 lua] "GET" "QueuedTrackingLock0"

At 1450765343 it sets the key to expire in 20 seconds and exactly 20.1 seconds later at 1450765363 it tries to expire the key again, notices the key does no longer exist (the lock expired) so it stops processing and roll backs the last tracked request set. So far the queue behaves actually as expected and maybe the queue is not the problem.

The question is why was there a break of 20 seconds. It usually takes like 100ms to insert a tracking request. Also I find it a bit weird that it is like exactly 20.1 seconds which almost feels like it was waiting for 20 seconds as it is so close to the 20 seconds. Do you use any custom plugins or have you modified the code in some ways? Wondering if there's eg a sleep(20) or similar somewhere.

It's a bit hard to debug without access to the server as it would be probably good to do a strace, have a look at the MySQL query log again etc to see what is actually happening. Not sure if you are familiar with strace etc?

[tsteur]

What you could also try would be to increase the 20 to eg 60 or more here: https://github.com/piwik/plugin-QueuedTracking/blob/0.2.4/Queue/Processor.php#L174 and here: https://github.com/piwik/plugin-QueuedTracking/blob/0.2.4/Queue/Processor.php#L161 and see how it behaves afterwards

[akrus]

Uhm, well, I changed it to 60 in both places and now it works... I don't have any plugin using Redis as I installed it specially for QueuedTracking (though it's also used for cache & session storage now).

[tsteur]

Are you using any other Piwik plugins? Also have you configured a different database for cache and session?

[akrus]

Yes I do, but as I tried disabling them also, I don't think they somehow affect this.

Yes, I configured different databases as it's mentioned in the configuration guide :)

Here's the list of all active plugins:

Actions
AdvancedCampaignReporting <-- tried disabling
Annotations
BulkTracking
CustomAlerts <-- tried disabling
CustomVariables
Dashboard
DevicePlugins
DevicesDetections
Ecommerce
Events
ExampleAPI
ExamplePlugin
ExampleRssWidget
Feedback
Goals
Heartbeat
ImageGraph
Live
LoginLdap <-- tried disabling
MobileMessaging
Monolog
MultiSites
Overlay
PrivacyManager
Provider
Referrers
ReferrersManager <-- tried disabling
Resolution
ScheduledReports
SegmentEditor
SEO
Transitions
UserCountry
UserCountryMap
UserLanguage
VisitFrequency
VisitorInterest
VisitsSummary
VisitTime
Widgetize

[tsteur]

Can you also try with 30 seconds instead of 60? If disabling all of the mentioned plugins doesn't have any affect there must be something else that makes it quite slow. I'll think about it but not sure what could make it that slow. Do you use the JavaScript tracker?

[akrus]

Tried 30 - not working, changed to 35 (and 40) and it started processing.

Yes, we use JS tracker.

[tsteur]

Is it processing the requests fast in general? Or is it adding more requests than it can process at the same time? You can monitor it with ./console queuedtracking:monitor. If it can't process fast enough, there must be something really slow but without having access to the actual system it is hard to tell what. Hope it works just fine with 40 seconds :)

[akrus]

Well, it works fine for some time, then stops, does nothing for a while and then continues (e.g. it goes fast from 20000 until 14000, then waits for 10 seconds, then continues from 14000 until 5000, again waits for 10 seconds and then continues until the end).
After all, I imported all the 270k records so it's much better now :)

[tsteur]

Sweet, thx for letting us know. Problem is after the next update you will have to update the timeout maybe again. I'm hoping the default 20 seconds timeout will work for you next time after all the records were processed now. I'm closing the issue and please let me know if you run into the issue again after an update. We'll then maybe need to increase the default timeout.

[woldra]

although this is already closed... we had similar issues from time to time. Debugging was quite a pain. But finally strace showed that process timed out caused by failed reverse DNS lookups (dropping instead of rejecting TCP Port 53). Maybe this helps...

[tsteur]

Did the failed DNS lookup take a long time? I wonder where we do DNS lookups

[woldra]

I think DNS lookups where triggered by "provider lookup".
For some IPs the answer was to big to fit in an UDP package so the resolver did a fallback to TCP. TCP was dropped on the firewall so it hung too long for process to finish. I added an iptables reject rule localy. No timeout, no problem ;)

[tsteur]

The Provider plugin should be disabled see https://github.com/piwik/plugin-QueuedTracking/blob/0.2.5/Commands/Process.php#L45 but it might not work anymore. I will create a new issue for this

[tsteur]

I tested it in #35 and Provider plugin should be disabled

[bodomic2]

Hi guys, I've just run in the same issue (on the very same system that akrus was initially reporting from) and could not recover by raising ttls as suggested.
I've tried all the suggestions provided here and still no luck.
Maybe you can advise me on how to process queue from Redis memory one by one?
./console queuedtracking:process does not do anything, queuedtracking:print-queued-requests gives some hope though.

[tsteur]

Maybe check if you can see anything in the server logs or so?

[bodomic2]

I'm sorry to repeat the research of akrus again but there is not much info in any logs or console debug messages. console just reports 0 workers and over 40k sets in queue, that's all.
If I increase Redis max memory, it continues to fill with incoming events without any progress to their processing.

[bhisecj]

Hello,

I faced the same issue. Solution:

In Matomo->General Settings->QueuedTracking set below value:

Number of requests that are processed in one batch=1.

It will process all pending request for that worker slowly.

[siva538]

While this is still searchable issue, we are continuing to observe this, due to the fact that, in case of any error in the processing step, the entire result would be represented incorrectly as 0 requests processed.

In the same thread the following command helped to debug further:

./console queuedtracking:process -vvv

It has generated a lot of output information, and fortunately we found the error -

DEBUG: Visit is known (IP = 192.168.0.0)
DEBUG: token_auth is authenticated in cache!
DEBUG: Failed to process a queued request setError query: SQLSTATE[22003]: Numeric value out of range: 1264 Out of range value for column 'visit_total_interactions' at row 1 In query: UPDATE matomo_log_visit SET idvisitor = ?, visit_last_action_time = ?, visit_exit_idaction_url = ?, visit_total_actions = visit_total_actions + 1 , visit_total_interactions = visit_total_interactions + 1 , visit_total_time = ? WHERE idsite = ? AND idvisit = ? Parameters: array (
DEBUG: 0 => 'o▒▒U▒▒',
DEBUG: 1 => '2018-09-18 08:24:13',
DEBUG: 2 => 8,
DEBUG: 3 => 56,
DEBUG: 4 => 3,
DEBUG: 5 => '13',
DEBUG: )
DEBUG: This worker processed 501 times, stopping now.

This worker finished queue processing with 0req/s (0 requests in 30.01 seconds)

@tsteur, can you please consider this request to increase the datatype of visit_total_interactions from smallint to int ?

[tsteur]

See matomo-org/matomo#11722

[brusch]

I'm having the same issue, but haven't found a solution yet to fix it.

This is the output of queuedtracking:test

Settings that will be used:
Backend: redis
NumQueueWorkers: 1
NumRequestsToProcess: 100
ProcessDuringTrackingRequest: 0
QueueEnabled: 1

Redis backend only settings (does not apply when using MySQL backend):
Host: 127.0.0.1
Port: 6379
Timeout: 0
Password:
Database: 10
UseSentinelBackend: 0
SentinelMasterName: mymaster

Version / stats:
PHP version: 7.2.16-1+0~20190307202415.17+stretch~1.gbpa7be82
Uname: Linux elements 4.9.0-8-amd64 #1 SMP Debian 4.9.144-3.1 (2019-02-19) x86_64
PHPRedis version: 4.2.0
Backend version: 3.2.6
Memory: array (
  'used_memory' => 76934360,
  'used_memory_human' => '73.37M',
  'used_memory_rss' => 80392192,
  'used_memory_rss_human' => '76.67M',
  'used_memory_peak' => 76934360,
  'used_memory_peak_human' => '73.37M',
  'total_system_memory' => 32949399552,
  'total_system_memory_human' => '30.69G',
  'used_memory_lua' => 36864,
  'used_memory_lua_human' => '36.00K',
  'maxmemory' => 2147483648,
  'maxmemory_human' => '2.00G',
  'maxmemory_policy' => 'noeviction',
  'mem_fragmentation_ratio' => 1.04,
  'mem_allocator' => 'jemalloc-3.6.0',
)
MaxMemory Eviction Policy config: noeviction
MaxMemory config: 2147483648

Performing some tests:
Redis is connected: 1
Connection works in general
Success for method set(testKey, value)
Success for method setnx(testnxkey, value)
Success for method setex(testexkey, 5, value)
Success for method set(testKeyWithNx, value, Array)
Success for method set(testKeyWithEx, value, Array)
Initial expire seems to be set correctly
setIfNotExists works fine
expireIfKeyHasValue seems to work fine
Extending expire seems to be set correctly
expireIfKeyHasValue correctly expires only when the value is correct
Expire is still set which is correct
deleteIfKeyHasValue seems to work fine
List feature seems to work fine

Done

Running queuedtracking:process -vvv does unfortunately give not much more information, output is just

Starting to process request sets, this can take a while

Then it sits there for a while and that's it.

I'm using latest Matomo 3.9.1 and all plugins are up to date. System integrity check also says everything is fine.

[tsteur]

Can you check if it prints some requests? ./console queuedtracking:print-queued-requests --queue-id=0 for example?

It might help to enable debugging logging for tracker: https://developer.matomo.org/api-reference/tracking-api#debugging-the-tracker

[brusch]

I guess I've found the problem, it seems that the database is already quite full and get's slow, I'm doing a cleanup now and see if it makes it any better.

[kwisatz]

While the symptoms might be the same, I'm thinking there are several distinct causes at play here. In my and @bhisecj 's case, the problem is simply that the number of items in the queue < Number of requests that are processed in one batch and thus won't start until you set it to a lower value (because it will wait for at least Number of requests that are processed in one batchto be present in the queue before it starts processing).

[RyanPriceDotCa]

As an update for anyone else that runs into this issue, I was experiencing the same issue and followed of the troubleshooting steps @tsteur recommended. Increasing the lock timeout time in Queue/Processor.php from 20 and 30 seconds to 60 seconds each appears to have the queue actually processing again.

We were up to over 550k items in the 4 queues and that appears to be dropping now. I hope that helps someone else too.