ENG-2031: Fixed the bug!

This is an ugly fix that I'm going to try and simplify. The blast
radius for these changes is larger than necessary, so I want to try
a simpler fix.

lib/broadway/topology/producer_stage.ex

@@ -422,9 +422,12 @@ defmodule Broadway.Topology.ProducerStage do
     Utility.maybe_log("Remaining rate limit: #{left}", state)
 
     cond do
-      messages == [] ->
+      messages == [] and next_message_weight == 0 ->
         {:open, [], []}
 
+      messages == [] ->
+        {:closed, [], []}
+
       # If no more messages are allowed, we're rate limited but we're able
       # to emit all messages that we have.
       left == 0 ->

lib/broadway/topology/rate_limiter.ex

@@ -21,6 +21,10 @@ defmodule Broadway.Topology.RateLimiter do
     end
   end
 
+  def rate_limit(counter, 0) when is_reference(counter) do
+    get_currently_allowed(counter)
+  end
+
   def rate_limit(counter, amount)
       when is_reference(counter) and is_integer(amount) and amount > 0 do
     :atomics.sub_get(counter, @atomics_index, amount)