Merge branch 'web-stack' of Tilo15/astralis into master

.usmignore

@@ -0,0 +1,9 @@
+# Build trees at any depth (directory-only: the directory and all beneath it)
+builddir*/
+
+# Database artifacts by basename at any depth
+*.sqlite*
+db.sqlite
+
+# Local agent session state (directory-only)
+.kilo/

MANIFEST.usm

@@ -0,0 +1,58 @@
+{
+  "name": "astralis",
+  "version": "0.1",
+  "summary": "Library for writing HTTP backends",
+  "licences": [],
+  "flags": ["ninjaStyleProgress"],
+  "provides": {
+    "lib:libastralis-0.1.so": "as-expected",
+    "inc:astralis-0.1.h": "as-expected",
+    "vapi:astralis-0.1.vapi": "as-expected",
+    "gir:astralis-0.1.gir": "as-expected",
+    "typelib:libastralis-0.1.typelib": "as-expected",
+    "pc:astralis-0.1.pc": "as-expected"
+  },
+  "depends": {
+    "runtime": [
+      "lib:libglib-2.0.so.0",
+      "lib:libgobject-2.0.so.0",
+      "lib:libgio-2.0.so.0",
+      "lib:libmicrohttpd.so.12",
+      "lib:libjson-glib-1.0.so.0",
+      "lib:libinvercargill-1.so",
+      "lib:libinversion-0.1.so",
+      "lib:libxml2.so.2",
+      "lib:libz.so.1",
+      "lib:libbrotlienc.so.1",
+      "lib:libzstd.so.1"
+    ],
+    "build": [
+      "bin:valac",
+      "bin:meson",
+      "bin:ninja",
+      "bin:gcc",
+      "bin:g-ir-compiler",
+      "pc:glib-2.0.pc",
+      "pc:gobject-2.0.pc",
+      "pc:gio-2.0.pc",
+      "pc:gio-unix-2.0.pc",
+      "pc:json-glib-1.0.pc",
+      "pc:zlib.pc",
+      "pc:libbrotlienc.pc",
+      "pc:libzstd.pc",
+      "pc:libxml-2.0.pc",
+      "pc:libmicrohttpd.pc",
+      "pc:invercargill-1.pc",
+      "pc:invercargill-json.pc",
+      "pc:inversion-0.1.pc"
+    ],
+    "manage": [
+      "bin:bash"
+    ]
+  },
+  "execs": {
+    "build": "usm-scripts/build.sh",
+    "install": "usm-scripts/install.sh",
+    "postInstall": "usm-scripts/post-install.sh"
+  }
+}

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/Endpoints/SseEndpoint.vala

@@ -39,10 +39,11 @@ namespace Astralis {
                 close();
                 throw new IOError.CLOSED("Cannot send to closed SSE stream");
             }
-            // Skip writing if the buffer is full (client is slow)
-            if (output.write_would_block) {
-                return;
-            }
+            // Backpressure is applied by AsyncOutput.write, which waits for
+            // the socket buffer to drain. Dropping events here (the old
+            // write_would_block skip) lost backlog events whenever a burst
+            // outran the client — SSE has no retransmit, so every event
+            // must be written.
             var formatted = event.format();
             yield output.write(new ByteBuffer.from_byte_array(formatted.data));
         }

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;

usm-scripts/build.sh

@@ -0,0 +1,10 @@
+#!/bin/bash
+set -e
+
+src_dir=$(pwd)
+build_dir=$1
+
+cd ${build_dir}
+
+meson setup ${src_dir} --prefix=${PREFIX} --libdir=${LIBDIR} --bindir=${BINDIR} --includedir=${INCLUDEDIR}
+ninja

usm-scripts/install.sh

@@ -0,0 +1,10 @@
+#!/bin/bash
+set -e
+
+src_dir=$(pwd)
+build_dir=$1
+install_dir=$2
+
+cd ${build_dir}
+
+meson install --destdir ${install_dir}

usm-scripts/post-install.sh

@@ -0,0 +1,2 @@
+#!/bin/bash
+ldconfig

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, ...);
 }