fix: recover Continue process after unexpected exceptions (#7463)

Solves CON-3587
Solves #7018
Solves #7121
Solves #5505
Solves #3981, and possibly other Stream closed-like errors

## Main change

Now, if some exception bubbles up at any stage (whether it's read, write, anything else within `processScope`), we catch it via `CoroutineExceptionHandler` and we restart the process + back-off. Type of exception is not important / now it's more like black-box approach.

This simplifies things quite a bit, since we now have only single piece of code for process recovery.

## Detailed changes

* The process now recovers from exceptions globally, without specifying the type of exception (more of a black-box approach).
* Remove `CoreMessengerManager`, since process recovery is now implemented in `ContinueProcessHandler`.
* Add a back-off utility for calculating delays between process recovery retries (just a cleanup).
* Restore `ContinuePluginService.dispose()` (because `Disposable` was accidentally removed during the last changes in the Next Edit - that's not critical, but it was causing exceptions when closing the application/during plugin disposal).
* Remove the artificial delay between reading messages (from now on, blocking happens directly on `readLine`).

Author: exigow

extensions/intellij/src/main/kotlin/com/github/continuedev/continueintellijextension/activities/ContinuePluginStartupActivity.kt

@@ -266,8 +266,7 @@ class ContinuePluginStartupActivity : StartupActivity, DumbAware {
                 project.service<ContinuePluginDisposable>()
             )
 
-            val coreMessengerManager = CoreMessengerManager(project, ideProtocolClient, coroutineScope)
-            continuePluginService.coreMessengerManager = coreMessengerManager
+            continuePluginService.coreMessenger = CoreMessenger(project, ideProtocolClient, coroutineScope)
         }
     }
 }

extensions/intellij/src/main/kotlin/com/github/continuedev/continueintellijextension/continue/CoreMessenger.kt

@@ -5,25 +5,24 @@ import com.github.continuedev.continueintellijextension.constants.MessageTypes
 import com.github.continuedev.continueintellijextension.`continue`.process.ContinueBinaryProcess
 import com.github.continuedev.continueintellijextension.`continue`.process.ContinueProcessHandler
 import com.github.continuedev.continueintellijextension.`continue`.process.ContinueSocketProcess
-import com.github.continuedev.continueintellijextension.services.ContinuePluginService
 import com.github.continuedev.continueintellijextension.utils.uuid
 import com.google.gson.Gson
 import com.google.gson.JsonSyntaxException
-import com.intellij.openapi.components.service
 import com.intellij.openapi.diagnostic.Logger
 import com.intellij.openapi.project.Project
 import kotlinx.coroutines.CoroutineScope
 
 class CoreMessenger(
     private val project: Project,
     private val ideProtocolClient: IdeProtocolClient,
-    val coroutineScope: CoroutineScope,
-    private val onUnexpectedExit: () -> Unit
+
+    // todo: this scope is public only to cancel the coroutines on dispose in ContinuePluginService
+    // todo: scope is created in ContinuePluginStartupActivity; refactor this/clean this up/encapsulate
+    val coroutineScope: CoroutineScope
 ) {
     private val gson = Gson()
     private val responseListeners = mutableMapOf<String, (Any?) -> Unit>()
-    private var process = startContinueProcess()
-    private val log = Logger.getInstance(CoreMessenger::class.java.simpleName)
+    private val process = startContinueProcess()
 
     fun request(messageType: String, data: Any?, messageId: String?, onResponse: (Any?) -> Unit) {
         val id = messageId ?: uuid()
@@ -34,14 +33,15 @@ class CoreMessenger(
 
     private fun startContinueProcess(): ContinueProcessHandler {
         val isTcp = System.getenv("USE_TCP")?.toBoolean() ?: false
-        val process = if (isTcp)
-            ContinueSocketProcess()
-        else
-            ContinueBinaryProcess(onUnexpectedExit)
-        return ContinueProcessHandler(coroutineScope, process, ::handleMessage)
+        return ContinueProcessHandler(coroutineScope, ::readProcessMessage) {
+            if (isTcp)
+                ContinueSocketProcess()
+            else
+                ContinueBinaryProcess()
+        }
     }
 
-    private fun handleMessage(json: String) {
+    private fun readProcessMessage(json: String) {
         val responseMap = tryToParse(json) ?: return
         val messageId = responseMap["messageId"].toString()
         val messageType = responseMap["messageType"].toString()
@@ -76,19 +76,16 @@ class CoreMessenger(
         try {
             gson.fromJson(json, Map::class.java)
         } catch (_: JsonSyntaxException) {
-            log.warn("Invalid message JSON: $json") // example: NODE_ENV undefined
+            LOG.warn("Invalid message JSON: $json") // example: NODE_ENV undefined
             null
         }
 
     fun restart() {
-        log.warn("Restarting Continue process")
         responseListeners.clear()
-        process.close()
-        process = startContinueProcess()
+        process.restart()
     }
 
-    fun close() {
-        log.warn("Closing Continue process")
-        process.close()
+    private companion object {
+        private val LOG = Logger.getInstance(CoreMessenger::class.java.simpleName)
     }
 }

extensions/intellij/src/main/kotlin/com/github/continuedev/continueintellijextension/continue/CoreMessengerManager.kt

@@ -0,0 +1,25 @@
+package com.github.continuedev.continueintellijextension.`continue`.process
+
+import kotlin.time.Duration
+import kotlin.time.Duration.Companion.seconds
+
+/**
+ * Starts at [initialDuration] and doubles on each call to [nextDuration] up to [maxDuration].
+ */
+class BackoffCalculator(
+    private val initialDuration: Duration = 1.seconds,
+    private val maxDuration: Duration = 30.seconds
+) {
+    private var currentTime: Duration = initialDuration
+
+    init {
+        require(initialDuration > 0.seconds)
+    }
+
+    fun nextDuration(): Duration {
+        val delay = currentTime
+        val next = currentTime * 2.0
+        currentTime = if (next <= maxDuration) next else maxDuration
+        return delay
+    }
+}

extensions/intellij/src/main/kotlin/com/github/continuedev/continueintellijextension/continue/process/BackoffCalculator.kt

@@ -1,12 +1,9 @@
 package com.github.continuedev.continueintellijextension.`continue`.process
 
-import com.github.continuedev.continueintellijextension.error.ContinuePostHogService
-import com.github.continuedev.continueintellijextension.error.ContinueSentryService
 import com.github.continuedev.continueintellijextension.proxy.ProxySettings
 import com.github.continuedev.continueintellijextension.utils.OS
 import com.github.continuedev.continueintellijextension.utils.getContinueBinaryPath
 import com.github.continuedev.continueintellijextension.utils.getOS
-import com.intellij.openapi.components.service
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.runBlocking
 import java.io.File
@@ -16,46 +13,26 @@ import java.nio.file.Files
 import java.nio.file.Paths
 import java.nio.file.attribute.PosixFilePermission
 
-class ContinueBinaryProcess(
-    private val onUnexpectedExit: () -> Unit
-) : ContinueProcess {
+class ContinueBinaryProcess() : ContinueProcess {
 
-    private val process = startBinaryProcess()
+    private val process = start()
     override val input: InputStream = process.inputStream
     override val output: OutputStream = process.outputStream
 
     override fun close() =
         process.destroy()
 
-    private fun startBinaryProcess(): Process {
-        val path = getContinueBinaryPath()
-        runBlocking(Dispatchers.IO) {
-            setPermissions()
-        }
-
-        val builder = ProcessBuilder(path)
-        builder.environment() += ProxySettings.getSettings().toContinueEnvVars()
-        return builder
-            .directory(File(path).parentFile)
-            .start()
-            .apply { onExit().thenRun(onUnexpectedExit).thenRun(::reportErrorTelemetry) }
-    }
+    private companion object {
 
-    private fun reportErrorTelemetry() {
-        var err = process.errorStream?.bufferedReader()?.readText()?.trim()
-        if (err != null) {
-            // There are often "⚡️Done in Xms" messages, and we want everything after the last one
-            val delimiter = "⚡ Done in"
-            val doneIndex = err.lastIndexOf(delimiter)
-            if (doneIndex != -1) {
-                err = err.substring(doneIndex + delimiter.length)
-            }
+        private fun start(): Process {
+            val path = getContinueBinaryPath()
+            setPermissions()
+            val builder = ProcessBuilder(path)
+            builder.environment() += ProxySettings.getSettings().toContinueEnvVars()
+            return builder
+                .directory(File(path).parentFile)
+                .start()
         }
-        service<ContinueSentryService>().reportMessage("Core process exited with output: $err")
-        service<ContinuePostHogService>().capture("jetbrains_core_exit", mapOf("error" to err))
-    }
-
-    private companion object {
 
         private fun setPermissions() {
             val os = getOS()

extensions/intellij/src/main/kotlin/com/github/continuedev/continueintellijextension/continue/process/ContinueBinaryProcess.kt

@@ -1,49 +1,62 @@
 package com.github.continuedev.continueintellijextension.`continue`.process
 
+import com.github.continuedev.continueintellijextension.error.ContinuePostHogService
 import com.github.continuedev.continueintellijextension.error.ContinueSentryService
 import com.intellij.openapi.components.service
 import com.intellij.openapi.diagnostic.Logger
 import kotlinx.coroutines.*
 import kotlinx.coroutines.channels.Channel
 import java.io.BufferedReader
-import java.io.IOException
 import java.io.InputStreamReader
 import java.io.OutputStreamWriter
 
 class ContinueProcessHandler(
-    parentScope: CoroutineScope,
-    private val process: ContinueProcess,
-    handleMessage: (String) -> (Unit)
+    private val parentScope: CoroutineScope,
+    private val readMessage: (String) -> (Unit),
+    private val createProcess: () -> ContinueProcess
 ) {
-    private val innerJob = Job()
-    private val scope = CoroutineScope(parentScope.coroutineContext + innerJob)
     private val pendingWrites = Channel<String>(Channel.UNLIMITED)
-    private val writer = OutputStreamWriter(process.output)
-    private val reader = BufferedReader(InputStreamReader(process.input))
-    private val log = Logger.getInstance(ContinueProcessHandler::class.java)
+    private val backoff = BackoffCalculator()
+    private var processScope: CoroutineScope? = null
+    private var process: ContinueProcess? = null
 
     init {
-        scope.launch(Dispatchers.IO) {
-            try {
-                while (isActive) {
-                    val line = reader.readLine()
-                    if (line != null && line.isNotEmpty()) {
-                        try {
-                            log.debug("Handle: $line")
-                            handleMessage(line)
-                        } catch (e: Exception) {
-                            service<ContinueSentryService>().report(e, "Error handling message: $line")
-                        }
-                    } else
-                        delay(100)
-                }
-            } catch (e: IOException) {
-                service<ContinueSentryService>().report(e)
+        restart()
+    }
+
+    fun restart() {
+        LOG.warn("Starting Continue process")
+        processScope?.cancel()
+        process?.close()
+
+        val handler = CoroutineExceptionHandler { _, e ->
+            service<ContinueSentryService>().report(e)
+            service<ContinuePostHogService>().capture("jetbrains_core_exit", mapOf("error" to e))
+
+            val backoffDuration = backoff.nextDuration()
+            LOG.warn("Process failed! Restarting in $backoffDuration")
+            parentScope.launch {
+                delay(backoffDuration)
+                restart()
+            }
+        }
+
+        val job = SupervisorJob(parentScope.coroutineContext.job)
+        processScope = CoroutineScope(parentScope.coroutineContext + job + handler)
+        process = createProcess()
+
+        val reader = BufferedReader(InputStreamReader(process!!.input))
+        val writer = OutputStreamWriter(process!!.output)
+
+        processScope!!.launch(Dispatchers.IO) {
+            while (isActive) {
+                val line = reader.readLine()
+                if (line != null && line.isNotEmpty())
+                    readMessage(line)
             }
         }
-        scope.launch(Dispatchers.IO) {
+        processScope!!.launch(Dispatchers.IO) {
             for (message in pendingWrites) {
-                log.debug("Write: $message")
                 writer.write(message)
                 writer.write("\r\n")
                 writer.flush()
@@ -54,12 +67,7 @@ class ContinueProcessHandler(
     fun write(message: String) =
         pendingWrites.trySend(message)
 
-    fun close() {
-        innerJob.cancel()
-        scope.launch(Dispatchers.IO) {
-            reader.close()
-            writer.close()
-            process.close()
-        }
+    private companion object {
+        private val LOG = Logger.getInstance(ContinueProcessHandler::class.java.simpleName)
     }
 }

extensions/intellij/src/main/kotlin/com/github/continuedev/continueintellijextension/continue/process/ContinueProcessHandler.kt

@@ -28,7 +28,7 @@ class ProxyPoolingActivity : StartupActivity {
 
     private fun onSettingsChanged(project: Project) {
         log.warn("Proxy settings changed, restarting")
-        project.service<ContinuePluginService>().coreMessengerManager?.coreMessenger?.restart()
+        project.service<ContinuePluginService>().coreMessenger?.restart()
     }
 }
 

extensions/intellij/src/main/kotlin/com/github/continuedev/continueintellijextension/proxy/ProxyPoolingActivity.kt

@@ -1,34 +1,28 @@
 package com.github.continuedev.continueintellijextension.services
 
 import com.github.continuedev.continueintellijextension.`continue`.CoreMessenger
-import com.github.continuedev.continueintellijextension.`continue`.CoreMessengerManager
 import com.github.continuedev.continueintellijextension.`continue`.DiffManager
 import com.github.continuedev.continueintellijextension.`continue`.IdeProtocolClient
 import com.github.continuedev.continueintellijextension.listeners.ActiveHandlerManager
 import com.github.continuedev.continueintellijextension.listeners.DocumentChangeTracker
-import com.github.continuedev.continueintellijextension.toolWindow.ContinuePluginToolWindowFactory
 import com.github.continuedev.continueintellijextension.utils.uuid
+import com.intellij.openapi.Disposable
 import com.intellij.openapi.components.Service
 import com.intellij.openapi.components.service
 import com.intellij.openapi.editor.EditorFactory
 import com.intellij.openapi.project.DumbAware
 import com.intellij.openapi.project.Project
 import com.intellij.openapi.util.Disposer
-import kotlinx.coroutines.CoroutineScope
-import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.cancel
 import kotlin.properties.Delegates
 
 @Service(Service.Level.PROJECT)
-class ContinuePluginService(private val project: Project) : DumbAware {
-    private val coroutineScope = CoroutineScope(Dispatchers.Main)
+class ContinuePluginService(private val project: Project) : Disposable, DumbAware {
     var listener: (() -> Unit)? = null
     var ideProtocolClient: IdeProtocolClient? by Delegates.observable(null) { _, _, _ ->
         synchronized(this) { listener?.also { listener = null }?.invoke() }
     }
-    var coreMessengerManager: CoreMessengerManager? = null
-    val coreMessenger: CoreMessenger?
-        get() = coreMessengerManager?.coreMessenger
+    var coreMessenger: CoreMessenger? = null
     var workspacePaths: Array<String>? = null
     var windowId: String = uuid()
     var diffManager: DiffManager? = null
@@ -57,14 +51,6 @@ class ContinuePluginService(private val project: Project) : DumbAware {
         }
     }
 
-    fun dispose() {
-        coroutineScope.cancel()
-        coreMessenger?.coroutineScope?.let {
-            it.cancel()
-            coreMessenger?.close()
-        }
-    }
-
     /**
      * Add a listener for protocolClient initialization.
      * Currently, only one needs to be processed. If there are more than one,
@@ -83,4 +69,8 @@ class ContinuePluginService(private val project: Project) : DumbAware {
             listener()
         }
     }
+
+    override fun dispose() {
+        coreMessenger?.coroutineScope?.cancel()
+    }
 }