rescue DeserializationError and mark job as failed

Closes collectiveidea#77

lib/delayed/backend/base.rb

@@ -100,6 +100,8 @@ def hook(name, *args)
           method = payload_object.method(name)
           method.arity == 0 ? method.call : method.call(self, *args)
         end
+      rescue DeserializationError
+        # do nothing
       end
 
       def reschedule_at

lib/delayed/backend/shared_spec.rb

@@ -333,6 +333,15 @@ def create_job(opts = {})
         end
       end
 
+      context "when the job raises a deserialization error" do
+        it "should mark the job as failed" do
+          Delayed::Worker.destroy_failed_jobs = false
+          job = described_class.create! :handler => "--- !ruby/object:JobThatDoesNotExist {}"
+          @worker.work_off
+          job.reload
+          job.failed_at.should_not be_nil
+        end
+      end
     end
 
     describe "failed jobs" do

lib/delayed/worker.rb

@@ -121,8 +121,11 @@ def run(job)
       end
       say "#{job.name} completed after %.4f" % runtime
       return true  # did work
-    rescue Exception => e
-      handle_failed_job(job, e)
+    rescue DeserializationError => error
+      job.last_error = "{#{error.message}\n#{error.backtrace.join('\n')}"
+      failed(job)
+    rescue Exception => error
+      handle_failed_job(job, error)
       return false  # work failed
     end
 
@@ -136,12 +139,16 @@ def reschedule(job, time = nil)
         job.save!
       else
         say "PERMANENTLY removing #{job.name} because of #{job.attempts} consecutive failures.", Logger::INFO
-        if job.respond_to?(:on_permanent_failure)
-          warn "[DEPRECATION] The #on_permanent_failure hook has been renamed to #failure."
-        end
-        job.hook(:failure)
-        self.class.destroy_failed_jobs ? job.destroy : job.update_attributes(:failed_at => Delayed::Job.db_time_now)
+        failed(job)
+      end
+    end
+
+    def failed(job)
+      job.hook(:failure)
+      if job.respond_to?(:on_permanent_failure)
+        warn "[DEPRECATION] The #on_permanent_failure hook has been renamed to #failure."
       end
+      self.class.destroy_failed_jobs ? job.destroy : job.update_attributes(:failed_at => Delayed::Job.db_time_now)
     end
 
     def say(text, level = Logger::INFO)
@@ -161,7 +168,7 @@ def handle_failed_job(job, error)
     # Run the next job we can get an exclusive lock on.
     # If no jobs are left we return nil
     def reserve_and_run_one_job
-      job = Delayed::Job.reserve(worker, self.class.max_run_time)
+      job = Delayed::Job.reserve(self, self.class.max_run_time)
       run(job) if job
     end
   end