Upgrade GrottoPress/pond shard to v2.0

CHANGELOG.md

@@ -5,6 +5,11 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](http://keepachangelog.com/en/1.0.0/)
 and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.html).
 
+## [Unreleased] - 
+
+### Changed
+- Upgrade `GrottoPress/pond` shard to v2.0
+
 ## [0.19.0] - 2024-02-21
 
 ### Changed

shard.yml

@@ -16,7 +16,7 @@ dependencies:
     version: ~> 0.4.0
   pond:
     github: GrottoPress/pond
-    version: ~> 1.0
+    version: ~> 2.0
   redis:
     github: jgaskins/redis
     version: ~> 0.8.0

src/spec.cr

@@ -6,6 +6,6 @@ module Mel
   end
 
   def sync(task : Task?)
-    task.try &.run.try { |fiber| Pond.drain(fiber) }
+    Pond.drain { |pond| task.try &.run(pond) }
   end
 end

src/worker.cr

@@ -60,9 +60,7 @@ module Mel
   end
 
   private def run_pending_tasks(pond)
-    Task.find_pending(-1).try do |tasks|
-      tasks.each &.run(force: true).try { |fiber| pond << fiber }
-    end
+    Task.find_pending(-1).try &.each &.run(pond, force: true)
   end
 
   private def run_tasks(pond)
@@ -71,7 +69,7 @@ module Mel
 
     while state.started?
       Task.find_lte(Time.local, batch_size(pond), delete: nil).try do |tasks|
-        tasks.each &.run(force: true).try { |fiber| pond << fiber }
+        tasks.each &.run(pond, force: true)
       end
 
       sleep jittered_poll_interval

src/worker/task.cr

@@ -1,14 +1,14 @@
 require "./task/**"
 
 abstract class Mel::Task
-  def run(*, force = false) : Fiber?
+  def run(pond : Pond, *, force = false) : Fiber?
     return log_not_due unless force || due?
     do_before_run
 
     self.attempts += 1
     set_run_time
 
-    spawn(name: id) do
+    pond.fill(name: id) do
       log_running
       job.run
     rescue error