Minion::Backend::PgExtra - A Pg Backend with Locks and Sequential Named Queues

pg_extra.t

use Mojo::Base -strict;

BEGIN { $ENV{MOJO_REACTOR} = 'Mojo::Reactor::Poll' }

use Test::More;

plan skip_all => 'set TEST_ONLINE to enable this test' unless $ENV{TEST_ONLINE};

use Minion;
use Mojo::IOLoop;
use Mojo::JSON 'true';
use Sys::Hostname 'hostname';
use Time::HiRes qw(time usleep);

# Isolate tests
require Mojo::Pg;
my $pg = Mojo::Pg->new($ENV{TEST_ONLINE});
$pg->db->query('drop schema if exists minion_test cascade');
$pg->db->query('create schema minion_test');
my $minion = Minion->new(PgExtra => $ENV{TEST_ONLINE});
$minion->backend->pg->search_path(['minion_test']);

# Nothing to repair
my $worker = $minion->repair->worker;
isa_ok $worker->minion->app, 'Mojolicious', 'has default application';

# Migrate up and down
is $minion->backend->pg->migrations->active, 1, 'active version is 1';
is $minion->backend->pg->migrations->migrate(0)->active, 0,
  'active version is 0';
is $minion->backend->pg->migrations->migrate->active, 1, 'active version is 1';

# Register and unregister
$worker->register;
like $worker->info->{started}, qr/^[\d.]+$/, 'has timestamp';
my $notified = $worker->info->{notified};
like $notified, qr/^[\d.]+$/, 'has timestamp';
my $id = $worker->id;
is $worker->register->id, $id, 'same id';
usleep 50000;
ok $worker->register->info->{notified} > $notified, 'new timestamp';
is $worker->unregister->info, undef, 'no information';
my $host = hostname;
is $worker->register->info->{host}, $host, 'right host';
is $worker->info->{pid}, $$, 'right pid';
is $worker->unregister->info, undef, 'no information';

# Repair missing worker
$minion->add_task(test => sub { });
my $worker2 = $minion->worker->register;
isnt $worker2->id, $worker->id, 'new id';
$id = $minion->enqueue('test');
my $job = $worker2->dequeue(0);
is $job->id, $id, 'right id';
is $worker2->info->{jobs}[0], $job->id, 'right id';
$id = $worker2->id;
undef $worker2;
is $job->info->{state}, 'active', 'job is still active';
ok !!$minion->backend->worker_info($id), 'is registered';
$minion->backend->pg->db->query(
  "update minion_workers
   set notified = now() - interval '1 second' * ? where id = ?",
  $minion->missing_after + 1, $id
);
$minion->repair;
ok !$minion->backend->worker_info($id), 'not registered';
like $job->info->{finished}, qr/^[\d.]+$/,       'has finished timestamp';
is $job->info->{state},      'failed',           'job is no longer active';
is $job->info->{result},     'Worker went away', 'right result';

# Repair abandoned job
$worker->register;
$id  = $minion->enqueue('test');
$job = $worker->dequeue(0);
is $job->id, $id, 'right id';
$worker->unregister;
$minion->repair;
is $job->info->{state},  'failed',           'job is no longer active';
is $job->info->{result}, 'Worker went away', 'right result';

# Repair old jobs
$worker->register;
$id = $minion->enqueue('test');
my $id2 = $minion->enqueue('test');
my $id3 = $minion->enqueue('test');
$worker->dequeue(0)->perform for 1 .. 3;
my $finished = $minion->backend->pg->db->query(
  'select extract(epoch from finished) as finished
   from minion_jobs
   where id = ?', $id2
)->hash->{finished};
$minion->backend->pg->db->query(
  'update minion_jobs set finished = to_timestamp(?) where id = ?',
  $finished - ($minion->remove_after + 1), $id2);
$finished = $minion->backend->pg->db->query(
  'select extract(epoch from finished) as finished
   from minion_jobs
   where id = ?', $id3
)->hash->{finished};
$minion->backend->pg->db->query(
  'update minion_jobs set finished = to_timestamp(?) where id = ?',
  $finished - ($minion->remove_after + 1), $id3);
$worker->unregister;
$minion->repair;
ok $minion->job($id), 'job has not been cleaned up';
ok !$minion->job($id2), 'job has been cleaned up';
ok !$minion->job($id3), 'job has been cleaned up';

# List workers
$worker  = $minion->worker->register;
$worker2 = $minion->worker->register;
my $batch = $minion->backend->list_workers(0, 10);
ok $batch->[0]{id},        'has id';
is $batch->[0]{host},      $host, 'right host';
is $batch->[0]{pid},       $$, 'right pid';
like $batch->[0]{started}, qr/^[\d.]+$/, 'has timestamp';
is $batch->[1]{host},      $host, 'right host';
is $batch->[1]{pid},       $$, 'right pid';
ok !$batch->[2], 'no more results';
$batch = $minion->backend->list_workers(0, 1);
is $batch->[0]{id}, $worker2->id, 'right id';
ok !$batch->[1], 'no more results';
$batch = $minion->backend->list_workers(1, 1);
is $batch->[0]{id}, $worker->id, 'right id';
ok !$batch->[1], 'no more results';
$worker->unregister;
$worker2->unregister;

# Reset
$minion->reset->repair;
ok !$minion->backend->pg->db->query(
  'select count(id) as count from minion_jobs')->hash->{count}, 'no jobs';
ok !$minion->backend->pg->db->query(
  'select count(id) as count from minion_workers')->hash->{count}, 'no workers';

# Stats
$minion->add_task(
  add => sub {
    my ($job, $first, $second) = @_;
    $job->finish({added => $first + $second});
  }
);
$minion->add_task(fail => sub { die "Intentional failure!\n" });
my $stats = $minion->stats;
is $stats->{active_workers},   0, 'no active workers';
is $stats->{inactive_workers}, 0, 'no inactive workers';
is $stats->{active_jobs},      0, 'no active jobs';
is $stats->{failed_jobs},      0, 'no failed jobs';
is $stats->{finished_jobs},    0, 'no finished jobs';
is $stats->{inactive_jobs},    0, 'no inactive jobs';
is $stats->{delayed_jobs},     0, 'no delayed jobs';
$worker = $minion->worker->register;
is $minion->stats->{inactive_workers}, 1, 'one inactive worker';
$minion->enqueue('fail');
$minion->enqueue('fail');
is $minion->stats->{inactive_jobs}, 2, 'two inactive jobs';
$job   = $worker->dequeue(0);
$stats = $minion->stats;
is $stats->{active_workers}, 1, 'one active worker';
is $stats->{active_jobs},    1, 'one active job';
is $stats->{inactive_jobs},  1, 'one inactive job';
$minion->enqueue('fail');
my $job2 = $worker->dequeue(0);
$stats = $minion->stats;
is $stats->{active_workers}, 1, 'one active worker';
is $stats->{active_jobs},    2, 'two active jobs';
is $stats->{inactive_jobs},  1, 'one inactive job';
ok $job2->finish, 'job finished';
ok $job->finish,  'job finished';
is $minion->stats->{finished_jobs}, 2, 'two finished jobs';
$job = $worker->dequeue(0);
ok $job->fail, 'job failed';
is $minion->stats->{failed_jobs}, 1, 'one failed job';
ok $job->retry, 'job retried';
is $minion->stats->{failed_jobs}, 0, 'no failed jobs';
ok $worker->dequeue(0)->finish(['works']), 'job finished';
$worker->unregister;
$stats = $minion->stats;
is $stats->{active_workers},   0, 'no active workers';
is $stats->{inactive_workers}, 0, 'no inactive workers';
is $stats->{active_jobs},      0, 'no active jobs';
is $stats->{failed_jobs},      0, 'no failed jobs';
is $stats->{finished_jobs},    3, 'three finished jobs';
is $stats->{inactive_jobs},    0, 'no inactive jobs';
is $stats->{delayed_jobs},     0, 'no delayed jobs';

# List jobs
$id = $minion->enqueue('add');
$batch = $minion->backend->list_jobs(0, 10);
ok $batch->[0]{id},        'has id';
is $batch->[0]{task},      'add', 'right task';
is $batch->[0]{state},     'inactive', 'right state';
is $batch->[0]{retries},   0, 'job has not been retried';
like $batch->[0]{created}, qr/^[\d.]+$/, 'has created timestamp';
is $batch->[1]{task},      'fail', 'right task';
is_deeply $batch->[1]{args}, [], 'right arguments';
is_deeply $batch->[1]{result}, ['works'], 'right result';
is $batch->[1]{state},      'finished',   'right state';
is $batch->[1]{priority},   0,            'right priority';
is $batch->[1]{retries},    1,            'job has been retried';
like $batch->[1]{created},  qr/^[\d.]+$/, 'has created timestamp';
like $batch->[1]{delayed},  qr/^[\d.]+$/, 'has delayed timestamp';
like $batch->[1]{finished}, qr/^[\d.]+$/, 'has finished timestamp';
like $batch->[1]{retried},  qr/^[\d.]+$/, 'has retried timestamp';
like $batch->[1]{started},  qr/^[\d.]+$/, 'has started timestamp';
is $batch->[2]{task},       'fail',       'right task';
is $batch->[2]{state},      'finished',   'right state';
is $batch->[2]{retries},    0,            'job has not been retried';
is $batch->[3]{task},       'fail',       'right task';
is $batch->[3]{state},      'finished',   'right state';
is $batch->[3]{retries},    0,            'job has not been retried';
ok !$batch->[4], 'no more results';
$batch = $minion->backend->list_jobs(0, 10, {state => 'inactive'});
is $batch->[0]{state},   'inactive', 'right state';
is $batch->[0]{retries}, 0,          'job has not been retried';
ok !$batch->[1], 'no more results';
$batch = $minion->backend->list_jobs(0, 10, {task => 'add'});
is $batch->[0]{task},    'add', 'right task';
is $batch->[0]{retries}, 0,     'job has not been retried';
ok !$batch->[1], 'no more results';
$batch = $minion->backend->list_jobs(0, 10, {queue => 'default'});
is $batch->[0]{queue}, 'default', 'right queue';
is $batch->[1]{queue}, 'default', 'right queue';
is $batch->[2]{queue}, 'default', 'right queue';
is $batch->[3]{queue}, 'default', 'right queue';
ok !$batch->[4], 'no more results';
$batch = $minion->backend->list_jobs(0, 10, {queue => 'does_not_exist'});
is_deeply $batch, [], 'no results';
$batch = $minion->backend->list_jobs(0, 1);
is $batch->[0]{state},   'inactive', 'right state';
is $batch->[0]{retries}, 0,          'job has not been retried';
ok !$batch->[1], 'no more results';
$batch = $minion->backend->list_jobs(1, 1);
is $batch->[0]{state},   'finished', 'right state';
is $batch->[0]{retries}, 1,          'job has been retried';
ok !$batch->[1], 'no more results';
ok $minion->job($id)->remove, 'job removed';

# Enqueue, dequeue and perform
is $minion->job(12345), undef, 'job does not exist';
$id = $minion->enqueue(add => [2, 2]);
ok $minion->job($id), 'job does exist';
my $info = $minion->job($id)->info;
is_deeply $info->{args}, [2, 2], 'right arguments';
is $info->{priority}, 0,          'right priority';
is $info->{state},    'inactive', 'right state';
$worker = $minion->worker;
is $worker->dequeue(0), undef, 'not registered';
ok !$minion->job($id)->info->{started}, 'no started timestamp';
$job = $worker->register->dequeue(0);
is $worker->info->{jobs}[0], $job->id, 'right job';
like $job->info->{created}, qr/^[\d.]+$/, 'has created timestamp';
like $job->info->{started}, qr/^[\d.]+$/, 'has started timestamp';
is_deeply $job->args, [2, 2], 'right arguments';
is $job->info->{state}, 'active', 'right state';
is $job->task,    'add', 'right task';
is $job->retries, 0,     'job has not been retried';
$id = $job->info->{worker};
is $minion->backend->worker_info($id)->{pid}, $$, 'right worker';
ok !$job->info->{finished}, 'no finished timestamp';
$job->perform;
is $worker->info->{jobs}[0], undef, 'no jobs';
like $job->info->{finished}, qr/^[\d.]+$/, 'has finished timestamp';
is_deeply $job->info->{result}, {added => 4}, 'right result';
is $job->info->{state}, 'finished', 'right state';
$worker->unregister;
$job = $minion->job($job->id);
is_deeply $job->args, [2, 2], 'right arguments';
is $job->retries, 0, 'job has not been retried';
is $job->info->{state}, 'finished', 'right state';
is $job->task, 'add', 'right task';

# Retry and remove
$id = $minion->enqueue(add => [5, 6]);
$job = $worker->register->dequeue(0);
is $job->info->{attempts}, 1, 'job will be attempted once';
is $job->info->{retries},  0, 'job has not been retried';
is $job->id, $id, 'right id';
ok $job->finish, 'job finished';
ok !$worker->dequeue(0), 'no more jobs';
$job = $minion->job($id);
ok !$job->info->{retried}, 'no retried timestamp';
ok $job->retry, 'job retried';
like $job->info->{retried}, qr/^[\d.]+$/, 'has retried timestamp';
is $job->info->{state},     'inactive',   'right state';
is $job->info->{retries},   1,            'job has been retried once';
$job = $worker->dequeue(0);
is $job->retries, 1, 'job has been retried once';
ok !$job->retry, 'job not retried';
is $job->id, $id, 'right id';
ok !$job->remove, 'job has not been removed';
ok $job->fail,  'job failed';
ok $job->retry, 'job retried';
is $job->info->{retries}, 2, 'job has been retried twice';
$job = $worker->dequeue(0);
is $job->info->{state}, 'active', 'right state';
ok $job->finish, 'job finished';
ok $job->remove, 'job has been removed';
is $job->info,   undef, 'no information';
$id = $minion->enqueue(add => [6, 5]);
$job = $minion->job($id);
is $job->info->{state},   'inactive', 'right state';
is $job->info->{retries}, 0,          'job has not been retried';
ok $job->retry, 'job retried';
is $job->info->{state},   'inactive', 'right state';
is $job->info->{retries}, 1,          'job has been retried once';
$job = $worker->dequeue(0);
is $job->id, $id, 'right id';
ok $job->fail,   'job failed';
ok $job->remove, 'job has been removed';
is $job->info,   undef, 'no information';
$id = $minion->enqueue(add => [5, 5]);
$job = $minion->job("$id");
ok $job->remove, 'job has been removed';
$worker->unregister;

# Jobs with priority
$minion->enqueue(add => [1, 2]);
$id = $minion->enqueue(add => [2, 4], {priority => 1});
$job = $worker->register->dequeue(0);
is $job->id, $id, 'right id';
is $job->info->{priority}, 1, 'right priority';
ok $job->finish, 'job finished';
isnt $worker->dequeue(0)->id, $id, 'different id';
$id = $minion->enqueue(add => [2, 5]);
$job = $worker->register->dequeue(0);
is $job->id, $id, 'right id';
is $job->info->{priority}, 0, 'right priority';
ok $job->finish, 'job finished';
ok $job->retry({priority => 100}), 'job retried with higher priority';
$job = $worker->dequeue(0);
is $job->id, $id, 'right id';
is $job->info->{retries},  1,   'job has been retried once';
is $job->info->{priority}, 100, 'high priority';
ok $job->finish, 'job finished';
ok $job->retry({priority => 0}), 'job retried with lower priority';
$job = $worker->dequeue(0);
is $job->id, $id, 'right id';
is $job->info->{retries},  2, 'job has been retried twice';
is $job->info->{priority}, 0, 'low priority';
ok $job->finish, 'job finished';
$worker->unregister;

# Delayed jobs
$id = $minion->enqueue(add => [2, 1] => {delay => 100});
is $minion->stats->{delayed_jobs}, 1, 'one delayed job';
is $worker->register->dequeue(0), undef, 'too early for job';
ok $minion->job($id)->info->{delayed} > time, 'delayed timestamp';
$minion->backend->pg->db->query(
  "update minion_jobs set delayed = now() - interval '1 day' where id = ?",
  $id);
$job = $worker->dequeue(0);
is $job->id, $id, 'right id';
like $job->info->{delayed}, qr/^[\d.]+$/, 'has delayed timestamp';
ok $job->finish, 'job finished';
ok $job->retry,  'job retried';
ok $minion->job($id)->info->{delayed} < time, 'no delayed timestamp';
ok $job->remove, 'job removed';
ok !$job->retry, 'job not retried';
$id = $minion->enqueue(add => [6, 9]);
$job = $worker->dequeue(0);
ok $job->info->{delayed} < time, 'no delayed timestamp';
ok $job->fail, 'job failed';
ok $job->retry({delay => 100}), 'job retried with delay';
is $job->info->{retries}, 1, 'job has been retried once';
ok $job->info->{delayed} > time, 'delayed timestamp';
ok $minion->job($id)->remove, 'job has been removed';
$worker->unregister;

# Events
my ($enqueue, $pid);
my $failed = $finished = 0;
$minion->once(enqueue => sub { $enqueue = pop });
$minion->once(
  worker => sub {
    my ($minion, $worker) = @_;
    $worker->on(
      dequeue => sub {
        my ($worker, $job) = @_;
        $job->on(failed   => sub { $failed++ });
        $job->on(finished => sub { $finished++ });
        $job->on(spawn    => sub { $pid = pop });
        $job->on(
          start => sub {
            my $job = shift;
            return unless $job->task eq 'switcheroo';
            $job->task('add')->args->[-1] += 1;
          }
        );
      }
    );
  }
);
$worker = $minion->worker->register;
$id = $minion->enqueue(add => [3, 3]);
is $enqueue, $id, 'enqueue event has been emitted';
$minion->enqueue(add => [4, 3]);
$job = $worker->dequeue(0);
is $failed,   0, 'failed event has not been emitted';
is $finished, 0, 'finished event has not been emitted';
my $result;
$job->on(finished => sub { $result = pop });
$job->finish('Everything is fine!');
$job->perform;
is $result,   'Everything is fine!', 'right result';
is $failed,   0,                     'failed event has not been emitted';
is $finished, 1,                     'finished event has been emitted once';
isnt $pid, $$, 'new process id';
$job = $worker->dequeue(0);
my $err;
$job->on(failed => sub { $err = pop });
$job->fail("test\n");
$job->fail;
is $err,      "test\n", 'right error';
is $failed,   1,        'failed event has been emitted once';
is $finished, 1,        'finished event has been emitted once';
$minion->add_task(switcheroo => sub { });
$minion->enqueue(switcheroo => [5, 3]);
$job = $worker->dequeue(0);
$job->perform;
is_deeply $job->info->{result}, {added => 9}, 'right result';
$worker->unregister;

# Queues
$id = $minion->enqueue(add => [100, 1]);
is $worker->register->dequeue(0 => {queues => ['test1']}), undef, 'wrong queue';
$job = $worker->dequeue(0);
is $job->id, $id, 'right id';
is $job->info->{queue}, 'default', 'right queue';
ok $job->finish, 'job finished';
$id = $minion->enqueue(add => [100, 3] => {queue => 'test1'});
is $worker->dequeue(0), undef, 'wrong queue';
$job = $worker->dequeue(0 => {queues => ['test1']});
is $job->id, $id, 'right id';
is $job->info->{queue}, 'test1', 'right queue';
ok $job->finish, 'job finished';
ok $job->retry({queue => 'test2'}), 'job retried';
$job = $worker->dequeue(0 => {queues => ['default', 'test2']});
is $job->id, $id, 'right id';
is $job->info->{queue}, 'test2', 'right queue';
ok $job->finish, 'job finished';
$worker->unregister;

# Failed jobs
$id = $minion->enqueue(add => [5, 6]);
$job = $worker->register->dequeue(0);
is $job->id, $id, 'right id';
is $job->info->{result}, undef, 'no result';
ok $job->fail, 'job failed';
ok !$job->finish, 'job not finished';
is $job->info->{state},  'failed',        'right state';
is $job->info->{result}, 'Unknown error', 'right result';
$id = $minion->enqueue(add => [6, 7]);
$job = $worker->dequeue(0);
is $job->id, $id, 'right id';
ok $job->fail('Something bad happened!'), 'job failed';
is $job->info->{state}, 'failed', 'right state';
is $job->info->{result}, 'Something bad happened!', 'right result';
$id  = $minion->enqueue('fail');
$job = $worker->dequeue(0);
is $job->id, $id, 'right id';
$job->perform;
is $job->info->{state}, 'failed', 'right state';
is $job->info->{result}, "Intentional failure!\n", 'right result';
$worker->unregister;

# Nested data structures
$minion->add_task(
  nested => sub {
    my ($job, $hash, $array) = @_;
    $job->finish([{23 => $hash->{first}[0]{second} x $array->[0][0]}]);
  }
);
$minion->enqueue(nested => [{first => [{second => 'test'}]}, [[3]]]);
$job = $worker->register->dequeue(0);
$job->perform;
is $job->info->{state}, 'finished', 'right state';
is_deeply $job->info->{result}, [{23 => 'testtesttest'}], 'right structure';
$worker->unregister;

# Perform job in a running event loop
$id = $minion->enqueue(add => [8, 9]);
Mojo::IOLoop->delay(sub { $minion->perform_jobs })->wait;
is $minion->job($id)->info->{state}, 'finished', 'right state';
is_deeply $minion->job($id)->info->{result}, {added => 17}, 'right result';

# Non-zero exit status
$minion->add_task(exit => sub { exit 1 });
$id  = $minion->enqueue('exit');
$job = $worker->register->dequeue(0);
is $job->id, $id, 'right id';
$job->perform;
is $job->info->{state}, 'failed', 'right state';
is $job->info->{result}, 'Non-zero exit status (1)', 'right result';
$worker->unregister;

# Multiple attempts while processing
is $minion->backoff->(0),  15,     'right result';
is $minion->backoff->(1),  16,     'right result';
is $minion->backoff->(2),  31,     'right result';
is $minion->backoff->(3),  96,     'right result';
is $minion->backoff->(4),  271,    'right result';
is $minion->backoff->(5),  640,    'right result';
is $minion->backoff->(25), 390640, 'right result';
$id = $minion->enqueue(exit => [] => {attempts => 2});
$job = $worker->register->dequeue(0);
is $job->id, $id, 'right id';
is $job->retries, 0, 'job has not been retried';
$job->perform;
is $job->info->{attempts}, 2,          'job will be attempted twice';
is $job->info->{state},    'inactive', 'right state';
is $job->info->{result}, 'Non-zero exit status (1)', 'right result';
ok $job->info->{retried} < $job->info->{delayed}, 'delayed timestamp';
$minion->backend->pg->db->query(
  'update minion_jobs set delayed = now() where id = ?', $id);
$job = $worker->register->dequeue(0);
is $job->id, $id, 'right id';
is $job->retries, 1, 'job has been retried once';
$job->perform;
is $job->info->{attempts}, 2,        'job will be attempted twice';
is $job->info->{state},    'failed', 'right state';
is $job->info->{result}, 'Non-zero exit status (1)', 'right result';
$worker->unregister;

# Multiple attempts during maintenance
$id = $minion->enqueue(exit => [] => {attempts => 2});
$job = $worker->register->dequeue(0);
is $job->id, $id, 'right id';
is $job->retries, 0, 'job has not been retried';
is $job->info->{attempts}, 2,        'job will be attempted twice';
is $job->info->{state},    'active', 'right state';
$worker->unregister;
$minion->repair;
is $job->info->{state},  'inactive',         'right state';
is $job->info->{result}, 'Worker went away', 'right result';
ok $job->info->{retried} < $job->info->{delayed}, 'delayed timestamp';
$minion->backend->pg->db->query(
  'update minion_jobs set delayed = now() where id = ?', $id);
$job = $worker->register->dequeue(0);
is $job->id, $id, 'right id';
is $job->retries, 1, 'job has been retried once';
$worker->unregister;
$minion->repair;
is $job->info->{state},  'failed',           'right state';
is $job->info->{result}, 'Worker went away', 'right result';

# A job needs to be dequeued again after a retry
$minion->add_task(restart => sub { });
$id  = $minion->enqueue('restart');
$job = $worker->register->dequeue(0);
is $job->id, $id, 'right id';
ok $job->finish, 'job finished';
is $job->info->{state}, 'finished', 'right state';
ok $job->retry, 'job retried';
is $job->info->{state}, 'inactive', 'right state';
$job2 = $worker->dequeue(0);
is $job->info->{state}, 'active', 'right state';
ok !$job->finish, 'job not finished';
is $job->info->{state}, 'active', 'right state';
is $job2->id, $id, 'right id';
ok $job2->finish, 'job finished';
ok !$job->retry, 'job not retried';
is $job->info->{state}, 'finished', 'right state';
$worker->unregister;

# Perform jobs concurrently
$id  = $minion->enqueue(add => [10, 11]);
$id2 = $minion->enqueue(add => [12, 13]);
$id3 = $minion->enqueue('test');
my $id4 = $minion->enqueue('exit');
$worker = $minion->worker->register;
$job    = $worker->dequeue(0);
$job2   = $worker->dequeue(0);
my $job3 = $worker->dequeue(0);
my $job4 = $worker->dequeue(0);
$pid = $job->start;
my $pid2 = $job2->start;
my $pid3 = $job3->start;
my $pid4 = $job4->start;
my ($first, $second, $third, $fourth);
usleep 50000
  until $first ||= $job->is_finished($pid)
  and $second  ||= $job2->is_finished($pid2)
  and $third   ||= $job3->is_finished($pid3)
  and $fourth  ||= $job4->is_finished($pid4);
is $minion->job($id)->info->{state}, 'finished', 'right state';
is_deeply $minion->job($id)->info->{result}, {added => 21}, 'right result';
is $minion->job($id2)->info->{state}, 'finished', 'right state';
is_deeply $minion->job($id2)->info->{result}, {added => 25}, 'right result';
is $minion->job($id3)->info->{state},  'finished', 'right state';
is $minion->job($id3)->info->{result}, undef,      'no result';
is $minion->job($id4)->info->{state},  'failed',   'right state';
is $minion->job($id4)->info->{result}, 'Non-zero exit status (1)',
  'right result';
$worker->unregister;

## PgExtra additional tests

# add resouces/tags
$minion->backend->add_resource(resource1 => [qw/tag1 tag2/]);
my $resource = $minion->backend->pg->db->query('select * from minion_resources where key=?', 'resource1')->hash;
is $resource->{key}, 'resource1', 'got correct resource';
is @{$resource->{tags}}, 2, 'correct number of tags';
my $tags = $minion->backend->pg->db->query('select id, name, to_json(enabled) as enabled from minion_resource_tags where id=any(?)', $resource->{tags})->expand->hashes;
is_deeply [sort map {$_->{name}} @$tags], [qw/tag1 tag2/], 'created the correct tags';
is_deeply [map {$_->{enabled}} @$tags], [true, true], 'tags default to enabled';

# adding a resource merges its tags
$minion->backend->add_resource(resource1 => [qw/tag3/]);
$resource = $minion->backend->pg->db->query('select * from minion_resources where key=?', 'resource1')->hash;
$tags = $minion->backend->pg->db->query('select id, name, to_json(enabled) as enabled from minion_resource_tags where id=any(?)', $resource->{tags})->expand->hashes;
is_deeply [sort map {$_->{name}} @$tags], [qw/tag1 tag2 tag3/], 'adding a tag merges it with existing ones';

# jobs with resources
$id  = $minion->enqueue('test', undef, {resource => 'resource1'});
$id2 = $minion->enqueue('test', undef, {resource => 'resource1'});
$worker = $minion->worker->register;
$job  = $worker->dequeue(0);
$job2 = $worker->dequeue(0);
is $job->id, $id, 'got correct job';
is $job2, undef, 'second job locked';
$job->finish;
$job2 = $worker->dequeue(0);
is $job2->id, $id2, 'got correct job';
$job2->finish;
$worker->unregister;

# Clean up once we are done
$pg->db->query('drop schema minion_test cascade');

done_testing();

PgExtra.pm

package Minion::Backend::PgExtra;
use Mojo::Base 'Minion::Backend';

use Carp 'croak';
use Mojo::IOLoop;
use Mojo::Pg 2.18;
use Sys::Hostname 'hostname';

has 'pg';

sub add_resource {
  my ($self, $name, $tags) = @_;
  $tags ||= [];

  if (@$tags) {
    $self->pg->db->query(<<'    SQL', $tags);
      insert into minion_resource_tags (name)
      values (unnest(?::text[]))
      on conflict do nothing
    SQL
  }

  $self->pg->db->query(<<'  SQL', $name, $tags);
    insert into minion_resources (key, tags)
    select ?, array_agg(t.id) from minion_resource_tags t
    where t.name = any(?)
    on conflict (key) do update
    set tags = (
      -- union of distinct elements of the existing and new tags
      select array_agg(id) from (
        select distinct unnest(minion_resources.tags || EXCLUDED.tags) id
        order by id
      ) g(id)
    )
  SQL
}

sub dequeue {
  my ($self, $id, $wait, $options) = @_;

  if ((my $job = $self->_try($id, $options))) { return $job }
  return undef if Mojo::IOLoop->is_running;

  my $db = $self->pg->db;
  $db->listen('minion.job')->on(notification => sub { Mojo::IOLoop->stop });
  my $timer = Mojo::IOLoop->timer($wait => sub { Mojo::IOLoop->stop });
  Mojo::IOLoop->start;
  $db->unlisten('*') and Mojo::IOLoop->remove($timer);
  undef $db;

  return $self->_try($id, $options);
}

sub enqueue {
  my ($self, $task, $args, $options) = (shift, shift, shift || [], shift || {});

  my $db = $self->pg->db;
  return $db->query(
    "insert into minion_jobs (args, attempts, delayed, priority, queue, task, resource)
     values (?, ?, (now() + (interval '1 second' * ?)), ?, ?, ?, (select id from minion_resources where key=?))
     returning id", {json => $args}, $options->{attempts} // 1,
    $options->{delay} // 0, $options->{priority} // 0,
    $options->{queue} // 'default', $task, $options->{resource}
  )->hash->{id};
}

sub fail_job   { shift->_update(1, @_) }
sub finish_job { shift->_update(0, @_) }

sub job_info {
  shift->pg->db->query(
    'select id, args, attempts, resource, parents,
       extract(epoch from created) as created,
       extract(epoch from delayed) as delayed,
       extract(epoch from finished) as finished, priority, queue, result,
       extract(epoch from retried) as retried, retries,
       extract(epoch from started) as started, state, task, worker
     from minion_jobs where id = ?', shift
  )->expand->hash;
}

sub list_jobs {
  my ($self, $offset, $limit, $options) = @_;

  return $self->pg->db->query(
    'select id from minion_jobs
     where (queue = $1 or $1 is null) and (state = $2 or $2 is null)
       and (task = $3 or $3 is null)
     order by id desc
     limit $4 offset $5', @$options{qw(queue state task)}, $limit, $offset
  )->arrays->map(sub { $self->job_info($_->[0]) })->to_array;
}

sub list_workers {
  my ($self, $offset, $limit) = @_;

  my $sql = 'select id from minion_workers order by id desc limit ? offset ?';
  return $self->pg->db->query($sql, $limit, $offset)
    ->arrays->map(sub { $self->worker_info($_->[0]) })->to_array;
}

sub new {
  my $self = shift->SUPER::new(pg => Mojo::Pg->new(@_));

  croak 'PostgreSQL 9.5 or later is required'
    if Mojo::Pg->new(@_)->db->dbh->{pg_server_version} < 90500;
  my $pg = $self->pg->auto_migrate(1)->max_connections(1);
  $pg->migrations->name('minion')->from_data;

  return $self;
}

sub register_worker {
  my ($self, $id) = @_;

  return shift->pg->db->query(
    "insert into minion_workers (id, host, pid)
     values (coalesce(?, nextval('minion_workers_id_seq')), ?, ?)
     on conflict(id) do update set notified = now()
     returning id", $id, $self->{host} //= hostname, $$
  )->hash->{id};
}

sub remove_job {
  !!shift->pg->db->query(
    "delete from minion_jobs
     where id = ? and state in ('inactive', 'failed', 'finished')
     returning 1", shift
  )->rows;
}

sub repair {
  my $self = shift;

  # Check worker registry
  my $db     = $self->pg->db;
  my $minion = $self->minion;
  $db->query(
    "delete from minion_workers
     where notified < now() - interval '1 second' * ?", $minion->missing_after
  );

  # Abandoned jobs
  my $fail = $db->query(
    "select id, retries from minion_jobs as j
     where state = 'active'
       and not exists(select 1 from minion_workers where id = j.worker)"
  )->hashes;
  $fail->each(sub { $self->fail_job(@$_{qw(id retries)}, 'Worker went away') });

  # Old jobs
  $db->query(
    "delete from minion_jobs
     where state = 'finished' and finished < now() - interval '1 second' * ?",
    $minion->remove_after
  );
}

sub reset { shift->pg->db->query('truncate minion_jobs, minion_workers') }

sub retry_job {
  my ($self, $id, $retries, $options) = (shift, shift, shift, shift || {});

  return !!$self->pg->db->query(
    "update minion_jobs
     set priority = coalesce(?, priority), queue = coalesce(?, queue),
       retried = now(), retries = retries + 1, state = 'inactive',
       delayed = (now() + (interval '1 second' * ?))
     where id = ? and retries = ?
       and state in ('inactive', 'failed', 'finished')
     returning 1", @$options{qw(priority queue)}, $options->{delay} // 0, $id,
    $retries
  )->rows;
}

sub stats {
  my $self = shift;

  my $stats = $self->pg->db->query(
    "select state::text || '_jobs', count(*) from minion_jobs group by state
     union all
     select 'delayed_jobs', count(*) from minion_jobs
     where state = 'inactive' and delayed > now()
     union all
     select 'inactive_workers', count(*) from minion_workers
     union all
     select 'active_workers', count(distinct worker) from minion_jobs
     where state = 'active'"
  )->arrays->reduce(sub { $a->{$b->[0]} = $b->[1]; $a }, {});
  $stats->{inactive_workers} -= $stats->{active_workers};
  $stats->{"${_}_jobs"} ||= 0 for qw(inactive active failed finished);

  return $stats;
}

sub unregister_worker {
  shift->pg->db->query('delete from minion_workers where id = ?', shift);
}

sub worker_info {
  shift->pg->db->query(
    "select id, extract(epoch from notified) as notified, array(
       select id from minion_jobs
       where state = 'active' and worker = minion_workers.id
     ) as jobs, host, pid, extract(epoch from started) as started
     from minion_workers
     where id = ?", shift
  )->hash;
}

sub _try {
  my ($self, $id, $options) = @_;

  my $sql = <<'  SQL';
    select * from minion_dequeue_job(?, ?, ?)
  SQL
  return $self->pg->db->query(
    $sql, $id, $options->{queues} || ['default'], [keys %{$self->minion->tasks}]
  )->expand->hash;
}

sub _update {
  my ($self, $fail, $id, $retries, $result) = @_;

  return undef unless my $row = $self->pg->db->query(
    "update minion_jobs
     set finished = now(), result = ?, state = ?
     where id = ? and retries = ? and state = 'active'
     returning attempts", {json => $result}, $fail ? 'failed' : 'finished',
    $id, $retries
  )->array;

  return 1 if !$fail || (my $attempts = $row->[0]) == 1;
  return 1 if $retries >= ($attempts - 1);
  my $delay = $self->minion->backoff->($retries);
  return $self->retry_job($id, $retries, {delay => $delay});
}

1;

=encoding utf8

=head1 NAME

Minion::Backend::Pg - PostgreSQL backend

=head1 SYNOPSIS

  use Minion::Backend::Pg;

  my $backend = Minion::Backend::Pg->new('postgresql://postgres@/test');

=head1 DESCRIPTION

L<Minion::Backend::Pg> is a backend for L<Minion> based on L<Mojo::Pg>. All
necessary tables will be created automatically with a set of migrations named
C<minion>. Note that this backend uses many bleeding edge features, so only the
latest, stable version of PostgreSQL is fully supported.

=head1 ATTRIBUTES

L<Minion::Backend::Pg> inherits all attributes from L<Minion::Backend> and
implements the following new ones.

=head2 pg

  my $pg   = $backend->pg;
  $backend = $backend->pg(Mojo::Pg->new);

L<Mojo::Pg> object used to store all data.

=head1 METHODS

L<Minion::Backend::Pg> inherits all methods from L<Minion::Backend> and
implements the following new ones.

=head2 dequeue

  my $job_info = $backend->dequeue($worker_id, 0.5);
  my $job_info = $backend->dequeue($worker_id, 0.5, {queues => ['important']});

Wait a given amount of time in seconds for a job, dequeue it and transition from
C<inactive> to C<active> state, or return C<undef> if queues were empty.

These options are currently available:

=over 2

=item queues

  queues => ['important']

One or more queues to dequeue jobs from, defaults to C<default>.

=back

These fields are currently available:

=over 2

=item args

  args => ['foo', 'bar']

Job arguments.

=item id

  id => '10023'

Job ID.

=item retries

  retries => 3

Number of times job has been retried.

=item task

  task => 'foo'

Task name.

=back

=head2 enqueue

  my $job_id = $backend->enqueue('foo');
  my $job_id = $backend->enqueue(foo => [@args]);
  my $job_id = $backend->enqueue(foo => [@args] => {priority => 1});

Enqueue a new job with C<inactive> state.

These options are currently available:

=over 2

=item attempts

  attempts => 25

Number of times performing this job will be attempted, with a delay based on
L<Minion/"backoff"> after the first attempt, defaults to C<1>.

=item delay

  delay => 10

Delay job for this many seconds (from now).

=item priority

  priority => 5

Job priority, defaults to C<0>.

=item queue

  queue => 'important'

Queue to put job in, defaults to C<default>.

=back

=head2 fail_job

  my $bool = $backend->fail_job($job_id, $retries);
  my $bool = $backend->fail_job($job_id, $retries, 'Something went wrong!');
  my $bool = $backend->fail_job(
    $job_id, $retries, {whatever => 'Something went wrong!'});

Transition from C<active> to C<failed> state, and if there are attempts
remaining, transition back to C<inactive> with a delay based on
L<Minion/"backoff">.

=head2 finish_job

  my $bool = $backend->finish_job($job_id, $retries);
  my $bool = $backend->finish_job($job_id, $retries, 'All went well!');
  my $bool = $backend->finish_job(
    $job_id, $retries, {whatever => 'All went well!'});

Transition from C<active> to C<finished> state.

=head2 job_info

  my $job_info = $backend->job_info($job_id);

Get information about a job, or return C<undef> if job does not exist.

  # Check job state
  my $state = $backend->job_info($job_id)->{state};

  # Get job result
  my $result = $backend->job_info($job_id)->{result};

These fields are currently available:

=over 2

=item args

  args => ['foo', 'bar']

Job arguments.

=item attempts

  attempts => 25

Number of times performing this job will be attempted.

=item created

  created => 784111777

Epoch time job was created.

=item delayed

  delayed => 784111777

Epoch time job was delayed to.

=item finished

  finished => 784111777

Epoch time job was finished.

=item priority

  priority => 3

Job priority.

=item queue

  queue => 'important'

Queue name.

=item result

  result => 'All went well!'

Job result.

=item retried

  retried => 784111777

Epoch time job has been retried.

=item retries

  retries => 3

Number of times job has been retried.

=item started

  started => 784111777

Epoch time job was started.

=item state

  state => 'inactive'

Current job state, usually C<active>, C<failed>, C<finished> or C<inactive>.

=item task

  task => 'foo'

Task name.

=item worker

  worker => '154'

Id of worker that is processing the job.

=back

=head2 list_jobs

  my $batch = $backend->list_jobs($offset, $limit);
  my $batch = $backend->list_jobs($offset, $limit, {state => 'inactive'});

Returns the same information as L</"job_info"> but in batches.

These options are currently available:

=over 2

=item queue

  queue => 'important'

List only jobs in this queue.

=item state

  state => 'inactive'

List only jobs in this state.

=item task

  task => 'test'

List only jobs for this task.

=back

=head2 list_workers

  my $batch = $backend->list_workers($offset, $limit);

Returns the same information as L</"worker_info"> but in batches.

=head2 new

  my $backend = Minion::Backend::Pg->new('postgresql://postgres@/test');

Construct a new L<Minion::Backend::Pg> object.

=head2 register_worker

  my $worker_id = $backend->register_worker;
  my $worker_id = $backend->register_worker($worker_id);

Register worker or send heartbeat to show that this worker is still alive.

=head2 remove_job

  my $bool = $backend->remove_job($job_id);

Remove C<failed>, C<finished> or C<inactive> job from queue.

=head2 repair

  $backend->repair;

Repair worker registry and job queue if necessary.

=head2 reset

  $backend->reset;

Reset job queue.

=head2 retry_job

  my $bool = $backend->retry_job($job_id, $retries);
  my $bool = $backend->retry_job($job_id, $retries, {delay => 10});

Transition from C<failed> or C<finished> state back to C<inactive>, already
C<inactive> jobs may also be retried to change options.

These options are currently available:

=over 2

=item delay

  delay => 10

Delay job for this many seconds (from now).

=item priority

  priority => 5

Job priority.

=item queue

  queue => 'important'

Queue to put job in.

=back

=head2 stats

  my $stats = $backend->stats;

Get statistics for jobs and workers.

These fields are currently available:

=over 2

=item active_jobs

  active_jobs => 100

Number of jobs in C<active> state.

=item active_workers

  active_workers => 100

Number of workers that are currently processing a job.

=item delayed_jobs

  delayed_jobs => 100

Number of jobs in C<inactive> state that are scheduled to run at specific time
in the future. Note that this field is EXPERIMENTAL and might change without
warning!

=item failed_jobs

  failed_jobs => 100

Number of jobs in C<failed> state.

=item finished_jobs

  finished_jobs => 100

Number of jobs in C<finished> state.

=item inactive_jobs

  inactive_jobs => 100

Number of jobs in C<inactive> state.

=item inactive_workers

  inactive_workers => 100

Number of workers that are currently not processing a job.

=back

=head2 unregister_worker

  $backend->unregister_worker($worker_id);

Unregister worker.

=head2 worker_info

  my $worker_info = $backend->worker_info($worker_id);

Get information about a worker, or return C<undef> if worker does not exist.

  # Check worker host
  my $host = $backend->worker_info($worker_id)->{host};

These fields are currently available:

=over 2

=item host

  host => 'localhost'

Worker host.

=item jobs

  jobs => ['10023', '10024', '10025', '10029']

Ids of jobs the worker is currently processing.

=item notified

  notified => 784111777

Epoch time worker sent the last heartbeat.

=item pid

  pid => 12345

Process id of worker.

=item started

  started => 784111777

Epoch time worker was started.

=back

=head1 SEE ALSO

L<Minion>, L<Mojolicious::Guides>, L<http://mojolicious.org>.

=cut

__DATA__

@@ minion
-- 1 up
create table if not exists minion_resource_tags (
  id bigserial primary key,
  name text not null,
  enabled boolean default true
);
create table if not exists minion_resources (
  id bigserial primary key,
  locked_by bigint, -- references minion_jobs(id),
  key text not null unique,
  tags bigint[] not null default '{}'::bigint[]
);
create type minion_state as enum ('inactive', 'active', 'failed', 'finished');
create table if not exists minion_jobs (
  id       bigserial not null primary key,
  args     jsonb not null,
  attempts int not null default 1,
  created  timestamp with time zone not null default now(),
  delayed  timestamp with time zone not null,
  finished timestamp with time zone,
  parents  bigint[] not null default '{}'::bigint[],
  priority int not null,
  queue    text not null default 'default',
  resource bigint references minion_resources(id),
  result   jsonb,
  retried  timestamp with time zone,
  retries  int not null default 0,
  started  timestamp with time zone,
  state    minion_state not null default 'inactive'::minion_state,
  task     text not null,
  worker   bigint,
  check(jsonb_typeof(args) = 'array')
);
create index on minion_jobs (state, priority desc, created);
create table if not exists minion_workers (
  id       bigserial not null primary key,
  host     text not null,
  notified timestamp with time zone not null default now(),
  pid      int not null,
  started  timestamp with time zone not null default now()
);
create or replace function minion_jobs_notify_workers() returns trigger as $$
  begin
    if new.delayed <= now() then
      notify "minion.job";
    end if;
    return null;
  end;
$$ language plpgsql;
set client_min_messages to warning;
drop trigger if exists minion_jobs_notify_workers_trigger on minion_jobs;
set client_min_messages to notice;
create trigger minion_jobs_notify_workers_trigger
  after insert or update of retries on minion_jobs
  for each row execute procedure minion_jobs_notify_workers();

create or replace function
  minion_dequeue_job(worker_id bigint, queues text[], tasks text[])
  returns table(
    id bigint,
    args jsonb,
    retries int,
    task text,
    lock bigint
  )
  as $$
  declare
    job record;
    resource record;
    job_id bigint;
  begin
    <<jobloop>>
    for job in
      -- select a candidate job
      select * from minion_jobs j
      where
        j.delayed <= now()
        and j.queue = any(queues)
        and j.state = 'inactive'
        and j.task = any(tasks)
        -- check for unfinished parent jobs
        and not exists (
          select 1 from minion_jobs p
          where p.id = any(j.parents) and p.state != 'finished'
        )
      order by priority desc, created
      for update
    loop
      -- if the job does not depend on a resource lock then it is good
      if job.resource is null then
        job_id := job.id;
        exit jobloop;
      end if;

      -- check if the resource is available
      select * from minion_resources r
      into resource
      where
        r.id = job.resource
        and not exists (
          select 1 from minion_resource_tags t
          where t.id = any(r.tags) and not t.enabled
        )
        and (
          locked_by is null
          or not exists (
            select 1 from minion_jobs l
            where l.id=r.locked_by and l.state='active'
          )
        )
      limit 1
      for update;
      continue jobloop when not found;

      -- lock the resource and end the candidate loop
      job_id := job.id;
      update minion_resources set locked_by = job_id;
      exit jobloop;
    end loop;

    -- if a candidate was found then mark it as dequeued
    if job_id is not null then
      return query
      update minion_jobs j
      set started = now(), state = 'active', worker = worker_id
      where j.id = job_id
      returning
        j.id,
        j.args,
        j.retries,
        j.task,
        j.resource as lock;
    end if;
  end;
$$ language plpgsql

-- 1 down
drop table if exists minion_jobs;
drop table if exists minion_workers;
drop table if exists minion_resources;
drop table if exists minion_resource_tags;
drop type if exists minion_state;
drop trigger if exists minion_jobs_notify_workers_trigger on minion_jobs;
drop function if exists minion_jobs_notify_workers();