diff --git a/nebula-api/src/main/kotlin/com/orbitalhq/nebula/core/Events.kt b/nebula-api/src/main/kotlin/com/orbitalhq/nebula/core/Events.kt index 4de8e17..9978f83 100644 --- a/nebula-api/src/main/kotlin/com/orbitalhq/nebula/core/Events.kt +++ b/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, - val stackState: NebulaStackState + val stackState: NebulaStackState, + val compilationErrors: List = emptyList() ) diff --git a/nebula-dsl/src/main/kotlin/com/orbitalhq/nebula/NebulaStack.kt b/nebula-dsl/src/main/kotlin/com/orbitalhq/nebula/NebulaStack.kt index 32f6280..b974ed0 100644 --- a/nebula-dsl/src/main/kotlin/com/orbitalhq/nebula/NebulaStack.kt +++ b/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> = emptyList() ) : InfraDsl, KafkaDsl, S3Dsl, HttpDsl, SqlDsl, HazelcastDsl, MongoDsl, TaxiPublisherDsl { + companion object { + private val logger = KotlinLogging.logger {} + } + private val _components = mutableListOf>() 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() { diff --git a/nebula-dsl/src/main/kotlin/com/orbitalhq/nebula/http/HttpDsl.kt b/nebula-dsl/src/main/kotlin/com/orbitalhq/nebula/http/HttpDsl.kt index 503d92a..b75ead0 100644 --- a/nebula-dsl/src/main/kotlin/com/orbitalhq/nebula/http/HttpDsl.kt +++ b/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.(ApplicationCall) -> Unit) { addRoute(HttpMethod.Post, path, handler) } + fun patch(path: String, handler: suspend PipelineContext.(ApplicationCall) -> Unit) { + addRoute(HttpMethod.Patch, path, handler) + } fun put(path: String, handler: suspend PipelineContext.(ApplicationCall) -> Unit) { addRoute(HttpMethod.Put, path, handler) diff --git a/nebula-dsl/src/main/kotlin/com/orbitalhq/nebula/http/HttpExecutor.kt b/nebula-dsl/src/main/kotlin/com/orbitalhq/nebula/http/HttpExecutor.kt index e5699d4..0a52c99 100644 --- a/nebula-dsl/src/main/kotlin/com/orbitalhq/nebula/http/HttpExecutor.kt +++ b/nebula-dsl/src/main/kotlin/com/orbitalhq/nebula/http/HttpExecutor.kt @@ -70,6 +70,7 @@ class HttpExecutor(private val config: HttpConfig, loggerNames: List 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}") } } diff --git a/nebula-dsl/src/main/kotlin/com/orbitalhq/nebula/sql/DatabaseExecutor.kt b/nebula-dsl/src/main/kotlin/com/orbitalhq/nebula/sql/DatabaseExecutor.kt index ef88ecc..e4d1faa 100644 --- a/nebula-dsl/src/main/kotlin/com/orbitalhq/nebula/sql/DatabaseExecutor.kt +++ b/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, private val dialect: SQLDialect, private val type: String, private val databaseName: String, private val componentName: ComponentName) { private val tables = mutableListOf() - fun table(name: String, ddl: String, data: List> = emptyList()) { + fun table(name: String, ddl: String, data: List> = emptyList()) { tables.add(TableConfig(name, ddl, data)) } - fun table(name: String, ddl: String, vararg data: Map) { + fun table(name: String, ddl: String, vararg data: Map) { 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>) \ No newline at end of file +data class TableConfig(val name: String, val ddl: String, val data: List>) \ No newline at end of file diff --git a/nebula-dsl/src/main/resources/META-INF/kotlin/script/templates/com.orbitalhq.nebula.NebulaScript.classname b/nebula-dsl/src/main/resources/META-INF/kotlin/script/templates/com.orbitalhq.nebula.NebulaScript.classname new file mode 100644 index 0000000..e69de29 diff --git a/nebula-dsl/src/test/kotlin/com/orbitalhq/nebula/StackStartFailureTest.kt b/nebula-dsl/src/test/kotlin/com/orbitalhq/nebula/StackStartFailureTest.kt new file mode 100644 index 0000000..b9aba17 --- /dev/null +++ b/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 { + override val type = "throwing" + override val componentInfo: ComponentInfo? = null + override val lifecycleEvents: Flux = Flux.never() + override val currentState: ComponentLifecycleEvent = NotStartedEvent + override val logStream = LogStream(name) + + override fun start(nebulaConfig: NebulaConfig, hostConfig: HostConfig): ComponentInfo { + throw IllegalStateException("Deliberate start failure") + } + + override fun stop() {} +} + +private class RecordingComponent(override val name: String) : InfrastructureComponent { + override val type = "recording" + var started = false + private set + + override var componentInfo: ComponentInfo? = null + private set + override val lifecycleEvents: Flux = 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 { + started = true + componentInfo = ComponentInfo(container = null, componentConfig = Unit, type = type, name = name, id = id) + return componentInfo!! + } + + override fun stop() {} +} diff --git a/nebula-dsl/src/test/kotlin/com/orbitalhq/nebula/sql/SqlExecutorFailureTest.kt b/nebula-dsl/src/test/kotlin/com/orbitalhq/nebula/sql/SqlExecutorFailureTest.kt new file mode 100644 index 0000000..a45b230 --- /dev/null +++ b/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() + 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 + } + } +}) diff --git a/nebula-runtime/src/main/kotlin/com/orbitalhq/nebula/runtime/server/AdminModels.kt b/nebula-runtime/src/main/kotlin/com/orbitalhq/nebula/runtime/server/AdminModels.kt index 1c56bef..c2a40ac 100644 --- a/nebula-runtime/src/main/kotlin/com/orbitalhq/nebula/runtime/server/AdminModels.kt +++ b/nebula-runtime/src/main/kotlin/com/orbitalhq/nebula/runtime/server/AdminModels.kt @@ -1,17 +1,10 @@ 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). @@ -19,7 +12,7 @@ data class CompilationErrorDto( data class FailedSubmission( val name: StackName, val source: String, - val compilationErrors: List + val compilationErrors: List ) /** @@ -32,12 +25,12 @@ data class AdminStackView( val name: StackName, val stackState: StackStateEvent?, val source: String, - val compilationErrors: List + val compilationErrors: List ) { 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, diff --git a/nebula-runtime/src/main/kotlin/com/orbitalhq/nebula/runtime/server/NebulaServer.kt b/nebula-runtime/src/main/kotlin/com/orbitalhq/nebula/runtime/server/NebulaServer.kt index 011a184..062f59d 100644 --- a/nebula-runtime/src/main/kotlin/com/orbitalhq/nebula/runtime/server/NebulaServer.kt +++ b/nebula-runtime/src/main/kotlin/com/orbitalhq/nebula/runtime/server/NebulaServer.kt @@ -9,6 +9,7 @@ import com.orbitalhq.nebula.NebulaStack import com.orbitalhq.nebula.NebulaStackWithSource import com.orbitalhq.nebula.StackName import com.orbitalhq.nebula.StackRunner +import com.orbitalhq.nebula.core.StackStateEvent import com.orbitalhq.nebula.runtime.NebulaScriptExecutor import io.github.oshai.kotlinlogging.KotlinLogging import io.ktor.http.* @@ -145,7 +146,7 @@ class NebulaServer( val failure = FailedSubmission( id, script, - exception.errors.map { it.toCompilationErrorDto() } + exception.errors.map { it.toCompilationError() } ) failedSubmissions[id] = failure call.respond(HttpStatusCode.UnprocessableEntity, failure) @@ -239,22 +240,54 @@ class NebulaServer( val payloadJson = frame.readText() logger.info { "Received updated stack submission: \n$payloadJson" } - val updateStacksRequest = - objectMapper.readValue(payloadJson) - - val stackMap = compile(updateStacksRequest, call.hostConfig()) - val eventStreams = stackMap.map { (name, stack) -> - stackExecutor.submit(stack, name, startAsync = true) - } - Flux.merge(eventStreams) - .subscribe { event -> - logger.info { "Emitting stack status event for stack ${event.stackName}" } - runBlocking { - val stackStatusJson = objectMapper.writeValueAsString(event) - send(Frame.Text(stackStatusJson)) - } + // Any failure handling a submission must not tear down the socket — + // the client would see nothing but a dropped connection. + try { + val updateStacksRequest = + objectMapper.readValue(payloadJson) + // Compile each stack independently: a broken script must not block + // the other stacks in the submission. Failures are reported back to + // the client as a StackStateEvent carrying the compilation errors, + // and recorded so the admin snapshot shows them too. + val eventStreams = updateStacksRequest.stacks.mapNotNull { (name, script) -> + scriptExecutor.compileToStackWithSource(script, call.hostConfig()).fold( + { compilationException -> + val errors = compilationException.errors.map { it.toCompilationError() } + logger.warn { "Stack $name failed to compile: ${errors.joinToString { it.message }}" } + failedSubmissions[name] = FailedSubmission(name, script, errors) + send( + Frame.Text( + objectMapper.writeValueAsString( + StackStateEvent( + stackName = name, + stateCounts = emptyMap(), + stackState = emptyMap(), + compilationErrors = errors + ) + ) + ) + ) + null + }, + { stackWithSource -> + failedSubmissions.remove(name) + stackExecutor.submit(stackWithSource.withName(name), name, startAsync = true) + } + ) } + Flux.merge(eventStreams) + .subscribe { event -> + logger.info { "Emitting stack status event for stack ${event.stackName}" } + runBlocking { + val stackStatusJson = objectMapper.writeValueAsString(event) + send(Frame.Text(stackStatusJson)) + } + + } + } catch (e: Exception) { + logger.error(e) { "Failed to process stack submission" } + } } } // Serve the management UI (bundled into the jar under resources/web). @@ -302,12 +335,6 @@ class NebulaServer( } } - private fun compile(updateStacksRequest: UpdateStackRSocketRequest, hostConfig: HostConfig): Map { - return updateStacksRequest.stacks.mapValues { (key, stackScript) -> - scriptExecutor.toStackWithSource(stackScript, hostConfig).withName(key) - } - } - } data class StackEventStreamRequest(val stackId: StackName) diff --git a/nebula-runtime/src/test/kotlin/com/orbitalhq/nebula/runtime/server/StreamStacksCompilationErrorTest.kt b/nebula-runtime/src/test/kotlin/com/orbitalhq/nebula/runtime/server/StreamStacksCompilationErrorTest.kt new file mode 100644 index 0000000..574907a --- /dev/null +++ b/nebula-runtime/src/test/kotlin/com/orbitalhq/nebula/runtime/server/StreamStacksCompilationErrorTest.kt @@ -0,0 +1,92 @@ +package com.orbitalhq.nebula.runtime.server + +import com.fasterxml.jackson.module.kotlin.jacksonObjectMapper +import com.fasterxml.jackson.module.kotlin.readValue +import com.orbitalhq.nebula.StackRunner +import com.orbitalhq.nebula.core.ComponentState +import com.orbitalhq.nebula.core.StackStateEvent +import io.kotest.core.spec.style.DescribeSpec +import io.kotest.matchers.collections.shouldNotBeEmpty +import io.kotest.matchers.maps.shouldBeEmpty +import io.kotest.matchers.shouldBe +import io.ktor.client.* +import io.ktor.client.engine.cio.* +import io.ktor.client.plugins.websocket.* +import io.ktor.websocket.* +import kotlinx.coroutines.withTimeout +import org.mockito.kotlin.any +import org.mockito.kotlin.eq +import org.mockito.kotlin.mock +import org.mockito.kotlin.verify +import org.mockito.kotlin.whenever +import reactor.core.publisher.Flux +import kotlin.time.Duration.Companion.minutes + +/** + * Covers the /stream/stacks contract for scripts that fail to compile: + * the submission must produce a StackStateEvent carrying the compilation + * errors (rather than silently killing the socket), and the same connection + * must remain usable for a subsequent, valid submission. + */ +class StreamStacksCompilationErrorTest : DescribeSpec({ + + describe("/stream/stacks compilation failures") { + + it("reports compilation errors back over the socket and keeps the connection alive") { + val runningEvent = StackStateEvent( + stackName = "my-stack", + stateCounts = mapOf(ComponentState.Running to 1), + stackState = emptyMap() + ) + val mockExecutor: StackRunner = mock {} + whenever(mockExecutor.submit(any(), eq("my-stack"), eq(true))).thenReturn(Flux.just(runningEvent)) + + val server = NebulaServer(port = 0, stackExecutor = mockExecutor) + val applicationEngine = server.start(wait = false) + val port = applicationEngine.resolvedConnectors().first().port + + val objectMapper = jacksonObjectMapper() + val client = HttpClient(CIO) { + install(WebSockets) + } + + try { + // Script compilation warms up the Kotlin scripting host, which can take + // a while on first use — hence the generous timeout. + withTimeout(3.minutes) { + client.webSocket("ws://localhost:$port/stream/stacks") { + send(Frame.Text(objectMapper.writeValueAsString( + UpdateStackRSocketRequest(mapOf("my-stack" to "this is not a valid nebula script")) + ))) + + val failureEvent = objectMapper.readValue(nextTextFrame()) + failureEvent.stackName shouldBe "my-stack" + failureEvent.compilationErrors.shouldNotBeEmpty() + failureEvent.stackState.shouldBeEmpty() + + // The socket must survive the failure: a corrected script submitted on + // the same connection compiles, is submitted, and its state is relayed. + send(Frame.Text(objectMapper.writeValueAsString( + UpdateStackRSocketRequest(mapOf("my-stack" to "stack {}")) + ))) + + val recoveredEvent = objectMapper.readValue(nextTextFrame()) + recoveredEvent.stackName shouldBe "my-stack" + recoveredEvent.compilationErrors shouldBe emptyList() + } + } + verify(mockExecutor).submit(any(), eq("my-stack"), eq(true)) + } finally { + client.close() + applicationEngine.stop() + } + } + } +}) + +private suspend fun DefaultClientWebSocketSession.nextTextFrame(): String { + while (true) { + val frame = incoming.receive() + if (frame is Frame.Text) return frame.readText() + } +}