Re-introduce support for io_close hook.

Author: samuel-williams-shopify

io.c

@@ -5547,22 +5547,28 @@ fptr_finalize_flush(rb_io_t *fptr, int noraise, int keepgvl)
         done = 1;
     }
 
+    // Wait for blocking operations to be interrupted and cleaned up BEFORE calling
+    // the scheduler hook. This ensures that if the scheduler closes the fd synchronously
+    // (or schedules an async close that completes quickly), blocking operations won't
+    // fail with EBADF.
+    rb_thread_io_close_wait(fptr);
+
+    // Call scheduler hook BEFORE setting fd = -1, so scheduler can access the fd
+    // via rb_io_descriptor if needed (e.g., for async close operations like io_uring_prep_close).
+    // The fd is still valid at this point. If the scheduler returns a truthy value,
+    // it will handle closing the fd asynchronously, so we skip the synchronous close below.
+    if (!done && fd >= 0) {
+        VALUE scheduler = rb_fiber_scheduler_current();
+        if (scheduler != Qnil) {
+            VALUE result = rb_fiber_scheduler_io_close(scheduler, fptr->self);
+            if (!UNDEF_P(result)) done = 1;
+        }
+    }
+
     fptr->fd = -1;
     fptr->stdio_file = 0;
     fptr->mode &= ~(FMODE_READABLE|FMODE_WRITABLE);
 
-    // wait for blocking operations to ensure they do not hit EBADF:
-    rb_thread_io_close_wait(fptr);
-
-    // Disable for now.
-    // if (!done && fd >= 0) {
-    //     VALUE scheduler = rb_fiber_scheduler_current();
-    //     if (scheduler != Qnil) {
-    //         VALUE result = rb_fiber_scheduler_io_close(scheduler, fptr->self);
-    //         if (!UNDEF_P(result)) done = 1;
-    //     }
-    // }
-
     if (!done && stdio_file) {
         // stdio_file is deallocated anyway even if fclose failed.
         if ((maygvl_fclose(stdio_file, noraise) < 0) && NIL_P(error)) {

thread.c

@@ -1773,6 +1773,15 @@ rb_io_blocking_operations(struct rb_io *io)
 static void
 rb_io_blocking_operation_enter(struct rb_io *io, struct rb_io_blocking_operation *blocking_operation)
 {
+    // Prevent new blocking operations from starting if the IO is being closed.
+    // closing_ec is set when closing starts and cleared after fd = -1 is set.
+    if (io->closing_ec) {
+        rb_execution_context_t *ec = GET_EC();
+        rb_vm_t *vm = ec->thread_ptr->vm;
+        VALUE error = vm->special_exceptions[ruby_error_stream_closed];
+        rb_exc_raise(error);
+    }
+
     ccan_list_add(rb_io_blocking_operations(io), &blocking_operation->list);
 }
 
@@ -2893,9 +2902,11 @@ rb_thread_io_close_wait(struct rb_io* io)
     }
     rb_mutex_unlock(wakeup_mutex);
 
-    // We are done closing:
+    // We are done waiting for blocking operations:
     io->wakeup_mutex = Qnil;
-    io->closing_ec = NULL;
+    // Note: closing_ec is NOT cleared here. Once fd = -1 is set, rb_thread_io_close_interrupt
+    // won't be called again (due to fd check), and new blocking operations will fail at
+    // rb_io_check_closed (due to fd check), so closing_ec becomes redundant but harmless.
 }
 
 void