Cover some more edge cases in EM::Pool:

 * Prevent leaks from untimely remove calls
 * Allow exceptions (for example in async-sinatra) from causing stale resources
 * Improve semantics for external resource management during error conditions

lib/em/pool.rb

@@ -122,7 +122,10 @@ def requeue resource
 
     def failure resource
       if @on_error
+        @contents.delete resource
         @on_error.call resource
+        # Prevent users from calling a leak.
+        @removed.delete resource
       else
         requeue resource
       end
@@ -140,7 +143,9 @@ def process work, resource
       else
         raise ArgumentError, "deferrable expected from work"
       end
+    rescue Exception
+      failure resource
+      raise
     end
-
   end
-end
+end

tests/test_pool.rb

@@ -115,6 +115,41 @@ def test_contents
     assert_equal [:res], pool.contents
   end
 
+  def test_contents_when_perform_errors_and_on_error_is_not_set
+    pool.add :res
+    assert_equal [:res], pool.contents
+
+    pool.perform do |r|
+      d = EM::DefaultDeferrable.new
+      d.fail
+      d
+    end
+
+    EM.run { EM.next_tick { EM.stop } }
+
+    assert_equal [:res], pool.contents
+  end
+
+  def test_contents_when_perform_errors_and_on_error_is_set
+    pool.add :res
+    res = nil
+    pool.on_error do |res|
+      res = res
+    end
+    assert_equal [:res], pool.contents
+
+    pool.perform do |r|
+      d = EM::DefaultDeferrable.new
+      d.fail 'foo'
+      d
+    end
+
+    EM.run { EM.next_tick { EM.stop } }
+
+    assert_equal :res, res
+    assert_equal [], pool.contents
+  end
+
   def test_num_waiting
     pool.add :res
     assert_equal 0, pool.num_waiting
@@ -125,4 +160,35 @@ def test_num_waiting
     assert_equal 10, pool.num_waiting
   end
 
-end
+  def test_exceptions_in_the_work_block_bubble_up_raise_and_fail_the_resource
+    pool.add :res
+
+    res = nil
+    pool.on_error { |r| res = r }
+    pool.perform { raise 'boom' }
+
+    assert_raises(RuntimeError) do
+      EM.run { EM.next_tick { EM.stop } }
+    end
+
+    assert_equal [], pool.contents
+    assert_equal :res, res
+  end
+
+  def test_removed_list_does_not_leak_on_errors
+    pool.add :res
+
+    pool.on_error do |r|
+      # This is actually the wrong thing to do, and not required, but some users
+      # might do it. When they do, they would find that @removed would cause a
+      # slow leak.
+      pool.remove r
+    end
+
+    pool.perform { d = EM::DefaultDeferrable.new; d.fail; d }
+
+    EM.run { EM.next_tick { EM.stop } }
+    assert_equal [], pool.instance_variable_get(:@removed)
+  end
+
+end