fix DBI connection, RT#39250

[freeside.git] / FS / bin / freeside-queued

diff --git a/FS/bin/freeside-queued b/FS/bin/freeside-queued
index 008616f..36871b2 100644 (file)
--- a/FS/bin/freeside-queued
+++ b/FS/bin/freeside-queued
@@ -1,7 +1,7 @@
 #!/usr/bin/perl -w
 
 use strict;
-use vars qw( $DEBUG $kids $max_kids %kids );
+use vars qw( $DEBUG $kids $max_kids $sleep_time %kids );
 use POSIX qw(:sys_wait_h);
 use IO::File;
 use Getopt::Std;
@@ -11,6 +11,9 @@ use FS::Conf;
 use FS::Record qw(qsearch);
 use FS::queue;
 use FS::queue_depend;
+use FS::queue_stat;
+use FS::Log;
+use FS::Cron::expire_user_pref qw( expire_user_pref );
 
 # no autoloading for non-FS classes...
 use Net::SSH 0.07;
@@ -45,6 +48,7 @@ while ( $@ ) {
   }
 }
 
+my $log = FS::Log->new('queue');
 logfile( "%%%FREESIDE_LOG%%%/queuelog.". $FS::UID::datasrc );
 
 warn "completing daemonization (detaching))\n" if $DEBUG;
@@ -54,6 +58,7 @@ daemonize2();
 
 my $conf = new FS::Conf;
 $max_kids = $conf->config('queued-max_kids') || 10;
+$sleep_time = $conf->config('queued-sleep_time') || 10;
 
 my $warnkids=0;
 while (1) {
@@ -63,6 +68,7 @@ while (1) {
   if ( $kids >= $max_kids ) {
     warn "WARNING: maximum $kids children reached\n" unless $warnkids++;
     &reap_kids;
+    expire_user_pref() unless $warnkids % 10;
     sleep 1; #waiting for signals is cheap
     next;
   }
@@ -102,17 +108,25 @@ while (1) {
 
   my $hashref = { 'status' => 'new' };
   if ( $opt{'s'} ) {
-    $hashref->{'status'} = 'Y';
+    $hashref->{'secure'} = 'Y';
   } elsif ( $opt{'n'} ) {
-    $hashref->{'status'} = '';
+    $hashref->{'secure'} = '';
   }
 
-  my @jobs = qsearch({
-    'table'     => 'queue',
-    'hashref'   => $hashref,
-    'extra_sql' => $nodepend,
-    'order_by'  => $order_by,
-  });
+  #qsearch dies when the db goes away
+  my @jobs = eval {
+    qsearch({
+      'table'     => 'queue',
+      'hashref'   => $hashref,
+      'extra_sql' => $nodepend,
+      'order_by'  => $order_by,
+    });
+  };
+  if ( $@ ) {
+    warn "WARNING: error searching for jobs, closing connection: $@";
+    undef $FS::UID::dbh;
+    next;
+  }
 
   unless ( @jobs ) {
     dbh->commit or do {
@@ -120,12 +134,17 @@ while (1) {
       undef $FS::UID::dbh;
       next;
     };
-    sleep 1;
+    expire_user_pref();
+    sleep $sleep_time;
     next;
   }
 
   foreach my $job ( @jobs ) {
 
+    my $start_date = time;
+
+    $log->debug('locking queue job', object => $job);
+
     my %hash = $job->hash;
     $hash{'status'} = 'locked';
     my $ljob = new FS::queue ( \%hash );
@@ -145,7 +164,12 @@ while (1) {
 
     $FS::UID::AutoCommit = 1;
 
-    my @args = $ljob->args;
+    my @args = eval { $ljob->args; };
+    if ( $@ ) {
+      warn "WARNING: error retrieving job arguments, closing connection: $@";
+      undef $FS::UID::dbh;
+      next;
+    }
     splice @args, 0, 1, $ljob if $args[0] eq '_JOB';
 
     defined( my $pid = fork ) or do {
@@ -155,7 +179,7 @@ while (1) {
       $hash{'statustext'} = "[freeside-queued] can't fork: $!";
       my $ljob = new FS::queue ( \%hash );
       my $error = $ljob->replace($job);
-      die $error if $error;
+      die $error if $error; #XXX still dying if we can't fork AND we can't connect to the db
       next; #don't increment the kid counter
     };
 
@@ -172,7 +196,7 @@ while (1) {
       dbh->{'private_profile'} = {} if UNIVERSAL::can(dbh, 'sprintProfile');
 
       #auto-use classes...
-      if (    $ljob->job =~ /(FS::(part_export|cust_main)::\w+)::/
+      if (    $ljob->job =~ /(FS::(part_export|cust_main|cust_pkg|part_pkg|Cron)::\w+)::/
            || $ljob->job =~ /(FS::\w+)::/
          )
       {
@@ -191,18 +215,46 @@ while (1) {
       }
 
       my $eval = "&". $ljob->job. '(@args);';
+      # don't put @args in the log, may expose passwords
+      $log->info('starting job ('.$ljob->job.')');
       warn 'running "&'. $ljob->job. '('. join(', ', @args). ")\n" if $DEBUG;
-      eval $eval; #throw away return value?  suppose so
+      local $FS::UID::AutoCommit = 0; # so that we can clean up failures
+      do {
+        # switch user only if a job user is available
+        local $FS::CurrentUser::CurrentUser = $ljob->access_user || $FS::CurrentUser::CurrentUser;
+        eval $eval; #throw away return value?  suppose so
+      };
       if ( $@ ) {
-        warn "job $eval failed";
+        dbh->rollback;
         my %hash = $ljob->hash;
-        $hash{'status'} = 'failed';
         $hash{'statustext'} = $@;
+        if ( $hash{'statustext'} =~ /\/misc\/queued_report/ ) { #use return?
+          $hash{'status'} = 'done'; 
+        } else {
+          $hash{'status'} = 'failed';
+          warn "job $eval failed";
+        }
         my $fjob = new FS::queue( \%hash );
         my $error = $fjob->replace($ljob);
         die $error if $error;
+        dbh->commit; # for the status change only
       } else {
         $ljob->delete;
+        dbh->commit; # for the job itself
+      }
+
+      if ( $ljob->job eq 'FS::cust_main::queued_bill' ) {
+        my $queue_stat = new FS::queue_stat {
+          'jobnum'      => $ljob->jobnum,
+          'job'         => $ljob->job,
+          'custnum'     => $ljob->custnum,
+          'insert_date' => $ljob->_date,
+          'start_date'  => $start_date,
+          'end_date'    => time,
+        };
+        my $error = $queue_stat->insert;
+        die $error if $error;
+        dbh->commit; #for the stat
       }
 
       if ( UNIVERSAL::can(dbh, 'sprintProfile') ) {
@@ -268,7 +320,7 @@ Job queue daemon.  Should be running at all times.
 
 -n: non-"secure" jobs only (other jobs)
 
-user: from the mapsecrets file - see config.html from the base documentation
+user: Typically "fs_queue"
 
 =head1 VERSION