allow releasing connections of the pool from pool_statements

remove old connections from PoolStatement::@connections either disposed or closed connections

src/db/database.cr

@@ -1,4 +1,5 @@
 require "http/params"
+require "weak_ref"
 
 module DB
   # Acts as an entry point for database access.
@@ -63,7 +64,7 @@ module DB
     end
 
     # :nodoc:
-    def checkout_some(candidates : Enumerable(Connection)) : {Connection, Bool}
+    def checkout_some(candidates : Enumerable(WeakRef(Connection))) : {Connection, Bool}
       @pool.checkout_some candidates
     end
 

src/db/pool.cr

@@ -1,3 +1,5 @@
+require "weak_ref"
+
 module DB
   class Pool(T)
     @initial_pool_size : Int32
@@ -47,18 +49,19 @@ module DB
     # ```
     # `selected` be a resource from the `candidates` list and `is_candidate` == `true`
     # or `selected` will be a new resource adn `is_candidate` == `false`
-    def checkout_some(candidates : Enumerable(T)) : {T, Bool}
+    def checkout_some(candidates : Enumerable(WeakRef(T))) : {T, Bool}
       # TODO honor candidates while waiting for availables
       # this will allow us to remove `candidates.includes?(resource)`
-      candidates.each do |resource|
-        if is_available?(resource)
+      candidates.each do |ref|
+        resource = ref.target
+        if resource && is_available?(resource)
           @available.delete resource
           return {resource, true}
         end
       end
 
       resource = checkout
-      {resource, candidates.includes?(resource)}
+      {resource, candidates.any? { |ref| ref.target == resource }}
     end
 
     def release(resource : T) : Nil

src/db/pool_statement.cr

@@ -7,7 +7,7 @@ module DB
     include StatementMethods
 
     # connections where the statement was prepared
-    @connections = Set(Connection).new
+    @connections = Set(WeakRef(Connection)).new
 
     def initialize(@db : Database, @query : String)
       # Prepares a statement on some connection
@@ -61,13 +61,22 @@ module DB
     # builds a statement over a real connection
     # the conneciton is registered in `@connections`
     private def build_statement
-      # TODO closed connections should be removed from @connections
-      # either by callbacks or by week references.
+      clean_connections
       conn, existing = @db.checkout_some(@connections)
-      @connections << conn unless existing
+      @connections << WeakRef.new(conn) unless existing
       conn.prepare(@query)
     end
 
+    private def clean_connections
+      # remove disposed or closed connections
+      @connections.each do |ref|
+        conn = ref.target
+        if !conn || conn.closed?
+          @connections.delete ref
+        end
+      end
+    end
+
     private def statement_with_retry
       @db.retry do
         return yield build_statement