Close the connection if an exception occurred during a query in order…

… to avoid leaving it in an undefined state.

ext/mysql.c

@@ -1157,34 +1157,57 @@ static int should_schedule_query(){
     return rb_thread_alone() != 1;
 }
 
-/* async_query(sql,timeout=nil)
-   optionally take a block
- */
-static VALUE async_query(int argc, VALUE* argv, VALUE obj)
-{
-    MYSQL* m = GetHandler(obj); 
+struct RealAsyncQueryArgs {
+    VALUE obj;
+    MYSQL* m;
     VALUE sql, timeout;
+};
 
-    rb_scan_args(argc, argv, "11", &sql, &timeout);
-
-    async_in_progress_set( obj, Qfalse );
+static VALUE real_async_query(VALUE _args)
+{
+    struct RealAsyncQueryArgs *args = (struct RealAsyncQueryArgs *) _args;
+
+    async_in_progress_set( args->obj, Qfalse );
 
-    busy(obj); 
+    busy(args->obj); 
 
-    send_query( obj, sql );
+    send_query( args->obj, args->sql );
 
     if ( should_schedule_query() ){
-      schedule_query(obj, timeout);
+      schedule_query(args->obj, args->timeout);
     } 
 
     if (rb_block_given_p()) {
-      rb_yield( get_result(obj) );
-      idle( obj );
-      return obj; 
+      rb_yield( get_result(args->obj) );
+      idle( args->obj );
+      return args->obj;
     }else{
-      idle( obj );
-      return get_result(obj); 
-    }  
+      idle( args->obj );
+      return get_result(args->obj);
+    }
+}
+
+/* async_query(sql,timeout=nil)
+   optionally take a block
+ */
+static VALUE async_query(int argc, VALUE* argv, VALUE obj)
+{
+    struct RealAsyncQueryArgs args;
+    int status;
+    VALUE result;
+
+    args.obj = obj;
+    args.m = GetHandler(obj);
+    rb_scan_args(argc, argv, "11", &args.sql, &args.timeout);
+
+    result = rb_protect(real_async_query, (VALUE) &args, &status);
+    if (status) {
+        my_close(obj);
+        rb_jump_tag(status);
+        return Qnil;
+    } else {
+        return result;
+    }
 }
 
 #if MYSQL_VERSION_ID >= 40100

lib/mysqlplus.rb

@@ -12,6 +12,9 @@ def ruby_async_query(sql, timeout = nil) # known to deadlock TODO
     send_query(sql)
     select [ (@sockets ||= {})[socket] ||= IO.new(socket) ], nil, nil, nil
     get_result
+  rescue Exception
+    close
+    raise
   end
 
   begin