Server: harden against crashes

server/src/main/kotlin/com/jaytux/phoebench/server/Main.kt

@@ -185,7 +185,7 @@ fun Application.module() {
 
             cseAuth(Routes.CSE.monitor,
                 { id, user -> transaction { accessibleProjectCSE(user, id, true) } },
-                ProjectHandler::monitorSetup) { _, _, bus, event ->
+                ProjectHandler::monitorSetup, ProjectHandler::monitorTeardown) { _, _, bus, event ->
                 try {
                     when(event) {
                         is ClientMonitorEvent.ApplicationFinished -> bus.end(ServerMonitorEvent.ApplicationEnd(Clock.System.now(), event.exitCode))

server/src/main/kotlin/com/jaytux/phoebench/server/handlers/Bridge.kt

@@ -282,6 +282,7 @@ inline fun <reified TParams: Any, reified TEvent: Any, TInter, TExtra> Route.wra
     api: CSERoute<TParams, TEvent>,
     noinline extra: suspend (ApplicationCall, TParams) -> TInter,
     noinline setup: suspend (TParams, TInter) -> TExtra,
+    noinline onClose: suspend (TParams, TInter, TExtra) -> Unit,
     noinline handler: suspend (TParams, TInter, TExtra, TEvent) -> Unit
 ) {
     webSocket(api.pattern) {
@@ -292,11 +293,15 @@ inline fun <reified TParams: Any, reified TEvent: Any, TInter, TExtra> Route.wra
 
             val converter = KotlinxWebsocketSerializationConverter(Json)
             val extra = setup(params, inter)
-            for(frame in incoming) {
+            try {
-                handler(params, inter, extra, converter.deserialize<TEvent>(frame))
+                for (frame in incoming) {
+                    handler(params, inter, extra, converter.deserialize<TEvent>(frame))
+                }
+            }
+            catch(e: ClosedReceiveChannelException) {
+                onClose(params, inter, extra)
             }
         }
-        catch(e: ClosedReceiveChannelException) {}
         catch(e: RouteError.CSERouteError) {
             close(CloseReason(e.code, e.message ?: "Unknown error"))
         }
@@ -310,6 +315,7 @@ inline fun <reified TParams: Any, reified TEvent: Any, TExtra> Route.wrapAuthCSE
     api: CSERoute<TParams, TEvent>,
     noinline verifyUser: suspend (TParams, User) -> Unit,
     noinline setup: suspend (TParams, User) -> TExtra,
+    noinline onClose: suspend (TParams, User, TExtra) -> Unit,
     noinline handler: suspend (TParams, User, TExtra, TEvent) -> Unit
 ) = wrapCSE<TParams, TEvent, User, TExtra>(
     api = api,
@@ -327,28 +333,32 @@ inline fun <reified TParams: Any, reified TEvent: Any, TExtra> Route.wrapAuthCSE
         user
     },
     setup = setup,
+    onClose = onClose,
     handler = handler
 )
 
 inline fun <reified TParams: Any, reified TEvent: Any, TExtra> Route.cse(
     api: CSERoute<TParams, TEvent>,
-    noinline setup: suspend (TParams, Unit) -> TExtra,
+    noinline setup: suspend (TParams) -> TExtra,
+    noinline onClose: suspend (TParams, TExtra) -> Unit,
     noinline handler: suspend (TParams, TExtra, TEvent) -> Unit
 ) {
     if(api.elevation != Elevation.UN_AUTH) throw IllegalArgumentException("CSE ${api.pattern} can only be used with ${api.elevation}")
-    wrapCSE(api, { _, _ -> }, setup) { params, _, extra, event -> handler(params, extra, event) }
+    wrapCSE(api, { _, _ -> }, { it, _ -> setup(it) }, { p, _, extra -> onClose(p, extra) }) { params, _, extra, event -> handler(params, extra, event) }
 }
 
 inline fun <reified TParams: Any, reified TEvent: Any, TExtra> Route.cseAuth(
     api: CSERoute<TParams, TEvent>, noinline verifyUser: suspend (TParams, User) -> Unit,
-    noinline setup: suspend (TParams, User) -> TExtra, noinline handler: suspend (TParams, User, TExtra, TEvent) -> Unit
+    noinline setup: suspend (TParams, User) -> TExtra, noinline onClose: suspend (TParams, User, TExtra) -> Unit,
+    noinline handler: suspend (TParams, User, TExtra, TEvent) -> Unit
 ) {
     if(api.elevation != Elevation.AUTH) throw IllegalArgumentException("CSE ${api.pattern} can only be used with ${api.elevation}")
-    wrapAuthCSE(api, verifyUser, setup, handler)
+    wrapAuthCSE(api, verifyUser, setup, onClose, handler)
 }
 
 inline fun <reified TParams: Any, reified TEvent: Any, TExtra> Route.cseAdmin(
     api: CSERoute<TParams, TEvent>, noinline setup: suspend (TParams, User) -> TExtra,
+    noinline onClose: suspend (TParams, User, TExtra) -> Unit,
     noinline handler: suspend (TParams, User, TExtra, TEvent) -> Unit
 ) {
     if(api.elevation != Elevation.ADMIN) throw IllegalArgumentException("CSE ${api.pattern} can only be used with ${api.elevation}")
@@ -356,5 +366,5 @@ inline fun <reified TParams: Any, reified TEvent: Any, TExtra> Route.cseAdmin(
         if(!user.isAdmin) {
             throw RouteError.CSERouteError("Admin access required", CloseReasons.NOT_AUTHORIZED)
         }
-    }, setup, handler)
+    }, setup, onClose, handler)
 }

server/src/main/kotlin/com/jaytux/phoebench/server/handlers/ProjectHandler.kt

@@ -402,6 +402,14 @@ object ProjectHandler {
         }
     }
 
+    suspend fun monitorTeardown(
+        id: Uuid, user: User,
+        bus: SSEBus.MonitorSSEBus<ServerMonitorEvent, ServerMonitorEvent.ApplicationStart, ServerMonitorEvent.Message,
+                ServerMonitorEvent.ApplicationEnd, ServerMonitorEvent.Backlog>
+    ) {
+        bus.end(ServerMonitorEvent.ApplicationEnd(Clock.System.now(), -1))
+    }
+
     fun deleteMonitor(user: User, req: Uuid) = transaction {
         checkMigration(user)
         accessibleProject(user, req, true)