Stream compilation errors

nebula-api/src/main/kotlin/com/orbitalhq/nebula/core/Events.kt

@@ -46,10 +46,28 @@ data class ContainerInfo(
 )
 
 
+/**
+ * A single diagnostic produced when compiling a submitted stack script.
+ * Carried on [StackStateEvent] so consumers (e.g. Orbital) can surface
+ * why a submission was rejected, and reused by the admin API.
+ */
+data class CompilationError(
+    val message: String,
+    val line: Int?,
+    val column: Int?,
+    val severity: String
+)
+
+/**
+ * State of a single stack. When a submission fails to compile, the event names
+ * the rejected stack and carries the [compilationErrors]; [stackState] is empty
+ * in that case (any previously-running version of the stack is left untouched).
+ */
 data class StackStateEvent(
     val stackName: String,
     val stateCounts: Map<ComponentState, Int>,
-    val stackState: NebulaStackState
+    val stackState: NebulaStackState,
+    val compilationErrors: List<CompilationError> = emptyList()
 )
 
 

nebula-dsl/src/main/kotlin/com/orbitalhq/nebula/NebulaStack.kt

@@ -13,6 +13,7 @@ import com.orbitalhq.nebula.s3.S3Dsl
 import com.orbitalhq.nebula.sql.SqlDsl
 import com.orbitalhq.nebula.taxi.TaxiPublisherDsl
 import com.orbitalhq.nebula.utils.NameGenerator
+import io.github.oshai.kotlinlogging.KotlinLogging
 import reactor.core.publisher.Flux
 import java.util.concurrent.atomic.AtomicBoolean
 
@@ -34,6 +35,10 @@ class NebulaStack(
     val name: StackName = NameGenerator.generateName(),
     initialComponents: List<InfrastructureComponent<*>> = emptyList()
 ) : InfraDsl, KafkaDsl, S3Dsl, HttpDsl, SqlDsl, HazelcastDsl, MongoDsl, TaxiPublisherDsl {
+    companion object {
+        private val logger = KotlinLogging.logger {}
+    }
+
     private val _components = mutableListOf<InfrastructureComponent<*>>()
 
     private val isStarted = AtomicBoolean(false)
@@ -68,9 +73,17 @@ class NebulaStack(
         markStarted()
         stackStateEventSource.listenForEvents(name, components)
         logStream.attachLogStreams(components)
-        return components.associate { component ->
-            component.type to component.start(config, hostConfig)
-        }
+        return components.mapNotNull { component ->
+            try {
+                component.type to component.start(config, hostConfig)
+            } catch (e: Exception) {
+                // The component's own event source is responsible for emitting Failed
+                // (which is what reaches clients); here we just stop the exception from
+                // killing the stack's start thread, and let the remaining components start.
+                logger.error(e) { "Component ${component.name} in stack $name failed to start" }
+                null
+            }
+        }.toMap()
     }
 
     fun markStarted() {

nebula-dsl/src/main/kotlin/com/orbitalhq/nebula/http/HttpDsl.kt

@@ -52,6 +52,9 @@ class HttpApiBuilder(private val port: Int = 0, private val componentName: Compo
     fun post(path: String, handler: suspend PipelineContext<Unit, ApplicationCall>.(ApplicationCall) -> Unit) {
         addRoute(HttpMethod.Post, path, handler)
     }
+    fun patch(path: String, handler: suspend PipelineContext<Unit, ApplicationCall>.(ApplicationCall) -> Unit) {
+        addRoute(HttpMethod.Patch, path, handler)
+    }
 
     fun put(path: String, handler: suspend PipelineContext<Unit, ApplicationCall>.(ApplicationCall) -> Unit) {
         addRoute(HttpMethod.Put, path, handler)

nebula-dsl/src/main/kotlin/com/orbitalhq/nebula/http/HttpExecutor.kt

@@ -70,6 +70,7 @@ class HttpExecutor(private val config: HttpConfig, loggerNames: List<LoggerName>
                         HttpMethod.Post -> post(route.path) {  route.handler(this, call)  }
                         HttpMethod.Put -> put(route.path) {  route.handler(this, call)  }
                         HttpMethod.Delete -> delete(route.path) {  route.handler(this, call) }
+                        HttpMethod.Patch -> patch(route.path) {  route.handler(this, call) }
                         else -> throw IllegalArgumentException("Unsupported HTTP method: ${route.method}")
                     }
                 }

nebula-dsl/src/main/kotlin/com/orbitalhq/nebula/sql/DatabaseExecutor.kt

@@ -7,6 +7,7 @@ import com.orbitalhq.nebula.StackRunner
 import com.orbitalhq.nebula.containerInfoFrom
 import com.orbitalhq.nebula.core.ComponentInfo
 import com.orbitalhq.nebula.core.ComponentLifecycleEvent
+import com.orbitalhq.nebula.core.ComponentState
 import com.orbitalhq.nebula.endpointFor
 import com.orbitalhq.nebula.events.ComponentLifecycleEventSource
 import com.orbitalhq.nebula.logging.LogStream
@@ -55,11 +56,20 @@ class DatabaseExecutor(private val config: DatabaseConfig, loggers: List<LoggerN
         databaseContainer = config.container.withDatabaseName(config.databaseName)
             .withNetwork(nebulaConfig.network)
             .withNetworkAliases(config.componentName)
-        eventSource.startContainerAndEmitEvents(databaseContainer, name)
+        // Table DDL and seed data run inside the guarded init step: a failure there
+        // (e.g. invalid DDL) emits Failed with the database's error message, rather
+        // than throwing out of the stack's start thread with nothing reported back.
+        eventSource.startContainerAndEmitEvents(databaseContainer, name) {
+            setupDataSource()
+            setupJooq()
+            createTablesAndLoadData()
+        }
 
-        setupDataSource()
-        setupJooq()
-        createTablesAndLoadData()
+        if (!databaseContainer.isRunning) {
+            // The container itself never came up (Failed has already been emitted) —
+            // its ports and jdbcUrl are unreadable, so there's no ComponentInfo to build.
+            error("Database container for component $name failed to start")
+        }
 
         // The internal port the DB listens on inside the container (5432, 3306, ...).
         val internalPort = databaseContainer.exposedPorts.first()
@@ -86,7 +96,11 @@ class DatabaseExecutor(private val config: DatabaseConfig, loggers: List<LoggerN
             id = id
 
         )
-        eventSource.running()
+        // Re-emit Running now that componentInfo is populated — unless the init step
+        // failed, where re-emitting would clobber the Failed state and its message.
+        if (currentState.state != ComponentState.Failed) {
+            eventSource.running()
+        }
         return componentInfo!!
     }
 

nebula-dsl/src/main/kotlin/com/orbitalhq/nebula/sql/SqlDsl.kt

@@ -35,11 +35,11 @@ interface SqlDsl : InfraDsl {
 class DatabaseBuilder(private val container: JdbcDatabaseContainer<*>, private val dialect: SQLDialect, private val type: String, private val databaseName: String, private val componentName: ComponentName) {
     private val tables = mutableListOf<TableConfig>()
 
-    fun table(name: String, ddl: String, data: List<Map<String, Any>> = emptyList()) {
+    fun table(name: String, ddl: String, data: List<Map<String, Any?>> = emptyList()) {
         tables.add(TableConfig(name, ddl, data))
     }
 
-    fun table(name: String, ddl: String, vararg data: Map<String, Any>) {
+    fun table(name: String, ddl: String, vararg data: Map<String, Any?>) {
         table(name, ddl, data.toList())
     }
 
@@ -60,4 +60,4 @@ data class DatabaseConfig(
     val componentName: String
 )
 
-data class TableConfig(val name: String, val ddl: String, val data: List<Map<String, Any>>)
+data class TableConfig(val name: String, val ddl: String, val data: List<Map<String, Any?>>)

nebula-dsl/src/test/kotlin/com/orbitalhq/nebula/StackStartFailureTest.kt

@@ -0,0 +1,71 @@
+package com.orbitalhq.nebula
+
+import com.orbitalhq.nebula.core.ComponentInfo
+import com.orbitalhq.nebula.core.ComponentLifecycleEvent
+import com.orbitalhq.nebula.core.ComponentState
+import com.orbitalhq.nebula.core.LifecycleUpdatedEvent
+import com.orbitalhq.nebula.core.NotStartedEvent
+import com.orbitalhq.nebula.logging.LogStream
+import io.kotest.assertions.throwables.shouldNotThrowAny
+import io.kotest.core.spec.style.DescribeSpec
+import io.kotest.matchers.shouldBe
+import reactor.core.publisher.Flux
+
+/**
+ * A component throwing out of start() must not kill the stack's start thread,
+ * and must not prevent the stack's other components from starting.
+ */
+class StackStartFailureTest : DescribeSpec({
+
+    describe("starting a stack where one component throws") {
+
+        it("still starts the remaining components and does not propagate the exception") {
+            val broken = ThrowingComponent("broken")
+            val healthy = RecordingComponent("healthy")
+            val stack = NebulaStack("test-stack", listOf(broken, healthy))
+
+            val componentInfos = shouldNotThrowAny {
+                stack.startComponents(NebulaConfig(), HostConfig.UNKNOWN)
+            }
+
+            healthy.started shouldBe true
+            // Only the component that started successfully contributes state
+            componentInfos.keys shouldBe setOf("recording")
+        }
+    }
+})
+
+private class ThrowingComponent(override val name: String) : InfrastructureComponent<Unit> {
+    override val type = "throwing"
+    override val componentInfo: ComponentInfo<Unit>? = null
+    override val lifecycleEvents: Flux<ComponentLifecycleEvent> = Flux.never()
+    override val currentState: ComponentLifecycleEvent = NotStartedEvent
+    override val logStream = LogStream(name)
+
+    override fun start(nebulaConfig: NebulaConfig, hostConfig: HostConfig): ComponentInfo<Unit> {
+        throw IllegalStateException("Deliberate start failure")
+    }
+
+    override fun stop() {}
+}
+
+private class RecordingComponent(override val name: String) : InfrastructureComponent<Unit> {
+    override val type = "recording"
+    var started = false
+        private set
+
+    override var componentInfo: ComponentInfo<Unit>? = null
+        private set
+    override val lifecycleEvents: Flux<ComponentLifecycleEvent> = Flux.never()
+    override val currentState: ComponentLifecycleEvent
+        get() = if (started) LifecycleUpdatedEvent(ComponentState.Running) else NotStartedEvent
+    override val logStream = LogStream(name)
+
+    override fun start(nebulaConfig: NebulaConfig, hostConfig: HostConfig): ComponentInfo<Unit> {
+        started = true
+        componentInfo = ComponentInfo(container = null, componentConfig = Unit, type = type, name = name, id = id)
+        return componentInfo!!
+    }
+
+    override fun stop() {}
+}

nebula-dsl/src/test/kotlin/com/orbitalhq/nebula/sql/SqlExecutorFailureTest.kt

@@ -0,0 +1,54 @@
+package com.orbitalhq.nebula.sql
+
+import com.orbitalhq.nebula.StackRunner
+import com.orbitalhq.nebula.core.ComponentState
+import com.orbitalhq.nebula.core.LifecycleEventWithMessage
+import com.orbitalhq.nebula.stack
+import com.orbitalhq.nebula.start
+import io.kotest.core.spec.style.DescribeSpec
+import io.kotest.matchers.shouldBe
+import io.kotest.matchers.string.shouldContain
+import io.kotest.matchers.types.shouldBeInstanceOf
+
+/**
+ * Invalid DDL (or bad seed data) must surface as a Failed component state carrying
+ * the database's error message — not as an exception thrown out of the stack's
+ * start thread, which previously left the component stuck reporting a healthy
+ * state with nothing reported to the user.
+ */
+class SqlExecutorFailureTest : DescribeSpec({
+
+    lateinit var infra: StackRunner
+
+    describe("a postgres stack with invalid DDL") {
+        afterTest {
+            infra.shutDownAll()
+        }
+
+        it("reports Failed with the database error instead of throwing out of start") {
+            infra = stack {
+                postgres {
+                    // Deliberately broken: references a type that doesn't exist.
+                    table(
+                        "users", """
+                        CREATE TABLE users (
+                            id NOT_A_REAL_TYPE PRIMARY KEY
+                        )
+                    """
+                    )
+                }
+            }.start()
+
+            val database = infra.database.single()
+            val state = database.currentState
+            state.state shouldBe ComponentState.Failed
+            state.shouldBeInstanceOf<LifecycleEventWithMessage>()
+            state.message shouldContain "not_a_real_type"
+
+            // The stack snapshot (what the /stream/stacks contract sends) reflects the failure too
+            val snapshot = infra.snapshot().single()
+            val componentState = snapshot.stackState.values.single().single().state
+            componentState.state shouldBe ComponentState.Failed
+        }
+    }
+})

nebula-runtime/src/main/kotlin/com/orbitalhq/nebula/runtime/server/AdminModels.kt

@@ -1,25 +1,18 @@
 package com.orbitalhq.nebula.runtime.server
 
 import com.orbitalhq.nebula.StackName
+import com.orbitalhq.nebula.core.CompilationError
 import com.orbitalhq.nebula.core.StackStateEvent
 import kotlin.script.experimental.api.ScriptDiagnostic
 
-/** A single compilation diagnostic, flattened for the admin API. */
-data class CompilationErrorDto(
-    val message: String,
-    val line: Int?,
-    val column: Int?,
-    val severity: String
-)
-
 /**
  * A stack that was submitted but failed to compile. Held in memory until a
  * valid version is submitted under the same name (which replaces it).
  */
 data class FailedSubmission(
     val name: StackName,
     val source: String,
-    val compilationErrors: List<CompilationErrorDto>
+    val compilationErrors: List<CompilationError>
 )
 
 /**
@@ -32,12 +25,12 @@ data class AdminStackView(
     val name: StackName,
     val stackState: StackStateEvent?,
     val source: String,
-    val compilationErrors: List<CompilationErrorDto>
+    val compilationErrors: List<CompilationError>
 ) {
     val failedCompilation: Boolean get() = compilationErrors.isNotEmpty()
 }
 
-fun ScriptDiagnostic.toCompilationErrorDto(): CompilationErrorDto = CompilationErrorDto(
+fun ScriptDiagnostic.toCompilationError(): CompilationError = CompilationError(
     message = message,
     line = location?.start?.line,
     column = location?.start?.col,