fix: request-scope lifecycle and epoll server — silent mid-run shard deaths

Three framework defects combined to kill long-running servers mid-test:

- SocketMonitor drove MHD in external select mode: fd_sets are fixed at
  FD_SETSIZE (1024), so once the process fd table grew past 1024 the
  per-fd watch rebuild read past the fd_set struct (undefined behavior)
  and connections on descriptors >= 1024 could not be served. The
  daemon now starts with MHD_USE_EPOLL and the monitor watches MHD's
  single epoll fd (MHD_get_daemon_info), removing the ceiling entirely.

- Pipeline.run created a request scope that nothing ever released.
  Scoped endpoints hold strong references back to the scope
  (EndpointRouter, statum ActionEndpoint/StatumAction), so every
  request leaked its whole object graph — including scoped
  Connection/OrmSession instances and their sqlite db+wal fd pair —
  until fd exhaustion silently killed the listener. The scope is now
  released in a finally block (Inversion.Scope.release clears the
  scoped-instance and local-registration caches, breaking every cycle
  through the scope in one cut).

- The MHD_OPTION_NOTIFY_COMPLETED closure was passed swapped: the
  callback pointer went into the cls slot and NULL into the callback
  slot, so request completion never fired and RequestContexts leaked
  on every request (respond() also removed the context from the
  server's set too early, relying on that leak to keep the object MHD
  still reads alive). The completion callback is now wired correctly,
  receives the server as cls, and is the single point where a request
  context is dropped.

src/Core/Pipeline.vala

@@ -14,19 +14,23 @@ namespace Astralis {
         public async HttpResult run(HttpContext context) throws Error {
             var scope = container.create_scope();
             scope.register_local_scoped<HttpContext>(() => context);
-            var components = scope.resolve_all<PipelineComponent>().to_immutable_buffer();
+            try {
+                var components = scope.resolve_all<PipelineComponent>().to_immutable_buffer();
 
-            if(components.length == 0) {
-                throw new PipelineError.NO_COMPONENTS("No components in pipeline");
-            }
+                if(components.length == 0) {
+                    throw new PipelineError.NO_COMPONENTS("No components in pipeline");
+                }
 
-            var current_context = new PipelineContext(context, null, null);
-            for(uint i = components.length; i > 0; i--) {
-                current_context = new PipelineContext(context, components[i-1], current_context); 
-            }
+                var current_context = new PipelineContext(context, null, null);
+                for(uint i = components.length; i > 0; i--) {
+                    current_context = new PipelineContext(context, components[i-1], current_context); 
+                }
 
-            var result = yield current_context.next();
-            return result;
+                var result = yield current_context.next();
+                return result;
+            } finally {
+                scope.release();
+            }
         }
     }
 

src/Server/Server.vala

@@ -105,11 +105,11 @@ namespace Astralis {
         }
 
         public void run() {
-            // External select mode: no USE_SELECT_INTERNALLY flag
-            // This means MHD creates no threads - we drive it from GLib main loop
-            // USE_TURBO enables faster processing by not explicitly closing connections
+            // External epoll mode: no internal polling flag, MHD exposes its
+            // epoll fd and we drive it from the GLib main loop. USE_TURBO
+            // enables faster processing by not explicitly closing connections.
             daemon = Daemon.start(
-                MHD.ALLOW_SUSPEND_RESUME | MHD.USE_TURBO,
+                MHD.ALLOW_SUSPEND_RESUME | MHD.USE_TURBO | MHD.USE_EPOLL,
                 (uint16) this.port,
                 null,
                 (connection, url, method, version, upload_data, upload_data_size, con_cls) => {
@@ -117,7 +117,7 @@ namespace Astralis {
                 },
                 MHD.OPTION_NOTIFY_COMPLETED,
                 (void*) request_completed_callback,
-                null,
+                (void*) this,
                 MHD.OPTION_END
             );
 
@@ -159,7 +159,6 @@ namespace Astralis {
             else if(context.handler_result != null) {
                 result = send_result(connection, context.handler_result, context);
             }
-            request_contexts.remove(context);
             return result;
         }
 
@@ -219,19 +218,28 @@ namespace Astralis {
 
         // Static callback wrapper for MHD_OPTION_NOTIFY_COMPLETED
         private static void request_completed_callback(void* cls, Connection connection, void** con_cls, RequestTerminationCode toe) {
-            // We don't have access to the Server instance here, so we handle cleanup directly
-            if (con_cls[0] == null) {
+            if (cls == null || con_cls[0] == null) {
                 return;
             }
-            
+
+            var server = (Server) ((Object*) cls);
             RequestContext context = (RequestContext) ((Object*) con_cls[0]);
-            
+
             // If the request was terminated abnormally, disconnect the output stream
             if (toe != RequestTerminationCode.COMPLETED_OK) {
                 if (context.response_context != null) {
                     context.response_context.body_output.close_connection();
+                    server.response_contexts.remove(context.response_context);
                 }
             }
+
+            // Drop the server's reference to the request context. The
+            // completion notification fires exactly once per request —
+            // normal completion and abnormal termination (client
+            // disconnects mid-request) alike — so this is the single
+            // point where the context's lifetime ends; MHD reads
+            // con_cls until then.
+            server.request_contexts.remove(context);
         }
 
         // Simple memory-based InputStream for request body

src/Server/SocketMonitor.vala

@@ -1,164 +1,88 @@
 using MHD;
-using Invercargill;
-using Invercargill.DataStructures;
 
 namespace Astralis {
 
-    /// Monitors MHD's file descriptors and integrates them with GLib's main loop.
-    /// This enables external select mode where MHD runs entirely within the GLib
+    /// Monitors MHD's epoll file descriptor and integrates it with GLib's main loop.
+    /// This enables external epoll mode where MHD runs entirely within the GLib
     /// main loop, eliminating cross-thread communication issues with suspend/resume.
+    ///
+    /// MHD is started with MHD_USE_EPOLL (no internal polling thread) and exposes
+    /// its epoll fd via MHD_get_daemon_info; a single GLib IO watch on that fd
+    /// drives MHD_run. Unlike the previous external-select integration this has
+    /// no FD_SETSIZE (1024) ceiling: select()'s fixed-size fd_set can neither
+    /// represent nor safely test descriptors >= 1024, which silently killed
+    /// long-running servers as soon as their fd table grew that high.
     internal class SocketMonitor : Object {
         private unowned MHD.Daemon daemon;
-        private Dictionary<int, WatchInfo> watches;
+        private IOChannel? epoll_channel;
+        private uint epoll_watch_id;
         private uint timeout_id;
         private bool running;
-        
-        private class WatchInfo {
-            public IOChannel channel;
-            public uint watch_id;
-            public bool monitor_read;
-            public bool monitor_write;
-            
-            public WatchInfo(IOChannel channel, uint watch_id, bool monitor_read, bool monitor_write) {
-                this.channel = channel;
-                this.watch_id = watch_id;
-                this.monitor_read = monitor_read;
-                this.monitor_write = monitor_write;
-            }
-        }
-        
+
         public SocketMonitor(MHD.Daemon daemon) {
             this.daemon = daemon;
-            this.watches = new Dictionary<int, WatchInfo>();
+            this.epoll_channel = null;
+            this.epoll_watch_id = 0;
             this.timeout_id = 0;
             this.running = false;
         }
-        
+
         /// Start monitoring MHD sockets and integrate with GLib main loop
         public void start() {
             if (running) {
                 return;
             }
+
+            var info = MHD.get_daemon_info(daemon, MHD.DaemonInfoType.EPOLL_FD);
+            if (info == null || ((!)info).epoll_fd < 0) {
+                error("Astralis: MHD daemon does not expose an epoll fd (external epoll mode is required on this platform)");
+            }
+            int epoll_fd = ((!)info).epoll_fd;
+
+            // The IOChannel does not own the fd; MHD keeps it open for the
+            // daemon's lifetime, and the watch holds a reference to the channel.
+            epoll_channel = new IOChannel.unix_new(epoll_fd);
+            IOCondition cond = IOCondition.IN | IOCondition.HUP | IOCondition.ERR | IOCondition.NVAL;
+            epoll_watch_id = epoll_channel.add_watch(cond, (source, condition) => {
+                if (!running) {
+                    return false;
+                }
+
+                // Run MHD processing
+                MHD.run(daemon);
+
+                // Reschedule timeout if MHD requested a new one
+                schedule_next_timeout();
+
+                // Continue monitoring
+                return true;
+            });
+
             running = true;
-            update_watches();
             schedule_next_timeout();
         }
-        
+
         /// Stop monitoring and clean up all GLib sources
         public void stop() {
             if (!running) {
                 return;
             }
             running = false;
-            
-            // Remove all IO channel watches
-            foreach (var entry in watches) {
-                Source.remove(entry.value.watch_id);
+
+            if (epoll_watch_id != 0) {
+                Source.remove(epoll_watch_id);
+                epoll_watch_id = 0;
+            }
+            if (epoll_channel != null) {
+                epoll_channel = null;
             }
-            watches.clear();
-            
-            // Remove timeout watcher
+
             if (timeout_id != 0) {
                 Source.remove(timeout_id);
                 timeout_id = 0;
             }
         }
-        
-        /// Update GLib IO channel watches based on MHD's current fd_set
-        private void update_watches() {
-            if (!running) {
-                return;
-            }
-            
-            // Get current fd_set from MHD
-            MHD.FDSet read_fds = MHD.FDSet();
-            MHD.FDSet write_fds = MHD.FDSet();
-            MHD.FDSet error_fds = MHD.FDSet();
-            int max_fd = 0;
-            
-            MHD.FDSet.zero(ref read_fds);
-            MHD.FDSet.zero(ref write_fds);
-            MHD.FDSet.zero(ref error_fds);
-            
-            MHD.get_fdset(daemon, ref read_fds, ref write_fds, ref error_fds, &max_fd);
-            
-            // Build set of active file descriptors
-            var active_fds = new Series<int>();
-            for (int fd = 0; fd <= max_fd; fd++) {
-                bool needs_read = MHD.FDSet.is_set(fd, ref read_fds);
-                bool needs_write = MHD.FDSet.is_set(fd, ref write_fds);
-                
-                if (needs_read || needs_write) {
-                    active_fds.add(fd);
-                    
-                    // Add new watch if not already monitoring this fd, or replace it if conditions changed
-                    WatchInfo? existing_watch = null;
-                    if (watches.try_get(fd, out existing_watch)) {
-                        if (existing_watch.monitor_read != needs_read || existing_watch.monitor_write != needs_write) {
-                            remove_watch_for_fd(fd);
-                            add_watch_for_fd(fd, needs_read, needs_write);
-                        }
-                    } else {
-                        add_watch_for_fd(fd, needs_read, needs_write);
-                    }
-                }
-            }
-            
-            // Remove watches for closed file descriptors
-            var to_remove = new Series<int>();
-            foreach (var entry in watches) {
-                if (!active_fds.contains(entry.key)) {
-                    to_remove.add(entry.key);
-                }
-            }
-            foreach (int fd in to_remove) {
-                remove_watch_for_fd(fd);
-            }
-        }
-        
-        /// Add a GLib IO channel watch for a file descriptor
-        private void add_watch_for_fd(int fd, bool monitor_read, bool monitor_write) {
-            var channel = new IOChannel.unix_new(fd);
-            
-            // Build IO condition flags - always monitor for errors/hangup
-            IOCondition cond = IOCondition.HUP | IOCondition.ERR | IOCondition.NVAL;
-            if (monitor_read) {
-                cond |= IOCondition.IN;
-            }
-            if (monitor_write) {
-                cond |= IOCondition.OUT;
-            }
-            
-            uint watch_id = channel.add_watch(cond, (source, condition) => {
-                if (!running) {
-                    return false;
-                }
-                
-                // Run MHD processing
-                MHD.run(daemon);
-                
-                // Socket set may have changed after processing
-                update_watches();
-                
-                // Reschedule timeout if MHD requested a new one
-                schedule_next_timeout();
-                
-                // Continue monitoring
-                return true;
-            });
-            
-            watches[fd] = new WatchInfo(channel, watch_id, monitor_read, monitor_write);
-        }
-        
-        /// Remove a GLib IO channel watch for a file descriptor
-        private void remove_watch_for_fd(int fd) {
-            WatchInfo? info = null;
-            if (watches.try_get(fd, out info)) {
-                Source.remove(info.watch_id);
-                watches.remove(fd);
-            }
-        }
-        
+
         private bool update_pending = false;
 
         public void force_update() {
@@ -166,37 +90,34 @@ namespace Astralis {
                 return;
             }
             update_pending = true;
-            
+
             Idle.add(() => {
                 update_pending = false;
                 if (!running) {
                     return false;
                 }
-                
+
                 // Run MHD processing
                 MHD.run(daemon);
-                
-                // Socket set may have changed
-                update_watches();
-                
+
                 // Reschedule timeout
                 schedule_next_timeout();
-                
+
                 return false;
             });
         }
-        
+
         /// Schedule the next timeout callback using MHD's recommended timeout
         private void schedule_next_timeout() {
             if (!running) {
                 return;
             }
-            
+
             if (timeout_id != 0) {
                 Source.remove(timeout_id);
                 timeout_id = 0;
             }
-            
+
             uint64 timeout_ms = 0;
             if (MHD.get_timeout(daemon, &timeout_ms) == MHD.Result.YES) {
                 if (timeout_ms == 0) {
@@ -206,7 +127,6 @@ namespace Astralis {
                         }
                         timeout_id = 0;
                         MHD.run(daemon);
-                        update_watches();
                         schedule_next_timeout();
                         return false; // run once, reschedule later
                     });
@@ -217,7 +137,6 @@ namespace Astralis {
                         }
                         timeout_id = 0;
                         MHD.run(daemon);
-                        update_watches();
                         schedule_next_timeout();
                         // Don't auto-repeat - we reschedule manually with potentially new timeout
                         return false;

vapi/libmicrohttpd.vapi

@@ -201,4 +201,28 @@ namespace MHD {
 
     [CCode (cname = "MHD_get_timeout")]
     public Result get_timeout(Daemon daemon, uint64* timeout);
+
+    [CCode (cname = "enum MHD_DaemonInfoType", cprefix = "MHD_DAEMON_INFO_")]
+    public enum DaemonInfoType {
+        KEY_SIZE,
+        MAC_KEY_SIZE,
+        LISTEN_FD,
+        EPOLL_FD,
+        CURRENT_CONNECTIONS,
+        FLAGS,
+        BIND_PORT
+    }
+
+    [CCode (cname = "union MHD_DaemonInfo", has_type_id = false)]
+    public struct DaemonInfo {
+        public size_t key_size;
+        public size_t mac_key_size;
+        public int listen_fd;
+        public uint16 port;
+        public int epoll_fd;
+        public uint num_connections;
+    }
+
+    [CCode (cname = "MHD_get_daemon_info")]
+    public unowned DaemonInfo? get_daemon_info(Daemon daemon, DaemonInfoType info_type, ...);
 }