mat/rpc_queue_connection_error_handling

lib/Mojo/WebSocketProxy/Backend/JobAsync.pm

@@ -13,6 +13,9 @@ use Job::Async;
 use JSON::MaybeUTF8 qw(encode_json_utf8 decode_json_utf8);
 use Log::Any qw($log);
 use MojoX::JSON::RPC::Client;
+use Syntax::Keyword::Try;
+
+my $PRC_QUEUE_TIMEOUT = $ENV{PRC_QUEUE_TIMEOUT} // 30;
 
 ## VERSION
 
@@ -107,9 +110,20 @@ sub call_rpc {
         # sequences in forked workers.
         Math::Random::Secure::srand() if Math::Random::Secure->can('srand');
         my $client_job = $jobman->client(redis => $self->{redis});
-        $client_job->start->retain;
         $client_job;
     };
+    # restart client at startup and after failure
+    $self->{start_client_future} //= do {
+        $self->client->start->on_done(
+            sub {
+                $log->infof('Queue backend client started successfully');
+            }
+            )->on_fail(
+            sub {
+                my $exception = shift;
+                $log->errorf('Queue backend client failed to start: %s', $exception);
+            });
+    };
 
     $req_storage->{call_params} ||= {};
     my $rpc_response_cb = $self->get_rpc_response_cb($c, $req_storage);
@@ -120,10 +134,16 @@ sub call_rpc {
     my $params = $self->make_call_params($c, $req_storage);
     $log->debugf("method %s has params = %s", $method, $params);
     $_->($c, $req_storage) for @$before_call_hook;
-    $self->client->submit(
-        name   => $req_storage->{name},
-        params => encode_json_utf8($params)
-        )->on_ready(
+
+    my $submit_future = $self->{start_client_future}->then(
+        sub {
+            $self->client->submit(
+                name   => $req_storage->{name},
+                params => encode_json_utf8($params));
+        });
+
+    my $timeout_future = Future::Mojo->new_timer($PRC_QUEUE_TIMEOUT)->then(sub { Future->fail('rpc queue timeout') });
+    Future->wait_any($submit_future, $timeout_future)->on_ready(
         sub {
             my ($f) = @_;
             $log->debugf('->submit completion: ', $f->state);
@@ -137,19 +157,35 @@ sub call_rpc {
             my $api_response;
 
             if ($f->is_done) {
-                my $result = MojoX::JSON::RPC::Client::ReturnObject->new(rpc_response => decode_json_utf8($f->get));
-
-                $_->($c, $req_storage, $result) for @$after_got_rpc_response_hook;
-
-                $api_response = $rpc_response_cb->($result->result);
-                stats_inc("rpc_queue.client.jobs.success", {tags => ["rpc:" . $req_storage->{name}, 'clientID:' . $self->client->id]});
+                try {
+                    my $result = MojoX::JSON::RPC::Client::ReturnObject->new(rpc_response => decode_json_utf8($f->get));
+
+                    $_->($c, $req_storage, $result) for @$after_got_rpc_response_hook;
+
+                    $api_response = $rpc_response_cb->($result->result);
+                    stats_inc("rpc_queue.client.jobs.success", {tags => ["rpc:" . $req_storage->{name}, 'clientID:' . $self->client->id]});
+                }
+                catch {
+                    $log->warnf("Failed to process response of method %s: %s", $method, $_);
+                    stats_inc("rpc_queue.client.jobs.fail",
+                        {tags => ["rpc:" . $req_storage->{name}, 'clientID:' . $self->client->id, 'error:' . $_]});
+                    $api_response = $c->wsp_error($msg_type, 'WrongResponse', 'Sorry, an error occurred while processing your request.');
+                };
             } else {
-                my ($failure) = $f->failure;
-                $log->warnf("method %s failed: %s", $method, $failure);
+                my $failure = $f->is_failed ? $f->failure // '' : 'request was cancelled';
+
+                $log->warnf("Method %s failed: %s", $method, $failure);
                 stats_inc("rpc_queue.client.jobs.fail",
                     {tags => ["rpc:" . $req_storage->{name}, 'clientID:' . $self->client->id, 'error:' . $failure]});
 
                 $api_response = $c->wsp_error($msg_type, 'WrongResponse', 'Sorry, an error occurred while processing your request.');
+
+                # force client to restart on failure (if it is not already restarting)
+                my $start_future = $self->{start_client_future};
+                if ( $start_future and ($start_future->state ne 'pending')) {
+                    $log->warnf('Client is going to restart after failure');
+                    delete $self->{start_client_future};
+                }
             }
 
             return unless $api_response;

t/25-job-async-backend.t

@@ -12,6 +12,8 @@ use IO::Async::Loop::Mojo;
 
 our @PENDING_JOBS;
 
+$ENV{PRC_QUEUE_TIMEOUT} = 2;
+
 my $loop = IO::Async::Loop::Mojo->new;
 our $LAST_ID = 999;
 package t::SampleClient {