Merge branch '1-x-stable'

Conflicts:
	examples/monit/resque.monit
	lib/resque/helpers.rb
	test/worker_test.rb

Author: hone

HISTORY.md

@@ -1,3 +1,13 @@
+## 1.21.0 (2012-07-02)
+
+* Add a flag to make sure failure hooks are only ran once (jakemack, #546)
+* Support updated MultiJSON API (@twinturbo)
+* Fix worker logging in monit example config (@twinturbo)
+* loosen dependency of redis-namespace to 1.x, support for redis-rb 3.0.x
+* change '%' to '$' in the 'stop program' command for monit
+* UTF8 sanitize exception messages when there's a failure (@brianmario, #507)
+* don't share a redis connection between parent and child (@jsanders, #588)
+
 ## 1.20.0 (2012-02-17)
 
 * Fixed demos for ruby 1.9 (@BMorearty, #445)

examples/monit/resque.monit

@@ -1,6 +1,6 @@
 check process resque_worker_QUEUE
   with pidfile /data/APP_NAME/current/tmp/pids/resque_worker_QUEUE.pid
   start program = "/usr/bin/env HOME=/home/user RACK_ENV=production PATH=/usr/local/bin:/usr/local/ruby/bin:/usr/bin:/bin:$PATH /bin/sh -l -c 'cd /data/APP_NAME/current; nohup bundle exec rake environment resque:work RAILS_ENV=production QUEUE=queue_name VERBOSE=1 PIDFILE=tmp/pids/resque_worker_QUEUE.pid >> log/resque_worker_QUEUE.log 2>&1'" as uid deploy and gid deploy
-  stop program = "/bin/sh -c 'cd /data/APP_NAME/current && kill -9 %(cat tmp/pids/resque_worker_QUEUE.pid) && rm -f tmp/pids/resque_worker_QUEUE.pid; exit 0;'"
+  stop program = "/bin/sh -c 'cd /data/APP_NAME/current && kill -9 $(cat tmp/pids/resque_worker_QUEUE.pid) && rm -f tmp/pids/resque_worker_QUEUE.pid; exit 0;'"
   if totalmem is greater than 300 MB for 10 cycles then restart  # eating up memory?
-  group resque_workers
+  group resque_workers

lib/resque.rb

@@ -17,6 +17,8 @@
 require 'resque/coder'
 require 'resque/multi_json_coder'
 
+require 'resque/vendor/utf8_util'
+
 module Resque
   include Helpers
   extend self

lib/resque/errors.rb

@@ -7,4 +7,7 @@ class NoClassError < RuntimeError; end
 
   # Raised when a worker was killed while processing a job.
   class DirtyExit < RuntimeError; end
+
+  # Raised when child process is TERM'd so job can rescue this to do shutdown work.
+  class TermException < SignalException; end
 end

lib/resque/failure/redis.rb

@@ -8,7 +8,7 @@ def save
           :failed_at => Time.now.strftime("%Y/%m/%d %H:%M:%S %Z"),
           :payload   => payload,
           :exception => exception.class.to_s,
-          :error     => exception.to_s,
+          :error     => UTF8Util.clean(exception.to_s),
           :backtrace => filter_backtrace(Array(exception.backtrace)),
           :worker    => worker.to_s,
           :queue     => queue

lib/resque/job.rb

@@ -32,6 +32,7 @@ class Job
     def initialize(queue, payload)
       @queue = queue
       @payload = payload
+      @failure_hooks_ran = false
     end
 
     # Creates a job by placing it on a queue. Expects a string queue
@@ -43,7 +44,9 @@ def self.create(queue, klass, *args)
       Resque.validate(klass, queue)
 
       if Resque.inline?
-        constantize(klass).perform(*decode(encode(args)))
+        # Instantiating a Resque::Job and calling perform on it so callbacks run
+        # decode(encode(args)) to ensure that args are normalized in the same manner as a non-inline job
+        new(:inline, {'class' => klass, 'args' => decode(encode(args))}).perform
       else
         Resque.push(queue, :class => klass.to_s, :args => args)
       end
@@ -210,14 +213,17 @@ def after_hooks
       @after_hooks ||= Plugin.after_hooks(payload_class)
     end
 
-    def failure_hooks 
+    def failure_hooks
       @failure_hooks ||= Plugin.failure_hooks(payload_class)
     end
-    
+
     def run_failure_hooks(exception)
-      job_args = args || []
-      failure_hooks.each { |hook| payload_class.send(hook, exception, *job_args) }
+      begin
+        job_args = args || []
+        failure_hooks.each { |hook| payload_class.send(hook, exception, *job_args) } unless @failure_hooks_ran
+      ensure
+        @failure_hooks_ran = true
+      end
     end
-
   end
 end

lib/resque/tasks.rb

@@ -14,6 +14,8 @@
       worker = Resque::Worker.new(*queues)
       worker.verbose = ENV['LOGGING'] || ENV['VERBOSE']
       worker.very_verbose = ENV['VVERBOSE']
+      worker.term_timeout = ENV['RESQUE_TERM_TIMEOUT'] || 4.0
+      worker.term_child = ENV['TERM_CHILD']
     rescue Resque::NoQueueError
       abort "set QUEUE env var, e.g. $ QUEUE=critical,high rake resque:work"
     end

lib/resque/vendor/utf8_util.rb

@@ -0,0 +1,26 @@
+module UTF8Util
+  # use '?' intsead of the unicode replace char, since that is 3 bytes
+  # and can increase the string size if it's done a lot
+  REPLACEMENT_CHAR = "?"
+
+  # Replace invalid UTF-8 character sequences with a replacement character
+  #
+  # Returns self as valid UTF-8.
+  def self.clean!(str)
+    raise NotImplementedError
+  end
+
+  # Replace invalid UTF-8 character sequences with a replacement character
+  #
+  # Returns a copy of this String as valid UTF-8.
+  def self.clean(str)
+    clean!(str.dup)
+  end
+
+end
+
+if RUBY_VERSION <= '1.9'
+  require 'resque/vendor/utf8_util/utf8_util_18'
+else
+  require 'resque/vendor/utf8_util/utf8_util_19'
+end

lib/resque/vendor/utf8_util/utf8_util_18.rb

@@ -0,0 +1,91 @@
+require 'strscan'
+
+module UTF8Util
+  HIGH_BIT_RANGE = /[\x80-\xff]/
+
+  # Check if this String is valid UTF-8
+  #
+  # Returns true or false.
+  def self.valid?(str)
+    sc = StringScanner.new(str)
+
+    while sc.skip_until(HIGH_BIT_RANGE)
+      sc.pos -= 1
+
+      if !sequence_length(sc)
+        return false
+      end
+    end
+
+    true
+  end
+
+  # Replace invalid UTF-8 character sequences with a replacement character
+  #
+  # Returns self as valid UTF-8.
+  def self.clean!(str)
+    sc = StringScanner.new(str)
+    while sc.skip_until(HIGH_BIT_RANGE)
+      pos = sc.pos = sc.pos-1
+
+      if !sequence_length(sc)
+        str[pos] = REPLACEMENT_CHAR
+      end
+    end
+
+    str
+  end
+
+  # Validate the UTF-8 sequence at the current scanner position.
+  #
+  # scanner - StringScanner instance so we can advance the pointer as we verify.
+  #
+  # Returns The length in bytes of this UTF-8 sequence, false if invalid.
+  def self.sequence_length(scanner)
+    leader = scanner.get_byte[0]
+
+    if (leader >> 5) == 0x6
+      if check_next_sequence(scanner)
+        return 2
+      else
+        scanner.pos -= 1
+      end
+    elsif (leader >> 4) == 0x0e
+      if check_next_sequence(scanner)
+        if check_next_sequence(scanner)
+          return 3
+        else
+          scanner.pos -= 2
+        end
+      else
+        scanner.pos -= 1
+      end
+    elsif (leader >> 3) == 0x1e
+      if check_next_sequence(scanner)
+        if check_next_sequence(scanner)
+          if check_next_sequence(scanner)
+            return 4
+          else
+            scanner.pos -= 3
+          end
+        else
+          scanner.pos -= 2
+        end
+      else
+        scanner.pos -= 1
+      end
+    end
+
+    false
+  end
+
+  private
+
+  # Read another byte off the scanner oving the scan position forward one place
+  #
+  # Returns nothing.
+  def self.check_next_sequence(scanner)
+    byte = scanner.get_byte[0]
+    (byte >> 6) == 0x2
+  end
+end

lib/resque/vendor/utf8_util/utf8_util_19.rb

@@ -0,0 +1,5 @@
+module UTF8Util
+  def self.clean!(str)
+    str.force_encoding("binary").encode("UTF-8", :invalid => :replace, :undef => :replace, :replace => REPLACEMENT_CHAR)
+  end
+end