Bug 284184: Allow Bugzilla to use an asynchronous job queue for sending mail.

Bugzilla.pm

@@ -42,6 +42,7 @@ use Bugzilla::Auth::Persist::Cookie;
 use Bugzilla::CGI;
 use Bugzilla::DB;
 use Bugzilla::Install::Localconfig qw(read_localconfig);
+use Bugzilla::JobQueue;
 use Bugzilla::Template;
 use Bugzilla::User;
 use Bugzilla::Error;
@@ -315,6 +316,12 @@ sub logout_request {
     # there. Don't rely on it: use Bugzilla->user->login instead!
 }
 
+sub job_queue {
+    my $class = shift;
+    $class->request_cache->{job_queue} ||= Bugzilla::JobQueue->new();
+    return $class->request_cache->{job_queue};
+}
+
 sub dbh {
     my $class = shift;
     # If we're not connected, then we must want the main db
@@ -714,4 +721,10 @@ Returns the local timezone of the Bugzilla installation,
 as a DateTime::TimeZone object. This detection is very time
 consuming, so we cache this information for future references.
 
+=item C<job_queue>
+
+Returns a L<Bugzilla::JobQueue> that you can use for queueing jobs.
+Will throw an error if job queueing is not correctly configured on
+this Bugzilla installation.
+
 =back

Bugzilla/Config/Common.pm

@@ -49,7 +49,7 @@ use base qw(Exporter);
        check_opsys check_shadowdb check_urlbase check_webdotbase
        check_netmask check_user_verify_class check_image_converter
        check_mail_delivery_method check_notification check_utf8
-       check_bug_status check_smtp_auth
+       check_bug_status check_smtp_auth check_theschwartz_available
 );
 
 # Checking functions for the various values
@@ -335,6 +335,15 @@ sub check_smtp_auth {
     return "";
 }
 
+sub check_theschwartz_available {
+    if (!eval { require TheSchwartz; require Daemon::Generic; }) {
+        return "Using the job queue requires that you have certain Perl"
+               . " modules installed. See the output of checksetup.pl"
+               . " for more information";
+    }
+    return "";
+}
+
 # OK, here are the parameter definitions themselves.
 #
 # Each definition is a hash with keys:

Bugzilla/Config/MTA.pm

@@ -58,6 +58,13 @@ sub get_param_list {
   },
 
   {
+   name => 'use_mailer_queue',
+   type => 'b',
+   default => 0,
+   checker => \&check_theschwartz_available,
+  },
+
+  {
    name => 'sendmailnow',
    type => 'b',
    default => 1
@@ -90,7 +97,6 @@ sub get_param_list {
    default => 7,
    checker => \&check_numeric
   },
-  
   {
    name => 'globalwatchers',
    type => 't',

Bugzilla/DB/Mysql.pm

@@ -63,7 +63,7 @@ sub new {
     my ($class, $user, $pass, $host, $dbname, $port, $sock) = @_;
 
     # construct the DSN from the parameters we got
-    my $dsn = "DBI:mysql:host=$host;database=$dbname";
+    my $dsn = "dbi:mysql:host=$host;database=$dbname";
     $dsn .= ";port=$port" if $port;
     $dsn .= ";mysql_socket=$sock" if $sock;
 
@@ -79,6 +79,9 @@ sub new {
     # a prefix 'private_'. See DBI documentation.
     $self->{private_bz_tables_locked} = "";
 
+    # Needed by TheSchwartz
+    $self->{private_bz_dsn} = $dsn;
+
     bless ($self, $class);
     
     # Bug 321645 - disable MySQL strict mode, if set

Bugzilla/DB/Oracle.pm

@@ -65,13 +65,15 @@ sub new {
     $ENV{'NLS_LANG'} = '.AL32UTF8' if Bugzilla->params->{'utf8'};
 
     # construct the DSN from the parameters we got
-    my $dsn = "DBI:Oracle:host=$host;sid=$dbname";
+    my $dsn = "dbi:Oracle:host=$host;sid=$dbname";
     $dsn .= ";port=$port" if $port;
     my $attrs = { FetchHashKeyName => 'NAME_lc',  
                   LongReadLen => ( Bugzilla->params->{'maxattachmentsize'}
                                      || 1000 ) * 1024, 
                 };
     my $self = $class->db_new($dsn, $user, $pass, $attrs);
+    # Needed by TheSchwartz
+    $self->{private_bz_dsn} = $dsn;
 
     bless ($self, $class);
 

Bugzilla/DB/Pg.pm

@@ -60,7 +60,7 @@ sub new {
     $dbname ||= 'template1';
 
     # construct the DSN from the parameters we got
-    my $dsn = "DBI:Pg:dbname=$dbname";
+    my $dsn = "dbi:Pg:dbname=$dbname";
     $dsn .= ";host=$host" if $host;
     $dsn .= ";port=$port" if $port;
 
@@ -75,6 +75,8 @@ sub new {
     # all class local variables stored in DBI derived class needs to have
     # a prefix 'private_'. See DBI documentation.
     $self->{private_bz_tables_locked} = "";
+    # Needed by TheSchwartz
+    $self->{private_bz_dsn} = $dsn;
 
     bless ($self, $class);
 

Bugzilla/DB/Schema.pm

@@ -1388,6 +1388,93 @@ use constant ABSTRACT_SCHEMA => {
         ],
      },
 
+    # THESCHWARTZ TABLES
+    # ------------------
+    # Note: In the standard TheSchwartz schema, most integers are unsigned,
+    # but we didn't implement unsigned ints for Bugzilla schemas, so we
+    # just create signed ints, which should be fine.
+
+    ts_funcmap => {
+        FIELDS => [
+            funcid   => {TYPE => 'INTSERIAL', PRIMARYKEY => 1, NOTNULL => 1},
+            funcname => {TYPE => 'varchar(255)', NOTNULL => 1},
+        ],
+        INDEXES => [
+            ts_funcmap_funcname_idx => {FIELDS => ['funcname'], 
+                                          TYPE => 'UNIQUE'},
+        ],
+    },
+
+    ts_job => {
+        FIELDS => [
+            # In a standard TheSchwartz schema, this is a BIGINT, but we
+            # don't have those and I didn't want to add them just for this.
+            jobid         => {TYPE => 'INTSERIAL', PRIMARYKEY => 1, 
+                              NOTNULL => 1},
+            funcid        => {TYPE => 'INT4', NOTNULL => 1},
+            # In standard TheSchwartz, this is a MEDIUMBLOB.
+            arg           => {TYPE => 'LONGBLOB'},
+            uniqkey       => {TYPE => 'varchar(255)'},
+            insert_time   => {TYPE => 'INT4'},
+            run_after     => {TYPE => 'INT4', NOTNULL => 1},
+            grabbed_until => {TYPE => 'INT4', NOTNULL => 1},
+            priority      => {TYPE => 'INT2'},
+            coalesce      => {TYPE => 'varchar(255)'},
+        ],
+        INDEXES => [
+            ts_job_funcid_idx => {FIELDS => [qw(funcid uniqkey)],
+                                  TYPE   => 'UNIQUE'},
+            # In a standard TheSchewartz schema, these both go in the other
+            # direction, but there's no reason to have three indexes that
+            # all start with the same column, and our naming scheme doesn't
+            # allow it anyhow.
+            ts_job_run_after_idx => [qw(run_after funcid)],
+            ts_job_coalesce_idx  => [qw(coalesce funcid)],
+        ],
+    },
+
+    ts_note => {
+        FIELDS => [
+            # This is a BIGINT in standard TheSchwartz schemas.
+            jobid   => {TYPE => 'INT4', NOTNULL => 1},
+            notekey => {TYPE => 'varchar(255)'},
+            value   => {TYPE => 'LONGBLOB'},
+        ],
+        INDEXES => [
+            ts_note_jobid_idx => {FIELDS => [qw(jobid notekey)], 
+                                    TYPE => 'UNIQUE'},
+        ],
+    },
+
+    ts_error => {
+        FIELDS => [
+            error_time => {TYPE => 'INT4', NOTNULL => 1},
+            jobid      => {TYPE => 'INT4', NOTNULL => 1},
+            message    => {TYPE => 'varchar(255)', NOTNULL => 1},
+            funcid     => {TYPE => 'INT4', NOTNULL => 1, DEFAULT => 0},
+        ],
+        INDEXES => [
+            ts_error_funcid_idx     => [qw(funcid error_time)],
+            ts_error_error_time_idx => ['error_time'],
+            ts_error_jobid_idx      => ['jobid'],
+        ],
+    },
+
+    ts_exitstatus => {
+        FIELDS => [
+            jobid           => {TYPE => 'INTSERIAL', PRIMARYKEY => 1,
+                                NOTNULL => 1},
+            funcid          => {TYPE => 'INT4', NOTNULL => 1, DEFAULT => 0},
+            status          => {TYPE => 'INT2'},
+            completion_time => {TYPE => 'INT4'},
+            delete_after    => {TYPE => 'INT4'},
+        ],
+        INDEXES => [
+            ts_exitstatus_funcid_idx       => ['funcid'],
+            ts_exitstatus_delete_after_idx => ['delete_after'],
+        ],
+    },
+
     # SCHEMA STORAGE
     # --------------
 

Bugzilla/Install/Filesystem.pm

@@ -114,6 +114,7 @@ sub FILESYSTEM {
         'customfield.pl'  => { perms => $owner_executable },
         'email_in.pl'     => { perms => $ws_executable },
         'sanitycheck.pl'  => { perms => $ws_executable },
+        'jobqueue.pl'     => { perms => $owner_executable },
         'install-module.pl' => { perms => $owner_executable },
 
         'docs/makedocs.pl'   => { perms => $owner_executable },

Bugzilla/Install/Requirements.pm

@@ -231,6 +231,20 @@ sub OPTIONAL_MODULES {
         feature => 'Inbound Email'
     },
 
+    # Mail Queueing
+    {
+        package => 'TheSchwartz',
+        module  => 'TheSchwartz',
+        version => 0,
+        feature => 'Mail Queueing',
+    },
+    {
+        package => 'Daemon-Generic',
+        module  => 'Daemon::Generic',
+        version => 0,
+        feature => 'Mail Queueing',
+    },
+
     # mod_perl
     {
         package => 'mod_perl',

Bugzilla/Job/Mailer.pm

+# -*- Mode: perl; indent-tabs-mode: nil -*-
+#
+# The contents of this file are subject to the Mozilla Public
+# License Version 1.1 (the "License"); you may not use this file
+# except in compliance with the License. You may obtain a copy of
+# the License at http://www.mozilla.org/MPL/
+#
+# Software distributed under the License is distributed on an "AS
+# IS" basis, WITHOUT WARRANTY OF ANY KIND, either express or
+# implied. See the License for the specific language governing
+# rights and limitations under the License.
+#
+# The Original Code is the Bugzilla Bug Tracking System.
+#
+# The Initial Developer of the Original Code is Mozilla Corporation.
+# Portions created by the Initial Developer are Copyright (C) 2008
+# Mozilla Corporation. All Rights Reserved.
+#
+# Contributor(s): 
+#   Mark Smith <mark@mozilla.com>
+#   Max Kanat-Alexander <mkanat@bugzilla.org>
+
+package Bugzilla::Job::Mailer;
+use Bugzilla::Mailer;
+BEGIN { eval "use base qw(TheSchwartz::Worker)"; }
+
+# The longest we expect a job to possibly take, in seconds.
+use constant grab_for => 300;
+# We don't want email to fail permanently very easily. Retry for 30 days.
+use constant max_retries => 725;
+
+# The first few retries happen quickly, but after that we wait an hour for
+# each retry.
+sub retry_delay {
+    my $num_retries = shift;
+    if ($num_retries < 5) {
+        return (10, 30, 60, 300, 600)[$num_retries];
+    }
+    # One hour
+    return 60*60;
+}
+
+sub work {
+    my ($class, $job) = @_;
+    my $msg = $job->arg->{msg};
+    my $success = eval { MessageToMTA($msg, 1); 1; };
+    if (!$success) {
+        $job->failed($@);
+        undef $@;
+    } 
+    else {
+        $job->completed;
+    }
+}
+
+1;

Bugzilla/JobQueue.pm

+# -*- Mode: perl; indent-tabs-mode: nil -*-
+#
+# The contents of this file are subject to the Mozilla Public
+# License Version 1.1 (the "License"); you may not use this file
+# except in compliance with the License. You may obtain a copy of
+# the License at http://www.mozilla.org/MPL/
+#
+# Software distributed under the License is distributed on an "AS
+# IS" basis, WITHOUT WARRANTY OF ANY KIND, either express or
+# implied. See the License for the specific language governing
+# rights and limitations under the License.
+#
+# The Original Code is the Bugzilla Bug Tracking System.
+#
+# The Initial Developer of the Original Code is Mozilla Corporation.
+# Portions created by the Initial Developer are Copyright (C) 2008
+# Mozilla Corporation. All Rights Reserved.
+#
+# Contributor(s): 
+#   Mark Smith <mark@mozilla.com>
+#   Max Kanat-Alexander <mkanat@bugzilla.org>
+
+package Bugzilla::JobQueue;
+
+use strict;
+
+use Bugzilla::Constants;
+use Bugzilla::Error;
+use Bugzilla::Install::Util qw(install_string);
+BEGIN { eval "use base qw(TheSchwartz)"; }
+
+# This maps job names for Bugzilla::JobQueue to the appropriate modules.
+# If you add new types of jobs, you should add a mapping here.
+use constant JOB_MAP => {
+    send_mail => 'Bugzilla::Job::Mailer',
+};
+
+sub new {
+    my $class = shift;
+
+    if (!eval { require TheSchwartz; }) {
+        ThrowCodeError('jobqueue_not_configured');
+    }
+
+    my $lc = Bugzilla->localconfig;
+    my $self = $class->SUPER::new(
+        databases => [{
+            dsn    => Bugzilla->dbh->{private_bz_dsn},
+            user   => $lc->{db_user},
+            pass   => $lc->{db_pass},
+            prefix => 'ts_',
+        }],
+    );
+
+    return $self;
+}
+
+# A way to get access to the underlying databases directly.
+sub bz_databases {
+    my $self = shift;
+    my @hashes = keys %{ $self->{databases} };
+    return map { $self->driver_for($_) } @hashes;
+}
+
+# inserts a job into the queue to be processed and returns immediately
+sub insert {
+    my $self = shift;
+    my $job = shift;
+
+    my $mapped_job = JOB_MAP->{$job};
+    ThrowCodeError('jobqueue_no_job_mapping', { job => $job })
+        if !$mapped_job;
+    unshift(@_, $mapped_job);
+
+    my $retval = $self->SUPER::insert(@_);
+    # XXX Need to get an error message here if insert fails, but
+    # I don't see any way to do that in TheSchwartz.
+    ThrowCodeError('jobqueue_insert_failed', { job => $job, errmsg => $@ })
+        if !$retval;
+ 
+    return $retval;
+}
+
+1;
+
+__END__
+
+=head1 NAME
+
+Bugzilla::JobQueue - Interface between Bugzilla and TheSchwartz.
+
+=head1 SYNOPSIS
+
+ use Bugzilla;
+
+ my $obj = Bugzilla->job_queue();
+ $obj->insert('send_mail', { msg => $message });
+
+=head1 DESCRIPTION
+
+Certain tasks should be done asyncronously.  The job queue system allows
+Bugzilla to use some sort of service to schedule jobs to happen asyncronously.
+
+=head2 Inserting a Job
+
+See the synopsis above for an easy to follow example on how to insert a
+job into the queue.  Give it a name and some arguments and the job will
+be sent away to be done later.

Bugzilla/JobQueue/Runner.pm

+# -*- Mode: perl; indent-tabs-mode: nil -*-
+#
+# The contents of this file are subject to the Mozilla Public
+# License Version 1.1 (the "License"); you may not use this file
+# except in compliance with the License. You may obtain a copy of
+# the License at http://www.mozilla.org/MPL/
+#
+# Software distributed under the License is distributed on an "AS
+# IS" basis, WITHOUT WARRANTY OF ANY KIND, either express or
+# implied. See the License for the specific language governing
+# rights and limitations under the License.
+#
+# The Original Code is the Bugzilla Bug Tracking System.
+#
+# The Initial Developer of the Original Code is Mozilla Corporation.
+# Portions created by the Initial Developer are Copyright (C) 2008
+# Mozilla Corporation. All Rights Reserved.
+#
+# Contributor(s): 
+#   Mark Smith <mark@mozilla.com>
+#   Max Kanat-Alexander <mkanat@bugzilla.org>
+
+# XXX In order to support Windows, we have to make gd_redirect_output
+# use Log4Perl or something instead of calling "logger". We probably
+# also need to use Win32::Daemon or something like that to daemonize.
+
+package Bugzilla::JobQueue::Runner;
+
+use strict;
+use File::Basename;
+use Pod::Usage;
+
+use Bugzilla::Constants;
+use Bugzilla::JobQueue;
+use Bugzilla::Util qw(get_text);
+BEGIN { eval "use base qw(Daemon::Generic)"; }
+
+# Required because of a bug in Daemon::Generic where it won't use the
+# "version" key from DAEMON_CONFIG.
+our $VERSION = BUGZILLA_VERSION;
+
+use constant DAEMON_CONFIG => (
+    progname => basename($0),
+    pidfile  => bz_locations()->{datadir} . '/' . basename($0) . '.pid',
+    version  => BUGZILLA_VERSION,
+);
+
+sub gd_preconfig {
+    return DAEMON_CONFIG;
+}
+
+sub gd_usage {
+    pod2usage({ -verbose => 0, -exitval => 'NOEXIT' });
+    return 0
+}
+
+sub gd_check {
+    my $self = shift;
+
+    # Get a count of all the jobs currently in the queue.
+    my $jq = Bugzilla->job_queue();
+    my @dbs = $jq->bz_databases();
+    my $count = 0;
+    foreach my $driver (@dbs) {
+        $count += $driver->select_one('SELECT COUNT(*) FROM ts_job', []);
+    }
+    print get_text('job_queue_depth', { count => $count }) . "\n";
+}
+
+sub gd_run {
+    my $self = shift;
+
+    my $jq = Bugzilla->job_queue();
+    $jq->set_verbose($self->{debug});
+    foreach my $module (values %{ Bugzilla::JobQueue::JOB_MAP() }) {
+        eval "use $module";
+        $jq->can_do($module);
+    }
+    $jq->work;
+}
+
+1;
+
+__END__
+
+=head1 NAME
+
+Bugzilla::JobQueue::Runner - A class representing the daemon that runs the
+job queue.
+
+ use Bugzilla::JobQueue::Runner;
+ Bugzilla::JobQueue::Runner->new();
+
+=head1 DESCRIPTION
+
+This is a subclass of L<Daemon::Generic> that is used by L<jobqueue>
+to run the Bugzilla job queue.

Bugzilla/Mailer.pm

@@ -53,10 +53,15 @@ use Email::MIME::Modifier;
 use Email::Send;
 
 sub MessageToMTA {
-    my ($msg) = (@_);
+    my ($msg, $send_now) = (@_);
     my $method = Bugzilla->params->{'mail_delivery_method'};
     return if $method eq 'None';
 
+    if (Bugzilla->params->{'use_mailer_queue'} and !$send_now) {
+        Bugzilla->job_queue->insert('send_mail', { msg => $msg });
+        return;
+    }
+
     my $email = ref($msg) ? $msg : Email::MIME->new($msg);
 
     # We add this header to uniquely identify all email that we

jobqueue.pl

+#!/usr/bin/perl -w
+# -*- Mode: perl; indent-tabs-mode: nil -*-
+#
+# The contents of this file are subject to the Mozilla Public
+# License Version 1.1 (the "License"); you may not use this file
+# except in compliance with the License. You may obtain a copy of
+# the License at http://www.mozilla.org/MPL/
+#
+# Software distributed under the License is distributed on an "AS
+# IS" basis, WITHOUT WARRANTY OF ANY KIND, either express or
+# implied. See the License for the specific language governing
+# rights and limitations under the License.
+#
+# The Original Code is the Bugzilla Bug Tracking System.
+#
+# The Initial Developer of the Original Code is Mozilla Corporation.
+# Portions created by the Initial Developer are Copyright (C) 2008
+# Mozilla Corporation. All Rights Reserved.
+#
+# Contributor(s): 
+#   Mark Smith <mark@mozilla.com>
+#   Max Kanat-Alexander <mkanat@bugzilla.org>
+
+use strict;
+use File::Basename;
+BEGIN { chdir dirname($0); }
+
+use lib qw(. lib);
+use Bugzilla;
+use Bugzilla::JobQueue::Runner;
+
+Bugzilla::JobQueue::Runner->new();
+
+=head1 NAME
+
+jobqueue.pl - Runs jobs in the background for Bugzilla.
+
+=head1 SYNOPSIS
+
+ ./jobqueue.pl [ -f ] [ -d ] { start | stop | restart | check | help | version } 
+
+   -f        Run in the foreground (don't detach)
+   -d        Output a lot of debugging information
+   start     Starts a new jobqueue daemon if there isn't one running already
+   stop	     Stops a running jobqueue daemon
+   restart   Stops a running jobqueue if one is running, and then
+             starts a new one.
+   check     Report the current status of the daemon.
+   help	     Display this usage info
+   version   Display the version of jobqueue.pl
+
+=head1 DESCRIPTION
+
+See L<Bugzilla::JobQueue> and L<Bugzilla::JobQueue::Runner>.

template/en/default/admin/params/mta.html.tmpl

@@ -45,6 +45,16 @@
   mailfrom => "The email address of the $terms.Bugzilla mail daemon.  Some email systems " _
               "require this to be a valid email address.",
 
+  use_mailer_queue => "In a large $terms.Bugzilla installation, updating"
+    _ " $terms.bugs can be very slow, because $terms.Bugzilla sends all"
+    _ " email at once. If you enable this parameter, $terms.Bugzilla will"
+    _ " queue all mail and then send it in the background. This requires"
+    _ " that you have installed certain Perl modules (as listed by"
+    _ " <code>checksetup.pl</code> for this feature), and that you are"
+    _ " running the <code>jobqueue.pl</code> daemon (otherwise your mail"
+    _ " won't get sent). This affects all mail sent by $terms.Bugzilla,"
+    _ " not just $terms.bug updates.",
+
   sendmailnow => "Sites using anything older than version 8.12 of 'sendmail' " _
                  "can achieve a significant performance increase in the " _
                  "UI -- at the cost of delaying the sending of mail -- by " _

template/en/default/global/code-error.html.tmpl

@@ -277,6 +277,21 @@
       given.
     [% END %]
 
+  [% ELSIF error == "jobqueue_insert_failed" %]
+   [% title = "Job Queue Failure" %]
+    Inserting a <code>[% job FILTER html %]</code> job into the Job 
+    Queue failed with the following error: [% errmsg FILTER html %]
+
+  [% ELSIF error == "jobqueue_not_configured" %]
+    Using the job queue system requires that certain Perl modules
+    be installed. Run <code>checksetup.pl</code> to see what modules
+    you are missing.
+
+  [% ELSIF error == "jobqueue_no_job_mapping" %]
+    <code>Bugzilla::JobQueue</code> has not been configured to handle
+    the job "[% job FILTER html %]".  You need to add this job type
+    to the <code>JOB_MAP</code> constant in <code>Bugzilla::JobQueue</code>.
+
   [% ELSIF error == "ldap_bind_failed" %]
     Failed to bind to the LDAP server. The error message was: 
     <code>[% errstr FILTER html %]</code>

template/en/default/global/messages.html.tmpl

@@ -451,6 +451,9 @@
       group.
     [% END %]
 
+  [% ELSIF message_tag == "job_queue_depth" %]
+    [% count FILTER html %] jobs in the queue.
+
   [% ELSIF message_tag == "keyword_created" %]
     [% title = "New Keyword Created" %]
     The keyword <em>[% name FILTER html %]</em> has been created.